From ab1f12296d48da3b0f665cc30a6352b4c0e9b9c6 Mon Sep 17 00:00:00 2001 From: root Date: Thu, 18 Jun 2026 01:28:28 -0500 Subject: [PATCH] profiler: multi-source quote aggregation with confidence scoring MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replaces the single-source quote feed with a 3-source consensus system, the differentiator for the index: pull each ticker from independent feeds, take the median, and score confidence by how many corroborate it. Disagreement becomes a signal instead of a silently-wrong number. Context: Stooq retired its free q/l/ CSV quote API (404s server-side; q/d/l behind a JS challenge), which broke the profiler basket. The interim Yahoo-only fix worked but had a hidden wrong-baseline bug — Yahoo's chartPreviousClose over a 5d window references the pre-window close, so day-change read +11.65% for CAT when the real figure is +1.11%. Cross-checking against Nasdaq + CNBC exposes exactly that class of quiet error, which a single source can never self-report. - quotes.ts — Yahoo + Nasdaq + CNBC fetchers (all keyless, browser UA; Yahoo prev-close now derived from the daily candle, not the buggy meta field), median consensus, confidence = corroborating sources / 3, outlier flagging, and a bun:sqlite hourly time-series cache. - quote_ingest.ts + ops/systemd/lakehouse-quote-ingest.{service,timer} — hourly snapshot into the cache (real-time not required per product call). Runs as root to match the mcp-server that co-owns the db; busy_timeout serializes the two writers. - /intelligence/ticker_quotes — serves cache-first (fast), aggregates cold tickers on-demand once and caches them. New /quote_ingest admin trigger. - profiler.html — per-card confidence badge (green/amber/red) + a per-source breakdown tooltip; "3-source consensus" labeling. - entity.ts — the earlier Stooq->Yahoo swap for the single-ticker brief (kept fetchStooqQuote dead-but-reversible). Verified live on devop.live/lakehouse/profiler: liquid names read 100% / 3-source consensus; thin microcaps (BLPG, ESHSF — Yahoo-only) honestly read 33% / 1-source. Full basket (17 tickers) seeded; hourly timer armed. Cache db (data/_quote_cache/) is gitignored. Co-Authored-By: Claude Opus 4.8 (1M context) --- mcp-server/entity.ts | 61 +++- mcp-server/index.ts | 72 +++-- mcp-server/profiler.html | 27 +- mcp-server/quote_ingest.ts | 31 ++ mcp-server/quotes.ts | 347 +++++++++++++++++++++ ops/systemd/install.sh | 2 + ops/systemd/lakehouse-quote-ingest.service | 17 + ops/systemd/lakehouse-quote-ingest.timer | 11 + 8 files changed, 531 insertions(+), 37 deletions(-) create mode 100644 mcp-server/quote_ingest.ts create mode 100644 mcp-server/quotes.ts create mode 100644 ops/systemd/lakehouse-quote-ingest.service create mode 100644 ops/systemd/lakehouse-quote-ingest.timer diff --git a/mcp-server/entity.ts b/mcp-server/entity.ts index 627dad0..15ec39d 100644 --- a/mcp-server/entity.ts +++ b/mcp-server/entity.ts @@ -1140,6 +1140,56 @@ export async function fetchStooqQuote(ticker: string): Promise<{ } } +// Yahoo Finance v8 chart API — the replacement quote source (2026-06-18). +// Stooq retired its free `q/l/` light-quote CSV endpoint (now 404s for +// non-browser clients; the historical `q/d/l/` CSV sits behind a JS +// browser-verification challenge), so fetchStooqQuote returns null for +// every symbol and the profiler basket renders no quotes. The v8 chart +// endpoint is free, key-less, single-symbol JSON and exposes everything +// the basket needs in `.chart.result[0].meta`: regularMarketPrice (last +// price), chartPreviousClose (for the conventional day-change), plus +// day high/low/volume. Same single-symbol shape as Stooq, so the +// per-ticker fan-out in /intelligence/ticker_quotes is unchanged. +// +// fetchStooqQuote is kept above (dead upstream, not deleted) so the +// swap is reversible if a keyed provider replaces this later. +export async function fetchYahooQuote(ticker: string): Promise<{ + price?: number; + prev_close?: number; + price_date?: string; + open?: number; + high?: number; + low?: number; + volume?: number; +} | 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 m = r?.meta; + if (!m || typeof m.regularMarketPrice !== "number") return null; + // Last daily candle for the open (meta has no regularMarketOpen). + const q = r?.indicators?.quote?.[0] ?? {}; + const lastOpen = Array.isArray(q.open) ? q.open[q.open.length - 1] : undefined; + const t = typeof m.regularMarketTime === "number" ? m.regularMarketTime : null; + return { + price: m.regularMarketPrice, + prev_close: typeof m.chartPreviousClose === "number" ? m.chartPreviousClose : undefined, + price_date: t ? new Date(t * 1000).toISOString().slice(0, 10) : undefined, + open: typeof lastOpen === "number" ? lastOpen : undefined, + high: typeof m.regularMarketDayHigh === "number" ? m.regularMarketDayHigh : undefined, + low: typeof m.regularMarketDayLow === "number" ? m.regularMarketDayLow : undefined, + volume: typeof m.regularMarketVolume === "number" ? m.regularMarketVolume : undefined, + }; + } catch { + return null; + } +} + // Cheap name-to-ticker lookup for batch use (e.g. profiler index over // 200 contractors). Hits the in-memory SEC tickers index and the // KNOWN_PARENT_MAP. No SEC profile fetch, no Stooq fetch. Returns @@ -1198,14 +1248,17 @@ export async function fetchTickerBrief(name: string): Promise { } const [profile, quote] = await Promise.all([ fetchSecProfile(hit.cik_str), - fetchStooqQuote(hit.ticker), + fetchYahooQuote(hit.ticker), // Stooq CSV retired 2026-06-18; see fetchYahooQuote ]); const cap_proxy = quote?.price && quote?.volume ? quote.price * quote.volume : undefined; + // Day change vs previous close (conventional); fall back to vs-open. const day_change_pct = - quote?.price && quote?.open && quote.open > 0 - ? ((quote.price - quote.open) / quote.open) * 100 - : undefined; + quote?.price && quote?.prev_close && quote.prev_close > 0 + ? ((quote.price - quote.prev_close) / quote.prev_close) * 100 + : quote?.price && quote?.open && quote.open > 0 + ? ((quote.price - quote.open) / quote.open) * 100 + : undefined; const brief: TickerBrief = { source: "sec+stooq", fetched_at: now, diff --git a/mcp-server/index.ts b/mcp-server/index.ts index c1d74bb..3029b1a 100644 --- a/mcp-server/index.ts +++ b/mcp-server/index.ts @@ -936,45 +936,57 @@ async function main() { } } - // Batch ticker quotes — used by the profiler page's scrolling - // ticker basket. Stooq's HTTP CSV API is single-symbol per call, - // so this fans out N tickers in parallel and returns a flat - // map. Non-US tickers (HOC.DE, SKA-B.ST, LLC.AX) won't have a - // Stooq.us entry; we surface that as null so the UI can render - // them with a "—" placeholder. + // Batch ticker quotes — the profiler basket. Multi-source + // consensus (Yahoo + Nasdaq + CNBC), cache-first from the local + // SQLite time-series; cold tickers are aggregated live once and + // cached. Each quote carries price + day_change_pct (median across + // sources), a confidence score (corroborating sources / 3), the + // per-source breakdown, and a stale flag. Dotted/foreign symbols + // (HOC.DE, SKA-B.ST) are skipped → null → "—" placeholder. + // See quotes.ts; refreshed hourly by quote_ingest.ts. if (url.pathname === "/intelligence/ticker_quotes" && req.method === "POST") { const start = Date.now(); const b = await json(); const tickers: string[] = Array.isArray(b.tickers) ? b.tickers.map((t: any) => String(t).toUpperCase()).filter(Boolean) : []; if (!tickers.length) return ok({ quotes: {}, duration_ms: 0 }); - const { fetchStooqQuote } = await import("./entity.js"); - const dedup = Array.from(new Set(tickers)).slice(0, 50); - const results = await Promise.all(dedup.map(async (t) => { - // Skip non-US suffixes — Stooq.us won't have them - if (t.includes(".")) return [t, null] as const; - try { - const q = await fetchStooqQuote(t); - if (!q || !q.price) return [t, null] as const; - const change_pct = q.open && q.open > 0 ? ((q.price - q.open) / q.open) * 100 : null; - return [t, { - ticker: t, - price: q.price, - price_date: q.price_date, - open: q.open, - high: q.high, - low: q.low, - day_change_pct: change_pct, - stooq_url: `https://stooq.com/q/?s=${t.toLowerCase()}.us`, - }] as const; - } catch { - return [t, null] as const; - } - })); + const { serveQuotes } = await import("./quotes.ts"); + const dedup = Array.from(new Set(tickers)).slice(0, 50).filter((t) => !t.includes(".")); + const served = await serveQuotes(dedup); const quotes: Record = {}; - for (const [t, q] of results) quotes[t] = q; + for (const t of dedup) { + const a = served[t]; + quotes[t] = a + ? { + ticker: t, + price: a.price, + day_change_pct: a.day_change_pct, + price_date: a.price_date, + confidence: a.confidence, + source_count: a.source_count, + spread_pp: a.spread_pp, + sources: a.sources, + stale: a.stale, + as_of: a.ts, + source_url: `https://finance.yahoo.com/quote/${t}`, + } + : null; + } return ok({ quotes, count: dedup.length, duration_ms: Date.now() - start }); } + // Manual quote-ingest trigger (admin / timer). POST {tickers?:[]} + // — with a list, ingests those; without, refreshes every tracked + // ticker. The systemd timer hits this hourly; also callable by + // hand. Returns the write summary. + if (url.pathname === "/intelligence/quote_ingest" && req.method === "POST") { + const start = Date.now(); + const b = await json().catch(() => ({})); + const list = Array.isArray(b?.tickers) ? b.tickers.map((t: any) => String(t).toUpperCase()) : undefined; + const { ingestQuotes } = await import("./quotes.ts"); + const r = await ingestQuotes(list); + return ok({ ...r, duration_ms: Date.now() - start }); + } + // Profiler index — directory of every contractor that has filed // a Chicago permit recently, ranked by permit count + total // cost. Each name in the response links to the full /contractor diff --git a/mcp-server/profiler.html b/mcp-server/profiler.html index 96ffdc4..87bf985 100644 --- a/mcp-server/profiler.html +++ b/mcp-server/profiler.html @@ -87,6 +87,11 @@ td.role .pill{display:inline-block;padding:2px 7px;border-radius:9px;font-size:9 .bk-card .ch.up{color:#3fb950} .bk-card .ch.down{color:#f85149} .bk-card .ch.flat{color:#545d68} +.bk-card .cf{font-family:ui-monospace,monospace;font-size:9px;margin-top:3px;font-weight:700;letter-spacing:0.3px;display:inline-block;padding:1px 5px;border-radius:4px} +.bk-card .cf.full{background:#0d2818;border:1px solid #2ea04366;color:#3fb950} +.bk-card .cf.part{background:#1a1410;border:1px solid #d2992266;color:#d29922} +.bk-card .cf.thin{background:#2d1316;border:1px solid #f8514966;color:#f85149} +.bk-card .cf.stale{opacity:0.6;border-style:dashed} .bk-card .meta{font-size:9px;color:#545d68;margin-top:5px;text-transform:uppercase;letter-spacing:0.6px} .bk-card .kind-bar{position:absolute;left:0;top:0;bottom:0;width:3px;border-radius:8px 0 0 8px} .bk-card .kind-bar.exact,.bk-card .kind-bar.direct{background:#3fb950} @@ -108,8 +113,8 @@ td.role .pill{display:inline-block;padding:2px 7px;border-radius:9px;font-size:9

