lakehouse/mcp-server/quotes.ts

// Multi-source quote aggregation + local hourly cache (2026-06-18).
//
// Why this exists: a single quote source can be confidently wrong and
// give you no way to know (Stooq died silently; Yahoo's chartPreviousClose
// over a 5d window references the pre-window close, so a Yahoo-only day
// change reads +1.5% when the real figure is -1.1%). The fix isn't a
// "better" single source — it's CONSENSUS. Pull the same ticker from
// several independent feeds, take the median, and score confidence by how
// many of them corroborate it. Disagreement becomes a signal, not a
// hidden bug. This is the differentiator: the index's numbers come with a
// provenance + confidence trail, not a single opaque vendor call.
//
// Real-time is not a requirement here (per product call): an hourly
// ingest job writes aggregates into a local SQLite time-series, and the
// serving path reads the cache. Cold tickers are aggregated on-demand and
// cached so the page never blocks on a missing symbol.
//
// Sources (all keyless from this host, verified 2026-06-18):
//   - Yahoo Finance v8 chart  (query1.finance.yahoo.com)
//   - Nasdaq quote API        (api.nasdaq.com)
//   - CNBC quote webservice   (quote.cnbc.com)

import { Database } from "bun:sqlite";

// Consumer finance endpoints (Nasdaq/CNBC) gate on a browser-shaped UA —
// the custom courtesy UA gets intermittently dropped. Low-volume hourly
// polling of public no-auth JSON, so a standard browser UA is the right
// call for reliability.
const UA =
  "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0 Safari/537.36";

// A source counts as "corroborating" the consensus if its day-change is
// within this many percentage points of the median. Legit feeds differ by
// timing/rounding (<0.1pp); a stale or wrong-baseline feed lands far
// outside and is flagged as an outlier.
const TOL_PP = 0.2;
// Confidence is scored against this many target sources, so a single
// source can never read as "fully confident" — corroboration needs peers.
const TARGET_SOURCES = 3;
// Serving marks a cached aggregate "stale" past this age (the hourly job
// should keep everything fresher than this); still served, just flagged.
export const STALE_AFTER_MS = 90 * 60 * 1000;

const DB_PATH = `${import.meta.dir}/../data/_quote_cache/quotes.db`;

export type SourceQuote = {
  source: string;
  price: number;
  day_change_pct: number;
  price_date?: string;
};

export type AggregateSource = SourceQuote & { outlier: boolean };

export type AggregateQuote = {
  ticker: string;
  price: number; // median price across corroborating sources
  day_change_pct: number; // median day change across sources
  confidence: number; // 0-100 — corroborating sources / TARGET_SOURCES
  source_count: number;
  spread_pp: number; // max-min day change across sources (dispersion)
  sources: AggregateSource[];
  price_date?: string;
  ts: number; // unix ms when aggregated
};

// ── per-source fetchers ─────────────────────────────────────────────
// Each returns a normalized SourceQuote or null (never throws). A null
// just means "this feed didn't answer for this symbol" — the aggregator
// works with whatever subset responds.

const num = (s: any): number | null => {
  if (typeof s === "number") return Number.isFinite(s) ? s : null;
  if (typeof s !== "string") return null;
  const n = parseFloat(s.replace(/[$,%\s]/g, ""));
  return Number.isFinite(n) ? n : null;
};

async function fromYahoo(ticker: string): Promise<SourceQuote | null> {
  try {
    const res = await fetch(
      `https://query1.finance.yahoo.com/v8/finance/chart/${encodeURIComponent(ticker.toUpperCase())}?range=5d&interval=1d`,
      { headers: { "User-Agent": UA }, signal: AbortSignal.timeout(8000) },
    );
    if (!res.ok) return null;
    const j: any = await res.json();
    const r = j?.chart?.result?.[0];
    const price = r?.meta?.regularMarketPrice;
    if (typeof price !== "number") return null;
    // Prev close = the daily close BEFORE the latest candle. meta's
    // chartPreviousClose is the pre-window close over range=5d (wrong
    // baseline), so derive from the candle array instead.
    const closes: number[] = (r?.indicators?.quote?.[0]?.close ?? []).filter(
      (c: any) => typeof c === "number",
    );
    const prev = closes.length >= 2 ? closes[closes.length - 2] : undefined;
    if (!prev || prev <= 0) return null;
    const t = typeof r?.meta?.regularMarketTime === "number" ? r.meta.regularMarketTime : null;
    return {
      source: "yahoo",
      price,
      day_change_pct: ((price - prev) / prev) * 100,
      price_date: t ? new Date(t * 1000).toISOString().slice(0, 10) : undefined,
    };
  } catch {
    return null;
  }
}

