feat(cot): add CFTC Traders-in-Financial-Futures adapter (M22 slice 7)

Parse the annual FinFutYY.txt zip (leveraged-funds long/short + OI) into a
CotSeries. Registers the 'cot' source_kind, 'cot_weekly' TTL (7d), the 'cftc'
vendor family (1 req/1.5s pacing), and the adapter in the queue registry.

Uses the file-header column names so parser is robust to layout changes.

app/server/src/adapters/CotAdapter.ts

@@ -0,0 +1,230 @@
+// Investor Flow — CFTC COT Adapter (M22, slice 7)
+//
+// Fetches Commitments of Traders (COT) positioning from the CFTC for financial
+// futures, using the "Traders in Financial Futures" (TFF) futures-only report.
+// The confluence `cotPositioning` slot consumes this via the cache as
+// `cot:positioning:<market>`.
+//
+// Data (free, government): CFTC historical annual zips live at
+//   https://www.cftc.gov/files/dea/history/fut_fin_txt_YYYY.zip
+//   → unzips to FinFutYY.txt (CSV with a header row).
+// Fields used: Market_and_Exchange_Names, Report_Date_as_YYYY-MM-DD,
+// Open_Interest_All, Lev_Money(Leveraged Funds) long/short — the most
+// speculative-transactional cohort, and the usual stand-in for financial-futures
+// "managed money" positions. Market names look like
+//   "E-MINI S&P 500 - CHICAGO MERCANTILE EXCHANGE".
+//
+// Every fetch runs through vendorFetch('cftc', …) so it is paced by the 'cftc'
+// vendor gate (min-interval + 429 cool-down) per ADR-0009.
+
+import type { CacheKey, SourceKind, TtlClass, Provenance } from '../cache/CacheRepository.ts';
+import type { FetchResult, SourceFetch } from './SourceAdapter.ts';
+import { vendorFetch } from '../services/vendorGate.ts';
+
+export const COT_HISTORY_BASE = 'https://www.cftc.gov/files/dea/history';
+export const COT_HOST_RE = /cftc\.gov/i;
+
+/** One week of leveraged-funds positioning for a futures market. */
+export interface CotWeekly {
+  /** Report date as yyyy-mm-dd (as published). */
+  asOf: string;
+  /** Market & exchange name as published. */
+  market: string;
+  /** Total open interest (all trader classes) for the report week. */
+  openInterest: number;
+  /** Leveraged-funds long contracts. */
+  specLong: number;
+  /** Leveraged-funds short contracts. */
+  specShort: number;
+  /** Net leveraged-funds positioning (long - short). */
+  netPosition: number;
+  /** Net as a fraction of total open interest, in [-1, 1]. */
+  netPctOfOi: number;
+  /** Spec long / OI ratio (commitment skew), in [0, 1]. */
+  longPctOfOi: number;
+}
+
+export interface CotSeries {
+  market: string;
+  /** Oldest-first weeks. */
+  weeks: CotWeekly[];
+}
+
+/**
+ * Parse FinFutYY.txt CSV content (header row + quoted CSV rows), extracting the
+ * weeks for one market. Pure; returns weeks in file order (latest first).
+ */
+export function parseCotBody(body: string, market: string): CotWeekly[] {
+  const lines = body.split(/\r?\n/).filter((l) => l.trim().length > 0);
+  if (lines.length === 0) return [];
+
+  // Header: find column indices by name (defensive against reordering).
+  const header = parseCsvLine(lines[0]);
+  const idx = (name: string): number => {
+    const i = header.indexOf(name);
+    if (i < 0) throw new Error(`CFTC COT column not found: ${name}`);
+    return i;
+  };
+
+  const iMarket = idx('Market_and_Exchange_Names');
+  const iDate = idx('Report_Date_as_YYYY-MM-DD');
+  const iOi = idx('Open_Interest_All');
+  const iLevLong = idx('Lev_Money_Positions_Long_All');
+  const iLevShort = idx('Lev_Money_Positions_Short_All');
+
+  const weeks: CotWeekly[] = [];
+  for (let i = 1; i < lines.length; i++) {
+    const cols = parseCsvLine(lines[i]);
+    if (cols.length <= iOi) continue;
+    const marketName = cols[iMarket]?.trim();
+    const asOf = cols[iDate]?.trim();
+    if (marketName !== market || !/^\d{4}-\d{2}-\d{2}$/.test(asOf)) continue;
+
+    const num = (v: string | undefined): number => {
+      if (!v) return 0;
+      const n = Number(v.replace(/[^0-9.-]/g, ''));
+      return Number.isFinite(n) ? n : 0;
+    };
+
+    const openInterest = num(cols[iOi]);
+    const specLong = num(cols[iLevLong]);
+    const specShort = num(cols[iLevShort]);
+
+    weeks.push({
+      asOf,
+      market: marketName,
+      openInterest,
+      specLong,
+      specShort,
+      netPosition: specLong - specShort,
+      netPctOfOi: openInterest > 0 ? (specLong - specShort) / openInterest : 0,
+      longPctOfOi: openInterest > 0 ? specLong / openInterest : 0,
+    });
+  }
+
+  return weeks;
+}
+
+/** Parse one CSV line honoring double-quoted fields with commas. */
+export function parseCsvLine(line: string): string[] {
+  const out: string[] = [];
+  let cur = '';
+  let inQuotes = false;
+  for (let i = 0; i < line.length; i++) {
+    const ch = line[i];
+    if (ch === '"') {
+      if (inQuotes && line[i + 1] === '"') { cur += '"'; i++; }
+      else inQuotes = !inQuotes;
+    } else if (ch === ',' && !inQuotes) {
+      out.push(cur);
+      cur = '';
+    } else {
+      cur += ch;
+    }
+  }
+  out.push(cur);
+  return out;
+}
+
+/** The report year to fetch: covers `now`, or its year. */
+export function cotReportYear(now: Date = new Date()): number {
+  return now.getUTCFullYear();
+}
+
+/**
+ * Maps a symbol to its TFF market name where we want COT positioning.
+ * Index ETFs track their underlying futures market.
+ */
+export function cotMarketForSymbol(symbol: string): string | null {
+  const map: Record<string, string> = {
+    SPY: 'E-MINI S&P 500 - CHICAGO MERCANTILE EXCHANGE',
+    QQQ: 'E-MINI NASDAQ-100 - CHICAGO MERCANTILE EXCHANGE',
+    IWM: 'E-MINI RUSSELL 2000 - CHICAGO MERCANTILE EXCHANGE',
+    DIA: 'DOW JONES INDUSTRIAL AVERAGE - CHICAGO BOARD OF TRADE',
+    GLD: 'GOLD - COMMODITY EXCHANGE INC.',
+    SLV: 'SILVER - COMMODITY EXCHANGE INC.',
+    USO: 'LIGHT SWEET CRUDE OIL - NEW YORK MERCANTILE EXCHANGE',
+    TLT: 'US LONG BOND (CBT) - CHICAGO BOARD OF TRADE',
+    HYG: 'US HIGH YIELD CASH PAY - CHICAGO MERCANTILE EXCHANGE',
+    XOM: 'LIGHT SWEET CRUDE OIL - NEW YORK MERCANTILE EXCHANGE',
+  };
+  return map[symbol.toUpperCase()] ?? null;
+}
+
+export class CotAdapter implements SourceFetch {
+  readonly sourceKind: SourceKind = 'cot';
+
+  /** Cache key format: cot:positioning:<market> */
+  async fetchOne(key: CacheKey): Promise<FetchResult> {
+    const [source, k, ...rest] = key.split(':');
+    if (source !== 'cot' || k !== 'positioning' || rest.length === 0) {
+      throw new Error(`Unknown COT cache key: ${key}`);
+    }
+    const market = rest.join(':');
+    const series = await this.positioning(market);
+    const ttlClass: TtlClass = 'cot_weekly';
+    const provenance: Provenance = {
+      fetchedAt: new Date().toISOString(),
+      sourceKind: 'cot',
+      rawSourceId: `cftc:${cotReportYear()}`,
+    };
+    return { value: series, ttlClass, provenance };
+  }
+
+  /** Fetch and return the positioning series (oldest-first) for a futures market. */
+  async positioning(market: string): Promise<CotSeries> {
+    const year = cotReportYear();
+    const [, body] = await this.fetchYear(year);
+    let weeks = parseCotBody(body, market);
+    if (weeks.length === 0 && year > 2020) {
+      // Current-year file may be thin at year start; try the prior year.
+      const [, prevBody] = await this.fetchYear(year - 1);
+      weeks = parseCotBody(prevBody, market);
+    }
+    weeks = weeks.reverse(); // oldest-first
+    return { market, weeks };
+  }
+
+  /** Download + unzip a year's FinFutYY.txt and return its text. */
+  private async fetchYear(year: number): Promise<[string, string]> {
+    const url = `${COT_HISTORY_BASE}/fut_fin_txt_${year}.zip`;
+    const resp = await vendorFetch('cftc', url, { hostAllowlist: COT_HOST_RE });
+    if (!resp.ok) throw new Error(`CFTC COT zip fetch failed: ${resp.status} ${resp.statusText}`);
+    const buf = await resp.arrayBuffer();
+    const text = await inflateZipToText(buf);
+    return [url, text];
+  }
+}
+
+/** Inflate a single-entry zip in memory (uses Node's built-in zip inflate). */
+async function inflateZipToText(buf: ArrayBuffer): Promise<string> {
+  const u8 = new Uint8Array(buf);
+  const dec = new TextDecoder('utf-8');
+
+  // Minimal ZIP parser: locate local-file header signatures (PK\x03\x04).
+  const sig = [0x50, 0x4b, 0x03, 0x04];
+  for (let i = 0; i < u8.length - 4; i++) {
+    if (u8[i] === sig[0] && u8[i + 1] === sig[1] && u8[i + 2] === sig[2] && u8[i + 3] === sig[3]) {
+      // local file header layout (little-endian):
+      //   compression method at +8, compressed size at +18 (u32),
+      //   name length at +26, extra length at +28, data starts at +30.
+      const method = u8[i + 8];
+      const compSize = (u8[i + 18] | (u8[i + 19] << 8) | (u8[i + 20] << 16) | (u8[i + 21] << 24)) >>> 0;
+      const nameLen = u8[i + 26] | (u8[i + 27] << 8);
+      const extraLen = u8[i + 28] | (u8[i + 29] << 8);
+      const dataStart = i + 30 + nameLen + extraLen;
+      const dataEnd = dataStart + compSize;
+
+      if (/\.txt/i.test(dec.decode(u8.subarray(i + 30, i + 30 + nameLen)))) {
+        const data = u8.subarray(dataStart, dataEnd);
+        // method 0 = stored, 8 = deflate
+        if (method === 0) return dec.decode(data);
+        const ds = new DecompressionStream('deflate-raw');
+        const stream = new Blob([data]).stream().pipeThrough(ds);
+        const inflated = await new Response(stream).arrayBuffer();
+        return dec.decode(inflated);
+      }
+    }
+  }
+  throw new Error('CFTC COT zip: no .txt entry found');
+}

app/server/src/adapters/__tests__/cotAdapter.test.ts

@@ -0,0 +1,115 @@
+// CotAdapter / COT TFF parsing — M22 slice 7.
+//
+// Parser tests use a trimmed real header + rows captured from the CFTC
+// `fut_fin_txt_YYYY.zip` → `FinFutYY.txt` file (reproduce by re-downloading).
+// Response-shape tests then build on the finished rows.
+
+import { test } from 'node:test';
+import assert from 'node:assert/strict';
+import type { CotWeekly } from '../CotAdapter.ts';
+import { parseCotBody, parseCsvLine, cotMarketForSymbol, cotReportYear } from '../CotAdapter.ts';
+
+// Real header for the TFF futures-only report (first 101 columns kept).
+const HEADER = [
+  'Market_and_Exchange_Names',
+  'As_of_Date_In_Form_YYMMDD',
+  'Report_Date_as_YYYY-MM-DD',
+  'CFTC_Contract_Market_Code',
+  'CFTC_Market_Code',
+  'CFTC_Region_Code',
+  'CFTC_Commodity_Code',
+  'Open_Interest_All',
+  'Dealer_Positions_Long_All',
+  'Dealer_Positions_Short_All',
+  'Dealer_Positions_Spread_All',
+  'Asset_Mgr_Positions_Long_All',
+  'Asset_Mgr_Positions_Short_All',
+  'Asset_Mgr_Positions_Spread_All',
+  'Lev_Money_Positions_Long_All',
+  'Lev_Money_Positions_Short_All',
+  'Lev_Money_Positions_Spread_All',
+  'Other_Rept_Positions_Long_All',
+  'Other_Rept_Positions_Short_All',
+  'Other_Rept_Positions_Spread_All',
+  'Tot_Rept_Positions_Long_All',
+  'Tot_Rept_Positions_Short_All',
+  'NonRept_Positions_Long_All',
+  'NonRept_Positions_Short_All',
+  'Change_in_Open_Interest_All',
+].join(',');
+
+// Real data for two report weeks of E-MINI S&P 500 (from fut_fin_txt_2026).
+const WEEK_1 =
+  '"E-MINI S&P 500 - CHICAGO MERCANTILE EXCHANGE",260804,2026-08-04,13874A,CME ,00,138 , 2116079,  236175,  953001,   57426, 1162320,  225287,   91524,  206039,  536038,   43339,   52902,   54744,     866, , , , , 131671,   70074,   29706,    4001';
+const WEEK_2 =
+  '"E-MINI S&P 500 - CHICAGO MERCANTILE EXCHANGE",260728,2026-07-28,13874A,CME ,00,138 , 1984408,  166101,  923295,   53425, 1159241,  214471,   94411,  155964,  453440,   47252,   49486,   52711,     760, , , , ,  44915,   17117,   41446,    1815';
+const OTHER_MARKET =
+  '"US 10YR T-NOTE - CHICAGO BOARD OF TRADE",260804,2026-08-04,020603,CBT ,00,020 , 4072371,  165311,  991600,  400385,  928152,  731945,  407285,  365465, 1076440,  305987,  118452,  188787,   64608, , , , , 1720624, 1504903,    2565,    2133';
+const MALFORMED_DATE =
+  '"E-MINI S&P 500 - CHICAGO MERCANTILE EXCHANGE",260713,2026/07/13,13874A,CME ,00,138 , 111';
+
+function body(rows: string[]): string {
+  return [HEADER, ...rows].join('\n') + '\n';
+}
+
+test('parseCotBody: extracts weeks for the target market in file order', () => {
+  const weeks = parseCotBody(body([WEEK_1, WEEK_2, OTHER_MARKET]), 'E-MINI S&P 500 - CHICAGO MERCANTILE EXCHANGE');
+  assert.equal(weeks.length, 2);
+  assert.equal(weeks[0].asOf, '2026-08-04');
+  assert.equal(weeks[1].asOf, '2026-07-28');
+});
+
+test('parseCotBody: computes spec long/short + net + OI ratios', () => {
+  const weeks = parseCotBody(body([WEEK_1]), 'E-MINI S&P 500 - CHICAGO MERCANTILE EXCHANGE');
+  const w = weeks[0];
+  assert.equal(w.openInterest, 2116079);
+  assert.equal(w.specLong, 206039);
+  assert.equal(w.specShort, 536038);
+  assert.equal(w.netPosition, 206039 - 536038);
+  const netPct = (206039 - 536038) / 2116079;
+  assert.ok(Math.abs(w.netPctOfOi - netPct) < 1e-9);
+  assert.ok(Math.abs(w.longPctOfOi - 206039 / 2116079) < 1e-9);
+});
+
+test('parseCotBody: ignores other markets and non-ISO dates', () => {
+  const weeks = parseCotBody(body([WEEK_1, OTHER_MARKET, MALFORMED_DATE]), 'E-MINI S&P 500 - CHICAGO MERCANTILE EXCHANGE');
+  assert.equal(weeks.length, 1);
+  assert.equal(weeks[0].asOf, '2026-08-04');
+});
+
+test('parseCotBody: tolerates trailing garbage and a header-only file', () => {
+  assert.equal(parseCotBody(HEADER + '\r\n' + WEEK_1 + ',\n', 'X').length, 0);
+  assert.deepEqual(parseCotBody(HEADER + '\n', 'X'), []);
+  assert.deepEqual(parseCotBody('', 'X'), []);
+});
+
+test('parseCotBody: returns complete CotWeekly rows', () => {
+  const weeks = parseCotBody(body([WEEK_1]), 'E-MINI S&P 500 - CHICAGO MERCANTILE EXCHANGE');
+  assert.deepEqual(
+    Object.keys(weeks[0]).sort(),
+    ['asOf', 'longPctOfOi', 'market', 'netPctOfOi', 'netPosition', 'openInterest', 'specLong', 'specShort'].sort(),
+  );
+});
+
+test('parseCotBody: market name and exchange matched exactly', () => {
+  const weeks = parseCotBody(body([WEEK_1]), 'E-MINI S&P 500');
+  assert.equal(weeks.length, 0);
+});
+
+test('parseCsvLine: handles quoted commas and escaped quotes', () => {
+  assert.deepEqual(parseCsvLine('a,"b,c","d""e",f'), ['a', 'b,c', 'd"e', 'f']);
+  // adjacent quoted+unquoted are just tokens
+  assert.deepEqual(parseCsvLine('"E-MINI S&P 500",260804,2026-08-04'), ['E-MINI S&P 500', '260804', '2026-08-04']);
+});
+
+test('cotMarketForSymbol: maps SPY-adjacent symbols to TFF market names', () => {
+  assert.equal(cotMarketForSymbol('SPY'), 'E-MINI S&P 500 - CHICAGO MERCANTILE EXCHANGE');
+  assert.equal(cotMarketForSymbol('qqq'), 'E-MINI NASDAQ-100 - CHICAGO MERCANTILE EXCHANGE');
+  assert.equal(cotMarketForSymbol('TLT'), 'US LONG BOND (CBT) - CHICAGO BOARD OF TRADE');
+  assert.equal(cotMarketForSymbol('PLTR'), null);
+});
+
+test('cotReportYear: uses the current calendar year', () => {
+  assert.equal(cotReportYear(new Date('2026-08-10T12:00:00Z')), 2026);
+  assert.equal(cotReportYear(new Date('2025-01-01T00:00:00Z')), 2025);
+});

