Add equity print quote joins

packages/bus/src/subjects.ts

@@ -4,6 +4,10 @@ export const STREAM_OPTION_NBBO = "OPTIONS_NBBO";
 export const SUBJECT_OPTION_NBBO = "options.nbbo";
 export const STREAM_EQUITY_PRINTS = "EQUITY_PRINTS";
 export const SUBJECT_EQUITY_PRINTS = "equities.prints";
+export const STREAM_EQUITY_QUOTES = "EQUITY_QUOTES";
+export const SUBJECT_EQUITY_QUOTES = "equities.quotes";
+export const STREAM_EQUITY_JOINS = "EQUITY_JOINS";
+export const SUBJECT_EQUITY_JOINS = "equities.joins";
 export const STREAM_FLOW_PACKETS = "FLOW_PACKETS";
 export const SUBJECT_FLOW_PACKETS = "flow.packets";
 export const STREAM_CLASSIFIER_HITS = "CLASSIFIER_HITS";

packages/storage/src/clickhouse.ts

@@ -3,6 +3,8 @@ import {
   AlertEventSchema,
   ClassifierHitEventSchema,
   EquityPrintSchema,
+  EquityQuoteSchema,
+  EquityPrintJoinSchema,
   FlowPacketSchema,
   OptionNBBOSchema,
   OptionPrintSchema
@@ -11,6 +13,8 @@ import type {
   AlertEvent,
   ClassifierHitEvent,
   EquityPrint,
+  EquityQuote,
+  EquityPrintJoin,
   FlowPacket,
   OptionNBBO,
   OptionPrint
@@ -26,6 +30,18 @@ import {
   EQUITY_PRINTS_TABLE,
   normalizeEquityPrint
 } from "./equity-prints";
+import {
+  equityQuotesTableDDL,
+  EQUITY_QUOTES_TABLE,
+  normalizeEquityQuote
+} from "./equity-quotes";
+import {
+  equityPrintJoinsTableDDL,
+  EQUITY_PRINT_JOINS_TABLE,
+  fromEquityPrintJoinRecord,
+  toEquityPrintJoinRecord,
+  type EquityPrintJoinRecord
+} from "./equity-print-joins";
 import {
   FLOW_PACKETS_TABLE,
   flowPacketsTableDDL,
@@ -88,6 +104,22 @@ export const ensureEquityPrintsTable = async (
   });
 };
 
+export const ensureEquityQuotesTable = async (
+  client: ClickHouseClient
+): Promise<void> => {
+  await client.exec({
+    query: equityQuotesTableDDL()
+  });
+};
+
+export const ensureEquityPrintJoinsTable = async (
+  client: ClickHouseClient
+): Promise<void> => {
+  await client.exec({
+    query: equityPrintJoinsTableDDL()
+  });
+};
+
 export const ensureFlowPacketsTable = async (
   client: ClickHouseClient
 ): Promise<void> => {
@@ -146,6 +178,30 @@ export const insertEquityPrint = async (
   });
 };
 