async function fromNasdaq(ticker: string): Promise<SourceQuote | null> {
  try {
    const res = await fetch(
      `https://api.nasdaq.com/api/quote/${encodeURIComponent(ticker.toUpperCase())}/info?assetclass=stocks`,
      { headers: { "User-Agent": UA, Accept: "application/json" }, signal: AbortSignal.timeout(8000) },
    );
    if (!res.ok) return null;
    const j: any = await res.json();
    const p = j?.data?.primaryData;
    const price = num(p?.lastSalePrice);
    const pct = num(p?.percentageChange);
    if (price == null || pct == null) return null;
    return { source: "nasdaq", price, day_change_pct: pct };
  } catch {
    return null;
  }
}

async function fromCnbc(ticker: string): Promise<SourceQuote | null> {
  try {
    const res = await fetch(
      `https://quote.cnbc.com/quote-html-webservice/restQuote/symbolType/symbol?symbols=${encodeURIComponent(ticker.toUpperCase())}&requestMethod=itv&output=json`,
      { headers: { "User-Agent": UA }, signal: AbortSignal.timeout(8000) },
    );
    if (!res.ok) return null;
    const j: any = await res.json();
    const q = j?.FormattedQuoteResult?.FormattedQuote?.[0];
    const price = num(q?.last);
    const pct = num(q?.change_pct);
    if (price == null || pct == null) return null;
    return { source: "cnbc", price, day_change_pct: pct, price_date: q?.last_time };
  } catch {
    return null;
  }
}

const FETCHERS = [fromYahoo, fromNasdaq, fromCnbc];

// ── aggregation ─────────────────────────────────────────────────────

function median(xs: number[]): number {
  const s = [...xs].sort((a, b) => a - b);
  const m = Math.floor(s.length / 2);
  return s.length % 2 ? s[m] : (s[m - 1] + s[m]) / 2;
}

// Fetch every source in parallel and fuse into one consensus quote.
// Returns null only if NO source answered.
export async function aggregateTicker(ticker: string): Promise<AggregateQuote | null> {
  const got = (await Promise.all(FETCHERS.map((f) => f(ticker)))).filter(
    (q): q is SourceQuote => q != null,
  );
  if (!got.length) return null;

  const changes = got.map((q) => q.day_change_pct);
  const consensusChange = median(changes);
  const consensusPrice = median(got.map((q) => q.price));
  const spread = Math.max(...changes) - Math.min(...changes);

  const sources: AggregateSource[] = got.map((q) => ({
    ...q,
    outlier: Math.abs(q.day_change_pct - consensusChange) > TOL_PP,
  }));
  // Confidence = how many independent sources corroborate the served
  // number, out of the target. 3 agree → 100; 2 of 3 → 67 (1 flagged);
  // a lone source → 33. Honest about thin or split evidence.
  const corroborating = sources.filter((s) => !s.outlier).length;
  const confidence = Math.round((100 * corroborating) / TARGET_SOURCES);

  return {
    ticker: ticker.toUpperCase(),
    price: consensusPrice,
    day_change_pct: consensusChange,
    confidence,
    source_count: got.length,
    spread_pp: spread,
    sources,
    price_date: got.find((q) => q.price_date)?.price_date,
    ts: Date.now(),
  };
}

// ── SQLite time-series cache ─────────────────────────────────────────

let _db: Database | null = null;
function db(): Database {
  if (_db) return _db;
  // Ensure the directory exists (Bun has no mkdir on Database open).
  const dir = DB_PATH.slice(0, DB_PATH.lastIndexOf("/"));
  try {
    require("node:fs").mkdirSync(dir, { recursive: true });
  } catch {}
  const d = new Database(DB_PATH, { create: true });
  d.exec("PRAGMA journal_mode = WAL;");
  // The serving path (gateway process) and the hourly ingest (timer
  // process) both write this db. WAL allows concurrent readers, but two
  // writers still serialize — busy_timeout makes the loser wait instead
  // of throwing SQLITE_BUSY.
  d.exec("PRAGMA busy_timeout = 5000;");
  d.exec(`
    CREATE TABLE IF NOT EXISTS quote_observations (
      ticker TEXT NOT NULL, ts INTEGER NOT NULL, source TEXT NOT NULL,
      price REAL, day_change_pct REAL,
      PRIMARY KEY (ticker, ts, source)
    );
    CREATE TABLE IF NOT EXISTS quote_aggregates (
      ticker TEXT NOT NULL, ts INTEGER NOT NULL,
      price REAL, day_change_pct REAL, confidence INTEGER,
      source_count INTEGER, spread_pp REAL, sources TEXT, price_date TEXT,
      PRIMARY KEY (ticker, ts)
    );
    CREATE INDEX IF NOT EXISTS idx_agg_ticker_ts ON quote_aggregates(ticker, ts DESC);
  `);
  _db = d;
  return d;
}