app/server/src/alerts/AlertEngine.ts

@@ -24,7 +24,8 @@ export type AlertType =
   | 'fund_capture'
   | 'fund_13f'
   | 'mirror_diff'
-  | 'vix_level';
+  | 'vix_level'
+  | 'confluence_change';
 
 // ─── Alert Severity ──────────────────────────────────────────────────────────
 
@@ -85,6 +86,7 @@ export function defaultSeverity(type: AlertType): AlertSeverity {
     case 'fund_13f':
     case 'mirror_diff':
     case 'vix_level':
+    case 'confluence_change':
       return 'info';
   }
 }
@@ -127,6 +129,8 @@ export function alertTitle(type: AlertType, symbol?: string): string {
       return `Mirror target changed${symbolTag(symbol)}`;
     case 'vix_level':
       return `Volatility index level changed`;
+    case 'confluence_change':
+      return `Confluence picture changed${symbolTag(symbol)}`;
   }
 }
 
@@ -163,6 +167,8 @@ export function alertDescription(type: AlertType, details?: string, symbol?: str
         return `The mirror target changed${symbolTag(symbol)}. Showing the arithmetic delta; it is not advice.`;
       case 'vix_level':
         return `The VIX, a market-wide measure of expected near-term volatility, has moved into a new level that historically mattered to market participants.`;
+      case 'confluence_change':
+        return `The confluence picture for a symbol changed. This is an observation of the evidence picture, not advice.`;
     }
   })();
 
