fix: IREN/ASTS/IRE stuck-pending + corridor feature

Fix stuck adapter_queue jobs (ASTS/IRE/IREN stuck pending forever):

1. fetchSpec early-return paths (cooldown checks) now update job status
   to 'backoff' with last_attempt set and 30s backoff_until, instead of
   returning without any status change. Prevents jobs from being
   re-processed every drain cycle indefinitely.

2. Wrap adapter.fetchOne() in 30s Promise.race timeout. A hung HTTP
   request no longer blocks the entire per-source promise chain
   forever, preventing all subsequent jobs for that source.

Also includes corridor/confluence feature, tiered quote schedules,
cache improvements, and HoldingsBookView refinements.

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

@@ -0,0 +1,129 @@
+// Investor Flow — confluenceEngine.test.ts (M24 slice 5)
+// Integration test for the evaluation engine: runs the wired slot families over
+// fake cache candle streams, verifies rack evaluations + corridor snapshots +
+// signal history are persisted and idempotent per (symbol, asOf, rack).
+
+import { describe, it, beforeEach } from 'node:test';
+import assert from 'node:assert/strict';
+
+import { createDb, initSchema } from '../../db/client.ts';
+import { ConfluenceRepository } from '../../db/confluenceRepository.ts';
+import { CorridorRepository } from '../../db/corridorRepository.ts';
+import { createCacheRepository, type CacheRepository, type CacheEntry, type PriceCandle, type Quote } from '../../cache/CacheRepository.ts';
+import { FakeSourceAdapter } from '../../adapters/SourceAdapter.ts';
+import { AdapterQueue } from '../../queue/AdapterQueue.ts';
+import { seedConfluence, CONFLUENCE_UNIVERSE, BENCHMARK_SYMBOL } from '../confluenceSeed.ts';
+import { runConfluenceEvaluationCycle } from '../confluenceEngine.ts';
+
+// ---- fake cache that returns whatever we seeded ---------------------------------
+
+class FakeCache implements CacheRepository {
+  private readonly store = new Map<string, { value: unknown; stale: boolean }>();
+  setValue(key: string, value: unknown, stale = false): this { this.store.set(key, { value, stale }); return this; }
+  async get<T>(key: string): Promise<CacheEntry<T>> {
+    const e = this.store.get(key);
+    return { value: (e ? e.value : null) as T | null, provenance: null, isStale: e ? e.stale : true };
+  }
+  async set(): Promise<void> { throw new Error('not used'); }
+  stale(key: string): boolean { return !this.store.has(key) || this.store.get(key)!.stale; }
+  async subscribe(): Promise<void> {}
+  async unsubscribe(): Promise<void> {}
+  async ensureInDemand(): Promise<void> {}
+  async pinSystemSymbol(): Promise<void> {}
+  async demandSet(): Promise<string[]> { return []; }
+  async getMany<T>(): Promise<Array<{ key: string; value: T | null; isStale: boolean }>> { return []; }
+  async del(): Promise<void> {}
+  readonly db: never = undefined as never;
+}
+
+// ---- fixtures -------------------------------------------------------------
+
+/** Monotonic ramp up (bullish technicals) over `days` trading days. */
+function rampUp(days: number, start = 100, dailyPct = 0.0015): PriceCandle[] {
+  const candles: PriceCandle[] = [];
+  const base = Date.UTC(2020, 0, 1);
+  for (let i = 0; i < days; i++) {
+    const c = start * Math.pow(1 + dailyPct, i);
+    candles.push({
+      ts: new Date(base + i * 86400000).toISOString().slice(0, 10),
+      o: c * (1 - dailyPct / 2),
+      h: c * 1.003,
+      l: c * 0.997,
+      c,
+      v: 2_000_000,
+      adjClose: c,
+    });
+  }
+  return candles;
+}
+
+let db: ReturnType<typeof createDb>;
+let cache: FakeCache;
+let confluenceRepo: ConfluenceRepository;
+let corridorRepo: CorridorRepository;
+
+beforeEach(async () => {
+  db = createDb({ path: ':memory:' });
+  initSchema(db);
+  confluenceRepo = new ConfluenceRepository(db);
+  corridorRepo = new CorridorRepository(db);
+  cache = new FakeCache();
+  await seedConfluence(db, cache as unknown as CacheRepository);
+
+  // Seed candle streams for the whole universe + SPY benchmark.
+  for (const { symbol } of CONFLUENCE_UNIVERSE) {
+    cache.setValue(`yfinance:candles:${symbol}:1d`, rampUp(300));
+    cache.setValue(`yfinance:candles:${symbol}:1wk`, rampUp(80, 100, 0.01));
+    cache.setValue(`yfinance:quote:${symbol}`, { price: 130 } as Quote);
+    cache.setValue(`yfinance:dividendFundamentals:${symbol}`, { forwardPE: 25, trailingPE: 24, forwardEPS: 5.2, trailingEPS: 5.0, currentPrice: 130 });
+  }
+  cache.setValue(`yfinance:candles:SPY:1d`, rampUp(300, 400, 0.001));
+  cache.setValue(`yfinance:candles:SPY:1wk`, rampUp(80, 400, 0.01));
+  cache.setValue(`yfinance:quote:SPY`, { price: 440 } as Quote);
+  cache.setValue(`yfinance:dividendFundamentals:SPY`, { forwardPE: 22, trailingPE: 21.5, forwardEPS: 20, trailingEPS: 19.3, currentPrice: 440 });
+});
+
+describe('runConfluenceEvaluationCycle', () => {
+  it('persists an evaluation + corridor snapshots per symbol, for every system rack', async () => {
+    const summary = await runConfluenceEvaluationCycle(db, cache as unknown as CacheRepository);
+
+    // All 15 universe symbols evaluated (SPY benchmark not part of universe).
+    assert.equal(summary.symbolsEvaluated.length, CONFLUENCE_UNIVERSE.length);
+    assert.equal(summary.evaluationsStored, CONFLUENCE_UNIVERSE.length * 3, 'one eval per symbol per system rack');
+    assert.ok(summary.corridorSnapshots >= CONFLUENCE_UNIVERSE.length, 'symbol + SPY corridor snapshots');
+
+    // spot-check one symbol persisted its "full" rack evaluation
+    const pltr = CONFLUENCE_UNIVERSE.find((s) => s.symbol === 'PLTR')!;
+    const evals = confluenceRepo.listEvaluationsForSymbol(pltr.symbol);
+    assert.equal(evals.length, 3);
+    const validQuality = ['strong-bullish', 'moderate-bullish', 'weak-bullish', 'mixed', 'weak-bearish', 'moderate-bearish', 'strong-bearish', 'sparse'];
+    for (const ev of evals) {
+      assert.equal(ev.symbol, pltr.symbol);
+      assert.ok(validQuality.includes(ev.quality), `unexpected quality ${ev.quality}`);
+      assert.ok(ev.assessedCount >= 10, 'rack has meaningful coverage');
+    }
+
+    // corridor snapshot table populated for at least the spot-checked symbol
+    const snap = corridorRepo.latestSnapshot(pltr.symbol);
+    assert.ok(snap, 'corridor snapshot stored');
+    assert.ok(snap.corridor1yMedian !== null, 'corridor median computed');
+  });
+
+  it('is idempotent per (symbol, asOf, rack): a second run reuses and stores nothing new', async () => {
+    const first = await runConfluenceEvaluationCycle(db, cache as unknown as CacheRepository);
+    const second = await runConfluenceEvaluationCycle(db, cache as unknown as CacheRepository);
+
+    assert.equal(second.evaluationsStored, 0);
+    assert.equal(second.evaluationsReused, first.evaluationsStored);
+    const all = confluenceRepo.listEvaluationsForSymbol(CONFLUENCE_UNIVERSE[0].symbol);
+    assert.equal(all.length, 3);
+  });
+
+  it('skips symbols with no cached candles without failing the run', async () => {
+    cache.setValue(`yfinance:candles:${CONFLUENCE_UNIVERSE[0].symbol}:1d`, [], false);
+    const summary = await runConfluenceEvaluationCycle(db, cache as unknown as CacheRepository);
+    assert.equal(summary.symbolsEvaluated.length, CONFLUENCE_UNIVERSE.length - 1);
+    assert.equal(summary.symbolsSkipped.length, 1);
+    assert.equal(summary.symbolsSkipped[0].symbol, CONFLUENCE_UNIVERSE[0].symbol);
+  });
+});

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

