spb/market-atlas

1/**2 * Market Atlas CLI:3 *   ma migrate                      apply SQL migrations4 *   ma seed                         reference data (countries, exchanges, holidays)5 *   ma connectors                   list connectors with rights/realtime status6 *   ma run <connector> [--file p]   run one poll (or normalize a fixture file offline) and print the batch7 *   ma replay <day> [connector]     re-run normalization on archived raw payloads of a day (no network)8 *   ma status                       quick DB/quotes/events status9 *   ma archive                      archive observation partitions older than the retention window10 *   ma backup                       pg_dump metadata to data/backups11 */12import { readFileSync, readdirSync, existsSync } from "node:fs";13import { join } from "node:path";14import { gunzipSync } from "node:zlib";15import { CONNECTORS } from "@market-atlas/connectors";16import { makeTestContext, normalizeFixture } from "@market-atlas/connector-sdk";17import { config } from "./config.js";18import { migrate } from "./db/migrate.js";19import { pool } from "./db/pool.js";20import { seed } from "./seed/index.js";21import { archiveOldPartitions, backupMetadata } from "./core/jobs.js";2223const [cmd, ...args] = process.argv.slice(2);2425async function main() {26  switch (cmd) {27    case "migrate": {28      const applied = await migrate();29      console.log(applied.length ? `applied: ${applied.join(", ")}` : "up to date");30      break;31    }32    case "seed":33      await migrate();34      await seed();35      console.log("seed ok");36      break;37    case "connectors":38      for (const c of CONNECTORS) {39        const m = c.metadata;40        console.log(`${m.id.padEnd(30)} ${m.sourceType.padEnd(12)} ${m.rightsStatus.padEnd(34)} ${m.realtimeStatus.padEnd(12)} ${m.assetClasses.join(",")}`);41      }42      break;43    case "run": {44      const id = args[0];45      const def = CONNECTORS.find((c) => c.metadata.id === id);46      if (!def) throw new Error(`unknown connector ${id}`);47      const fileIdx = args.indexOf("--file");48      if (fileIdx >= 0) {49        const path = args[fileIdx + 1]!;50        const text = readFileSync(path, "utf8");51        const payload = path.endsWith(".json") ? JSON.parse(text) : text;52        const kind = args[args.indexOf("--kind") + 1] ?? "poll";53        const batch = normalizeFixture(def, args.includes("--kind") ? kind : "poll", payload);54        console.log(JSON.stringify(batch, null, 2));55        break;56      }57      if (!def.poll) throw new Error("streaming connector: use the running service or --file");58      const ctx = makeTestContext(def, { fetchImpl: fetch, secrets: process.env as Record<string, string> });59      const raws = await def.poll(ctx);60      let total = 0;61      for (const r of raws) {62        const b = def.normalize(r);63        total += b.observations.length;64        console.log(`kind=${r.kind} observations=${b.observations.length} events=${b.events?.length ?? 0} instruments=${b.instruments?.length ?? 0} filings=${b.filings?.length ?? 0}`);65        for (const o of b.observations.slice(0, 8)) console.log(`  ${o.symbol.padEnd(12)} ${o.field.padEnd(15)} ${String(o.value).padEnd(14)} ${o.sourceTimestamp ? new Date(o.sourceTimestamp).toISOString() : "-"} ${o.realtimeStatus}`);66      }67      console.log(`total observations: ${total}`);68      break;69    }70    case "replay": {71      const day = args[0];72      if (!day) throw new Error("usage: ma replay YYYY-MM-DD [connector]");73      const dir = join(config.dataDir, "raw", day);74      if (!existsSync(dir)) throw new Error(`no raw archive for ${day}`);75      const connectors = args[1] ? [args[1]] : readdirSync(dir);76      for (const cid of connectors) {77        const def = CONNECTORS.find((c) => c.metadata.id === cid);78        if (!def) continue;79        let n = 0,80          obs = 0,81          failed = 0;82        for (const f of readdirSync(join(dir, cid))) {83          const rec = JSON.parse(gunzipSync(readFileSync(join(dir, cid, f))).toString("utf8"));84          try {85            const b = def.normalize({ connectorId: cid, sourceId: def.metadata.sourceId, kind: rec.kind, payload: rec.payload, receivedAt: Date.parse(rec.received_at) });86            obs += b.observations.length;87          } catch {88            failed++;89          }90          n++;91        }92        console.log(`${cid}: ${n} payloads → ${obs} observations, ${failed} normalization failures`);93      }94      break;95    }96    case "status": {97      const [i, q, e, o, c] = await Promise.all([98        pool.query<{ n: number }>("select count(*)::int as n from instruments"),99        pool.query<{ n: number }>("select count(*)::int as n from canonical_quotes where price is not null"),100        pool.query<{ n: number }>("select count(*)::int as n from market_events where ts > now() - interval '24 hours'"),101        pool.query<{ n: number }>("select coalesce(sum(observations),0)::bigint as n from daily_stats"),102        pool.query("select id, status, messages_total, reliability_score, last_message_at from connectors order by id"),103      ]);104      console.log(`instruments: ${i.rows[0]?.n}  quotes: ${q.rows[0]?.n}  events 24h: ${e.rows[0]?.n}  observations total: ${o.rows[0]?.n}`);105      for (const r of c.rows) console.log(`  ${String(r.id).padEnd(30)} ${String(r.status).padEnd(12)} msgs=${r.messages_total} score=${r.reliability_score ?? "-"} last=${r.last_message_at ?? "-"}`);106      break;107    }108    case "archive":109      console.log(await archiveOldPartitions());110      break;111    case "backup":112      console.log(await backupMetadata());113      break;114    default:115      console.log("usage: ma migrate|seed|connectors|run <id> [--file f --kind k]|replay <day> [connector]|status|archive|backup");116  }117  await pool.end();118}119120main().catch((err) => {121  console.error(err instanceof Error ? err.message : err);122  process.exit(1);123});124