feat: dealer flow, mirror portfolio (M21), options convexity, FINRA short interest, alert producers, vendor gate

Snapshot of in-progress module work across multiple slices: - Dealer Flow: dealerExposureEngine, dealerMapService, dealerMapExplain, dealerMapIntegrity, dealerMapReplay, dealerStudyEngine, hanStyleLevels - Mirror Portfolio (M21): fundRepository, captureIngest, mirrorAlertProducers, fund holdings strip, live book, position capture ingest - Options: BSM, NormalizedOptionSurface types, OptionsChainRouter, ConvexityGate, option legs panel - Alert producers: vixLevel, rotation, thesis, unlock, portfolioRisk, mirror (fund_capture, fund_13f, mirror_diff) - FINRA short interest adapter + queue integration - SEC company tickers adapter + ingest (symbol search index seed) - Vendor gate (rate-limit-first data plane, ADR-0009) - CUSIP registry, reverse 13F refresh, stock float service - LRU cache, portfolio backtest engine - Frontend: dealer-flow, funds, journal, lab, monitor, plan, portfolio, reports, screener, strategies, theses, guided-start, exits, more pages - Volume profile, workspace profile, visibility-aware poll - ADRs 0010 (mirror math not advice), 0011 (symbol search index) - VENDOR_INTEGRATIONS.md, END_USER_TEST.md - .gitignore: exclude DBs, .DS_Store, local config, agent scratch
2026-08-10 13:36:26 -04:00
parent 04fc11b2fd
commit ac94acf9e3
229 changed files with 32617 additions and 3934 deletions
@@ -5,19 +5,37 @@
 // the injected scheduler (SourceAdapter/AdapterQueue satisfy `CacheScheduler`).
 import { DatabaseSync } from 'node:sqlite';
 import { db as defaultDb } from '../db/client.ts';
+import {
+  CANDLE_FRESH_MS,
+  quoteTtlMs,
+  SYMBOL_META_INCOMPLETE_TTL_MS,
+} from '../queue/sourceRatePolicy.ts';
+import { KvReadCache } from './LruCache.ts';

-export type SourceKind = 'yfinance' | 'nasdaq' | 'finra-bulk' | 'sec' | 'sec-fetch' | 'reddit' | 'x' | 'macro' | 'llm' | 'sec-lint-holders' | 'sec-lint-insiders' | 'fred';
+export type SourceKind = 'yfinance' | 'nasdaq' | 'finra-bulk' | 'finra-si' | 'sec' | 'sec-fetch' | 'sec-sc-fetch' | 'sec-tickers' | 'reddit' | 'x' | 'macro' | 'llm' | 'sec-lint-holders' | 'sec-lint-insiders' | 'fred';
 export type TickerKind = 'equity' | 'crypto' | 'etf' | 'index';
 export type CacheKey = string; // `${SourceKind}:${kind}:${id}` e.g. 'yfinance:quote:NVDA', 'yfinance:candles:NVDA:1d'
 export type TtlClass =
  | 'live_quote' | 'intraday' | 'daily_permanent' | 'options_snapshot'
  | 'filing_immutable' | 'quarterly_immutable' | 'thread_7d'
  | 'macro_event' | 'regime_classification' | 'llm_summary' | 'symbol_meta'
-  | 'short_interest';
+  | 'short_interest' | 'dividend_fundamentals' | 'fred_macro';

 export interface Provenance { fetchedAt: string; sourceKind: SourceKind; rawSourceId?: string; }

-export interface Quote { symbol: string; price: number; bid?: number | null; ask?: number | null; change?: number | null; changePercent?: number | null; iv?: number | null; }
+export interface Quote {
+  symbol: string;
+  price: number;
+  bid?: number | null;
+  ask?: number | null;
+  change?: number | null;
+  changePercent?: number | null;
+  iv?: number | null;
+  /** Which session the `price` print came from (pre/post when Yahoo has extended data). */
+  session?: 'regular' | 'pre' | 'post' | null;
+  /** RTH last (close during extended); useful when `price` is pre/post. */
+  regularPrice?: number | null;
+}
 export interface PriceCandle { ts: string; o: number; h: number; l: number; c: number; v: number; adjClose?: number | null; }
 export interface SymbolMeta { symbol: string; name?: string | null; sector?: string | null; industry?: string | null; exchange?: string | null; tickerKind: TickerKind; peers?: string[] | null; description?: string | null; website?: string | null; marketCap?: number | null; currency?: string | null; employees?: number | null; country?: string | null; city?: string | null; }
 export interface PriceAdjustment { symbol: string; exDate: string; type: "split" | "dividend"; ratio: number }
@@ -27,10 +45,18 @@ export interface CacheScheduler { queue(key: CacheKey): Promise<void>; }

 export interface CacheEntry<T> { value: T | null; provenance: Provenance | null; isStale: boolean; }

