Merge pull request #61 from usherlabs/archive-fills-transfers

Archive durable fill and transfer execution facts

Author: outerlook

schema/clickhouse/broker_execution.sql

@@ -65,6 +65,93 @@ ENGINE = MergeTree
 PARTITION BY toYYYYMM(parseDateTimeBestEffortOrZero(broker_observed_timestamp))
 ORDER BY (exchange, symbol, broker_observed_timestamp);
 
+-- CEX value movements: withdrawals, deposits, and sub<->master internal transfers.
+--
+-- Column names/types/ORDER BY match the fiet-maker consumer contract
+-- (docs/CEX_EXECUTION_ARCHIVE_CONTRACT.md + scripts/sql/clickhouse-schema.sql):
+-- MergeTree with a 90-day TTL, DateTime64(3,'UTC') timestamps, string quantities,
+-- result_index UInt32, error_summary. Two ADDITIVE columns not in the consumer
+-- contract: fee_amount / fee_currency (the ccxt withdrawal object exposes the fee,
+-- the dominant small-commit cost). broker_observed_timestamp is emitted as an
+-- ISO-8601 UTC string and parsed on insert via the forwarder's
+-- date_time_input_format=best_effort (see services/archive-forwarder/index.ts).
+--
+-- Engine is plain MergeTree (contract), so re-observed rows are NOT collapsed;
+-- dedup, when needed, is at read time (GROUP BY / argMax over exchange,
+-- account_selector, symbol, external_id, lifecycle_action).
+CREATE TABLE IF NOT EXISTS broker_execution.transfer_events
+(
+    broker_observed_timestamp DateTime64(3, 'UTC'),
+    source LowCardinality(String),
+    deployment_id LowCardinality(String),
+    schema_version LowCardinality(String),
+    account_selector LowCardinality(String),
+    exchange LowCardinality(String),
+    symbol LowCardinality(String),
+    event_kind LowCardinality(String),
+    lifecycle_action LowCardinality(String),
+    status LowCardinality(String) DEFAULT '',
+    asset_symbol LowCardinality(String) DEFAULT '',
+    amount String DEFAULT '',
+    address String DEFAULT '',
+    network LowCardinality(String) DEFAULT '',
+    external_id String DEFAULT '',
+    txid String DEFAULT '',
+    result_index UInt32 DEFAULT 0,
+    fee_amount String DEFAULT '',
+    fee_currency LowCardinality(String) DEFAULT '',
+    exchange_timestamp Nullable(DateTime64(3, 'UTC')),
+    error_summary String DEFAULT '',
+    payload_json String DEFAULT ''
+)
+ENGINE = MergeTree
+PARTITION BY toDate(broker_observed_timestamp)
+ORDER BY (account_selector, broker_observed_timestamp, exchange, symbol, event_kind, lifecycle_action)
+TTL toDateTime(broker_observed_timestamp) + toIntervalDay(90)
+SETTINGS ttl_only_drop_parts = 1;
+
+-- Per-fill execution facts from the venue trade-history endpoint (fetchMyTrades),
+-- captured by the broker-internal fill poller. GetOrderDetails/createOrder payloads
+-- carry no per-trade breakdown and no fee on most venues, so per-fill truth
+-- (incl. fee) requires this endpoint; hence event_kind is stamped
+-- "trade_history_fill" rather than the contract fixture's "create_order_fill".
+--
+-- Column names/types/ORDER BY match the fiet-maker consumer contract: MergeTree +
+-- 90-day TTL, DateTime64 timestamps, string quantities, fill_index UInt32. Plain
+-- MergeTree (contract): the poller re-scans a lookback window after a restart, so
+-- the same trade can be re-inserted; dedup is at read time (GROUP BY / argMax over
+-- exchange, account_selector, symbol, order_id, fill_id).
+CREATE TABLE IF NOT EXISTS broker_execution.fill_events
+(
+    broker_observed_timestamp DateTime64(3, 'UTC'),
+    source LowCardinality(String),
+    deployment_id LowCardinality(String),
+    schema_version LowCardinality(String),
+    account_selector LowCardinality(String),
+    exchange LowCardinality(String),
+    symbol LowCardinality(String),
+    event_kind LowCardinality(String),
+    order_id String,
+    client_order_id String DEFAULT '',
+    fill_id String DEFAULT '',
+    fill_index UInt32 DEFAULT 0,
+    side LowCardinality(String) DEFAULT '',
+    order_type LowCardinality(String) DEFAULT '',
+    price String DEFAULT '',
+    base_quantity String DEFAULT '',
+    quote_quantity String DEFAULT '',
+    fee_amount String DEFAULT '',
+    fee_currency LowCardinality(String) DEFAULT '',
+    fee_rate String DEFAULT '',
+    exchange_timestamp Nullable(DateTime64(3, 'UTC')),
+    payload_json String DEFAULT ''
+)
+ENGINE = MergeTree
+PARTITION BY toDate(broker_observed_timestamp)
+ORDER BY (symbol, account_selector, broker_observed_timestamp, exchange, order_id, fill_index)
+TTL toDateTime(broker_observed_timestamp) + toIntervalDay(90)
+SETTINGS ttl_only_drop_parts = 1;
+
 -- Pre-order top-of-book snapshots captured immediately before an order action,
 -- joinable to order_events via market_metadata_hash and the order identifiers.
 CREATE TABLE IF NOT EXISTS broker_execution.market_metadata_snapshots

