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

Fix contract-focused options tape hydration

Browse source
This commit is contained in:
dirtydishes 2026-05-07 23:37:32 -04:00
parent 73e25ddf70
commit b73e62bdba
8 changed files with 657 additions and 171 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -82,7 +82,7 @@ import {
fetchClassifierHitsByPacketIds,
fetchRecentOptionPrints
} from "@islandflow/storage";
import type { EquityPrintQueryFilters, OptionPrintQueryFilters } from "@islandflow/storage";
import type { EquityPrintQueryFilters } from "@islandflow/storage";
import {
AlertEventSchema,
ClassifierHitEventSchema,
@ -99,11 +99,6 @@ import {
LiveSubscriptionSchema,
matchesFlowPacketFilters,
matchesOptionPrintFilters,
OptionFlowFilters,
OptionFlowViewSchema,
OptionNbboSideSchema,
OptionSecurityTypeSchema,
OptionTypeSchema,
FlowPacketSchema,
SmartMoneyEventSchema,
OptionNBBOSchema,
@ -113,6 +108,7 @@ import {
import { createClient } from "redis";
import { z } from "zod";
import { HOT_LIVE_REDIS_KEYS, LiveStateManager, shouldFanoutLiveEvent } from "./live";
import { parseOptionPrintQuery } from "./option-queries";
const service = "api";
const logger = createLogger({ service });
@ -224,33 +220,6 @@ const equityPrintRangeSchema = z.object({
end_ts: z.coerce.number().int().nonnegative(),
limit: limitSchema.optional()
});
const optionSideListSchema = z
.string()
.transform((value) =>
value
.split(",")
.map((entry) => entry.trim())
.filter(Boolean)
)
.pipe(z.array(OptionNbboSideSchema));
const optionTypeListSchema = z
.string()
.transform((value) =>
value
.split(",")
.map((entry) => entry.trim())
.filter(Boolean)
)
.pipe(z.array(OptionTypeSchema));
const optionSecuritySchema = z.enum(["stock", "etf", "all"]);
const optionFilterQuerySchema = z.object({
view: OptionFlowViewSchema.optional(),
security: optionSecuritySchema.optional(),
side: optionSideListSchema.optional(),
type: optionTypeListSchema.optional(),
min_notional: z.coerce.number().nonnegative().optional()
});
type Channel =
| "options"
| "options-nbbo"
@ -351,43 +320,6 @@ const applyDeliverPolicy = (
}
};
const parseOptionPrintFilters = (
url: URL
): {
view: z.infer<typeof OptionFlowViewSchema>;
storageFilters: Parameters<typeof fetchRecentOptionPrints>[3];
liveFilters: OptionFlowFilters;
} => {
const parsed = optionFilterQuerySchema.parse({
view: url.searchParams.get("view") ?? undefined,
security: url.searchParams.get("security") ?? undefined,
side: url.searchParams.get("side") ?? undefined,
type: url.searchParams.get("type") ?? undefined,
min_notional: url.searchParams.get("min_notional") ?? undefined
});
const view = parsed.view ?? "signal";
const security = parsed.security ?? (view === "raw" ? "all" : "stock");
const storageFilters = {
view,
security,
minNotional: parsed.min_notional,
nbboSides: parsed.side,
optionTypes: parsed.type
} as const;
const liveFilters: OptionFlowFilters = {
view,
securityTypes:
security === "all"
? undefined
: ([security] as Array<z.infer<typeof OptionSecurityTypeSchema>>),
nbboSides: parsed.side,
optionTypes: parsed.type,
minNotional: parsed.min_notional
};
return { view, storageFilters, liveFilters };
};
const parseReplayParams = (url: URL): { afterTs: number; afterSeq: number; limit: number } => {
const params = replayParamsSchema.parse({
after_ts: url.searchParams.get("after_ts") ?? undefined,
@ -605,15 +537,6 @@ const parseScopeList = (url: URL, ...keys: string[]): string[] | undefined => {
return unique.length > 0 ? unique : undefined;
};
const parseLiveOptionPrintFilters = (url: URL): OptionPrintQueryFilters => {
const { storageFilters } = parseOptionPrintFilters(url);
return {
...storageFilters,
underlyingIds: parseScopeList(url, "underlying_id", "underlying_ids"),
optionContractId: url.searchParams.get("option_contract_id") ?? undefined
};
};
const parseLiveEquityPrintFilters = (url: URL): EquityPrintQueryFilters => ({
underlyingIds: parseScopeList(url, "underlying_id", "underlying_ids")
});
@ -1399,7 +1322,7 @@ const run = async () => {
try {
const limit = parseLimit(url.searchParams.get("limit"));
const source = parseReplaySource(url) ?? undefined;
const { storageFilters } = parseOptionPrintFilters(url);
const { storageFilters } = parseOptionPrintQuery(url);
const data = await fetchRecentOptionPrints(clickhouse, limit, source, storageFilters);
return jsonResponse({ data });
} catch (error) {
@ -1525,7 +1448,7 @@ const run = async () => {
try {
const { beforeTs, beforeSeq, limit } = parseBeforeParams(url);
const source = parseReplaySource(url) ?? undefined;
const storageFilters = parseLiveOptionPrintFilters(url);
const { storageFilters } = parseOptionPrintQuery(url);
const data = await fetchOptionPrintsBefore(
clickhouse,
beforeTs,
@ -1668,7 +1591,7 @@ const run = async () => {
try {
const { afterTs, afterSeq, limit } = parseReplayParams(url);
const source = parseReplaySource(url) ?? undefined;
const { storageFilters } = parseOptionPrintFilters(url);
const { storageFilters } = parseOptionPrintQuery(url);
const data = await fetchOptionPrintsAfter(
clickhouse,
afterTs,

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

@ -345,6 +345,30 @@ const snapshotLimitFor = (subscription: LiveSubscription, configuredLimit: numbe
return Math.max(1, Math.min(configuredLimit, Math.floor(requested)));
};
export const buildOptionSnapshotFilters = (
subscription: Extract<LiveSubscription, { channel: "options" }>
): OptionPrintQueryFilters => {
if (subscription.option_contract_id) {
return {
view: "raw",
optionContractId: subscription.option_contract_id
};
}
return {
view: subscription.filters?.view ?? "signal",
security:
subscription.filters?.securityTypes?.length === 1
? subscription.filters.securityTypes[0]
: "all",
nbboSides: subscription.filters?.nbboSides,
optionTypes: subscription.filters?.optionTypes,
minNotional: subscription.filters?.minNotional,
underlyingIds: subscription.underlying_ids,
optionContractId: subscription.option_contract_id
};
};
const candleRedisKey = (underlyingId: string, intervalMs: number): string =>
`live:equity-candles:${underlyingId}:${intervalMs}`;
@ -489,18 +513,7 @@ export class LiveStateManager {
if (subscription.filters?.view === "raw" || scoped) {
this.stats.scopedClickHouseSnapshots += 1;
const limit = snapshotLimitFor(subscription, this.generic.options.limit);
const storageFilters: OptionPrintQueryFilters = {
view: subscription.filters?.view ?? "signal",
security:
subscription.filters?.securityTypes?.length === 1
? subscription.filters.securityTypes[0]
: "all",
nbboSides: subscription.filters?.nbboSides,
optionTypes: subscription.filters?.optionTypes,
minNotional: subscription.filters?.minNotional,
underlyingIds: subscription.underlying_ids,
optionContractId: subscription.option_contract_id
};
const storageFilters = buildOptionSnapshotFilters(subscription);
const items = await fetchRecentOptionPrints(
this.clickhouse,
limit,

107
services/api/src/option-queries.ts Normal file
View file

@ -0,0 +1,107 @@
import type { OptionPrintQueryFilters } from "@islandflow/storage";
import {
OptionFlowViewSchema,
OptionNbboSideSchema,
OptionSecurityTypeSchema,
OptionTypeSchema,
type OptionFlowFilters
} from "@islandflow/types";
import { z } from "zod";
const optionSideListSchema = z
.string()
.transform((value) =>
value
.split(",")
.map((entry) => entry.trim())
.filter(Boolean)
)
.pipe(z.array(OptionNbboSideSchema));
const optionTypeListSchema = z
.string()
.transform((value) =>
value
.split(",")
.map((entry) => entry.trim())
.filter(Boolean)
)
.pipe(z.array(OptionTypeSchema));
const optionSecuritySchema = z.enum(["stock", "etf", "all"]);
const optionFilterQuerySchema = z.object({
view: OptionFlowViewSchema.optional(),
security: optionSecuritySchema.optional(),
side: optionSideListSchema.optional(),
type: optionTypeListSchema.optional(),
min_notional: z.coerce.number().nonnegative().optional()
});
export type ParsedOptionPrintQuery = {
scope: {
underlyingIds?: string[];
optionContractId?: string;
};
flowFilters: OptionFlowFilters;
storageFilters: OptionPrintQueryFilters;
isContractDrilldown: boolean;
};
const parseScopeList = (url: URL, ...keys: string[]): string[] | undefined => {
const values = keys
.flatMap((key) => url.searchParams.getAll(key))
.flatMap((value) => value.split(","))
.map((value) => value.trim().toUpperCase())
.filter(Boolean);
const unique = Array.from(new Set(values));
return unique.length > 0 ? unique : undefined;
};
export const parseOptionPrintQuery = (url: URL): ParsedOptionPrintQuery => {
const parsed = optionFilterQuerySchema.parse({
view: url.searchParams.get("view") ?? undefined,
security: url.searchParams.get("security") ?? undefined,
side: url.searchParams.get("side") ?? undefined,
type: url.searchParams.get("type") ?? undefined,
min_notional: url.searchParams.get("min_notional") ?? undefined
});
const scope = {
underlyingIds: parseScopeList(url, "underlying_id", "underlying_ids"),
optionContractId: url.searchParams.get("option_contract_id") ?? undefined
};
const view = parsed.view ?? "signal";
const security = parsed.security ?? (view === "raw" ? "all" : "stock");
const flowFilters: OptionFlowFilters = {
view,
securityTypes:
security === "all"
? undefined
: ([security] as Array<z.infer<typeof OptionSecurityTypeSchema>>),
nbboSides: parsed.side,
optionTypes: parsed.type,
minNotional: parsed.min_notional
};
const isContractDrilldown = Boolean(scope.optionContractId);
const storageFilters: OptionPrintQueryFilters = isContractDrilldown
? {
view: "raw",
optionContractId: scope.optionContractId
}
: {
view,
security,
minNotional: parsed.min_notional,
nbboSides: parsed.side,
optionTypes: parsed.type,
underlyingIds: scope.underlyingIds,
optionContractId: scope.optionContractId
};
return {
scope,
flowFilters,
storageFilters,
isContractDrilldown
};
};

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

@ -1,6 +1,7 @@
import { describe, expect, it } from "bun:test";
import type { ClickHouseClient } from "@islandflow/storage";
import {
buildOptionSnapshotFilters,
HOT_LIVE_REDIS_KEYS,
LiveStateManager,
isLiveItemFresh,
@ -450,6 +451,74 @@ describe("LiveStateManager", () => {
expect(isLiveItemFresh("options", snapshot.items[0], now)).toBe(false);
});
it("builds raw contract-only snapshot filters for focused option subscriptions", () => {
expect(
buildOptionSnapshotFilters({
channel: "options",
filters: {
view: "signal",
minNotional: 500_000,
nbboSides: ["A"],
optionTypes: ["call"],
securityTypes: ["stock"]
},
underlying_ids: ["AAPL"],
option_contract_id: "AAPL-2025-01-17-200-C"
})
).toEqual({
view: "raw",
optionContractId: "AAPL-2025-01-17-200-C"
});
});
it("returns raw contract rows for focused option snapshots even when broad filters would reject them", async () => {
const manager = new LiveStateManager(
makeClickHouse((query) => {
expect(query).toContain("option_contract_id = 'AAPL-2025-01-17-200-C'");
expect(query).not.toContain("signal_pass = 1");
expect(query).not.toContain("notional >=");
expect(query).not.toContain("nbbo_side IN");
expect(query).not.toContain("option_type IN");
return [
{
source_ts: 1_000,
ingest_ts: 1_001,
seq: 1,
trace_id: "opt-raw",
ts: 1_000,
option_contract_id: "AAPL-2025-01-17-200-C",
underlying_id: "AAPL",
option_type: "put",
nbbo_side: "B",
notional: 50_000,
signal_pass: false,
price: 1,
size: 5,
exchange: "X"
}
];
}),
null
);
const snapshot = await manager.getSnapshot({
channel: "options",
filters: {
view: "signal",
minNotional: 500_000,
nbboSides: ["A"],
optionTypes: ["call"],
securityTypes: ["stock"]
},
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-raw"
]);
});
it("seeds scoped equity snapshots from clickhouse rows older than 24h", async () => {
const now = Date.now();
const staleTs = now - 25 * 60 * 60 * 1000;

59
services/api/tests/option-queries.test.ts Normal file
View file

@ -0,0 +1,59 @@
import { describe, expect, it } from "bun:test";
import { parseOptionPrintQuery } from "../src/option-queries";
describe("parseOptionPrintQuery", () => {
it("keeps broad option flow filters for non-contract requests", () => {
const url = new URL(
"http://localhost/prints/options?view=signal&security=stock&side=A&type=call&min_notional=500000&underlying_ids=AAPL,MSFT"
);
expect(parseOptionPrintQuery(url)).toEqual({
scope: {
underlyingIds: ["AAPL", "MSFT"],
optionContractId: undefined
},
flowFilters: {
view: "signal",
securityTypes: ["stock"],
nbboSides: ["A"],
optionTypes: ["call"],
minNotional: 500000
},
storageFilters: {
view: "signal",
security: "stock",
nbboSides: ["A"],
optionTypes: ["call"],
minNotional: 500000,
underlyingIds: ["AAPL", "MSFT"],
optionContractId: undefined
},
isContractDrilldown: false
});
});
it("switches contract requests to raw contract-only storage filters", () => {
const url = new URL(
"http://localhost/replay/options?view=signal&security=stock&side=A&type=call&min_notional=500000&underlying_id=AAPL&option_contract_id=AAPL-2025-01-17-200-C"
);
expect(parseOptionPrintQuery(url)).toEqual({
scope: {
underlyingIds: ["AAPL"],
optionContractId: "AAPL-2025-01-17-200-C"
},
flowFilters: {
view: "signal",
securityTypes: ["stock"],
nbboSides: ["A"],
optionTypes: ["call"],
minNotional: 500000
},
storageFilters: {
view: "raw",
optionContractId: "AAPL-2025-01-17-200-C"
},
isContractDrilldown: true
});
});
});