// One row per ticker per HOUR bucket — re-running within the hour
// overwrites (idempotent), so a manual re-trigger never double-counts.
function hourBucket(tsMs: number): number {
  return Math.floor(tsMs / 3_600_000) * 3_600_000;
}

export function writeAggregate(agg: AggregateQuote): void {
  const d = db();
  const bucket = hourBucket(agg.ts);
  const obs = d.prepare(
    "INSERT OR REPLACE INTO quote_observations (ticker, ts, source, price, day_change_pct) VALUES (?, ?, ?, ?, ?)",
  );
  const tx = d.transaction((a: AggregateQuote) => {
    for (const s of a.sources) obs.run(a.ticker, bucket, s.source, s.price, s.day_change_pct);
    d.prepare(
      `INSERT OR REPLACE INTO quote_aggregates
        (ticker, ts, price, day_change_pct, confidence, source_count, spread_pp, sources, price_date)
        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`,
    ).run(
      a.ticker,
      bucket,
      a.price,
      a.day_change_pct,
      a.confidence,
      a.source_count,
      a.spread_pp,
      JSON.stringify(a.sources),
      a.price_date ?? null,
    );
  });
  tx(agg);
}

function rowToAggregate(row: any): AggregateQuote {
  return {
    ticker: row.ticker,
    price: row.price,
    day_change_pct: row.day_change_pct,
    confidence: row.confidence,
    source_count: row.source_count,
    spread_pp: row.spread_pp,
    sources: JSON.parse(row.sources || "[]"),
    price_date: row.price_date ?? undefined,
    ts: row.ts,
  };
}

export function readCachedAggregate(ticker: string): AggregateQuote | null {
  const row = db()
    .prepare("SELECT * FROM quote_aggregates WHERE ticker = ? ORDER BY ts DESC LIMIT 1")
    .get(ticker.toUpperCase()) as any;
  return row ? rowToAggregate(row) : null;
}

// Distinct tickers currently tracked in the cache — the universe the
// hourly job refreshes when called with no explicit list.
export function trackedTickers(): string[] {
  return (db().prepare("SELECT DISTINCT ticker FROM quote_aggregates").all() as any[]).map(
    (r) => r.ticker,
  );
}

// ── serving + ingest ────────────────────────────────────────────────

// Serve aggregates for a request. Cache-first; tickers never seen are
// aggregated live ONCE and cached (so the page never blocks on a cold
// symbol), then flagged stale if the cached row is older than the hourly
// window. Refreshing warm tickers is the ingest job's responsibility, not
// the request path — keeps serving fast.
export async function serveQuotes(
  tickers: string[],
): Promise<Record<string, (AggregateQuote & { stale: boolean }) | null>> {
  const out: Record<string, (AggregateQuote & { stale: boolean }) | null> = {};
  const cold: string[] = [];
  for (const t of tickers) {
    const c = readCachedAggregate(t);
    if (c) out[t] = { ...c, stale: Date.now() - c.ts > STALE_AFTER_MS };
    else cold.push(t);
  }
  if (cold.length) {
    const fresh = await Promise.all(cold.map((t) => aggregateTicker(t)));
    for (let i = 0; i < cold.length; i++) {
      const a = fresh[i];
      if (a) {
        writeAggregate(a);
        out[cold[i]] = { ...a, stale: false };
      } else {
        out[cold[i]] = null;
      }
    }
  }
  return out;
}

// The hourly ingest. Aggregates each ticker from all sources and writes
// the snapshot. With no list, refreshes everything already tracked.
export async function ingestQuotes(
  tickers?: string[],
): Promise<{ requested: number; written: number; failed: string[] }> {
  const list = (tickers?.length ? tickers : trackedTickers()).map((t) => t.toUpperCase());
  const uniq = Array.from(new Set(list)).filter((t) => t && !t.includes("."));
  let written = 0;
  const failed: string[] = [];
  // Serial fan-out to stay polite to the free endpoints. One retry on a
  // total miss: all three fetches share a call, so a transient blip drops
  // them together (the most liquid ticker fails as easily as a thin one).
  for (const t of uniq) {
    let a = await aggregateTicker(t);
    if (!a) {
      await new Promise((r) => setTimeout(r, 600));
      a = await aggregateTicker(t);
    }
    if (a) {
      writeAggregate(a);
      written++;
    } else {
      failed.push(t);
    }
  }
  return { requested: uniq.length, written, failed };
}