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

Restore continuous live tape history

Browse source
This commit is contained in:
dirtydishes 2026-05-07 00:39:26 -04:00
parent d81b4c0cfb
commit 034d24f8ac
5 changed files with 483 additions and 117 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -112,7 +112,7 @@ import {
} from "@islandflow/types";
import { createClient } from "redis";
import { z } from "zod";
import { LIVE_FEED_LOOKBACK_MS, LiveStateManager, shouldFanoutLiveEvent } from "./live";
import { LiveStateManager, shouldFanoutLiveEvent } from "./live";
const service = "api";
const logger = createLogger({ service });
@ -617,14 +617,12 @@ const parseLiveOptionPrintFilters = (url: URL): OptionPrintQueryFilters => {
return {
...storageFilters,
underlyingIds: parseScopeList(url, "underlying_id", "underlying_ids"),
optionContractId: url.searchParams.get("option_contract_id") ?? undefined,
sinceTs: Date.now() - LIVE_FEED_LOOKBACK_MS
optionContractId: url.searchParams.get("option_contract_id") ?? undefined
};
};
const parseLiveEquityPrintFilters = (url: URL): EquityPrintQueryFilters => ({
underlyingIds: parseScopeList(url, "underlying_id", "underlying_ids"),
sinceTs: Date.now() - LIVE_FEED_LOOKBACK_MS
underlyingIds: parseScopeList(url, "underlying_id", "underlying_ids")
});
const matchesScopedOptionSubscription = (

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

@ -408,13 +408,7 @@ export class LiveStateManager {
const config = this.generic[channel];
if (this.redis?.isOpen) {
const payloads = await this.redis.lRange(config.redisKey, 0, config.limit - 1);
const cached = normalizeGenericItems(
channel,
parseJsonList(payloads, config.parse).filter((item) =>
isWithinLiveFeedLookback(channel, item)
),
config
);
const cached = normalizeGenericItems(channel, parseJsonList(payloads, config.parse), config);
if (cached.length > 0) {
this.genericItems.set(channel, cached);
this.stats.genericHydrateFromRedis += 1;
@ -434,9 +428,7 @@ export class LiveStateManager {
const fresh = normalizeGenericItems(
channel,
(await config.fetchRecent(this.clickhouse, config.limit)).filter((item) =>
isWithinLiveFeedLookback(channel, item)
),
await config.fetchRecent(this.clickhouse, config.limit),
config
);
this.stats.genericHydrateFromClickHouse += 1;
@ -467,8 +459,7 @@ export class LiveStateManager {
optionTypes: subscription.filters?.optionTypes,
minNotional: subscription.filters?.minNotional,
underlyingIds: subscription.underlying_ids,
optionContractId: subscription.option_contract_id,
sinceTs: Date.now() - LIVE_FEED_LOOKBACK_MS
optionContractId: subscription.option_contract_id
};
const items = await fetchRecentOptionPrints(
this.clickhouse,
@ -487,7 +478,6 @@ export class LiveStateManager {
const config = this.generic.options;
const limit = snapshotLimitFor(subscription, config.limit);
const items = (this.genericItems.get("options") ?? []).filter((item) =>
isWithinLiveFeedLookback("options", item) &&
matchesOptionPrintFilters(item, subscription.filters)
).slice(0, limit);
return {
@ -501,7 +491,6 @@ export class LiveStateManager {
const config = this.generic.flow;
const limit = snapshotLimitFor(subscription, config.limit);
const items = (this.genericItems.get("flow") ?? []).filter((item) =>
isWithinLiveFeedLookback("flow", item) &&
matchesFlowPacketFilters(item, subscription.filters)
).slice(0, limit);
return {
@ -516,8 +505,7 @@ export class LiveStateManager {
const limit = snapshotLimitFor(subscription, config.limit);
if (subscription.underlying_ids?.length) {
const filters: EquityPrintQueryFilters = {
underlyingIds: subscription.underlying_ids,
sinceTs: Date.now() - LIVE_FEED_LOOKBACK_MS
underlyingIds: subscription.underlying_ids
};
const items = await fetchRecentEquityPrints(this.clickhouse, limit, filters);
return {
@ -527,9 +515,7 @@ export class LiveStateManager {
next_before: nextBeforeForItems(items, config.cursor)
};
}
const items = (this.genericItems.get("equities") ?? []).filter((item) =>
isWithinLiveFeedLookback("equities", item)
).slice(0, limit);
const items = (this.genericItems.get("equities") ?? []).slice(0, limit);
return {
subscription,
items,
@ -568,9 +554,7 @@ export class LiveStateManager {
default: {
const config = this.generic[subscription.channel];
const limit = snapshotLimitFor(subscription, config.limit);
const items = (this.genericItems.get(subscription.channel) ?? []).filter((item) =>
isWithinLiveFeedLookback(subscription.channel, item)
).slice(0, limit);
const items = (this.genericItems.get(subscription.channel) ?? []).slice(0, limit);
return {
subscription,
items,

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

@ -7,15 +7,17 @@ import {
shouldFanoutLiveEvent
} from "../src/live";
const makeClickHouse = (): ClickHouseClient =>
const makeClickHouse = (
queryResolver?: (query: string) => unknown[]
): ClickHouseClient =>
({
exec: async () => {},
insert: async () => {},
ping: async () => ({ success: true }),
close: async () => {},
query: async () => ({
query: async ({ query }: { query: string }) => ({
async json<T>() {
return [] as T;
return (queryResolver?.(query) ?? []) as T;
}
})
}) as ClickHouseClient;
@ -408,6 +410,160 @@ describe("LiveStateManager", () => {
]);
});
it("seeds scoped option snapshots from clickhouse rows older than 24h", async () => {
const now = Date.now();
const staleTs = now - 25 * 60 * 60 * 1000;
const manager = new LiveStateManager(
makeClickHouse((query) =>
query.includes("FROM option_prints")
? [
{
source_ts: staleTs,
ingest_ts: staleTs + 1,
seq: 1,
trace_id: "opt-ancient",
ts: staleTs,
option_contract_id: "AAPL-2025-01-17-200-C",
underlying_id: "AAPL",
price: 1,
size: 10,
exchange: "X",
signal_pass: true
}
]
: []
),
null
);
const snapshot = await manager.getSnapshot({
channel: "options",
underlying_ids: ["AAPL"],
option_contract_id: "AAPL-2025-01-17-200-C"
});
expect((snapshot.items as Array<{ trace_id: string }>).map((item) => item.trace_id)).toEqual([
"opt-ancient"
]);
expect(snapshot.next_before).toEqual({ ts: staleTs, seq: 1 });
expect(isLiveItemFresh("options", snapshot.items[0], now)).toBe(false);
});
it("seeds scoped equity snapshots from clickhouse rows older than 24h", async () => {
const now = Date.now();
const staleTs = now - 25 * 60 * 60 * 1000;
const manager = new LiveStateManager(
makeClickHouse((query) =>
query.includes("FROM equity_prints")
? [
{
source_ts: staleTs,
ingest_ts: staleTs + 1,
seq: 1,
trace_id: "eq-ancient",
ts: staleTs,
underlying_id: "AAPL",
price: 100,
size: 10,
exchange: "X",
offExchangeFlag: false
}
]
: []
),
null
);
const snapshot = await manager.getSnapshot({
channel: "equities",
underlying_ids: ["AAPL"]
});
expect((snapshot.items as Array<{ trace_id: string }>).map((item) => item.trace_id)).toEqual([
"eq-ancient"
]);
expect(snapshot.next_before).toEqual({ ts: staleTs, seq: 1 });
expect(isLiveItemFresh("equities", snapshot.items[0], now)).toBe(false);
});
it("hydrates retained rows older than 24h into generic live snapshots and keeps them stale", async () => {
const redis = makeRedis();
const now = Date.now();
const staleTs = now - 25 * 60 * 60 * 1000;
await redis.lPush(
"live:options",
JSON.stringify({
source_ts: staleTs,
ingest_ts: staleTs + 1,
seq: 1,
trace_id: "opt-retained",
ts: staleTs,
option_contract_id: "AAPL-2025-01-17-200-C",
underlying_id: "AAPL",
price: 1,
size: 10,
exchange: "X",
signal_pass: true
})
);
await redis.hSet("live:cursors", "options", JSON.stringify({ ts: staleTs, seq: 1 }));
await redis.lPush(
"live:equities",
JSON.stringify({
source_ts: staleTs,
ingest_ts: staleTs + 1,
seq: 2,
trace_id: "eq-retained",
ts: staleTs,
underlying_id: "AAPL",
price: 100,
size: 10,
exchange: "X",
offExchangeFlag: false
})
);
await redis.hSet("live:cursors", "equities", JSON.stringify({ ts: staleTs, seq: 2 }));
await redis.lPush(
"live:flow",
JSON.stringify({
source_ts: staleTs,
ingest_ts: staleTs + 1,
seq: 3,
trace_id: "flow-retained",
id: "flow-retained",
members: ["opt-retained"],
features: {},
join_quality: {}
})
);
await redis.hSet("live:cursors", "flow", JSON.stringify({ ts: staleTs, seq: 3 }));
const manager = new LiveStateManager(makeClickHouse(), redis as never);
await manager.hydrate();
const [optionsSnapshot, equitiesSnapshot, flowSnapshot] = await Promise.all([
manager.getSnapshot({ channel: "options" }),
manager.getSnapshot({ channel: "equities" }),
manager.getSnapshot({ channel: "flow" })
]);
expect((optionsSnapshot.items as Array<{ trace_id: string }>).map((item) => item.trace_id)).toEqual([
"opt-retained"
]);
expect((equitiesSnapshot.items as Array<{ trace_id: string }>).map((item) => item.trace_id)).toEqual([
"eq-retained"
]);
expect((flowSnapshot.items as Array<{ id: string }>).map((item) => item.id)).toEqual([
"flow-retained"
]);
expect(isLiveItemFresh("options", optionsSnapshot.items[0], now)).toBe(false);
expect(isLiveItemFresh("equities", equitiesSnapshot.items[0], now)).toBe(false);
expect(isLiveItemFresh("flow", flowSnapshot.items[0], now)).toBe(false);
});
it("keeps only the newest NBBO quote per contract across hydrate and ingest", async () => {
const redis = makeRedis();
const now = Date.now();