Staffing Co-Pilot · Profiler Index

@@ -303,6 +308,7 @@ function buildBasket(){ var tk=document.createElement('div'); tk.className='tk'; tk.textContent=b.ticker; card.appendChild(tk); var px=document.createElement('div'); px.className='px'; px.textContent='—'; card.appendChild(px); var ch=document.createElement('div'); ch.className='ch flat'; ch.textContent=' '; card.appendChild(ch); + var cf=document.createElement('div'); cf.className='cf'; cf.textContent=''; card.appendChild(cf); var meta=document.createElement('div'); meta.className='meta'; meta.textContent=b.count+' attribution'+(b.count===1?'':'s')+' · '+b.kinds.join('+'); card.appendChild(meta); @@ -328,7 +334,7 @@ function buildBasket(){ }).then(function(r){return r.json()}).then(function(qd){ var quotes=qd.quotes||{}; lastQuotes = quotes; - document.getElementById('bk-meta').textContent='quotes via Stooq · '+(qd.duration_ms||0)+'ms'; + document.getElementById('bk-meta').textContent='quotes via 3-source consensus (Yahoo·Nasdaq·CNBC) · '+(qd.duration_ms||0)+'ms'; Array.prototype.forEach.call(track.children, function(card){ var t=card.dataset.ticker; var q=quotes[t]; if(!q || !q.price) return; @@ -338,6 +344,21 @@ function buildBasket(){ if(q.day_change_pct==null){ ch.textContent='close '+(q.price_date||''); ch.className='ch flat'; } else if(q.day_change_pct>=0){ ch.textContent='+'+q.day_change_pct.toFixed(2)+'%'; ch.className='ch up'; } else { ch.textContent=q.day_change_pct.toFixed(2)+'%'; ch.className='ch down'; } + // Multi-source confidence badge: corroborating sources / 3. Green + // at full agreement, amber when one source is missing/divergent, + // red when thin. The whole differentiator made visible per-ticker. + var cf=card.querySelector('.cf'); + if(cf && q.confidence!=null){ + var cls = q.confidence>=100 ? 'cf full' : q.confidence>=67 ? 'cf part' : 'cf thin'; + if(q.stale) cls += ' stale'; + cf.className = cls; + cf.textContent = (q.source_count||0)+'/3 ✓ '+q.confidence+'%'+(q.stale?' · stale':''); + // Tooltip: per-source breakdown so a staffer can audit the number. + var lines=(q.sources||[]).map(function(s){ + return ' '+s.source+': '+(s.day_change_pct>=0?'+':'')+s.day_change_pct.toFixed(2)+'%'+(s.outlier?' ⚠ outlier':''); + }).join('\n'); + card.title=(q.ticker||t)+' — '+q.confidence+'% confidence ('+(q.source_count||0)+'/3 sources, spread '+(q.spread_pp!=null?q.spread_pp.toFixed(2):'?')+'pp)\n'+lines; + } }); updateThesisMetrics(); }).catch(function(){ diff --git a/mcp-server/quote_ingest.ts b/mcp-server/quote_ingest.ts new file mode 100644 index 0000000..2f9721e --- /dev/null +++ b/mcp-server/quote_ingest.ts @@ -0,0 +1,31 @@ +// Hourly quote ingest — CLI entry for the systemd timer (and manual runs). +// +// bun run quote_ingest.ts AAPL MSFT CAT # ingest a specific list +// bun run quote_ingest.ts # refresh every tracked ticker +// +// Writes multi-source consensus snapshots into the local SQLite cache +// (data/_quote_cache/quotes.db). See quotes.ts for the aggregation logic. +import { ingestQuotes, trackedTickers } from "./quotes.ts"; + +const args = process.argv.slice(2).map((s) => s.toUpperCase()); +const t0 = Date.now(); +const tracked = trackedTickers(); +const target = args.length ? args : tracked; + +if (!target.length) { + console.log( + "[quote_ingest] nothing to ingest — cache is empty and no tickers passed.\n" + + " Seed it by passing symbols (e.g. `bun run quote_ingest.ts AAPL CAT HD`),\n" + + " or just hit the profiler once so the serving path warms the cache.", + ); + process.exit(0); +} + +console.log( + `[quote_ingest] ${args.length ? "explicit list" : "refreshing tracked universe"}: ${target.length} tickers`, +); +const r = await ingestQuotes(target); +console.log( + `[quote_ingest] done in ${Date.now() - t0}ms — ${r.written}/${r.requested} written` + + (r.failed.length ? ` · no source answered for: ${r.failed.join(", ")}` : ""), +); diff --git a/mcp-server/quotes.ts b/mcp-server/quotes.ts new file mode 100644 index 0000000..45a14f1 --- /dev/null +++ b/mcp-server/quotes.ts @@ -0,0 +1,347 @@ +// 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 { + 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 { + 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 { + 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 { + 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> { + const out: Record = {}; + 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 }; +} diff --git a/ops/systemd/install.sh b/ops/systemd/install.sh index 53a02c6..0e96567 100755 --- a/ops/systemd/install.sh +++ b/ops/systemd/install.sh @@ -21,6 +21,8 @@ UNITS=( lakehouse-context7-bridge.service lakehouse-retention-sweep.service lakehouse-retention-sweep.timer + lakehouse-quote-ingest.service + lakehouse-quote-ingest.timer ) if [[ $EUID -ne 0 ]]; then diff --git a/ops/systemd/lakehouse-quote-ingest.service b/ops/systemd/lakehouse-quote-ingest.service new file mode 100644 index 0000000..de990d8 --- /dev/null +++ b/ops/systemd/lakehouse-quote-ingest.service @@ -0,0 +1,17 @@ +[Unit] +Description=Lakehouse quote ingest — multi-source consensus snapshot into the local SQLite cache (Yahoo+Nasdaq+CNBC) +After=network-online.target lakehouse-agent.service +Wants=network-online.target + +[Service] +# Oneshot driven by lakehouse-quote-ingest.timer. Runs as root to match +# the lakehouse-agent (mcp-server) service that owns the cache db — both +# write data/_quote_cache/quotes.db, so they must share a user (see +# README "Why both services run as root"). busy_timeout in quotes.ts +# serializes the concurrent writes. +Type=oneshot +WorkingDirectory=/home/profit/lakehouse/mcp-server +ExecStart=/home/profit/.bun/bin/bun run /home/profit/lakehouse/mcp-server/quote_ingest.ts +# No args → refreshes every ticker already tracked in the cache. Cold +# tickers (new basket entries) warm on-demand via the serving path. +TimeoutStartSec=300 diff --git a/ops/systemd/lakehouse-quote-ingest.timer b/ops/systemd/lakehouse-quote-ingest.timer new file mode 100644 index 0000000..5525e19 --- /dev/null +++ b/ops/systemd/lakehouse-quote-ingest.timer @@ -0,0 +1,11 @@ +[Unit] +Description=Hourly trigger for Lakehouse multi-source quote ingest + +[Timer] +OnCalendar=hourly +Persistent=true +# Spread the wake so we don't hit the free quote endpoints on the dot. +RandomizedDelaySec=120 + +[Install] +WantedBy=timers.target