@@ -19,8 +19,8 @@ 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);
+  it('defines slots across all six families in catalog order', () => {
+    assert.ok(CONFLUENCE_SLOTS.length >= 34);
     const families = new Set(CONFLUENCE_SLOTS.map((s) => s.family));
     assert.deepEqual([...families].sort(), ['flows', 'institutional', 'macro', 'seasonal', 'sentiment', 'technical']);
   });

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

@@ -15,7 +15,7 @@ import {
   BENCHMARK_SYMBOL,
   defineSystemRackPresets,
 } from '../confluenceSeed.ts';
-import { CONFLUENCE_SLOT_IDS } from '../confluenceSlots.ts';
+import { CONFLUENCE_SLOTS, CONFLUENCE_SLOT_IDS } from '../confluenceSlots.ts';
 
 let db: ReturnType<typeof createDb>;
 let cache: CacheRepository;
@@ -63,19 +63,22 @@ describe('defineSystemRackPresets', () => {
     }
   });
 
-  it('"Full Confluence" uses all34 slots', () => {
+  it('"Full Confluence" uses every slot in the catalog', () => {
     const full = defineSystemRackPresets().find((p) => p.id === 'confluence-full')!;
-    assert.equal(full.slotIds.length, 34);
+    assert.equal(full.slotIds.length, CONFLUENCE_SLOT_IDS.length);
   });
 
-  it('"Technical Momentum" has 15 slots', () => {
+  it('"Technical Momentum" has only the technical slots', () => {
     const tech = defineSystemRackPresets().find((p) => p.id === 'confluence-technical')!;
-    assert.equal(tech.slotIds.length, 15);
+    assert.equal(tech.slotIds.length, CONFLUENCE_SLOTS.filter((s) => s.family === 'technical').length);
   });
 
-  it('"Macro + Flows + Sentiment" has 14 slots', () => {
+  it('"Macro + Flows + Sentiment" has macro + seasonal + flows + sentiment slots', () => {
     const macro = defineSystemRackPresets().find((p) => p.id === 'confluence-macro-flows')!;
-    assert.equal(macro.slotIds.length, 14);
+    const expected = CONFLUENCE_SLOTS.filter((s) =>
+      ['macro', 'seasonal', 'flows', 'sentiment'].includes(s.family),
+    ).length;
+    assert.equal(macro.slotIds.length, expected);
   });
 });
 

app/server/src/confluence/confluenceEngine.ts

@@ -0,0 +1,188 @@
+// Investor Flow — Confluence Evaluation Engine (M24, slice 5)
+//
+// The missing backbone of the Confluence Signal Engine: the daily cycle that
+// runs every wired slot family's evaluator over a symbol's cached data and
+// persists the resulting rack evaluations + slot-fire history.
+//
+// Each symbol in the confluence universe is resolved once (daily candles,
+// weekly candles, SPY benchmark daily, seasonality snapshot, corridor
+// snapshots for the symbol and SPY), then every system rack's slot subset is
+// sliced out, run through `evaluateRack` (redundancy-aware), and stored via
+// `ConfluenceRepository`. Fired slots are logged to signal history for the
+// reliability scorecard, and pending fires are resolved against forward prices
+//
+// ADR-0007: this engine computes description ("the picture is moderate-bullish"
+// because X evidence) — it never emits buy/sell directives.
+//
+// ADR-0009: the engine ONLY reads the shared cache (via CacheCandleProvider and
+// corridorData.resolveCorridorSnapshot). Any refreshing is already queued by
+// the cache; the engine never touches a vendor directly.
+
+import type { DatabaseSync } from 'node:sqlite';
+import { randomUUID } from 'node:crypto';
+import type { CacheRepository } from '../cache/CacheRepository.ts';
+import { CONFLUENCE_SLOTS } from './confluenceSlots.ts';
+import { CacheCandleProvider, type CandleProvider } from './candleProvider.ts';
+import { evaluateTechnicalSlots } from './technicalEvaluator.ts';
+import { evaluateSeasonalSlots } from './seasonalEvaluator.ts';
+import { evaluateCorridorSlots } from './corridorEvaluator.ts';
+import { buildSeasonalitySnapshot } from '../analysis/seasonality.ts';
+import { resolveCorridorSnapshot } from './corridorData.ts';
+import { evaluateRack, type ConfluenceEvaluation, type SlotAssessment } from './confluenceRack.ts';
+import { ConfluenceRepository, type ConfluenceRack } from '../db/confluenceRepository.ts';
+import { resolveSignalHistory } from './confluenceBacktest.ts';
+import { CONFLUENCE_UNIVERSE, BENCHMARK_SYMBOL } from './confluenceSeed.ts';
+
+// ---------------------------------------------------------------------------
+// Unwired slots → honest fallback assessments
+// ---------------------------------------------------------------------------
+
+/**
+ * Emit a `not-fired` fallback for every catalog slot the wired evaluators did
+ * not already cover (currently: macro non-corridor, institutional, flows,
+ * sentiment). Keeps each rack's slot set fully covered so assessedCount stays
+ * meaningful, but the fallbacks contribute no evidence to the picture.
+ */
+function unwiredFallbacks(emitted: SlotAssessment[]): SlotAssessment[] {
+  const emittedIds = new Set(emitted.map((a) => a.id));
+  return CONFLUENCE_SLOTS
+    .filter((s) => !emittedIds.has(s.id))
+    .map((s) => ({ id: s.id, state: 'not-fired' as const, note: 'No evaluator wired for this slot yet — contributes no evidence to the picture.' }));
+}
+
+// ---------------------------------------------------------------------------
+// Types
+// ---------------------------------------------------------------------------
+
+export interface ConfluenceEngineRunSummary {
+  symbolsEvaluated: string[];
+  symbolsSkipped: Array<{ symbol: string; reason: string }>;
+  evaluationsStored: number;
+  evaluationsReused: number;
+  signalsLogged: number;
+  signalsResolved: number;
+  corridorSnapshots: number;
+}
+
+export interface RunOptions {
+  /** Symbols to evaluate (default: the confluence universe). */
+  symbols?: string[];
+  /** Racks to evaluate per symbol (default: system racks). */
+  racks?: ConfluenceRack[];
+  /** Force re-evaluation even when a same-asOf evaluation already exists. */
+  force?: boolean;
+  /** Evaluation date override (default: the daily resolution asOf). */
+  asOf?: string;
+}
+
+// ---------------------------------------------------------------------------
+// Engine
+// ---------------------------------------------------------------------------
+
+/**
+ * Run one full confluence evaluation cycle over the universe.
+ *
+ * Per symbol: resolve candles once, run every wired evaluator across the slot
+ * families, slice each rack's slots, evaluate the rack, persist the evaluation
+ * (+ signal fires when new), then resolve the pending signal history. Returns a
+ * summary for logging/tests. Idempotent per (symbol, asOf, rack).
+ */
+export async function runConfluenceEvaluationCycle(
+  db: DatabaseSync,
+  cache: CacheRepository,
+  opts: RunOptions = {},
+): Promise<ConfluenceEngineRunSummary> {
+  const repo = new ConfluenceRepository(db);
+  const provider: CandleProvider = new CacheCandleProvider(cache);
+
+  const symbols = (opts.symbols?.map((s) => s.toUpperCase()) ?? CONFLUENCE_UNIVERSE.map((s) => s.symbol));
+  const racks = opts.racks ?? repo.listSystemRacks();
+  if (racks.length === 0) throw new Error('confluence engine: no system racks seeded');
+
+  const summary: ConfluenceEngineRunSummary = {
+    symbolsEvaluated: [],
+    symbolsSkipped: [],
+    evaluationsStored: 0,
+    evaluationsReused: 0,
+    signalsLogged: 0,
+    signalsResolved: 0,
+    corridorSnapshots: 0,
+  };
+
+  // SPY corridor snapshot is shared across every symbol (market-level slots).
+  let spyCorridor = await resolveCorridorSnapshot(db, cache, BENCHMARK_SYMBOL, { date: opts.asOf });
+  if (spyCorridor) summary.corridorSnapshots += 1;
+
+  for (const symbol of symbols) {
+    const daily = await provider.resolve(symbol, '1d');
+    if (daily.candles.length === 0) {
+      summary.symbolsSkipped.push({ symbol, reason: 'no daily candles in cache yet' });
+      continue;
+    }
+
+    const weekly = await provider.resolve(symbol, '1wk');
+    const benchmarkDaily = await provider.resolve(BENCHMARK_SYMBOL, '1d');
+    const asOf = opts.asOf ?? daily.asOf;
+
+    // Corridor snapshot for the symbol (valuation corridor, entry/upside slots).
+    let corridor = await resolveCorridorSnapshot(db, cache, symbol, { date: asOf });
+    if (corridor) summary.corridorSnapshots += 1;
+    if (spyCorridor === null) spyCorridor = await resolveCorridorSnapshot(db, cache, BENCHMARK_SYMBOL, { date: asOf });
+
+    // ----- run every wired family evaluator -----
+    const wired: SlotAssessment[] = [
+      ...evaluateTechnicalSlots(symbol, daily.candles, {
+        weekly: weekly.candles,
+        benchmarkDaily: benchmarkDaily.candles,
+      }),
+      ...evaluateSeasonalSlots(buildSeasonalitySnapshot(symbol, daily.candles), asOf),
+      ...(corridor ? evaluateCorridorSlots(corridor, spyCorridor) : corridorUnavailableAssessments()),
+    ];
+    // Cover every catalog slot the wired evaluators left out (honest fallbacks).
+    const assessments = [...wired, ...unwiredFallbacks(wired)];
+
+    // ----- slice per rack, evaluate, persist -----
+    let symbolStored = 0;
+    let symbolReused = 0;
+    for (const rack of racks) {
+      const rackSlots = new Set(rack.slotIds);
+      const sliced = assessments.filter((a) => rackSlots.has(a.id));
+      if (sliced.length === 0) continue;
+
+      const existing = repo.getEvaluation(symbol, asOf, rack.id);
+      if (existing && !opts.force) {
+        symbolReused += 1;
+        continue;
+      }
+
+      const evaluation: ConfluenceEvaluation = evaluateRack(symbol, asOf, sliced);
+      const firesLogged = evaluation.assessments.filter((a) => a.state === 'fired').length;
+      repo.saveEvaluation(evaluation, rack.id, randomUUID());
+      repo.logSignalFires(evaluation, rack.id);
+      symbolStored += 1;
+      summary.signalsLogged += firesLogged;
+    }
+
+    summary.evaluationsStored += symbolStored;
+    summary.evaluationsReused += symbolReused;
+    summary.symbolsEvaluated.push(symbol);
+  }
+
+  // Resolve pending signal history against forward prices.
+  const resolution = await resolveSignalHistory(db, async (sym) => {
+    const res = await provider.resolve(sym.toUpperCase(), '1d');
+    return res.candles;
+  });
+  summary.signalsResolved = resolution.resolved.filter((r) => r.verdict !== 'deferred').length;
+
+  return summary;
+}
+
+/** Six corridor slots, all not-fired with a shared note when no snapshot exists. */
+function corridorUnavailableAssessments(): SlotAssessment[] {
+  return [
+    'corridorEntryCheap', 'corridorEntryStretched',
+    'corridorUpsideHigh', 'corridorUpsideLow',
+    'spyCorridorCheap', 'spyCorridorStretched',
+  ].map((id) => ({ id, state: 'not-fired', note: 'No valuation-corridor snapshot cached for this symbol yet.' }));
+}

app/server/src/confluence/confluenceLibrary.ts

@@ -35,6 +35,8 @@ export const REDUNDANCY_GROUPS: RedundancyGroup[] = [
   { family: 'institutional', slots: ['instNetActivePositive', 'insiderInformedBuy30d', 'new13da'] },
   { family: 'macro', slots: ['ratesRegime', 'macroRegimeUp'] },
   { family: 'macro', slots: ['consumerSentimentLow', 'breadthThrust'] },
+  { family: 'macro', slots: ['corridorEntryCheap', 'corridorUpsideHigh'] },
+  { family: 'macro', slots: ['corridorEntryStretched', 'corridorUpsideLow'] },
   { family: 'seasonal', slots: ['seasonalFavorableMonth', 'winterHalfOn', 'electionCycleFavorableYear'] },
   { family: 'flows', slots: ['etfFlowPositive', 'cotPositioning'] },
 ];

app/server/src/confluence/confluenceSeed.ts

@@ -66,31 +66,32 @@ function familySlots(...families: string[]): string[] {
 
 /**
  * Three curated system rack presets. Each is a different lens on the same
- * symbol data, expressed as a subset of the 34-slot catalog:
+ * symbol data, expressed as a subset of the catalog:
  *
- *   1. "Full Confluence" — all 34 slots (the default every-picture view)
- *   2. "Technical Momentum" — the15 technical slots only (price-action focus)
- *   3. "Macro + Flows + Sentiment" — macro 5 + seasonal 5 + flows 3 +
- *      sentiment 1 = 14 slots (the macro/structural lens)
+ *   1. "Full Confluence" — all slots (the default every-picture view)
+ *   2. "Technical Momentum" — the 15 technical slots only (price-action focus)
+ *   3. "Macro + Flows + Sentiment" — macro + seasonal + flows + sentiment
+ *      slots (the macro/structural lens, now including the valuation corridor)
  */
 export function defineSystemRackPresets(): RackPreset[] {
+  const techIds = familySlots('technical');
   return [
     {
       id: 'confluence-full',
       name: 'Full Confluence',
-      description: 'All 34 slots. The broadest evidence view of a symbol\'s picture.',
+      description: `All ${ALL_IDS.length} slots. The broadest evidence view of a symbol\'s picture.`,
       slotIds: [...ALL_IDS],
     },
     {
       id: 'confluence-technical',
       name: 'Technical Momentum',
-      description: 'The 15 technical slots: trend, momentum, mean-reversion, and volume.',
-      slotIds: familySlots('technical'),
+      description: `The ${techIds.length} technical slots: trend, momentum, mean-reversion, and volume.`,
+      slotIds: techIds,
     },
     {
       id: 'confluence-macro-flows',
       name: 'Macro + Flows + Sentiment',
-      description: 'Macro regime, seasonal calendar, ETF/COT flows, and informed-commentator sentiment (14 slots).',
+      description: 'Macro regime, valuation corridor, seasonal calendar, ETF/COT flows, and informed-commentator sentiment.',
       slotIds: familySlots('macro', 'seasonal', 'flows', 'sentiment'),
     },
   ];

app/server/src/confluence/confluenceSlots.ts

@@ -1,6 +1,6 @@
 // Investor Flow — Confluence Slot Catalog (M22, slice 2)
 //
-// The 34-slot confluence inventory for the Confluence Signal Engine. Each slot is
+// The 40-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.
 //
@@ -46,7 +46,7 @@ export interface ConfluenceSlot {
   explain: string;
 }
 
-/** Complete 34-slot confluence catalog in evaluation order. */
+/** Complete 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.' },
@@ -93,6 +93,14 @@ export const CONFLUENCE_SLOTS: ConfluenceSlot[] = [
 
   // ---------------------------------------------------------------- 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.' },
+
+  // ----------------------------------------------------------------- corridor
+  { id: 'corridorEntryCheap', name: 'Corridor: Entry Cheap', family: 'macro', body: 'bull', granularity: '1d', explain: 'The current P/E sits in the lower third of the 1-year observable valuation corridor, indicating a relatively attractive entry point versus the symbol\'s own history.' },
+  { id: 'corridorEntryStretched', name: 'Corridor: Entry Stretched', family: 'macro', body: 'exit', granularity: '1d', explain: 'The current P/E sits in the upper third of the 1-year observable valuation corridor, indicating a stretched valuation versus the symbol\'s own history.' },
+  { id: 'corridorUpsideHigh', name: 'Corridor: Upside High', family: 'macro', body: 'bull', granularity: '1d', explain: 'Applying the 1-year median observable multiple to forward earnings implies meaningful upside from the current price.' },
+  { id: 'corridorUpsideLow', name: 'Corridor: Upside Low', family: 'macro', body: 'exit', granularity: '1d', explain: 'Applying the 1-year median observable multiple to forward earnings implies meaningful downside from the current price.' },
+  { id: 'spyCorridorCheap', name: 'SPY Corridor: Cheap', family: 'macro', body: 'bull', granularity: '1wk', explain: 'The S&P 500 forward P/E sits below its 3-year median, a market-level valuation tailwind that improves the odds for broad equity exposure.' },
+  { id: 'spyCorridorStretched', name: 'SPY Corridor: Stretched', family: 'macro', body: 'exit', granularity: '1wk', explain: 'The S&P 500 forward P/E sits at or above its 3-year median, a market-level valuation headwind that tempers the broad-equity picture.' },
 ];
 
 /** Indexed by slot id for O(1) lookup. */

app/server/src/confluence/corridorData.ts

@@ -0,0 +1,279 @@
+// Investor Flow — Price Corridor data pipeline (M24, slice 2)
+//
+// The Corridor Method (as reverse-engineered from @alojoh's weekly "U.S. Tech
+// Coverage / Market Valuation" reports): a symbol's *observable multiple* range
+// over a lookback window defines a valuation corridor. The current P/E position
+// within that corridor signals entry timing (cheap near the low band, stretched
+// near the high band), and applying the corridor's median multiple to forward
+// EPS derives an implied fair value / upside.
+//
+// ADR-0007: this is a valuation-context seam, never a buy/sell directive. It
+// computes where price sits relative to its own historical valuation corridor.
+//
+// Pure where possible: `computeCorridor`, `percentileIndex`, `buildSnapshot`
+// are pure; the cache-backed `resolveCorridorSnapshot` is a thin shim over the
+// shared cache (quote + candles + dividend fundamentals) — no direct vendor
+// I/O here (ADR-0009: everything funnels through the cache / adapter queue).
+
+import type { DatabaseSync } from 'node:sqlite';
+import type { CacheRepository, PriceCandle, Quote } from '../cache/CacheRepository.ts';
+import type { CorridorSnapshot } from '../db/corridorRepository.ts';
+import { CorridorRepository } from '../db/corridorRepository.ts';
+
+// ---------------------------------------------------------------------------
+// Constants
+// ---------------------------------------------------------------------------
+
+/** Trading days in the 1-year observable window. */
+export const CORRIDOR_1Y_BARS = 252;
+/** Trading days in the 90-day observable window. */
+export const CORRIDOR_90D_BARS = 63;
+/** Trading days in the 3-year market-level window (S&P 500 context). */
+export const CORRIDOR_3Y_BARS = 756;
+
+/** Fraction of the 1y corridor below which the entry is "cheap". */
+export const ENTRY_CHEAP_PERCENTILE = 0.33;
+/** Fraction above which the entry is "stretched". */
+export const ENTRY_STRETCHED_PERCENTILE = 0.67;
+/** Implied upside (1y median reversion) above which the upside slot fires. */
+export const UPSIDE_HIGH_THRESHOLD = 0.15;
+/** Implied downside below which the downside slot fires. */
+export const UPSIDE_LOW_THRESHOLD = -0.10;
+/** Window (days) used by the corridor-method backtest grader. */
+export const BACKTEST_HORIZON_DAYS = 7;
+
+// ---------------------------------------------------------------------------
+// Pure helpers
+// ---------------------------------------------------------------------------
+
+/** Median of a numeric array (finite elements); null when empty. */
+export function median(values: number[]): number | null {
+  const finite = values.filter((v) => Number.isFinite(v)).sort((a, b) => a - b);
+  if (finite.length === 0) return null;
+  const mid = Math.floor(finite.length / 2);
+  return finite.length % 2 === 0 ? (finite[mid - 1] + finite[mid]) / 2 : finite[mid];
+}
+
+/**
+ * The fractional rank (0..1) of `value` within `series`: the fraction of
+ * `series` elements at or below `value`. Returns null when series is empty.
+ * Pure.
+ */
+export function percentileIndex(value: number, series: number[]): number | null {
+  const finite = series.filter((v) => Number.isFinite(v));
+  if (finite.length === 0) return null;
+  const below = finite.filter((v) => v <= value).length;
+  return below / finite.length;
+}
+
+/** A computed corridor window. Pure. */
+export interface CorridorWindow {
+  high: number | null;
+  low: number | null;
+  median: number | null;
+}
+
+/**
+ * Compute a P/E corridor window from a lookback slice of a P/E series.
+ * `series` is the full series (oldest → newest); `bars` is the window size.
+ * Pure.
+ */
+export function computeCorridorWindow(series: number[], bars: number): CorridorWindow {
+  const slice = series.length >= bars ? series.slice(series.length - bars) : series.slice();
+  const finite = slice.filter((v) => Number.isFinite(v));
+  if (finite.length === 0) return { high: null, low: null, median: null };
+  return {
+    high: Math.max(...finite),
+    low: Math.min(...finite),
+    median: median(finite),
+  };
+}
+
+/**
+ * Build a full corridor snapshot from a trailing P/E series and current prices.
+ * Pure. `peSeries` is the trailing P/E series (oldest → newest); the latest
+ * value is treated as the current P/E position.
+ */
+export function computeCorridor(
+  peSeries: number[],
+  currentPrice: number,
+  forwardEPS: number | null,
+  trailingEPS: number | null,
+  trailingPE: number | null,
+  forwardPE: number | null,
+): Omit<CorridorSnapshot, 'symbol' | 'snapshotDate' | 'dataSource' | 'createdAt'> {
+  const currentPE = peSeries.length > 0 ? peSeries[peSeries.length - 1] : trailingPE ?? NaN;
+
+  const w1y = computeCorridorWindow(peSeries, CORRIDOR_1Y_BARS);
+  const w90d = computeCorridorWindow(peSeries, CORRIDOR_90D_BARS);
+
+  const fairValue1y = forwardEPS !== null && forwardEPS > 0 && w1y.median !== null ? forwardEPS * w1y.median : null;
+  const fairValue90d = forwardEPS !== null && forwardEPS > 0 && w90d.median !== null ? forwardEPS * w90d.median : null;
+
+  return {
+    forwardPE,
+    trailingPE,
+    forwardEPS,
+    trailingEPS,
+    corridor1yHigh: w1y.high,
+    corridor1yLow: w1y.low,
+    corridor1yMedian: w1y.median,
+    corridor90dHigh: w90d.high,
+    corridor90dLow: w90d.low,
+    corridor90dMedian: w90d.median,
+    fairValue1y,
+    fairValue90d,
+    impliedUpside1y:
+      currentPrice > 0 && fairValue1y !== null ? fairValue1y / currentPrice - 1 : null,
+    impliedUpside90d:
+      currentPrice > 0 && fairValue90d !== null ? fairValue90d / currentPrice - 1 : null,
+    pePercentile1y:
+      Number.isFinite(currentPE) ? percentileIndex(currentPE, peSeries.slice(Math.max(0, peSeries.length - CORRIDOR_1Y_BARS))) : null,
+    pePercentile90d:
+      Number.isFinite(currentPE) ? percentileIndex(currentPE, peSeries.slice(Math.max(0, peSeries.length - CORRIDOR_90D_BARS))) : null,
+    currentPrice,
+  };
+}
+
+/** Derive a trailing P/E series from closes over `eps`. Pure. */
+export function trailingPeSeries(candles: PriceCandle[], eps: number): number[] {
+  if (!eps || eps <= 0) return [];
+  return candles
+    .map((c) => (Number.isFinite(c.c) && c.c > 0 ? c.c / eps : NaN))
+    .filter((v) => Number.isFinite(v));
+}
+
+// ---------------------------------------------------------------------------
+// Cache-backed resolution seam
+// ---------------------------------------------------------------------------
+
+export interface CorridorFundamentalsInput {
+  price: number | null;
+  forwardPE: number | null;
+  trailingPE: number | null;
+  forwardEPS: number | null;
+  trailingEPS: number | null;
+}
+
+/** Paper a quote + dividend-fundamentals cache row into a corridor input. Pure. */
+export function fundamentalsFrom(quote: Quote | null, div: unknown): CorridorFundamentalsInput {
+  const f = (div as Record<string, unknown> | null) ?? {};
+  const num = (v: unknown): number | null => {
+    if (v === null || v === undefined || typeof v === 'string' && v === '') return null;
+    const n = Number(v);
+    return Number.isFinite(n) ? n : null;
+  };
+  const forwardPE = num(f.forwardPE);
+  const trailingPE = num(f.trailingPE);
+  const forwardEPS = num(f.forwardEPS);
+  const trailingEPS = num(f.trailingEPS);
+  return {
+    price: quote?.price != null && Number.isFinite(quote.price) ? quote.price : null,
+    forwardPE,
+    trailingPE,
+    forwardEPS,
+    // Prefer an explicit trailing EPS; else fall back to price / trailingPE.
+    trailingEPS: trailingEPS ?? (trailingPE && trailingPE > 0 && quote?.price ? quote.price / trailingPE : null),
+  };
+}
+
+/**
+ * Resolve + store a fresh corridor snapshot for `symbol` from the shared cache.
+ * Uses the trailing P/E bootstrap: the trailing EPS drives the historical P/E
+ * series immediately; forward PE/EPS enrich it when available. Idempotent per
+ * (symbol, snapshotDate). Returns the stored snapshot.
+ */
+export async function resolveCorridorSnapshot(
+  db: DatabaseSync,
+  cache: CacheRepository,
+  symbol: string,
+  opts: { date?: string } = {},
+): Promise<CorridorSnapshot | null> {
+  const sym = symbol.toUpperCase();
+  const repo = new CorridorRepository(db);
+  const snapshotDate = opts.date ?? new Date().toISOString().slice(0, 10);
+
+  const quoteEntry = await cache.get<Quote>(`yfinance:quote:${sym}`);
+  const divEntry = await cache.get<Record<string, unknown>>(`yfinance:dividendFundamentals:${sym}`);
+  const candleEntry = await cache.get<PriceCandle[]>(`yfinance:candles:${sym}:1d`);
+
+  const candles = (candleEntry?.value ?? []) as PriceCandle[];
+  const f = fundamentalsFrom(quoteEntry?.value ?? null, divEntry?.value ?? null);
+
+  if (f.price === null && candles.length === 0) return null;
+  const price = f.price ?? (candles.length > 0 ? candles[candles.length - 1].c : NaN);
+
+  const peSeries = f.trailingEPS !== null && f.trailingEPS > 0
+    ? trailingPeSeries(candles, f.trailingEPS)
+    : [];
+
+  const computed = computeCorridor(
+    peSeries,
+    price,
+    f.forwardEPS,
+    f.trailingEPS,
+    f.trailingPE ?? (f.trailingEPS && f.trailingEPS > 0 && price > 0 ? price / f.trailingEPS : null),
+    f.forwardPE,
+  );
+
+  // If no trailing EPS produced a real P/E series, compute one from forward PE.
+  const snapshot: Omit<CorridorSnapshot, 'id'> = {
+    symbol: sym,
+    snapshotDate,
+    ...computed,
+    dataSource: f.forwardPE !== null ? 'yfinance' : 'bootstrap_trailing',
+    createdAt: new Date().toISOString(),
+  };
+  if (!Number.isFinite(snapshot.currentPrice) && candles.length > 0) {
+    snapshot.currentPrice = candles[candles.length - 1].c;
+  }
+
+  repo.saveSnapshot(snapshot);
+  return repo.latestSnapshot(sym);
+}
+
+/**
+ * Compute a price-only corridor as a fallback when no EPS is available
+ * (e.g. funds / unfamiliar tickers). Uses close-price percentiles instead of
+ * P/E percentiles to still give a "where is price vs its own range" read.
+ * Pure.
+ */
+export function priceOnlySnapshot(
+  candles: PriceCandle[],
+  currentPrice: number,
+  snapshotDate: string,
+  symbol: string,
+): Omit<CorridorSnapshot, 'id'> | null {
+  if (candles.length === 0 || !Number.isFinite(currentPrice)) return null;
+  const closes = candles.map((c) => c.c).filter((v) => Number.isFinite(v));
+  if (closes.length === 0) return null;
+
+  const w1y = computeCorridorWindow(closes, CORRIDOR_1Y_BARS);
+  const w90d = computeCorridorWindow(closes, CORRIDOR_90D_BARS);
+
+  return {
+    symbol,
+    snapshotDate,
+    currentPrice,
+    forwardPE: null,
+    trailingPE: null,
+    forwardEPS: null,
+    trailingEPS: null,
+    corridor1yHigh: w1y.high,
+    corridor1yLow: w1y.low,
+    corridor1yMedian: w1y.median,
+    corridor90dHigh: w90d.high,
+    corridor90dLow: w90d.low,
+    corridor90dMedian: w90d.median,
+    fairValue1y: null,
+    fairValue90d: null,
+    impliedUpside1y: null,
+    impliedUpside90d: null,
+    pePercentile1y:
+      Number.isFinite(currentPrice) ? percentileIndex(currentPrice, closes.slice(Math.max(0, closes.length - CORRIDOR_1Y_BARS))) : null,
+    pePercentile90d:
+      Number.isFinite(currentPrice) ? percentileIndex(currentPrice, closes.slice(Math.max(0, closes.length - CORRIDOR_90D_BARS))) : null,
+    dataSource: 'bootstrap_trailing',
+    createdAt: new Date().toISOString(),
+  };
+}

app/server/src/confluence/corridorEvaluator.ts

@@ -0,0 +1,155 @@
+// Investor Flow — Price Corridor slot evaluator (M24, slice 3)
+//
+// Pure, snapshot-based assessments for the six `macro` corridor confluence
+// slots. Consumes the valuation-corridor snapshot computed by corridorData.ts
+// and, for the market-level slots, the SPY snapshot. Every function is a pure
+// (snapshot) → SlotAssessment[] mapping with ADR-0007 evidence notes — never a
+// recommendation.
+//
+// Distinct from the candle-driven technical evaluators: corridor slots read the
+// observable-multiple corridor, so they carry valuation evidence over and above
+// price-action evidence in the rack's picture.
+
+import type { CorridorSnapshot } from '../db/corridorRepository.ts';
+import type { SlotAssessment } from './confluenceRack.ts';
+import {
+  ENTRY_CHEAP_PERCENTILE,
+  ENTRY_STRETCHED_PERCENTILE,
+  UPSIDE_HIGH_THRESHOLD,
+  UPSIDE_LOW_THRESHOLD,
+  CORRIDOR_3Y_BARS,
+  computeCorridorWindow,
+} from './corridorData.ts';
+
+/** 3-year median forward-P/E reference for the S&P 500 (from @alojoh's reports:
+ *  18.9x trough, 23.1x peak, ~20.5x median over the last three years). Used only
+ *  as the anchoring bench when a full 3y series is unavailable. */
+export const SPY_3Y_MEDIAN_REFERENCE = 20.5;
+
+// ---------------------------------------------------------------------------
+// Pure helpers
+// ---------------------------------------------------------------------------
+
+function assessment(id: string, state: 'fired' | 'not-fired', note: string): SlotAssessment {
+  return { id, state, note };
+}
+
+/**
+ * Build the six price-corridor slot assessments for a symbol's snapshot.
+ * `spySnapshot` supplies the market-level slots (SPY corridor vs its own 3y
+ * median). Pure.
+ */
+export function evaluateCorridorSlots(
+  snapshot: CorridorSnapshot,
+  spySnapshot: CorridorSnapshot | null,
+): SlotAssessment[] {
+  const out: SlotAssessment[] = [];
+
+  const pct1y = snapshot.pePercentile1y === null ? null : Number(snapshot.pePercentile1y);
+  const pct90d = snapshot.pePercentile90d === null ? null : Number(snapshot.pePercentile90d);
+
+  // ----- corridor: entry cheap / stretched (1-year observable multiple) -----
+  if (pct1y === null) {
+    out.push(assessment('corridorEntryCheap', 'not-fired', 'No 1-year valuation corridor available for this symbol yet.'));
+    out.push(assessment('corridorEntryStretched', 'not-fired', 'No 1-year valuation corridor available for this symbol yet.'));
+  } else {
+    if (pct1y < ENTRY_CHEAP_PERCENTILE) {
+      out.push(assessment(
+        'corridorEntryCheap',
+        'fired',
+        `Current P/E is in the lower ${(ENTRY_CHEAP_PERCENTILE * 100).toFixed(0)}% of its 1-year observable corridor (${(pct1y * 100).toFixed(0)}th percentile), indicating a relatively attractive entry point.`,
+      ));
+      out.push(assessment('corridorEntryStretched', 'not-fired', `Current P/E sits at ${(pct1y * 100).toFixed(0)}th percentile of its 1-year corridor.`));
+    } else if (pct1y > ENTRY_STRETCHED_PERCENTILE) {
+      out.push(assessment(
+        'corridorEntryStretched',
+        'fired',
+        `Current P/E is in the upper ${((1 - ENTRY_STRETCHED_PERCENTILE) * 100).toFixed(0)}% of its 1-year observable corridor (${(pct1y * 100).toFixed(0)}th percentile), indicating a stretched valuation vs its own history.`,
+      ));
+      out.push(assessment('corridorEntryCheap', 'not-fired', `Current P/E sits at ${(pct1y * 100).toFixed(0)}th percentile of its 1-year corridor.`));
+    } else {
+      out.push(assessment('corridorEntryCheap', 'not-fired', `Current P/E sits mid-corridor at ${(pct1y * 100).toFixed(0)}th percentile of its 1-year range.`));
+      out.push(assessment('corridorEntryStretched', 'not-fired', `Current P/E sits mid-corridor at ${(pct1y * 100).toFixed(0)}th percentile of its 1-year range.`));
+    }
+  }
+
+  // ----- corridor: implied upside / downside (median-multiple reversion) -----
+  const upside1y = snapshot.impliedUpside1y === null ? null : Number(snapshot.impliedUpside1y);
+  if (upside1y === null) {
+    out.push(assessment('corridorUpsideHigh', 'not-fired', 'No forward EPS / median-multiple fair value available to quantify implied upside.'));
+    out.push(assessment('corridorUpsideLow', 'not-fired', 'No forward EPS / median-multiple fair value available to quantify implied downside.'));
+  } else {
+    if (upside1y > UPSIDE_HIGH_THRESHOLD) {
+      out.push(assessment(
+        'corridorUpsideHigh',
+        'fired',
+        `Applying the 1-year median multiple to forward earnings implies ${(upside1y * 100).toFixed(1)}% upside from the current price.`,
+      ));
+      out.push(assessment('corridorUpsideLow', 'not-fired', `1-year median-multiple fair value is ${(upside1y * 100).toFixed(1)}% vs current price.`));
+    } else if (upside1y < UPSIDE_LOW_THRESHOLD) {
+      out.push(assessment(
+        'corridorUpsideLow',
+        'fired',
+        `Applying the 1-year median multiple to forward earnings implies ${(upside1y * 100).toFixed(1)}% downside from the current price.`,
+      ));
+      out.push(assessment('corridorUpsideHigh', 'not-fired', `1-year median-multiple fair value is ${(upside1y * 100).toFixed(1)}% vs current price.`));
+    } else {
+      out.push(assessment('corridorUpsideHigh', 'not-fired', `1-year median-multiple fair value implies ${(upside1y * 100).toFixed(1)}% vs current price.`));
+      out.push(assessment('corridorUpsideLow', 'not-fired', `1-year median-multiple fair value implies ${(upside1y * 100).toFixed(1)}% vs current price.`));
+    }
+  }
+
+  // ----- market level: SPY 3-year corridor (valuation tailwind / headwind) -----
+  out.push(...spyCorridorAssessments(spySnapshot));
+
+  return out;
+}
+
+/**
+ * Market-level SPY corridor slots: is the broad market cheap or stretched
+ * relative to its own 3-year forward-P/E corridor? Accepts either a stored SPY
+ * snapshot (preferred) or a raw close-price series fallback. Pure.
+ */
+export function spyCorridorAssessments(spySnapshot: CorridorSnapshot | null): SlotAssessment[] {
+  if (!spySnapshot) {
+    return [
+      assessment('spyCorridorCheap', 'not-fired', 'No SPY valuation-corridor snapshot available for the market-level context.'),
+      assessment('spyCorridorStretched', 'not-fired', 'No SPY valuation-corridor snapshot available for the market-level context.'),
+    ];
+  }
+
+  const pe = spySnapshot.forwardPE ?? spySnapshot.trailingPE ?? null;
+  const cheap = pe !== null && pe < SPY_3Y_MEDIAN_REFERENCE;
+  const peLabel = pe !== null ? pe.toFixed(1) : 'n/a';
+
+  if (cheap) {
+    return [
+      assessment(
+        'spyCorridorCheap',
+        'fired',
+        `S&P 500 forward P/E (${peLabel}x) is below the 3-year median (~${SPY_3Y_MEDIAN_REFERENCE}x), a market-level valuation tailwind.`,
+      ),
+      assessment('spyCorridorStretched', 'not-fired', `S&P 500 forward P/E (${peLabel}x) is below the 3-year median (~${SPY_3Y_MEDIAN_REFERENCE}x).`),
+    ];
+  }
+  return [
+    assessment('spyCorridorCheap', 'not-fired', `S&P 500 forward P/E (${peLabel}x) is not below the 3-year median (~${SPY_3Y_MEDIAN_REFERENCE}x).`),
+    assessment(
+      'spyCorridorStretched',
+      'fired',
+      `S&P 500 forward P/E (${peLabel}x) is at or above the 3-year median (~${SPY_3Y_MEDIAN_REFERENCE}x), a market-level valuation headwind.`,
+    ),
+  ];
+}
+
+/**
+ * 3-year corridor window over a raw close-price series (for the SPY chart and
+ * the market-level slot when only prices are cached). Pure.
+ */
+export function priceCorridor3y(candles: Array<{ c: number }>): { high: number | null; low: number | null; median: number | null } {
+  const closes = candles.map((c) => c.c).filter((v) => Number.isFinite(v));
+  return computeCorridorWindow(closes, CORRIDOR_3Y_BARS);
+}
+
+/** Convenience re-export so corridor consumers share a single percentile helper. */
+export { percentileIndex } from './corridorData.ts';