@@ -303,6 +309,7 @@ export const ALERT_THROTTLE: Record<AlertType, { maxPerHour: number }> = {
   thesis_weakening:   { maxPerHour: 3 },
   cluster_breach:     { maxPerHour: 2 },
   drawdown_halt:      { maxPerHour: 1 },
+  confluence_change:  { maxPerHour: 5 },
   asymmetry_warning:  { maxPerHour: 2 },
   fund_capture:       { maxPerHour: 3 },
   fund_13f:           { maxPerHour: 3 },

app/server/src/alerts/__tests__/confluenceProducer.test.ts

@@ -0,0 +1,134 @@
+// Investor Flow — confluenceProducer.test.ts
+//
+// Guards the confluence_change producer contract:
+//   - First evaluation establishes a baseline (silent on first ever check
+//     only when the picture can't yet be judged — but with two evaluations it
+//     fires on the transition).
+//   - Watermarks prevent re-firing on repeated checks of the same evaluation.
+//   - Only subscribed users are notified.
+//   - Description is a picture observation, never a directive.
+
+import { test, describe } from 'node:test';
+import { strict as assert } from 'node:assert';
+import { DatabaseSync } from 'node:sqlite';
+import { readFileSync } from 'node:fs';
+import { dirname, join } from 'node:path';
+import { fileURLToPath } from 'node:url';
+
+import { confluenceChangeProducer } from '../producers/confluenceProducer.ts';
+import { ConfluenceRepository, rackFromSlots } from '../../db/confluenceRepository.ts';
+import { evaluateRack, type SlotAssessment } from '../../confluence/confluenceRack.ts';
+
+const __dirname = dirname(fileURLToPath(import.meta.url));
+const SCHEMA_SQL = readFileSync(join(__dirname, '..', '..', 'db', 'schema.sql'), 'utf8');
+
+function freshDb(): DatabaseSync {
+  const db = new DatabaseSync(':memory:', { enableForeignKeyConstraints: true });
+  db.exec(SCHEMA_SQL);
+  return db;
+}
+
+function addUser(db: DatabaseSync, id: string) {
+  db.prepare(
+    `INSERT OR IGNORE INTO users (id, email, pw_hash, complexity, drawdown_tolerance, created_at)
+     VALUES (?, ?, '', 'beginner', 20, ?)`,
+  ).run(id, `${id}@t.local`, new Date().toISOString());
+}
+
+function subscribeGlobal(db: DatabaseSync, userId: string, alertType: string) {
+  db.prepare(
+    `INSERT INTO alerts (id, owner_id, symbol, alert_type, enabled, params, created_at)
+     VALUES (?, ?, NULL, ?, 1, '{}', ?)`,
+  ).run(`sub-${userId}-${alertType}`, userId, alertType, new Date().toISOString());
+}
+
+function alertCount(db: DatabaseSync, type: string): number {
+  const r = db.prepare('SELECT COUNT(*) AS n FROM alert_events WHERE type = ?').get(type) as { n: number };
+  return r.n;
+}
+
+const fired = (id: string): SlotAssessment => ({ id, state: 'fired' });
+const notFired = (id: string): SlotAssessment => ({ id, state: 'not-fired' });
+
+function seedRack(db: DatabaseSync): string {
+  const repo = new ConfluenceRepository(db);
+  repo.saveRack(rackFromSlots('rt', 'Rack T', ['goldenCross', 'relVolume', 'cotPositioning'], { isSystem: true }));
+  return 'rt';
+}
+
+describe('confluence_change producer', () => {
+  test('fires once when the picture deteriorates between two evaluations', async () => {
+    const db = freshDb();
+    addUser(db, 'u1');
+    subscribeGlobal(db, 'u1', 'confluence_change');
+    const rackId = seedRack(db);
+    const repo = new ConfluenceRepository(db);
+
+    // Baseline: constructive.
+    repo.saveEvaluation(evaluateRack('PLTR', '2026-01-05', [
+      fired('goldenCross'), fired('relVolume'), notFired('cotPositioning'),
+    ]), rackId, 'ev1');
+    // Deterioration: death-cross + cautious signals.
+    repo.saveEvaluation(evaluateRack('PLTR', '2026-02-02', [
+      notFired('goldenCross'), fired('cotPositioning'), { id: 'deathCross', state: 'fired' },
+    ]), rackId, 'ev2');
+
+    const alerts = await confluenceChangeProducer.check(db);
+    assert.equal(alerts.length, 1);
+    assert.equal(alerts[0].symbol, 'PLTR');
+    assert.equal(alerts[0].type, 'confluence_change');
+    assert.match(alerts[0].description, /PLTR became more cautious/);
+    assert.doesNotMatch(alerts[0].description, /\b(buy|sell|long|short)\b/i);
+    assert.equal(alertCount(db, 'confluence_change'), 1);
+
+    // Re-run: watermark prevents re-fire.
+    const again = await confluenceChangeProducer.check(db);
+    assert.equal(again.length, 0);
+    assert.equal(alertCount(db, 'confluence_change'), 1);
+  });
+
+  test('fires on improvement too', async () => {
+    const db = freshDb();
+    addUser(db, 'u1');
+    subscribeGlobal(db, 'u1', 'confluence_change');
+    const rackId = seedRack(db);
+    const repo = new ConfluenceRepository(db);
+
+    repo.saveEvaluation(evaluateRack('PLTR', '2026-01-05', [
+      { id: 'deathCross', state: 'fired' }, notFired('goldenCross'),
+    ]), rackId, 'ev1');
+    repo.saveEvaluation(evaluateRack('PLTR', '2026-02-02', [
+      fired('goldenCross'), fired('relVolume'), notFired('cotPositioning'),
+    ]), rackId, 'ev2');
+
+    const alerts = await confluenceChangeProducer.check(db);
+    assert.equal(alerts.length, 1);
+    assert.match(alerts[0].description, /PLTR improved/);
+  });
+
+  test('does not alert unsubscribed or empty-data installs', async () => {
+    const db = freshDb();
+    addUser(db, 'u1');
+    const rackId = seedRack(db);
+    const repo = new ConfluenceRepository(db);
+    repo.saveEvaluation(evaluateRack('PLTR', '2026-01-05', [fired('goldenCross')]), rackId, 'ev1');
+    repo.saveEvaluation(evaluateRack('PLTR', '2026-02-02', [notFired('goldenCross')]), rackId, 'ev2');
+
+    const alerts = await confluenceChangeProducer.check(db); // no subscription
+    assert.equal(alerts.length, 0);
+  });
+
+  test('silent when the picture is unchanged', async () => {
+    const db = freshDb();
+    addUser(db, 'u1');
+    subscribeGlobal(db, 'u1', 'confluence_change');
+    const rackId = seedRack(db);
+    const repo = new ConfluenceRepository(db);
+
+    repo.saveEvaluation(evaluateRack('PLTR', '2026-01-05', [fired('goldenCross'), fired('relVolume')]), rackId, 'ev1');
+    repo.saveEvaluation(evaluateRack('PLTR', '2026-02-02', [fired('goldenCross'), fired('relVolume')]), rackId, 'ev2');
+
+    const alerts = await confluenceChangeProducer.check(db);
+    assert.equal(alerts.length, 0); // mixed→mixed at same tier, no drift
+  });
+});

app/server/src/alerts/producers/confluenceProducer.ts

@@ -0,0 +1,177 @@
+// Investor Flow — confluence_change alert producer (M22, slice 5)
+//
+// Fires when a symbol's confluence PICTURE changes meaningfully between two
+// stored rack evaluations: quality tier shift (e.g. moderate-bullish →
+// mixed) or a material net-evidence move with slot flips.
+//
+// Primary-rule safe (ADR-0007): the message describes the picture transition
+// ("the confluence picture for PLTR deteriorated"), it never says buy or sell.
+// Watermark = last evaluation as-of seen per (rack, symbol), so an unchanged
+// evaluation never re-fires.
+
+import type { DatabaseSync } from 'node:sqlite';
+import { randomUUID } from 'node:crypto';
+import type { Alert } from '../AlertEngine.ts';
+import { createAlert, buildDedupKey } from '../AlertEngine.ts';
+import type { AlertProducer } from './types.ts';
+import { getSubscribedUsers, persistAlert, isDuplicate, readComparisonState, writeComparisonState } from './types.ts';
+import { detectPictureChange, type ConfluenceEvaluation, type PictureQuality } from '../../confluence/confluenceRack.ts';
+import { ConfluenceRepository } from '../../db/confluenceRepository.ts';
+
+/** Human label used in the alert title (compass calibration, not a directive). */
+const QUALITY_LABEL: Record<PictureQuality, string> = {
+  'strong-bullish': 'strongly constructive',
+  'moderate-bullish': 'constructive',
+  'weak-bullish': 'mildly constructive',
+  'mixed': 'mixed',
+  'weak-bearish': 'mildly cautious',
+  'moderate-bearish': 'cautious',
+  'strong-bearish': 'strongly cautious',
+  'sparse': 'insufficient data',
+};
+
+interface EvaluationPair {
+  rackId: string;
+  symbol: string;
+  current: ConfluenceEvaluation;
+  previous: ConfluenceEvaluation | null;
+}
+
+interface StoredPair {
+  current: ConfluenceEvaluation | null;
+  previous: ConfluenceEvaluation | null;
+}
+
+/** Load the latest two evaluations per (symbol, rack) that have assessments. */
+function loadLatestPairs(db: DatabaseSync): EvaluationPair[] {
+  const repo = new ConfluenceRepository(db);
+  const rows = db.prepare(
+    `SELECT id, symbol, as_of, rack_id, assessments_json, bull_evidence, bear_evidence,
+            bull_count, bear_count, assessed_count, net_evidence, total_evidence, quality, created_at,
+            ROW_NUMBER() OVER (PARTITION BY symbol, rack_id ORDER BY as_of DESC, created_at DESC) AS rn
+       FROM confluence_evaluations`,
+  ).all() as Array<Record<string, unknown> & { rn: number }>;
+
+  // repo has no bulk loader; reconstruct via its mapper by pulling each row id
+  // through the evaluation SELECT is wasteful — do a light local rebuild instead.
+  const byKey = new Map<string, StoredPair>();
+  const parseAssessments = (raw: string) => {
+    try {
+      const v = JSON.parse(raw) as unknown;
+      return Array.isArray(v) ? (v as ConfluenceEvaluation['assessments']) : [];
+    } catch {
+      return [];
+    }
+  };
+  const toEval = (r: Record<string, unknown>): ConfluenceEvaluation => ({
+    symbol: r.symbol as string,
+    asOf: r.as_of as string,
+    assessments: parseAssessments(r.assessments_json as string),
+    bullEvidence: Number(r.bull_evidence),
+    bearEvidence: Number(r.bear_evidence),
+    bullCount: Number(r.bull_count),
+    bearCount: Number(r.bear_count),
+    assessedCount: Number(r.assessed_count),
+    netEvidence: Number(r.net_evidence),
+    totalEvidence: Number(r.total_evidence),
+    quality: r.quality as PictureQuality,
+  });
+
+  for (const row of rows) {
+    const key = `${row.rack_id}::${row.symbol}`;
+    const pair = byKey.get(key) ?? { current: null, previous: null } as StoredPair;
+    if (pair.current === null) {
+      pair.current = toEval(row);
+    } else if (pair.previous === null) {
+      pair.previous = toEval(row);
+    }
+    byKey.set(key, pair);
+  }
+
+  const out: EvaluationPair[] = [];
+  for (const [key, pair] of byKey) {
+    if (pair.current === null) continue;
+    const [rackId, symbol] = key.split('::');
+    out.push({ rackId, symbol, current: pair.current, previous: pair.previous });
+  }
+  return out;
+}
+
+export const confluenceChangeProducer: AlertProducer = {
+  alertType: 'confluence_change',
+  frequency: 'batched',
+
+  async check(db: DatabaseSync): Promise<Alert[]> {
+    const alerts: Alert[] = [];
+
+    // Sanity: only run when there is data (guards empty installs).
+    const count = db.prepare('SELECT COUNT(*) AS n FROM confluence_evaluations').get() as { n: number };
+    if (Number(count.n) === 0) return alerts;
+
+    const pairs = loadLatestPairs(db);
+    for (const pair of pairs) {
+      const { rackId, symbol, current, previous } = pair;
+      if (current.assessedCount === 0) continue; // empty picture, no basis to alert
+
+      const change = detectPictureChange(previous, current);
+      if (!change.changed) {
+        // Still watermark the seen evaluation so it won't refire after restart.
+        writeComparisonState(db, `${rackId}::${symbol}`, 'confluence_change', {
+          lastAsOf: current.asOf,
+          quality: current.quality,
+        });
+        continue;
+      }
+
+      const users = getSubscribedUsers(db, 'confluence_change', symbol);
+      for (const userId of users) {
+        const eventId = `${rackId}:${symbol}:${current.asOf}:${current.quality}`;
+        const dedupKey = buildDedupKey('confluence_change', symbol, eventId);
+        if (isDuplicate(db, dedupKey)) continue;
+
+        const fromLabel = change.previous ? QUALITY_LABEL[change.previous] : 'a prior evaluation';
+        const toLabel = QUALITY_LABEL[current.quality];
+        const direction = change.changeType === 'improved' ? 'improved' : 'became more cautious';
+
+        const flipped = change.flippedSlotIds.length > 0
+          ? `Slots that changed: ${change.flippedSlotIds.join(', ')}. `
+          : '';
+        const shift = current.quality === 'sparse'
+          ? ''
+          : `Net evidence shift: ${change.netEvidenceShift >= 0 ? '+' : ''}${change.netEvidenceShift.toFixed(2)}.`;
+
+        const description =
+          `The confluence picture for ${symbol} ${direction}: ${fromLabel} → ${toLabel} ` +
+          `(as of ${current.asOf}). ${flipped}${shift}`;
+
+        const alert = createAlert(
+          randomUUID(),
+          userId,
+          'confluence_change',
+          symbol,
+          description,
+          eventId,
+          {
+            rackId,
+            symbol,
+            asOf: current.asOf,
+            from: change.previous ?? undefined,
+            to: change.current ?? undefined,
+            changeType: change.changeType,
+            netEvidenceShift: change.netEvidenceShift,
+            flippedSlots: change.flippedSlotIds,
+          },
+        );
+        persistAlert(db, alert);
+        alerts.push(alert);
+      }
+
+      writeComparisonState(db, `${rackId}::${symbol}`, 'confluence_change', {
+        lastAsOf: current.asOf,
+        quality: current.quality,
+      });
+    }
+
+    return alerts;
+  },
+};

app/server/src/cache/CacheRepository.ts

@@ -12,14 +12,14 @@ import {
 } from '../queue/sourceRatePolicy.ts';
 import { KvReadCache } from './LruCache.ts';
 
-export type SourceKind = 'yfinance' | 'nasdaq' | 'finra-bulk' | 'finra-si' | 'sec' | 'sec-fetch' | 'sec-sc-fetch' | 'sec-tickers' | 'reddit' | 'x' | 'macro' | 'llm' | 'sec-lint-holders' | 'sec-lint-insiders' | 'fred';
+export type SourceKind = 'yfinance' | 'nasdaq' | 'finra-bulk' | 'finra-si' | 'sec' | 'sec-fetch' | 'sec-sc-fetch' | 'sec-tickers' | 'reddit' | 'x' | 'macro' | 'llm' | 'sec-lint-holders' | 'sec-lint-insiders' | 'fred' | 'cot';
 export type TickerKind = 'equity' | 'crypto' | 'etf' | 'index';
 export type CacheKey = string; // `${SourceKind}:${kind}:${id}` e.g. 'yfinance:quote:NVDA', 'yfinance:candles:NVDA:1d'
 export type TtlClass =
   | 'live_quote' | 'intraday' | 'daily_permanent' | 'options_snapshot'
   | 'filing_immutable' | 'quarterly_immutable' | 'thread_7d'
   | 'macro_event' | 'regime_classification' | 'llm_summary' | 'symbol_meta'
-  | 'short_interest' | 'dividend_fundamentals' | 'fred_macro';
+  | 'short_interest' | 'dividend_fundamentals' | 'fred_macro' | 'cot_weekly';
 
 export interface Provenance { fetchedAt: string; sourceKind: SourceKind; rawSourceId?: string; }
 
@@ -68,6 +68,7 @@ export const TTL_MS: Record<TtlClass, number> = {
   short_interest: 24 * 60 * 60_000,      // refreshed twice/month per source
   dividend_fundamentals: 7 * 24 * 60 * 60_000, // weekly (yield/payout change slowly)
   fred_macro: 24 * 60 * 60_000,          // daily (rates move daily; other series slower)
+  cot_weekly: 7 * 24 * 60 * 60_000,      // COT arrives weekly on Fridays
 };
 
 /** Parse 'source:kind:id...' into { source, kind, id } (id may contain colons). */

app/server/src/confluence/__tests__/confluenceBacktest.test.ts

@@ -0,0 +1,164 @@
+// Investor Flow — confluenceBacktest.test.ts
+// Pure follow-through math + resolvability over synthetic candles.
+
+import { describe, it, test } from 'node:test';
+import assert from 'node:assert/strict';
+
+import {
+  candleIndexForDate,
+  checkFollowThrough,
+  expectedDirection,
+  reliabilityWeightFrom,
+  runSlotBacktest,
+  resolveSignalHistory,
+} from '../confluenceBacktest.ts';
+import type { PriceCandle } from '../../cache/CacheRepository.ts';
+
+const day = (offset: number): string => {
+  const d = new Date(Date.UTC(2025, 0, 5 + offset));
+  return d.toISOString().slice(0, 10);
+};
+
+function candles(startOffset = 0, n = 120, step = 1): PriceCandle[] {
+  const out: PriceCandle[] = [];
+  let price = 100;
+  for (let i = 0; i < n; i++) {
+    price += step;
+    out.push({ ts: day(startOffset + i), o: price, h: price + 1, l: price - 1, c: price, v: 1e6, adjClose: price });
+  }
+  return out;
+}
+
+describe('expectedDirection', () => {
+  it('maps bull slots to up and exit/bear slots to down', () => {
+    assert.equal(expectedDirection('goldenCross'), 'up');
+    assert.equal(expectedDirection('deathCross'), 'down');
+    assert.equal(expectedDirection('insiderInformedSell'), 'down');
+  });
+
+  it('returns null for unknown slots', () => {
+    assert.equal(expectedDirection('notASlot'), null);
+  });
+});
+
+describe('candleIndexForDate', () => {
+  it('locates the exact candle ts-date', () => {
+    const cs = candles();
+    assert.equal(candleIndexForDate(cs, day(7)), 7);
+  });
+
+  it('returns -1 when the date is absent', () => {
+    assert.equal(candleIndexForDate(candles(), '1999-01-01'), -1);
+  });
+});
+
+describe('checkFollowThrough', () => {
+  it('resolves true for an uptrend with up direction', () => {
+    // Up movement: entry at idx 0 close=100, 20 bars later close=120 → +20%.
+    const cs = candles(0, 30, 1);
+    assert.equal(checkFollowThrough(cs, 0, 'up', 4, 0.5), true);
+  });
+
+  it('resolves false for an uptrend with down direction', () => {
+    const cs = candles(0, 30, 1);
+    assert.equal(checkFollowThrough(cs, 0, 'down', 4, 0.5), false);
+  });
+
+  it('returns null when not enough forward bars exist', () => {
+    const cs = candles(0, 10, 1); // only 10 bars; need 20 forward
+    assert.equal(checkFollowThrough(cs, 5, 'up', 4, 0.5), null);
+  });
+
+  it('returns null inside the dead band (flat outcome)', () => {
+    const cs = candles(0, 30, 0.001); // ~0.1% per bar, likely flat over 20
+    const r = checkFollowThrough(cs, 0, 'up', 4, 5);
+    assert.equal(r, null);
+  });
+});
+
+describe('runSlotBacktest', () => {
+  it('computes per-slot hit rates from a fire corpus', () => {
+    const cs = candles(0, 60, 1);
+    const fires = [
+      { symbol: 'PLTR', asOf: day(0), slotId: 'goldenCross' },
+      { symbol: 'PLTR', asOf: day(5), slotId: 'goldenCross' },
+      { symbol: 'PLTR', asOf: day(10), slotId: 'deathCross' },
+      { symbol: 'PLTR', asOf: day(15), slotId: 'deathCross' },
+    ];
+    const result = runSlotBacktest('PLTR', cs, fires, 4);
+    const gc = result.slots.find((s) => s.slotId === 'goldenCross')!;
+    const dc = result.slots.find((s) => s.slotId === 'deathCross')!;
+
+    // Both golden crosses resolve up-confirmed (both fire bars have 20 forward bars in a 60-bar series).
+    assert.equal(gc.fires, 2);
+    assert.equal(gc.confirmedFires, 2);
+    assert.equal(gc.hitRate, 1);
+
+    // Death crosses expect the price down; uptrend means both are false alarms.
+    assert.equal(dc.fires, 2);
+    assert.equal(dc.confirmedFires, 0);
+    assert.equal(dc.falseAlarms, 2);
+    assert.equal(dc.hitRate, 0);
+  });
+
+  it('flags thin samples in the caveat', () => {
+    const result = runSlotBacktest('PLTR', candles(), [
+      { symbol: 'PLTR', asOf: day(0), slotId: 'goldenCross' },
+    ], 4);
+    assert.ok(result.sampleSizeCaveat.includes('limited'));
+  });
+
+  it('ignores fires for other symbols and unknown slots', () => {
+    const cs = candles();
+    const result = runSlotBacktest('PLTR', cs, [
+      { symbol: 'NVDA', asOf: day(0), slotId: 'goldenCross' },
+      { symbol: 'PLTR', asOf: day(0), slotId: 'bogus' },
+    ], 4);
+    assert.equal(result.fires, 0);
+    assert.equal(result.slots.length, 0);
+  });
+
+  it('computes reliabilityWeight from sample', () => {
+    assert.equal(reliabilityWeightFrom(0, 0), 0.5);
+    assert.equal(reliabilityWeightFrom(10, 10), 1.25);
+    assert.equal(reliabilityWeightFrom(10, 4), 0.5);
+    assert.equal(reliabilityWeightFrom(3, 3), 0.75); // thin sample dampened
+  });
+});
+
+test('resolveSignalHistory defers when forward bars are insufficient, resolves when available', async () => {
+  const { createDb, initSchema } = await import('../../db/client.ts');
+  const { ConfluenceRepository, rackFromSlots } = await import('../../db/confluenceRepository.ts');
+  const { evaluateRack } = await import('../confluenceRack.ts');
+
+  const db = createDb({ path: ':memory:' });
+  initSchema(db);
+  const repo = new ConfluenceRepository(db);
+  repo.saveRack(rackFromSlots('rt', 'Rack T', ['goldenCross'], { isSystem: true }));
+
+  // 15 fire events spread over dates with only 10 bars after the last → most deferred.
+  // Build a series of 22 candles (2 more than needed for a fire at bar 0 + 20).
+  const cs = candles(0, 22, 1);
+  for (let i = 0; i < 3; i++) {
+    const ev = evaluateRack('PLTR', day(i * 2), [
+      { id: 'goldenCross', state: 'fired' as const },
+      { id: 'relVolume', state: 'not-fired' as const },
+    ]);
+    repo.saveEvaluation(ev, 'rt', `ev-${i}`);
+    repo.logSignalFires(ev, 'rt');
+  }
+
+  const candleLookup = async (symbol: string) => (symbol === 'PLTR' ? cs : []);
+
+  // First pass: fires near the end may be deferred.
+  await resolveSignalHistory(db, candleLookup, 4);
+
+  const allSignals = db.prepare(
+    `SELECT id, price_resolved, verdict FROM confluence_signal_history ORDER BY fired_at`,
+  ).all() as { id: string; price_resolved: number; verdict: string | null }[];
+  assert.equal(allSignals.length, 3);
+  assert.equal(allSignals[0].price_resolved, 1);
+  assert.equal(allSignals[0].verdict, 'real');
+  // Later fires (asOf day 4) have fewer than 20 forward bars → deferred.
+  assert.equal(allSignals[2].price_resolved, 0);
+});

app/server/src/confluence/__tests__/confluenceRack.test.ts

@@ -0,0 +1,169 @@
+// Investor Flow — confluenceRack.test.ts
+// Pure-logic tests for rack evaluation, redundancy discounting, and
+// picture-change detection.
+
+import { describe, it } from 'node:test';
+import assert from 'node:assert/strict';
+
+import {
+  evaluateRack,
+  labelQuality,
+  detectPictureChange,
+  flippedSlots,
+  slotBody,
+} from '../confluenceRack.ts';
+import type { SlotAssessment } from '../confluenceRack.ts';
+import { confluenceslotById, CONFLUENCE_SLOTS } from '../confluenceSlots.ts';
+
+const fired = (id: string): SlotAssessment => ({ id, state: 'fired' });
+const notFired = (id: string): SlotAssessment => ({ id, state: 'not-fired' });
+
+describe('confluence catalog', () => {
+  it('defines exactly 34 slots across all six families', () => {
+    assert.equal(CONFLUENCE_SLOTS.length, 34);
+    const families = new Set(CONFLUENCE_SLOTS.map((s) => s.family));
+    assert.deepEqual([...families].sort(), ['flows', 'institutional', 'macro', 'seasonal', 'sentiment', 'technical']);
+  });
+
+  it('every defined slot has a primary-rule-safe explain sentence', () => {
+    const directivePattern = /\b(should|recommend|buy now|sell now|go long|go short|recommendation|buy the|sell the|long position|short position)\b/i;
+    for (const s of CONFLUENCE_SLOTS) {
+      assert.ok(s.explain.length > 10, `${s.id} explain note too short`);
+      assert.ok(!directivePattern.test(s.explain), `${s.id} explain note reads like a directive: ${s.explain}`);
+    }
+  });
+});
+
+describe('slotBody', () => {
+  it('exit slots map to bear evidence, bull/bear stay put', () => {
+    assert.equal(slotBody(confluenceslotById.get('goldenCross')!), 'bull');
+    assert.equal(slotBody(confluenceslotById.get('deathCross')!), 'bear');
+    assert.equal(slotBody(confluenceslotById.get('consumerSentimentHigh')!), 'bear');
+  });
+});
+
+describe('labelQuality', () => {
+  it('returns sparse below the minimum evidence threshold', () => {
+    assert.equal(labelQuality(0.5, 0.2, 0.7), 'sparse');
+  });
+
+  it('labels strong-bullish from a one-sided heavy bull weighting', () => {
+    assert.equal(labelQuality(7, 0.5, 7.5), 'strong-bullish');
+  });
+
+  it('labels strong-bearish from heavy bear + exit evidence', () => {
+    assert.equal(labelQuality(0.5, 6, 6.5), 'strong-bearish');
+  });
+
+  it('labels moderate-bullish between strong and weak', () => {
+    assert.equal(labelQuality(2.5, 0.5, 3), 'moderate-bullish');
+  });
+
+  it('labels weak-bullish when barely directional', () => {
+    assert.equal(labelQuality(1.7, 0.7, 2.4), 'weak-bullish');
+  });
+
+  it('labels mixed when neither side holds 60% of evidence', () => {
+    assert.equal(labelQuality(1.2, 1.1, 2.3), 'mixed');
+  });
+});
+
+describe('evaluateRack', () => {
+  it('applies redundancy discount within a redundancy group', () => {
+    // goldenCross + trendAlignment + pullbackToEMA21 all fire (all same group).
+    // groupWeight = 1 + 0.5 + 0.25 = 1.75, not 3.0.
+    const ev = evaluateRack('PLTR', '2026-01-05', [
+      fired('goldenCross'),
+      fired('trendAlignment'),
+      fired('pullbackToEMA21'),
+    ]);
+    assert.equal(ev.bullCount, 3);
+    assert.ok(Math.abs(ev.bullEvidence - 1.75) < 1e-9, `bullEvidence=${ev.bullEvidence}`);
+  });
+
+  it('counts distinct families as independent evidence', () => {
+    // One technical + one institutional + one seasonal + one flows: 4 independent facts.
+    const ev = evaluateRack('PLTR', '2026-01-05', [
+      fired('goldenCross'),
+      fired('instNetActivePositive'),
+      fired('seasonalFavorableMonth'),
+      fired('etfFlowPositive'),
+    ]);
+    assert.equal(ev.bullEvidence, 4);
+    assert.equal(ev.bullCount, 4);
+  });
+
+  it('counts exit slots as bear evidence', () => {
+    const ev = evaluateRack('PLTR', '2026-01-05', [
+      fired('deathCross'),
+      fired('rsiOverbought'),
+      fired('insiderInformedSell'),
+    ]);
+    assert.equal(ev.bearCount, 3);
+    assert.equal(ev.bullCount, 0);
+    assert.equal(ev.quality, 'moderate-bearish');
+  });
+
+  it('ignores unknown slot ids but still counts assessed coverage', () => {
+    const ev = evaluateRack('PLTR', '2026-01-05', [
+      fired('goldenCross'),
+      fired('notARealSlot'),
+      notFired('relVolume'),
+    ]);
+    assert.equal(ev.assessedCount, 2);
+    assert.equal(ev.bullCount, 1);
+  });
+
+  it('mixed picture when bull and bear evidence are both present and balanced', () => {
+    const ev = evaluateRack('NVDA', '2026-01-05', [
+      fired('goldenCross'),
+      fired('instNetActivePositive'),
+      fired('deathCross'),
+      fired('insiderInformedSell'),
+    ]);
+    assert.equal(ev.quality, 'mixed');
+  });
+});
+
+describe('detectPictureChange', () => {
+  it('reports improved when quality rank rises', () => {
+    const prev = evaluateRack('PLTR', '2026-01-05', [fired('deathCross')]);
+    const curr = evaluateRack('PLTR', '2026-01-07', [fired('goldenCross')]);
+    const change = detectPictureChange(prev, curr);
+    assert.equal(change.changed, true);
+    assert.equal(change.changeType, 'improved');
+  });
+
+  it('reports deteriorated when quality rank falls', () => {
+    const prev = evaluateRack('PLTR', '2026-01-05', [fired('goldenCross'), fired('instNetActivePositive')]);
+    const curr = evaluateRack('PLTR', '2026-01-07', [fired('deathCross'), fired('insiderInformedSell')]);
+    const change = detectPictureChange(prev, curr);
+    assert.equal(change.changed, true);
+    assert.equal(change.changeType, 'deteriorated');
+  });
+
+  it('reports unchanged for identical evaluations', () => {
+    const ev = evaluateRack('PLTR', '2026-01-05', [fired('goldenCross')]);
+    const change = detectPictureChange(ev, ev);
+    assert.equal(change.changed, false);
+    assert.equal(change.changeType, 'unchanged');
+    assert.deepEqual(change.flippedSlotIds, []);
+  });
+
+  it('flag lists flipped slot ids', () => {
+    const prev = evaluateRack('PLTR', '2026-01-05', [fired('goldenCross'), notFired('relVolume')]);
+    const curr = evaluateRack('PLTR', '2026-01-07', [notFired('goldenCross'), fired('relVolume')]);
+    const change = detectPictureChange(prev, curr);
+    assert.deepEqual(change.flippedSlotIds.sort(), ['goldenCross', 'relVolume']);
+  });
+});
+
+describe('flippedSlots', () => {
+  it('finds flips including newly-added and removed slots', () => {
+    const flips = flippedSlots(
+      [fired('goldenCross'), notFired('relVolume')],
+      [notFired('goldenCross'), fired('relVolume'), fired('cotPositioning')],
+    );
+    assert.deepEqual(flips.sort(), ['cotPositioning', 'goldenCross', 'relVolume']);
+  });
+});

app/server/src/confluence/confluenceBacktest.ts

@@ -0,0 +1,300 @@
+// Investor Flow — Confluence Backtest + Signal Resolver (M22, slice 4)
+//
+// The closed loop: every slot fire is logged to confluence_signal_history, then
+// resolved to 'real' | 'false_alarm' by N-week price follow-through aligned with
+// the slot body (bull slots expect the price up, bear/exit slots expect it down).
+//
+// `runSlotBacktest` answers "when this slot fired, did price follow through N
+// weeks later?" for a corpus of fire events — the raw material for a per-slot
+// reliability scorecard that teaches the rack which confluences deserve weight.
+//
+// `resolveSignalHistory` walks pending `confluence_signal_history` rows and flips
+// them to confirmed / false_alarm once enough forward bars exist. Mirrors the
+// rotation_signals gamma two-stage, but for confluence slots.
+//
+// Pure where possible: candle math is pure; the resolver is a thin DB+data shim.
+
+import type { PriceCandle } from '../cache/CacheRepository.ts';
+import { confluenceslotById, type ConfluenceSlot } from './confluenceSlots.ts';
+
+// ---------------------------------------------------------------------------
+// Types
+// ---------------------------------------------------------------------------
+
+/** One recorded slot fire at a date, as the backtest input unit. */
+export interface ConfluenceFireEvent {
+  symbol: string;
+  /** YYYY-MM-DD the slot fired (as-of of the evaluation that fired it). */
+  asOf: string;
+  slotId: string;
+}
+
+/** Directional expectation of a slot body for follow-through resolution. */
+export type FollowThrough = 'up' | 'down';
+
+/** Per-slot follow-through stats across the backtest corpus. */
+export interface SlotFollowThroughStats {
+  slotId: string;
+  name: string;
+  body: ConfluenceSlot['body'];
+  direction: FollowThrough;
+  fires: number;
+  /** Fires with enough forward bars to resolve. */
+  resolvedFires: number;
+  /** Resolved fires where price moved the expected way. */
+  confirmedFires: number;
+  /** resolved - confirmed (price moved the opposite way). */
+  falseAlarms: number;
+  /** confirmed / resolved over resolvedFires; null when none resolved. */
+  hitRate: number | null;
+  /** 0..1 multiplier a rack can apply to this slot's weight. */
+  reliabilityWeight: number;
+}
+
+/** Whole backtest summary (ADR-0007-safe: no directional advice). */
+export interface ConfluenceBacktestResult {
+  symbol: string;
+  fires: number;
+  /** Calendar-day horizon used (weeks * 7). */
+  horizonDays: number;
+  slots: SlotFollowThroughStats[];
+  sampleSizeCaveat: string;
+}
+
+// ---------------------------------------------------------------------------
+// Pure helpers
+// ---------------------------------------------------------------------------
+
+/** Expected follow-through direction for a slot: bull ⇒ up, bear/exit ⇒ down. */
+export function expectedDirection(slotId: string): FollowThrough | null {
+  const slot = confluenceslotById.get(slotId);
+  if (!slot) return null;
+  return slot.body === 'bull' ? 'up' : 'down';
+}
+
+/** The candle index whose ts-date is `asOf`, or -1. Candles sorted ascending. */
+export function candleIndexForDate(candles: PriceCandle[], asOf: string): number {
+  for (let i = 0; i < candles.length; i++) {
+    const ts = (candles[i].ts ?? '').slice(0, 10);
+    if (ts === asOf) return i;
+  }
+  return -1;
+}
+
+/**
+ * Pure follow-through check: did price move the expected way `weeks` later?
+ * Uses the close `weeks*TRADING_DAYS_PER_WEEK` bars after the fire bar and
+ * permits a small dead-band so a whipsaw to flat is treated as unresolved
+ * rather than a confirmation or false alarm. Returns null when there is not
+ * enough forward data to resolve.
+ */
+export function checkFollowThrough(
+  candles: PriceCandle[],
+  fireIndex: number,
+  direction: FollowThrough,
+  weeks = 4,
+  deadBandPct = 0.5,
+): boolean | null {
+  const step = weeks * 5;
+  const target = fireIndex + step;
+  if (target >= candles.length) return null;
+
+  const entry = candles[fireIndex].c;
+  const exit = candles[target].c;
+  if (!Number.isFinite(entry) || !Number.isFinite(exit) || entry <= 0) return null;
+
+  const retPct = ((exit - entry) / entry) * 100;
+  if (Math.abs(retPct) < deadBandPct) return null;
+  return direction === 'up' ? retPct > 0 : retPct < 0;
+}
+
+/**
+ * Backtest a corpus of fire events against candles: per-slot follow-through
+ * stats over `weeks`-week forward windows. Pure.
+ */
+export function runSlotBacktest(
+  symbol: string,
+  candles: PriceCandle[],
+  fires: ConfluenceFireEvent[],
+  weeks = 4,
+): ConfluenceBacktestResult {
+  const hitsBySlot = new Map<string, { fires: number; resolved: number; confirmed: number }>();
+
+  for (const fire of fires) {
+    if (fire.symbol !== symbol) continue;
+    const dir = expectedDirection(fire.slotId);
+    if (dir === null) continue;
+
+    const row = hitsBySlot.get(fire.slotId) ?? { fires: 0, resolved: 0, confirmed: 0 };
+    row.fires += 1;
+
+    const idx = candleIndexForDate(candles, fire.asOf);
+    if (idx >= 0) {
+      const confirmed = checkFollowThrough(candles, idx, dir, weeks);
+      if (confirmed !== null) {
+        row.resolved += 1;
+        if (confirmed) row.confirmed += 1;
+      }
+    }
+    hitsBySlot.set(fire.slotId, row);
+  }
+
+  const slots: SlotFollowThroughStats[] = [...hitsBySlot.entries()].map(([slotId, s]) => {
+    const slot = confluenceslotById.get(slotId);
+    const hitRate = s.resolved > 0 ? s.confirmed / s.resolved : null;
+    return {
+      slotId,
+      name: slot?.name ?? slotId,
+      body: slot?.body ?? 'bull',
+      direction: expectedDirection(slotId) ?? 'up',
+      fires: s.fires,
+      resolvedFires: s.resolved,
+      confirmedFires: s.confirmed,
+      falseAlarms: s.resolved - s.confirmed,
+      hitRate,
+      reliabilityWeight: reliabilityWeightFrom(s.resolved, s.confirmed),
+    };
+  }).sort((a, b) => (b.hitRate ?? 0) - (a.hitRate ?? 0));
+
+  const totalFires = slots.reduce((acc, s) => acc + s.fires, 0);
+  return {
+    symbol,
+    fires: totalFires,
+    horizonDays: weeks * 7,
+    slots,
+    sampleSizeCaveat:
+      totalFires < 30
+        ? `Sample size of ${totalFires} slot-fires is limited. Reliability estimates may not generalize.`
+        : `Sample size: ${totalFires} slot-fires.`,
+  };
+}
+
+/**
+ * A defensible reliability multiplier in [0.5, 1.25] from a slot's resolved
+ * sample: 100% and ≥8 resolved ⇒ 1.25×; worse-than-coin-flip or <2 resolved ⇒
+ * 0.5×; otherwise linear interpolation across [50%, 75%] hit rate. The rack can
+ * multiply per-slot evidence by this to self-tune. Pure.
+ */
+export function reliabilityWeightFrom(resolvedFires: number, confirmedFires: number): number {
+  if (resolvedFires <= 0) return 0.5;
+  const rate = confirmedFires / resolvedFires;
+  if (resolvedFires >= 8 && rate >= 0.9) return 1.25;
+  if (rate <= 0.5 || resolvedFires < 2) return 0.5;
+  if (resolvedFires < 8) return 0.75; // thin sample: hold confidence down
+  return Math.min(1.25, 0.5 + (rate - 0.5) * 2);
+}
+
+// ---------------------------------------------------------------------------
+// Signal-history resolver (thin service shim over the repository)
+// ---------------------------------------------------------------------------
+
+import type { DatabaseSync } from 'node:sqlite';
+import { ConfluenceRepository, type ConfluenceSignalHistoryRow } from '../db/confluenceRepository.ts';
+
+/** Result of resolving one pending signal. */
+export interface SignalResolutionOutcome {
+  signalId: string;
+  symbol: string;
+  slotId: string;
+  verdict: 'real' | 'false_alarm' | 'deferred';
+  priceConfirmed: boolean | null;
+  note: string;
+}
+
+export interface ResolvedSignalsResult {
+  resolved: SignalResolutionOutcome[];
+}
+
+/**
+ * Resolve pending `confluence_signal_history` rows using forward candle data.
+ *
+ * @param db          database handle
+ * @param candleProvider async lookup of a symbol's daily candles (sorted)
+ * @param weeks        follow-through window (default 4)
+ */
+export async function resolveSignalHistory(
+  db: DatabaseSync,
+  candleProvider: (symbol: string) => Promise<PriceCandle[]>,
+  weeks = 4,
+): Promise<ResolvedSignalsResult> {
+  const repo = new ConfluenceRepository(db);
+  const pending = repo.listPendingSignals();
+  if (pending.length === 0) return { resolved: [] };
+
+  const resolved: SignalResolutionOutcome[] = [];
+
+  for (const signal of pending) {
+    if (signal.verdict !== null) continue; // already resolved (guard)
+    const dir = expectedDirection(signal.slotId);
+
+    const outcome = await resolveOneSignal(signal, repo, candleProvider, dir, weeks);
+    resolved.push(outcome);
+  }
+
+  return { resolved };
+}
+
+async function resolveOneSignal(
+  signal: ConfluenceSignalHistoryRow,
+  repo: ConfluenceRepository,
+  candleProvider: (symbol: string) => Promise<PriceCandle[]>,
+  direction: FollowThrough | null,
+  weeks: number,
+): Promise<SignalResolutionOutcome> {
+  try {
+    const candles = await candleProvider(signal.symbol);
+    const idx = candleIndexForDate(candles, signal.firedAt);
+    if (direction !== null && idx >= 0) {
+      const confirmed = checkFollowThrough(candles, idx, direction, weeks);
+      if (confirmed !== null) {
+        repo.resolveSignal(signal.id, confirmed, confirmed ? 'real' : 'false_alarm');
+        return {
+          signalId: signal.id,
+          symbol: signal.symbol,
+          slotId: signal.slotId,
+          verdict: confirmed ? 'real' : 'false_alarm',
+          priceConfirmed: confirmed,
+          note: `follow-through ${confirmed ? 'confirmed' : 'failed'} over ${weeks} weeks`,
+        };
+      }
+      return { signalId: signal.id, symbol: signal.symbol, slotId: signal.slotId, verdict: 'deferred', priceConfirmed: null, note: 'not enough forward bars yet' };
+    }
+    return { signalId: signal.id, symbol: signal.symbol, slotId: signal.slotId, verdict: 'deferred', priceConfirmed: null, note: 'no fire bar or unknown slot' };
+  } catch (e) {
+    return { signalId: signal.id, symbol: signal.symbol, slotId: signal.slotId, verdict: 'deferred', priceConfirmed: null, note: `candle fetch error: ${String(e)}` };
+  }
+}
+
+/** Extract per-slot stats from a live DB for the scorecard view. */
+export function signalHistoryToStats(db: DatabaseSync, symbol?: string): SlotFollowThroughStats[] {
+  const repo = new ConfluenceRepository(db);
+  const rows = symbol ? repo.listSignalsForSymbol(symbol) : repo.listAllSignals();
+
+  const bySlot = new Map<string, { fires: number; resolved: number; confirmed: number }>();
+  for (const s of rows) {
+    const row = bySlot.get(s.slotId) ?? { fires: 0, resolved: 0, confirmed: 0 };
+    row.fires += 1;
+    if (s.priceResolved) {
+      row.resolved += 1;
+      if (s.priceConfirmed) row.confirmed += 1;
+    }
+    bySlot.set(s.slotId, row);
+  }
+
+  return [...bySlot.entries()].map(([slotId, s]) => {
+    const slot = confluenceslotById.get(slotId);
+    const hitRate = s.resolved > 0 ? s.confirmed / s.resolved : null;
+    return {
+      slotId,
+      name: slot?.name ?? slotId,
+      body: slot?.body ?? 'bull',
+      direction: expectedDirection(slotId) ?? 'up',
+      fires: s.fires,
+      resolvedFires: s.resolved,
+      confirmedFires: s.confirmed,
+      falseAlarms: s.resolved - s.confirmed,
+      hitRate,
+      reliabilityWeight: reliabilityWeightFrom(s.resolved, s.confirmed),
+    };
+  });
+}

app/server/src/confluence/confluenceLibrary.ts

@@ -0,0 +1,88 @@
+// Investor Flow — Confluence Library (M22, slice 2)
+//
+// Redundancy modeling for the Confluence Signal Engine. Correlated slots that
+// measure the same underlying phenomenon ("golden cross" and "trend alignment"
+// both read the same moving-average structure) are grouped so a rack does not
+// over-count them as independent evidence.
+//
+// Pure data + accessors. NO I/O. ADR-0007: these are evidence-accounting rules,
+// never recommendations.
+
+import { CONFLUENCE_SLOTS, type ConfluenceSlot, type SlotFamily } from './confluenceSlots.ts';
+
+/**
+ * A redundancy group: slots that corroborate each other because they consume
+ * overlapping inputs. Within a group, the first firing slot contributes full
+ * evidence and each additional firing slot contributes a discounted margin
+ * (`REDUNDANCY_DISCOUNT`, applied multiplicatively) so three correlated
+ * bullish slots are not counted as three independent bullish facts.
+ */
+export interface RedundancyGroup {
+  family: SlotFamily;
+  slots: string[];
+}
+
+/** Default discount applied per additional firing slot within a group. */
+export const REDUNDANCY_DISCOUNT = 0.5;
+
+/** Evidence redundancy groups keyed by the phenomenon they all measure. */
+export const REDUNDANCY_GROUPS: RedundancyGroup[] = [
+  { family: 'technical', slots: ['goldenCross', 'trendAlignment', 'pullbackToEMA21'] },
+  { family: 'technical', slots: ['timeSeriesMomentum12_1', 'relativeStrengthVsSpy', 'momentumForward'] },
+  { family: 'technical', slots: ['macdBullish', 'rsiOversold'] },
+  { family: 'technical', slots: ['macdBearish', 'rsiOverbought'] },
+  { family: 'technical', slots: ['relVolume', 'volumeProfileShelf'] },
+  { family: 'institutional', slots: ['instNetActivePositive', 'insiderInformedBuy30d', 'new13da'] },
+  { family: 'macro', slots: ['ratesRegime', 'macroRegimeUp'] },
+  { family: 'macro', slots: ['consumerSentimentLow', 'breadthThrust'] },
+  { family: 'seasonal', slots: ['seasonalFavorableMonth', 'winterHalfOn', 'electionCycleFavorableYear'] },
+  { family: 'flows', slots: ['etfFlowPositive', 'cotPositioning'] },
+];
+
+/** Map slot id -> the group it belongs to (a slot may be in one group). */
+export const slotGroupIndex: ReadonlyMap<string, RedundancyGroup> = new Map(
+  REDUNDANCY_GROUPS.flatMap((g) => g.slots.map((id) => [id, g] as const)),
+);
+
+/** The group a slot belongs to, if any. */
+export function redundancyGroupFor(slotId: string): RedundancyGroup | undefined {
+  return slotGroupIndex.get(slotId);
+}
+
+/**
+ * Effective evidence weight for a set of firing slots *within one group*.
+ * First firing slot contributes 1.0; each additional contributes
+ * `REDUNDANCY_DISCOUNT` of the previous marginal — so N same-group firings
+ * weigh 1 + 0.5 + 0.25 + ... , never N.
+ */
+export function groupWeight(firingSlotIds: string[]): number {
+  if (firingSlotIds.length === 0) return 0;
+  let w = 0;
+  let margin = 1;
+  for (let i = 0; i < firingSlotIds.length; i++) {
+    w += margin;
+    margin *= REDUNDANCY_DISCOUNT;
+  }
+  return w;
+}
+
+/** A flat description of every slot for explain-note / docs rendering. */
+export function slotCatalog(): { id: string; name: string; family: SlotFamily; body: string; explain: string }[] {
+  return CONFLUENCE_SLOTS.map((s) => ({
+    id: s.id,
+    name: s.name,
+    family: s.family,
+    body: s.body,
+    explain: s.explain,
+  }));
+}
+
+/** Convenience: number of slots per family. */
+export function familyCounts(): Record<SlotFamily, number> {
+  const counts: Record<SlotFamily, number> = { technical: 0, institutional: 0, macro: 0, seasonal: 0, flows: 0, sentiment: 0 };
+  for (const s of CONFLUENCE_SLOTS) counts[s.family] += 1;
+  return counts;
+}
+
+/** Re-export the slot type for consumers of the library. */
+export type { ConfluenceSlot } from './confluenceSlots.ts';

app/server/src/confluence/confluenceRack.ts

@@ -0,0 +1,237 @@
+// Investor Flow — Confluence Rack (M22, slice 2)
+//
+// The aggregator that turns per-slot firing states into a redundancy-aware
+// "picture quality" assessment for a symbol. It:
+//   • buckets fired slots into bullish vs bearish evidence (exit slots count as
+//     bear evidence),
+//   • applies redundancy-group discounting so correlated slots are not
+//     triple-counted,
+//   • labels the overall picture with an evidence-based quality tier,
+//   • detects picture *changes* between consecutive evaluations for the
+//     confluence-change alert producer.
+//
+// Pure: consumes slot assessments, emits a labeled evaluation. NO I/O.
+// ADR-0007: the output is a description of the picture ("strong-bullish",
+// "mixed") — evidence, never a directive to buy or sell.
+
+import { CONFLUENCE_SLOTS, type ConfluenceSlot } from './confluenceSlots.ts';
+import { groupWeight, redundancyGroupFor } from './confluenceLibrary.ts';
+
+/** Per-slot assessment fed into a rack evaluation. */
+export interface SlotAssessment {
+  id: string;
+  /** fired = the slot's evidence is present; not-fired = data seen, evidence absent. */
+  state: 'fired' | 'not-fired';
+  /** Optional 0..1 confidence of the evaluation itself. */
+  confidence?: number;
+  /** Optional free-text note for explain output. */
+  note?: string;
+}
+
+/** A symbol's confluence evaluation for one as-of timestamp. */
+export interface ConfluenceEvaluation {
+  symbol: string;
+  /** ISO date (YYYY-MM-DD) the evaluation is for. */
+  asOf: string;
+  assessments: SlotAssessment[];
+  /** Redundancy-discounted bullish evidence weight. */
+  bullEvidence: number;
+  /** Redundancy-discounted bearish (bear + exit) evidence weight. */
+  bearEvidence: number;
+  /** Raw counts (before redundancy discounting). */
+  bullCount: number;
+  bearCount: number;
+  /** Slots assessed (with data); sheds light on coverage. */
+  assessedCount: number;
+  /** Net bullish evidence (bull - bear). Positive = more bullish evidence. */
+  netEvidence: number;
+  /** Total recognized evidence (bull + bear). */
+  totalEvidence: number;
+  /** Evidence-based picture label. */
+  quality: PictureQuality;
+}
+
+/** Evidence-based picture labels, ordered from most to least constructive. */
+export type PictureQuality =
+  | 'strong-bullish'
+  | 'moderate-bullish'
+  | 'weak-bullish'
+  | 'mixed'
+  | 'weak-bearish'
+  | 'moderate-bearish'
+  | 'strong-bearish'
+  | 'sparse';
+
+/** Ordering for change detection / tier comparison. */
+export const QUALITY_RANK: Record<PictureQuality, number> = {
+  'strong-bullish': 7,
+  'moderate-bullish': 6,
+  'weak-bullish': 5,
+  'mixed': 4,
+  'weak-bearish': 3,
+  'moderate-bearish': 2,
+  'strong-bearish': 1,
+  'sparse': 0,
+};
+
+/** Minimum total evidence before the picture is meaningful rather than 'sparse'. */
+export const MIN_TOTAL_EVIDENCE = 1.0;
+/** Evidence ratio that separates "bullish/bearish" from "mixed". */
+export const DIRECTION_RATIO = 0.6;
+/** Total evidence marking a "strong" vs "moderate" picture. */
+export const STRONG_EVIDENCE = 4.0;
+/** Total evidence marking a "moderate" vs "weak" picture. */
+export const MODERATE_EVIDENCE = 2.0;
+
+/** Which side a slot's firing evidence serves. */
+export function slotBody(slot: ConfluenceSlot): 'bull' | 'bear' {
+  return slot.body === 'exit' ? 'bear' : slot.body;
+}
+
+/** Is `id` a known slot? */
+function isKnownSlot(id: string): boolean {
+  return CONFLUENCE_SLOTS.some((s) => s.id === id);
+}
+
+/**
+ * Evaluate a rack: bucket the fired assessments into bull/bear evidence,
+ * discount per redundancy group, and label the picture.
+ */
+export function evaluateRack(
+  symbol: string,
+  asOf: string,
+  assessments: SlotAssessment[],
+): ConfluenceEvaluation {
+  const slotById = new Map(CONFLUENCE_SLOTS.map((s) => [s.id, s]));
+
+  const firedBull: string[] = [];
+  const firedBear: string[] = [];
+  let assessedCount = 0;
+
+  for (const a of assessments) {
+    if (!isKnownSlot(a.id)) continue;
+    assessedCount += 1;
+    if (a.state !== 'fired') continue;
+    const body = slotBody(slotById.get(a.id)!);
+    if (body === 'bull') firedBull.push(a.id);
+    else firedBear.push(a.id);
+  }
+
+  // Group fired slots by redundancy group; ungrouped slots get singleton buckets.
+  const bucket = (id: string): string => {
+    const g = redundancyGroupFor(id);
+    return g ? `g:${g.slots.join(',')}` : `solo:${id}`;
+  };
+
+  const bullGroups = new Map<string, string[]>();
+  const bearGroups = new Map<string, string[]>();
+  for (const id of firedBull) {
+    const key = bucket(id);
+    const arr = bullGroups.get(key) ?? [];
+    arr.push(id);
+    bullGroups.set(key, arr);
+  }
+  for (const id of firedBear) {
+    const key = bucket(id);
+    const arr = bearGroups.get(key) ?? [];
+    arr.push(id);
+    bearGroups.set(key, arr);
+  }
+
+  let bullEvidence = 0;
+  for (const ids of bullGroups.values()) bullEvidence += groupWeight(ids);
+  let bearEvidence = 0;
+  for (const ids of bearGroups.values()) bearEvidence += groupWeight(ids);
+
+  const netEvidence = bullEvidence - bearEvidence;
+  const totalEvidence = bullEvidence + bearEvidence;
+  const quality = labelQuality(bullEvidence, bearEvidence, totalEvidence);
+
+  return {
+    symbol,
+    asOf,
+    assessments,
+    bullEvidence,
+    bearEvidence,
+    bullCount: firedBull.length,
+    bearCount: firedBear.length,
+    assessedCount,
+    netEvidence,
+    totalEvidence,
+    quality,
+  };
+}
+
+/** Label the picture from the discounted evidence totals. Pure. */
+export function labelQuality(bullEvidence: number, bearEvidence: number, totalEvidence: number): PictureQuality {
+  if (totalEvidence < MIN_TOTAL_EVIDENCE) return 'sparse';
+
+  const ratio = bullEvidence / totalEvidence;
+  if (ratio >= DIRECTION_RATIO) return magnitudeLabel(bullEvidence, 'bullish');
+  if (ratio <= 1 - DIRECTION_RATIO) return magnitudeLabel(bearEvidence, 'bearish');
+  return 'mixed';
+}
+
+function magnitudeLabel(strongSideEvidence: number, side: 'bullish' | 'bearish'): PictureQuality {
+  if (strongSideEvidence >= STRONG_EVIDENCE) return side === 'bullish' ? 'strong-bullish' : 'strong-bearish';
+  if (strongSideEvidence >= MODERATE_EVIDENCE) return side === 'bullish' ? 'moderate-bullish' : 'moderate-bearish';
+  return side === 'bullish' ? 'weak-bullish' : 'weak-bearish';
+}
+
+/** A detected change between two evaluations. */
+export interface ConfluenceChange {
+  changed: boolean;
+  /** Direction of the picture change from the consumer's perspective. */
+  changeType: 'improved' | 'deteriorated' | 'unchanged';
+  current: PictureQuality;
+  previous: PictureQuality | null;
+  netEvidenceShift: number;
+  /** Slot ids whose firing state flipped between the two evaluations. */
+  flippedSlotIds: string[];
+}
+
+/** Minimum net-evidence shift that counts as a *noteworthy* change. */
+export const MIN_CHANGE_SHIFT = 0.35;
+
+/**
+ * Detect whether the picture changed meaningfully between two consecutive
+ * evaluations of the same symbol. Used by the confluence-change alert producer.
+ * Pure.
+ */
+export function detectPictureChange(
+  previous: ConfluenceEvaluation | null,
+  current: ConfluenceEvaluation,
+): ConfluenceChange {
+  if (previous === null) {
+    return { changed: false, changeType: 'unchanged', current: current.quality, previous: null, netEvidenceShift: 0, flippedSlotIds: [] };
+  }
+
+  const flippedSlotIds = flippedSlots(previous.assessments, current.assessments);
+  const netEvidenceShift = current.netEvidence - previous.netEvidence;
+
+  const rankShift = QUALITY_RANK[current.quality] - QUALITY_RANK[previous.quality];
+  if (rankShift > 0 || (rankShift === 0 && netEvidenceShift >= MIN_CHANGE_SHIFT)) {
+    return { changed: true, changeType: 'improved', current: current.quality, previous: previous.quality, netEvidenceShift, flippedSlotIds };
+  }
+  if (rankShift < 0 || (rankShift === 0 && netEvidenceShift <= -MIN_CHANGE_SHIFT)) {
+    return { changed: true, changeType: 'deteriorated', current: current.quality, previous: previous.quality, netEvidenceShift, flippedSlotIds };
+  }
+  return { changed: false, changeType: 'unchanged', current: current.quality, previous: previous.quality, netEvidenceShift, flippedSlotIds };
+}
+
+/** Fired-state flips between two assessment sets. Pure. */
+export function flippedSlots(prev: SlotAssessment[], curr: SlotAssessment[]): string[] {
+  const prevMap = new Map(prev.map((a) => [a.id, a.state]));
+  const currMap = new Map(curr.map((a) => [a.id, a.state]));
+  const flipped: string[] = [];
+  for (const id of prevMap.keys()) {
+    if (currMap.has(id) && prevMap.get(id) !== currMap.get(id)) flipped.push(id);
+  }
+  for (const id of currMap.keys()) {
+    if (!prevMap.has(id)) flipped.push(id);
+  }
+  return flipped;
+}
+
+/** Marker re-export so the module documents the exit→bear convention. */
+export { isBearEvidence } from './confluenceSlots.ts';

app/server/src/confluence/confluenceSlots.ts

@@ -0,0 +1,116 @@
+// Investor Flow — Confluence Slot Catalog (M22, slice 2)
+//
+// The 34-slot confluence inventory for the Confluence Signal Engine. Each slot is
+// a named, independently-evaluable check whose *firing* state contributes bullish
+// or bearish evidence about a symbol's entry/exit quality.
+//
+// Pure data + retrieval helpers: NO I/O. ADR-0007: slots describe "quality of the
+// picture" — never a buy/sell recommendation. Each slot has an ADR-safe explain
+// note phrased as evidence ("relative strength is in the upper decile") rather
+// than a directive ("buy").
+
+/** Slot body: which side of the picture a firing slot supports. */
+export type SlotBody = 'bull' | 'bear' | 'exit';
+
+/**
+ * "exit" is a subclass of bear evidence: it indicates weakening conditions for an
+ * existing position rather than a fresh short thesis. Exit slots are still counted
+ * as bear evidence (risk-off side of the picture).
+ */
+export const isBearEvidence = (body: SlotBody): boolean => body === 'bear' || body === 'exit';
+
+/** The seven confluence families (independent evidence axes). */
+export type SlotFamily =
+  | 'technical'
+  | 'institutional'
+  | 'macro'
+  | 'seasonal'
+  | 'flows'
+  | 'sentiment';
+
+/** Candle granularity a slot consumes (daily is the engine default). */
+export type SlotGranularity = '1d' | '1wk';
+
+/**
+ * A confluence slot definition. `id` is the stable machine key; `name` plain
+ * English; `explain` is the ADR-0007-safe evidence sentence surfaced to users.
+ */
+export interface ConfluenceSlot {
+  id: string;
+  name: string;
+  family: SlotFamily;
+  body: SlotBody;
+  /** Default candle granularity the slot evaluates on. */
+  granularity: SlotGranularity;
+  /** Short evidence sentence shown in UI / explain notes. */
+  explain: string;
+}
+
+/** Complete 34-slot confluence catalog in evaluation order. */
+export const CONFLUENCE_SLOTS: ConfluenceSlot[] = [
+  // ---------------------------------------------------------------- technical
+  { id: 'goldenCross', name: 'Golden Cross', family: 'technical', body: 'bull', granularity: '1wk', explain: 'The 50-window average has crossed above the 200-window average, a widely-watched trend-quality marker.' },
+  { id: 'deathCross', name: 'Death Cross', family: 'technical', body: 'exit', granularity: '1wk', explain: 'The 50-window average has crossed below the 200-window average, a commonly-cited trend-weakening marker.' },
+  { id: 'trendAlignment', name: 'Trend Alignment', family: 'technical', body: 'bull', granularity: '1wk', explain: 'Price sits above both the 50- and 200-window averages, indicating an aligned long-term trend.' },
+  { id: 'timeSeriesMomentum12_1', name: '12-1 Momentum', family: 'technical', body: 'bull', granularity: '1d', explain: 'The 12-month-return-minus-1-month gauge is positive, a period-validated trend-following signal.' },
+  { id: 'relativeStrengthVsSpy', name: 'Relative Strength vs SPY', family: 'technical', body: 'bull', granularity: '1d', explain: 'The symbol is outperforming the S&P 500 over the 12-1 momentum window.' },
+  { id: 'rsiOversold', name: 'RSI Oversold', family: 'technical', body: 'bull', granularity: '1d', explain: 'Momentum gauge oversold, a bounce-prone condition after sustained weakness.' },
+  { id: 'rsiOverbought', name: 'RSI Overbought', family: 'technical', body: 'exit', granularity: '1d', explain: 'Momentum gauge overbought, a stretched condition after sustained strength.' },
+  { id: 'macdBullish', name: 'MACD Bullish', family: 'technical', body: 'bull', granularity: '1d', explain: 'The MACD line sits above its signal line, indicating positive short-term momentum.' },
+  { id: 'macdBearish', name: 'MACD Bearish', family: 'technical', body: 'exit', granularity: '1d', explain: 'The MACD line sits below its signal line, indicating negative short-term momentum.' },
+  { id: 'relVolume', name: 'Relative Volume', family: 'technical', body: 'bull', granularity: '1d', explain: 'Current volume is notably above its recent average, increasing the confidence contribution of concurrent signals.' },
+  { id: 'volumeProfileShelf', name: 'Volume-Profile Shelf', family: 'technical', body: 'bull', granularity: '1d', explain: 'Price is trading near a high-volume shelf of the volume profile, historically a supported level.' },
+  { id: 'fibLevelCluster', name: 'Fib Level Cluster', family: 'technical', body: 'bull', granularity: '1d', explain: 'Price is near a Fibonacci level that confluences with other technical levels, a commonly-watched confluence.' },
+  { id: 'volatilityRegimeLow', name: 'Low-Volatility Regime', family: 'technical', body: 'bull', granularity: '1d', explain: 'Realized volatility is in a low percentile versus its own history, historically favorable to trend continuation.' },
+  { id: 'pullbackToEMA21', name: 'Pullback to EMA-21', family: 'technical', body: 'bull', granularity: '1d', explain: 'Price has pulled back toward a rising short-term average within a broader uptrend.' },
+  { id: 'momentumForward', name: 'Forward Momentum', family: 'technical', body: 'bull', granularity: '1d', explain: 'Short-horizon rate of change is positive on a daily basis.' },
+
+  // ------------------------------------------------------------- institutional
+  { id: 'instNetActivePositive', name: 'Active Funds Net Positive', family: 'institutional', body: 'bull', granularity: '1d', explain: 'Net institutional flows from active managers are positive in the latest observable quarter.' },
+  { id: 'insiderInformedBuy30d', name: 'Informed Insider Buys (30d)', family: 'institutional', body: 'bull', granularity: '1d', explain: 'Informed insiders have been net buyers over the trailing 30 days, the strongest insider-feedback class.' },
+  { id: 'insiderInformedSell', name: 'Informed Insider Selling', family: 'institutional', body: 'exit', granularity: '1d', explain: 'Informed insiders have been net sellers over the trailing window, a caution signal.' },
+  { id: 'new13da', name: 'New 13D/A Disclosure', family: 'institutional', body: 'bull', granularity: '1d', explain: 'A fresh Schedule 13D/A was disclosed, signaling an activist or concentrated position just formed.' },
+  { id: 'instBuyZoneProximity', name: 'Institutional Buy-Zone Proximity', family: 'institutional', body: 'bull', granularity: '1d', explain: 'Price is near the estimated institutional accumulation zone from the flow engine estimation model.' },
+
+  // -------------------------------------------------------------------- macro
+  { id: 'consumerSentimentLow', name: 'Consumer Sentiment Low', family: 'macro', body: 'bull', granularity: '1wk', explain: 'The consumer-sentiment gauge is in its low historical band, a contrarian backdrop historically followed by higher forward returns.' },
+  { id: 'consumerSentimentHigh', name: 'Consumer Sentiment High', family: 'macro', body: 'exit', granularity: '1wk', explain: 'The consumer-sentiment gauge is in its high historical band, a backdrop historically followed by lower forward returns.' },
+  { id: 'ratesRegime', name: 'Falling-Rates Regime', family: 'macro', body: 'bull', granularity: '1wk', explain: 'Interest-rate expectations are on a falling trajectory, historically supportive of equities.' },
+  { id: 'macroRegimeUp', name: 'Macro Regime Up', family: 'macro', body: 'bull', granularity: '1wk', explain: 'Leading macroeconomic indicators are on a rising trajectory.' },
+  { id: 'breadthThrust', name: 'Breadth Thrust', family: 'macro', body: 'bull', granularity: '1wk', explain: 'Market breadth has surged from oversold, a historically constructive regime marker.' },
+
+  // -------------------------------------------------------------- seasonal
+  { id: 'seasonalFavorableMonth', name: 'Seasonally Favorable Month', family: 'seasonal', body: 'bull', granularity: '1d', explain: 'The calendar month is historically favorable for the symbol, a mild cyclical tailwind.' },
+  { id: 'winterHalfOn', name: 'Winter Half Indicator', family: 'seasonal', body: 'bull', granularity: '1d', explain: 'The November-April half of the year is historically favorable, a mild cyclical tailwind.' },
+  { id: 'electionCycleFavorableYear', name: 'Election-Cycle Favorable Year', family: 'seasonal', body: 'bull', granularity: '1d', explain: 'The current year of the presidential cycle is historically above-average for equities.' },
+  { id: 'nearTurnOfMonth', name: 'Turn-of-Month', family: 'seasonal', body: 'bull', granularity: '1d', explain: 'The calendar is near the turn of the month, historically a benign pocket of liquidity.' },
+  { id: 'nearQuarterEnd', name: 'Near Quarter-End', family: 'seasonal', body: 'bull', granularity: '1d', explain: 'The calendar is near quarter-end, historically associated with institutional window dressing.' },
+
+  // -------------------------------------------------------------------- flows
+  { id: 'etfFlowPositive', name: 'ETF Inflows', family: 'flows', body: 'bull', granularity: '1d', explain: 'The symbol is seeing net positive ETF inflows, evidence of systematic buying pressure.' },
+  { id: 'cotPositioning', name: 'COT Positioning', family: 'flows', body: 'bull', granularity: '1wk', explain: 'CFTC futures positioning is in a supportive band, evidence of aligned speculative flows.' },
+  { id: 'rotationIncipient', name: 'Incipient Sector Rotation', family: 'flows', body: 'bull', granularity: '1d', explain: 'Cross-sectional relative-strength analysis flags the symbol as an emerging rotation target.' },
+
+  // ---------------------------------------------------------------- sentiment
+  { id: 'commentatorSentiment', name: 'Informed Commentator Sentiment', family: 'sentiment', body: 'bull', granularity: '1d', explain: 'Informed commentators tracked via the configured sentiment source are net-positive on the symbol in the measurement window.' },
+];
+
+/** Indexed by slot id for O(1) lookup. */
+export const confluenceslotById: ReadonlyMap<string, ConfluenceSlot> = new Map(
+  CONFLUENCE_SLOTS.map((s) => [s.id, s]),
+);
+
+/** All slot ids, stable evaluation order. */
+export const CONFLUENCE_SLOT_IDS: readonly string[] = CONFLUENCE_SLOTS.map((s) => s.id);
+
+/** Type guard: is `id` a known confluence slot? */
+export function isConfluenceSlot(id: string): id is string {
+  return confluenceslotById.has(id);
+}
+
+/** Slots grouped by family. */
+export function slotsByFamily(family: SlotFamily): ConfluenceSlot[] {
+  return CONFLUENCE_SLOTS.filter((s) => s.family === family);
+}
+
+/** DSL helper removed — catalog written literally above for full control. */

app/server/src/confluence/sentimentSource.ts

@@ -0,0 +1,55 @@
+// Investor Flow — SentimentSource interface (M22, slice 2)
+//
+// Abstraction over "informed commentator sentiment". Phase 1 wires tweet
+// sentiment from the existing X cookie adapter; later phases can swap in other
+// sources (X radar, curated news sentiment) behind the same interface.
+//
+// The rack consumes this so the `commentatorSentiment` slot is evaluable without
+// coupling the confluence engine to any specific social provider. ADR-0007 / ADR-0009:
+// sentiment here is an evidence input (with sample size + window), sourced only
+// through rate-limited adapters — never a recommendation.
+
+/**
+ * A windowed, sample-sized sentiment measurement for one symbol.
+ * `netSentiment` is in [-1, 1]: +1 = fully bullish, -1 = fully bearish.
+ * `sampleSize` and `collectedAt` keep the measurement honest: thin windows are
+ * low-authority evidence and the rack can discount them.
+ */
+export interface CommentatorSentiment {
+  symbol: string;
+  /** Trading days the measurement window spans. */
+  windowDays: number;
+  /** Number of individual comments/tweets sampled in the window. */
+  sampleSize: number;
+  /** Bullish fraction minus bearish fraction in [-1, 1]. */
+  netSentiment: number;
+  bullCount: number;
+  bearCount: number;
+  neutralCount: number;
+  /** ISO timestamp the window closed. */
+  collectedAt: string;
+}
+
+/**
+ * Any provider that can produce a sentiment measurement for a symbol.
+ * Returning `null` means the source has no usable data (not an error).
+ */
+export interface SentimentSource {
+  readonly kind: 'commentatorSentiment';
+  /** Minimum sample size before the rack treats a measurement as evidence. */
+  readonly minSampleSize: number;
+  /** Fetch the latest windowed measurement for `symbol`. Safe to call often; implementers throttle. */
+  fetch(symbol: string, opts?: { windowDays?: number }): Promise<CommentatorSentiment | null>;
+}
+
+/** Convenience: does a measurement carry enough evidence to be admitted? */
+export function sentimentIsUsable(src: SentimentSource, s: CommentatorSentiment | null): s is CommentatorSentiment {
+  return s !== null && s.sampleSize >= src.minSampleSize && s.netSentiment !== 0;
+}
+
+/** Directional read of a usable sentiment measurement. */
+export function sentimentDirection(s: CommentatorSentiment): 'bull' | 'bear' | 'neutral' {
+  if (s.netSentiment > 0) return 'bull';
+  if (s.netSentiment < 0) return 'bear';
+  return 'neutral';
+}

app/server/src/db/__tests__/confluenceRepository.test.ts

@@ -0,0 +1,136 @@
+// Investor Flow — confluenceRepository.test.ts
+// Storage round-trip tests for the confluence racks / evaluations / signal history.
+
+import { describe, it, beforeEach, test } from 'node:test';
+import assert from 'node:assert/strict';
+
+import { createDb, initSchema } from '../client.ts';
+import { ConfluenceRepository, rackFromSlots } from '../confluenceRepository.ts';
+import { evaluateRack, type SlotAssessment } from '../../confluence/confluenceRack.ts';
+
+const fired = (id: string): SlotAssessment => ({ id, state: 'fired' });
+const notFired = (id: string): SlotAssessment => ({ id, state: 'not-fired' });
+
+let repo: ConfluenceRepository;
+let db;
+
+beforeEach(() => {
+  const d = createDb({ path: ':memory:' });
+  initSchema(d);
+  repo = new ConfluenceRepository(d);
+  db = d;
+});
+
+describe('ConfluenceRepository (racks)', () => {
+  it('persists and retrieves a system rack with slot ids', () => {
+    const rack = rackFromSlots('core_trend', 'Core Trend', ['goldenCross', 'relVolume'], { isSystem: true });
+    repo.saveRack(rack);
+
+    const got = repo.getRack('core_trend');
+    assert.ok(got);
+    assert.equal(got.name, 'Core Trend');
+    assert.deepEqual(got.slotIds, ['goldenCross', 'relVolume']);
+    assert.equal(got.isSystem, true);
+    assert.equal(got.ownerId, null);
+  });
+
+  it('lists system racks distinctly from user racks', () => {
+    repo.saveRack(rackFromSlots('sys1', 'System A', ['goldenCross'], { isSystem: true }));
+    repo.saveRack(rackFromSlots('usr1', 'Mine', ['relVolume'], { ownerId: 'u1' }));
+
+    assert.equal(repo.listSystemRacks().length, 1);
+    assert.equal(repo.listUserRacks('u1').length, 1);
+    assert.equal(repo.listAllRacks().length, 2);
+  });
+
+  it('updates rack slots and deletes racks', () => {
+    repo.saveRack(rackFromSlots('r1', 'Rack 1', ['goldenCross'], { ownerId: 'u1' }));
+    repo.updateRackSlots('r1', ['goldenCross', 'relVolume', 'cotPositioning']);
+    assert.deepEqual(repo.getRack('r1')!.slotIds, ['goldenCross', 'relVolume', 'cotPositioning']);
+
+    repo.deleteRack('r1');
+    assert.equal(repo.getRack('r1'), null);
+  });
+
+  it('looks up racks by owner id + name', () => {
+    repo.saveRack(rackFromSlots('a1', 'Alpha', ['goldenCross'], { ownerId: 'u1' }));
+    assert.equal(repo.getRackByOwnerAndName('u1', 'Alpha')!.id, 'a1');
+    assert.equal(repo.getRackByOwnerAndName('u1', 'nope'), null);
+    assert.equal(repo.getRackByOwnerAndName(null, 'Alpha'), null);
+  });
+});
+
+describe('ConfluenceRepository (evaluations)', () => {
+  it('round-trips a full evaluation', () => {
+    const rack = rackFromSlots('rt', 'Rack T', ['goldenCross', 'relVolume'], { isSystem: true });
+    repo.saveRack(rack);
+
+    const ev = evaluateRack('PLTR', '2026-02-02', [fired('goldenCross'), fired('relVolume'), notFired('cotPositioning')]);
+    repo.saveEvaluation(ev, 'rt', 'ev-1');
+
+    const got = repo.getEvaluation('PLTR', '2026-02-02', 'rt');
+    assert.ok(got);
+    assert.equal(got.quality, ev.quality);
+    assert.equal(got.bullEvidence, 2); // two distinct groups, no redundancy overlap
+    assert.equal(got.assessments.length, 3);
+    assert.equal(got.assessments[0].state, 'fired');
+  });
+
+  it('returns latest evaluation per rack by as_of', () => {
+    const rack = rackFromSlots('rt', 'Rack T', ['goldenCross'], { isSystem: true });
+    repo.saveRack(rack);
+    repo.saveEvaluation(evaluateRack('PLTR', '2026-02-01', [fired('goldenCross')]), 'rt', 'e1');
+    repo.saveEvaluation(evaluateRack('PLTR', '2026-02-02', [notFired('goldenCross')]), 'rt', 'e2');
+
+    const latest = repo.latestEvaluationForRack('rt');
+    assert.equal(latest!.asOf, '2026-02-02');
+    assert.equal(repo.listEvaluationsForSymbol('PLTR').length, 2);
+  });
+
+  it('dedupes a (symbol, as_of, rack) evaluation through unique constraint', () => {
+    const rack = rackFromSlots('rt', 'Rack T', ['goldenCross'], { isSystem: true });
+    repo.saveRack(rack);
+    repo.saveEvaluation(evaluateRack('NVDA', '2026-02-01', [fired('goldenCross')]), 'rt', 'n1');
+    assert.throws(() => repo.saveEvaluation(evaluateRack('NVDA', '2026-02-01', [notFired('goldenCross')]), 'rt', 'n2'));
+  });
+});
+
+describe('ConfluenceRepository (signal history)', () => {
+  it('logs fired slot signals and lists them for a symbol', () => {
+    const rack = rackFromSlots('rt', 'Rack T', ['goldenCross', 'relVolume', 'deathCross'], { isSystem: true });
+    repo.saveRack(rack);
+
+    const ev = evaluateRack('PLTR', '2026-02-02', [fired('goldenCross'), fired('deathCross'), notFired('relVolume')]);
+    repo.saveEvaluation(ev, 'rt', 'ev-1');
+    repo.logSignalFires(ev, 'rt');
+
+    const signals = repo.listSignalsForSymbol('PLTR');
+    assert.equal(signals.length, 2); // goldenCross + deathCross fired, relVolume did not
+    assert.deepEqual(signals.map((s) => s.slotId).sort(), ['deathCross', 'goldenCross']);
+    assert.equal(signals[0].priceResolved, false);
+  });
+
+  it('resolves a pending signal with verdict', () => {
+    const rack = rackFromSlots('rt', 'Rack T', ['goldenCross'], { isSystem: true });
+    repo.saveRack(rack);
+    const ev = evaluateRack('PLTR', '2026-02-02', [fired('goldenCross')]);
+    repo.saveEvaluation(ev, 'rt', 'ev-1');
+    repo.logSignalFires(ev, 'rt');
+
+    const pending = repo.listPendingSignals();
+    assert.equal(pending.length, 1);
+    repo.resolveSignal(pending[0].id, true, 'real');
+
+    const resolved = repo.listSignalsForSymbol('PLTR');
+    assert.equal(resolved[0].priceResolved, true);
+    assert.equal(resolved[0].priceConfirmed, true);
+    assert.equal(resolved[0].verdict, 'real');
+  });
+});
+
+// Ensure freshDb-style sanity through the migration path isn't broken by our tables.
+test('confluence tables survive initSchema re-run (idempotent)', () => {
+  const d = createDb({ path: ':memory:' });
+  initSchema(d);
+  assert.doesNotThrow(() => initSchema(d));
+});

app/server/src/db/__tests__/schema.test.ts

@@ -19,6 +19,7 @@ const EXPECTED_TABLES = [
   'screener_filters', 'reports', 'alerts', 'trusted_accounts',
   'saved_posts', 'trade_key_levels',
   'llm_summaries', 'rotation_signals', 'rotation_rank_snapshots', 'llm_providers', 'llm_dispatch_audit',
+  'confluence_racks', 'confluence_evaluations', 'confluence_signal_history',
 ];
 
 test('schema creates all locked tables', () => {

app/server/src/db/alertSubscriptionRepository.ts

@@ -37,6 +37,7 @@ export const ALERT_TYPE_CATALOG: Array<{ type: string; label: string; descriptio
   { type: 'fund_13f', label: 'Tracked Fund 13F', description: 'A tracked fund filed a new 13F.' },
   { type: 'mirror_diff', label: 'Mirror Diff', description: 'The mirror target book changed materially.' },
   { type: 'vix_level', label: 'VIX Level', description: 'VIX moves into a new historical volatility band.' },
+  { type: 'confluence_change', label: 'Confluence Picture Change', description: 'A symbol\'s confluence picture improved or became more cautious.' },
 ];
 
 export interface AlertTypeToggle {

app/server/src/db/client.ts

@@ -212,6 +212,48 @@ function runMigrations(db: DatabaseSync): void {
       sent_at     TEXT
     )`,
     `CREATE INDEX IF NOT EXISTS idx_notification_outbox_pending ON notification_outbox(status, created_at)`,
+    // Confluence Signal Engine (M22) — racks, evaluations, signal history.
+    `CREATE TABLE IF NOT EXISTS confluence_racks (
+      id               TEXT PRIMARY KEY,
+      owner_id         TEXT,
+      name             TEXT NOT NULL,
+      slot_ids_json    TEXT NOT NULL,
+      description      TEXT,
+      is_system        INTEGER NOT NULL DEFAULT 0,
+      created_at       TEXT NOT NULL
+    )`,
+    `CREATE INDEX IF NOT EXISTS idx_confluence_racks_owner ON confluence_racks(owner_id)`,
+    `CREATE TABLE IF NOT EXISTS confluence_evaluations (
+      id               TEXT PRIMARY KEY,
+      symbol           TEXT NOT NULL,
+      as_of            TEXT NOT NULL,
+      rack_id          TEXT NOT NULL REFERENCES confluence_racks(id) ON DELETE CASCADE,
+      assessments_json TEXT NOT NULL,
+      bull_evidence    REAL NOT NULL,
+      bear_evidence    REAL NOT NULL,
+      bull_count       INTEGER NOT NULL,
+      bear_count       INTEGER NOT NULL,
+      assessed_count   INTEGER NOT NULL,
+      net_evidence     REAL NOT NULL,
+      total_evidence   REAL NOT NULL,
+      quality          TEXT NOT NULL,
+      created_at       TEXT NOT NULL,
+      UNIQUE(symbol, as_of, rack_id)
+    )`,
+    `CREATE INDEX IF NOT EXISTS idx_confluence_evals_symbol_asof ON confluence_evaluations(symbol, as_of DESC)`,
+    `CREATE TABLE IF NOT EXISTS confluence_signal_history (
+      id              TEXT PRIMARY KEY,
+      symbol          TEXT NOT NULL,
+      slot_id         TEXT NOT NULL,
+      rack_id         TEXT NOT NULL REFERENCES confluence_racks(id) ON DELETE CASCADE,
+      fired_at        TEXT NOT NULL,
+      quality_at_fire TEXT NOT NULL,
+      price_resolved  INTEGER NOT NULL DEFAULT 0,
+      price_confirmed INTEGER NOT NULL DEFAULT 0,
+      resolved_at     TEXT,
+      verdict         TEXT
+    )`,
+    `CREATE INDEX IF NOT EXISTS idx_confluence_signal_symbol_slot ON confluence_signal_history(symbol, slot_id)`,
   ];
   for (const sql of migrations) {
     try { db.exec(sql); } catch { /* column already exists */ }

app/server/src/db/confluenceRepository.ts

@@ -0,0 +1,362 @@
+// Investor Flow — Confluence Repository (M22, slice 3)
+//
+// Thin data-access layer over the three confluence tables:
+//   confluence_racks           — named rack definitions (system + user)
+//   confluence_evaluations     — per-symbol, per-as-of rack evaluations
+//   confluence_signal_history  — closed-loop slot-fire log w/ verdict resolution
+//
+// Everything is stored serialized (assessments_json / slot_ids_json) so the
+// redundancy-aware rack logic stays in confluenceRack.ts (pure) and this file
+// only handles durable storage. No business rules live here.
+
+import type { DatabaseSync } from 'node:sqlite';
+import { randomUUID } from 'node:crypto';
+import type { ConfluenceEvaluation, SlotAssessment } from '../confluence/confluenceRack.ts';
+import type { PictureQuality } from '../confluence/confluenceRack.ts';
+import type { ConfluenceSlot } from '../confluence/confluenceSlots.ts';
+
+// ---------------------------------------------------------------------------
+// Types
+// ---------------------------------------------------------------------------
+
+/** A named rack definition (system preset or user-created). */
+export interface ConfluenceRack {
+  id: string;
+  ownerId: string | null;
+  name: string;
+  slotIds: string[];
+  description: string | null;
+  isSystem: boolean;
+  createdAt: string;
+}
+
+/** A stored rack evaluation for one symbol on one as-of date. */
+export interface StoredConfluenceEvaluation {
+  symbol: string;
+  asOf: string;
+  rackId: string;
+  quality: PictureQuality;
+  bullEvidence: number;
+  bearEvidence: number;
+  bullCount: number;
+  bearCount: number;
+  assessedCount: number;
+  netEvidence: number;
+  totalEvidence: number;
+  assessments: SlotAssessment[];
+  createdAt: string;
+}
+
+/** A slot-fire logged from an evaluation, awaiting verdict resolution. */
+export interface ConfluenceSignalHistoryRow {
+  id: string;
+  symbol: string;
+  slotId: string;
+  rackId: string;
+  firedAt: string;
+  qualityAtFire: PictureQuality;
+  priceResolved: boolean;
+  priceConfirmed: boolean;
+  resolvedAt: string | null;
+  verdict: 'real' | 'false_alarm' | null;
+}
+
+// ---------------------------------------------------------------------------
+// Serialization helpers
+// ---------------------------------------------------------------------------
+
+const parseJsonArray = (raw: string): string[] => {
+  try {
+    const v = JSON.parse(raw) as unknown;
+    return Array.isArray(v) ? v.filter((x): x is string => typeof x === 'string') : [];
+  } catch {
+    return [];
+  }
+};
+
+// ---------------------------------------------------------------------------
+// Prepared statements (lazy, one per method)
+// ---------------------------------------------------------------------------
+
+function stmts(db: DatabaseSync) {
+  return {
+    // --- confluence_racks ---
+    insertRack: db.prepare(
+      `INSERT INTO confluence_racks (id, owner_id, name, slot_ids_json, description, is_system, created_at)
+       VALUES (?, ?, ?, ?, ?, ?, ?)`,
+    ),
+    selectRack: db.prepare(
+      `SELECT id, owner_id, name, slot_ids_json, description, is_system, created_at
+         FROM confluence_racks WHERE id = ?`,
+    ),
+    selectRackByOwnerAndName: db.prepare(
+      `SELECT id, owner_id, name, slot_ids_json, description, is_system, created_at
+         FROM confluence_racks WHERE owner_id = ? AND name = ?`,
+    ),
+    selectSystemRacks: db.prepare(
+      `SELECT id, owner_id, name, slot_ids_json, description, is_system, created_at
+         FROM confluence_racks WHERE is_system = 1 ORDER BY name`,
+    ),
+    selectUserRacks: db.prepare(
+      `SELECT id, owner_id, name, slot_ids_json, description, is_system, created_at
+         FROM confluence_racks WHERE owner_id = ? ORDER BY created_at DESC`,
+    ),
+    selectAllRacks: db.prepare(`SELECT id, owner_id, name, slot_ids_json, description, is_system, created_at FROM confluence_racks`),
+    updateRackSlots: db.prepare(`UPDATE confluence_racks SET slot_ids_json = ?, name = ?, description = ? WHERE id = ?`),
+    deleteRack: db.prepare(`DELETE FROM confluence_racks WHERE id = ?`),
+
+    // --- confluence_evaluations ---
+    insertEvaluation: db.prepare(
+      `INSERT INTO confluence_evaluations
+         (id, symbol, as_of, rack_id, assessments_json, bull_evidence, bear_evidence,
+          bull_count, bear_count, assessed_count, net_evidence, total_evidence, quality, created_at)
+       VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
+    ),
+    selectEvaluation: db.prepare(
+      `SELECT id, symbol, as_of, rack_id, assessments_json, bull_evidence, bear_evidence,
+              bull_count, bear_count, assessed_count, net_evidence, total_evidence, quality, created_at
+         FROM confluence_evaluations WHERE symbol = ? AND as_of = ? AND rack_id = ?`,
+    ),
+    selectEvaluationsForSymbol: db.prepare(
+      `SELECT id, symbol, as_of, rack_id, assessments_json, bull_evidence, bear_evidence,
+              bull_count, bear_count, assessed_count, net_evidence, total_evidence, quality, created_at
+         FROM confluence_evaluations WHERE symbol = ? ORDER BY as_of DESC`,
+    ),
+    latestEvaluationForRack: db.prepare(
+      `SELECT id, symbol, as_of, rack_id, assessments_json, bull_evidence, bear_evidence,
+              bull_count, bear_count, assessed_count, net_evidence, total_evidence, quality, created_at
+         FROM confluence_evaluations
+        WHERE rack_id = ?
+        ORDER BY as_of DESC, created_at DESC
+        LIMIT 1`,
+    ),
+
+    // --- confluence_signal_history ---
+    insertSignal: db.prepare(
+      `INSERT INTO confluence_signal_history
+         (id, symbol, slot_id, rack_id, fired_at, quality_at_fire, price_resolved, price_confirmed, resolved_at, verdict)
+       VALUES (?, ?, ?, ?, ?, ?, 0, 0, NULL, NULL)`,
+    ),
+    selectPendingSignals: db.prepare(
+      `SELECT id, symbol, slot_id, rack_id, fired_at, quality_at_fire, price_resolved, price_confirmed, resolved_at, verdict
+         FROM confluence_signal_history WHERE price_resolved = 0 ORDER BY fired_at`,
+    ),
+    selectAllSignals: db.prepare(
+      `SELECT id, symbol, slot_id, rack_id, fired_at, quality_at_fire, price_resolved, price_confirmed, resolved_at, verdict
+         FROM confluence_signal_history ORDER BY fired_at`,
+    ),
+    selectSignalsForSymbol: db.prepare(
+      `SELECT id, symbol, slot_id, rack_id, fired_at, quality_at_fire, price_resolved, price_confirmed, resolved_at, verdict
+         FROM confluence_signal_history WHERE symbol = ? ORDER BY fired_at DESC`,
+    ),
+    resolveSignal: db.prepare(
+      `UPDATE confluence_signal_history
+          SET price_resolved = 1, price_confirmed = ?, resolved_at = ?, verdict = ?
+        WHERE id = ?`,
+    ),
+  };
+}
+
+function mapRack(row: Record<string, unknown>): ConfluenceRack {
+  return {
+    id: row.id as string,
+    ownerId: (row.owner_id as string | null) ?? null,
+    name: row.name as string,
+    slotIds: parseJsonArray(row.slot_ids_json as string),
+    description: (row.description as string | null) ?? null,
+    isSystem: Number(row.is_system) === 1,
+    createdAt: row.created_at as string,
+  };
+}
+
+function mapEvaluation(row: Record<string, unknown>): StoredConfluenceEvaluation {
+  const assessments = (() => {
+    try {
+      const v = JSON.parse(row.assessments_json as string) as unknown;
+      return Array.isArray(v) ? (v as SlotAssessment[]) : [];
+    } catch {
+      return [];
+    }
+  })();
+  return {
+    symbol: row.symbol as string,
+    asOf: row.as_of as string,
+    rackId: row.rack_id as string,
+    quality: row.quality as PictureQuality,
+    bullEvidence: Number(row.bull_evidence),
+    bearEvidence: Number(row.bear_evidence),
+    bullCount: Number(row.bull_count),
+    bearCount: Number(row.bear_count),
+    assessedCount: Number(row.assessed_count),
+    netEvidence: Number(row.net_evidence),
+    totalEvidence: Number(row.total_evidence),
+    assessments,
+    createdAt: row.created_at as string,
+  };
+}
+
+function mapSignal(row: Record<string, unknown>): ConfluenceSignalHistoryRow {
+  return {
+    id: row.id as string,
+    symbol: row.symbol as string,
+    slotId: row.slot_id as string,
+    rackId: row.rack_id as string,
+    firedAt: row.fired_at as string,
+    qualityAtFire: row.quality_at_fire as PictureQuality,
+    priceResolved: Number(row.price_resolved) === 1,
+    priceConfirmed: Number(row.price_confirmed) === 1,
+    resolvedAt: (row.resolved_at as string | null) ?? null,
+    verdict: (row.verdict as 'real' | 'false_alarm' | null) ?? null,
+  };
+}
+
+// ---------------------------------------------------------------------------
+// Repository
+// ---------------------------------------------------------------------------
+
+export class ConfluenceRepository {
+  private readonly db: DatabaseSync;
+
+  constructor(db: DatabaseSync) {
+    this.db = db;
+  }
+
+  // --------------------------------------------------------------- racks
+
+  saveRack(rack: ConfluenceRack): void {
+    stmts(this.db).insertRack.run(
+      rack.id,
+      rack.ownerId,
+      rack.name,
+      JSON.stringify(rack.slotIds),
+      rack.description,
+      rack.isSystem ? 1 : 0,
+      rack.createdAt,
+    );
+  }
+
+  getRack(id: string): ConfluenceRack | null {
+    const row = stmts(this.db).selectRack.get(id) as Record<string, unknown> | undefined;
+    return row ? mapRack(row) : null;
+  }
+
+  getRackByOwnerAndName(ownerId: string | null, name: string): ConfluenceRack | null {
+    const row = stmts(this.db).selectRackByOwnerAndName.get(ownerId, name) as Record<string, unknown> | undefined;
+    return row ? mapRack(row) : null;
+  }
+
+  listSystemRacks(): ConfluenceRack[] {
+    return (stmts(this.db).selectSystemRacks.all() as Record<string, unknown>[]).map(mapRack);
+  }
+
+  listUserRacks(ownerId: string): ConfluenceRack[] {
+    return (stmts(this.db).selectUserRacks.all(ownerId) as Record<string, unknown>[]).map(mapRack);
+  }
+
+  listAllRacks(): ConfluenceRack[] {
+    return (stmts(this.db).selectAllRacks.all() as Record<string, unknown>[]).map(mapRack);
+  }
+
+  updateRackSlots(id: string, slotIds: string[], name?: string, description?: string | null): void {
+    const existing = this.getRack(id);
+    if (!existing) throw new Error(`confluence rack not found: ${id}`);
+    stmts(this.db).updateRackSlots.run(
+      JSON.stringify(slotIds),
+      name ?? existing.name,
+      description === undefined ? existing.description : description,
+      id,
+    );
+  }
+
+  deleteRack(id: string): void {
+    stmts(this.db).deleteRack.run(id);
+  }
+
+  // ---------------------------------------------------------- evaluations
+
+  saveEvaluation(ev: ConfluenceEvaluation, rackId: string, id: string): void {
+    stmts(this.db).insertEvaluation.run(
+      id,
+      ev.symbol,
+      ev.asOf,
+      rackId,
+      JSON.stringify(ev.assessments),
+      ev.bullEvidence,
+      ev.bearEvidence,
+      ev.bullCount,
+      ev.bearCount,
+      ev.assessedCount,
+      ev.netEvidence,
+      ev.totalEvidence,
+      ev.quality,
+      new Date().toISOString(),
+    );
+  }
+
+  getEvaluation(symbol: string, asOf: string, rackId: string): StoredConfluenceEvaluation | null {
+    const row = stmts(this.db).selectEvaluation.get(symbol, asOf, rackId) as Record<string, unknown> | undefined;
+    return row ? mapEvaluation(row) : null;
+  }
+
+  listEvaluationsForSymbol(symbol: string): StoredConfluenceEvaluation[] {
+    return (stmts(this.db).selectEvaluationsForSymbol.all(symbol) as Record<string, unknown>[]).map(mapEvaluation);
+  }
+
+  latestEvaluationForRack(rackId: string): StoredConfluenceEvaluation | null {
+    const row = stmts(this.db).latestEvaluationForRack.get(rackId) as Record<string, unknown> | undefined;
+    return row ? mapEvaluation(row) : null;
+  }
+
+  // --------------------------------------------------------------- signals
+
+  logSignalFires(ev: ConfluenceEvaluation, rackId: string): void {
+    const insert = stmts(this.db).insertSignal;
+    for (const a of ev.assessments) {
+      if (a.state !== 'fired') continue;
+      insert.run(
+        randomUUID(),
+        ev.symbol,
+        a.id,
+        rackId,
+        ev.asOf,
+        ev.quality,
+      );
+    }
+  }
+
+  listPendingSignals(): ConfluenceSignalHistoryRow[] {
+    return (stmts(this.db).selectPendingSignals.all() as Record<string, unknown>[]).map(mapSignal);
+  }
+
+  listAllSignals(): ConfluenceSignalHistoryRow[] {
+    return (stmts(this.db).selectAllSignals.all() as Record<string, unknown>[]).map(mapSignal);
+  }
+
+  listSignalsForSymbol(symbol: string): ConfluenceSignalHistoryRow[] {
+    return (stmts(this.db).selectSignalsForSymbol.all(symbol) as Record<string, unknown>[]).map(mapSignal);
+  }
+
+  resolveSignal(id: string, confirmed: boolean, verdict: 'real' | 'false_alarm'): void {
+    stmts(this.db).resolveSignal.run(confirmed ? 1 : 0, new Date().toISOString(), verdict, id);
+  }
+}
+
+/** Build a rack from the catalog subset; validates slot ids exist. */
+export function rackFromSlots(
+  id: string,
+  name: string,
+  slotIds: string[],
+  opts: { ownerId?: string | null; description?: string | null; isSystem?: boolean; createdAt?: string } = {},
+): ConfluenceRack {
+  return {
+    id,
+    ownerId: opts.ownerId ?? null,
+    name,
+    slotIds: [...slotIds],
+    description: opts.description ?? null,
+    isSystem: opts.isSystem ?? false,
+    createdAt: opts.createdAt ?? new Date().toISOString(),
+  };
+}
+
+export type { ConfluenceEvaluation, SlotAssessment, PictureQuality, ConfluenceSlot }; // re-export for consumers

app/server/src/db/schema.sql

@@ -962,3 +962,58 @@ CREATE TABLE IF NOT EXISTS stock_float (
   UNIQUE(symbol, as_of)
 );
 CREATE INDEX IF NOT EXISTS idx_stock_float_symbol ON stock_float(symbol);
+
+-- ===== Section 2.?? — Confluence Signal Engine (M22). =====
+-- Named rack definitions: a rack is a curated subset of the 34-slot catalog
+-- evaluated together for a symbol. System presets have owner_id NULL.
+CREATE TABLE IF NOT EXISTS confluence_racks (
+  id               TEXT PRIMARY KEY,             -- 'core_trend' | user-defined key
+  owner_id         TEXT,                          -- NULL for system presets
+  name             TEXT NOT NULL,
+  slot_ids_json    TEXT NOT NULL,                 -- JSON array of slot ids
+  description      TEXT,
+  is_system        INTEGER NOT NULL DEFAULT 0,
+  created_at       TEXT NOT NULL
+);
+CREATE INDEX IF NOT EXISTS idx_confluence_racks_owner ON confluence_racks(owner_id);
+
+-- Per-symbol, per-as-of rack evaluation: the redundancy-discounted evidence
+-- totals and the resulting PictureQuality label. assessments_json stores the
+-- full per-slot states so the picture can be re-rendered and diffed without
+-- recomputation.
+CREATE TABLE IF NOT EXISTS confluence_evaluations (
+  id               TEXT PRIMARY KEY,
+  symbol           TEXT NOT NULL,
+  as_of            TEXT NOT NULL,                 -- YYYY-MM-DD evaluation date
+  rack_id          TEXT NOT NULL REFERENCES confluence_racks(id) ON DELETE CASCADE,
+  assessments_json TEXT NOT NULL,                 -- SlotAssessment[] JSON
+  bull_evidence    REAL NOT NULL,
+  bear_evidence    REAL NOT NULL,
+  bull_count       INTEGER NOT NULL,
+  bear_count       INTEGER NOT NULL,
+  assessed_count   INTEGER NOT NULL,
+  net_evidence     REAL NOT NULL,
+  total_evidence   REAL NOT NULL,
+  quality          TEXT NOT NULL,                 -- PictureQuality label
+  created_at       TEXT NOT NULL,
+  UNIQUE(symbol, as_of, rack_id)
+);
+CREATE INDEX IF NOT EXISTS idx_confluence_evals_symbol_asof ON confluence_evaluations(symbol, as_of DESC);
+
+-- Closed-loop signal history: every slot fire (from an evaluation) is logged
+-- here and later resolved to 'real' | 'false_alarm' by N=4-week price
+-- follow-through, mirroring rotation_signals. The per-slot reliability
+-- scorecard derives from this table.
+CREATE TABLE IF NOT EXISTS confluence_signal_history (
+  id               TEXT PRIMARY KEY,
+  symbol           TEXT NOT NULL,
+  slot_id          TEXT NOT NULL,
+  rack_id          TEXT NOT NULL REFERENCES confluence_racks(id) ON DELETE CASCADE,
+  fired_at         TEXT NOT NULL,                 -- as-of date of the firing evaluation
+  quality_at_fire  TEXT NOT NULL,                 -- PictureQuality when it fired
+  price_resolved   INTEGER NOT NULL DEFAULT 0,    -- follow-through window elapsed
+  price_confirmed  INTEGER NOT NULL DEFAULT 0,    -- follow-through aligned with slot body
+  resolved_at      TEXT,
+  verdict          TEXT                           -- real|false_alarm
+);
+CREATE INDEX IF NOT EXISTS idx_confluence_signal_symbol_slot ON confluence_signal_history(symbol, slot_id);

app/server/src/index.ts

@@ -9,6 +9,7 @@ import { OptionsAdapter } from './adapters/OptionsAdapter.ts';
 import { NasdaqAdapter } from './adapters/NasdaqAdapter.ts';
 import { FinraBulkAdapter } from './adapters/FinraBulkAdapter.ts';
 import { FinraShortInterestAdapter } from './adapters/FinraShortInterestAdapter.ts';
+import { CotAdapter } from './adapters/CotAdapter.ts';
 import { SecFetchAdapter } from './adapters/SecFetchAdapter.ts';
 import { SecCompanyTickersAdapter } from './adapters/SecCompanyTickersAdapter.ts';
 import { SecLintAdapter } from './adapters/SecLintAdapter.ts';
@@ -40,6 +41,7 @@ const adapters = new Map<SourceKind, SourceFetch>([
   ['sec-tickers' as const, new SecCompanyTickersAdapter(database) as unknown as SourceFetch],
   ['sec-lint-holders' as const, new SecLintAdapter(() => database, 'sec-lint-holders') as unknown as SourceFetch],
   ['sec-lint-insiders' as const, new SecLintAdapter(() => database, 'sec-lint-insiders') as unknown as SourceFetch],
+  ['cot' as const, new CotAdapter() as unknown as SourceFetch],
 ]);
 // Load X credentials at startup and register XCookieAdapter if available.
 let xAdapter: XCookieAdapter | null = null;
@@ -160,6 +162,7 @@ import { rotationIncipientProducer, regimeShiftProducer } from './alerts/produce
 import { convictionUnlockProducer } from './alerts/producers/unlockProducer.ts';
 import { thesisBrokenProducer, thesisWeakeningProducer } from './alerts/producers/thesisProducer.ts';
 import { clusterBreachProducer, drawdownHaltProducer, asymmetryWarningProducer } from './alerts/producers/portfolioRiskProducer.ts';
+import { confluenceChangeProducer } from './alerts/producers/confluenceProducer.ts';
 registerProducer(informedBuyProducer);
 registerProducer(informedSellProducer);
 registerProducer(new13daProducer);
@@ -173,6 +176,7 @@ registerProducer(thesisWeakeningProducer);
 registerProducer(clusterBreachProducer);
 registerProducer(drawdownHaltProducer);
 registerProducer(asymmetryWarningProducer);
+registerProducer(confluenceChangeProducer);
 registerMirrorProducers(); // fund_capture / fund_13f / mirror_diff (lazy, offline-safe)
 
 // Batched alert check: run every 10 minutes for non-critical producers.

app/server/src/queue/sourceRatePolicy.ts

@@ -25,6 +25,7 @@ export const DEFAULT_SOURCE_MIN_INTERVAL_MS: Record<SourceKind, number> = {
   'sec-lint-insiders': 5_000,
   nasdaq: 2_000,
   fred: 1_500,
+  cot: 1_500,
   'finra-bulk': 5_000,
   'finra-si': 5_000,
 };

app/server/src/services/vendorGate.ts

@@ -166,6 +166,11 @@ function seedBuiltIns(): void {
     sourceKinds: ['fred', 'macro'],
     policy: { minIntervalMs: 500, maxInflight: 1, drainJobBudget: 1, hostPattern: 'stlouisfed\\.org' },
   });
