Merge pull request #105 from usherlabs/refactor/external-market-data-collector

Use external broker for market data collection

Author: outerlook

.env.sample

@@ -37,10 +37,9 @@ CEX_BROKER_ARCHIVE_FORWARDER_URL=http://localhost:8090/archive
 # the container filesystem is not durable across container replacement.
 CEX_BROKER_ARCHIVE_DEAD_LETTER_PATH=./archive-loss.jsonl
 CEX_BROKER_DEPLOYMENT_ID=local-dev
-# OHLCV collector: required JSON array config and reconnect bootstrap coverage.
-# 1000 one-minute bars cover roughly 16 hours on every (re)subscription.
-CEX_BROKER_OHLCV_ARCHIVE_BOOTSTRAP_LIMIT=1000
-CEX_BROKER_OHLCV_COLLECTOR_CONFIG=./ohlcv-subscriptions.json
+# Market-data collector: external broker target and feed-intent JSON document.
+CEX_BROKER_URL=localhost:8086
+CEX_BROKER_MARKET_DATA_COLLECTOR_CONFIG=./market-data-subscriptions.json
 # ClickHouse (forwarder, candle-viewer, Python)
 CLICKHOUSE_HOST=localhost
 CLICKHOUSE_PORT=8123

…ub/workflows/publish-ohlcv-collector.yml …kflows/publish-market-data-collector.yml.github/workflows/publish-ohlcv-collector.yml renamed to .github/workflows/publish-market-data-collector.yml .github/workflows/publish-ohlcv-collector.yml renamed to .github/workflows/publish-market-data-collector.yml

@@ -1,11 +1,11 @@
-# Publishes the OHLCV collector image (services/ohlcv-collector).
+# Publishes the multi-feed market-data collector image (services/ohlcv-collector).
 # The collector is repo-runtime only — it is NOT part of the npm package
 # (files: ["dist"]), so it ships as its own container image.
 #
 # Triggers: semver tags (alongside the broker publish) and manual dispatch
 # (to publish from a branch without cutting an npm release).
 
-name: Publish OHLCV Collector
+name: Publish Market Data Collector
 
 on:
   push:
@@ -15,14 +15,14 @@ on:
   workflow_dispatch:
 
 env:
-  IMAGE_NAME: ghcr.io/usherlabs/cex-broker-ohlcv-collector
+  IMAGE_NAME: ghcr.io/usherlabs/cex-broker-market-data-collector
 
 permissions:
   contents: read
   packages: write
 
 jobs:
-  publish-ohlcv-collector:
+  publish-market-data-collector:
     runs-on: ubuntu-latest
     steps:
       - name: Checkout repo

docs/canonical-market-data-replay.md

