dirtydishes/islandflow
1
0
Fork 0
Code Issues 2 Pull requests 4 Projects Releases Packages Wiki Activity Actions

Implement first-pass load reduction controls

Browse source
This commit is contained in:
dirtydishes 2026-05-08 02:46:41 -04:00
parent 5d488fd7f5
commit e7f4805ccc
17 changed files with 1191 additions and 608 deletions
Download patch file Download diff file Expand all files Collapse all files

187
services/api/src/index.ts
View file

@ -1,5 +1,5 @@
import { readEnv } from "@islandflow/config";
import { createLogger } from "@islandflow/observability";
import { createLogger, createMetrics } from "@islandflow/observability";
import {
SUBJECT_ALERTS,
SUBJECT_CLASSIFIER_HITS,
@ -23,6 +23,7 @@ import {
STREAM_SMART_MONEY_EVENTS,
STREAM_OPTION_NBBO,
STREAM_OPTION_SIGNAL_PRINTS,
buildStreamConfig,
buildDurableConsumer,
connectJetStreamWithRetry,
ensureStream,
@ -107,11 +108,17 @@ import {
} from "@islandflow/types";
import { createClient } from "redis";
import { z } from "zod";
import { HOT_LIVE_REDIS_KEYS, LiveStateManager, shouldFanoutLiveEvent } from "./live";
import {
HOT_LIVE_REDIS_KEYS,
LiveStateManager,
resolveLiveStateConfig,
shouldFanoutLiveEvent
} from "./live";
import { parseOptionPrintQuery } from "./option-queries";
const service = "api";
const logger = createLogger({ service });
const metrics = createMetrics({ service });
const DeliverPolicySchema = z.enum(["new", "all", "last", "last_per_subject"]);
@ -617,148 +624,17 @@ const run = async () => {
{ attempts: 120, delayMs: 500 }
);
await ensureStream(jsm, {
name: STREAM_OPTION_SIGNAL_PRINTS,
subjects: [SUBJECT_OPTION_SIGNAL_PRINTS],
retention: "limits",
storage: "file",
discard: "old",
max_msgs_per_subject: -1,
max_msgs: -1,
max_bytes: -1,
max_age: 0,
num_replicas: 1
});
await ensureStream(jsm, {
name: STREAM_OPTION_NBBO,
subjects: [SUBJECT_OPTION_NBBO],
retention: "limits",
storage: "file",
discard: "old",
max_msgs_per_subject: -1,
max_msgs: -1,
max_bytes: -1,
max_age: 0,
num_replicas: 1
});
await ensureStream(jsm, {
name: STREAM_EQUITY_PRINTS,
subjects: [SUBJECT_EQUITY_PRINTS],
retention: "limits",
storage: "file",
discard: "old",
max_msgs_per_subject: -1,
max_msgs: -1,
max_bytes: -1,
max_age: 0,
num_replicas: 1
});
await ensureStream(jsm, {
name: STREAM_EQUITY_QUOTES,
subjects: [SUBJECT_EQUITY_QUOTES],
retention: "limits",
storage: "file",
discard: "old",
max_msgs_per_subject: -1,
max_msgs: -1,
max_bytes: -1,
max_age: 0,
num_replicas: 1
});
await ensureStream(jsm, {
name: STREAM_EQUITY_CANDLES,
subjects: [SUBJECT_EQUITY_CANDLES],
retention: "limits",
storage: "file",
discard: "old",
max_msgs_per_subject: -1,
max_msgs: -1,
max_bytes: -1,
max_age: 0,
num_replicas: 1
});
await ensureStream(jsm, {
name: STREAM_EQUITY_JOINS,
subjects: [SUBJECT_EQUITY_JOINS],
retention: "limits",
storage: "file",
discard: "old",
max_msgs_per_subject: -1,
max_msgs: -1,
max_bytes: -1,
max_age: 0,
num_replicas: 1
});
await ensureStream(jsm, {
name: STREAM_INFERRED_DARK,
subjects: [SUBJECT_INFERRED_DARK],
retention: "limits",
storage: "file",
discard: "old",
max_msgs_per_subject: -1,
max_msgs: -1,
max_bytes: -1,
max_age: 0,
num_replicas: 1
});
await ensureStream(jsm, {
name: STREAM_FLOW_PACKETS,
subjects: [SUBJECT_FLOW_PACKETS],
retention: "limits",
storage: "file",
discard: "old",
max_msgs_per_subject: -1,
max_msgs: -1,
max_bytes: -1,
max_age: 0,
num_replicas: 1
});
await ensureStream(jsm, {
name: STREAM_SMART_MONEY_EVENTS,
subjects: [SUBJECT_SMART_MONEY_EVENTS],
retention: "limits",
storage: "file",
discard: "old",
max_msgs_per_subject: -1,
max_msgs: -1,
max_bytes: -1,
max_age: 0,
num_replicas: 1
});
await ensureStream(jsm, {
name: STREAM_CLASSIFIER_HITS,
subjects: [SUBJECT_CLASSIFIER_HITS],
retention: "limits",
storage: "file",
discard: "old",
max_msgs_per_subject: -1,
max_msgs: -1,
max_bytes: -1,
max_age: 0,
num_replicas: 1
});
await ensureStream(jsm, {
name: STREAM_ALERTS,
subjects: [SUBJECT_ALERTS],
retention: "limits",
storage: "file",
discard: "old",
max_msgs_per_subject: -1,
max_msgs: -1,
max_bytes: -1,
max_age: 0,
num_replicas: 1
});
await ensureStream(jsm, buildStreamConfig(STREAM_OPTION_SIGNAL_PRINTS, SUBJECT_OPTION_SIGNAL_PRINTS, "derived"));
await ensureStream(jsm, buildStreamConfig(STREAM_OPTION_NBBO, SUBJECT_OPTION_NBBO, "raw"));
await ensureStream(jsm, buildStreamConfig(STREAM_EQUITY_PRINTS, SUBJECT_EQUITY_PRINTS, "raw"));
await ensureStream(jsm, buildStreamConfig(STREAM_EQUITY_QUOTES, SUBJECT_EQUITY_QUOTES, "raw"));
await ensureStream(jsm, buildStreamConfig(STREAM_EQUITY_CANDLES, SUBJECT_EQUITY_CANDLES, "derived"));
await ensureStream(jsm, buildStreamConfig(STREAM_EQUITY_JOINS, SUBJECT_EQUITY_JOINS, "derived"));
await ensureStream(jsm, buildStreamConfig(STREAM_INFERRED_DARK, SUBJECT_INFERRED_DARK, "derived"));
await ensureStream(jsm, buildStreamConfig(STREAM_FLOW_PACKETS, SUBJECT_FLOW_PACKETS, "derived"));
await ensureStream(jsm, buildStreamConfig(STREAM_SMART_MONEY_EVENTS, SUBJECT_SMART_MONEY_EVENTS, "derived"));
await ensureStream(jsm, buildStreamConfig(STREAM_CLASSIFIER_HITS, SUBJECT_CLASSIFIER_HITS, "derived"));
await ensureStream(jsm, buildStreamConfig(STREAM_ALERTS, SUBJECT_ALERTS, "derived"));
const clickhouse = createClickHouseClient({
url: env.CLICKHOUSE_URL,
@ -804,7 +680,7 @@ const run = async () => {
redis = null;
}
const liveState = new LiveStateManager(clickhouse, redis);
const liveState = new LiveStateManager(clickhouse, redis, resolveLiveStateConfig());
await liveState.hydrate();
const warnLiveLag = (
channel: keyof typeof HOT_LIVE_REDIS_KEYS,
@ -1069,6 +945,11 @@ const run = async () => {
return;
}
const optionItem = ingestChannel === "options" ? (item as Parameters<typeof matchesOptionPrintFilters>[0]) : null;
const equityItem = ingestChannel === "equities" ? (item as Parameters<typeof matchesScopedEquitySubscription>[0]) : null;
const flowItem = ingestChannel === "flow" ? (item as Parameters<typeof matchesFlowPacketFilters>[0]) : null;
let matchedSubscriptions = 0;
for (const [key, candidate] of matchingSubscriptions) {
const sockets = subscriptionSockets.get(key);
if (!sockets || sockets.size === 0) {
@ -1077,26 +958,29 @@ const run = async () => {
if (
candidate.channel === "options" &&
(!matchesOptionPrintFilters(OptionPrintSchema.parse(item), candidate.filters) ||
!matchesScopedOptionSubscription(OptionPrintSchema.parse(item), candidate))
(!optionItem ||
!matchesOptionPrintFilters(optionItem, candidate.filters) ||
!matchesScopedOptionSubscription(optionItem, candidate))
) {
continue;
}
if (
candidate.channel === "equities" &&
!matchesScopedEquitySubscription(EquityPrintSchema.parse(item), candidate)
(!equityItem || !matchesScopedEquitySubscription(equityItem, candidate))
) {
continue;
}
if (
candidate.channel === "flow" &&
!matchesFlowPacketFilters(FlowPacketSchema.parse(item), candidate.filters)
(!flowItem || !matchesFlowPacketFilters(flowItem, candidate.filters))
) {
continue;
}
matchedSubscriptions += 1;
for (const socket of sockets) {
sendLiveMessage(socket, {
op: "event",
@ -1106,6 +990,10 @@ const run = async () => {
});
}
}
if (matchedSubscriptions > 0) {
metrics.count("api.live.subscription_match_count", matchedSubscriptions);
}
};
const pumpOptions = async () => {
@ -1931,6 +1819,7 @@ const run = async () => {
logger.info("service stopping", { signal });
server.stop();
clearInterval(liveStateMetricsTimer);
await liveState.close();
if (redis && redis.isOpen) {
try {

439
services/api/src/live.ts
View file

@ -41,12 +41,15 @@ import {
type EquityPrint,
type LiveChannel
} from "@islandflow/types";
import { createMetrics } from "@islandflow/observability";
import type { RedisClientType } from "redis";
const CURSOR_HASH_KEY = "live:cursors";
export const LIVE_FEED_LOOKBACK_MS = 24 * 60 * 60 * 1000;
const DEFAULT_GENERIC_LIMIT = 10000;
const metrics = createMetrics({ service: "api" });
const DEFAULT_GENERIC_LIMIT = 1000;
const MAX_GENERIC_LIMIT = 100000;
const MIN_GENERIC_LIMIT = 1;
const GENERIC_LIMIT_ENV_KEYS: Record<LiveGenericChannel, string> = {
@ -67,6 +70,23 @@ const CHART_LIMITS = {
overlay: 1500
} as const;
const DEFAULT_LIVE_LIMITS: GenericLiveLimits = {
options: 1000,
nbbo: 1000,
equities: 1000,
"equity-quotes": 500,
"equity-joins": 500,
flow: 500,
"smart-money": 300,
"classifier-hits": 300,
alerts: 300,
"inferred-dark": 300
};
const DEFAULT_SCOPED_CACHE_MAX_KEYS = 32;
const DEFAULT_REDIS_FLUSH_INTERVAL_MS = 250;
const DEFAULT_REDIS_FLUSH_MAX_ITEMS = 100;
type GenericFeedConfig = {
redisKey: string;
cursorField: string;
@ -93,6 +113,13 @@ export const HOT_LIVE_REDIS_KEYS = {
export type GenericLiveLimits = Record<LiveGenericChannel, number>;
type LiveStateConfig = {
limits: GenericLiveLimits;
scopedCacheMaxKeys: number;
redisFlushIntervalMs: number;
redisFlushMaxItems: number;
};
const parseGenericLimit = (
env: NodeJS.ProcessEnv,
channel: LiveGenericChannel,
@ -117,17 +144,77 @@ const parseGenericLimit = (
return bounded;
};
export const resolveGenericLiveLimits = (env: NodeJS.ProcessEnv = process.env): GenericLiveLimits => ({
options: parseGenericLimit(env, "options", DEFAULT_GENERIC_LIMIT),
nbbo: parseGenericLimit(env, "nbbo", DEFAULT_GENERIC_LIMIT),
equities: parseGenericLimit(env, "equities", DEFAULT_GENERIC_LIMIT),
"equity-quotes": parseGenericLimit(env, "equity-quotes", DEFAULT_GENERIC_LIMIT),
"equity-joins": parseGenericLimit(env, "equity-joins", DEFAULT_GENERIC_LIMIT),
flow: parseGenericLimit(env, "flow", DEFAULT_GENERIC_LIMIT),
"smart-money": parseGenericLimit(env, "smart-money", DEFAULT_GENERIC_LIMIT),
"classifier-hits": parseGenericLimit(env, "classifier-hits", DEFAULT_GENERIC_LIMIT),
alerts: parseGenericLimit(env, "alerts", DEFAULT_GENERIC_LIMIT),
"inferred-dark": parseGenericLimit(env, "inferred-dark", DEFAULT_GENERIC_LIMIT)
const parseGenericLimitFallback = (env: NodeJS.ProcessEnv, fallback: number): number => {
const raw = env.LIVE_LIMIT_DEFAULT;
if (!raw || raw.trim().length === 0) {
return fallback;
}
const parsed = Number(raw);
if (!Number.isFinite(parsed)) {
console.warn(`Invalid LIVE_LIMIT_DEFAULT="${raw}", using ${fallback}`);
return fallback;
}
return Math.max(MIN_GENERIC_LIMIT, Math.min(MAX_GENERIC_LIMIT, Math.floor(parsed)));
};
export const resolveGenericLiveLimits = (env: NodeJS.ProcessEnv = process.env): GenericLiveLimits => {
const liveLimitDefault = parseGenericLimitFallback(env, DEFAULT_GENERIC_LIMIT);
return {
options: parseGenericLimit(env, "options", env.LIVE_LIMIT_DEFAULT ? liveLimitDefault : DEFAULT_LIVE_LIMITS.options),
nbbo: parseGenericLimit(env, "nbbo", env.LIVE_LIMIT_DEFAULT ? liveLimitDefault : DEFAULT_LIVE_LIMITS.nbbo),
equities: parseGenericLimit(
env,
"equities",
env.LIVE_LIMIT_DEFAULT ? liveLimitDefault : DEFAULT_LIVE_LIMITS.equities
),
"equity-quotes": parseGenericLimit(
env,
"equity-quotes",
env.LIVE_LIMIT_DEFAULT ? liveLimitDefault : DEFAULT_LIVE_LIMITS["equity-quotes"]
),
"equity-joins": parseGenericLimit(
env,
"equity-joins",
env.LIVE_LIMIT_DEFAULT ? liveLimitDefault : DEFAULT_LIVE_LIMITS["equity-joins"]
),
flow: parseGenericLimit(env, "flow", env.LIVE_LIMIT_DEFAULT ? liveLimitDefault : DEFAULT_LIVE_LIMITS.flow),
"smart-money": parseGenericLimit(
env,
"smart-money",
env.LIVE_LIMIT_DEFAULT ? liveLimitDefault : DEFAULT_LIVE_LIMITS["smart-money"]
),
"classifier-hits": parseGenericLimit(
env,
"classifier-hits",
env.LIVE_LIMIT_DEFAULT ? liveLimitDefault : DEFAULT_LIVE_LIMITS["classifier-hits"]
),
alerts: parseGenericLimit(env, "alerts", env.LIVE_LIMIT_DEFAULT ? liveLimitDefault : DEFAULT_LIVE_LIMITS.alerts),
"inferred-dark": parseGenericLimit(
env,
"inferred-dark",
env.LIVE_LIMIT_DEFAULT ? liveLimitDefault : DEFAULT_LIVE_LIMITS["inferred-dark"]
)
};
};
const parsePositiveInt = (value: string | undefined, fallback: number): number => {
const parsed = Number(value);
if (!Number.isFinite(parsed)) {
return fallback;
}
return Math.max(1, Math.floor(parsed));
};
export const resolveLiveStateConfig = (env: NodeJS.ProcessEnv = process.env): LiveStateConfig => ({
limits: resolveGenericLiveLimits(env),
scopedCacheMaxKeys: parsePositiveInt(env.LIVE_SCOPED_CACHE_MAX_KEYS, DEFAULT_SCOPED_CACHE_MAX_KEYS),
redisFlushIntervalMs: parsePositiveInt(
env.LIVE_REDIS_FLUSH_INTERVAL_MS,
DEFAULT_REDIS_FLUSH_INTERVAL_MS
),
redisFlushMaxItems: parsePositiveInt(env.LIVE_REDIS_FLUSH_MAX_ITEMS, DEFAULT_REDIS_FLUSH_MAX_ITEMS)
});
type RedisLike = Pick<
@ -378,7 +465,50 @@ const candleCursorField = (underlyingId: string, intervalMs: number): string =>
const overlayRedisKey = (underlyingId: string): string => `live:equity-overlay:${underlyingId}`;
const overlayCursorField = (underlyingId: string): string => `equities:${underlyingId}`;
const dropMatchingCursor = <T>(
items: T[],
target: Cursor,
cursorOf: (item: T) => Cursor
): T[] => items.filter((item) => compareCursors(cursorOf(item), target) !== 0);
const insertNewestFirst = <T>(
items: T[],
item: T,
cursorOf: (item: T) => Cursor,
limit: number
): { items: T[]; outOfOrder: boolean } => {
const cursor = cursorOf(item);
const deduped = dropMatchingCursor(items, cursor, cursorOf);
const head = deduped[0];
const outOfOrder = head ? compareCursors(cursor, cursorOf(head)) > 0 : false;
if (!outOfOrder) {
return {
items: [item, ...deduped].slice(0, limit),
outOfOrder: false
};
}
return {
items: sortGenericItems([...deduped, item], cursorOf).slice(0, limit),
outOfOrder: true
};
};
type BufferedRedisWrite = {
listKey: string;
cursorField: string;
items: unknown[];
limit: number;
cursor: Cursor | null;
updates: number;
};
const isLiveStateConfig = (value: GenericLiveLimits | LiveStateConfig): value is LiveStateConfig =>
"limits" in value;
export class LiveStateManager {
private readonly config: LiveStateConfig;
private readonly generic: {
[K in LiveGenericChannel]: GenericFeedConfig;
};
@ -386,14 +516,22 @@ export class LiveStateManager {
private readonly genericCursors = new Map<string, Cursor | null>();
private readonly candleItems = new Map<string, EquityCandle[]>();
private readonly candleCursors = new Map<string, Cursor | null>();
private readonly candleAccess = new Map<string, number>();
private readonly overlayItems = new Map<string, EquityPrint[]>();
private readonly overlayCursors = new Map<string, Cursor | null>();
private readonly overlayAccess = new Map<string, number>();
private readonly pendingRedisWrites = new Map<string, BufferedRedisWrite>();
private readonly redisFlushTimer: ReturnType<typeof setInterval> | null;
private readonly stats = {
genericHydrateFromRedis: 0,
genericHydrateFromClickHouse: 0,
genericCacheSnapshots: 0,
scopedClickHouseSnapshots: 0,
trimOperations: 0,
redisFlushCount: 0,
redisFlushItems: 0,
cacheEvictions: 0,
outOfOrderEvents: 0,
cacheDepthByKey: new Map<string, number>(),
freshnessAgeMsByKey: new Map<string, number>()
};
@ -401,9 +539,31 @@ export class LiveStateManager {
constructor(
private readonly clickhouse: ClickHouseClient,
private readonly redis: RedisLike | null,
limits: GenericLiveLimits = resolveGenericLiveLimits()
config: GenericLiveLimits | LiveStateConfig = resolveLiveStateConfig()
) {
this.generic = getGenericConfig(limits);
this.config = isLiveStateConfig(config)
? config
: {
limits: config,
scopedCacheMaxKeys: DEFAULT_SCOPED_CACHE_MAX_KEYS,
redisFlushIntervalMs: DEFAULT_REDIS_FLUSH_INTERVAL_MS,
redisFlushMaxItems: DEFAULT_REDIS_FLUSH_MAX_ITEMS
};
this.generic = getGenericConfig(this.config.limits);
this.redisFlushTimer =
this.redis && this.redis.isOpen
? setInterval(() => {
void this.flushRedisWrites();
}, this.config.redisFlushIntervalMs)
: null;
this.redisFlushTimer?.unref?.();
}
async close(): Promise<void> {
if (this.redisFlushTimer) {
clearInterval(this.redisFlushTimer);
}
await this.flushRedisWrites();
}
getStatsSnapshot(): {
@ -412,6 +572,10 @@ export class LiveStateManager {
genericCacheSnapshots: number;
scopedClickHouseSnapshots: number;
trimOperations: number;
redisFlushCount: number;
redisFlushItems: number;
cacheEvictions: number;
outOfOrderEvents: number;
cacheDepthByKey: Record<string, number>;
freshnessAgeMsByKey: Record<string, number>;
} {
@ -421,6 +585,10 @@ export class LiveStateManager {
genericCacheSnapshots: this.stats.genericCacheSnapshots,
scopedClickHouseSnapshots: this.stats.scopedClickHouseSnapshots,
trimOperations: this.stats.trimOperations,
redisFlushCount: this.stats.redisFlushCount,
redisFlushItems: this.stats.redisFlushItems,
cacheEvictions: this.stats.cacheEvictions,
outOfOrderEvents: this.stats.outOfOrderEvents,
cacheDepthByKey: Object.fromEntries(this.stats.cacheDepthByKey),
freshnessAgeMsByKey: Object.fromEntries(this.stats.freshnessAgeMsByKey)
};
@ -435,6 +603,23 @@ export class LiveStateManager {
};
}
async flushRedisWrites(): Promise<void> {
if (!this.redis?.isOpen) {
return;
}
const writes = Array.from(this.pendingRedisWrites.values());
this.pendingRedisWrites.clear();
for (const write of writes) {
await this.persistList(write.listKey, write.cursorField, write.items, write.limit, write.cursor);
this.stats.redisFlushCount += 1;
this.stats.redisFlushItems += write.items.length;
metrics.count("api.live.redis_flush_count", 1);
metrics.count("api.live.redis_flush_items", write.items.length);
}
}
private getChannelHealth(channel: LiveHotChannel): LiveChannelHealth {
const listKey = HOT_LIVE_REDIS_KEYS[channel];
const thresholdMs = LIVE_FRESHNESS_THRESHOLDS[channel];
@ -449,6 +634,34 @@ export class LiveStateManager {
};
}
private touchAccess(accessMap: Map<string, number>, key: string): void {
accessMap.set(key, Date.now());
}
private evictScopedCachesIfNeeded(
itemsMap: Map<string, unknown[]>,
cursorsMap: Map<string, Cursor | null>,
accessMap: Map<string, number>
): void {
while (itemsMap.size > this.config.scopedCacheMaxKeys) {
const oldest = [...accessMap.entries()].sort((a, b) => a[1] - b[1])[0];
if (!oldest) {
break;
}
const [key] = oldest;
itemsMap.delete(key);
cursorsMap.delete(
key.startsWith("live:equity-candles:")
? key.replace("live:", "")
: key.replace("live:equity-overlay:", "equities:")
);
accessMap.delete(key);
this.stats.cacheDepthByKey.delete(key);
this.stats.cacheEvictions += 1;
metrics.count("api.live.cache_evictions", 1);
}
}
private updateFreshnessMetric(listKey: string, channel: LiveChannel, item: unknown, now = Date.now()): void {
const ts =
channel === "equity-candles" || channel === "equity-overlay"
@ -462,6 +675,32 @@ export class LiveStateManager {
}
}
private queueRedisWrite(
listKey: string,
cursorField: string,
items: unknown[],
limit: number,
cursor: Cursor | null
): void {
if (!this.redis?.isOpen) {
return;
}
const existing = this.pendingRedisWrites.get(listKey);
const write: BufferedRedisWrite = {
listKey,
cursorField,
items: [...items],
limit,
cursor,
updates: (existing?.updates ?? 0) + 1
};
this.pendingRedisWrites.set(listKey, write);
if (write.updates >= this.config.redisFlushMaxItems) {
void this.flushRedisWrites();
}
}
async hydrate(): Promise<void> {
const channels = Object.keys(this.generic) as LiveGenericChannel[];
await Promise.all(channels.map((channel) => this.hydrateGeneric(channel)));
@ -477,23 +716,16 @@ export class LiveStateManager {
this.stats.genericHydrateFromRedis += 1;
this.stats.cacheDepthByKey.set(config.redisKey, cached.length);
this.updateFreshnessMetric(config.redisKey, channel, cached[0]);
this.genericCursors.set(config.cursorField, parseCursor(await this.redis.hGet(CURSOR_HASH_KEY, config.cursorField)));
await this.persistList(
config.redisKey,
this.genericCursors.set(
config.cursorField,
cached,
config.limit,
this.genericCursors.get(config.cursorField) ?? null
parseCursor(await this.redis.hGet(CURSOR_HASH_KEY, config.cursorField))
);
await this.persistList(config.redisKey, config.cursorField, cached, config.limit, this.genericCursors.get(config.cursorField) ?? null);
return;
}
}
const fresh = normalizeGenericItems(
channel,
await config.fetchRecent(this.clickhouse, config.limit),
config
);
const fresh = normalizeGenericItems(channel, await config.fetchRecent(this.clickhouse, config.limit), config);
this.stats.genericHydrateFromClickHouse += 1;
this.stats.cacheDepthByKey.set(config.redisKey, fresh.length);
this.genericItems.set(channel, fresh);
@ -508,32 +740,26 @@ export class LiveStateManager {
async getSnapshot(subscription: LiveSubscription): Promise<FeedSnapshot<unknown>> {
switch (subscription.channel) {
case "options": {
const scoped =
Boolean(subscription.underlying_ids?.length) || Boolean(subscription.option_contract_id);
const scoped = Boolean(subscription.underlying_ids?.length) || Boolean(subscription.option_contract_id);
if (subscription.filters?.view === "raw" || scoped) {
this.stats.scopedClickHouseSnapshots += 1;
const limit = snapshotLimitFor(subscription, this.generic.options.limit);
const storageFilters = buildOptionSnapshotFilters(subscription);
const items = await fetchRecentOptionPrints(
this.clickhouse,
limit,
undefined,
storageFilters
);
const items = await fetchRecentOptionPrints(this.clickhouse, limit, undefined, storageFilters);
return {
subscription,
items,
watermark: items[0] ? { ts: items[0].ts, seq: items[0].seq } : null,
next_before: nextBeforeForItems(items, (item) => ({ ts: item.ts, seq: item.seq }))
next_before: nextBeforeForItems(items, (entry) => ({ ts: entry.ts, seq: entry.seq }))
};
}
const config = this.generic.options;
this.stats.genericCacheSnapshots += 1;
const limit = snapshotLimitFor(subscription, config.limit);
const items = (this.genericItems.get("options") ?? []).filter((item) =>
matchesOptionPrintFilters(item, subscription.filters)
).slice(0, limit);
const items = (this.genericItems.get("options") ?? [])
.filter((entry) => matchesOptionPrintFilters(entry, subscription.filters))
.slice(0, limit);
return {
subscription,
items,
@ -545,9 +771,9 @@ export class LiveStateManager {
const config = this.generic.flow;
this.stats.genericCacheSnapshots += 1;
const limit = snapshotLimitFor(subscription, config.limit);
const items = (this.genericItems.get("flow") ?? []).filter((item) =>
matchesFlowPacketFilters(item, subscription.filters)
).slice(0, limit);
const items = (this.genericItems.get("flow") ?? [])
.filter((entry) => matchesFlowPacketFilters(entry, subscription.filters))
.slice(0, limit);
return {
subscription,
items,
@ -560,9 +786,7 @@ export class LiveStateManager {
const limit = snapshotLimitFor(subscription, config.limit);
if (subscription.underlying_ids?.length) {
this.stats.scopedClickHouseSnapshots += 1;
const filters: EquityPrintQueryFilters = {
underlyingIds: subscription.underlying_ids
};
const filters: EquityPrintQueryFilters = { underlyingIds: subscription.underlying_ids };
const items = await fetchRecentEquityPrints(this.clickhouse, limit, filters);
return {
subscription,
@ -586,12 +810,13 @@ export class LiveStateManager {
if (!this.candleItems.has(key)) {
await this.hydrateCandles(subscription.underlying_id, subscription.interval_ms);
}
this.touchAccess(this.candleAccess, key);
const items = this.candleItems.get(key) ?? [];
return {
subscription,
items,
watermark: this.candleCursors.get(cursorField) ?? null,
next_before: nextBeforeForItems(items, (item) => ({ ts: item.ts, seq: item.seq }))
next_before: nextBeforeForItems(items, (entry) => ({ ts: entry.ts, seq: entry.seq }))
};
}
case "equity-overlay": {
@ -600,12 +825,13 @@ export class LiveStateManager {
if (!this.overlayItems.has(key)) {
await this.hydrateOverlay(subscription.underlying_id);
}
this.touchAccess(this.overlayAccess, key);
const items = this.overlayItems.get(key) ?? [];
return {
subscription,
items,
watermark: this.overlayCursors.get(cursorField) ?? null,
next_before: nextBeforeForItems(items, (item) => ({ ts: item.ts, seq: item.seq }))
next_before: nextBeforeForItems(items, (entry) => ({ ts: entry.ts, seq: entry.seq }))
};
}
default: {
@ -629,48 +855,52 @@ export class LiveStateManager {
const candle = EquityCandleSchema.parse(item);
const key = candleRedisKey(candle.underlying_id, candle.interval_ms);
const cursorField = candleCursorField(candle.underlying_id, candle.interval_ms);
const previousCursor = this.candleCursors.get(cursorField) ?? null;
const items = this.candleItems.get(key) ?? [];
const next = [candle, ...items]
.sort((a, b) => (b.ts - a.ts) || (b.seq - a.seq))
.slice(0, CHART_LIMITS.candles);
this.candleItems.set(key, next);
this.stats.cacheDepthByKey.set(key, next.length);
const nextState = insertNewestFirst(
this.candleItems.get(key) ?? [],
candle,
(entry) => ({ ts: entry.ts, seq: entry.seq }),
CHART_LIMITS.candles
);
const cursor = { ts: candle.ts, seq: candle.seq };
this.candleItems.set(key, nextState.items);
this.candleCursors.set(cursorField, cursor);
if (next.length > 0) {
this.updateFreshnessMetric(key, "equity-candles", next[0]);
this.touchAccess(this.candleAccess, key);
this.evictScopedCachesIfNeeded(this.candleItems as Map<string, unknown[]>, this.candleCursors, this.candleAccess);
if (nextState.outOfOrder) {
this.stats.outOfOrderEvents += 1;
metrics.count("api.live.out_of_order_events", 1);
}
const outOfOrder = previousCursor ? compareCursors(cursor, previousCursor) > 0 : false;
if (outOfOrder) {
await this.persistList(key, cursorField, next, CHART_LIMITS.candles, cursor);
} else {
await this.persistItem(key, cursorField, candle, CHART_LIMITS.candles, cursor, next.length);
this.stats.cacheDepthByKey.set(key, nextState.items.length);
if (nextState.items.length > 0) {
this.updateFreshnessMetric(key, "equity-candles", nextState.items[0]);
}
this.queueRedisWrite(key, cursorField, nextState.items, CHART_LIMITS.candles, cursor);
return cursor;
}
case "equity-overlay": {
const print = EquityPrintSchema.parse(item);
const key = overlayRedisKey(print.underlying_id);
const cursorField = overlayCursorField(print.underlying_id);
const previousCursor = this.overlayCursors.get(cursorField) ?? null;
const items = this.overlayItems.get(key) ?? [];
const next = [print, ...items]
.sort((a, b) => (b.ts - a.ts) || (b.seq - a.seq))
.slice(0, CHART_LIMITS.overlay);
this.overlayItems.set(key, next);
this.stats.cacheDepthByKey.set(key, next.length);
const nextState = insertNewestFirst(
this.overlayItems.get(key) ?? [],
print,
(entry) => ({ ts: entry.ts, seq: entry.seq }),
CHART_LIMITS.overlay
);
const cursor = { ts: print.ts, seq: print.seq };
this.overlayItems.set(key, nextState.items);
this.overlayCursors.set(cursorField, cursor);
if (next.length > 0) {
this.updateFreshnessMetric(key, "equity-overlay", next[0]);
this.touchAccess(this.overlayAccess, key);
this.evictScopedCachesIfNeeded(this.overlayItems as Map<string, unknown[]>, this.overlayCursors, this.overlayAccess);
if (nextState.outOfOrder) {
this.stats.outOfOrderEvents += 1;
metrics.count("api.live.out_of_order_events", 1);
}
const outOfOrder = previousCursor ? compareCursors(cursor, previousCursor) > 0 : false;
if (outOfOrder) {
await this.persistList(key, cursorField, next, CHART_LIMITS.overlay, cursor);
} else {
await this.persistItem(key, cursorField, print, CHART_LIMITS.overlay, cursor, next.length);
this.stats.cacheDepthByKey.set(key, nextState.items.length);
if (nextState.items.length > 0) {
this.updateFreshnessMetric(key, "equity-overlay", nextState.items[0]);
}
this.queueRedisWrite(key, cursorField, nextState.items, CHART_LIMITS.overlay, cursor);
return cursor;
}
default: {
@ -679,22 +909,28 @@ export class LiveStateManager {
if (!isWithinLiveFeedLookback(channel, parsed)) {
return null;
}
const previousCursor = this.genericCursors.get(config.cursorField) ?? null;
const items = this.genericItems.get(channel) ?? [];
const next = normalizeGenericItems(channel, [parsed, ...items], config);
this.genericItems.set(channel, next);
this.stats.cacheDepthByKey.set(config.redisKey, next.length);
const cursor = config.cursor(parsed);
const nextState =
channel === "nbbo"
? {
items: normalizeGenericItems(channel, [parsed, ...(this.genericItems.get(channel) ?? [])], config),
outOfOrder: false
}
: insertNewestFirst(this.genericItems.get(channel) ?? [], parsed, config.cursor, config.limit);
if (nextState.outOfOrder) {
this.stats.outOfOrderEvents += 1;
metrics.count("api.live.out_of_order_events", 1);
}
this.genericItems.set(channel, nextState.items);
this.genericCursors.set(config.cursorField, cursor);
if (next.length > 0) {
this.updateFreshnessMetric(config.redisKey, channel, next[0]);
}
const outOfOrder = previousCursor ? compareCursors(cursor, previousCursor) > 0 : false;
if (channel === "nbbo" || outOfOrder) {
await this.persistList(config.redisKey, config.cursorField, next, config.limit, cursor);
} else {
await this.persistItem(config.redisKey, config.cursorField, parsed, config.limit, cursor, next.length);
this.stats.cacheDepthByKey.set(config.redisKey, nextState.items.length);
if (nextState.items.length > 0) {
this.updateFreshnessMetric(config.redisKey, channel, nextState.items[0]);
}
this.queueRedisWrite(config.redisKey, config.cursorField, nextState.items, config.limit, cursor);
return cursor;
}
}
@ -708,6 +944,8 @@ export class LiveStateManager {
const cached = parseJsonList(payloads, (value) => EquityCandleSchema.parse(value));
if (cached.length > 0) {
this.candleItems.set(key, cached);
this.touchAccess(this.candleAccess, key);
this.evictScopedCachesIfNeeded(this.candleItems as Map<string, unknown[]>, this.candleCursors, this.candleAccess);
this.stats.cacheDepthByKey.set(key, cached.length);
this.updateFreshnessMetric(key, "equity-candles", cached[0]);
this.candleCursors.set(cursorField, parseCursor(await this.redis.hGet(CURSOR_HASH_KEY, cursorField)));
@ -717,6 +955,8 @@ export class LiveStateManager {
const fresh = await fetchRecentEquityCandles(this.clickhouse, underlyingId, intervalMs, CHART_LIMITS.candles);
this.candleItems.set(key, fresh);
this.touchAccess(this.candleAccess, key);
this.evictScopedCachesIfNeeded(this.candleItems as Map<string, unknown[]>, this.candleCursors, this.candleAccess);
this.stats.cacheDepthByKey.set(key, fresh.length);
if (fresh.length > 0) {
this.updateFreshnessMetric(key, "equity-candles", fresh[0]);
@ -734,6 +974,8 @@ export class LiveStateManager {
const cached = parseJsonList(payloads, (value) => EquityPrintSchema.parse(value));
if (cached.length > 0) {
this.overlayItems.set(key, cached);
this.touchAccess(this.overlayAccess, key);
this.evictScopedCachesIfNeeded(this.overlayItems as Map<string, unknown[]>, this.overlayCursors, this.overlayAccess);
this.stats.cacheDepthByKey.set(key, cached.length);
this.updateFreshnessMetric(key, "equity-overlay", cached[0]);
this.overlayCursors.set(cursorField, parseCursor(await this.redis.hGet(CURSOR_HASH_KEY, cursorField)));
@ -742,9 +984,11 @@ export class LiveStateManager {
}
const fresh = (await fetchRecentEquityPrints(this.clickhouse, CHART_LIMITS.overlay)).filter(
(item) => item.underlying_id === underlyingId
(entry) => entry.underlying_id === underlyingId
);
this.overlayItems.set(key, fresh);
this.touchAccess(this.overlayAccess, key);
this.evictScopedCachesIfNeeded(this.overlayItems as Map<string, unknown[]>, this.overlayCursors, this.overlayAccess);
this.stats.cacheDepthByKey.set(key, fresh.length);
if (fresh.length > 0) {
this.updateFreshnessMetric(key, "equity-overlay", fresh[0]);
@ -754,25 +998,6 @@ export class LiveStateManager {
await this.persistList(key, cursorField, fresh, CHART_LIMITS.overlay, watermark);
}
private async persistItem<T>(
listKey: string,
cursorField: string,
item: T,
limit: number,
cursor: Cursor | null,
depth: number
): Promise<void> {
if (!this.redis?.isOpen) {
return;
}
await this.redis.lPush(listKey, JSON.stringify(item));
await this.redis.lTrim(listKey, 0, limit - 1);
this.stats.trimOperations += 1;
this.stats.cacheDepthByKey.set(listKey, Math.min(depth, limit));
await this.redis.hSet(CURSOR_HASH_KEY, cursorField, JSON.stringify(cursor));
}
private async persistList<T>(
listKey: string,
cursorField: string,
@ -784,7 +1009,7 @@ export class LiveStateManager {
return;
}
const payloads = items.map((item) => JSON.stringify(item));
const payloads = items.map((entry) => JSON.stringify(entry));
await this.redis.lTrim(listKey, 1, 0);
this.stats.trimOperations += 1;
if (payloads.length > 0) {

116
services/api/tests/live.test.ts
View file

@ -66,9 +66,9 @@ describe("LiveStateManager", () => {
expect(limits.options).toBe(777);
expect(limits.nbbo).toBe(100000);
expect(limits.flow).toBe(10000);
expect(limits["equity-quotes"]).toBe(10000);
expect(limits.alerts).toBe(10000);
expect(limits.flow).toBe(500);
expect(limits["equity-quotes"]).toBe(500);
expect(limits.alerts).toBe(300);
});
it("hydrates snapshots from redis generic windows", async () => {
@ -204,13 +204,121 @@ describe("LiveStateManager", () => {
]);
const persisted = await redis.lRange("live:flow", 0, 99);
expect(persisted).toHaveLength(2);
await manager.flushRedisWrites();
const flushed = await redis.lRange("live:flow", 0, 99);
expect(persisted).toHaveLength(0);
expect(flushed).toHaveLength(2);
const stats = manager.getStatsSnapshot();
expect(stats.trimOperations).toBeGreaterThan(0);
expect(stats.redisFlushCount).toBeGreaterThan(0);
expect(stats.cacheDepthByKey["live:flow"]).toBe(2);
});
it("reorders out-of-order live events without dropping newest-first semantics", async () => {
const now = Date.now();
const manager = new LiveStateManager(makeClickHouse(), null, {
limits: {
options: 1000,
nbbo: 1000,
equities: 1000,
"equity-quotes": 500,
"equity-joins": 500,
flow: 3,
"smart-money": 300,
"classifier-hits": 300,
alerts: 300,
"inferred-dark": 300
},
scopedCacheMaxKeys: 32,
redisFlushIntervalMs: 250,
redisFlushMaxItems: 100
});
await manager.ingest("flow", {
source_ts: now,
ingest_ts: now + 1,
seq: 2,
trace_id: "flow-2",
id: "flow-2",
members: [],
features: {},
join_quality: {}
});
await manager.ingest("flow", {
source_ts: now - 1_000,
ingest_ts: now - 999,
seq: 1,
trace_id: "flow-1",
id: "flow-1",
members: [],
features: {},
join_quality: {}
});
const snapshot = await manager.getSnapshot({ channel: "flow" });
expect((snapshot.items as Array<{ id: string }>).map((item) => item.id)).toEqual([
"flow-2",
"flow-1"
]);
expect(manager.getStatsSnapshot().outOfOrderEvents).toBe(1);
});
it("evicts least-recently-used scoped candle caches past the configured key limit", async () => {
const manager = new LiveStateManager(makeClickHouse(), null, {
limits: resolveGenericLiveLimits(),
scopedCacheMaxKeys: 1,
redisFlushIntervalMs: 250,
redisFlushMaxItems: 100
});
await manager.ingest("equity-candles", {
source_ts: 100,
ingest_ts: 101,
seq: 1,
trace_id: "candle:SPY:60000:100",
ts: 100,
interval_ms: 60000,
underlying_id: "SPY",
open: 1,
high: 2,
low: 1,
close: 2,
volume: 10,
trade_count: 1
});
await manager.ingest("equity-candles", {
source_ts: 200,
ingest_ts: 201,
seq: 2,
trace_id: "candle:QQQ:60000:200",
ts: 200,
interval_ms: 60000,
underlying_id: "QQQ",
open: 3,
high: 4,
low: 3,
close: 4,
volume: 20,
trade_count: 2
});
const qqqSnapshot = await manager.getSnapshot({
channel: "equity-candles",
underlying_id: "QQQ",
interval_ms: 60000
});
const spySnapshot = await manager.getSnapshot({
channel: "equity-candles",
underlying_id: "SPY",
interval_ms: 60000
});
expect(qqqSnapshot.items).toHaveLength(1);
expect(spySnapshot.items).toEqual([]);
expect(manager.getStatsSnapshot().cacheEvictions).toBeGreaterThan(0);
});
it("filters option and flow snapshots using subscription filters", async () => {
const manager = new LiveStateManager(makeClickHouse(), null);
const now = Date.now();

28
services/candles/src/index.ts
View file

@ -5,6 +5,7 @@ import {
SUBJECT_EQUITY_PRINTS,
STREAM_EQUITY_CANDLES,
STREAM_EQUITY_PRINTS,
buildStreamConfig,
buildDurableConsumer,
connectJetStreamWithRetry,
ensureStream,
@ -240,31 +241,8 @@ const run = async () => {
{ attempts: 120, delayMs: 500 }
);
await ensureStream(jsm, {
name: STREAM_EQUITY_PRINTS,
subjects: [SUBJECT_EQUITY_PRINTS],
retention: "limits",
storage: "file",
discard: "old",
max_msgs_per_subject: -1,
max_msgs: -1,
max_bytes: -1,
max_age: 0,
num_replicas: 1
});
await ensureStream(jsm, {
name: STREAM_EQUITY_CANDLES,
subjects: [SUBJECT_EQUITY_CANDLES],
retention: "limits",
storage: "file",
discard: "old",
max_msgs_per_subject: -1,
max_msgs: -1,
max_bytes: -1,
max_age: 0,
num_replicas: 1
});
await ensureStream(jsm, buildStreamConfig(STREAM_EQUITY_PRINTS, SUBJECT_EQUITY_PRINTS, "raw"));
await ensureStream(jsm, buildStreamConfig(STREAM_EQUITY_CANDLES, SUBJECT_EQUITY_CANDLES, "derived"));
const clickhouse = createClickHouseClient({
url: env.CLICKHOUSE_URL,

395
services/compute/src/index.ts
View file

@ -26,6 +26,7 @@ import {
STREAM_SMART_MONEY_EVENTS,
STREAM_OPTION_NBBO,
STREAM_OPTION_SIGNAL_PRINTS,
buildStreamConfig,
buildDurableConsumer,
connectJetStreamWithRetry,
ensureStream,
@ -40,12 +41,13 @@ import {
ensureInferredDarkTable,
ensureFlowPacketsTable,
ensureSmartMoneyEventsTable,
insertAlert,
insertClassifierHit,
insertEquityPrintJoin,
insertInferredDark,
insertFlowPacket,
insertSmartMoneyEvent
ClickHouseBatchWriter,
enqueueAlertInsert,
enqueueClassifierHitInsert,
enqueueEquityPrintJoinInsert,
enqueueFlowPacketInsert,
enqueueInferredDarkInsert,
enqueueSmartMoneyEventInsert,
} from "@islandflow/storage";
import {
AlertEventSchema,
@ -82,7 +84,12 @@ import {
type DarkInferenceConfig
} from "./dark-inference";
import { buildEquityPrintJoin, type EquityQuoteJoin } from "./equity-joins";
import { createRedisClient, updateRollingStats, type RollingStatsConfig } from "./rolling-stats";
import {
createRedisClient,
RollingWindowStore,
type RollingStatsConfig,
type RollingWindowStoreConfig
} from "./rolling-stats";
import { summarizeStructure, type ContractLeg } from "./structures";
import {
buildStructureFlowPacket,
@ -103,6 +110,8 @@ const envSchema = z.object({
CLUSTER_WINDOW_MS: z.coerce.number().int().positive().default(500),
ROLLING_WINDOW_SIZE: z.coerce.number().int().positive().default(50),
ROLLING_TTL_SEC: z.coerce.number().int().nonnegative().default(86400),
ROLLING_CACHE_FLUSH_INTERVAL_MS: z.coerce.number().int().positive().default(30_000),
ROLLING_CACHE_MAX_KEYS: z.coerce.number().int().positive().default(20_000),
COMPUTE_DELIVER_POLICY: z.enum(["new", "all", "last", "last_per_subject"]).default("new"),
COMPUTE_CONSUMER_RESET: z
.preprocess((value) => {
@ -119,6 +128,8 @@ const envSchema = z.object({
}, z.boolean())
.default(false),
NBBO_MAX_AGE_MS: z.coerce.number().int().positive().default(1000),
COMPUTE_NBBO_CACHE_MAX_KEYS: z.coerce.number().int().positive().default(20_000),
COMPUTE_NBBO_CACHE_TTL_MS: z.coerce.number().int().positive().default(900_000),
EQUITY_QUOTE_MAX_AGE_MS: z.coerce.number().int().positive().default(1000),
DARK_INFER_WINDOW_MS: z.coerce.number().int().positive().default(60000),
DARK_INFER_COOLDOWN_MS: z.coerce.number().int().nonnegative().default(30000),
@ -269,6 +280,9 @@ const clusters = new Map<string, ClusterState>();
const nbboCache = new Map<string, OptionNBBO>();
const equityQuoteCache = new Map<string, EquityQuote>();
const darkInferenceState = createDarkInferenceState();
const nbboCacheTouchedAt = new Map<string, number>();
const equityQuoteCacheTouchedAt = new Map<string, number>();
const darkInferenceTouchedAt = new Map<string, number>();
const recentLegsByKey = new Map<string, LegEvidence[]>();
const recentLegsByRoot = new Map<string, LegEvidence[]>();
const recentStructureEmits = new Map<string, number>();
@ -278,6 +292,20 @@ const runtimeState = {
};
const MAX_RECENT_LEGS = 20;
const EQUITY_QUOTE_CACHE_MAX_KEYS = 2_000;
const EQUITY_QUOTE_CACHE_TTL_MS = 900_000;
const DARK_INFERENCE_TTL_MS = 900_000;
const CACHE_PRUNE_INTERVAL_MS = 60_000;
const emitCounters = {
flowPackets: 0,
structurePackets: 0,
smartMoneyEvents: 0,
classifierHits: 0,
alerts: 0,
equityJoins: 0,
darkEvents: 0
};
const rollingKey = (metric: string, contractId: string): string => {
return `rolling:${metric}:${contractId}`;
@ -479,8 +507,8 @@ const pruneRecentStructureEmits = (anchorTs: number): void => {
};
const emitStructurePacketIfNeeded = async (
clickhouse: ReturnType<typeof createClickHouseClient>,
js: Awaited<ReturnType<typeof connectJetStreamWithRetry>>["js"],
batchWriter: ClickHouseBatchWriter,
legs: LegEvidence[],
summary: ReturnType<typeof summarizeStructure>,
currentContractId: string
@ -512,16 +540,11 @@ const emitStructurePacketIfNeeded = async (
const packet = buildStructureFlowPacket(plan, summary);
const validated = FlowPacketSchema.parse(packet);
await insertFlowPacket(clickhouse, validated);
enqueueFlowPacketInsert(batchWriter, validated);
await publishJson(js, SUBJECT_FLOW_PACKETS, validated);
await emitClassifiers(clickhouse, js, validated);
logger.info("emitted structure flow packet", {
id: validated.id,
type: summary.type,
legs: summary.legs,
strikes: summary.strikes
});
emitCounters.flowPackets += 1;
emitCounters.structurePackets += 1;
await emitClassifiers(js, batchWriter, validated);
};
const applyDeliverPolicy = (
@ -606,6 +629,7 @@ const updateNbboCache = (nbbo: OptionNBBO): void => {
(nbbo.ts === existing.ts && nbbo.seq >= existing.seq)
) {
nbboCache.set(nbbo.option_contract_id, nbbo);
nbboCacheTouchedAt.set(nbbo.option_contract_id, Date.now());
}
};
@ -617,6 +641,7 @@ const updateEquityQuoteCache = (quote: EquityQuote): void => {
(quote.ts === existing.ts && quote.seq >= existing.seq)
) {
equityQuoteCache.set(quote.underlying_id, quote);
equityQuoteCacheTouchedAt.set(quote.underlying_id, Date.now());
}
};
@ -626,6 +651,7 @@ const selectNbbo = (contractId: string, ts: number): NbboJoin => {
return { nbbo: null, ageMs: env.NBBO_MAX_AGE_MS + 1, stale: true };
}
nbboCacheTouchedAt.set(contractId, Date.now());
const ageMs = Math.abs(ts - nbbo.ts);
const stale = ageMs > env.NBBO_MAX_AGE_MS;
return { nbbo, ageMs, stale };
@ -637,11 +663,77 @@ const selectEquityQuote = (underlyingId: string, ts: number): EquityQuoteJoin =>
return { quote: null, ageMs: env.EQUITY_QUOTE_MAX_AGE_MS + 1, stale: true };
}
equityQuoteCacheTouchedAt.set(underlyingId, Date.now());
const ageMs = Math.abs(ts - quote.ts);
const stale = ageMs > env.EQUITY_QUOTE_MAX_AGE_MS;
return { quote, ageMs, stale };
};
const pruneTimedMap = <T>(
values: Map<string, T>,
touchedAt: Map<string, number>,
maxKeys: number,
ttlMs: number,
now = Date.now()
): number => {
let removed = 0;
for (const [key, touched] of touchedAt) {
if (now - touched > ttlMs) {
touchedAt.delete(key);
values.delete(key);
removed += 1;
}
}
if (values.size <= maxKeys) {
return removed;
}
const overflow = values.size - maxKeys;
const oldest = [...touchedAt.entries()].sort((a, b) => a[1] - b[1]).slice(0, overflow);
for (const [key] of oldest) {
touchedAt.delete(key);
values.delete(key);
removed += 1;
}
return removed;
};
const pruneComputeCaches = (rollingStore: RollingWindowStore, now = Date.now()) => {
const nbboRemoved = pruneTimedMap(
nbboCache,
nbboCacheTouchedAt,
env.COMPUTE_NBBO_CACHE_MAX_KEYS,
env.COMPUTE_NBBO_CACHE_TTL_MS,
now
);
const quoteRemoved = pruneTimedMap(
equityQuoteCache,
equityQuoteCacheTouchedAt,
EQUITY_QUOTE_CACHE_MAX_KEYS,
EQUITY_QUOTE_CACHE_TTL_MS,
now
);
const darkRemoved = pruneTimedMap(
darkInferenceState.lastEmittedByUnderlying,
darkInferenceTouchedAt,
EQUITY_QUOTE_CACHE_MAX_KEYS,
DARK_INFERENCE_TTL_MS,
now
);
const rollingRemoved = rollingStore.prune(now);
logger.info("compute cache summary", {
nbbo_cache_size: nbboCache.size,
equity_quote_cache_size: equityQuoteCache.size,
dark_inference_cache_size: darkInferenceState.lastEmittedByUnderlying.size,
rolling_cache_size: rollingStore.size,
removed: nbboRemoved + quoteRemoved + darkRemoved + rollingRemoved
});
};
const classifyPlacement = (price: number, join: NbboJoin): NbboPlacement => {
if (!Number.isFinite(price)) {
return "MISSING";
@ -679,10 +771,9 @@ const classifyPlacement = (price: number, join: NbboJoin): NbboPlacement => {
};
const flushCluster = async (
clickhouse: ReturnType<typeof createClickHouseClient>,
js: Awaited<ReturnType<typeof connectJetStreamWithRetry>>["js"],
redis: ReturnType<typeof createRedisClient>,
rollingConfig: RollingStatsConfig,
batchWriter: ClickHouseBatchWriter,
rollingStore: RollingWindowStore,
cluster: ClusterState
): Promise<void> => {
if (cluster.flushed) {
@ -784,12 +875,7 @@ const flushCluster = async (
prefix: string
): Promise<void> => {
try {
const snapshot = await updateRollingStats(
redis,
rollingKey(metric, cluster.contractId),
value,
rollingConfig
);
const snapshot = rollingStore.update(rollingKey(metric, cluster.contractId), value);
features[`${prefix}_mean`] = roundTo(snapshot.mean);
features[`${prefix}_std`] = roundTo(snapshot.stddev);
features[`${prefix}_z`] = roundTo(snapshot.zscore);
@ -824,7 +910,7 @@ const flushCluster = async (
features.structure_rights = summary.rights;
}
await emitStructurePacketIfNeeded(clickhouse, js, legs, summary, currentLeg.contractId);
await emitStructurePacketIfNeeded(js, batchWriter, legs, summary, currentLeg.contractId);
const rootKey = buildRootKey(currentLeg);
const rootCandidates = [
@ -834,7 +920,7 @@ const flushCluster = async (
const rollLegs = [currentLeg, ...rootCandidates];
const rollSummary = summarizeStructure(rollLegs);
if (rollSummary?.type === "roll") {
await emitStructurePacketIfNeeded(clickhouse, js, rollLegs, rollSummary, currentLeg.contractId);
await emitStructurePacketIfNeeded(js, batchWriter, rollLegs, rollSummary, currentLeg.contractId);
}
storeRecentLeg(currentLeg, anchorTs);
@ -873,16 +959,10 @@ const flushCluster = async (
const validated = FlowPacketSchema.parse(packet);
try {
await insertFlowPacket(clickhouse, validated);
enqueueFlowPacketInsert(batchWriter, validated);
await publishJson(js, SUBJECT_FLOW_PACKETS, validated);
await emitClassifiers(clickhouse, js, validated);
logger.info("emitted flow packet", {
id: validated.id,
contract: cluster.contractId,
count: cluster.members.length
});
emitCounters.flowPackets += 1;
await emitClassifiers(js, batchWriter, validated);
} catch (error) {
if (isExpectedShutdownNatsError(error)) {
logger.info("skipped flow packet publish during shutdown", {
@ -899,8 +979,8 @@ const flushCluster = async (
};
const emitClassifiers = async (
clickhouse: ReturnType<typeof createClickHouseClient>,
js: Awaited<ReturnType<typeof connectJetStreamWithRetry>>["js"],
batchWriter: ClickHouseBatchWriter,
packet: FlowPacket
): Promise<void> => {
let smartMoneyEvent: SmartMoneyEvent;
@ -915,8 +995,9 @@ const emitClassifiers = async (
: packet.source_ts;
const eventCalendarMatch = underlyingId ? eventCalendarProvider.findNextEvent(underlyingId, referenceTs) : null;
smartMoneyEvent = SmartMoneyEventSchema.parse(buildSmartMoneyEventFromPacket(packet, { eventCalendarMatch }));
await insertSmartMoneyEvent(clickhouse, smartMoneyEvent);
enqueueSmartMoneyEventInsert(batchWriter, smartMoneyEvent);
await publishJson(js, SUBJECT_SMART_MONEY_EVENTS, smartMoneyEvent);
emitCounters.smartMoneyEvents += 1;
} catch (error) {
if (isExpectedShutdownNatsError(error)) {
return;
@ -945,8 +1026,9 @@ const emitClassifiers = async (
for (const hit of hitEvents) {
try {
await insertClassifierHit(clickhouse, hit);
enqueueClassifierHitInsert(batchWriter, hit);
await publishJson(js, SUBJECT_CLASSIFIER_HITS, hit);
emitCounters.classifierHits += 1;
} catch (error) {
if (isExpectedShutdownNatsError(error)) {
continue;
@ -981,8 +1063,9 @@ const emitClassifiers = async (
});
try {
await insertAlert(clickhouse, alert);
enqueueAlertInsert(batchWriter, alert);
await publishJson(js, SUBJECT_ALERTS, alert);
emitCounters.alerts += 1;
} catch (error) {
if (isExpectedShutdownNatsError(error)) {
return;
@ -995,17 +1078,21 @@ const emitClassifiers = async (
};
const emitEquityJoin = async (
clickhouse: ReturnType<typeof createClickHouseClient>,
js: Awaited<ReturnType<typeof connectJetStreamWithRetry>>["js"],
batchWriter: ClickHouseBatchWriter,
print: EquityPrint
): Promise<void> => {
const join = selectEquityQuote(print.underlying_id, print.ts);
const payload: EquityPrintJoin = EquityPrintJoinSchema.parse(buildEquityPrintJoin(print, join));
try {
await insertEquityPrintJoin(clickhouse, payload);
enqueueEquityPrintJoinInsert(batchWriter, payload);
} catch (error) {
logger.error("failed to emit equity print join", {
if (isExpectedShutdownNatsError(error)) {
return;
}
logger.error("failed to queue equity print join", {
error: error instanceof Error ? error.message : String(error),
trace_id: payload.trace_id
});
@ -1014,6 +1101,7 @@ const emitEquityJoin = async (
try {
await publishJson(js, SUBJECT_EQUITY_JOINS, payload);
emitCounters.equityJoins += 1;
} catch (error) {
if (isExpectedShutdownNatsError(error)) {
return;
@ -1024,20 +1112,26 @@ const emitEquityJoin = async (
});
}
await emitDarkInferences(clickhouse, js, payload);
await emitDarkInferences(js, batchWriter, payload);
};
const emitDarkInferences = async (
clickhouse: ReturnType<typeof createClickHouseClient>,
js: Awaited<ReturnType<typeof connectJetStreamWithRetry>>["js"],
batchWriter: ClickHouseBatchWriter,
join: EquityPrintJoin
): Promise<void> => {
const events = evaluateDarkInferences(join, darkInferenceConfig, darkInferenceState);
for (const event of events) {
const validated: InferredDarkEvent = InferredDarkEventSchema.parse(event);
try {
await insertInferredDark(clickhouse, validated);
enqueueInferredDarkInsert(batchWriter, validated);
await publishJson(js, SUBJECT_INFERRED_DARK, validated);
emitCounters.darkEvents += 1;
const underlyingId =
typeof join.features?.underlying_id === "string" ? join.features.underlying_id : null;
if (underlyingId) {
darkInferenceTouchedAt.set(underlyingId, Date.now());
}
} catch (error) {
if (isExpectedShutdownNatsError(error)) {
continue;
@ -1051,10 +1145,9 @@ const emitDarkInferences = async (
};
const flushEligibleClusters = async (
clickhouse: ReturnType<typeof createClickHouseClient>,
js: Awaited<ReturnType<typeof connectJetStreamWithRetry>>["js"],
redis: ReturnType<typeof createRedisClient>,
rollingConfig: RollingStatsConfig,
batchWriter: ClickHouseBatchWriter,
rollingStore: RollingWindowStore,
currentTs: number,
skipContractId: string
): Promise<void> => {
@ -1065,7 +1158,7 @@ const flushEligibleClusters = async (
if (currentTs - cluster.endTs > env.CLUSTER_WINDOW_MS) {
clusters.delete(contractId);
await flushCluster(clickhouse, js, redis, rollingConfig, cluster);
await flushCluster(js, batchWriter, rollingStore, cluster);
}
}
};
@ -1081,135 +1174,16 @@ const run = async () => {
{ attempts: 120, delayMs: 500 }
);
await ensureStream(jsm, {
name: STREAM_OPTION_SIGNAL_PRINTS,
subjects: [SUBJECT_OPTION_SIGNAL_PRINTS],
retention: "limits",
storage: "file",
discard: "old",
max_msgs_per_subject: -1,
max_msgs: -1,
max_bytes: -1,
max_age: 0,
num_replicas: 1
});
await ensureStream(jsm, {
name: STREAM_OPTION_NBBO,
subjects: [SUBJECT_OPTION_NBBO],
retention: "limits",
storage: "file",
discard: "old",
max_msgs_per_subject: -1,
max_msgs: -1,
max_bytes: -1,
max_age: 0,
num_replicas: 1
});
await ensureStream(jsm, {
name: STREAM_EQUITY_PRINTS,
subjects: [SUBJECT_EQUITY_PRINTS],
retention: "limits",
storage: "file",
discard: "old",
max_msgs_per_subject: -1,
max_msgs: -1,
max_bytes: -1,
max_age: 0,
num_replicas: 1
});
await ensureStream(jsm, {
name: STREAM_EQUITY_QUOTES,
subjects: [SUBJECT_EQUITY_QUOTES],
retention: "limits",
storage: "file",
discard: "old",
max_msgs_per_subject: -1,
max_msgs: -1,
max_bytes: -1,
max_age: 0,
num_replicas: 1
});
await ensureStream(jsm, {
name: STREAM_FLOW_PACKETS,
subjects: [SUBJECT_FLOW_PACKETS],
retention: "limits",
storage: "file",
discard: "old",
max_msgs_per_subject: -1,
max_msgs: -1,
max_bytes: -1,
max_age: 0,
num_replicas: 1
});
await ensureStream(jsm, {
name: STREAM_SMART_MONEY_EVENTS,
subjects: [SUBJECT_SMART_MONEY_EVENTS],
retention: "limits",
storage: "file",
discard: "old",
max_msgs_per_subject: -1,
max_msgs: -1,
max_bytes: -1,
max_age: 0,
num_replicas: 1
});
await ensureStream(jsm, {
name: STREAM_EQUITY_JOINS,
subjects: [SUBJECT_EQUITY_JOINS],
retention: "limits",
storage: "file",
discard: "old",
max_msgs_per_subject: -1,
max_msgs: -1,
max_bytes: -1,
max_age: 0,
num_replicas: 1
});
await ensureStream(jsm, {
name: STREAM_INFERRED_DARK,
subjects: [SUBJECT_INFERRED_DARK],
retention: "limits",
storage: "file",
discard: "old",
max_msgs_per_subject: -1,
max_msgs: -1,
max_bytes: -1,
max_age: 0,
num_replicas: 1
});
await ensureStream(jsm, {
name: STREAM_CLASSIFIER_HITS,
subjects: [SUBJECT_CLASSIFIER_HITS],
retention: "limits",
storage: "file",
discard: "old",
max_msgs_per_subject: -1,
max_msgs: -1,
max_bytes: -1,
max_age: 0,
num_replicas: 1
});
await ensureStream(jsm, {
name: STREAM_ALERTS,
subjects: [SUBJECT_ALERTS],
retention: "limits",
storage: "file",
discard: "old",
max_msgs_per_subject: -1,
max_msgs: -1,
max_bytes: -1,
max_age: 0,
num_replicas: 1
});
await ensureStream(jsm, buildStreamConfig(STREAM_OPTION_SIGNAL_PRINTS, SUBJECT_OPTION_SIGNAL_PRINTS, "derived"));
await ensureStream(jsm, buildStreamConfig(STREAM_OPTION_NBBO, SUBJECT_OPTION_NBBO, "raw"));
await ensureStream(jsm, buildStreamConfig(STREAM_EQUITY_PRINTS, SUBJECT_EQUITY_PRINTS, "raw"));
await ensureStream(jsm, buildStreamConfig(STREAM_EQUITY_QUOTES, SUBJECT_EQUITY_QUOTES, "raw"));
await ensureStream(jsm, buildStreamConfig(STREAM_FLOW_PACKETS, SUBJECT_FLOW_PACKETS, "derived"));
await ensureStream(jsm, buildStreamConfig(STREAM_SMART_MONEY_EVENTS, SUBJECT_SMART_MONEY_EVENTS, "derived"));
await ensureStream(jsm, buildStreamConfig(STREAM_EQUITY_JOINS, SUBJECT_EQUITY_JOINS, "derived"));
await ensureStream(jsm, buildStreamConfig(STREAM_INFERRED_DARK, SUBJECT_INFERRED_DARK, "derived"));
await ensureStream(jsm, buildStreamConfig(STREAM_CLASSIFIER_HITS, SUBJECT_CLASSIFIER_HITS, "derived"));
await ensureStream(jsm, buildStreamConfig(STREAM_ALERTS, SUBJECT_ALERTS, "derived"));
const clickhouse = createClickHouseClient({
url: env.CLICKHOUSE_URL,
@ -1242,6 +1216,51 @@ const run = async () => {
windowSize: env.ROLLING_WINDOW_SIZE,
ttlSeconds: env.ROLLING_TTL_SEC
};
const rollingStore = new RollingWindowStore({
...rollingConfig,
flushIntervalMs: env.ROLLING_CACHE_FLUSH_INTERVAL_MS,
maxKeys: env.ROLLING_CACHE_MAX_KEYS
} satisfies RollingWindowStoreConfig);
const batchWriter = new ClickHouseBatchWriter(clickhouse, {
flushIntervalMs: 100,
maxRows: 250,
onError: (table, error, rowCount) => {
logger.error("batched clickhouse insert failed", {
table,
row_count: rowCount,
error: error instanceof Error ? error.message : String(error),
action: "dropped"
});
}
});
const rollingFlushTimer = setInterval(() => {
void rollingStore.flushToRedis(redis);
}, env.ROLLING_CACHE_FLUSH_INTERVAL_MS);
const pruneTimer = setInterval(() => {
pruneComputeCaches(rollingStore);
}, CACHE_PRUNE_INTERVAL_MS);
const summaryTimer = setInterval(() => {
logger.info("compute minute summary", {
flow_packets_emitted: emitCounters.flowPackets,
structure_packets_emitted: emitCounters.structurePackets,
smart_money_events_emitted: emitCounters.smartMoneyEvents,
classifier_hits_emitted: emitCounters.classifierHits,
alerts_emitted: emitCounters.alerts,
equity_joins_emitted: emitCounters.equityJoins,
dark_events_emitted: emitCounters.darkEvents,
rolling_stats_cache_size: rollingStore.size
});
emitCounters.flowPackets = 0;
emitCounters.structurePackets = 0;
emitCounters.smartMoneyEvents = 0;
emitCounters.classifierHits = 0;
emitCounters.alerts = 0;
emitCounters.equityJoins = 0;
emitCounters.darkEvents = 0;
}, 60_000);
rollingFlushTimer.unref?.();
pruneTimer.unref?.();
summaryTimer.unref?.();
await retry("clickhouse table init", 120, 500, async () => {
await ensureFlowPacketsTable(clickhouse);
@ -1578,7 +1597,7 @@ const run = async () => {
try {
const print = EquityPrintSchema.parse(equitySubscription.decode(msg));
await emitEquityJoin(clickhouse, js, print);
await emitEquityJoin(js, batchWriter, print);
msg.ack();
} catch (error) {
logger.error("failed to process equity print", {
@ -1602,11 +1621,16 @@ const run = async () => {
runtimeState.shuttingDown = true;
runtimeState.shutdownPromise = (async () => {
logger.info("service stopping", { signal });
clearInterval(rollingFlushTimer);
clearInterval(pruneTimer);
clearInterval(summaryTimer);
for (const cluster of [...clusters.values()]) {
await flushCluster(clickhouse, js, redis, rollingConfig, cluster);
await flushCluster(js, batchWriter, rollingStore, cluster);
}
clusters.clear();
await batchWriter.close();
await rollingStore.flushToRedis(redis);
try {
await nc.drain();
@ -1655,10 +1679,9 @@ const run = async () => {
try {
const print = OptionPrintSchema.parse(subscription.decode(msg));
await flushEligibleClusters(
clickhouse,
js,
redis,
rollingConfig,
batchWriter,
rollingStore,
print.ts,
print.option_contract_id
);
@ -1674,7 +1697,7 @@ const run = async () => {
updateCluster(existing, print);
} else {
clusters.delete(print.option_contract_id);
await flushCluster(clickhouse, js, redis, rollingConfig, existing);
await flushCluster(js, batchWriter, rollingStore, existing);
clusters.set(print.option_contract_id, buildCluster(print));
}

143
services/compute/src/rolling-stats.ts
View file

@ -5,6 +5,11 @@ export type RollingStatsConfig = {
ttlSeconds: number;
};
export type RollingWindowStoreConfig = RollingStatsConfig & {
flushIntervalMs: number;
maxKeys: number;
};
export type RollingSnapshot = {
baselineCount: number;
mean: number;
@ -12,6 +17,12 @@ export type RollingSnapshot = {
zscore: number;
};
type RollingWindowEntry = {
values: number[];
updatedAt: number;
dirty: boolean;
};
const toNumbers = (values: string[]): number[] => {
return values
.map((value) => Number(value))
@ -49,26 +60,120 @@ export const createRedisClient = (url: string) => {
return createClient({ url });
};
export const updateRollingStats = async (
client: ReturnType<typeof createClient>,
key: string,
value: number,
config: RollingStatsConfig
): Promise<RollingSnapshot> => {
const limit = Math.max(0, config.windowSize - 1);
const existing = await client.lRange(key, 0, limit);
const baseline = toNumbers(existing);
const snapshot = computeSnapshot(baseline, value);
const getOldestKey = (store: Map<string, RollingWindowEntry>): string | null => {
let oldestKey: string | null = null;
let oldestUpdatedAt = Number.POSITIVE_INFINITY;
const multi = client.multi();
multi.lPush(key, value.toString());
if (config.windowSize > 0) {
multi.lTrim(key, 0, config.windowSize - 1);
for (const [key, entry] of store) {
if (entry.updatedAt < oldestUpdatedAt) {
oldestUpdatedAt = entry.updatedAt;
oldestKey = key;
}
}
if (config.ttlSeconds > 0) {
multi.expire(key, config.ttlSeconds);
}
await multi.exec();
return snapshot;
return oldestKey;
};
export class RollingWindowStore {
private readonly store = new Map<string, RollingWindowEntry>();
private readonly ttlMs: number;
private readonly windowSize: number;
private readonly maxKeys: number;
constructor(private readonly config: RollingWindowStoreConfig) {
this.ttlMs = Math.max(0, config.ttlSeconds * 1000);
this.windowSize = Math.max(1, config.windowSize);
this.maxKeys = Math.max(1, config.maxKeys);
}
get size(): number {
return this.store.size;
}
update(key: string, value: number, now = Date.now()): RollingSnapshot {
this.prune(now);
const existing = this.store.get(key);
const baseline = existing?.values ?? [];
const snapshot = computeSnapshot(baseline, value);
const nextValues = [value, ...baseline].slice(0, this.windowSize);
this.store.set(key, {
values: nextValues,
updatedAt: now,
dirty: true
});
this.enforceMaxKeys();
return snapshot;
}
prune(now = Date.now()): number {
if (this.ttlMs <= 0) {
return 0;
}
let removed = 0;
for (const [key, entry] of this.store) {
if (now - entry.updatedAt > this.ttlMs) {
this.store.delete(key);
removed += 1;
}
}
return removed;
}
async hydrateFromRedis(
client: ReturnType<typeof createClient>,
keys: string[],
now = Date.now()
): Promise<void> {
for (const key of keys) {
const values = toNumbers(await client.lRange(key, 0, this.windowSize - 1));
if (values.length === 0) {
continue;
}
this.store.set(key, {
values,
updatedAt: now,
dirty: false
});
}
this.enforceMaxKeys();
}
async flushToRedis(client: ReturnType<typeof createClient>): Promise<number> {
let flushed = 0;
for (const [key, entry] of this.store) {
if (!entry.dirty) {
continue;
}
const multi = client.multi();
multi.lTrim(key, 1, 0);
for (let idx = entry.values.length - 1; idx >= 0; idx -= 1) {
const value = entry.values[idx];
if (typeof value === "number" && Number.isFinite(value)) {
multi.lPush(key, value.toString());
}
}
if (this.config.ttlSeconds > 0) {
multi.expire(key, this.config.ttlSeconds);
}
await multi.exec();
entry.dirty = false;
flushed += 1;
}
return flushed;
}
private enforceMaxKeys(): void {
while (this.store.size > this.maxKeys) {
const oldestKey = getOldestKey(this.store);
if (!oldestKey) {
break;
}
this.store.delete(oldestKey);
}
}
}

16
services/compute/tests/rolling-stats.test.ts
View file

@ -1,5 +1,5 @@
import { describe, expect, test } from "bun:test";
import { computeSnapshot, computeStats } from "../src/rolling-stats";
import { computeSnapshot, computeStats, RollingWindowStore } from "../src/rolling-stats";
describe("rolling stats helpers", () => {
test("computeStats handles empty baseline", () => {
@ -21,4 +21,18 @@ describe("rolling stats helpers", () => {
expect(snapshot.baselineCount).toBe(3);
expect(snapshot.zscore).toBeCloseTo(1.84, 2);
});
test("RollingWindowStore prunes stale keys by ttl", () => {
const store = new RollingWindowStore({
windowSize: 3,
ttlSeconds: 1,
flushIntervalMs: 30_000,
maxKeys: 10
});
store.update("rolling:premium:ABC", 10, 0);
expect(store.size).toBe(1);
expect(store.prune(1_500)).toBe(1);
expect(store.size).toBe(0);
});
});

33
services/ingest-equities/src/index.ts
View file

@ -5,6 +5,7 @@ import {
SUBJECT_EQUITY_QUOTES,
STREAM_EQUITY_PRINTS,
STREAM_EQUITY_QUOTES,
buildStreamConfig,
connectJetStreamWithRetry,
ensureStream,
publishJson
@ -194,31 +195,8 @@ const run = async () => {
{ attempts: 120, delayMs: 500 }
);
await ensureStream(jsm, {
name: STREAM_EQUITY_PRINTS,
subjects: [SUBJECT_EQUITY_PRINTS],
retention: "limits",
storage: "file",
discard: "old",
max_msgs_per_subject: -1,
max_msgs: -1,
max_bytes: -1,
max_age: 0,
num_replicas: 1
});
await ensureStream(jsm, {
name: STREAM_EQUITY_QUOTES,
subjects: [SUBJECT_EQUITY_QUOTES],
retention: "limits",
storage: "file",
discard: "old",
max_msgs_per_subject: -1,
max_msgs: -1,
max_bytes: -1,
max_age: 0,
num_replicas: 1
});
await ensureStream(jsm, buildStreamConfig(STREAM_EQUITY_PRINTS, SUBJECT_EQUITY_PRINTS, "raw"));
await ensureStream(jsm, buildStreamConfig(STREAM_EQUITY_QUOTES, SUBJECT_EQUITY_QUOTES, "raw"));
const clickhouse = createClickHouseClient({
url: env.CLICKHOUSE_URL,
@ -251,11 +229,6 @@ const run = async () => {
try {
await insertEquityPrint(clickhouse, print);
await publishJson(js, SUBJECT_EQUITY_PRINTS, print);
logger.info("published equity print", {
trace_id: print.trace_id,
seq: print.seq,
underlying_id: print.underlying_id
});
} catch (error) {
if (isExpectedShutdownError(error)) {
return;

119
services/ingest-options/src/index.ts
View file

@ -9,6 +9,7 @@ import {
STREAM_OPTION_NBBO,
STREAM_OPTION_PRINTS,
STREAM_OPTION_SIGNAL_PRINTS,
buildStreamConfig,
buildDurableConsumer,
connectJetStreamWithRetry,
ensureStream,
@ -109,7 +110,9 @@ const envSchema = z.object({
return value;
}, z.boolean())
.default(false),
TESTING_THROTTLE_MS: z.coerce.number().int().nonnegative().default(200)
TESTING_THROTTLE_MS: z.coerce.number().int().nonnegative().default(200),
OPTION_CONTEXT_MAX_KEYS: z.coerce.number().int().positive().default(20_000),
OPTION_CONTEXT_TTL_MS: z.coerce.number().int().positive().default(900_000)
});
const env = readEnv(envSchema);
@ -143,6 +146,44 @@ const state = {
const nbboHistoryByContract: ContextHistory<OptionNBBO> = new Map();
const equityQuoteHistoryByUnderlying: ContextHistory<EquityQuote> = new Map();
const OPTION_CONTEXT_PRUNE_INTERVAL_MS = 60_000;
const pruneContextHistory = <T extends { ts: number }>(
history: ContextHistory<T>,
maxKeys: number,
ttlMs: number,
now = Date.now()
): number => {
let removed = 0;
for (const [key, items] of history) {
const filtered = items.filter((item) => now - item.ts <= ttlMs);
if (filtered.length === 0) {
history.delete(key);
removed += 1;
continue;
}
if (filtered.length !== items.length) {
history.set(key, filtered);
}
}
if (history.size <= maxKeys) {
return removed;
}
const overflow = history.size - maxKeys;
const oldestKeys = [...history.entries()]
.map(([key, items]) => [key, items.at(-1)?.ts ?? Number.NEGATIVE_INFINITY] as const)
.sort((a, b) => a[1] - b[1])
.slice(0, overflow);
for (const [key] of oldestKeys) {
history.delete(key);
removed += 1;
}
return removed;
};
const getErrorMessage = (error: unknown): string => {
return error instanceof Error ? error.message : String(error);
@ -305,57 +346,10 @@ const run = async () => {
{ attempts: 120, delayMs: 500 }
);
await ensureStream(jsm, {
name: STREAM_OPTION_PRINTS,
subjects: [SUBJECT_OPTION_PRINTS],
retention: "limits",
storage: "file",
discard: "old",
max_msgs_per_subject: -1,
max_msgs: -1,
max_bytes: -1,
max_age: 0,
num_replicas: 1
});
await ensureStream(jsm, {
name: STREAM_OPTION_NBBO,
subjects: [SUBJECT_OPTION_NBBO],
retention: "limits",
storage: "file",
discard: "old",
max_msgs_per_subject: -1,
max_msgs: -1,
max_bytes: -1,
max_age: 0,
num_replicas: 1
});
await ensureStream(jsm, {
name: STREAM_OPTION_SIGNAL_PRINTS,
subjects: [SUBJECT_OPTION_SIGNAL_PRINTS],
retention: "limits",
storage: "file",
discard: "old",
max_msgs_per_subject: -1,
max_msgs: -1,
max_bytes: -1,
max_age: 0,
num_replicas: 1
});
await ensureStream(jsm, {
name: STREAM_EQUITY_QUOTES,
subjects: [SUBJECT_EQUITY_QUOTES],
retention: "limits",
storage: "file",
discard: "old",
max_msgs_per_subject: -1,
max_msgs: -1,
max_bytes: -1,
max_age: 0,
num_replicas: 1
});
await ensureStream(jsm, buildStreamConfig(STREAM_OPTION_PRINTS, SUBJECT_OPTION_PRINTS, "raw"));
await ensureStream(jsm, buildStreamConfig(STREAM_OPTION_NBBO, SUBJECT_OPTION_NBBO, "raw"));
await ensureStream(jsm, buildStreamConfig(STREAM_OPTION_SIGNAL_PRINTS, SUBJECT_OPTION_SIGNAL_PRINTS, "derived"));
await ensureStream(jsm, buildStreamConfig(STREAM_EQUITY_QUOTES, SUBJECT_EQUITY_QUOTES, "raw"));
const clickhouse = createClickHouseClient({
url: env.CLICKHOUSE_URL,
@ -400,14 +394,6 @@ const run = async () => {
if (print.signal_pass) {
await publishJson(js, SUBJECT_OPTION_SIGNAL_PRINTS, print);
}
logger.info("published option print", {
trace_id: print.trace_id,
seq: print.seq,
option_contract_id: print.option_contract_id,
signal_pass: print.signal_pass,
nbbo_side: print.nbbo_side,
notional: print.notional
});
} catch (error) {
if (isExpectedShutdownError(error)) {
return;
@ -475,6 +461,18 @@ const run = async () => {
}
})();
const pruneTimer = setInterval(() => {
const removed =
pruneContextHistory(nbboHistoryByContract, env.OPTION_CONTEXT_MAX_KEYS, env.OPTION_CONTEXT_TTL_MS) +
pruneContextHistory(equityQuoteHistoryByUnderlying, env.OPTION_CONTEXT_MAX_KEYS, env.OPTION_CONTEXT_TTL_MS);
logger.info("option context cache summary", {
nbbo_context_keys: nbboHistoryByContract.size,
equity_quote_context_keys: equityQuoteHistoryByUnderlying.size,
removed
});
}, OPTION_CONTEXT_PRUNE_INTERVAL_MS);
pruneTimer.unref?.();
const shutdown = async (signal: string) => {
if (state.shutdownPromise) {
return state.shutdownPromise;
@ -483,6 +481,7 @@ const run = async () => {
state.shuttingDown = true;
state.shutdownPromise = (async () => {
logger.info("service stopping", { signal });
clearInterval(pruneTimer);
await stopAdapter();
try {

18
services/replay/src/index.ts
View file

@ -11,6 +11,7 @@ import {
STREAM_OPTION_NBBO,
STREAM_OPTION_PRINTS,
STREAM_OPTION_SIGNAL_PRINTS,
buildStreamConfig,
connectJetStreamWithRetry,
ensureStream,
publishJson
@ -180,19 +181,6 @@ const parseStreamList = (value: string): ReplayStreamKind[] => {
return result;
};
const buildStreamConfig = (name: string, subject: string) => ({
name,
subjects: [subject],
retention: "limits",
storage: "file",
discard: "old",
max_msgs_per_subject: -1,
max_msgs: -1,
max_bytes: -1,
max_age: 0,
num_replicas: 1
});
const buildStartCursor = (startTs: number): ReplayCursor => {
if (startTs <= 0) {
return { ts: 0, seq: 0 };
@ -304,10 +292,10 @@ const run = async () => {
for (const kind of streamKinds) {
const def = STREAM_DEFS[kind];
await ensureStream(jsm, buildStreamConfig(def.streamName, def.subject));
await ensureStream(jsm, buildStreamConfig(def.streamName, def.subject, "raw"));
}
if (streamKinds.includes("options")) {
await ensureStream(jsm, buildStreamConfig(STREAM_OPTION_SIGNAL_PRINTS, SUBJECT_OPTION_SIGNAL_PRINTS));
await ensureStream(jsm, buildStreamConfig(STREAM_OPTION_SIGNAL_PRINTS, SUBJECT_OPTION_SIGNAL_PRINTS, "derived"));
}
const clickhouse = createClickHouseClient({