Add smart money event calendar enrichment

.beads/issues.jsonl

@@ -4,7 +4,7 @@
 {"_type":"issue","id":"islandflow-ayo","title":"Drop stale backlog events from live fanout","description":"Follow-up to live freshness rollout: /ws/live was still fanning out stale backlog events for freshness-gated channels, which kept tape panes in Live feed behind despite active synthetic ingest. Gate fanout and cache ingest by freshness for options/nbbo/equities/flow.","status":"closed","priority":1,"issue_type":"bug","assignee":"dirtydishes","owner":"dishes@dpdrm.com","created_at":"2026-04-28T21:26:39Z","created_by":"dirtydishes","updated_at":"2026-04-28T21:26:44Z","started_at":"2026-04-28T21:26:44Z","closed_at":"2026-04-28T21:26:44Z","close_reason":"Completed","dependency_count":0,"dependent_count":0,"comment_count":0}
 {"_type":"issue","id":"islandflow-0v6","title":"Fix tape freshness, NBBO coverage, pause controls, and filter popup","description":"Implement the tape fixes requested for synthetic options notional sizing, strict live freshness, live-mode pause/resume behavior, stronger NBBO snapshot coverage, and moving flow filters behind a popup. Includes server-side live cache changes, web terminal state/UI changes, and tests for synthetic pricing, live snapshot freshness/NBBO retention, and live pause/filter interactions.","status":"closed","priority":1,"issue_type":"task","assignee":"dirtydishes","owner":"dishes@dpdrm.com","created_at":"2026-04-28T21:02:52Z","created_by":"dirtydishes","updated_at":"2026-04-28T21:13:38Z","started_at":"2026-04-28T21:02:57Z","closed_at":"2026-04-28T21:13:38Z","close_reason":"Completed","dependency_count":0,"dependent_count":0,"comment_count":0}
 {"_type":"issue","id":"islandflow-e4r","title":"Implement smart-money flow filtering and synthetic firehose modes","description":"Implement the approved multi-surface plan for named synthetic market profiles, options raw-vs-signal filtering, live/API filter contracts, Tape page client-side flow filters, firehose-readiness improvements, tests, and README updates.","status":"closed","priority":1,"issue_type":"feature","assignee":"dirtydishes","owner":"dishes@dpdrm.com","created_at":"2026-04-28T20:10:49Z","created_by":"dirtydishes","updated_at":"2026-04-28T20:29:29Z","started_at":"2026-04-28T20:10:53Z","closed_at":"2026-04-28T20:29:29Z","close_reason":"Implemented synthetic market profiles, options signal-path filtering, signal-aware API/replay contracts, Tape page filters, tests, and README updates. Follow-up tracked in islandflow-biq.","dependency_count":0,"dependent_count":0,"comment_count":0}
