profiler: multi-source quote aggregation with confidence scoring

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) <noreply@anthropic.com>
This commit is contained in:
root 2026-06-18 01:28:28 -05:00
parent fcd53168a0
commit ab1f12296d
8 changed files with 531 additions and 37 deletions

View File

@ -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<TickerBrief> {
}
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,

View File

@ -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<string, any> = {};
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

View File

@ -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
<div class="bar">
<h1>Staffing Co-Pilot · Profiler Index</h1>
<nav>
<a href="" id="back-dashboard">← Dashboard</a>
<a href="" id="back-console">Console</a>
<a href="." id="back-dashboard">← Dashboard</a>
<a href="console" id="back-console">Console</a>
</nav>
</div>
<div class="content">
@ -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(){

View File

@ -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(", ")}` : ""),
);

347
mcp-server/quotes.ts Normal file
View File

@ -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<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 };
}

View File

@ -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

View File

@ -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

View File

@ -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