spb/market-atlas

1import type { Observation } from "@market-atlas/market-model";2import { pool } from "../db/pool.js";3import { logger } from "../logger.js";4import { telemetry } from "./telemetry.js";56interface Sample {7  v: number;8  t: number; // source ts or receive ts9}1011/**12 * Statistical shared-upstream detection. For every instrument, recent LAST_PRICE observations of each13 * source are kept; two sources whose values coincide (identical price within a rounding epsilon and14 * within 2 s) on the vast majority of aligned samples very likely share an upstream vendor. The result is15 * a `similarity` in [0, 1] per source pair; pairs above the threshold are treated as one family by the16 * consensus engine (independence weight 0.25) even without contractual knowledge of the vendor.17 */18export class LineageEngine {19  private samples = new Map<string, Map<string, Sample[]>>(); // instrumentId -> sourceId -> ring20  private similarity = new Map<string, { similarity: number; samples: number; instruments: number }>(); // "a|b" sorted21  private inferredFamily = new Map<string, string>(); // sourceId -> inferred family id22  readonly maxPerSource = 400;23  readonly threshold = 0.92;24  readonly minSamples = 120;2526  observe(o: Observation): void {27    if (o.field !== "LAST_PRICE") return;28    let m = this.samples.get(o.instrumentId);29    if (!m) {30      m = new Map();31      this.samples.set(o.instrumentId, m);32    }33    let ring = m.get(o.sourceId);34    if (!ring) {35      ring = [];36      m.set(o.sourceId, ring);37    }38    ring.push({ v: o.value, t: o.sourceTimestamp ?? o.receivedAt });39    if (ring.length > this.maxPerSource) ring.splice(0, ring.length - this.maxPerSource);40  }4142  /** Family override for a source when statistical lineage says it mirrors another source. */43  familyOf(sourceId: string): string | null {44    return this.inferredFamily.get(sourceId) ?? null;45  }4647  pairs(): Array<{ sourceA: string; sourceB: string; similarity: number; samples: number; instruments: number; likelySharedUpstream: boolean }> {48    return [...this.similarity.entries()]49      .map(([k, v]) => {50        const [sourceA, sourceB] = k.split("|") as [string, string];51        return { sourceA, sourceB, ...v, likelySharedUpstream: v.similarity >= this.threshold && v.samples >= this.minSamples };52      })53      .sort((a, b) => b.similarity - a.similarity);54  }5556  /** Recompute similarities (every 5 min) and persist them. */57  async recompute(): Promise<void> {58    const acc = new Map<string, { matched: number; total: number; instruments: Set<string> }>();59    for (const [instrumentId, bySource] of this.samples) {60      const sources = [...bySource.keys()];61      for (let i = 0; i < sources.length; i++) {62        for (let j = i + 1; j < sources.length; j++) {63          const a = bySource.get(sources[i]!)!;64          const b = bySource.get(sources[j]!)!;65          if (a.length < 30 || b.length < 30) continue;66          const { matched, total } = align(a, b);67          if (!total) continue;68          const key = [sources[i]!, sources[j]!].sort().join("|");69          const e = acc.get(key) ?? { matched: 0, total: 0, instruments: new Set<string>() };70          e.matched += matched;71          e.total += total;72          e.instruments.add(instrumentId);73          acc.set(key, e);74        }75      }76    }77    this.similarity.clear();78    this.inferredFamily.clear();79    const rows: unknown[][] = [];80    for (const [key, e] of acc) {81      const sim = e.matched / e.total;82      this.similarity.set(key, { similarity: Math.round(sim * 1000) / 1000, samples: e.total, instruments: e.instruments.size });83      if (sim >= this.threshold && e.total >= this.minSamples) {84        const [a, b] = key.split("|") as [string, string];85        const fam = this.inferredFamily.get(a) ?? this.inferredFamily.get(b) ?? `inferred:${a}+${b}`;86        this.inferredFamily.set(a, fam);87        this.inferredFamily.set(b, fam);88        telemetry.inc("lineage_shared_upstream_pairs_total");89      }90      const [a, b] = key.split("|");91      rows.push([a, b, Math.round(sim * 1000) / 1000, e.total, e.instruments.size]);92    }93    if (!rows.length) return;94    try {95      const tuples = rows.map((_, i) => `($${i * 5 + 1},$${i * 5 + 2},$${i * 5 + 3},$${i * 5 + 4},$${i * 5 + 5})`);96      await pool.query(97        `insert into source_lineage (source_a, source_b, similarity, samples, instruments) values ${tuples.join(",")}98         on conflict (source_a, source_b) do update set similarity = excluded.similarity, samples = excluded.samples, instruments = excluded.instruments, computed_at = now()`,99        rows.flat(),100      );101    } catch (err) {102      logger.warn({ err: err instanceof Error ? err.message : String(err) }, "lineage persist failed");103    }104  }105}106107/** Count how many samples of `a` have a sample of `b` within 2 s with an identical value (relative 1e-6). */108function align(a: Sample[], b: Sample[]): { matched: number; total: number } {109  let j = 0;110  let matched = 0;111  let total = 0;112  const sb = [...b].sort((x, y) => x.t - y.t);113  for (const s of [...a].sort((x, y) => x.t - y.t)) {114    while (j < sb.length && sb[j]!.t < s.t - 2000) j++;115    let k = j;116    let hit = false;117    let found = false;118    while (k < sb.length && sb[k]!.t <= s.t + 2000) {119      found = true;120      if (Math.abs(sb[k]!.v - s.v) <= Math.abs(s.v) * 1e-6) {121        hit = true;122        break;123      }124      k++;125    }126    if (found) {127      total++;128      if (hit) matched++;129    }130  }131  return { matched, total };132}133134export const lineage = new LineageEngine();135