-{"_type":"issue","id":"islandflow-b6d","title":"Finish smart-money event-calendar enrichment","description":"Finish the smart-money event-calendar provider layer in services/refdata and connect days-to-event / expiry-after-event enrichment into compute using timestamp-available data only.","status":"open","priority":2,"issue_type":"task","owner":"dishes@dpdrm.com","created_at":"2026-05-04T21:35:26Z","created_by":"dirtydishes","updated_at":"2026-05-04T21:35:26Z","dependency_count":0,"dependent_count":0,"comment_count":0}
+{"_type":"issue","id":"islandflow-b6d","title":"Finish smart-money event-calendar enrichment","description":"Finish the smart-money event-calendar provider layer in services/refdata and connect days-to-event / expiry-after-event enrichment into compute using timestamp-available data only.","status":"closed","priority":2,"issue_type":"task","assignee":"dirtydishes","owner":"dishes@dpdrm.com","created_at":"2026-05-04T21:35:26Z","created_by":"dirtydishes","updated_at":"2026-05-04T23:21:09Z","started_at":"2026-05-04T23:18:29Z","closed_at":"2026-05-04T23:21:09Z","close_reason":"Completed event-calendar provider and compute enrichment","dependency_count":0,"dependent_count":0,"comment_count":0}
 {"_type":"issue","id":"islandflow-e60","title":"Add smart-money replay evaluation harness","description":"Add replay-style live-vs-batch consistency tests plus evaluation utilities for parent-event precision/recall, calibration, abstention rate, and economic sanity checks.","status":"open","priority":2,"issue_type":"task","owner":"dishes@dpdrm.com","created_at":"2026-05-04T21:35:25Z","created_by":"dirtydishes","updated_at":"2026-05-04T21:35:25Z","dependency_count":0,"dependent_count":0,"comment_count":0}
 {"_type":"issue","id":"islandflow-020","title":"Rebuild synthetic smart-money scenarios","description":"Rework services/ingest-options synthetic generation around labeled parent-event templates for the six core smart-money profiles plus neutral background noise, with deterministic test/demo modes and hidden labels for tests.","status":"open","priority":2,"issue_type":"task","owner":"dishes@dpdrm.com","created_at":"2026-05-04T21:35:24Z","created_by":"dirtydishes","updated_at":"2026-05-04T21:35:24Z","dependency_count":0,"dependent_count":0,"comment_count":0}
 {"_type":"issue","id":"islandflow-zs0","title":"Migrate terminal UI to smart-money profiles","description":"Migrate apps/web terminal rendering to consume SmartMoneyEvent directly: primary profile, probability ladder, reason codes, and suppression/abstention state, while preserving legacy alert/classifier displays during the bridge.","status":"open","priority":2,"issue_type":"task","owner":"dishes@dpdrm.com","created_at":"2026-05-04T21:35:23Z","created_by":"dirtydishes","updated_at":"2026-05-04T21:35:23Z","dependency_count":0,"dependent_count":0,"comment_count":0}

SMART_MONEY_REBUILD_PLAN.md

@@ -24,7 +24,7 @@ Acceptance: live and replay produce the same event ID for the same packet.
 ### Phase 3: Feature Engineering
 - [x] Build typed features for aggressor mix, spread/quote quality, timing, strike concentration, DTE, moneyness, structure markers, and event alignment fields.
 - [x] Keep batch-only validation fields out of live scoring.
-- [ ] Connect an external event-calendar feed through `services/refdata`.
+- [x] Connect an external event-calendar feed through `services/refdata`.
 
 Acceptance: missing event-calendar fields produce neutral `null` feature values and do not block scoring.
 

bun.lock

@@ -81,6 +81,7 @@
         "@islandflow/bus": "workspace:*",
         "@islandflow/config": "workspace:*",
         "@islandflow/observability": "workspace:*",
+        "@islandflow/refdata": "workspace:*",
         "@islandflow/storage": "workspace:*",
         "@islandflow/types": "workspace:*",
         "redis": "^5.10.0",

services/compute/package.json

@@ -9,6 +9,7 @@
     "@islandflow/bus": "workspace:*",
     "@islandflow/config": "workspace:*",
     "@islandflow/observability": "workspace:*",
+    "@islandflow/refdata": "workspace:*",
     "@islandflow/storage": "workspace:*",
     "@islandflow/types": "workspace:*",
     "redis": "^5.10.0",

services/compute/src/index.ts

@@ -1,5 +1,10 @@
 import { readEnv } from "@islandflow/config";
 import { createLogger } from "@islandflow/observability";