services/archive-forwarder/index.ts

@@ -17,6 +17,12 @@ const clickhouse = createClient({
 	username: config.clickhouse.username,
 	password: config.clickhouse.password,
 	database: config.clickhouse.database,
+	// broker_execution.transfer_events/fill_events use DateTime64 columns; producers
+	// emit broker_observed_timestamp/exchange_timestamp as ISO-8601 UTC strings, so
+	// the inserter must best-effort-parse them (basic mode rejects the 'T'/'Z' form).
+	// Strictly more lenient than basic, so it never breaks the existing String/Int64
+	// timestamp columns on the other archive tables.
+	clickhouse_settings: { date_time_input_format: "best_effort" },
 });
 const inserter = createClickHouseInserter(clickhouse);
 
@@ -82,11 +88,18 @@ const server = Bun.serve({
 			}
 
 			if (parsed.rejectedRowCount > 0) {
+				// Name the offending tables: an unknown table (e.g. a forgotten
+				// SUPPORTED_TABLES entry for a new archive table) would otherwise reject
+				// the whole batch with only a count, hiding which table is at fault.
+				console.warn(
+					`Rejected ${parsed.rejectedRowCount}/${parsed.inputRowCount} archive row(s) from ${parsed.batch.source}; tables: ${parsed.rejectedTables.join(", ")}`,
+				);
 				return Response.json(
 					{
 						error: "Malformed archive rows in batch",
 						rejected: parsed.rejectedRowCount,
 						inputRows: parsed.inputRowCount,
+						rejectedTables: parsed.rejectedTables,
 					},
 					{ status: 400 },
 				);

services/archive-forwarder/router.ts

@@ -8,6 +8,10 @@ export type ParsedArchiveBatch =
 			batch: ArchiveBatchRequest;
 			inputRowCount: number;
 			rejectedRowCount: number;
+			// Distinct table names among rejected rows (unknown/unsupported tables),
+			// so the caller can name them in a WARN instead of dropping silently.
+			// A rejected row with no string `table` contributes "(malformed)".
+			rejectedTables: string[];
 	  }
 	| { ok: false };
 
@@ -45,6 +49,7 @@ export function parseArchiveBatchRequest(body: unknown): ParsedArchiveBatch {
 	}
 	const inputRowCount = record.rows.length;
 	const rows = record.rows.filter(isValidArchiveRow);
+	const rejectedTables = collectRejectedTables(record.rows);
 	return {
 		ok: true,
 		batch: {
@@ -54,9 +59,22 @@ export function parseArchiveBatchRequest(body: unknown): ParsedArchiveBatch {
 		},
 		inputRowCount,
 		rejectedRowCount: inputRowCount - rows.length,
+		rejectedTables,
 	};
 }
 
+function collectRejectedTables(rows: unknown[]): string[] {
+	const tables = new Set<string>();
+	for (const entry of rows) {
+		if (isValidArchiveRow(entry)) {
+			continue;
+		}
+		const table = (entry as { table?: unknown } | null)?.table;
+		tables.add(typeof table === "string" ? table : "(malformed)");
+	}
+	return [...tables];
+}
+
 /** @deprecated Use parseArchiveBatchRequest returning ParsedArchiveBatch */
 export function parseArchiveBatchRequestLegacy(
 	body: unknown,

services/archive-forwarder/types.ts

@@ -25,6 +25,8 @@ export const SUPPORTED_TABLES = [
 	"market_data.cex_trades",
 	"broker_execution.order_events",
 	"broker_execution.market_metadata_snapshots",
+	"broker_execution.transfer_events",
+	"broker_execution.fill_events",
 	"strategy_data.policy_evaluation_events",
 	"strategy_data.strategy_policy_snapshots",
 	"strategy_data.inventory_settlement_events",

src/handlers/execute-action/context.ts

@@ -13,6 +13,7 @@ import {
 	resolveGrpcError,
 	stableGrpcErrorCode,
 } from "../../helpers/grpc/status";
+import type { OrderActivityTracker } from "../../helpers/order-activity-tracker";
 import type { OtelMetrics } from "../../helpers/otel";
 import { getErrorMessage } from "../../helpers/shared/errors";
 import type { PolicyConfig } from "../../types";
@@ -38,6 +39,7 @@ export type ExecuteActionContext = {
 	verityProverUrl: string;
 	otelMetrics?: OtelMetrics;
 	brokerArchiver?: BrokerExecutionArchiver;
+	orderActivityTracker?: OrderActivityTracker;
 };
 
 export function requireSymbol(

src/handlers/execute-action/deposit.ts

@@ -1,5 +1,6 @@
 import * as grpc from "@grpc/grpc-js";
 import ccxt from "@usherlabs/ccxt";
+import { archiveTransferEventInBackground } from "../../helpers/broker-execution-archive";
 import {
 	depositField,
 	depositMatchesTransaction,
@@ -19,7 +20,13 @@ import type { ExecuteActionContext } from "./context";
 import { parsePayloadForAction, rejectWithGrpcError } from "./context";
 
 export async function handleDeposit(ctx: ExecuteActionContext): Promise<void> {
-	const { normalizedCex, symbol, selectedBrokerAccount, broker } = ctx;
+	const {
+		normalizedCex,
+		symbol,
+		selectedBrokerAccount,
+		broker,
+		brokerArchiver,
+	} = ctx;
 
 	if (!symbol) {
 		return ctx.wrappedCallback(
@@ -117,6 +124,40 @@ export async function handleDeposit(ctx: ExecuteActionContext): Promise<void> {
 			const status = normalizeDepositStatus(
 				depositField(deposit, ["status", "state"]),
 			);
+			const depositTxid = String(
+				depositField(deposit, ["txid", "txId", "tx_hash", "txHash"]) ??
+					value.transactionHash,
+			);
+			const creditedAt = depositField(deposit, [
+				"creditedAt",
+				"credited_at",
+				"updated",
+				"updatedAt",
+				"timestamp",
+				"datetime",
+			]);
+			// Observed deposit lifecycle fact (contract lifecycle_action
+			// "observe_deposit"). MergeTree keeps all observations; dedup, if needed,
+			// is at read time over (exchange, account, symbol, external_id, status).
+			archiveTransferEventInBackground(brokerArchiver, {
+				exchange: normalizedCex,
+				accountSelector: selectedBrokerAccount?.label,
+				assetSymbol: symbol,
+				transfer: {
+					eventKind: "deposit",
+					lifecycleAction: "observe_deposit",
+					status,
+					amount:
+						observedAmount !== undefined ? String(observedAmount) : undefined,
+					address: String(observedAddress ?? value.recipientAddress),
+					network: depositNetwork?.exchangeNetworkId,
+					externalId: depositTxid,
+					txid: depositTxid,
+					exchangeTimestamp:
+						typeof creditedAt === "string" ? creditedAt : undefined,
+					payload: deposit,
+				},
+			});
 			log.info(
 				`Amount ${value.amount} at ${value.transactionHash} . Paid to ${value.recipientAddress}`,
 			);

src/handlers/execute-action/handler.ts

@@ -7,6 +7,7 @@ import type { BrokerExecutionArchiver } from "../../helpers/broker-execution-arc
 import { Action, getActionName, resolveAction } from "../../helpers/constants";
 import { selectBrokerAccountForCex } from "../../helpers/grpc/broker";
 import { log } from "../../helpers/logger";
+import type { OrderActivityTracker } from "../../helpers/order-activity-tracker";
 import type { OtelMetrics } from "../../helpers/otel";
 import { safeLogError } from "../../helpers/shared/errors";
 import {
@@ -27,6 +28,7 @@ export type ExecuteActionDeps = {
 	verityProverUrl: string;
 	otelMetrics?: OtelMetrics;
 	brokerArchiver?: BrokerExecutionArchiver;
+	orderActivityTracker?: OrderActivityTracker;
 };
 
 export function createExecuteActionHandler(deps: ExecuteActionDeps) {
@@ -38,6 +40,7 @@ export function createExecuteActionHandler(deps: ExecuteActionDeps) {
 		verityProverUrl,
 		otelMetrics,
 		brokerArchiver,
+		orderActivityTracker,
 	} = deps;
 
 	return async (
@@ -151,6 +154,7 @@ export function createExecuteActionHandler(deps: ExecuteActionDeps) {
 				verityProverUrl,
 				otelMetrics,
 				brokerArchiver,
+				orderActivityTracker,
 			};
 
 			if (action === Action.Call) {

src/handlers/execute-action/internal-transfer.ts

@@ -7,6 +7,10 @@ import {
 	transferBinanceInternal,
 	verityHttpClientOverridePredicate,
 } from "../../helpers";
+import {
+	archiveTransferEventInBackground,
+	normalizeCcxtTransactionForArchive,
+} from "../../helpers/broker-execution-archive";
 import { mapCcxtErrorToGrpcStatus } from "../../helpers/grpc/status";
 import { log } from "../../helpers/logger";
 import { getErrorMessage, safeLogError } from "../../helpers/shared/errors";
@@ -25,6 +29,7 @@ export async function handleInternalTransfer(
 		verity,
 		useVerity,
 		verityProverUrl,
+		brokerArchiver,
 	} = ctx;
 
 	if (!symbol) {
@@ -103,6 +108,22 @@ export async function handleInternalTransfer(
 			symbol,
 			transferPayload.amount,
 		);
+		// account_selector is the source; the destination is kept in payload_json.
+		const normalized = normalizeCcxtTransactionForArchive(result);
+		archiveTransferEventInBackground(brokerArchiver, {
+			exchange: normalizedCex,
+			accountSelector: fromSelector,
+			assetSymbol: symbol,
+			transfer: {
+				eventKind: "internal_transfer",
+				lifecycleAction: "submit_internal_transfer",
+				status: normalized.status ?? "ok",
+				amount: normalized.amount ?? String(transferPayload.amount),
+				network: "internal",
+				externalId: normalized.externalId,
+				payload: { from: fromSelector, to: toSelector, result },
+			},
+		});
 		ctx.wrappedCallback(null, {
 			proof: verity.proof,
 			result: JSON.stringify(result),

src/handlers/execute-action/orders.ts

@@ -36,6 +36,7 @@ async function handleCreateOrder(ctx: ExecuteActionContext): Promise<void> {
 		verityProverUrl,
 		otelMetrics,
 		brokerArchiver,
+		orderActivityTracker,
 	} = ctx;
 	const verityProof = verity.proof;
 
@@ -82,6 +83,14 @@ async function handleCreateOrder(ctx: ExecuteActionContext): Promise<void> {
 			side: resolution.side,
 			requestedQuantity: resolution.amountBase ?? orderValue.amount,
 		};
+		// Mark this (account, symbol) so the fill poller scans it for trade history.
+		if (selectedBrokerAccount?.label) {
+			orderActivityTracker?.record(
+				cex,
+				selectedBrokerAccount.label,
+				resolution.symbol,
+			);
+		}
 		const telemetryIds = extractOrderTelemetryIds(orderValue.params);
 		const submissionTimestamp = new Date().toISOString();
 		const marketMetadataHash = await captureMarketMetadataSnapshot(
@@ -183,6 +192,7 @@ async function handleGetOrderDetails(ctx: ExecuteActionContext): Promise<void> {
 		verityProverUrl,
 		otelMetrics,
 		brokerArchiver,
+		orderActivityTracker,
 	} = ctx;
 	const verityProof = verity.proof;
 
@@ -207,6 +217,9 @@ async function handleGetOrderDetails(ctx: ExecuteActionContext): Promise<void> {
 			symbol,
 			{ ...getOrderValue.params },
 		);
+		if (selectedBrokerAccount?.label && symbol) {
+			orderActivityTracker?.record(cex, selectedBrokerAccount.label, symbol);
+		}
 		const getOrderContext = {
 			action: "GetOrderDetails" as const,
 			cex,
@@ -285,6 +298,7 @@ async function handleCancelOrder(ctx: ExecuteActionContext): Promise<void> {
 		verityProverUrl,
 		otelMetrics,
 		brokerArchiver,
+		orderActivityTracker,
 	} = ctx;
 	const verityProof = verity.proof;
 
@@ -312,6 +326,9 @@ async function handleCancelOrder(ctx: ExecuteActionContext): Promise<void> {
 			symbol,
 			cancelOrderValue.params ?? {},
 		);
+		if (selectedBrokerAccount?.label && symbol) {
+			orderActivityTracker?.record(cex, selectedBrokerAccount.label, symbol);
+		}
 		emitOrderExecutionTelemetryInBackground(
 			otelMetrics,
 			cancelOrderContext,