diff --git a/src/apps/history-writer.mjs b/src/apps/history-writer.mjs index 1863c9d..cef3f64 100644 --- a/src/apps/history-writer.mjs +++ b/src/apps/history-writer.mjs @@ -28,6 +28,7 @@ import { loadLatestPortfolioMetric, loadPortfolioMetricInputs, refreshIntentRequestOutcomes, + refreshQuoteLifecycleStatistics, refreshQuoteOutcomes, runQuoteLifecycleRetentionMaintenance, seedTradingConfig, @@ -202,6 +203,10 @@ const state = { last_quote_lifecycle_rollup_at: null, quote_lifecycle_rollup_count: 0, quote_lifecycle_rollup_error: null, + last_quote_lifecycle_statistics_at: null, + quote_lifecycle_statistics_count: 0, + quote_lifecycle_statistics_error: null, + latest_quote_lifecycle_statistics: null, last_retention_maintenance_at: null, retention_maintenance: null, retention_summary: null, @@ -233,6 +238,9 @@ await refreshIntentRequestOutcomeAttributions().then((records) => publishIntentR await refreshQuoteLifecycleRetentionState().catch((error) => { state.retention_error = serializeError(error); }); +await refreshQuoteLifecycleStatisticsState({ persist: true }).catch((error) => { + state.quote_lifecycle_statistics_error = serializeError(error); +}); let quoteLifecycleRetentionMaintenanceInFlight = null; const quoteLifecycleRetentionMaintenanceTimer = setInterval(() => { @@ -340,9 +348,15 @@ async function maybeRunQuoteLifecycleRetentionMaintenance({ force = false } = {} quoteLifecycleRetentionMaintenanceInFlight = (async () => { try { - const result = await runQuoteLifecycleRetentionMaintenance(pool, { - now: new Date(nowMs).toISOString(), + const nowIso = new Date(nowMs).toISOString(); + const statisticsResult = await refreshQuoteLifecycleStatisticsState({ + now: nowIso, + persist: true, }); + const result = await runQuoteLifecycleRetentionMaintenance(pool, { + now: nowIso, + }); + result.quote_lifecycle_statistics = statisticsResult; applyRetentionMaintenanceResult(result); await refreshQuoteLifecycleRetentionState(); if (result.status === 'success' && Number(result.pruned_counts?.deletedCount || 0) > 0) { @@ -419,6 +433,33 @@ async function refreshQuoteLifecycleRetentionState() { return summary; } +async function refreshQuoteLifecycleStatisticsState({ + now = new Date().toISOString(), + persist = false, +} = {}) { + let result = null; + try { + result = persist + ? await refreshQuoteLifecycleStatistics(pool, { now }) + : null; + } catch (error) { + state.quote_lifecycle_statistics_error = serializeError(error); + throw error; + } + if (!result) return state.latest_quote_lifecycle_statistics; + + state.last_quote_lifecycle_statistics_at = result.computed_at; + state.quote_lifecycle_statistics_count = result.statistic_row_count; + state.quote_lifecycle_statistics_error = null; + state.latest_quote_lifecycle_statistics = { + computed_at: result.computed_at, + quote_count: result.quote_count, + statistic_row_count: result.statistic_row_count, + evidence_counts: result.evidence_counts, + }; + return state.latest_quote_lifecycle_statistics; +} + async function handleWrittenHistoryEvent({ topic, partition, @@ -560,6 +601,10 @@ const controlApi = startControlApi({ last_retention_maintenance_at: state.last_retention_maintenance_at, retention_error: state.retention_error, storage_pressure: state.storage_pressure, + last_quote_lifecycle_statistics_at: state.last_quote_lifecycle_statistics_at, + quote_lifecycle_statistics_count: state.quote_lifecycle_statistics_count, + quote_lifecycle_statistics_error: state.quote_lifecycle_statistics_error, + latest_quote_lifecycle_statistics: state.latest_quote_lifecycle_statistics, last_quote_lifecycle_rollup_at: state.last_quote_lifecycle_rollup_at, quote_lifecycle_rollup_count: state.quote_lifecycle_rollup_count, quote_lifecycle_rollup_error: state.quote_lifecycle_rollup_error, diff --git a/src/apps/operator-dashboard.mjs b/src/apps/operator-dashboard.mjs index ead8fac..705b625 100644 --- a/src/apps/operator-dashboard.mjs +++ b/src/apps/operator-dashboard.mjs @@ -13,6 +13,7 @@ import { applyDashboardLiveEvent, buildDashboardBootstrap, buildDashboardControlErrorResponse, + buildLiveQuoteLifecycleStatistics, buildLiveQuoteLifecycleRows, buildQuoteLifecycleLookupResponse, buildLiveStatusBar, @@ -47,6 +48,7 @@ import { loadPairConfigSummary, loadQuoteLifecycleLookupEvidence, loadQuoteLifecycleRetentionSummary, + loadQuoteLifecycleStatistics, loadRecentAlertTransitions, loadRecentDepositStatuses, loadRecentEnvironmentStatuses, @@ -60,6 +62,7 @@ import { loadSubmissionPage, loadSubmissionSummary, pauseTradingPair, + refreshQuoteLifecycleStatistics, seedTradingConfig, setTradingPairMode, } from '../lib/postgres.mjs'; @@ -231,6 +234,7 @@ webSocketServer.on('connection', (socket, _req, authContext) => { webSockets.add(socket); dashboardRuntimeState.websocket_clients = webSockets.size; const recentLifecycleRows = buildLiveQuoteLifecycleRows(liveState); + const quoteLifecycleStatistics = buildLiveQuoteLifecycleStatistics(liveState); dashboardRuntimeState.latest_maker_competitiveness = buildMakerCompetitivenessSummary({ lifecycleRows: recentLifecycleRows, }); @@ -240,6 +244,7 @@ webSocketServer.on('connection', (socket, _req, authContext) => { live: { recent_quotes: liveState.recent_quotes, recent_lifecycle_rows: recentLifecycleRows, + quote_lifecycle_statistics: quoteLifecycleStatistics, maker_competitiveness: dashboardRuntimeState.latest_maker_competitiveness, status_bar: buildLiveStatusBar(liveState), }, @@ -380,6 +385,14 @@ async function handleApiRequest({ req, res, url, auth }) { return sendJson(res, 200, submissionPage); } + if (req.method === 'GET' && url.pathname === '/api/strategy/quote-lifecycle/statistics') { + const payload = await loadQuoteLifecycleStatisticsPayload({ + grains: url.searchParams.get('grains') || undefined, + limit: Number(url.searchParams.get('limit') || 120), + }); + return sendJson(res, 200, payload); + } + const lifecycleLookupMatch = req.method === 'GET' ? url.pathname.match(/^\/api\/strategy\/quote-lifecycle\/(.+)$/) : null; @@ -484,6 +497,7 @@ async function loadBootstrapPayload({ auth, page, pageSize }) { recentQuoteOutcomes, quoteLifecycleRetention, quoteLifecycleRollups, + quoteLifecycleStatistics, recentIntentRequests, recentAlertTransitions, recentEnvironmentStatuses, @@ -567,6 +581,12 @@ async function loadBootstrapPayload({ auth, page, pageSize }) { [], sourceErrors, ), + safeSourceLoad( + 'quote_lifecycle_statistics', + () => loadQuoteLifecycleStatistics(pool, { limit: 120 }), + [], + sourceErrors, + ), safeSourceLoad( 'recent_intent_requests', () => loadRecentIntentRequests(pool, { @@ -613,6 +633,14 @@ async function loadBootstrapPayload({ auth, page, pageSize }) { recentQuoteOutcomes, quoteLifecycleRetention, quoteLifecycleRollups, + quoteLifecycleStatistics: { + persisted: { + source: 'persisted', + generated_at: quoteLifecycleStatistics[0]?.computed_at || null, + rows: quoteLifecycleStatistics, + }, + live: buildLiveQuoteLifecycleStatistics(liveState), + }, recentIntentRequests, recentAlertTransitions, recentEnvironmentStatuses, @@ -644,6 +672,32 @@ async function loadQuoteLifecycleLookupPayload({ identifier }) { }); } +async function loadQuoteLifecycleStatisticsPayload({ grains, limit } = {}) { + const refreshed = await refreshQuoteLifecycleStatistics(pool, { + now: new Date().toISOString(), + grains, + }); + const rows = await loadQuoteLifecycleStatistics(pool, { + grains, + limit, + }); + return { + ok: true, + refreshed_at: refreshed.computed_at, + persisted: { + source: 'persisted', + generated_at: rows[0]?.computed_at || refreshed.computed_at, + rows, + }, + live: buildLiveQuoteLifecycleStatistics(liveState), + refresh: { + statistic_row_count: refreshed.statistic_row_count, + quote_count: refreshed.quote_count, + evidence_counts: refreshed.evidence_counts, + }, + }; +} + async function loadServiceSnapshots() { const services = listDashboardServices(config); return Promise.all(services.map((service) => loadServiceSnapshot(service))); diff --git a/src/core/operator-dashboard.mjs b/src/core/operator-dashboard.mjs index f12c61f..29227af 100644 --- a/src/core/operator-dashboard.mjs +++ b/src/core/operator-dashboard.mjs @@ -3,6 +3,7 @@ import { bridgeDepositObservedAt } from './bridge-assets.mjs'; import { summarizeFundingObservations } from './funding-observations.mjs'; import { buildMakerCompetitivenessSummary } from './maker-competitiveness.mjs'; import { resolveDashboardRequestAuth } from './operator-dashboard-auth.mjs'; +import { buildQuoteLifecycleStatistics } from './quote-lifecycle-statistics.mjs'; import { TERMINAL_SETTLEMENT_ATTRIBUTION_STATUSES } from './quote-outcomes.mjs'; import { inferServiceFreshnessTimestamp as inferRuntimeFreshnessTimestamp } from './runtime-health.mjs'; @@ -417,6 +418,27 @@ export function buildLiveQuoteLifecycleRows(state, { flashQuoteId = null, flashA )); } +export function buildLiveQuoteLifecycleStatistics(state, { + computedAt = new Date().toISOString(), +} = {}) { + const lifecycleRows = deriveQuoteLifecycleRows({ + recentQuotes: state.recent_quotes, + recentTradeDecisions: state.recent_trade_decisions, + recentExecuteTradeCommands: state.recent_execute_trade_commands, + recentExecutionResults: state.recent_execution_results, + recentQuoteOutcomes: state.recent_quote_outcomes, + limit: null, + }).map((row) => enrichLifecycleRowForUi({ config: state.config, row })); + return { + source: 'live_recent', + generated_at: computedAt, + rows: buildQuoteLifecycleStatistics({ + lifecycleRows, + computedAt, + }), + }; +} + export function applyDashboardLiveEvent(state, { topic, event }) { if (!event?.payload) return []; @@ -538,6 +560,7 @@ export function buildDashboardBootstrap({ recentQuoteOutcomes = [], quoteLifecycleRetention = null, quoteLifecycleRollups = [], + quoteLifecycleStatistics = null, recentIntentRequests = [], recentAlertTransitions, recentEnvironmentStatuses = [], @@ -633,6 +656,7 @@ export function buildDashboardBootstrap({ recentExecutionResults, recentQuoteOutcomes, quoteLifecycleRollups, + quoteLifecycleStatistics, }), system: buildSystemSummary({ servicesByName, @@ -1728,6 +1752,7 @@ function buildStrategySummary({ recentExecutionResults = [], recentQuoteOutcomes = [], quoteLifecycleRollups = [], + quoteLifecycleStatistics = null, }) { const strategyState = servicesByName['strategy-engine']?.state || {}; const executorState = servicesByName['trade-executor']?.state || {}; @@ -1796,6 +1821,18 @@ function buildStrategySummary({ ? recentDecisions : [...durableDecisionsById.values()].slice(0, 20), recent_lifecycle_rows: lifecycleRows, + quote_lifecycle_statistics: quoteLifecycleStatistics || { + persisted: { + source: 'persisted', + generated_at: null, + rows: [], + }, + live: { + source: 'live_recent', + generated_at: null, + rows: [], + }, + }, trade_funnel: tradeFunnel, maker_competitiveness: makerCompetitiveness, skipped_counts: strategyState.skipped_counts || {}, @@ -2051,6 +2088,18 @@ function buildSystemSummary({ || historyWriterState.retention_summary || null, quote_lifecycle_rollups: quoteLifecycleRollups || [], + quote_lifecycle_statistics: + historyWriterState.latest_quote_lifecycle_statistics + || null, + last_quote_lifecycle_statistics_at: + historyWriterState.last_quote_lifecycle_statistics_at + || null, + quote_lifecycle_statistics_count: + historyWriterState.quote_lifecycle_statistics_count + ?? null, + quote_lifecycle_statistics_error: + historyWriterState.quote_lifecycle_statistics_error + || null, last_retention_maintenance_at: quoteLifecycleRetention?.latest_run?.completed_at || historyWriterState.last_retention_maintenance_at @@ -2687,6 +2736,14 @@ function buildQuoteLifecycleUpdate(state, { flashQuoteId = null } = {}) { type: 'quote_lifecycle.updated', recent_lifecycle_rows: lifecycleRows, maker_competitiveness: buildMakerCompetitivenessSummary({ lifecycleRows, generatedAt: receivedAt }), + quote_lifecycle_statistics: { + source: 'live_recent', + generated_at: receivedAt, + rows: buildQuoteLifecycleStatistics({ + lifecycleRows, + computedAt: receivedAt, + }), + }, flash_quote_id: flashQuoteId || null, received_at: receivedAt, }; diff --git a/src/core/quote-lifecycle-statistics.mjs b/src/core/quote-lifecycle-statistics.mjs new file mode 100644 index 0000000..3de2de5 --- /dev/null +++ b/src/core/quote-lifecycle-statistics.mjs @@ -0,0 +1,474 @@ +export const QUOTE_LIFECYCLE_STATISTIC_GRAINS = Object.freeze([ + 'all_time', + 'month', + 'week', + 'day', + 'hour', + 'five_minute', +]); + +export const QUOTE_LIFECYCLE_STATISTIC_BUCKETS = Object.freeze([ + 'success', + 'not_filled', + 'submission_failed', + 'submitted_no_reply', + 'awaiting_executor', + 'approved_no_command', + 'blocked_before_submit', + 'rejected_by_strategy', + 'observed_no_decision', + 'unavailable', +]); + +const BUCKET_RANK = Object.freeze({ + unavailable: 0, + observed_no_decision: 1, + approved_no_command: 2, + awaiting_executor: 3, + rejected_by_strategy: 4, + blocked_before_submit: 4, + submitted_no_reply: 5, + submission_failed: 6, + not_filled: 7, + success: 8, +}); + +export function normalizeQuoteLifecycleStatisticsGrains(grains = QUOTE_LIFECYCLE_STATISTIC_GRAINS) { + const requested = Array.isArray(grains) ? grains : String(grains || '').split(','); + const normalized = requested + .map((grain) => String(grain || '').trim()) + .filter((grain) => QUOTE_LIFECYCLE_STATISTIC_GRAINS.includes(grain)); + return normalized.length ? [...new Set(normalized)] : [...QUOTE_LIFECYCLE_STATISTIC_GRAINS]; +} + +export function emptyQuoteLifecycleBucketCounts() { + return Object.fromEntries(QUOTE_LIFECYCLE_STATISTIC_BUCKETS.map((bucket) => [bucket, 0])); +} + +export function classifyQuoteLifecycleStatisticsBucket(row = null) { + if (!row) return 'unavailable'; + + const lifecycleState = normalizeToken(row.lifecycle_state); + const executionStatus = normalizeToken(row.execution?.status || row.stage_details?.execution_status); + const outcomeStatus = normalizeToken( + row.outcome_status + || row.outcome?.outcome_status + || row.execution?.outcome_status + || row.stage_details?.durable_outcome_status + || row.stage_details?.execution_outcome_status, + ); + const reasonCode = normalizeToken( + row.reason_code + || row.execution?.failure_category + || row.execution?.result_code + || row.outcome?.outcome_reason, + ); + + if ( + lifecycleState === 'completed' + || ['completed', 'filled', 'settled', 'finalized'].includes(outcomeStatus) + ) { + return 'success'; + } + + if ( + lifecycleState === 'not_filled' + || ['not_filled', 'expired', 'cancelled'].includes(outcomeStatus) + ) { + return 'not_filled'; + } + + if ( + lifecycleState === 'failed' + || executionStatus === 'failed' + || reasonCode === 'submission_failed' + || reasonCode === 'quote_not_found_or_finished' + ) { + return 'submission_failed'; + } + + if ( + lifecycleState === 'submitted' + || lifecycleState === 'awaiting_outcome' + || executionStatus === 'submitted' + || outcomeStatus === 'awaiting_outcome' + ) { + return 'submitted_no_reply'; + } + + if (lifecycleState === 'command_emitted') return 'awaiting_executor'; + if (lifecycleState === 'evaluated') return 'approved_no_command'; + if (lifecycleState === 'blocked') return 'blocked_before_submit'; + if (lifecycleState === 'rejected') return 'rejected_by_strategy'; + if (lifecycleState === 'observed') return 'observed_no_decision'; + return 'unavailable'; +} + +export function quoteLifecycleStatisticWindow(timestamp, grain) { + if (grain === 'all_time') { + return { + grain, + window_start: null, + window_end: null, + }; + } + + const date = new Date(timestamp || ''); + if (Number.isNaN(date.getTime())) return null; + + const start = new Date(date.getTime()); + switch (grain) { + case 'five_minute': + start.setUTCSeconds(0, 0); + start.setUTCMinutes(Math.floor(start.getUTCMinutes() / 5) * 5); + break; + case 'hour': + start.setUTCMinutes(0, 0, 0); + break; + case 'day': + start.setUTCHours(0, 0, 0, 0); + break; + case 'week': { + start.setUTCHours(0, 0, 0, 0); + const daysSinceMonday = (start.getUTCDay() + 6) % 7; + start.setUTCDate(start.getUTCDate() - daysSinceMonday); + break; + } + case 'month': + start.setUTCDate(1); + start.setUTCHours(0, 0, 0, 0); + break; + default: + return null; + } + + const end = new Date(start.getTime()); + switch (grain) { + case 'five_minute': + end.setUTCMinutes(end.getUTCMinutes() + 5); + break; + case 'hour': + end.setUTCHours(end.getUTCHours() + 1); + break; + case 'day': + end.setUTCDate(end.getUTCDate() + 1); + break; + case 'week': + end.setUTCDate(end.getUTCDate() + 7); + break; + case 'month': + end.setUTCMonth(end.getUTCMonth() + 1); + break; + default: + return null; + } + + return { + grain, + window_start: start.toISOString(), + window_end: end.toISOString(), + }; +} + +export function buildQuoteLifecycleStatistics({ + lifecycleRows = [], + grains = QUOTE_LIFECYCLE_STATISTIC_GRAINS, + computedAt = new Date().toISOString(), +} = {}) { + const normalizedGrains = normalizeQuoteLifecycleStatisticsGrains(grains); + const groups = new Map(); + const deduped = dedupeLifecycleRows(lifecycleRows); + + if (normalizedGrains.includes('all_time')) { + ensureStatisticGroup(groups, { + grain: 'all_time', + window_start: null, + window_end: null, + computedAt, + }); + } + + for (const entry of deduped.identified) { + const row = entry.row; + const bucket = classifyQuoteLifecycleStatisticsBucket(row); + const activityAt = entry.first_activity_at; + addRowToStatisticGroup(groups, { + grain: 'all_time', + window_start: null, + window_end: null, + bucket, + computedAt, + hasQuoteIdentifier: true, + hasTimestamp: Boolean(activityAt), + enabled: normalizedGrains.includes('all_time'), + }); + + if (!activityAt) continue; + for (const grain of normalizedGrains.filter((candidate) => candidate !== 'all_time')) { + const window = quoteLifecycleStatisticWindow(activityAt, grain); + if (!window) continue; + addRowToStatisticGroup(groups, { + ...window, + bucket, + computedAt, + hasQuoteIdentifier: true, + hasTimestamp: true, + enabled: true, + }); + } + } + + for (const row of deduped.missingIdentifier) { + const bucket = 'unavailable'; + const activityAt = firstLifecycleTimestamp(row); + addRowToStatisticGroup(groups, { + grain: 'all_time', + window_start: null, + window_end: null, + bucket, + computedAt, + hasQuoteIdentifier: false, + hasTimestamp: Boolean(activityAt), + enabled: normalizedGrains.includes('all_time'), + }); + + if (!activityAt) continue; + for (const grain of normalizedGrains.filter((candidate) => candidate !== 'all_time')) { + const window = quoteLifecycleStatisticWindow(activityAt, grain); + if (!window) continue; + addRowToStatisticGroup(groups, { + ...window, + bucket, + computedAt, + hasQuoteIdentifier: false, + hasTimestamp: true, + enabled: true, + }); + } + } + + return [...groups.values()] + .map((group) => ({ + ...group, + stat_id: quoteLifecycleStatisticId(group), + bucket_counts: normalizeBucketCounts(group.bucket_counts), + })) + .sort(compareStatisticRows); +} + +export function normalizeQuoteLifecycleStatisticsRow(row = {}) { + return { + stat_id: row.stat_id || quoteLifecycleStatisticId(row), + grain: row.grain || 'all_time', + window_start: toIsoTimestamp(row.window_start), + window_end: toIsoTimestamp(row.window_end), + quote_count: Number(row.quote_count || 0), + missing_identifier_count: Number(row.missing_identifier_count || 0), + missing_timestamp_count: Number(row.missing_timestamp_count || 0), + bucket_counts: normalizeBucketCounts(row.bucket_counts), + computed_at: toIsoTimestamp(row.computed_at), + }; +} + +export function quoteLifecycleStatisticId(row = {}) { + const grain = row.grain || 'all_time'; + const windowStart = row.window_start || 'all'; + return `quote-lifecycle-stat:${grain}:${windowStart}`; +} + +function dedupeLifecycleRows(rows = []) { + const identifiedByQuote = new Map(); + const missingIdentifier = []; + + for (const row of rows || []) { + const quoteId = String(row?.quote_id || '').trim(); + if (!quoteId) { + missingIdentifier.push(row); + continue; + } + + const activityAt = firstLifecycleTimestamp(row); + const bucket = classifyQuoteLifecycleStatisticsBucket(row); + const existing = identifiedByQuote.get(quoteId); + if (!existing) { + identifiedByQuote.set(quoteId, { + row, + first_activity_at: activityAt, + latest_stage_at: latestLifecycleTimestamp(row), + bucket, + }); + continue; + } + + const firstActivityAt = earlierTimestamp(existing.first_activity_at, activityAt); + const latestStageAt = laterTimestamp(existing.latest_stage_at, latestLifecycleTimestamp(row)); + const existingRank = BUCKET_RANK[existing.bucket] ?? 0; + const nextRank = BUCKET_RANK[bucket] ?? 0; + const useNextRow = nextRank > existingRank + || (nextRank === existingRank && timestampValue(latestLifecycleTimestamp(row)) >= timestampValue(existing.latest_stage_at)); + identifiedByQuote.set(quoteId, { + row: useNextRow ? row : existing.row, + first_activity_at: firstActivityAt, + latest_stage_at: latestStageAt, + bucket: useNextRow ? bucket : existing.bucket, + }); + } + + return { + identified: [...identifiedByQuote.values()], + missingIdentifier, + }; +} + +function addRowToStatisticGroup(groups, { + grain, + window_start, + window_end, + bucket, + computedAt, + hasQuoteIdentifier, + hasTimestamp, + enabled, +}) { + if (!enabled) return; + const group = ensureStatisticGroup(groups, { + grain, + window_start, + window_end, + computedAt, + }); + if (hasQuoteIdentifier) { + group.quote_count += 1; + } else { + group.missing_identifier_count += 1; + } + if (!hasTimestamp) group.missing_timestamp_count += 1; + group.bucket_counts[bucket] = Number(group.bucket_counts[bucket] || 0) + 1; +} + +function ensureStatisticGroup(groups, { + grain, + window_start, + window_end, + computedAt, +}) { + const key = `${grain}:${window_start || 'all'}`; + if (!groups.has(key)) { + groups.set(key, { + stat_id: null, + grain, + window_start, + window_end, + quote_count: 0, + missing_identifier_count: 0, + missing_timestamp_count: 0, + bucket_counts: emptyQuoteLifecycleBucketCounts(), + computed_at: toIsoTimestamp(computedAt) || new Date().toISOString(), + }); + } + return groups.get(key); +} + +function compareStatisticRows(left, right) { + const grainOrder = new Map(QUOTE_LIFECYCLE_STATISTIC_GRAINS.map((grain, index) => [grain, index])); + const grainComparison = (grainOrder.get(left.grain) ?? 99) - (grainOrder.get(right.grain) ?? 99); + if (grainComparison !== 0) return grainComparison; + const windowComparison = timestampValue(right.window_start) - timestampValue(left.window_start); + return Number.isFinite(windowComparison) ? windowComparison : 0; +} + +function firstLifecycleTimestamp(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, + ]); +} + +function latestLifecycleTimestamp(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, + ]); +} + +function normalizeBucketCounts(value = {}) { + const source = typeof value === 'string' ? parseJsonObject(value) : value || {}; + return Object.fromEntries( + QUOTE_LIFECYCLE_STATISTIC_BUCKETS.map((bucket) => [bucket, Number(source[bucket] || 0)]), + ); +} + +function parseJsonObject(value) { + try { + const parsed = JSON.parse(value); + return parsed && typeof parsed === 'object' && !Array.isArray(parsed) ? parsed : {}; + } catch { + return {}; + } +} + +function normalizeToken(value) { + return String(value || '') + .trim() + .toLowerCase() + .replace(/[\s-]+/g, '_'); +} + +function timestampValue(value) { + const parsed = Date.parse(value || ''); + return Number.isFinite(parsed) ? parsed : -Infinity; +} + +function toIsoTimestamp(value) { + if (!value) return null; + const date = new Date(value); + return Number.isNaN(date.getTime()) ? null : date.toISOString(); +} + +function earliestTimestamp(values = []) { + let earliest = null; + for (const value of values) { + const iso = toIsoTimestamp(value); + if (!iso) continue; + earliest = earliestTimestampPair(earliest, iso); + } + return earliest; +} + +function latestTimestamp(values = []) { + let latest = null; + for (const value of values) { + const iso = toIsoTimestamp(value); + if (!iso) continue; + latest = laterTimestamp(latest, iso); + } + return latest; +} + +function earliestTimestampPair(left, right) { + if (!left) return toIsoTimestamp(right); + if (!right) return toIsoTimestamp(left); + return new Date(Math.min(timestampValue(left), timestampValue(right))).toISOString(); +} + +function earlierTimestamp(left, right) { + if (!left) return toIsoTimestamp(right); + if (!right) return toIsoTimestamp(left); + return new Date(Math.min(timestampValue(left), timestampValue(right))).toISOString(); +} + +function laterTimestamp(left, right) { + if (!left) return toIsoTimestamp(right); + if (!right) return toIsoTimestamp(left); + return new Date(Math.max(timestampValue(left), timestampValue(right))).toISOString(); +} diff --git a/src/lib/postgres.mjs b/src/lib/postgres.mjs index 14cd1fb..05f4890 100644 --- a/src/lib/postgres.mjs +++ b/src/lib/postgres.mjs @@ -1,8 +1,15 @@ import { Pool } from 'pg'; import { deriveIntentRequestOutcomeRecords } from '../core/intent-request-outcomes.mjs'; +import { deriveQuoteLifecycleRows } from '../core/operator-dashboard.mjs'; import { buildCashEquivalentValuationAssets } from '../core/portfolio-metrics.mjs'; import { deriveQuoteOutcomeRecords } from '../core/quote-outcomes.mjs'; +import { + QUOTE_LIFECYCLE_STATISTIC_GRAINS, + buildQuoteLifecycleStatistics, + normalizeQuoteLifecycleStatisticsGrains, + normalizeQuoteLifecycleStatisticsRow, +} from '../core/quote-lifecycle-statistics.mjs'; import { DEFAULT_QUOTE_LIFECYCLE_RETENTION_POLICY, QUOTE_LIFECYCLE_DATA_CLASSES, @@ -53,6 +60,7 @@ const RETENTION_POLICIES_TABLE = 'retention_policies'; const QUOTE_LIFECYCLE_ROLLUPS_TABLE = 'quote_lifecycle_rollups'; const QUOTE_LIFECYCLE_ROLLUP_WATERMARKS_TABLE = 'quote_lifecycle_rollup_watermarks'; const QUOTE_LIFECYCLE_RETENTION_RUNS_TABLE = 'quote_lifecycle_retention_runs'; +const QUOTE_LIFECYCLE_STATISTICS_TABLE = 'quote_lifecycle_statistics'; const QUOTE_LIFECYCLE_ROLLUP_WINDOWS_PER_PASS = 48; const SUPPORTED_ASSET_IMPORT_RUNS_TABLE = 'supported_asset_import_runs'; const TRADING_ASSETS_TABLE = 'trading_assets'; @@ -389,6 +397,28 @@ export async function ensureQuoteLifecycleRetentionSchema(pool) { ON ${QUOTE_LIFECYCLE_ROLLUPS_TABLE} (source_data_class, window_end DESC) `); + await pool.query(` + CREATE TABLE IF NOT EXISTS ${QUOTE_LIFECYCLE_STATISTICS_TABLE} ( + stat_id TEXT PRIMARY KEY, + grain TEXT NOT NULL, + window_start TIMESTAMPTZ, + window_end TIMESTAMPTZ, + quote_count INTEGER NOT NULL DEFAULT 0, + missing_identifier_count INTEGER NOT NULL DEFAULT 0, + missing_timestamp_count INTEGER NOT NULL DEFAULT 0, + bucket_counts JSONB NOT NULL DEFAULT '{}'::jsonb, + computed_at TIMESTAMPTZ NOT NULL + ) + `); + await pool.query(` + CREATE INDEX IF NOT EXISTS ${QUOTE_LIFECYCLE_STATISTICS_TABLE}_grain_window_idx + ON ${QUOTE_LIFECYCLE_STATISTICS_TABLE} (grain, window_start DESC NULLS LAST) + `); + await pool.query(` + CREATE INDEX IF NOT EXISTS ${QUOTE_LIFECYCLE_STATISTICS_TABLE}_computed_at_idx + ON ${QUOTE_LIFECYCLE_STATISTICS_TABLE} (computed_at DESC) + `); + await pool.query(` CREATE TABLE IF NOT EXISTS ${QUOTE_LIFECYCLE_ROLLUP_WATERMARKS_TABLE} ( rollup_key TEXT PRIMARY KEY, @@ -3149,6 +3179,193 @@ export async function loadRecentQuoteLifecycleRollups(pool, { limit = 50 } = {}) return result.rows.map(normalizeQuoteLifecycleRollupRow); } +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 statistics = buildQuoteLifecycleStatistics({ + lifecycleRows, + grains, + computedAt: now, + }); + await upsertQuoteLifecycleStatistics(pool, statistics); + return { + computed_at: now, + 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, + }, + statistics, + }; +} + +export async function loadQuoteLifecycleStatistics(pool, { + grains = QUOTE_LIFECYCLE_STATISTIC_GRAINS, + limit = 96, +} = {}) { + const normalizedGrains = normalizeQuoteLifecycleStatisticsGrains(grains); + const normalizedLimit = Math.max(1, Math.min(1000, Number(limit) || 96)); + const result = await pool.query( + ` + WITH ranked AS ( + SELECT + *, + ROW_NUMBER() OVER ( + PARTITION BY grain + ORDER BY window_start DESC NULLS LAST + ) AS grain_rank + FROM ${QUOTE_LIFECYCLE_STATISTICS_TABLE} + WHERE grain = ANY($1::text[]) + ) + SELECT * + FROM ranked + WHERE grain = 'all_time' + OR grain_rank <= $2 + ORDER BY + CASE 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, + window_start DESC NULLS LAST + `, + [normalizedGrains, normalizedLimit], + ); + return result.rows.map(normalizeQuoteLifecycleStatisticsRow); +} + +async function loadQuoteLifecycleStatisticsEvidence(pool) { + const [ + quotesResult, + decisionsResult, + commandsResult, + executionResultsResult, + outcomesResult, + ] = await Promise.all([ + pool.query(` + SELECT observed_at, ingested_at, payload + FROM swap_demand_events + ORDER BY COALESCE(observed_at, ingested_at) ASC + `), + pool.query(` + SELECT observed_at, ingested_at, payload + FROM trade_decisions + ORDER BY COALESCE(observed_at, ingested_at) ASC + `), + pool.query(` + SELECT observed_at, ingested_at, payload + FROM execute_trade_commands + ORDER BY COALESCE(observed_at, ingested_at) ASC + `), + pool.query(` + SELECT + r.observed_at AS result_observed_at, + r.ingested_at AS result_ingested_at, + r.payload AS result_payload, + c.ingested_at AS command_ingested_at, + c.payload AS command_payload, + d.payload AS decision_payload, + o.payload AS outcome_payload + FROM trade_execution_results r + LEFT JOIN execute_trade_commands c + ON c.decision_key = r.decision_key + LEFT JOIN trade_decisions d + ON d.decision_key = COALESCE(c.payload->>'decision_id', r.payload->>'decision_id') + LEFT JOIN ${QUOTE_OUTCOMES_TABLE} o + ON o.quote_id = r.quote_id + ORDER BY COALESCE(r.observed_at, r.ingested_at) ASC + `), + pool.query(` + SELECT + quote_id, + decision_id, + command_id, + execution_result_status, + execution_result_code, + submitted_at, + command_at, + outcome_status, + outcome_observed_at, + outcome_source, + attribution_status, + attribution_method, + attributed_inventory_delta, + computed_at, + payload + FROM ${QUOTE_OUTCOMES_TABLE} + ORDER BY COALESCE(outcome_observed_at, computed_at) ASC + `), + ]); + + return { + recentQuotes: quotesResult.rows.map(normalizeRecentQuoteRow), + recentTradeDecisions: decisionsResult.rows.map((row) => ({ + observed_at: toIsoTimestamp(row.observed_at), + ingested_at: toIsoTimestamp(row.ingested_at), + payload: row.payload, + })), + recentExecuteTradeCommands: commandsResult.rows.map((row) => normalizeExecuteTradeCommandRow(row)), + recentExecutionResults: executionResultsResult.rows.map((row) => normalizeExecutionResultRow(row)), + recentQuoteOutcomes: outcomesResult.rows.map((row) => normalizeQuoteOutcomeRow(row)), + }; +} + +async function upsertQuoteLifecycleStatistics(pool, statistics = []) { + for (const row of statistics || []) { + await pool.query( + ` + INSERT INTO ${QUOTE_LIFECYCLE_STATISTICS_TABLE} ( + stat_id, + grain, + window_start, + window_end, + quote_count, + missing_identifier_count, + missing_timestamp_count, + bucket_counts, + computed_at + ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8::jsonb,$9) + ON CONFLICT (stat_id) DO UPDATE SET + quote_count = EXCLUDED.quote_count, + missing_identifier_count = EXCLUDED.missing_identifier_count, + missing_timestamp_count = EXCLUDED.missing_timestamp_count, + bucket_counts = EXCLUDED.bucket_counts, + computed_at = EXCLUDED.computed_at + `, + [ + row.stat_id, + row.grain, + row.window_start, + row.window_end, + row.quote_count, + row.missing_identifier_count, + row.missing_timestamp_count, + JSON.stringify(row.bucket_counts || {}), + row.computed_at, + ], + ); + } +} + async function countSuccessfulQuoteEvidence(pool) { const result = await pool.query(` SELECT COUNT(*)::INT AS count diff --git a/src/operator-dashboard/static/pages/StrategyPage.jsx b/src/operator-dashboard/static/pages/StrategyPage.jsx index 136ed1c..f0ab027 100644 --- a/src/operator-dashboard/static/pages/StrategyPage.jsx +++ b/src/operator-dashboard/static/pages/StrategyPage.jsx @@ -31,6 +31,26 @@ const QUOTE_LIFECYCLE_PROBLEM_FILTERS = [ ['strategy_rejected', 'Strategy rejected'], ['submitted_awaiting_outcome', 'Submitted or awaiting outcome'], ]; +const QUOTE_LIFECYCLE_STAT_GRAINS = [ + ['five_minute', '5m'], + ['hour', 'Hour'], + ['day', 'Day'], + ['week', 'Week'], + ['month', 'Month'], + ['all_time', 'All-time'], +]; +const QUOTE_LIFECYCLE_STAT_BUCKETS = [ + ['success', 'Success'], + ['submitted_no_reply', 'Submitted no reply'], + ['submission_failed', 'Submission failed'], + ['awaiting_executor', 'Awaiting executor'], + ['approved_no_command', 'Approved no command'], + ['rejected_by_strategy', 'Rejected by strategy'], + ['blocked_before_submit', 'Blocked before submit'], + ['not_filled', 'Not filled'], + ['observed_no_decision', 'Observed no decision'], + ['unavailable', 'Unavailable'], +]; async function copyIdentifier(value) { if (!value || !navigator?.clipboard?.writeText) return; @@ -57,6 +77,21 @@ async function fetchQuoteLifecycleLookup(identifier) { return payload; } +async function fetchQuoteLifecycleStatistics() { + const response = await fetch('/api/strategy/quote-lifecycle/statistics?limit=160', { + headers: { + accept: 'application/json', + }, + }); + const text = await response.text(); + const payload = text.trim() ? JSON.parse(text) : null; + + if (!response.ok) { + throw new Error(payload?.error || text.trim() || `HTTP ${response.status}`); + } + return payload; +} + function useNow(intervalMs = 1000) { const [now, setNow] = useState(() => Date.now()); @@ -201,6 +236,41 @@ function plainCodeLabel(value, fallback = 'Unavailable') { return text.replaceAll('_', ' '); } +function formatInteger(value) { + const number = Number(value); + if (!Number.isFinite(number)) return '0'; + return number.toLocaleString(); +} + +function statisticBucketCount(row, bucket) { + return Number(row?.bucket_counts?.[bucket] || 0); +} + +function quoteLifecycleStatisticRows(statistics, source = 'persisted') { + return statistics?.[source]?.rows || []; +} + +function latestStatisticRow(rows, grain) { + const matches = (rows || []).filter((row) => row?.grain === grain); + if (grain === 'all_time') return matches[0] || null; + return matches.sort((left, right) => ( + Date.parse(right.window_start || '') - Date.parse(left.window_start || '') + ))[0] || null; +} + +function formatStatisticWindow(row) { + if (!row) return 'Unavailable'; + if (row.grain === 'all_time') return 'All-time'; + const start = formatTimestamp(row.window_start); + const end = formatTimestamp(row.window_end); + return `${start} to ${end}`; +} + +function statisticFreshnessLabel(value) { + if (!value) return 'Unavailable'; + return formatTimestamp(value); +} + function strategyDecisionStatus(decision) { if (decision?.decision === 'approved') return 'Strategy approved'; if (decision?.decision === 'rejected') return 'Strategy rejected'; @@ -852,10 +922,125 @@ function MakerCompetitivenessSection({ summary, pairConfig }) { ); } -function QuoteLifecycleTable({ items }) { +function QuoteLifecycleStatisticsPanel({ + error, + grain, + liveRows, + onGrainChange, + onRefresh, + persistedRows, + state, +}) { + const allTime = latestStatisticRow(persistedRows, 'all_time'); + const selectedRows = grain === 'all_time' + ? [allTime].filter(Boolean) + : (persistedRows || []).filter((row) => row?.grain === grain).slice(0, 8); + const selectedCurrent = latestStatisticRow(persistedRows, grain); + const liveCurrent = latestStatisticRow(liveRows, grain); + const persistedGeneratedAt = persistedRows?.[0]?.computed_at || null; + const liveGeneratedAt = liveRows?.[0]?.computed_at || null; + const rows = fixedRows(selectedRows, grain === 'all_time' ? 1 : 8); + + return ( +
+
+ + + + +
+ {error ?
{error}
: null} +
+
+ All-time unique quotes + {formatInteger(allTime?.quote_count)} +
+
+ {grain === 'all_time' ? 'All-time selected' : 'Current persisted window'} + {formatInteger(selectedCurrent?.quote_count)} +
+
+ Live recent window + {formatInteger(liveCurrent?.quote_count)} +
+
+ Unavailable evidence + {formatInteger((selectedCurrent?.missing_identifier_count || 0) + (selectedCurrent?.missing_timestamp_count || 0))} +
+
+ + + + + + + + + + + + + + + + {rows.map((row, index) => { + if (!row) { + return ( + + + + ); + } + const other = statisticBucketCount(row, 'approved_no_command') + + statisticBucketCount(row, 'blocked_before_submit') + + statisticBucketCount(row, 'not_filled') + + statisticBucketCount(row, 'observed_no_decision'); + const unavailable = statisticBucketCount(row, 'unavailable') + + Number(row.missing_identifier_count || 0) + + Number(row.missing_timestamp_count || 0); + return ( + + + + + + + + + + + ); + })} + +
WindowUnique quotesSuccessSubmitted no replySubmission failedAwaiting executorRejectedOther / unavailable
{index === 0 ? 'No persisted quote statistics are available yet.' : ''}
+
{formatStatisticWindow(row)}
+
{plainCodeLabel(row.grain)}
+ 		{formatInteger(row.quote_count)}{formatInteger(statisticBucketCount(row, 'success'))}{formatInteger(statisticBucketCount(row, 'submitted_no_reply'))}{formatInteger(statisticBucketCount(row, 'submission_failed'))}{formatInteger(statisticBucketCount(row, 'awaiting_executor'))}{formatInteger(statisticBucketCount(row, 'rejected_by_strategy'))} +
{formatInteger(other)} other
+
{formatInteger(unavailable)} unavailable
+
+ +
+ ); +} + +function QuoteLifecycleTable({ items, statistics }) { const [expanded, setExpanded] = useState(() => new Set()); const [showStrategyRejected, setShowStrategyRejected] = useState(true); const [problemFilter, setProblemFilter] = useState('all'); + const [statGrain, setStatGrain] = useState('five_minute'); + const [statsState, setStatsState] = useState('idle'); + const [statsError, setStatsError] = useState(null); + const [statisticsSnapshot, setStatisticsSnapshot] = useState(() => statistics || null); const latestItemsRef = useRef(items || []); const [quoteDisplayPaused, setQuoteDisplayPaused] = useState(false); const [displayItems, setDisplayItems] = useState(() => items || []); @@ -872,6 +1057,14 @@ function QuoteLifecycleTable({ items }) { selectedInvestigationRef.current = selectedInvestigation; }, [selectedInvestigation]); + useEffect(() => { + if (!statistics) return; + setStatisticsSnapshot((current) => ({ + persisted: statistics.persisted || current?.persisted || { source: 'persisted', generated_at: null, rows: [] }, + live: statistics.live || current?.live || { source: 'live_recent', generated_at: null, rows: [] }, + })); + }, [statistics]); + useEffect(() => () => { if (lookupTimerRef.current) window.clearTimeout(lookupTimerRef.current); }, []); @@ -1008,13 +1201,41 @@ function QuoteLifecycleTable({ items }) { setDisplayNow(Date.now()); } + async function refreshStatistics() { + setStatsState('loading'); + setStatsError(null); + try { + const payload = await fetchQuoteLifecycleStatistics(); + setStatisticsSnapshot((current) => ({ + persisted: payload?.persisted || current?.persisted || { source: 'persisted', generated_at: null, rows: [] }, + live: payload?.live || current?.live || { source: 'live_recent', generated_at: null, rows: [] }, + })); + setStatsState('loaded'); + } catch (error) { + setStatsState('error'); + setStatsError(error?.message || 'Statistics refresh failed'); + } + } + function toggleQuoteDisplayPaused() { if (quoteDisplayPaused) applyLatestLifecycleDisplay(); setQuoteDisplayPaused((paused) => !paused); } + const persistedStatisticRows = quoteLifecycleStatisticRows(statisticsSnapshot, 'persisted'); + const liveStatisticRows = quoteLifecycleStatisticRows(statisticsSnapshot, 'live'); + return ( <> + refreshStatistics().catch(() => {})} + persistedRows={persistedStatisticRows} + state={statsState} + />