+import {
+  createEmptyEventCalendarProvider,
+  loadEventCalendarProviderFromFile,
+  type EventCalendarProvider
+} from "@islandflow/refdata/event-calendar";
 import {
   SUBJECT_ALERTS,
   SUBJECT_CLASSIFIER_HITS,
@@ -135,10 +140,12 @@ const envSchema = z.object({
   CLASSIFIER_MIN_AGGRESSOR_RATIO: z.coerce.number().min(0).max(1).default(0.55),
   CLASSIFIER_0DTE_MAX_ATM_PCT: z.coerce.number().min(0).max(1).default(0.01),
   CLASSIFIER_0DTE_MIN_PREMIUM: z.coerce.number().positive().default(20_000),
-  CLASSIFIER_0DTE_MIN_SIZE: z.coerce.number().int().positive().default(400)
+  CLASSIFIER_0DTE_MIN_SIZE: z.coerce.number().int().positive().default(400),
+  SMART_MONEY_EVENT_CALENDAR_PATH: z.string().optional()
 });
 
 const env = readEnv(envSchema);
+let eventCalendarProvider: EventCalendarProvider = createEmptyEventCalendarProvider();
 
 const classifierConfig: ClassifierConfig = {
   sweepMinPremium: env.CLASSIFIER_SWEEP_MIN_PREMIUM,
@@ -898,7 +905,16 @@ const emitClassifiers = async (
 ): Promise<void> => {
   let smartMoneyEvent: SmartMoneyEvent;
   try {
-    smartMoneyEvent = SmartMoneyEventSchema.parse(buildSmartMoneyEventFromPacket(packet));
+    const underlyingId =
+      typeof packet.features.underlying_id === "string"
+        ? packet.features.underlying_id
+        : parseContractId(typeof packet.features.option_contract_id === "string" ? packet.features.option_contract_id : "")?.root;
+    const referenceTs =
+      typeof packet.features.end_ts === "number" && Number.isFinite(packet.features.end_ts)
+        ? packet.features.end_ts
+        : packet.source_ts;
+    const eventCalendarMatch = underlyingId ? eventCalendarProvider.findNextEvent(underlyingId, referenceTs) : null;
+    smartMoneyEvent = SmartMoneyEventSchema.parse(buildSmartMoneyEventFromPacket(packet, { eventCalendarMatch }));
     await insertSmartMoneyEvent(clickhouse, smartMoneyEvent);
     await publishJson(js, SUBJECT_SMART_MONEY_EVENTS, smartMoneyEvent);
   } catch (error) {
@@ -1200,6 +1216,19 @@ const run = async () => {
     database: env.CLICKHOUSE_DATABASE
   });
 
+  if (env.SMART_MONEY_EVENT_CALENDAR_PATH) {
+    try {
+      eventCalendarProvider = await loadEventCalendarProviderFromFile(env.SMART_MONEY_EVENT_CALENDAR_PATH);
+      logger.info("smart money event calendar loaded", { path: env.SMART_MONEY_EVENT_CALENDAR_PATH });
+    } catch (error) {
+      eventCalendarProvider = createEmptyEventCalendarProvider();
+      logger.warn("smart money event calendar unavailable; scoring will use neutral event features", {
+        path: env.SMART_MONEY_EVENT_CALENDAR_PATH,
+        error: error instanceof Error ? error.message : String(error)
+      });
+    }
+  }
+
   const redis = createRedisClient(env.REDIS_URL);
   redis.on("error", (error) => {
     logger.warn("redis client error", { error: error instanceof Error ? error.message : String(error) });

services/compute/src/parent-events.ts

@@ -8,6 +8,7 @@ import {
   type SmartMoneyProfileId,
   type SmartMoneyProfileScore
 } from "@islandflow/types";
+import type { EventCalendarMatch } from "@islandflow/refdata/event-calendar";
 import { parseContractId } from "./contracts";
 
 const MS_PER_DAY = 86_400_000;
@@ -97,7 +98,11 @@ const inferDirection = (packet: FlowPacket): SmartMoneyDirection => {
   return "neutral";
 };
 
-const buildFeatures = (packet: FlowPacket): SmartMoneyFeatures => {
+export type SmartMoneyParentEventOptions = {
+  eventCalendarMatch?: EventCalendarMatch | null;
+};
+
+const buildFeatures = (packet: FlowPacket, options: SmartMoneyParentEventOptions = {}): SmartMoneyFeatures => {
   const contractId = stringFeature(packet, "option_contract_id");
   const contract = parseContractId(contractId);
   const underlyingMid = numberFeature(packet, "underlying_mid");
@@ -108,7 +113,8 @@ const buildFeatures = (packet: FlowPacket): SmartMoneyFeatures => {
   const structureLegs = Math.max(0, Math.round(numberFeature(packet, "structure_legs")));
   const strikeCount = Math.max(1, Math.round(numberFeature(packet, "structure_strikes") || (contract ? 1 : 0)));
   const specialCount = numberFeature(packet, "special_print_count");
-  const eventTs = numberFeature(packet, "corporate_event_ts");
+  const calendarEventTs = options.eventCalendarMatch?.event_ts ?? null;
+  const eventTs = calendarEventTs ?? numberFeature(packet, "corporate_event_ts");
   const referenceTs = getReferenceTs(packet);
   const expiryTs = contract ? Date.parse(`${contract.expiry}T00:00:00Z`) : Number.NaN;
 
@@ -259,8 +265,11 @@ const evaluateProfiles = (
   return scores.sort((a, b) => b.probability - a.probability);
 };
 
-export const buildSmartMoneyEventFromPacket = (packet: FlowPacket): SmartMoneyEvent => {
+export const buildSmartMoneyEventFromPacket = (
-  const features = buildFeatures(packet);
+  packet: FlowPacket,
+  options: SmartMoneyParentEventOptions = {}
+): SmartMoneyEvent => {
+  const features = buildFeatures(packet, options);
   const suppressed = detectSuppression(packet, features);
   const profileScores = evaluateProfiles(packet, features, suppressed);
   const primary = profileScores[0] ?? null;

services/compute/tests/parent-events.test.ts

@@ -55,4 +55,58 @@ describe("smart money parent events", () => {
     expect(event.primary_profile_id).toBeNull();
     expect(event.suppressed_reasons).toContain("stale_or_missing_quote_context");
   });
+
+  it("uses timestamp-available event calendar matches for event-driven scoring", () => {
+    const packet = buildFlowPacket({
+      id: "flowpacket:event-driven",
+      source_ts: Date.parse("2025-01-15T15:00:00Z"),
+      features: {
+        option_contract_id: "AAPL-2025-02-07-225-C",
+        underlying_id: "AAPL",
+        count: 1,
+        window_ms: 450,
+        total_size: 1800,
+        total_premium: 160_000,
+        total_notional: 16_000_000,
+        nbbo_coverage_ratio: 0.5,
+        nbbo_aggressive_ratio: 0.4,
+        nbbo_aggressive_buy_ratio: 0.4,
+        nbbo_aggressive_sell_ratio: 0.1,
+        nbbo_inside_ratio: 0.08,
+        underlying_mid: 224
+      }
+    });
+
+    const event = buildSmartMoneyEventFromPacket(packet, {
+      eventCalendarMatch: {
+        underlying_id: "AAPL",
+        event_ts: Date.parse("2025-01-31T21:00:00Z"),
+        event_kind: "earnings",
+        announced_ts: Date.parse("2024-12-20T21:00:00Z"),
+        days_to_event: 16.25
+      }
+    });
+
+    expect(event.features.days_to_event).toBeCloseTo(16.25);
+    expect(event.features.expiry_after_event).toBe(true);
+    expect(event.primary_profile_id).toBe("event_driven");
+  });
+
+  it("keeps event-calendar features neutral when no match is available", () => {
+    const packet = buildFlowPacket({
+      id: "flowpacket:no-calendar",
+      source_ts: Date.parse("2025-01-15T15:00:00Z"),
+      features: {
+        option_contract_id: "AAPL-2025-02-07-225-C",
+        underlying_id: "AAPL",
+        total_premium: 160_000,
+        nbbo_coverage_ratio: 0.92
+      }
+    });
+
+    const event = buildSmartMoneyEventFromPacket(packet);
+    expect(event.features.days_to_event).toBeNull();
+    expect(event.features.expiry_after_event).toBeNull();
+    expect(event.features.pre_event_concentration).toBeNull();
+  });
 });

services/refdata/package.json

@@ -2,6 +2,9 @@
   "name": "@islandflow/refdata",
   "private": true,
   "type": "module",
+  "exports": {
+    "./event-calendar": "./src/event-calendar.ts"
+  },
   "scripts": {
     "dev": "bun run src/index.ts"
   },

services/refdata/src/event-calendar.ts

@@ -0,0 +1,116 @@
+export type EventCalendarKind = "earnings" | "dividend" | "corporate_action" | "m_and_a" | "news" | "other";
+
+export type EventCalendarEntry = {
+  underlying_id: string;
+  event_ts: number;
+  event_kind: EventCalendarKind;
+  announced_ts: number;
+  source?: string;
+  source_event_id?: string;
+};
+
+export type EventCalendarMatch = EventCalendarEntry & {
+  days_to_event: number;
+};
+
+export type EventCalendarProvider = {
+  findNextEvent(underlyingId: string, asOfTs: number): EventCalendarMatch | null;
+};
+
+const MS_PER_DAY = 86_400_000;
+
+const EVENT_KINDS = new Set<EventCalendarKind>([
+  "earnings",
+  "dividend",
+  "corporate_action",
+  "m_and_a",
+  "news",
+  "other"
+]);
+
+const normalizeUnderlying = (underlyingId: string): string => underlyingId.trim().toUpperCase();
+
+const asNumber = (value: unknown): number | null => {
+  if (typeof value === "number" && Number.isFinite(value)) {
+    return value;
+  }
+  if (typeof value === "string" && value.trim()) {
+    const parsed = Number(value);
+    if (Number.isFinite(parsed)) {
+      return parsed;
+    }
+    const ts = Date.parse(value);
+    return Number.isFinite(ts) ? ts : null;
+  }
+  return null;
+};
+
+const asString = (value: unknown): string | null => (typeof value === "string" && value.trim() ? value.trim() : null);
+
+export const parseEventCalendarEntries = (value: unknown): EventCalendarEntry[] => {
+  const rows = Array.isArray(value) ? value : [];
+  return rows.flatMap((row): EventCalendarEntry[] => {
+    if (!row || typeof row !== "object") {
+      return [];
+    }
+
+    const record = row as Record<string, unknown>;
+    const underlying = asString(record.underlying_id ?? record.underlying ?? record.symbol);
+    const eventTs = asNumber(record.event_ts ?? record.event_time ?? record.event_date);
+    const announcedTs = asNumber(record.announced_ts ?? record.available_ts ?? record.as_of_ts ?? record.created_ts) ?? 0;
+    const rawKind = asString(record.event_kind ?? record.kind ?? record.type) ?? "other";
+    const eventKind = EVENT_KINDS.has(rawKind as EventCalendarKind) ? (rawKind as EventCalendarKind) : "other";
+
+    if (!underlying || eventTs === null || eventTs < 0 || announcedTs < 0) {
+      return [];
+    }
+
+    return [
+      {
+        underlying_id: normalizeUnderlying(underlying),
+        event_ts: Math.trunc(eventTs),
+        event_kind: eventKind,
+        announced_ts: Math.trunc(announcedTs),
+        ...(asString(record.source) ? { source: asString(record.source) ?? undefined } : {}),
+        ...(asString(record.source_event_id ?? record.id)
+          ? { source_event_id: asString(record.source_event_id ?? record.id) ?? undefined }
+          : {})
+      }
+    ];
+  });
+};
+
+export const createStaticEventCalendarProvider = (entries: EventCalendarEntry[]): EventCalendarProvider => {
+  const byUnderlying = new Map<string, EventCalendarEntry[]>();
+  for (const entry of entries) {
+    const key = normalizeUnderlying(entry.underlying_id);
+    const normalized = { ...entry, underlying_id: key };
+    const bucket = byUnderlying.get(key) ?? [];
+    bucket.push(normalized);
+    byUnderlying.set(key, bucket);
+  }
+
+  for (const bucket of byUnderlying.values()) {
+    bucket.sort((a, b) => a.event_ts - b.event_ts || a.announced_ts - b.announced_ts);
+  }
+
+  return {
+    findNextEvent(underlyingId, asOfTs) {
+      const key = normalizeUnderlying(underlyingId);
+      if (!key || !Number.isFinite(asOfTs)) {
+        return null;
+      }
+
+      const bucket = byUnderlying.get(key) ?? [];
+      const entry = bucket.find((candidate) => candidate.announced_ts <= asOfTs && candidate.event_ts >= asOfTs);
+      return entry ? { ...entry, days_to_event: (entry.event_ts - asOfTs) / MS_PER_DAY } : null;
+    }
+  };
+};
+
+export const createEmptyEventCalendarProvider = (): EventCalendarProvider => createStaticEventCalendarProvider([]);
+
+export const loadEventCalendarProviderFromFile = async (path: string): Promise<EventCalendarProvider> => {
+  const text = await Bun.file(path).text();
+  return createStaticEventCalendarProvider(parseEventCalendarEntries(JSON.parse(text)));
+};

services/refdata/src/index.ts

@@ -1,10 +1,28 @@
 import { createLogger } from "@islandflow/observability";
+import { createEmptyEventCalendarProvider, loadEventCalendarProviderFromFile } from "./event-calendar";
 
 const service = "refdata";
 const logger = createLogger({ service });
 
 logger.info("service starting");
 
+const eventCalendarPath = process.env.REFDATA_EVENT_CALENDAR_PATH ?? process.env.SMART_MONEY_EVENT_CALENDAR_PATH;
+
+if (eventCalendarPath) {
+  try {
+    await loadEventCalendarProviderFromFile(eventCalendarPath);
+    logger.info("event calendar loaded", { path: eventCalendarPath });
+  } catch (error) {
+    logger.warn("event calendar unavailable", {
+      path: eventCalendarPath,
+      error: error instanceof Error ? error.message : String(error)
+    });
+  }
+} else {
+  createEmptyEventCalendarProvider();
+  logger.info("event calendar disabled");
+}
+
 const shutdown = (signal: string) => {
   logger.info("service stopping", { signal });
   process.exit(0);

services/refdata/tests/event-calendar.test.ts

@@ -0,0 +1,31 @@
+import { describe, expect, it } from "bun:test";
+import { createStaticEventCalendarProvider, parseEventCalendarEntries } from "../src/event-calendar";
+
+describe("event calendar refdata", () => {
+  it("parses provider rows and filters by timestamp availability", () => {
+    const entries = parseEventCalendarEntries([
+      {
+        symbol: "aapl",
+        event_date: "2025-01-31T21:00:00Z",
+        event_kind: "earnings",
+        announced_ts: "2025-01-20T21:00:00Z",
+        source: "fixture"
+      },
+      {
+        symbol: "AAPL",
+        event_date: "2025-02-28T21:00:00Z",
+        type: "mystery",
+        announced_ts: "2025-02-01T21:00:00Z"
+      }
+    ]);
+
+    const provider = createStaticEventCalendarProvider(entries);
+    const beforeAnnouncement = provider.findNextEvent("AAPL", Date.parse("2025-01-15T15:00:00Z"));
+    const afterAnnouncement = provider.findNextEvent("aapl", Date.parse("2025-01-21T15:00:00Z"));
+
+    expect(beforeAnnouncement).toBeNull();
+    expect(afterAnnouncement?.event_kind).toBe("earnings");
+    expect(afterAnnouncement?.underlying_id).toBe("AAPL");
+    expect(afterAnnouncement?.days_to_event).toBeGreaterThan(0);
+  });
+});