+// Module-level LRU cache for kv_cache reads — shared across all handler instances
+let kvReadCache = new KvReadCache({ max: 2000, ttl: 5 * 60 * 1000 });
+
+/** Set the LRU cache instance (for testing or runtime configuration) */
+export function setKvReadCache(cache: KvReadCache): void {
+  kvReadCache = cache;
+}
+
 export const TTL_MS: Record<TtlClass, number> = {
-  live_quote: 60_000,                       // 1min (mkt hrs); after-hours 15min refined in adapter slice
+  live_quote: 60_000,                       // 1min (mkt hrs); after-hours uses quoteTtlMs()
  intraday: 5 * 60_000,
-  daily_permanent: Number.POSITIVE_INFINITY, // immutable once fetched; stale only when absent
+  daily_permanent: Number.POSITIVE_INFINITY, // full history kept; EOD refresh is schedule-driven
  options_snapshot: 15 * 60_000,
  filing_immutable: Number.POSITIVE_INFINITY,
  quarterly_immutable: Number.POSITIVE_INFINITY,
@@ -40,6 +66,8 @@ export const TTL_MS: Record<TtlClass, number> = {
  llm_summary: Number.POSITIVE_INFINITY,    // cached forever by prompt-hash
  symbol_meta: 7 * 24 * 60 * 60_000,         // weekly (slow-changing sector/industry)
  short_interest: 24 * 60 * 60_000,      // refreshed twice/month per source
+  dividend_fundamentals: 7 * 24 * 60 * 60_000, // weekly (yield/payout change slowly)
+  fred_macro: 24 * 60 * 60_000,          // daily (rates move daily; other series slower)
 };

 /** Parse 'source:kind:id...' into { source, kind, id } (id may contain colons). */
@@ -73,19 +101,48 @@ function tsAgeMs(ts: string | null, now: number): number {
 const quoteHandler: KindHandler = {
  ttlClass: 'live_quote',
  read(d, symbol) {
-    const r = d.prepare('SELECT symbol,price,bid,ask,change,change_pct,iv,observed_at FROM quotes WHERE symbol=?').get(symbol) as Record<string, unknown> | undefined;
+    const r = d.prepare(
+      'SELECT symbol,price,bid,ask,change,change_pct,iv,session,regular_price,observed_at FROM quotes WHERE symbol=?',
+    ).get(symbol) as Record<string, unknown> | undefined;
    if (!r) return null;
+    const sessionRaw = r.session == null ? null : String(r.session);
+    const session =
+      sessionRaw === 'regular' || sessionRaw === 'pre' || sessionRaw === 'post'
+        ? sessionRaw
+        : null;
    return {
-      value: { symbol: r.symbol, price: r.price, bid: r.bid, ask: r.ask, change: r.change, changePercent: r.change_pct, iv: r.iv } as Quote,
+      value: {
+        symbol: r.symbol,
+        price: r.price,
+        bid: r.bid,
+        ask: r.ask,
+        change: r.change,
+        changePercent: r.change_pct,
+        iv: r.iv,
+        session,
+        regularPrice: r.regular_price as number | null | undefined,
+      } as Quote,
      stalenessTs: r.observed_at as string,
    };
  },
  write(d, symbol, value, provenance) {
    const q = value as Quote;
-    d.prepare('INSERT OR REPLACE INTO quotes (symbol,price,bid,ask,change,change_pct,iv,observed_at) VALUES (?,?,?,?,?,?,?,?)')
-      .run(symbol, q.price, q.bid ?? null, q.ask ?? null, q.change ?? null, q.changePercent ?? null, q.iv ?? null, provenance.fetchedAt);
+    d.prepare(
+      'INSERT OR REPLACE INTO quotes (symbol,price,bid,ask,change,change_pct,iv,session,regular_price,observed_at) VALUES (?,?,?,?,?,?,?,?,?,?)',
+    ).run(
+      symbol,
+      q.price,
+      q.bid ?? null,
+      q.ask ?? null,
+      q.change ?? null,
+      q.changePercent ?? null,
+      q.iv ?? null,
+      q.session ?? null,
+      q.regularPrice ?? null,
+      provenance.fetchedAt,
+    );
  },
-  isStale(ts, now) { return tsAgeMs(ts, now) > TTL_MS.live_quote; },
+  isStale(ts, now) { return tsAgeMs(ts, now) > quoteTtlMs(new Date(now)); },
 };

 const candlesHandler: KindHandler = {
@@ -96,16 +153,53 @@ const candlesHandler: KindHandler = {
    const rows = d.prepare('SELECT ts,o,h,l,c,v,adj_close,observed_at FROM price_candles WHERE symbol=? AND timeframe=? ORDER BY ts ASC').all(symbol, timeframe) as Array<Record<string, unknown>>;
    if (!rows.length) return null;
    const value: PriceCandle[] = rows.map((r) => ({ ts: r.ts as string, o: r.o as number, h: r.h as number, l: r.l as number, c: r.c as number, v: r.v as number, adjClose: r.adj_close as number | null }));
-    return { value, stalenessTs: rows[rows.length - 1].observed_at as string };
+    // Staleness for SWR uses last *bar* ts (not observed_at) so schedule + get agree on lag.
+    return { value, stalenessTs: rows[rows.length - 1].ts as string };
  },
  write(d, id, value, provenance) {
    const [symbol, timeframe] = id.split(':');
    const ins = d.prepare('INSERT OR REPLACE INTO price_candles (symbol,timeframe,ts,o,h,l,c,v,adj_close,observed_at) VALUES (?,?,?,?,?,?,?,?,?,?)');
    for (const c of value as PriceCandle[]) ins.run(symbol, timeframe, c.ts, c.o, c.h, c.l, c.c, c.v, c.adjClose ?? null, provenance.fetchedAt);
  },
-  isStale(ts) { return ts === null; }, // permanent: stale only when absent
+  // Do not thrash on every read: schedule owns EOD refresh. Stale only when absent.
+  isStale(ts) { return ts === null; },
 };

+/** True when daily candles are missing or last bar is older than CANDLE_FRESH_MS. */
+export function needsCandleRefresh(d: DatabaseSync, symbol: string, timeframe = '1d', now = Date.now()): boolean {
+  const row = d.prepare(
+    'SELECT MAX(ts) AS last_ts FROM price_candles WHERE symbol=? AND timeframe=?',
+  ).get(symbol, timeframe) as { last_ts: string | null } | undefined;
+  if (!row?.last_ts) return true;
+  return tsAgeMs(row.last_ts, now) > CANDLE_FRESH_MS;
+}
+
+/** Last daily candle bar timestamp, or null if none. */
+export function lastCandleTs(d: DatabaseSync, symbol: string, timeframe = '1d'): string | null {
+  const row = d.prepare(
+    'SELECT MAX(ts) AS last_ts FROM price_candles WHERE symbol=? AND timeframe=?',
+  ).get(symbol, timeframe) as { last_ts: string | null } | undefined;
+  return row?.last_ts ?? null;
+}
+
+/** True when quote is missing or past session-aware TTL. */
+export function needsQuoteRefresh(d: DatabaseSync, symbol: string, now = Date.now()): boolean {
+  const row = d.prepare('SELECT observed_at FROM quotes WHERE symbol=?').get(symbol) as { observed_at: string } | undefined;
+  if (!row?.observed_at) return true;
+  return tsAgeMs(row.observed_at, now) > quoteTtlMs(new Date(now));
+}
+
+/** True when symbol meta missing, incomplete (no name), or past weekly TTL. */
+export function needsSymbolMetaRefresh(d: DatabaseSync, symbol: string, now = Date.now()): boolean {
+  const row = d.prepare('SELECT name, sector, updated_at FROM symbols WHERE symbol=?').get(symbol) as
+    | { name: string | null; sector: string | null; updated_at: string | null }
+    | undefined;
+  if (!row?.updated_at) return true;
+  const incomplete = !row.name;
+  const ttl = incomplete ? SYMBOL_META_INCOMPLETE_TTL_MS : TTL_MS.symbol_meta;
+  return tsAgeMs(row.updated_at, now) > ttl;
+}
+
 const adjustmentsHandler: KindHandler = {
  ttlClass: 'daily_permanent',
  read(d, symbol) {
@@ -138,7 +232,10 @@ const symbolHandler: KindHandler = {
    d.prepare('INSERT OR REPLACE INTO symbols (symbol,name,sector,industry,exchange,ticker_kind,peers,updated_at) VALUES (?,?,?,?,?,?,?,?)')
      .run(symbol, s.name ?? null, s.sector ?? null, s.industry ?? null, s.exchange ?? null, s.tickerKind, s.peers ? JSON.stringify(s.peers) : null, provenance.fetchedAt);
  },
-  isStale(ts, now) { return tsAgeMs(ts, now) > TTL_MS.symbol_meta; },
+  isStale(ts, now) {
+    // Age-only here; incomplete rows (null name) are handled in needsSymbolMetaRefresh + get override.
+    return tsAgeMs(ts, now) > TTL_MS.symbol_meta;
+  },
 };

 // ----- Options handlers (slice 15) -----
@@ -178,10 +275,15 @@ const optionsChainHandler: KindHandler = {
    for (const r of rows) {
      const strike = typeof r.strike === 'number' ? r.strike : 0;
      const right = r.right === 'put' ? 'put' : 'call';
+      // OptionChainRow nests greeks; accept flat fields too for older writers.
+      const g = (r.greeks && typeof r.greeks === 'object')
+        ? (r.greeks as Record<string, unknown>)
+        : null;
      ins.run(
        symbol, expiry, strike, right,
        numOrNull(r.bid), numOrNull(r.ask), numOrNull(r.impliedVolatility),
-        numOrNull(r.delta), numOrNull(r.gamma), numOrNull(r.theta), numOrNull(r.vega),
+        numOrNull(g?.delta ?? r.delta), numOrNull(g?.gamma ?? r.gamma),
+        numOrNull(g?.theta ?? r.theta), numOrNull(g?.vega ?? r.vega),
        numOrNull(r.openInterest), numOrNull(r.volume),
        provenance.fetchedAt
      );
@@ -190,6 +292,33 @@ const optionsChainHandler: KindHandler = {
  isStale(ts, now) { return tsAgeMs(ts, now) > TTL_MS.options_snapshot; },
 };

+// Helper: LRU cache-aware read wrapper for handlers that use kv_cache
+function withKvCacheRead(
+  handler: KindHandler,
+  getCacheKey: (id: string) => string | null,
+): KindHandler {
+  return {
+    ...handler,
+    read(d, id) {
+      const cacheKey = getCacheKey(id);
+      if (!cacheKey) return handler.read(d, id);
+      
+      // Check LRU cache first
+      const cached = kvReadCache.get(cacheKey);
+      if (cached) return cached as { value: unknown; stalenessTs: string } | null;
+      
+      const result = handler.read(d, id);
+      
+      // Populate LRU cache on hit
+      if (result && !cacheKey.includes(':*')) {
+        kvReadCache.set(cacheKey, result.value as unknown, result.stalenessTs);
+      }
+      
+      return result;
+    },
+  };
+}
+
 const optionsExpiryDatesHandler: KindHandler = {
  ttlClass: 'intraday',
  read(d, symbol) {
@@ -261,9 +390,13 @@ function parseGreeksId(id: string): { symbol: string; expiry: string; strike: st
 const fetchHandler: KindHandler = {
  ttlClass: 'daily_permanent',
  read(d, id) {
-    const r = d.prepare('SELECT value, observed_at FROM kv_cache WHERE key=?').get(`sec-fetch:${id}`) as { value: string; observed_at: string } | undefined;
+    const cacheKey = `sec-fetch:${id}`;
+    const r = d.prepare('SELECT value, observed_at FROM kv_cache WHERE key=?').get(cacheKey) as { value: string; observed_at: string } | undefined;
    if (!r) return null;
-    try { return { value: JSON.parse(r.value), stalenessTs: r.observed_at }; } catch { return null; }
+    try {
+      const value = JSON.parse(r.value);
+      return { value, stalenessTs: r.observed_at };
+    } catch { return null; }
  },
  write(d, id, value, provenance) {
    d.prepare('INSERT OR REPLACE INTO kv_cache (key, value, observed_at) VALUES (?,?,?)').run(`sec-fetch:${id}`, JSON.stringify(value), provenance.fetchedAt);
@@ -292,9 +425,13 @@ const lintInsidersHandler: KindHandler = {
 const shortInterestHandler: KindHandler = {
  ttlClass: 'short_interest',
  read(d, id) {
-    const r = d.prepare('SELECT value, observed_at FROM kv_cache WHERE key=?').get(`yfinance:shortinterest:${id}`) as { value: string; observed_at: string } | undefined;
+    const cacheKey = `yfinance:shortinterest:${id}`;
+    const r = d.prepare('SELECT value, observed_at FROM kv_cache WHERE key=?').get(cacheKey) as { value: string; observed_at: string } | undefined;
    if (!r) return null;
-    try { return { value: JSON.parse(r.value), stalenessTs: r.observed_at }; } catch { return null; }
+    try {
+      const value = JSON.parse(r.value);
+      return { value, stalenessTs: r.observed_at };
+    } catch { return null; }
  },
  write(d, id, value, provenance) {
    d.prepare('INSERT OR REPLACE INTO kv_cache (key, value, observed_at) VALUES (?,?,?)').run(`yfinance:shortinterest:${id}`, JSON.stringify(value), provenance.fetchedAt);
@@ -305,9 +442,13 @@ const shortInterestHandler: KindHandler = {
 const nasdaqShortinterestHandler: KindHandler = {
  ttlClass: 'short_interest',
  read(d, id) {
-    const r = d.prepare('SELECT value, observed_at FROM kv_cache WHERE key=?').get(`nasdaq:nasdaqShortinterest:${id}`) as { value: string; observed_at: string } | undefined;
+    const cacheKey = `nasdaq:nasdaqShortinterest:${id}`;
+    const r = d.prepare('SELECT value, observed_at FROM kv_cache WHERE key=?').get(cacheKey) as { value: string; observed_at: string } | undefined;
    if (!r) return null;
-    try { return { value: JSON.parse(r.value), stalenessTs: r.observed_at }; } catch { return null; }
+    try {
+      const value = JSON.parse(r.value);
+      return { value, stalenessTs: r.observed_at };
+    } catch { return null; }
  },
  write(d, id, value, provenance) {
    d.prepare('INSERT OR REPLACE INTO kv_cache (key, value, observed_at) VALUES (?,?,?)').run(`nasdaq:nasdaqShortinterest:${id}`, JSON.stringify(value), provenance.fetchedAt);
@@ -346,32 +487,143 @@ const finraShortInterestHandler: KindHandler = {
  },
 };

+const finraSiBiweeklyHandler: KindHandler = {
+  ttlClass: 'short_interest',
+  read(d, symbol) {
+    const r = d.prepare(
+      `SELECT current_short_position, previous_short_position, avg_daily_volume, days_to_cover, change_percent, settlement_date, ingested_at
+       FROM finra_short_interest_biweekly WHERE symbol=? ORDER BY settlement_date DESC LIMIT 1`
+    ).get(symbol) as Record<string, unknown> | undefined;
+    if (!r) return null;
+    return {
+      value: {
+        currentShortPosition: r.current_short_position as number,
+        previousShortPosition: r.previous_short_position as number | null,
+        avgDailyVolume: r.avg_daily_volume as number | null,
+        daysToCover: r.days_to_cover as number | null,
+        changePercent: r.change_percent as number | null,
+        settlementDate: r.settlement_date as string,
+      },
+      stalenessTs: r.ingested_at as string,
+    };
+  },
+  write(_d, _symbol, _value, _provenance) {},
+  isStale(ts, now) {
+    if (ts === null) return false;
+    return tsAgeMs(ts, now) > TTL_MS.short_interest;
+  },
+};
+
+const dividendFundamentalsHandler: KindHandler = {
+  ttlClass: 'dividend_fundamentals',
+  read(d, symbol) {
+    const r = d.prepare(
+      `SELECT dividend_yield, payout_ratio, dividend_rate, ex_dividend_date, trailing_eps, forward_pe, growth_streak_years, observed_at FROM dividend_fundamentals WHERE symbol=?`
+    ).get(symbol) as Record<string, unknown> | undefined;
+    if (!r) return null;
+    return { value: { symbol, dividendYield: r.dividend_yield, payoutRatio: r.payout_ratio, dividendRate: r.dividend_rate, exDividendDate: r.ex_dividend_date, trailingEps: r.trailing_eps, forwardPE: r.forward_pe, growthStreakYears: r.growth_streak_years }, stalenessTs: r.observed_at as string };
+  },
+  write(d, symbol, value, provenance) {
+    d.prepare('INSERT OR REPLACE INTO dividend_fundamentals (symbol, dividend_yield, payout_ratio, dividend_rate, ex_dividend_date, trailing_eps, forward_pe, growth_streak_years, observed_at) VALUES (?,?,?,?,?,?,?,?)').run(symbol, (value as any).dividendYield ?? null, (value as any).payoutRatio ?? null, (value as any).dividendRate ?? null, (value as any).exDividendDate ?? null, (value as any).trailingEps ?? null, (value as any).forwardPE ?? null, (value as any).growthStreakYears ?? null, provenance.fetchedAt);
+  },
+  isStale(ts, now) { return ts ? tsAgeMs(ts, now) > 7 * 86400000 : true; },
+};
+
+/** ETF composition — same kv key as market.sectorHoldings for cache sharing. */
+const topHoldingsHandler: KindHandler = {
+  ttlClass: 'symbol_meta',
+  read(d, symbol) {
+    const key = `etf:topHoldings:${symbol.toUpperCase()}`;
+    const r = d.prepare('SELECT value, observed_at FROM kv_cache WHERE key=?').get(key) as
+      | { value: string; observed_at: string }
+      | undefined;
+    if (!r?.value) return null;
+    try {
+      const parsed = JSON.parse(r.value) as { holdings?: unknown };
+      return { value: parsed, stalenessTs: r.observed_at };
+    } catch {
+      return null;
+    }
+  },
+  write(d, symbol, value, provenance) {
+    const key = `etf:topHoldings:${symbol.toUpperCase()}`;
+    const payload = typeof value === 'string' ? value : JSON.stringify(value);
+    d.prepare('INSERT OR REPLACE INTO kv_cache (key, value, observed_at) VALUES (?,?,?)')
+      .run(key, payload, provenance.fetchedAt);
+  },
+  isStale(ts, now) { return tsAgeMs(ts, now) > TTL_MS.symbol_meta; },
+};
+
+// Wrap kv_cache-reading handlers with LRU cache
+const wrappedOptionsExpiryDates = withKvCacheRead(optionsExpiryDatesHandler, (id) => `options_expiry:${id}`);
+const wrappedFetch = withKvCacheRead(fetchHandler, (id) => `sec-fetch:${id}`);
+const wrappedShortInterest = withKvCacheRead(shortInterestHandler, (id) => `yfinance:shortinterest:${id}`);
+const wrappedNasdaqShortInterest = withKvCacheRead(nasdaqShortinterestHandler, (id) => `nasdaq:nasdaqShortinterest:${id}`);
+const wrappedTopHoldings = withKvCacheRead(topHoldingsHandler, (symbol) => `etf:topHoldings:${(symbol as string).toUpperCase()}`);
+
+/** FRED macro series — kv_cache keyed `fred:series:{seriesId}` (warmed by the queue, off request path). */
+const seriesHandler: KindHandler = {
+  ttlClass: 'fred_macro',
+  read(d, seriesId) {
+    const cacheKey = `fred:series:${seriesId}`;
+    const r = d.prepare('SELECT value, observed_at FROM kv_cache WHERE key=?').get(cacheKey) as
+      | { value: string; observed_at: string }
+      | undefined;
+    if (!r?.value) return null;
+    try {
+      return { value: JSON.parse(r.value), stalenessTs: r.observed_at };
+    } catch { return null; }
+  },
+  write(d, seriesId, value, provenance) {
+    d.prepare('INSERT OR REPLACE INTO kv_cache (key, value, observed_at) VALUES (?,?,?)')
+      .run(`fred:series:${seriesId}`, JSON.stringify(value), provenance.fetchedAt);
+  },
+  isStale(ts, now) { return tsAgeMs(ts, now) > TTL_MS.fred_macro; },
+};
+
 const HANDLERS = new Map<string, KindHandler>([
  ['quote', quoteHandler],
  ['candles', candlesHandler],
  ['symbol', symbolHandler],
  ['adjustments', adjustmentsHandler],
  ['chain', optionsChainHandler],
-  ['expiry_dates', optionsExpiryDatesHandler],
+  ['expiry_dates', wrappedOptionsExpiryDates],
  ['greeks', greeksHandler],
-  ['fetch', fetchHandler],
+  ['fetch', wrappedFetch],
  ['holders', lintHoldersHandler],
  ['insiders', lintInsidersHandler],
-  ['shortinterest', shortInterestHandler],
-  ['nasdaqShortinterest', nasdaqShortinterestHandler],
+  ['shortinterest', wrappedShortInterest],
+  ['nasdaqShortinterest', wrappedNasdaqShortInterest],
  ['finraShortinterest', finraShortInterestHandler],
+  ['finraSiBiweekly', finraSiBiweeklyHandler],
+  ['dividendFundamentals', dividendFundamentalsHandler],
+  ['topHoldings', wrappedTopHoldings],
+  ['series', seriesHandler],
 ]);

 export interface CacheRepository {
  get<T>(key: CacheKey): Promise<CacheEntry<T>>;
  set<T>(key: CacheKey, value: T, ttlClass: TtlClass, provenance: Provenance): Promise<void>;
  stale(key: CacheKey): boolean;
+  /**
+   * Watchlist/portfolio ownership: bumps refcount and seeds cache on first demand.
+   * Do NOT call from page-view paths (rotation, seasonality, condition) — use ensureInDemand.
+   */
  subscribe(symbol: string, tickerKind: TickerKind): Promise<void>;
  unsubscribe(symbol: string): Promise<void>;
+  /**
+   * Request-path pin: mark symbol in demand + queue missing/stale data without inflating refcount.
+   * Safe to call on every Market Outlook / ticker context load.
+   */
+  ensureInDemand(symbol: string, tickerKind: TickerKind): Promise<void>;
+  /** Permanent system pin (rotation universe, SPY, VIX) — survives unsubscribe. */
+  pinSystemSymbol(symbol: string, tickerKind: TickerKind): Promise<void>;
  demandSet(): Promise<string[]>;
  getMany<T>(keys: CacheKey[]): Promise<Array<{ key: CacheKey; value: T | null; isStale: boolean }>>;
  /** Delete a cache entry by key (or, for wildcard keys ending in `:*`, all matching entries). */
  del(key: CacheKey): Promise<void>;
+  /** Underlying DB for schedule TTL checks (queue only). */
+  readonly db: DatabaseSync;
 }

 export class CacheRepositoryImpl implements CacheRepository {
@@ -381,6 +633,7 @@ export class CacheRepositoryImpl implements CacheRepository {
    this._db = opts.db;
    this._scheduler = opts.scheduler;
  }
+  get db(): DatabaseSync { return this._db; }
  private handler(kind: string): KindHandler {
    const h = HANDLERS.get(kind);
    if (!h) throw new Error(`unknown cache kind: ${kind}`);
@@ -391,7 +644,14 @@ export class CacheRepositoryImpl implements CacheRepository {
    const h = this.handler(kind);
    const row = h.read(this._db, id);
    const now = Date.now();
-    const stale = h.isStale(row ? row.stalenessTs : null, now);
+    let stale = h.isStale(row ? row.stalenessTs : null, now);
+    // Incomplete symbol meta (null name) is always treated as stale for SWR re-fetch.
+    if (kind === 'symbol' && row) {
+      const meta = row.value as SymbolMeta;
+      if (!meta?.name && tsAgeMs(row.stalenessTs, now) > SYMBOL_META_INCOMPLETE_TTL_MS) {
+        stale = true;
+      }
+    }
    if (stale) {
      try { await this._scheduler.queue(key); } catch { /* background refresh; never block readers */ }
    }
@@ -406,33 +666,94 @@ export class CacheRepositoryImpl implements CacheRepository {
    const h = this.handler(kind);
    if (h.ttlClass !== ttlClass) throw new Error(`ttlClass mismatch for kind '${kind}': expected ${h.ttlClass}, got ${ttlClass}`);
    h.write(this._db, id, value, provenance);
+    // Touch demand last_refreshed when a quote lands.
+    if (kind === 'quote') {
+      try {
+        this._db.prepare('UPDATE symbol_demand SET last_refreshed_at=? WHERE symbol=?').run(provenance.fetchedAt, id);
+      } catch { /* ignore */ }
+    }
  }
  stale(key: CacheKey): boolean {
    const { kind, id } = parseCacheKey(key);
    const h = this.handler(kind);
    const row = h.read(this._db, id);
+    if (kind === 'symbol' && row) {
+      const meta = row.value as SymbolMeta;
+      if (!meta?.name) return true;
+    }
    return h.isStale(row ? row.stalenessTs : null, Date.now());
  }
+
+  private ensureDemandRow(symbol: string, tickerKind: TickerKind): void {
+    this._db.prepare(
+      'INSERT OR IGNORE INTO symbol_demand (symbol,refcount,ticker_kind,in_demand,last_refreshed_at,system_pin) VALUES (?,?,?,?,?,?)',
+    ).run(symbol, 0, tickerKind, 1, null, 0);
+  }
+
+  /** Queue only what is missing or past TTL — no refcount change. */
+  private async queueIfNeeded(symbol: string): Promise<void> {
+    const d = this._db;
+    if (needsQuoteRefresh(d, symbol)) {
+      try { await this._scheduler.queue(`yfinance:quote:${symbol}`); } catch { /* ignore */ }
+    }
+    if (needsCandleRefresh(d, symbol)) {
+      try { await this._scheduler.queue(`yfinance:candles:${symbol}:1d`); } catch { /* ignore */ }
+    }
+    if (needsSymbolMetaRefresh(d, symbol)) {
+      try { await this._scheduler.queue(`yfinance:symbol:${symbol}`); } catch { /* ignore */ }
+    }
+  }
+
  async subscribe(symbol: string, tickerKind: TickerKind): Promise<void> {
    const d = this._db;
    d.prepare('INSERT OR IGNORE INTO symbol_demand (symbol,refcount,ticker_kind,in_demand,last_refreshed_at) VALUES (?,?,?,?,?)').run(symbol, 0, tickerKind, 1, null);
    const prev = d.prepare('SELECT refcount FROM symbol_demand WHERE symbol=?').get(symbol) as { refcount: number } | undefined;
    const before = prev?.refcount ?? 0;
-    d.prepare('UPDATE symbol_demand SET refcount = refcount + 1, in_demand = 1 WHERE symbol=?').run(symbol);
+    d.prepare('UPDATE symbol_demand SET refcount = refcount + 1, in_demand = 1, ticker_kind=? WHERE symbol=?').run(tickerKind, symbol);
    if (before === 0) {
-      // First demand: schedule initial cache population (slice 1: yfinance quote + symbol meta + candles + adjustments)
-      for (const k of [`yfinance:quote:${symbol}`, `yfinance:symbol:${symbol}`, `yfinance:candles:${symbol}:1d`, `yfinance:adjustments:${symbol}`, `yfinance:shortinterest:${symbol}`, `nasdaq:nasdaqShortinterest:${symbol}`]) {
+      // First user demand: full seed once (not every schedule tick).
+      for (const k of [
+        `yfinance:quote:${symbol}`,
+        `yfinance:symbol:${symbol}`,
+        `yfinance:candles:${symbol}:1d`,
+        `yfinance:adjustments:${symbol}`,
+        `yfinance:shortinterest:${symbol}`,
+        `nasdaq:nasdaqShortinterest:${symbol}`,
+      ]) {
        try { await this._scheduler.queue(k); } catch { /* ignore */ }
      }
+    } else {
+      // Subsequent demand: re-check staleness and queue missing kinds (no refcount change).
+      await this.queueIfNeeded(symbol);
    }
  }
+
+  async ensureInDemand(symbol: string, tickerKind: TickerKind): Promise<void> {
+    this.ensureDemandRow(symbol, tickerKind);
+    this._db.prepare('UPDATE symbol_demand SET in_demand = 1, ticker_kind=COALESCE(ticker_kind, ?) WHERE symbol=?').run(tickerKind, symbol);
+    await this.queueIfNeeded(symbol);
+  }
+
+  async pinSystemSymbol(symbol: string, tickerKind: TickerKind): Promise<void> {
+    this.ensureDemandRow(symbol, tickerKind);
+    this._db.prepare(
+      'UPDATE symbol_demand SET in_demand = 1, system_pin = 1, ticker_kind=? WHERE symbol=?',
+    ).run(tickerKind, symbol);
+    await this.queueIfNeeded(symbol);
+  }
+
  async unsubscribe(symbol: string): Promise<void> {
    const d = this._db;
    d.prepare('UPDATE symbol_demand SET refcount = MAX(refcount - 1, 0) WHERE symbol=?').run(symbol);
-    d.prepare('UPDATE symbol_demand SET in_demand = 0 WHERE symbol=? AND refcount = 0').run(symbol);
+    // Keep system pins in demand even when no user owns them.
+    d.prepare(
+      'UPDATE symbol_demand SET in_demand = 0 WHERE symbol=? AND refcount = 0 AND COALESCE(system_pin, 0) = 0',
+    ).run(symbol);
  }
  async demandSet(): Promise<string[]> {
-    return (this._db.prepare('SELECT symbol FROM symbol_demand WHERE refcount > 0 ORDER BY symbol').all() as Array<{ symbol: string }>).map((r) => r.symbol);
+    return (this._db.prepare(
+      'SELECT symbol FROM symbol_demand WHERE in_demand = 1 OR COALESCE(system_pin, 0) = 1 ORDER BY symbol',
+    ).all() as Array<{ symbol: string }>).map((r) => r.symbol);
  }
  async getMany<T>(keys: CacheKey[]): Promise<Array<{ key: CacheKey; value: T | null; isStale: boolean }>> {
    return Promise.all(keys.map(async (key) => {
@@ -440,6 +761,14 @@ export class CacheRepositoryImpl implements CacheRepository {
      return { key, value: e.value, isStale: e.isStale };
    }));
  }
+  private clearKvCachePrefix(prefix: string): void {
+    // Invalidate all LRU cache entries matching the prefix
+    for (const key of kvReadCache.keys()) {
+      if (key.startsWith(prefix)) {
+        kvReadCache.delete(key);
+      }
+    }
+  }
  async del(key: CacheKey): Promise<void> {
    const { source, kind, id } = parseCacheKey(key);
    const d = this._db;
@@ -464,22 +793,41 @@ export class CacheRepositoryImpl implements CacheRepository {
        d.prepare('DELETE FROM options_chains WHERE symbol=? AND expiry=?').run(symbol, expiry);
        break;
      }
-      case 'expiry_dates':
-        d.prepare('DELETE FROM kv_cache WHERE key=?').run(`options_expiry:${id}`);
+      case 'expiry_dates': {
+        const cacheKey = `options_expiry:${id}`;
+        d.prepare('DELETE FROM kv_cache WHERE key=?').run(cacheKey);
+        kvReadCache.delete(cacheKey);
        break;
-      case 'shortinterest':
-        d.prepare('DELETE FROM kv_cache WHERE key=?').run(`yfinance:shortinterest:${id}`);
+      }
+      case 'shortinterest': {
+        const cacheKey = `yfinance:shortinterest:${id}`;
+        d.prepare('DELETE FROM kv_cache WHERE key=?').run(cacheKey);
+        kvReadCache.delete(cacheKey);
        break;
-      case 'nasdaqShortinterest':
-        d.prepare('DELETE FROM kv_cache WHERE key=?').run(`nasdaq:nasdaqShortinterest:${id}`);
+      }
+      case 'nasdaqShortinterest': {
+        const cacheKey = `nasdaq:nasdaqShortinterest:${id}`;
+        d.prepare('DELETE FROM kv_cache WHERE key=?').run(cacheKey);
+        kvReadCache.delete(cacheKey);
        break;
+      }
      case 'finraShortinterest':
        d.prepare('DELETE FROM finra_short_interest WHERE symbol=?').run(id);
        break;
+      case 'finraSiBiweekly':
+        d.prepare('DELETE FROM finra_short_interest_biweekly WHERE symbol=?').run(id);
+        break;
+      case 'dividendFundamentals':
+        d.prepare('DELETE FROM dividend_fundamentals WHERE symbol=?').run(id);
+        break;
      default: {
        // Unknown/wildcard kind (e.g. `x:cashtag:*`): best-effort delete from kv_cache via LIKE.
        const like = key.endsWith(':*') ? `${key.slice(0, -1)}%` : key;
        d.prepare('DELETE FROM kv_cache WHERE key LIKE ?').run(like);
+        // Invalidate LRU cache for this prefix
+        if (key.endsWith(':*')) {
+          this.clearKvCachePrefix(key.slice(0, -1));
+        }
      }
    }
    void source;