@@ -18,7 +18,7 @@ CEX_BROKER_ARCHIVE_FORWARDER_URL=http://archive-forwarder:8090/archive
 CEX_BROKER_ARCHIVE_DEAD_LETTER_PATH=/var/lib/cex-broker/archive-loss.jsonl
 ```
 
-`CEX_BROKER_ARCHIVE_SOURCE` is closed to `broker_read|broker_write` and defaults to `broker_write` for existing deployments. The writer stamps this immutable value into envelopes, rows, and loss records. It is never inferred from API-key presence. Production FIET-901 collector startup requires `broker_read` and a non-empty deployment-owned capture bundle. Development capture explicitly uses `CEX_BROKER_MARKET_CAPTURE_ENVIRONMENT=development` and generates a `development:<deployment>` bundle when none is supplied.
+`CEX_BROKER_ARCHIVE_SOURCE` is closed to `broker_read|broker_write` and defaults to `broker_write` for existing deployments. The writer stamps this immutable value into envelopes, rows, and loss records. It is never inferred from API-key presence. Production market-data collection requires the separately deployed broker to use `broker_read` and a non-empty deployment-owned capture bundle. Development capture explicitly uses `CEX_BROKER_MARKET_CAPTURE_ENVIRONMENT=development` and generates a `development:<deployment>` bundle when none is supplied.
 
 Credential resolution uses the broker's established fixed precedence and requires no archive-specific credential configuration:
 
@@ -52,8 +52,6 @@ Run it with `bun run start-market-data-collector`. The collector starts no loopb
 
 Each entry has an independent reconnect supervisor and health state. OHLCV retains bootstrap/catch-up and stamps `broker_bootstrap_fetch_v1` separately from live capture. ORDERBOOK, TICKER, and TRADES record unrecoverable gaps after reconnect rather than synthesize missing events.
 
-`CEX_BROKER_OHLCV_COLLECTOR_CONFIG` remains supported as an OHLCV-only array. Move those entries into the canonical document and add `feed: "OHLCV"`; archive identity stays on the broker. The `start-ohlcv-collector` command and existing service directory/image remain compatibility wrappers for both formats.
-
 An external CCXT or Hummingbot fallback is an optional out-of-band producer of the shared capture contract, not a broker-collector implementation. It must declare its provider, versioned fallback source mode, reason, configured exchange, and configured pair. Cross-venue or cross-pair substitution is rejected.
 
 ## Capture and integrity contract

package.json

@@ -36,7 +36,7 @@
 		"prepack": "bun run build",
 		"start-broker": "bun run ./src/cli.ts",
 		"start-archive-forwarder": "bun run services/archive-forwarder/index.ts",
-		"start-ohlcv-collector": "bun run services/ohlcv-collector/index.ts",
+		"start-market-data-collector": "bun run services/ohlcv-collector/index.ts",
 		"start-archive-watch": "bun run examples/archive-watch-subscribe.ts",
 		"start-candle-viewer": "bun run research/candle-viewer/server.ts",
 		"dev:candle-viewer": "bunx nodemon --watch research/candle-viewer --ext ts,html --signal SIGTERM --exec \"bun run start-candle-viewer\"",

services/ohlcv-collector/Dockerfile

@@ -1,12 +1,12 @@
-# OHLCV collector: runs a loopback-only, keyless public broker and keeps one
-# Subscribe(OHLCV) stream alive for every configured pair. The broker forwards
-# captured candles to the separately deployed archive-forwarder.
+# Market-data collector: keeps one external Subscribe stream alive for every
+# configured feed intent against CEX_BROKER_URL.
 #
 # Build from the repository root:
-#   docker build -f services/ohlcv-collector/Dockerfile -t cex-broker-ohlcv-collector .
+#   docker build -f services/ohlcv-collector/Dockerfile -t cex-broker-market-data-collector .
 #
-# Mount a JSON array of {"exchange","symbol","timeframe"} entries and set
-# CEX_BROKER_OHLCV_COLLECTOR_CONFIG to its container path. timeframe defaults to 1m.
+# Set CEX_BROKER_URL to the external broker target and
+# CEX_BROKER_MARKET_DATA_COLLECTOR_CONFIG to a JSON document containing
+# {"subscriptions":[...]} feed intents.
 FROM oven/bun:1.3
 
 WORKDIR /app
@@ -18,11 +18,7 @@ RUN bun install --frozen-lockfile
 COPY src ./src
 COPY services/ohlcv-collector ./services/ohlcv-collector
 
-# Every reconnect invokes the broker's OHLCV bootstrap. At 1m, 1000 bars provide
-# roughly 16 hours of gap coverage without adding a collector-side write/backfill path.
-ENV CEX_BROKER_OHLCV_ARCHIVE_BOOTSTRAP_LIMIT=1000
-
 # The collector needs no elevated privileges; run as the base image's bun user.
 USER bun
 
-CMD ["bun", "run", "services/ohlcv-collector/index.ts"]
+CMD ["bun", "run", "start-market-data-collector"]

services/ohlcv-collector/collector.ts

@@ -3,7 +3,7 @@ import { SubscriptionType } from "../../src/helpers/constants";
 import { log } from "../../src/helpers/logger";
 import type { OtelMetrics } from "../../src/helpers/otel";
 import { CEX_BROKER_PACKAGE_DEFINITION } from "../../src/proto-package-definition";
-import type { MarketDataSubscription, OhlcvSubscription } from "./config";
+import type { MarketDataSubscription } from "./config";
 
 type SubscribeResponse = {
 	data: string;
@@ -20,16 +20,14 @@ type SubscribeClient = grpc.Client & {
 
 type CollectorMetrics = Pick<OtelMetrics, "recordCounter">;
 
-export type OhlcvCollectorOptions = {
+export type MarketDataCollectorOptions = {
 	brokerUrl: string;
-	subscriptions: Array<OhlcvSubscription | MarketDataSubscription>;
+	subscriptions: MarketDataSubscription[];
 	metrics?: CollectorMetrics;
 	retry?: Partial<RetryPolicy>;
 };
 
-type CollectorSubscription =
-	| (OhlcvSubscription & { feed: "OHLCV"; bootstrapLimit?: number })
-	| MarketDataSubscription;
+type CollectorSubscription = MarketDataSubscription;
 
 export type CollectorFeedHealth = {
 	state: "connecting" | "healthy" | "backoff" | "stopped";
@@ -113,20 +111,16 @@ function waitForDelay(delayMs: number, signal: AbortSignal): Promise<boolean> {
 	});
 }
 
-export class OhlcvCollector {
+export class MarketDataCollector {
 	readonly #client: SubscribeClient;
 	readonly #subscriptions: CollectorSubscription[];
 	readonly #metrics?: CollectorMetrics;
 	readonly #retry: RetryPolicy;
 	readonly #health = new Map<string, CollectorFeedHealth>();
 	#started = false;
 
-	constructor(options: OhlcvCollectorOptions) {
-		this.#subscriptions = options.subscriptions.map((subscription) =>
-			"feed" in subscription
-				? subscription
-				: { ...subscription, feed: "OHLCV" as const },
-		);
+	constructor(options: MarketDataCollectorOptions) {
+		this.#subscriptions = options.subscriptions;
 		this.#metrics = options.metrics;
 		this.#retry = { ...DEFAULT_RETRY_POLICY, ...options.retry };
 		this.#client = new grpcObject.cex_broker.cex_service(
@@ -161,7 +155,7 @@ export class OhlcvCollector {
 
 	async run(signal: AbortSignal): Promise<void> {
 		if (this.#started) {
-			throw new Error("OHLCV collector can only be started once");
+			throw new Error("Market-data collector can only be started once");
 		}
 		this.#started = true;
 		try {
@@ -370,5 +364,3 @@ export class OhlcvCollector {
 		return Math.max(0, Math.round(baseDelay * jitter));
 	}
 }
-
-export const MarketDataCollector = OhlcvCollector;

services/ohlcv-collector/config.ts

@@ -1,6 +1,5 @@
 import { z } from "zod";
 
-export const OHLCV_COLLECTOR_CONFIG_ENV = "CEX_BROKER_OHLCV_COLLECTOR_CONFIG";
 export const MARKET_DATA_COLLECTOR_CONFIG_ENV =
 	"CEX_BROKER_MARKET_DATA_COLLECTOR_CONFIG";
 
@@ -47,19 +46,10 @@ const marketSubscriptionSchema = z.discriminatedUnion("feed", [
 
 const marketConfigSchema = z
 	.object({
-		captureBundleId: z.string().trim().optional(),
-		environment: z.enum(["development", "production"]).default("production"),
 		subscriptions: z.array(marketSubscriptionSchema).min(1),
 	})
 	.strict()
 	.superRefine((config, context) => {
-		if (config.environment === "production" && !config.captureBundleId) {
-			context.addIssue({
-				code: "custom",
-				message: "captureBundleId is required in production",
-				path: ["captureBundleId"],
-			});
-		}
 		const seen = new Set<string>();
 		for (const [index, subscription] of config.subscriptions.entries()) {
 			const options =
@@ -133,74 +123,3 @@ export async function loadMarketDataCollectorConfig(
 		throw error;
 	}
 }
-
-const subscriptionSchema = z
-	.object({
-		exchange: z
-			.string()
-			.trim()
-			.min(1)
-			.transform((value) => value.toLowerCase()),
-		symbol: z.string().trim().min(1),
-		timeframe: z.string().trim().min(1).default("1m"),
-	})
-	.strict();
-
-const configSchema = z
-	.array(subscriptionSchema)
-	.min(1, "at least one OHLCV subscription is required")
-	.superRefine((subscriptions, context) => {
-		const seen = new Set<string>();
-		for (const [index, subscription] of subscriptions.entries()) {
-			const key = `${subscription.exchange}\u0000${subscription.symbol}\u0000${subscription.timeframe}`;
-			if (seen.has(key)) {
-				context.addIssue({
-					code: "custom",
-					message: "duplicate OHLCV subscription",
-					path: [index],
-				});
-			}
-			seen.add(key);
-		}
-	});
-
-export type OhlcvSubscription = z.infer<typeof subscriptionSchema>;
-
-export function parseOhlcvCollectorConfig(input: unknown): OhlcvSubscription[] {
-	const result = configSchema.safeParse(input);
-	if (!result.success) {
-		throw new Error(
-			`Invalid OHLCV collector config: ${formatZodIssues(result.error)}`,
-		);
-	}
-	return result.data;
-}
-
-export async function loadOhlcvCollectorConfig(
-	configPath = process.env[OHLCV_COLLECTOR_CONFIG_ENV],
-): Promise<OhlcvSubscription[]> {
-	const path = configPath?.trim();
-	if (!path) {
-		throw new Error(`${OHLCV_COLLECTOR_CONFIG_ENV} must point to a JSON file`);
-	}
-
-	let contents: string;
-	try {
-		contents = await Bun.file(path).text();
-	} catch (error) {
-		throw new Error(`Failed to read OHLCV collector config at ${path}`, {
-			cause: error,
-		});
-	}
-
-	let parsed: unknown;
-	try {
-		parsed = JSON.parse(contents);
-	} catch (error) {
-		throw new Error(`OHLCV collector config at ${path} is not valid JSON`, {
-			cause: error,
-		});
-	}
-
-	return parseOhlcvCollectorConfig(parsed);
-}