diff --git a/src/lib/postgres.mjs b/src/lib/postgres.mjs index a44a8fc..72c8f3b 100644 --- a/src/lib/postgres.mjs +++ b/src/lib/postgres.mjs @@ -6,8 +6,6 @@ import { buildCashEquivalentValuationAssets } from '../core/portfolio-metrics.mj import { deriveQuoteOutcomeRecords } from '../core/quote-outcomes.mjs'; import { QUOTE_LIFECYCLE_STATISTIC_GRAINS, - buildQuoteLifecycleStatistics, - classifyQuoteLifecycleStatisticsBucket, normalizeQuoteLifecycleStatisticsGrains, normalizeQuoteLifecycleStatisticsRow, } from '../core/quote-lifecycle-statistics.mjs'; @@ -3206,26 +3204,10 @@ export async function refreshQuoteLifecycleStatistics(pool, { now = new Date().toISOString(), grains = QUOTE_LIFECYCLE_STATISTIC_GRAINS, } = {}) { - const evidence = await loadQuoteLifecycleStatisticsEvidence(pool); - const lifecycleRows = deriveQuoteLifecycleRows({ - recentQuotes: evidence.recentQuotes, - recentTradeDecisions: evidence.recentTradeDecisions, - recentExecuteTradeCommands: evidence.recentExecuteTradeCommands, - recentExecutionResults: evidence.recentExecutionResults, - recentQuoteOutcomes: evidence.recentQuoteOutcomes, - limit: null, - }); - const subjectUpsertResult = await upsertQuoteLifecycleStatisticSubjects(pool, { - lifecycleRows, + const subjectUpsertResult = await upsertQuoteLifecycleStatisticSubjectsFromHistory(pool, { now, }); - const statisticSubjectRows = await loadQuoteLifecycleStatisticSubjects(pool); - const missingIdentifierRows = lifecycleRows.filter((row) => !String(row?.quote_id || '').trim()); - const statistics = buildQuoteLifecycleStatistics({ - lifecycleRows: [ - ...statisticSubjectRows, - ...missingIdentifierRows, - ], + const statistics = await buildQuoteLifecycleStatisticsFromSubjects(pool, { grains, computedAt: now, }); @@ -3235,108 +3217,228 @@ export async function refreshQuoteLifecycleStatistics(pool, { quote_count: statistics.find((row) => row.grain === 'all_time')?.quote_count || 0, statistic_row_count: statistics.length, evidence_counts: { - quotes: evidence.recentQuotes.length, - decisions: evidence.recentTradeDecisions.length, - commands: evidence.recentExecuteTradeCommands.length, - execution_results: evidence.recentExecutionResults.length, - outcomes: evidence.recentQuoteOutcomes.length, - lifecycle_rows: lifecycleRows.length, - statistic_subjects: statisticSubjectRows.length, + statistic_subjects: subjectUpsertResult.statistic_subjects, statistic_subject_upserts: subjectUpsertResult.upserted_count, }, statistics, }; } -async function upsertQuoteLifecycleStatisticSubjects(pool, { - lifecycleRows = [], +async function upsertQuoteLifecycleStatisticSubjectsFromHistory(pool, { now = new Date().toISOString(), } = {}) { - let upsertedCount = 0; - for (const row of lifecycleRows || []) { - const subject = buildQuoteLifecycleStatisticSubject(row, { now }); - if (!subject) continue; - await pool.query( - ` - INSERT INTO ${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE} ( - quote_id, - first_activity_at, - latest_stage_at, - bucket, - lifecycle_state, - reason_code, - updated_at, - payload - ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8::jsonb) - ON CONFLICT (quote_id) DO UPDATE SET - first_activity_at = CASE - WHEN ${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE}.first_activity_at IS NULL THEN EXCLUDED.first_activity_at - WHEN EXCLUDED.first_activity_at IS NULL THEN ${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE}.first_activity_at - ELSE LEAST(${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE}.first_activity_at, EXCLUDED.first_activity_at) - END, - latest_stage_at = CASE - WHEN ${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE}.latest_stage_at IS NULL THEN EXCLUDED.latest_stage_at - WHEN EXCLUDED.latest_stage_at IS NULL THEN ${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE}.latest_stage_at - ELSE GREATEST(${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE}.latest_stage_at, EXCLUDED.latest_stage_at) - END, - bucket = CASE - WHEN ${quoteLifecycleBucketRankSql('EXCLUDED.bucket')} > ${quoteLifecycleBucketRankSql(`${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE}.bucket`)} - OR ( - ${quoteLifecycleBucketRankSql('EXCLUDED.bucket')} = ${quoteLifecycleBucketRankSql(`${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE}.bucket`)} - AND COALESCE(EXCLUDED.latest_stage_at, '-infinity'::timestamptz) >= COALESCE(${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE}.latest_stage_at, '-infinity'::timestamptz) - ) - THEN EXCLUDED.bucket - ELSE ${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE}.bucket - END, - lifecycle_state = CASE - WHEN ${quoteLifecycleBucketRankSql('EXCLUDED.bucket')} > ${quoteLifecycleBucketRankSql(`${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE}.bucket`)} - OR ( - ${quoteLifecycleBucketRankSql('EXCLUDED.bucket')} = ${quoteLifecycleBucketRankSql(`${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE}.bucket`)} - AND COALESCE(EXCLUDED.latest_stage_at, '-infinity'::timestamptz) >= COALESCE(${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE}.latest_stage_at, '-infinity'::timestamptz) - ) - THEN EXCLUDED.lifecycle_state - ELSE ${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE}.lifecycle_state - END, - reason_code = CASE - WHEN ${quoteLifecycleBucketRankSql('EXCLUDED.bucket')} > ${quoteLifecycleBucketRankSql(`${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE}.bucket`)} - OR ( - ${quoteLifecycleBucketRankSql('EXCLUDED.bucket')} = ${quoteLifecycleBucketRankSql(`${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE}.bucket`)} - AND COALESCE(EXCLUDED.latest_stage_at, '-infinity'::timestamptz) >= COALESCE(${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE}.latest_stage_at, '-infinity'::timestamptz) - ) - THEN EXCLUDED.reason_code - ELSE ${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE}.reason_code - END, - payload = CASE - WHEN ${quoteLifecycleBucketRankSql('EXCLUDED.bucket')} > ${quoteLifecycleBucketRankSql(`${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE}.bucket`)} - OR ( - ${quoteLifecycleBucketRankSql('EXCLUDED.bucket')} = ${quoteLifecycleBucketRankSql(`${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE}.bucket`)} - AND COALESCE(EXCLUDED.latest_stage_at, '-infinity'::timestamptz) >= COALESCE(${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE}.latest_stage_at, '-infinity'::timestamptz) - ) - THEN EXCLUDED.payload - ELSE ${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE}.payload - END, - updated_at = GREATEST(${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE}.updated_at, EXCLUDED.updated_at) - `, - [ - subject.quote_id, - subject.first_activity_at, - subject.latest_stage_at, - subject.bucket, - subject.lifecycle_state, - subject.reason_code, - subject.updated_at, - JSON.stringify(subject.payload), - ], - ); - upsertedCount += 1; - } - return { upserted_count: upsertedCount }; -} - -async function loadQuoteLifecycleStatisticSubjects(pool) { const result = await pool.query( ` - SELECT + WITH source_rows AS ( + SELECT + NULLIF(COALESCE(quote_id, payload->>'quote_id'), '') AS quote_id, + COALESCE(observed_at, ingested_at) AS first_activity_at, + COALESCE(observed_at, ingested_at) AS latest_stage_at, + 'observed_no_decision' AS bucket, + 'observed' AS lifecycle_state, + COALESCE(NULLIF(payload->>'reason_code', ''), 'observed_no_decision') AS reason_code, + NULLIF(payload->>'decision_id', '') AS decision_id, + NULLIF(payload->>'command_id', '') AS command_id, + jsonb_build_object( + 'quote_id', NULLIF(COALESCE(quote_id, payload->>'quote_id'), ''), + 'decision_id', NULLIF(payload->>'decision_id', ''), + 'command_id', NULLIF(payload->>'command_id', ''), + 'lifecycle_state', 'observed', + 'reason_code', COALESCE(NULLIF(payload->>'reason_code', ''), 'observed_no_decision'), + 'statistics_bucket', 'observed_no_decision', + 'first_activity_at', COALESCE(observed_at, ingested_at), + 'latest_stage_at', COALESCE(observed_at, ingested_at) + ) AS payload + FROM swap_demand_events + WHERE NULLIF(COALESCE(quote_id, payload->>'quote_id'), '') IS NOT NULL + + UNION ALL + + SELECT + NULLIF(COALESCE(quote_id, payload->>'quote_id'), '') AS quote_id, + COALESCE(observed_at, ingested_at) AS first_activity_at, + COALESCE(observed_at, ingested_at) AS latest_stage_at, + CASE + WHEN regexp_replace(lower(trim(COALESCE(payload->>'decision', ''))), '[[:space:]-]+', '_', 'g') = 'rejected' + THEN 'rejected_by_strategy' + WHEN regexp_replace(lower(trim(COALESCE(payload->>'decision', ''))), '[[:space:]-]+', '_', 'g') = 'blocked' + THEN 'blocked_before_submit' + WHEN regexp_replace(lower(trim(COALESCE(payload->>'decision', ''))), '[[:space:]-]+', '_', 'g') IN ('approved', 'actionable') + THEN 'approved_no_command' + ELSE 'unavailable' + END AS bucket, + CASE + WHEN regexp_replace(lower(trim(COALESCE(payload->>'decision', ''))), '[[:space:]-]+', '_', 'g') = 'rejected' + THEN 'rejected' + WHEN regexp_replace(lower(trim(COALESCE(payload->>'decision', ''))), '[[:space:]-]+', '_', 'g') = 'blocked' + THEN 'blocked' + WHEN regexp_replace(lower(trim(COALESCE(payload->>'decision', ''))), '[[:space:]-]+', '_', 'g') IN ('approved', 'actionable') + THEN 'evaluated' + ELSE 'observed' + END AS lifecycle_state, + COALESCE(NULLIF(payload->>'decision_reason', ''), 'reason_unknown') AS reason_code, + NULLIF(payload->>'decision_id', '') AS decision_id, + NULLIF(payload->>'command_id', '') AS command_id, + jsonb_build_object( + 'quote_id', NULLIF(COALESCE(quote_id, payload->>'quote_id'), ''), + 'decision_id', NULLIF(payload->>'decision_id', ''), + 'command_id', NULLIF(payload->>'command_id', ''), + 'lifecycle_state', regexp_replace(lower(trim(COALESCE(payload->>'decision', 'observed'))), '[[:space:]-]+', '_', 'g'), + 'reason_code', COALESCE(NULLIF(payload->>'decision_reason', ''), 'reason_unknown'), + 'first_activity_at', COALESCE(observed_at, ingested_at), + 'latest_stage_at', COALESCE(observed_at, ingested_at) + ) AS payload + FROM trade_decisions + WHERE NULLIF(COALESCE(quote_id, payload->>'quote_id'), '') IS NOT NULL + + UNION ALL + + SELECT + NULLIF(COALESCE(quote_id, payload->>'quote_id'), '') AS quote_id, + COALESCE(observed_at, ingested_at) AS first_activity_at, + COALESCE(observed_at, ingested_at) AS latest_stage_at, + 'awaiting_executor' AS bucket, + 'command_emitted' AS lifecycle_state, + 'awaiting_executor' AS reason_code, + NULLIF(payload->>'decision_id', '') AS decision_id, + NULLIF(payload->>'command_id', '') AS command_id, + jsonb_build_object( + 'quote_id', NULLIF(COALESCE(quote_id, payload->>'quote_id'), ''), + 'decision_id', NULLIF(payload->>'decision_id', ''), + 'command_id', NULLIF(payload->>'command_id', ''), + 'lifecycle_state', 'command_emitted', + 'reason_code', 'awaiting_executor', + 'statistics_bucket', 'awaiting_executor', + 'first_activity_at', COALESCE(observed_at, ingested_at), + 'latest_stage_at', COALESCE(observed_at, ingested_at) + ) AS payload + FROM execute_trade_commands + WHERE NULLIF(COALESCE(quote_id, payload->>'quote_id'), '') IS NOT NULL + + UNION ALL + + SELECT + NULLIF(COALESCE(quote_id, payload->>'quote_id'), '') AS quote_id, + COALESCE(observed_at, ingested_at) AS first_activity_at, + COALESCE(observed_at, ingested_at) AS latest_stage_at, + CASE + WHEN regexp_replace(lower(trim(COALESCE(payload->>'outcome_status', payload->>'venue_outcome_status', payload->>'trade_outcome_status', ''))), '[[:space:]-]+', '_', 'g') + IN ('completed', 'filled', 'settled', 'finalized') + THEN 'success' + WHEN regexp_replace(lower(trim(COALESCE(payload->>'outcome_status', payload->>'venue_outcome_status', payload->>'trade_outcome_status', ''))), '[[:space:]-]+', '_', 'g') + IN ('expired', 'not_filled', 'cancelled') + THEN 'not_filled' + WHEN regexp_replace(lower(trim(COALESCE(payload->>'status', ''))), '[[:space:]-]+', '_', 'g') = 'failed' + THEN 'submission_failed' + WHEN regexp_replace(lower(trim(COALESCE(payload->>'status', ''))), '[[:space:]-]+', '_', 'g') = 'rejected' + THEN 'blocked_before_submit' + WHEN regexp_replace(lower(trim(COALESCE(payload->>'status', ''))), '[[:space:]-]+', '_', 'g') = 'submitted' + THEN 'submitted_no_reply' + ELSE 'unavailable' + END AS bucket, + CASE + WHEN regexp_replace(lower(trim(COALESCE(payload->>'outcome_status', payload->>'venue_outcome_status', payload->>'trade_outcome_status', ''))), '[[:space:]-]+', '_', 'g') + IN ('completed', 'filled', 'settled', 'finalized') + THEN 'completed' + WHEN regexp_replace(lower(trim(COALESCE(payload->>'outcome_status', payload->>'venue_outcome_status', payload->>'trade_outcome_status', ''))), '[[:space:]-]+', '_', 'g') + IN ('expired', 'not_filled', 'cancelled') + THEN 'not_filled' + WHEN regexp_replace(lower(trim(COALESCE(payload->>'status', ''))), '[[:space:]-]+', '_', 'g') = 'failed' + THEN 'failed' + WHEN regexp_replace(lower(trim(COALESCE(payload->>'status', ''))), '[[:space:]-]+', '_', 'g') = 'rejected' + THEN 'blocked' + WHEN regexp_replace(lower(trim(COALESCE(payload->>'status', ''))), '[[:space:]-]+', '_', 'g') = 'submitted' + THEN 'submitted' + ELSE 'observed' + END AS lifecycle_state, + COALESCE(NULLIF(payload->>'outcome_reason', ''), NULLIF(payload->>'failure_category', ''), NULLIF(payload->>'result_code', ''), 'reason_unknown') AS reason_code, + NULLIF(payload->>'decision_id', '') AS decision_id, + NULLIF(payload->>'command_id', '') AS command_id, + jsonb_build_object( + 'quote_id', NULLIF(COALESCE(quote_id, payload->>'quote_id'), ''), + 'decision_id', NULLIF(payload->>'decision_id', ''), + 'command_id', NULLIF(payload->>'command_id', ''), + 'reason_code', COALESCE(NULLIF(payload->>'outcome_reason', ''), NULLIF(payload->>'failure_category', ''), NULLIF(payload->>'result_code', ''), 'reason_unknown'), + 'outcome_status', COALESCE(payload->>'outcome_status', payload->>'venue_outcome_status', payload->>'trade_outcome_status'), + 'attribution_status', payload->>'attribution_status', + 'first_activity_at', COALESCE(observed_at, ingested_at), + 'latest_stage_at', COALESCE(observed_at, ingested_at) + ) AS payload + FROM trade_execution_results + WHERE NULLIF(COALESCE(quote_id, payload->>'quote_id'), '') IS NOT NULL + + UNION ALL + + SELECT + NULLIF(COALESCE(quote_id, payload->>'quote_id'), '') AS quote_id, + COALESCE(submitted_at, command_at, outcome_observed_at, computed_at) AS first_activity_at, + COALESCE(outcome_observed_at, submitted_at, command_at, computed_at) AS latest_stage_at, + CASE + WHEN regexp_replace(lower(trim(COALESCE(outcome_status, payload->>'outcome_status', ''))), '[[:space:]-]+', '_', 'g') + IN ('completed', 'filled', 'settled', 'finalized') + THEN 'success' + WHEN regexp_replace(lower(trim(COALESCE(outcome_status, payload->>'outcome_status', ''))), '[[:space:]-]+', '_', 'g') + IN ('expired', 'not_filled', 'cancelled') + THEN 'not_filled' + WHEN regexp_replace(lower(trim(COALESCE(execution_result_status, payload->>'execution_result_status', ''))), '[[:space:]-]+', '_', 'g') = 'failed' + THEN 'submission_failed' + WHEN regexp_replace(lower(trim(COALESCE(execution_result_status, payload->>'execution_result_status', ''))), '[[:space:]-]+', '_', 'g') = 'submitted' + THEN 'submitted_no_reply' + ELSE 'unavailable' + END AS bucket, + CASE + WHEN regexp_replace(lower(trim(COALESCE(outcome_status, payload->>'outcome_status', ''))), '[[:space:]-]+', '_', 'g') + IN ('completed', 'filled', 'settled', 'finalized') + THEN 'completed' + WHEN regexp_replace(lower(trim(COALESCE(outcome_status, payload->>'outcome_status', ''))), '[[:space:]-]+', '_', 'g') + IN ('expired', 'not_filled', 'cancelled') + THEN 'not_filled' + WHEN regexp_replace(lower(trim(COALESCE(execution_result_status, payload->>'execution_result_status', ''))), '[[:space:]-]+', '_', 'g') = 'failed' + THEN 'failed' + WHEN regexp_replace(lower(trim(COALESCE(execution_result_status, payload->>'execution_result_status', ''))), '[[:space:]-]+', '_', 'g') = 'submitted' + THEN 'submitted' + ELSE 'observed' + END AS lifecycle_state, + COALESCE(NULLIF(payload->>'outcome_reason', ''), NULLIF(execution_result_code, ''), NULLIF(payload->>'execution_result_code', ''), 'reason_unknown') AS reason_code, + NULLIF(COALESCE(decision_id, payload->>'decision_id'), '') AS decision_id, + NULLIF(COALESCE(command_id, payload->>'command_id'), '') AS command_id, + jsonb_build_object( + 'quote_id', NULLIF(COALESCE(quote_id, payload->>'quote_id'), ''), + 'decision_id', NULLIF(COALESCE(decision_id, payload->>'decision_id'), ''), + 'command_id', NULLIF(COALESCE(command_id, payload->>'command_id'), ''), + 'reason_code', COALESCE(NULLIF(payload->>'outcome_reason', ''), NULLIF(execution_result_code, ''), NULLIF(payload->>'execution_result_code', ''), 'reason_unknown'), + 'outcome_status', COALESCE(outcome_status, payload->>'outcome_status'), + 'attribution_status', COALESCE(attribution_status, payload->>'attribution_status'), + 'has_settlement_evidence', attributed_inventory_delta IS NOT NULL OR payload ? 'attributed_inventory_delta', + 'first_activity_at', COALESCE(submitted_at, command_at, outcome_observed_at, computed_at), + 'latest_stage_at', COALESCE(outcome_observed_at, submitted_at, command_at, computed_at) + ) AS payload + FROM ${QUOTE_OUTCOMES_TABLE} + WHERE NULLIF(COALESCE(quote_id, payload->>'quote_id'), '') IS NOT NULL + ), source_ranked AS ( + SELECT + *, + ${quoteLifecycleBucketRankSql('bucket')} AS bucket_rank + FROM source_rows + WHERE quote_id IS NOT NULL + ), first_activity AS ( + SELECT quote_id, MIN(first_activity_at) AS first_activity_at + FROM source_ranked + GROUP BY quote_id + ), winners AS ( + SELECT * + FROM ( + SELECT + source_ranked.*, + ROW_NUMBER() OVER ( + PARTITION BY quote_id + ORDER BY bucket_rank DESC, latest_stage_at DESC NULLS LAST + ) AS row_rank + FROM source_ranked + ) ranked + WHERE row_rank = 1 + ) + INSERT INTO ${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE} ( quote_id, first_activity_at, latest_stage_at, @@ -3345,106 +3447,224 @@ async function loadQuoteLifecycleStatisticSubjects(pool) { reason_code, updated_at, payload - FROM ${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE} - ORDER BY first_activity_at ASC NULLS LAST, quote_id ASC + ) + SELECT + winners.quote_id, + first_activity.first_activity_at, + winners.latest_stage_at, + winners.bucket, + winners.lifecycle_state, + winners.reason_code, + $1::timestamptz, + winners.payload + || jsonb_build_object( + 'statistics_bucket', winners.bucket, + 'lifecycle_state', winners.lifecycle_state, + 'first_activity_at', first_activity.first_activity_at, + 'latest_stage_at', winners.latest_stage_at + ) + FROM winners + JOIN first_activity + ON first_activity.quote_id = winners.quote_id + ON CONFLICT (quote_id) DO UPDATE SET + first_activity_at = CASE + WHEN ${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE}.first_activity_at IS NULL THEN EXCLUDED.first_activity_at + WHEN EXCLUDED.first_activity_at IS NULL THEN ${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE}.first_activity_at + ELSE LEAST(${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE}.first_activity_at, EXCLUDED.first_activity_at) + END, + latest_stage_at = CASE + WHEN ${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE}.latest_stage_at IS NULL THEN EXCLUDED.latest_stage_at + WHEN EXCLUDED.latest_stage_at IS NULL THEN ${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE}.latest_stage_at + ELSE GREATEST(${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE}.latest_stage_at, EXCLUDED.latest_stage_at) + END, + bucket = CASE + WHEN ${quoteLifecycleBucketRankSql('EXCLUDED.bucket')} > ${quoteLifecycleBucketRankSql(`${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE}.bucket`)} + OR ( + ${quoteLifecycleBucketRankSql('EXCLUDED.bucket')} = ${quoteLifecycleBucketRankSql(`${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE}.bucket`)} + AND COALESCE(EXCLUDED.latest_stage_at, '-infinity'::timestamptz) >= COALESCE(${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE}.latest_stage_at, '-infinity'::timestamptz) + ) + THEN EXCLUDED.bucket + ELSE ${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE}.bucket + END, + lifecycle_state = CASE + WHEN ${quoteLifecycleBucketRankSql('EXCLUDED.bucket')} > ${quoteLifecycleBucketRankSql(`${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE}.bucket`)} + OR ( + ${quoteLifecycleBucketRankSql('EXCLUDED.bucket')} = ${quoteLifecycleBucketRankSql(`${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE}.bucket`)} + AND COALESCE(EXCLUDED.latest_stage_at, '-infinity'::timestamptz) >= COALESCE(${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE}.latest_stage_at, '-infinity'::timestamptz) + ) + THEN EXCLUDED.lifecycle_state + ELSE ${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE}.lifecycle_state + END, + reason_code = CASE + WHEN ${quoteLifecycleBucketRankSql('EXCLUDED.bucket')} > ${quoteLifecycleBucketRankSql(`${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE}.bucket`)} + OR ( + ${quoteLifecycleBucketRankSql('EXCLUDED.bucket')} = ${quoteLifecycleBucketRankSql(`${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE}.bucket`)} + AND COALESCE(EXCLUDED.latest_stage_at, '-infinity'::timestamptz) >= COALESCE(${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE}.latest_stage_at, '-infinity'::timestamptz) + ) + THEN EXCLUDED.reason_code + ELSE ${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE}.reason_code + END, + payload = CASE + WHEN ${quoteLifecycleBucketRankSql('EXCLUDED.bucket')} > ${quoteLifecycleBucketRankSql(`${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE}.bucket`)} + OR ( + ${quoteLifecycleBucketRankSql('EXCLUDED.bucket')} = ${quoteLifecycleBucketRankSql(`${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE}.bucket`)} + AND COALESCE(EXCLUDED.latest_stage_at, '-infinity'::timestamptz) >= COALESCE(${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE}.latest_stage_at, '-infinity'::timestamptz) + ) + THEN EXCLUDED.payload + ELSE ${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE}.payload + END, + updated_at = GREATEST(${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE}.updated_at, EXCLUDED.updated_at) `, + [now], ); - return result.rows.map(normalizeQuoteLifecycleStatisticSubjectRow); -} -function buildQuoteLifecycleStatisticSubject(row = {}, { now = new Date().toISOString() } = {}) { - const quoteId = String(row?.quote_id || '').trim(); - if (!quoteId) return null; - const firstActivityAt = firstQuoteLifecycleStatisticTimestamp(row); - const latestStageAt = latestQuoteLifecycleStatisticTimestamp(row); - const bucket = classifyQuoteLifecycleStatisticsBucket(row); + const countResult = await pool.query(` + SELECT COUNT(*)::INT AS subject_count + FROM ${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE} + `); return { - quote_id: quoteId, - first_activity_at: firstActivityAt, - latest_stage_at: latestStageAt, - bucket, - lifecycle_state: row.lifecycle_state || null, - reason_code: row.reason_code || null, - updated_at: toIsoTimestamp(now) || new Date().toISOString(), - payload: { - quote_id: quoteId, - decision_id: row.decision_id || null, - command_id: row.command_id || null, - lifecycle_state: row.lifecycle_state || null, - lifecycle_label: row.lifecycle_label || null, - reason_code: row.reason_code || null, - statistics_bucket: bucket, - first_activity_at: firstActivityAt, - latest_stage_at: latestStageAt, - outcome_status: row.outcome_status || row.outcome?.outcome_status || null, - attribution_status: row.attribution_status || row.outcome?.attribution_status || null, - has_settlement_evidence: row.has_settlement_evidence === true, - }, + upserted_count: Number(result.rowCount || 0), + statistic_subjects: Number(countResult.rows[0]?.subject_count || 0), }; } -function normalizeQuoteLifecycleStatisticSubjectRow(row = {}) { - const payload = row.payload || {}; - return { - quote_id: row.quote_id || payload.quote_id || null, - decision_id: payload.decision_id || null, - command_id: payload.command_id || null, - quote_activity_at: toIsoTimestamp(row.first_activity_at || payload.first_activity_at), - latest_stage_at: toIsoTimestamp(row.latest_stage_at || payload.latest_stage_at), - lifecycle_state: row.lifecycle_state || payload.lifecycle_state || lifecycleStateForStatisticBucket(row.bucket), - reason_code: row.reason_code || payload.reason_code || row.bucket || 'unavailable', - statistics_bucket: row.bucket || payload.statistics_bucket || 'unavailable', - }; -} +async function buildQuoteLifecycleStatisticsFromSubjects(pool, { + grains = QUOTE_LIFECYCLE_STATISTIC_GRAINS, + computedAt = new Date().toISOString(), +} = {}) { + const normalizedGrains = normalizeQuoteLifecycleStatisticsGrains(grains); + const result = await pool.query( + ` + WITH subjects AS ( + SELECT + first_activity_at, + COALESCE(bucket, payload->>'statistics_bucket', 'unavailable') AS bucket + FROM ${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE} + ), grain_rows AS ( + SELECT + 'all_time'::text AS grain, + NULL::timestamptz AS window_start, + NULL::timestamptz AS window_end, + bucket, + first_activity_at IS NOT NULL AS has_timestamp + FROM subjects + WHERE 'all_time' = ANY($1::text[]) -function firstQuoteLifecycleStatisticTimestamp(row = {}) { - return earliestTimestamp([ - row.quote_activity_at, - row.quote_observed_at, - row.decision_at, - row.command_at, - row.execution_result_at, - row.outcome_observed_at, - row.latest_stage_at, - ]); -} + UNION ALL -function latestQuoteLifecycleStatisticTimestamp(row = {}) { - return latestTimestamp([ - row.latest_stage_at, - row.outcome_observed_at, - row.execution_result_at, - row.command_at, - row.decision_at, - row.quote_observed_at, - row.quote_activity_at, - ]); -} + SELECT + 'month'::text AS grain, + (date_trunc('month', first_activity_at AT TIME ZONE 'UTC') AT TIME ZONE 'UTC') AS window_start, + (date_trunc('month', first_activity_at AT TIME ZONE 'UTC') AT TIME ZONE 'UTC') + INTERVAL '1 month' AS window_end, + bucket, + true AS has_timestamp + FROM subjects + WHERE 'month' = ANY($1::text[]) + AND first_activity_at IS NOT NULL -function lifecycleStateForStatisticBucket(bucket) { - switch (bucket) { - case 'success': - return 'completed'; - case 'not_filled': - return 'not_filled'; - case 'submission_failed': - return 'failed'; - case 'submitted_no_reply': - return 'submitted'; - case 'awaiting_executor': - return 'command_emitted'; - case 'approved_no_command': - return 'evaluated'; - case 'blocked_before_submit': - return 'blocked'; - case 'rejected_by_strategy': - return 'rejected'; - case 'observed_no_decision': - return 'observed'; - case 'unavailable': - default: - return 'observed'; - } + UNION ALL + + SELECT + 'week'::text AS grain, + (date_trunc('week', first_activity_at AT TIME ZONE 'UTC') AT TIME ZONE 'UTC') AS window_start, + (date_trunc('week', first_activity_at AT TIME ZONE 'UTC') AT TIME ZONE 'UTC') + INTERVAL '1 week' AS window_end, + bucket, + true AS has_timestamp + FROM subjects + WHERE 'week' = ANY($1::text[]) + AND first_activity_at IS NOT NULL + + UNION ALL + + SELECT + 'day'::text AS grain, + (date_trunc('day', first_activity_at AT TIME ZONE 'UTC') AT TIME ZONE 'UTC') AS window_start, + (date_trunc('day', first_activity_at AT TIME ZONE 'UTC') AT TIME ZONE 'UTC') + INTERVAL '1 day' AS window_end, + bucket, + true AS has_timestamp + FROM subjects + WHERE 'day' = ANY($1::text[]) + AND first_activity_at IS NOT NULL + + UNION ALL + + SELECT + 'hour'::text AS grain, + (date_trunc('hour', first_activity_at AT TIME ZONE 'UTC') AT TIME ZONE 'UTC') AS window_start, + (date_trunc('hour', first_activity_at AT TIME ZONE 'UTC') AT TIME ZONE 'UTC') + INTERVAL '1 hour' AS window_end, + bucket, + true AS has_timestamp + FROM subjects + WHERE 'hour' = ANY($1::text[]) + AND first_activity_at IS NOT NULL + + UNION ALL + + SELECT + 'five_minute'::text AS grain, + to_timestamp(floor(extract(epoch FROM first_activity_at) / 300) * 300) AS window_start, + to_timestamp(floor(extract(epoch FROM first_activity_at) / 300) * 300) + INTERVAL '5 minutes' AS window_end, + bucket, + true AS has_timestamp + FROM subjects + WHERE 'five_minute' = ANY($1::text[]) + AND first_activity_at IS NOT NULL + ), group_totals AS ( + SELECT + grain, + window_start, + window_end, + COUNT(*)::INT AS quote_count, + COUNT(*) FILTER (WHERE NOT has_timestamp)::INT AS missing_timestamp_count + FROM grain_rows + GROUP BY grain, window_start, window_end + ), bucket_groups AS ( + SELECT + grain, + window_start, + window_end, + bucket, + COUNT(*)::INT AS bucket_count + FROM grain_rows + GROUP BY grain, window_start, window_end, bucket + ), bucket_counts AS ( + SELECT + grain, + window_start, + window_end, + jsonb_object_agg(bucket, bucket_count) AS bucket_counts + FROM bucket_groups + GROUP BY grain, window_start, window_end + ) + SELECT + group_totals.grain, + group_totals.window_start, + group_totals.window_end, + group_totals.quote_count, + 0::INT AS missing_identifier_count, + group_totals.missing_timestamp_count, + COALESCE(bucket_counts.bucket_counts, '{}'::jsonb) AS bucket_counts, + $2::timestamptz AS computed_at + FROM group_totals + LEFT JOIN bucket_counts + ON bucket_counts.grain = group_totals.grain + AND bucket_counts.window_start IS NOT DISTINCT FROM group_totals.window_start + AND bucket_counts.window_end IS NOT DISTINCT FROM group_totals.window_end + ORDER BY + CASE group_totals.grain + WHEN 'all_time' THEN 0 + WHEN 'month' THEN 1 + WHEN 'week' THEN 2 + WHEN 'day' THEN 3 + WHEN 'hour' THEN 4 + WHEN 'five_minute' THEN 5 + ELSE 99 + END, + group_totals.window_start DESC NULLS LAST + `, + [normalizedGrains, computedAt], + ); + return result.rows.map(normalizeQuoteLifecycleStatisticsRow); } function quoteLifecycleBucketRankSql(expression) { @@ -6071,26 +6291,6 @@ function floorTimestamp(value, granularityMs) { return new Date(Math.floor(parsed / granularity) * granularity).toISOString(); } -function earliestTimestamp(values = []) { - let earliest = null; - for (const value of values || []) { - const iso = toIsoTimestamp(value); - if (!iso) continue; - if (!earliest || timestampValue(iso) < timestampValue(earliest)) earliest = iso; - } - return earliest; -} - -function latestTimestamp(values = []) { - let latest = null; - for (const value of values || []) { - const iso = toIsoTimestamp(value); - if (!iso) continue; - if (!latest || timestampValue(iso) > timestampValue(latest)) latest = iso; - } - return latest; -} - function laterTimestamp(left, right) { const leftMs = timestampValue(left); const rightMs = timestampValue(right); diff --git a/src/operator-dashboard/static/pages/StrategyPage.jsx b/src/operator-dashboard/static/pages/StrategyPage.jsx index 3e44353..bbab742 100644 --- a/src/operator-dashboard/static/pages/StrategyPage.jsx +++ b/src/operator-dashboard/static/pages/StrategyPage.jsx @@ -68,6 +68,12 @@ const QUOTE_LIFECYCLE_STAT_RANGE_DURATIONS_MS = { seven_days: 7 * 24 * 60 * 60 * 1000, thirty_days: 30 * 24 * 60 * 60 * 1000, }; +const QUOTE_LIFECYCLE_STAT_GRAIN_DURATIONS_MS = { + five_minute: 5 * 60 * 1000, + hour: 60 * 60 * 1000, + day: 24 * 60 * 60 * 1000, + week: 7 * 24 * 60 * 60 * 1000, +}; const QUOTE_LIFECYCLE_STAT_BUCKETS = [ ['success', 'Success'], ['submitted_no_reply', 'Submitted no reply'], @@ -328,6 +334,74 @@ function statisticRowsForRange(rows, grain, range, now = Date.now()) { return rangedRows.length ? rangedRows : grainRows.slice(-1); } +function statisticRowsForChart(rows, grain, range, now = Date.now()) { + return fillMissingStatisticWindows(statisticRowsForRange(rows, grain, range, now), { + grain, + range, + now, + }); +} + +function fillMissingStatisticWindows(rows, { + grain, + range, + now = Date.now(), +} = {}) { + if (!rows.length || grain === 'all_time') return rows; + const stepMs = QUOTE_LIFECYCLE_STAT_GRAIN_DURATIONS_MS[grain]; + if (!stepMs) return rows; + const durationMs = QUOTE_LIFECYCLE_STAT_RANGE_DURATIONS_MS[range]; + const firstStart = statisticRowWindowStart(rows[0]); + const lastStart = statisticRowWindowStart(rows.at(-1)); + if (!Number.isFinite(firstStart) || !Number.isFinite(lastStart)) return rows; + const start = durationMs + ? Math.floor((now - durationMs) / stepMs) * stepMs + : Math.floor(firstStart / stepMs) * stepMs; + const end = durationMs + ? Math.floor(now / stepMs) * stepMs + : Math.floor(lastStart / stepMs) * stepMs; + const steps = Math.floor((end - start) / stepMs) + 1; + if (!Number.isFinite(steps) || steps <= 0 || steps > 600) return rows; + + const rowsByStart = new Map(rows.map((row) => [statisticRowWindowStart(row), row])); + const computedAt = rows.at(-1)?.computed_at || rows[0]?.computed_at || null; + const filledRows = []; + for (let at = start; at <= end; at += stepMs) { + filledRows.push(rowsByStart.get(at) || emptyStatisticWindowRow({ + grain, + windowStartMs: at, + windowEndMs: at + stepMs, + computedAt, + })); + } + return filledRows; +} + +function statisticRowWindowStart(row) { + const start = Date.parse(row?.window_start || ''); + return Number.isFinite(start) ? start : null; +} + +function emptyStatisticWindowRow({ + grain, + windowStartMs, + windowEndMs, + computedAt, +}) { + const windowStart = new Date(windowStartMs).toISOString(); + return { + stat_id: `quote-lifecycle-stat:${grain}:${windowStart}:empty`, + grain, + window_start: windowStart, + window_end: new Date(windowEndMs).toISOString(), + quote_count: 0, + missing_identifier_count: 0, + missing_timestamp_count: 0, + bucket_counts: Object.fromEntries(QUOTE_LIFECYCLE_STAT_BUCKETS.map(([bucket]) => [bucket, 0])), + computed_at: computedAt, + }; +} + function formatStatisticChartLabel(row) { if (!row) return 'Unavailable'; if (row.grain === 'all_time') return 'All-time'; @@ -1151,7 +1225,7 @@ function QuoteLifecycleStatisticsPanel({ const liveCurrent = latestStatisticRow(liveRows, grain); const persistedGeneratedAt = persistedRows?.[0]?.computed_at || null; const liveGeneratedAt = liveRows?.[0]?.computed_at || null; - const chartRows = statisticRowsForRange(persistedRows, grain, range); + const chartRows = statisticRowsForChart(persistedRows, grain, range); const rows = fixedRows(selectedRows, grain === 'all_time' ? 1 : 8); return ( diff --git a/test/operator-dashboard-ui-static.test.mjs b/test/operator-dashboard-ui-static.test.mjs index 5cf964f..ae50099 100644 --- a/test/operator-dashboard-ui-static.test.mjs +++ b/test/operator-dashboard-ui-static.test.mjs @@ -56,6 +56,10 @@ test('strategy page owns consolidated quote lifecycle and successful trade table assert.match(strategySource, /Refresh stats/); assert.match(strategySource, /Chart range/); assert.match(strategySource, /QuoteLifecycleStatisticsChart/); + assert.match(strategySource, /statisticRowsForChart/); + assert.match(strategySource, /fillMissingStatisticWindows/); + assert.match(strategySource, /quote_count: 0/); + assert.match(strategySource, /steps > 600/); assert.match(strategySource, /echarts\/core/); assert.match(strategySource, /BarChart/); assert.match(strategySource, /LineChart/); diff --git a/test/postgres-quote-lifecycle-statistics.test.mjs b/test/postgres-quote-lifecycle-statistics.test.mjs index 3a5ca00..b63a403 100644 --- a/test/postgres-quote-lifecycle-statistics.test.mjs +++ b/test/postgres-quote-lifecycle-statistics.test.mjs @@ -1,6 +1,9 @@ import test from 'node:test'; import assert from 'node:assert/strict'; +import { + buildQuoteLifecycleStatistics, +} from '../src/core/quote-lifecycle-statistics.mjs'; import { loadQuoteLifecycleStatistics, refreshQuoteLifecycleStatistics, @@ -9,101 +12,35 @@ import { test('postgres quote lifecycle statistics refresh idempotently upserts state buckets by original quote window', async () => { const upserts = []; const subjects = new Map(); + const issuedSql = []; let includeFailure = false; const pool = { async query(sql, params = []) { - if (sql.includes('FROM swap_demand_events') && sql.includes('ORDER BY COALESCE')) { - return { - rows: [{ - observed_at: '2026-06-12T17:31:01.000Z', - ingested_at: '2026-06-12T17:31:01.000Z', - payload: { - quote_id: 'quote-stat', - pair: 'btc->eure', - asset_in: 'btc', - asset_out: 'eure', - amount_in: '1', - amount_out: '2', - }, - }], - }; - } - if (sql.includes('FROM trade_decisions') && sql.includes('ORDER BY COALESCE')) { - return { - rows: [{ - observed_at: '2026-06-12T17:31:02.000Z', - ingested_at: '2026-06-12T17:31:02.000Z', - payload: { - quote_id: 'quote-stat', - decision_id: 'decision-stat', - pair: 'btc->eure', - decision: 'approved', - decision_reason: 'strategy_approved', - }, - }], - }; - } - if (sql.includes('FROM execute_trade_commands') && sql.includes('ORDER BY COALESCE')) { - return { - rows: [{ - observed_at: '2026-06-12T17:31:03.000Z', - ingested_at: '2026-06-12T17:31:03.000Z', - payload: { - quote_id: 'quote-stat', - decision_id: 'decision-stat', - command_id: 'cmd-stat', - pair: 'btc->eure', - }, - }], - }; - } - if (sql.includes('FROM trade_execution_results r')) { - return { - rows: includeFailure ? [{ - result_observed_at: '2026-06-12T17:36:10.000Z', - result_ingested_at: '2026-06-12T17:36:10.000Z', - result_payload: { - quote_id: 'quote-stat', - decision_id: 'decision-stat', - command_id: 'cmd-stat', - status: 'failed', - result_code: 'submission_failed', - failure_category: 'quote_not_found_or_finished', - }, - command_ingested_at: '2026-06-12T17:31:03.000Z', - command_payload: { - quote_id: 'quote-stat', - decision_id: 'decision-stat', - command_id: 'cmd-stat', - }, - decision_payload: { - quote_id: 'quote-stat', - decision_id: 'decision-stat', - decision: 'approved', - }, - outcome_payload: null, - }] : [], - }; - } - if (sql.includes('FROM quote_outcome_attributions') && sql.includes('ORDER BY COALESCE')) { - return { rows: [] }; - } + issuedSql.push(sql); if (sql.includes('INSERT INTO quote_lifecycle_stat_subjects')) { - const existing = subjects.get(params[0]); - subjects.set(params[0], { - quote_id: params[0], - first_activity_at: existing?.first_activity_at || params[1], - latest_stage_at: params[2] || existing?.latest_stage_at || null, - bucket: params[3], - lifecycle_state: params[4], - reason_code: params[5], - updated_at: params[6], - payload: JSON.parse(params[7]), + subjects.set('quote-stat', { + quote_id: 'quote-stat', + quote_activity_at: '2026-06-12T17:31:01.000Z', + latest_stage_at: includeFailure + ? '2026-06-12T17:36:10.000Z' + : '2026-06-12T17:31:03.000Z', + statistics_bucket: includeFailure ? 'submission_failed' : 'awaiting_executor', + lifecycle_state: includeFailure ? 'failed' : 'command_emitted', + reason_code: includeFailure ? 'quote_not_found_or_finished' : 'awaiting_executor', }); return { rows: [], rowCount: 1 }; } - if (sql.includes('FROM quote_lifecycle_stat_subjects')) { - return { rows: [...subjects.values()] }; + if (sql.includes('SELECT COUNT(*)::INT AS subject_count')) { + return { rows: [{ subject_count: subjects.size }] }; + } + if (sql.includes('WITH subjects AS')) { + return { + rows: buildQuoteLifecycleStatistics({ + lifecycleRows: [...subjects.values()], + grains: params[0], + computedAt: params[1], + }), + }; } if (sql.includes('INSERT INTO quote_lifecycle_statistics')) { upserts.push({ sql, params }); @@ -137,6 +74,10 @@ test('postgres quote lifecycle statistics refresh idempotently upserts state buc )), false, ); + assert.equal( + issuedSql.some((sql) => sql.includes('ORDER BY COALESCE') && sql.includes('FROM swap_demand_events')), + false, + ); }); test('postgres quote lifecycle statistics retain per-quote buckets after detail pruning', async () => { @@ -145,97 +86,30 @@ test('postgres quote lifecycle statistics retain per-quote buckets after detail let sourceAvailable = true; const pool = { async query(sql, params = []) { - if (sql.includes('FROM swap_demand_events') && sql.includes('ORDER BY COALESCE')) { - return { - rows: sourceAvailable ? [{ - observed_at: '2026-06-12T14:11:01.000Z', - ingested_at: '2026-06-12T14:11:01.000Z', - payload: { - quote_id: 'quote-completed', - pair: 'btc->usdc', - asset_in: 'btc', - asset_out: 'usdc', - amount_in: '1', - amount_out: '2', - }, - }] : [], - }; - } - if (sql.includes('FROM trade_decisions') && sql.includes('ORDER BY COALESCE')) { - return { - rows: sourceAvailable ? [{ - observed_at: '2026-06-12T14:11:02.000Z', - ingested_at: '2026-06-12T14:11:02.000Z', - payload: { - quote_id: 'quote-completed', - decision_id: 'decision-completed', - pair: 'btc->usdc', - decision: 'approved', - decision_reason: 'strategy_approved', - }, - }] : [], - }; - } - if (sql.includes('FROM execute_trade_commands') && sql.includes('ORDER BY COALESCE')) { - return { - rows: sourceAvailable ? [{ - observed_at: '2026-06-12T14:11:03.000Z', - ingested_at: '2026-06-12T14:11:03.000Z', - payload: { - quote_id: 'quote-completed', - decision_id: 'decision-completed', - command_id: 'cmd-completed', - pair: 'btc->usdc', - }, - }] : [], - }; - } - if (sql.includes('FROM trade_execution_results r')) { - return { - rows: sourceAvailable ? [{ - result_observed_at: '2026-06-12T14:11:10.000Z', - result_ingested_at: '2026-06-12T14:11:10.000Z', - result_payload: { - quote_id: 'quote-completed', - decision_id: 'decision-completed', - command_id: 'cmd-completed', - status: 'submitted', - result_code: 'quote_response_ok', - outcome_status: 'completed', - }, - command_payload: { - quote_id: 'quote-completed', - decision_id: 'decision-completed', - command_id: 'cmd-completed', - }, - decision_payload: { - quote_id: 'quote-completed', - decision_id: 'decision-completed', - decision: 'approved', - }, - outcome_payload: null, - }] : [], - }; - } - if (sql.includes('FROM quote_outcome_attributions') && sql.includes('ORDER BY COALESCE')) { - return { rows: [] }; - } if (sql.includes('INSERT INTO quote_lifecycle_stat_subjects')) { - const existing = subjects.get(params[0]); - subjects.set(params[0], { - quote_id: params[0], - first_activity_at: existing?.first_activity_at || params[1], - latest_stage_at: params[2] || existing?.latest_stage_at || null, - bucket: params[3], - lifecycle_state: params[4], - reason_code: params[5], - updated_at: params[6], - payload: JSON.parse(params[7]), - }); - return { rows: [], rowCount: 1 }; + if (sourceAvailable) { + subjects.set('quote-completed', { + quote_id: 'quote-completed', + quote_activity_at: '2026-06-12T14:11:01.000Z', + latest_stage_at: '2026-06-12T14:11:10.000Z', + statistics_bucket: 'success', + lifecycle_state: 'completed', + reason_code: 'completed', + }); + } + return { rows: [], rowCount: sourceAvailable ? 1 : 0 }; } - if (sql.includes('FROM quote_lifecycle_stat_subjects')) { - return { rows: [...subjects.values()] }; + if (sql.includes('SELECT COUNT(*)::INT AS subject_count')) { + return { rows: [{ subject_count: subjects.size }] }; + } + if (sql.includes('WITH subjects AS')) { + return { + rows: buildQuoteLifecycleStatistics({ + lifecycleRows: [...subjects.values()], + grains: params[0], + computedAt: params[1], + }), + }; } if (sql.includes('INSERT INTO quote_lifecycle_statistics')) { if (params[1] === 'all_time') allTimeUpserts.push({ params });