feat: Unraid deploy, dealer-flow heatmap, confluence zones, 13F capture

Ship Node production images, Unraid compose, and Gitea CI/CD (test then
push registry images; cron script if no runner). Rebuild dealer flow as a
heatmap-first map with integrity gates and chart helpers. Add confluence
zone rules, session clock, capture evidence, and tighter 13F/queue/options
paths, plus the matching UI and tests.

app/server/src/cache/CacheRepository.ts

@@ -42,7 +42,11 @@ export interface SymbolMeta { symbol: string; name?: string | null; sector?: str
 export interface PriceAdjustment { symbol: string; exDate: string; type: "split" | "dividend"; ratio: number }
 
 /** Port CacheRepository depends on to schedule background refreshes. SourceAdapter/AdapterQueue satisfy this. */
-export interface CacheScheduler { queue(key: CacheKey): Promise<void>; }
+export interface CacheScheduler {
+  queue(key: CacheKey): Promise<void>;
+  /** Jump this key to the front of the next drain (header symbol / page view). */
+  prioritize?(key: CacheKey): Promise<void>;
+}
 
 export interface CacheEntry<T> { value: T | null; provenance: Provenance | null; isStale: boolean; }
 
@@ -89,7 +93,7 @@ interface KindHandler {
   /** Write value to cache, stamping provenance. */
   write(d: DatabaseSync, id: string, value: unknown, provenance: Provenance): void;
   /** Staleness verdict given the staleness timestamp (null = never cached) and now. */
-  isStale(stalenessTs: string | null, now: number): boolean;
+  isStale(stalenessTs: string | null, now: number, id?: string): boolean;
 }
 
 function tsAgeMs(ts: string | null, now: number): number {
@@ -155,16 +159,27 @@ const candlesHandler: KindHandler = {
     const rows = d.prepare('SELECT ts,o,h,l,c,v,adj_close,observed_at FROM price_candles WHERE symbol=? AND timeframe=? ORDER BY ts ASC').all(symbol, timeframe) as Array<Record<string, unknown>>;
     if (!rows.length) return null;
     const value: PriceCandle[] = rows.map((r) => ({ ts: r.ts as string, o: r.o as number, h: r.h as number, l: r.l as number, c: r.c as number, v: r.v as number, adjClose: r.adj_close as number | null }));
-    // Staleness for SWR uses last *bar* ts (not observed_at) so schedule + get agree on lag.
-    return { value, stalenessTs: rows[rows.length - 1].ts as string };
+    const last = rows[rows.length - 1]!;
+    // Minute TFs: freshness is last fetch (observed_at). Daily: last bar ts so schedule + get agree.
+    const tf = timeframe ?? '';
+    const stalenessTs = (tf === '1m' || tf === '5m')
+      ? (last.observed_at as string)
+      : (last.ts as string);
+    return { value, stalenessTs };
   },
   write(d, id, value, provenance) {
     const [symbol, timeframe] = id.split(':');
     const ins = d.prepare('INSERT OR REPLACE INTO price_candles (symbol,timeframe,ts,o,h,l,c,v,adj_close,observed_at) VALUES (?,?,?,?,?,?,?,?,?,?)');
     for (const c of value as PriceCandle[]) ins.run(symbol, timeframe, c.ts, c.o, c.h, c.l, c.c, c.v, c.adjClose ?? null, provenance.fetchedAt);
   },
-  // Do not thrash on every read: schedule owns EOD refresh. Stale only when absent.
-  isStale(ts) { return ts === null; },
+  // Daily: schedule owns refresh (stale only when absent). 1m/5m: short observed_at TTL.
+  isStale(ts, now, id) {
+    if (ts === null) return true;
+    const tf = id?.split(':')[1];
+    if (tf === '1m') return tsAgeMs(ts, now) > 60_000;
+    if (tf === '5m') return tsAgeMs(ts, now) > TTL_MS.intraday;
+    return false;
+  },
 };
 
 /** True when daily candles are missing or last bar is older than CANDLE_FRESH_MS. */
@@ -630,7 +645,7 @@ export interface CacheRepository {
   /** Permanent system pin (rotation universe, SPY, VIX) — survives unsubscribe. */
   pinSystemSymbol(symbol: string, tickerKind: TickerKind): Promise<void>;
   demandSet(): Promise<string[]>;
-  getMany<T>(keys: CacheKey[]): Promise<Array<{ key: CacheKey; value: T | null; isStale: boolean }>>;
+  getMany<T>(keys: CacheKey[]): Promise<Array<{ key: CacheKey; value: T | null; isStale: boolean; fetchedAt: string | null }>>;
   /** Delete a cache entry by key (or, for wildcard keys ending in `:*`, all matching entries). */
   del(key: CacheKey): Promise<void>;
   /** Underlying DB for schedule TTL checks (queue only). */
@@ -655,7 +670,7 @@ export class CacheRepositoryImpl implements CacheRepository {
     const h = this.handler(kind);
     const row = h.read(this._db, id);
     const now = Date.now();
-    let stale = h.isStale(row ? row.stalenessTs : null, now);
+    let stale = h.isStale(row ? row.stalenessTs : null, now, id);
     // Incomplete symbol meta (null name) is always treated as stale for SWR re-fetch.
     if (kind === 'symbol' && row) {
       const meta = row.value as SymbolMeta;
@@ -692,7 +707,7 @@ export class CacheRepositoryImpl implements CacheRepository {
       const meta = row.value as SymbolMeta;
       if (!meta?.name) return true;
     }
-    return h.isStale(row ? row.stalenessTs : null, Date.now());
+    return h.isStale(row ? row.stalenessTs : null, Date.now(), id);
   }
 
   private ensureDemandRow(symbol: string, tickerKind: TickerKind): void {
@@ -702,10 +717,14 @@ export class CacheRepositoryImpl implements CacheRepository {
   }
 
   /** Queue only what is missing or past TTL — no refcount change. */
-  private async queueIfNeeded(symbol: string): Promise<void> {
+  private async queueIfNeeded(symbol: string, opts?: { prioritizeQuote?: boolean }): Promise<void> {
     const d = this._db;
     if (needsQuoteRefresh(d, symbol)) {
-      try { await this._scheduler.queue(`yfinance:quote:${symbol}`); } catch { /* ignore */ }
+      const key = `yfinance:quote:${symbol}`;
+      try {
+        if (opts?.prioritizeQuote && this._scheduler.prioritize) await this._scheduler.prioritize(key);
+        else await this._scheduler.queue(key);
+      } catch { /* ignore */ }
     }
     if (needsCandleRefresh(d, symbol)) {
       try { await this._scheduler.queue(`yfinance:candles:${symbol}:1d`); } catch { /* ignore */ }
@@ -754,7 +773,7 @@ export class CacheRepositoryImpl implements CacheRepository {
     this._db.prepare(
       "UPDATE symbol_demand SET tier = MIN(tier, 2), last_viewed_at = ?, in_demand = 1 WHERE symbol=?",
     ).run(now, symbol);
-    await this.queueIfNeeded(symbol);
+    await this.queueIfNeeded(symbol, { prioritizeQuote: true });
   }
 
   async pinSystemSymbol(symbol: string, tickerKind: TickerKind): Promise<void> {
@@ -778,10 +797,10 @@ export class CacheRepositoryImpl implements CacheRepository {
       'SELECT symbol FROM symbol_demand WHERE in_demand = 1 OR COALESCE(system_pin, 0) = 1 ORDER BY symbol',
     ).all() as Array<{ symbol: string }>).map((r) => r.symbol);
   }
-  async getMany<T>(keys: CacheKey[]): Promise<Array<{ key: CacheKey; value: T | null; isStale: boolean }>> {
+  async getMany<T>(keys: CacheKey[]): Promise<Array<{ key: CacheKey; value: T | null; isStale: boolean; fetchedAt: string | null }>> {
     return Promise.all(keys.map(async (key) => {
       const e = await this.get<T>(key);
-      return { key, value: e.value, isStale: e.isStale };
+      return { key, value: e.value, isStale: e.isStale, fetchedAt: e.provenance?.fetchedAt ?? null };
     }));
   }
   private clearKvCachePrefix(prefix: string): void {