+export const insertEquityQuote = async (
+  client: ClickHouseClient,
+  quote: EquityQuote
+): Promise<void> => {
+  const record = normalizeEquityQuote(quote);
+  await client.insert({
+    table: EQUITY_QUOTES_TABLE,
+    values: [record],
+    format: "JSONEachRow"
+  });
+};
+
+export const insertEquityPrintJoin = async (
+  client: ClickHouseClient,
+  join: EquityPrintJoin
+): Promise<void> => {
+  const record = toEquityPrintJoinRecord(join);
+  await client.insert({
+    table: EQUITY_PRINT_JOINS_TABLE,
+    values: [record],
+    format: "JSONEachRow"
+  });
+};
+
 export const insertFlowPacket = async (
   client: ClickHouseClient,
   packet: FlowPacket
@@ -253,6 +309,20 @@ const normalizeOptionNbboRow = (row: unknown): unknown => {
   return row;
 };
 
+const normalizeEquityQuoteRow = (row: unknown): unknown => {
+  if (row && typeof row === "object") {
+    return normalizeNumericFields(row as Record<string, unknown>, [
+      "source_ts",
+      "ingest_ts",
+      "seq",
+      "ts",
+      "bid",
+      "ask"
+    ]);
+  }
+
+  return row;
+};
 
 const normalizeEquityRow = (row: unknown): unknown => {
   if (row && typeof row === "object") {
@@ -278,6 +348,25 @@ const normalizeEquityRow = (row: unknown): unknown => {
   return row;
 };
 
+const normalizeEquityPrintJoinRow = (row: unknown): EquityPrintJoinRecord | null => {
+  if (!row || typeof row !== "object") {
+    return null;
+  }
+
+  const record = row as Record<string, unknown>;
+  return {
+    source_ts: coerceNumber(record.source_ts) as number,
+    ingest_ts: coerceNumber(record.ingest_ts) as number,
+    seq: coerceNumber(record.seq) as number,
+    trace_id: String(record.trace_id ?? ""),
+    id: String(record.id ?? ""),
+    print_trace_id: String(record.print_trace_id ?? ""),
+    quote_trace_id: String(record.quote_trace_id ?? ""),
+    features_json: String(record.features_json ?? "{}"),
+    join_quality_json: String(record.join_quality_json ?? "{}")
+  };
+};
+
 const normalizeFlowPacketRow = (row: unknown): FlowPacketRecord | null => {
   if (!row || typeof row !== "object") {
     return null;
@@ -376,6 +465,38 @@ export const fetchRecentEquityPrints = async (
   return EquityPrintSchema.array().parse(rows.map(normalizeEquityRow));
 };
 
+export const fetchRecentEquityQuotes = async (
+  client: ClickHouseClient,
+  limit: number
+): Promise<EquityQuote[]> => {
+  const safeLimit = clampLimit(limit);
+  const result = await client.query({
+    query: `SELECT * FROM ${EQUITY_QUOTES_TABLE} ORDER BY ts DESC, seq DESC LIMIT ${safeLimit}`,
+    format: "JSONEachRow"
+  });
+
+  const rows = await result.json<unknown[]>();
+  return EquityQuoteSchema.array().parse(rows.map(normalizeEquityQuoteRow));
+};
+
+export const fetchRecentEquityPrintJoins = async (
+  client: ClickHouseClient,
+  limit: number
+): Promise<EquityPrintJoin[]> => {
+  const safeLimit = clampLimit(limit);
+  const result = await client.query({
+    query: `SELECT * FROM ${EQUITY_PRINT_JOINS_TABLE} ORDER BY source_ts DESC, seq DESC LIMIT ${safeLimit}`,
+    format: "JSONEachRow"
+  });
+
+  const rows = await result.json<unknown[]>();
+  const records = rows
+    .map(normalizeEquityPrintJoinRow)
+    .filter((record): record is EquityPrintJoinRecord => record !== null);
+  const joins = records.map(fromEquityPrintJoinRecord);
+  return EquityPrintJoinSchema.array().parse(joins);
+};
+
 export const fetchRecentFlowPackets = async (
   client: ClickHouseClient,
   limit: number
@@ -486,3 +607,45 @@ export const fetchEquityPrintsAfter = async (
   const rows = await result.json<unknown[]>();
   return EquityPrintSchema.array().parse(rows.map(normalizeEquityRow));
 };
+
+export const fetchEquityQuotesAfter = async (
+  client: ClickHouseClient,
+  afterTs: number,
+  afterSeq: number,
+  limit: number
+): Promise<EquityQuote[]> => {
+  const safeLimit = clampLimit(limit);
+  const safeAfterTs = clampCursor(afterTs);
+  const safeAfterSeq = clampCursor(afterSeq);
+
+  const result = await client.query({
+    query: `SELECT * FROM ${EQUITY_QUOTES_TABLE} WHERE (ts, seq) > (${safeAfterTs}, ${safeAfterSeq}) ORDER BY ts ASC, seq ASC LIMIT ${safeLimit}`,
+    format: "JSONEachRow"
+  });
+
+  const rows = await result.json<unknown[]>();
+  return EquityQuoteSchema.array().parse(rows.map(normalizeEquityQuoteRow));
+};
+
+export const fetchEquityPrintJoinsAfter = async (
+  client: ClickHouseClient,
+  afterTs: number,
+  afterSeq: number,
+  limit: number
+): Promise<EquityPrintJoin[]> => {
+  const safeLimit = clampLimit(limit);
+  const safeAfterTs = clampCursor(afterTs);
+  const safeAfterSeq = clampCursor(afterSeq);
+
+  const result = await client.query({
+    query: `SELECT * FROM ${EQUITY_PRINT_JOINS_TABLE} WHERE (source_ts, seq) > (${safeAfterTs}, ${safeAfterSeq}) ORDER BY source_ts ASC, seq ASC LIMIT ${safeLimit}`,
+    format: "JSONEachRow"
+  });
+
+  const rows = await result.json<unknown[]>();
+  const records = rows
+    .map(normalizeEquityPrintJoinRow)
+    .filter((record): record is EquityPrintJoinRecord => record !== null);
+  const joins = records.map(fromEquityPrintJoinRecord);
+  return EquityPrintJoinSchema.array().parse(joins);
+};

packages/storage/src/equity-print-joins.ts

@@ -0,0 +1,74 @@
+import type { EquityPrintJoin } from "@islandflow/types";
+
+export const EQUITY_PRINT_JOINS_TABLE = "equity_print_joins";
+
+export type EquityPrintJoinRecord = {
+  source_ts: number;
+  ingest_ts: number;
+  seq: number;
+  trace_id: string;
+  id: string;
+  print_trace_id: string;
+  quote_trace_id: string;
+  features_json: string;
+  join_quality_json: string;
+};
+
+export const equityPrintJoinsTableDDL = (): string => {
+  return `
+CREATE TABLE IF NOT EXISTS ${EQUITY_PRINT_JOINS_TABLE} (
+  source_ts UInt64,
+  ingest_ts UInt64,
+  seq UInt64,
+  trace_id String,
+  id String,
+  print_trace_id String,
+  quote_trace_id String,
+  features_json String,
+  join_quality_json String
+)
+ENGINE = MergeTree
+ORDER BY (source_ts, seq)
+`;
+};
+
+export const toEquityPrintJoinRecord = (join: EquityPrintJoin): EquityPrintJoinRecord => {
+  return {
+    source_ts: join.source_ts,
+    ingest_ts: join.ingest_ts,
+    seq: join.seq,
+    trace_id: join.trace_id,
+    id: join.id,
+    print_trace_id: join.print_trace_id,
+    quote_trace_id: join.quote_trace_id,
+    features_json: JSON.stringify(join.features),
+    join_quality_json: JSON.stringify(join.join_quality)
+  };
+};
+
+const safeJson = (value: string, fallback: Record<string, unknown>): Record<string, unknown> => {
+  try {
+    const parsed = JSON.parse(value);
+    if (parsed && typeof parsed === "object") {
+      return parsed as Record<string, unknown>;
+    }
+  } catch {
+    // ignore
+  }
+
+  return fallback;
+};
+
+export const fromEquityPrintJoinRecord = (record: EquityPrintJoinRecord): EquityPrintJoin => {
+  return {
+    source_ts: record.source_ts,
+    ingest_ts: record.ingest_ts,
+    seq: record.seq,
+    trace_id: record.trace_id,
+    id: record.id,
+    print_trace_id: record.print_trace_id,
+    quote_trace_id: record.quote_trace_id,
+    features: safeJson(record.features_json, {}),
+    join_quality: safeJson(record.join_quality_json, {}) as Record<string, number>
+  };
+};

packages/storage/src/equity-quotes.ts

@@ -0,0 +1,24 @@
+import type { EquityQuote } from "@islandflow/types";
+
+export const EQUITY_QUOTES_TABLE = "equity_quotes";
+
+export const equityQuotesTableDDL = (): string => {
+  return `
+CREATE TABLE IF NOT EXISTS ${EQUITY_QUOTES_TABLE} (
+  source_ts UInt64,
+  ingest_ts UInt64,
+  seq UInt64,
+  trace_id String,
+  ts UInt64,
+  underlying_id String,
+  bid Float64,
+  ask Float64
+)
+ENGINE = MergeTree
+ORDER BY (ts, underlying_id)
+`;
+};
+
+export const normalizeEquityQuote = (quote: EquityQuote): EquityQuote => {
+  return quote;
+};

packages/storage/src/index.ts

@@ -3,5 +3,7 @@ export * from "./classifier-hits";
 export * from "./alerts";
 export * from "./flow-packets";
 export * from "./equity-prints";
+export * from "./equity-quotes";
+export * from "./equity-print-joins";
 export * from "./option-prints";
 export * from "./option-nbbo";

packages/storage/tests/equity-print-joins.test.ts

@@ -0,0 +1,42 @@
+import { describe, expect, it } from "bun:test";
+import {
+  equityPrintJoinsTableDDL,
+  EQUITY_PRINT_JOINS_TABLE,
+  fromEquityPrintJoinRecord,
+  toEquityPrintJoinRecord
+} from "../src/equity-print-joins";
+
+const join = {
+  source_ts: 100,
+  ingest_ts: 120,
+  seq: 1,
+  trace_id: "equityjoin:trace-1",
+  id: "equityjoin:trace-1",
+  print_trace_id: "trace-1",
+  quote_trace_id: "quote-1",
+  features: {
+    underlying_id: "SPY",
+    price: 450.12,
+    size: 200,
+    off_exchange_flag: true,
+    quote_placement: "A"
+  },
+  join_quality: {
+    quote_age_ms: 15
+  }
+};
+
+describe("equity-print-joins storage helpers", () => {
+  it("includes the correct table name in the DDL", () => {
+    const ddl = equityPrintJoinsTableDDL();
+    expect(ddl).toContain(EQUITY_PRINT_JOINS_TABLE);
+    expect(ddl).toContain("CREATE TABLE IF NOT EXISTS");
+  });
+
+  it("round-trips equity print join records", () => {
+    const record = toEquityPrintJoinRecord(join);
+    const restored = fromEquityPrintJoinRecord(record);
+    expect(restored.features).toEqual(join.features);
+    expect(restored.join_quality).toEqual(join.join_quality);
+  });
+});

packages/storage/tests/equity-quotes.test.ts

@@ -0,0 +1,30 @@
+import { describe, expect, it } from "bun:test";
+import {
+  equityQuotesTableDDL,
+  EQUITY_QUOTES_TABLE,
+  normalizeEquityQuote
+} from "../src/equity-quotes";
+
+const baseQuote = {
+  source_ts: 100,
+  ingest_ts: 200,
+  seq: 1,
+  trace_id: "trace-1",
+  ts: 100,
+  underlying_id: "SPY",
+  bid: 450.1,
+  ask: 450.2
+};
+
+describe("equity-quotes storage helpers", () => {
+  it("keeps required fields intact", () => {
+    const normalized = normalizeEquityQuote(baseQuote);
+    expect(normalized).toEqual(baseQuote);
+  });
+
+  it("includes the correct table name in the DDL", () => {
+    const ddl = equityQuotesTableDDL();
+    expect(ddl).toContain(EQUITY_QUOTES_TABLE);
+    expect(ddl).toContain("CREATE TABLE IF NOT EXISTS");
+  });
+});

packages/types/src/events.ts

@@ -59,6 +59,18 @@ export const EquityQuoteSchema = EventMetaSchema.merge(
 
 export type EquityQuote = z.infer<typeof EquityQuoteSchema>;
 
+export const EquityPrintJoinSchema = EventMetaSchema.merge(
+  z.object({
+    id: z.string().min(1),
+    print_trace_id: z.string().min(1),
+    quote_trace_id: z.string(),
+    features: z.record(z.union([z.string(), z.number(), z.boolean()])),
+    join_quality: z.record(z.number())
+  })
+);
+
+export type EquityPrintJoin = z.infer<typeof EquityPrintJoinSchema>;
+
 export const FlowPacketSchema = EventMetaSchema.merge(
   z.object({
     id: z.string().min(1),