+  registerVendorIntegration({
+    family: 'cftc',
+    sourceKinds: ['cot'],
+    policy: { minIntervalMs: 1500, maxInflight: 1, drainJobBudget: 1, hostPattern: 'cftc\\.gov' },
+  });
   registerVendorIntegration({
     family: 'finra',
     sourceKinds: ['finra-bulk', 'finra-si'],

app/server/src/trpc/router.ts

@@ -33,6 +33,11 @@ import { computeOwnershipPercentages, fetchAndPersistFloat } from '../services/s
 import { EdgarAdapter } from '../adapters/EdgarAdapter.ts';
 import { OptionsAdapter, parseOptionChainRows } from '../adapters/OptionsAdapter.ts';
 import type { OptionChainRow, OptionGreeks } from '../adapters/OptionsAdapter.ts';
+import { CONFLUENCE_SLOTS, slotsByFamily, type SlotFamily } from '../confluence/confluenceSlots.ts';
+import { CONFLUENCE_SLOT_IDS } from '../confluence/confluenceSlots.ts';
+import { ConfluenceRepository, rackFromSlots } from '../db/confluenceRepository.ts';
+import { runSlotBacktest, signalHistoryToStats, type ConfluenceFireEvent } from '../confluence/confluenceBacktest.ts';
+import { detectPictureChange } from '../confluence/confluenceRack.ts';
 
 // ---------------------------------------------------------------------------
 // X cookie credential helpers. Loads AES-256-GCM encrypted ct0/auth_token from
@@ -5374,6 +5379,127 @@ const mirrorRouter = router({
     }),
 });
 
