philipp/unrip
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Make quote statistics refresh set based
All checks were successful
deploy / deploy (push) Successful in 1m5s
Details

Browse source 
Proof: replaced Node-side full lifecycle statistics refresh with PostgreSQL set-based subject upsert and aggregation, added bounded zero-window chart filling so missing 5m/hour windows are visible, and verified with targeted stats/UI tests, full npm test, and operator dashboard bundle build.

Assumptions: persisted quote statistics represent retained unique lifecycle subjects, not every raw upstream quote message; raw firehose volume remains separately retained and can exceed unique lifecycle counts by orders of magnitude.

Still fake: this does not repair stale upstream market/inventory ingestion or reconstruct already-pruned raw quote detail; fee-aware PnL and venue-native fill truth remain out of scope.
This commit is contained in:
philipp 2026-06-14 12:23:43 +02:00
parent 503a69e71c
commit 9d97ed7d39
4 changed files with 552 additions and 400 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

540
src/lib/postgres.mjs
View file

@ -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,29 +3217,227 @@ 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(
const result = await pool.query(
`
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,
@ -3267,7 +3447,25 @@ async function upsertQuoteLifecycleStatisticSubjects(pool, {
reason_code,
updated_at,
payload
) VALUES ($1,$2,$3,$4,$5,$6,$7,$8::jsonb)
)
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
@ -3317,134 +3515,156 @@ async function upsertQuoteLifecycleStatisticSubjects(pool, {
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),
],
[now],
);
upsertedCount += 1;
}
return { upserted_count: upsertedCount };
const countResult = await pool.query(`
SELECT COUNT(*)::INT AS subject_count
FROM ${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE}
`);
return {
upserted_count: Number(result.rowCount || 0),
statistic_subjects: Number(countResult.rows[0]?.subject_count || 0),
};
}
async function loadQuoteLifecycleStatisticSubjects(pool) {
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
quote_id,
first_activity_at,
latest_stage_at,
bucket,
lifecycle_state,
reason_code,
updated_at,
payload
COALESCE(bucket, payload->>'statistics_bucket', 'unavailable') AS bucket
FROM ${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE}
ORDER BY first_activity_at ASC NULLS LAST, quote_id ASC
`,
);
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);
return {
quote_id: quoteId,
first_activity_at: firstActivityAt,
latest_stage_at: latestStageAt,
), grain_rows AS (
SELECT
'all_time'::text AS grain,
NULL::timestamptz AS window_start,
NULL::timestamptz AS window_end,
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,
},
};
}
first_activity_at IS NOT NULL AS has_timestamp
FROM subjects
WHERE 'all_time' = ANY($1::text[])
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',
};
}
UNION ALL
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,
]);
}
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 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,
]);
}
UNION ALL
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';
}
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);

76
src/operator-dashboard/static/pages/StrategyPage.jsx
View file

@ -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 (

4
test/operator-dashboard-ui-static.test.mjs
View file

@ -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/);

226
test/postgres-quote-lifecycle-statistics.test.mjs
View file

@ -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]),
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: 1 };
}
if (sql.includes('FROM quote_lifecycle_stat_subjects')) {
return { rows: [...subjects.values()] };
return { rows: [], rowCount: sourceAvailable ? 1 : 0 };
}
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 });