Implement live tape scroll-gated history

services/api/src/index.ts

@@ -49,6 +49,7 @@ import {
   fetchSmartMoneyEventsBefore,
   fetchFlowPacketsAfter,
   fetchFlowPacketById,
+  fetchFlowPacketsByMemberTraceIds,
   fetchFlowPacketsBefore,
   fetchRecentAlerts,
   fetchRecentClassifierHits,
@@ -76,6 +77,9 @@ import {
   fetchOptionPrintsBefore,
   fetchOptionPrintsAfter,
   fetchOptionPrintsByTraceIds,
+  fetchNearestOptionNBBOForPrints,
+  fetchSmartMoneyEventsByPacketIds,
+  fetchClassifierHitsByPacketIds,
   fetchRecentOptionPrints
 } from "@islandflow/storage";
 import type { EquityPrintQueryFilters, OptionPrintQueryFilters } from "@islandflow/storage";
@@ -303,6 +307,28 @@ const jsonResponse = (body: unknown, status = 200): Response => {
   });
 };
 
+const readJsonBody = async (req: Request): Promise<unknown> => {
+  const text = await req.text();
+  if (!text.trim()) {
+    return {};
+  }
+  return JSON.parse(text);
+};
+
+const optionsSupportLookupSchema = z.object({
+  trace_ids: z.array(z.string().min(1)).default([]),
+  nbbo_context: z
+    .array(
+      z.object({
+        trace_id: z.string().min(1),
+        option_contract_id: z.string().min(1),
+        ts: z.number().int().nonnegative()
+      })
+    )
+    .optional()
+    .default([])
+});
+
 const parseLimit = (value: string | null): number => {
   if (value === null) {
     return env.REST_DEFAULT_LIMIT;
@@ -1608,6 +1634,33 @@ const run = async () => {
         return jsonResponse({ data });
       }
 
+      if (req.method === "POST" && url.pathname === "/lookup/options-support") {
+        try {
+          const body = optionsSupportLookupSchema.parse(await readJsonBody(req));
+          const packets = await fetchFlowPacketsByMemberTraceIds(clickhouse, body.trace_ids);
+          const packetIds = packets.map((packet) => packet.id);
+          const [smartMoney, classifierHits, nbboByTraceId] = await Promise.all([
+            fetchSmartMoneyEventsByPacketIds(clickhouse, packetIds),
+            fetchClassifierHitsByPacketIds(clickhouse, packetIds),
+            fetchNearestOptionNBBOForPrints(clickhouse, body.nbbo_context)
+          ]);
+          return jsonResponse({
+            packets,
+            smart_money: smartMoney,
+            classifier_hits: classifierHits,
+            nbbo_by_trace_id: nbboByTraceId
+          });
+        } catch (error) {
+          return jsonResponse(
+            {
+              error: "invalid options support lookup",
+              detail: error instanceof Error ? error.message : String(error)
+            },
+            400
+          );
+        }
+      }
+
       if (req.method === "GET" && url.pathname === "/equity-joins/by-id") {
         const ids = url.searchParams.getAll("id");
         const data = await fetchEquityPrintJoinsByIds(clickhouse, ids);

services/api/src/live.ts

@@ -327,6 +327,14 @@ const nextBeforeForItems = <T>(items: T[], cursorOf: (item: T) => Cursor): Curso
   return last ? cursorOf(last) : null;
 };
 
+const snapshotLimitFor = (subscription: LiveSubscription, configuredLimit: number): number => {
+  const requested = "snapshot_limit" in subscription ? subscription.snapshot_limit : undefined;
+  if (!requested) {
+    return configuredLimit;
+  }
+  return Math.max(1, Math.min(configuredLimit, Math.floor(requested)));
+};
+
 const candleRedisKey = (underlyingId: string, intervalMs: number): string =>
   `live:equity-candles:${underlyingId}:${intervalMs}`;
 
@@ -448,6 +456,7 @@ export class LiveStateManager {
         const scoped =
           Boolean(subscription.underlying_ids?.length) || Boolean(subscription.option_contract_id);
         if (subscription.filters?.view === "raw" || scoped) {
+          const limit = snapshotLimitFor(subscription, this.generic.options.limit);
           const storageFilters: OptionPrintQueryFilters = {
             view: subscription.filters?.view ?? "signal",
             security:
@@ -463,7 +472,7 @@ export class LiveStateManager {
           };
           const items = await fetchRecentOptionPrints(
             this.clickhouse,
-            this.generic.options.limit,
+            limit,
             undefined,
             storageFilters
           );
@@ -476,10 +485,11 @@ 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 {
           subscription,
           items,
@@ -489,10 +499,11 @@ export class LiveStateManager {
       }
       case "flow": {
         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 {
           subscription,
           items,
@@ -502,12 +513,13 @@ export class LiveStateManager {
       }
       case "equities": {
         const config = this.generic.equities;
+        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
           };
-          const items = await fetchRecentEquityPrints(this.clickhouse, config.limit, filters);
+          const items = await fetchRecentEquityPrints(this.clickhouse, limit, filters);
           return {
             subscription,
             items,
@@ -517,7 +529,7 @@ export class LiveStateManager {
         }
         const items = (this.genericItems.get("equities") ?? []).filter((item) =>
           isWithinLiveFeedLookback("equities", item)
-        );
+        ).slice(0, limit);
         return {
           subscription,
           items,
@@ -555,9 +567,10 @@ 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);
         return {
           subscription,
           items,