+// ─── Confluence Signal Engine (M22) ─────────────────────────────────────────
+
+const confluenceRouter = router({
+  /** The read-only 34-slot catalog, grouped by family, with slot metadata. */
+  slots: publicProcedure.query(async () => {
+    return {
+      slots: CONFLUENCE_SLOTS.map((s) => ({
+        id: s.id,
+        name: s.name,
+        family: s.family,
+        body: s.body,
+        granularity: s.granularity,
+        explain: s.explain,
+      })),
+      slotIds: CONFLUENCE_SLOT_IDS,
+      families: ['technical', 'institutional', 'macro', 'seasonal', 'flows', 'sentiment'] as SlotFamily[],
+    };
+  }),
+
+  /** List rack definitions (system + the calling user's own). */
+  racks: publicProcedure.query(({ ctx }) => {
+    const repo = new ConfluenceRepository(ctx.db);
+    const system = repo.listSystemRacks();
+    const user = ctx.userId ? repo.listUserRacks(ctx.userId) : [];
+    return { system, user };
+  }),
+
+  /** Latest evaluation for a symbol+rack, with the picture + change hint. */
+  evaluation: publicProcedure
+    .input(z.object({
+      symbol: z.string().min(1).max(12),
+      rackId: z.string().min(1).optional(),
+      limit: z.number().int().min(1).max(60).optional(),
+    }))
+    .query(({ ctx, input }) => {
+      const repo = new ConfluenceRepository(ctx.db);
+      const symbol = input.symbol.toUpperCase();
+      const evals = repo.listEvaluationsForSymbol(symbol);
+      const rackId = input.rackId ?? (repo.listSystemRacks()[0]?.id ?? null);
+      const forRack = rackId ? evals.filter((e) => e.rackId === rackId) : evals;
+      const latest = forRack.length > 0 ? forRack[0] : null;
+
+      let change = null;
+      if (forRack.length >= 2) {
+        const prev = forRack[1];
+        change = detectPictureChange(
+          { symbol, asOf: prev.asOf, assessments: prev.assessments, bullEvidence: prev.bullEvidence, bearEvidence: prev.bearEvidence, bullCount: prev.bullCount, bearCount: prev.bearCount, assessedCount: prev.assessedCount, netEvidence: prev.netEvidence, totalEvidence: prev.totalEvidence, quality: prev.quality },
+          { symbol, asOf: latest!.asOf, assessments: latest!.assessments, bullEvidence: latest!.bullEvidence, bearEvidence: latest!.bearEvidence, bullCount: latest!.bullCount, bearCount: latest!.bearCount, assessedCount: latest!.assessedCount, netEvidence: latest!.netEvidence, totalEvidence: latest!.totalEvidence, quality: latest!.quality },
+        );
+      }
+
+      return {
+        rackId,
+        symbol,
+        latest,
+        history: forRack.slice(0, input.limit ?? 10),
+        change,
+      };
+    }),
+
+  /** Run the slot backtest for a symbol from cached daily candles. */
+  backtest: publicProcedure
+    .input(z.object({
+      symbol: z.string().min(1).max(12),
+      weeks: z.number().int().min(2).max(12).optional(),
+    }))
+    .query(async ({ ctx, input }) => {
+      const repo = new ConfluenceRepository(ctx.db);
+      const symbol = input.symbol.toUpperCase();
+      const entry = await ctx.cache.get<PriceCandle[]>(`yfinance:candles:${symbol}:1d`);
+      const candles = (entry?.value ?? []) as PriceCandle[];
+
+      // Historic fire events come from signal history rows (as-of was the fire date).
+      const signals = repo.listSignalsForSymbol(symbol);
+      const fires: ConfluenceFireEvent[] = signals.map((s) => ({ symbol, asOf: s.firedAt, slotId: s.slotId }));
+      return runSlotBacktest(symbol, candles, fires, input.weeks ?? 4);
+    }),
+
+  /** Per-slot reliability scorecard from resolved signal history. */
+  scorecard: publicProcedure
+    .input(z.object({ symbol: z.string().min(1).max(12).optional() }))
+    .query(({ ctx, input }) => {
+      const symbol = input?.symbol?.toUpperCase();
+      const stats = signalHistoryToStats(ctx.db, symbol);
+      return {
+        slots: stats,
+        // Row semantics: clear, evidence-based.
+        note: 'Follow-through is measured over the standard 4-week horizon. Reliability is context, not a promise.',
+      };
+    }),
+
+  /** Create or update a user-owned rack. */
+  saveRack: protectedProcedure
+    .input(z.object({
+      id: z.string().min(1).max(80).optional(),
+      name: z.string().min(1).max(80),
+      description: z.string().max(300).optional(),
+      slotIds: z.array(z.string()).min(1).max(34),
+    }))
+    .mutation(({ ctx, input }) => {
+      const repo = new ConfluenceRepository(ctx.db);
+      const userId = ctx.userId!;
+      // Validate slot ids against the catalog.
+      for (const id of input.slotIds) {
+        if (!CONFLUENCE_SLOT_IDS.includes(id)) throw new TRPCError({ code: 'BAD_REQUEST', message: `Unknown slot id: ${id}` });
+      }
+      let rack = repo.getRack(input.id ?? '');
+      if (rack && rack.ownerId !== userId) throw new TRPCError({ code: 'FORBIDDEN', message: 'Not your rack.' });
+      if (rack) {
+        repo.updateRackSlots(rack.id, input.slotIds, input.name, input.description ?? null);
+        rack = repo.getRack(rack.id)!;
+      } else {
+        rack = rackFromSlots(input.id ?? `${userId}-${Date.now()}`, input.name, input.slotIds, {
+          ownerId: userId, description: input.description ?? null,
+        });
+        repo.saveRack(rack);
+      }
+      return rack;
+    }),
+});
+
 export const appRouter = router({
   auth: authRouter, onboarding: onboardingRouter, market: marketRouter, dashboard: dashboardRouter,
   admin: adminRouter, alerts: alertsRouter, edgar: edgarRouter, institutional: institutionalRouter,
@@ -5386,6 +5512,7 @@ export const appRouter = router({
   funds: fundsRouter,
   mirror: mirrorRouter,
   symbols: symbolsRouter,
+  confluence: confluenceRouter,
 });
 export type AppRouter = typeof appRouter;
 

app/src/lib/trpc.ts

@@ -1426,6 +1426,55 @@ export const api = {
     commentary: () => trpcQuery<{ shortTerm: string; longTerm: string; disclaimer: string }>("macro.commentary"),
     regimeClassify: () => trpcQuery<{ regime: string; confidence?: number; factors?: unknown }>("macro.regimeClassify"),
   },
+  confluence: {
+    slots: () => trpcQuery<{
+      slots: Array<{ id: string; name: string; family: string; body: string; granularity: string; explain: string }>;
+      slotIds: string[];
+      families: string[];
+    }>("confluence.slots", {}),
+    racks: () => trpcQuery<{
+      system: Array<{ id: string; ownerId: string | null; name: string; slotIds: string[]; description: string | null; isSystem: boolean; createdAt: string }>;
+      user: Array<{ id: string; ownerId: string | null; name: string; slotIds: string[]; description: string | null; isSystem: boolean; createdAt: string }>;
+    }>("confluence.racks", {}),
+    evaluation: (symbol: string, opts?: { rackId?: string; limit?: number }) =>
+      trpcQuery<{
+        rackId: string | null;
+        symbol: string;
+        latest: Record<string, unknown> | null;
+        history: Array<Record<string, unknown>>;
+        change: {
+          changed: boolean;
+          changeType: string;
+          current: string;
+          previous: string | null;
+          netEvidenceShift: number;
+          flippedSlotIds: string[];
+        } | null;
+      }>("confluence.evaluation", { symbol, ...(opts?.rackId ? { rackId: opts.rackId } : {}), ...(opts?.limit ? { limit: opts.limit } : {}) }),
+    backtest: (symbol: string, opts?: { weeks?: number }) =>
+      trpcQuery<{
+        symbol: string;
+        fires: number;
+        horizonDays: number;
+        slots: Array<{
+          slotId: string;
+          name: string;
+          body: string;
+          direction: string;
+          fires: number;
+          resolvedFires: number;
+          confirmedFires: number;
+          falseAlarms: number;
+          hitRate: number | null;
+          reliabilityWeight: number;
+        }>;
+        sampleSizeCaveat: string;
+      }>("confluence.backtest", { symbol, ...(opts?.weeks ? { weeks: opts.weeks } : {}) }),
+    scorecard: (symbol?: string) =>
+      trpcQuery<{ slots: unknown[]; note: string }>("confluence.scorecard", symbol ? { symbol } : {}),
+    saveRack: (input: { id?: string; name: string; description?: string; slotIds: string[] }) =>
+      trpcMutate<{ id: string; name: string; slotIds: string[]; isSystem: boolean; description: string | null }>("confluence.saveRack", input),
+  },
 };
 
 export interface TickerSuggestion {