fix: scan overlap protection + snapshot routing + worker history tables

Four fixes to keep the dashboard and worker honest while running real money:

A) Snapshot routing fix (worker.py:_sync_to_supabase)
Top-level fields (status/portfolio/closedTrades/performance/eligibleStocks)
were being copied from markets[_CURRENT_MARKET_ID], which flipped
between BIST and US depending on which market the worker happened to
process last. Result: a BIST winning trade vanished from top-level
while the US tab "inherited" it; performance showed winRate=0 even
though IZENR returned +12.93%.

Pin the top-level view to a canonical primary market (BIST by default,
override via TOPLEVEL_MARKET_ID env var). Per-market detail is still
under markets.{id}. Add `topLevelMarketId` + keep `activeMarketId` for
back-compat, and surface `currentlyProcessingMarketId` for diagnostics.

C) Overlap protection (worker.py:_latest_scan_payload_for_market,
app.py:_load_scan_payload, app.py:/api/eligible)
When a fresh chunked scan starts, the work file (eligible=2, stage2=9)
has a newer timestamp than the previous completed final file
(eligible=15, stage2=95). Old sort key was timestamp-only, so the
in-progress scan would shadow the completed one for hours until the
new run finished — risking the worker trading on a partial eligible
list at the next market open.

Sort key now ranks `completed=True` above `completed=False` first,
timestamp only as tiebreaker. Staleness of completed scans is still
enforced separately via SCAN_MAX_AGE_HOURS so we don't pin to ancient
data.

B) Boot-scan staleness threshold raised 72h → 168h (7 days), driven by
BOOT_SCAN_MAX_AGE_HOURS env var. The 72h figure was the root cause of
the 17→20 May "scan gap" — nightly scans were getting skipped because
the previous completion was still within 72h, but the container had
been asleep during the actual 19:00 TR window. Nightly re-runs are
still enforced separately via SCAN_MAX_AGE_HOURS (default 20h) so
freshness isn't sacrificed.

D) Worker history tables (supabase/migrations/20260521_worker_history.sql
+ worker.py:_sync_worker_history_to_db)
snapshot.json is overwritten on every sync — fine for "what's
happening now", useless for "show me every trade in the last 3
months". Three new public-read tables (worker_trades, worker_signals,
worker_equity) give us durable, queryable per-row history. UPSERT via
PostgREST `Prefer: resolution=merge-duplicates` keyed on UNIQUE
indexes (market_id+source_id / market_id+date+symbol / market_id+date)
so re-uploads are idempotent. Service-role writes only; anon SELECT.

The DB sync is best-effort: failures log a warning but never block the
snapshot.json upload that the live dashboard depends on.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

app.py

@@ -339,8 +339,14 @@ def _daily_scan_scheduler():
                 if finished:
                     scan_dt = datetime.fromisoformat(str(finished).replace("Z", "+00:00"))
                     age_hours = (datetime.now(timezone(timedelta(hours=0))) - scan_dt).total_seconds() / 3600
-                    if age_hours > 72:
-                        print(f"[scheduler] {market_id} scan results stale ({age_hours:.0f}h old)")
+                    # Boot-scan staleness threshold: previously 72h (3 days),
+                    # which combined with HF free-tier sleep meant scans would
+                    # be skipped during normal nightly windows. Now driven by
+                    # BOOT_SCAN_MAX_AGE_HOURS (default 168h = 7 days). Nightly
+                    # re-runs are still enforced separately via SCAN_MAX_AGE_HOURS.
+                    boot_max_age = float(os.environ.get("BOOT_SCAN_MAX_AGE_HOURS", "168"))
+                    if age_hours > boot_max_age:
+                        print(f"[scheduler] {market_id} scan results stale ({age_hours:.0f}h > {boot_max_age:.0f}h)")
                         try:
                             from trading.telegram_notify import notify as _tgn
                             _tgn("scan_stale", market_id=market_id, age_hours=age_hours)
@@ -348,7 +354,7 @@ def _daily_scan_scheduler():
                             pass
                         return True
                     else:
-                        print(f"[scheduler] {market_id} scan results fresh ({age_hours:.0f}h old)")
+                        print(f"[scheduler] {market_id} scan results fresh ({age_hours:.0f}h old, <{boot_max_age:.0f}h)")
                         return False
                 else:
                     return True
@@ -465,6 +471,19 @@ def _scan_stage_counts(data: dict) -> dict:
 
 
 def _load_scan_payload(market_id: str) -> tuple[Optional[dict], str]:
+    """Pick the best scan payload to expose to consumers.
+
+    Overlap protection: a COMPLETED scan is preferred over an in-progress (work)
+    scan even when the work scan has a newer timestamp. Without this, a freshly
+    started chunked scan with only a handful of stage2 results would shadow the
+    previous fully-completed scan (e.g. eligible=2 instead of eligible=15)
+    until it finishes hours/days later.
+
+    Tiebreaker: when both candidates share the same completed flag, the newer
+    timestamp wins. A staleness check is applied separately in
+    `_scan_already_completed` (SCAN_MAX_AGE_HOURS) so we still re-run scans
+    whose completed snapshot has aged out.
+    """
     from trading.market_registry import get_scan_results_path
 
     work_file = get_scan_results_path(market_id, completed=False)
@@ -483,12 +502,16 @@ def _load_scan_payload(market_id: str) -> tuple[Optional[dict], str]:
         return None, "none"
     loaded.sort(
         key=lambda item: (
+            # 1) completed scans rank above in-progress scans
+            bool(item[1].get("completed")),
+            # 2) tiebreak by max known timestamp
             max(
                 _scan_timestamp(item[1].get("updated_at")),
                 _scan_timestamp(item[1].get("scan_finished")),
                 _scan_timestamp(item[1].get("chunk_paused_at")),
                 _scan_timestamp(item[1].get("scan_started")),
             ),
+            # 3) final source-label beats work-label when everything else ties
             1 if item[0] == "final" else 0,
         ),
         reverse=True,
@@ -2436,7 +2459,20 @@ def get_eligible_stocks(market: str = Query("bist")):
             _timestamp_of((work_data or {}).get("scan_started")),
         )
 
-        using_work_file = bool(work_data) and (not final_data or work_ts >= final_ts)
+        # Overlap protection: prefer a completed (final) scan over an
+        # in-progress (work) scan, even if the work scan is newer. Falls back
+        # to the work scan only when no completed scan exists, or when the
+        # work scan itself is already completed and newer.
+        final_completed = bool((final_data or {}).get("completed"))
+        work_completed = bool((work_data or {}).get("completed"))
+        if work_data and (
+            not final_data
+            or (work_completed and (work_ts >= final_ts or not final_completed))
+            or (not final_completed and not work_completed and work_ts >= final_ts)
+        ):
+            using_work_file = True
+        else:
+            using_work_file = False
         data = work_data if using_work_file else final_data
         if data is None:
             raise RuntimeError("Scan payload is empty")

supabase/migrations/20260521_worker_history.sql

@@ -0,0 +1,140 @@
+-- =====================================================================
+-- Worker history tables: durable per-row storage of every automated
+-- trade, signal, and equity point produced by the trading worker.
+--
+-- Why this exists:
+--   The dashboard already gets a denormalised snapshot from
+--   storage/v1/object/public/trading/snapshot.json. That blob is
+--   overwritten on every sync — fine for "what's happening now", useless
+--   for "show me every trade in the last 3 months" or "average win rate
+--   by symbol". These tables give us durable, queryable history.
+--
+-- Visibility:
+--   PUBLIC READ (anon SELECT), WRITE only via service_role. The dashboard
+--   can pull historical analytics without auth; the worker writes with
+--   its SUPABASE_SERVICE_ROLE_KEY.
+--
+-- Idempotency:
+--   - worker_trades: UNIQUE(market_id, source_id) — `source_id` is the
+--     row id from the worker's SQLite `trades` table, so re-uploading the
+--     same trade is a no-op via ON CONFLICT.
+--   - worker_signals: UNIQUE(market_id, date, symbol) — one signal per
+--     symbol per day.
+--   - worker_equity: UNIQUE(market_id, date) — one equity point per day.
+-- =====================================================================
+
+CREATE TABLE IF NOT EXISTS worker_trades (
+  id BIGSERIAL PRIMARY KEY,
+  market_id TEXT NOT NULL,
+  source_id INTEGER NOT NULL,
+  symbol TEXT NOT NULL,
+  side TEXT NOT NULL CHECK (side IN ('BUY', 'SELL')),
+  entry_date DATE,
+  entry_price NUMERIC(14, 4),
+  exit_date DATE,
+  exit_price NUMERIC(14, 4),
+  quantity INTEGER NOT NULL,
+  entry_commission NUMERIC(12, 4),
+  exit_commission NUMERIC(12, 4),
+  gross_pnl NUMERIC(14, 4),
+  net_pnl NUMERIC(14, 4),
+  return_pct NUMERIC(10, 4),
+  holding_days INTEGER,
+  signal_confidence NUMERIC(10, 4),
+  predicted_return NUMERIC(10, 4),
+  entry_reason TEXT,
+  exit_reason TEXT,
+  is_closed BOOLEAN NOT NULL DEFAULT FALSE,
+  created_at TIMESTAMPTZ DEFAULT NOW(),
+  updated_at TIMESTAMPTZ DEFAULT NOW()
+);
+
+CREATE UNIQUE INDEX IF NOT EXISTS idx_worker_trades_dedup
+  ON worker_trades(market_id, source_id);
+CREATE INDEX IF NOT EXISTS idx_worker_trades_market_entry
+  ON worker_trades(market_id, entry_date DESC);
+CREATE INDEX IF NOT EXISTS idx_worker_trades_symbol
+  ON worker_trades(symbol);
+CREATE INDEX IF NOT EXISTS idx_worker_trades_closed
+  ON worker_trades(market_id, is_closed, exit_date DESC);
+
+COMMENT ON TABLE worker_trades IS
+  'Per-row durable record of every automated worker trade (paper or live). '
+  'Snapshot.json is denormalised; this table is the source of truth for '
+  'historical analytics (win rate by symbol/date, PnL distribution, etc.).';
+
+-- ---------------------------------------------------------------------
+
+CREATE TABLE IF NOT EXISTS worker_signals (
+  id BIGSERIAL PRIMARY KEY,
+  market_id TEXT NOT NULL,
+  date DATE NOT NULL,
+  symbol TEXT NOT NULL,
+  signal TEXT NOT NULL,
+  confidence NUMERIC(10, 4),
+  predicted_return NUMERIC(10, 4),
+  ml_signal TEXT,
+  tech_signal TEXT,
+  action TEXT,
+  created_at TIMESTAMPTZ DEFAULT NOW()
+);
+
+CREATE UNIQUE INDEX IF NOT EXISTS idx_worker_signals_dedup
+  ON worker_signals(market_id, date, symbol);
+CREATE INDEX IF NOT EXISTS idx_worker_signals_market_date
+  ON worker_signals(market_id, date DESC);
+
+COMMENT ON TABLE worker_signals IS
+  'Per-day per-symbol signal output. One row per (market, date, symbol). '
+  'Used to compare ML predictions against realised outcomes over time.';
+
+-- ---------------------------------------------------------------------
+
+CREATE TABLE IF NOT EXISTS worker_equity (
+  id BIGSERIAL PRIMARY KEY,
+  market_id TEXT NOT NULL,
+  date DATE NOT NULL,
+  equity NUMERIC(14, 4) NOT NULL,
+  cash NUMERIC(14, 4) NOT NULL,
+  positions_count INTEGER NOT NULL DEFAULT 0,
+  unrealized_pnl NUMERIC(14, 4),
+  realized_pnl_today NUMERIC(14, 4),
+  created_at TIMESTAMPTZ DEFAULT NOW(),
+  updated_at TIMESTAMPTZ DEFAULT NOW()
+);
+
+CREATE UNIQUE INDEX IF NOT EXISTS idx_worker_equity_dedup
+  ON worker_equity(market_id, date);
+CREATE INDEX IF NOT EXISTS idx_worker_equity_market_date
+  ON worker_equity(market_id, date DESC);
+
+COMMENT ON TABLE worker_equity IS
+  'Daily equity-curve point per market. Append-only on new days, upsert '
+  'on intraday rewrites.';
+
+-- =====================================================================
+-- RLS: anonymous read, service-role write.
+-- =====================================================================
+
+ALTER TABLE worker_trades  ENABLE ROW LEVEL SECURITY;
+ALTER TABLE worker_signals ENABLE ROW LEVEL SECURITY;
+ALTER TABLE worker_equity  ENABLE ROW LEVEL SECURITY;
+
+DROP POLICY IF EXISTS worker_trades_anon_read  ON worker_trades;
+DROP POLICY IF EXISTS worker_signals_anon_read ON worker_signals;
+DROP POLICY IF EXISTS worker_equity_anon_read  ON worker_equity;
+
+CREATE POLICY worker_trades_anon_read
+  ON worker_trades FOR SELECT TO anon, authenticated
+  USING (TRUE);
+
+CREATE POLICY worker_signals_anon_read
+  ON worker_signals FOR SELECT TO anon, authenticated
+  USING (TRUE);
+
+CREATE POLICY worker_equity_anon_read
+  ON worker_equity FOR SELECT TO anon, authenticated
+  USING (TRUE);
+
+-- Service-role bypasses RLS automatically; no explicit insert/update
+-- policies needed for the worker. Anon/authenticated cannot mutate.

trading/worker.py

@@ -133,6 +133,17 @@ def _scan_payload_timestamp(data: Optional[Dict[str, Any]]) -> float:
 
 
 def _latest_scan_payload_for_market(market_id: str) -> Dict[str, Any]:
+    """Return the best scan payload for downstream consumers.
+
+    Selection policy (overlap protection): a COMPLETED scan is always preferred
+    over an in-progress (work) scan, even if the work scan has a newer
+    timestamp. This prevents the dashboard/worker from seeing a partial
+    eligible list (e.g. eligible=2 of 95) while a new chunked scan is mid-flight
+    and overwriting the prior eligible list (e.g. eligible=15).
+
+    Tiebreaker: if both candidates are completed (or both in-progress), pick the
+    one with the newer timestamp.
+    """
     work_path = get_scan_results_path(market_id, completed=False)
     final_path = get_scan_results_path(market_id, completed=True)
 
@@ -155,7 +166,12 @@ def _latest_scan_payload_for_market(market_id: str) -> Dict[str, Any]:
     if not candidates:
         return {}
 
-    candidates.sort(key=_scan_payload_timestamp, reverse=True)
+    # Sort: (completed=True ranks first, then newer timestamp wins as tiebreak).
+    # reverse=True so True > False and larger timestamp > smaller.
+    candidates.sort(
+        key=lambda d: (bool(d.get("completed")), _scan_payload_timestamp(d)),
+        reverse=True,
+    )
     return candidates[0]
 
 
@@ -469,10 +485,27 @@ def _sync_to_supabase() -> None:
         import urllib.error
 
         markets = {market_id: _build_market_snapshot(market_id) for market_id in _SUPPORTED_MARKET_IDS}
-        current_market = markets[_CURRENT_MARKET_ID]
-        state_data = _load_json_file(_state_file())
-        status_data = _load_json_file(_status_file())
-        scan_results = _load_json_file(_scan_results_file())
+
+        # ── Snapshot routing fix ──────────────────────────────────────────
+        # Previously the top-level fields (status/portfolio/closedTrades/
+        # performance/eligibleStocks) were copied from `markets[_CURRENT_MARKET_ID]`,
+        # which changed depending on which market the worker happened to be
+        # processing last. The result was a confusing dashboard where switching
+        # the active market silently swapped what "the headline numbers" meant
+        # — e.g. a BIST winning trade vanished from top-level while the US
+        # tab inherited it.
+        #
+        # We now pin the top-level view to a canonical primary market (BIST by
+        # default, overridable via TOPLEVEL_MARKET_ID env var). Per-market
+        # detail is still available under `markets.{id}`.
+        topl_id = os.environ.get("TOPLEVEL_MARKET_ID", "bist").strip().lower()
+        if topl_id not in markets:
+            topl_id = _CURRENT_MARKET_ID if _CURRENT_MARKET_ID in markets else next(iter(markets))
+        current_market = markets[topl_id]
+
+        state_data = _load_json_file(_state_file_for_market(topl_id))
+        status_data = _load_json_file(_status_file_for_market(topl_id))
+        scan_results = _load_json_file(_scan_results_file_for_market(topl_id))
         open_trades = current_market["openTrades"]
         closed_trades = current_market["closedTrades"]
         trades = open_trades + closed_trades
@@ -496,7 +529,7 @@ def _sync_to_supabase() -> None:
 
         # Read kill switch status from disk (file-based, survives restarts)
         from trading.monitoring import KillSwitch
-        _ks_path = get_market_storage_dir(_CURRENT_MARKET_ID) / ".kill_switch"
+        _ks_path = get_market_storage_dir(topl_id) / ".kill_switch"
         _ks = KillSwitch(path=_ks_path)
 
         # Use actual initial_cash from state, not hardcoded 100K
@@ -558,7 +591,14 @@ def _sync_to_supabase() -> None:
             },
             "eligibleStocks": current_market["eligibleStocks"],
             "markets": markets,
-            "activeMarketId": _CURRENT_MARKET_ID,
+            # Pinned market for top-level fields (see snapshot routing fix above).
+            "topLevelMarketId": topl_id,
+            # Backwards-compat: keep the older `activeMarketId` field so any
+            # existing frontend reading it gets the same value.
+            "activeMarketId": topl_id,
+            # The market the worker is currently processing (for diagnostics
+            # only — top-level fields are NOT tied to this anymore).
+            "currentlyProcessingMarketId": _CURRENT_MARKET_ID,
             "_syncedAt": datetime.now(timezone.utc).isoformat(),
         }
 
@@ -583,12 +623,134 @@ def _sync_to_supabase() -> None:
                 cash, len(positions), len(trades),
             )
 
+        # ── Per-row history sync to Supabase DB tables ────────────────────
+        # snapshot.json above is fast/denormalised for the live dashboard.
+        # These rows give us durable, queryable history for analytics
+        # (win-rate by symbol, PnL distribution, ML calibration over time).
+        # Idempotent via UNIQUE indexes (see 20260521_worker_history.sql).
+        try:
+            _sync_worker_history_to_db(markets)
+        except Exception as exc:
+            logger.warning("Worker history DB sync failed: %s", exc)
+
     except urllib.error.HTTPError as e:
         logger.warning("Supabase sync HTTP %s: %s", e.code, e.read().decode()[:200])
     except Exception as e:
         logger.warning("Supabase sync failed: %s", e)
 
 
+def _sync_worker_history_to_db(markets: Dict[str, Dict[str, Any]]) -> None:
+    """Upsert per-row trade/signal/equity history to Supabase Postgres.
+
+    Uses PostgREST `Prefer: resolution=merge-duplicates` so re-uploading the
+    same row is a no-op via the UNIQUE indexes defined in migration
+    20260521_worker_history.sql. Errors are logged but never raised — DB
+    sync is best-effort and must not block the dashboard snapshot upload.
+    """
+    if not _SUPABASE_SERVICE_KEY:
+        return
+
+    import urllib.request
+    import urllib.error
+
+    def _post_rows(table: str, rows: List[Dict[str, Any]], on_conflict: str) -> None:
+        if not rows:
+            return
+        url = (
+            f"{_SUPABASE_URL}/rest/v1/{table}"
+            f"?on_conflict={on_conflict}"
+        )
+        req = urllib.request.Request(
+            url,
+            data=json.dumps(rows, default=str).encode("utf-8"),
+            headers={
+                "Content-Type": "application/json",
+                "Authorization": f"Bearer {_SUPABASE_SERVICE_KEY}",
+                "apikey": _SUPABASE_SERVICE_KEY,
+                # merge-duplicates → UPSERT; return=minimal → no body needed
+                "Prefer": "resolution=merge-duplicates,return=minimal",
+            },
+            method="POST",
+        )
+        try:
+            with urllib.request.urlopen(req, timeout=30) as resp:
+                resp.read()
+            logger.info("DB sync %s: %d row(s) upserted", table, len(rows))
+        except urllib.error.HTTPError as e:
+            logger.warning("DB sync %s HTTP %s: %s", table, e.code, e.read().decode()[:200])
+        except Exception as e:
+            logger.warning("DB sync %s failed: %s", table, e)
+
+    trades_rows: List[Dict[str, Any]] = []
+    signals_rows: List[Dict[str, Any]] = []
+    equity_rows: List[Dict[str, Any]] = []
+
+    for market_id, snap in markets.items():
+        for t in (snap.get("openTrades") or []) + (snap.get("closedTrades") or []):
+            src_id = t.get("id")
+            if src_id is None:
+                continue
+            trades_rows.append({
+                "market_id": market_id,
+                "source_id": int(src_id),
+                "symbol": t.get("symbol"),
+                "side": t.get("side"),
+                "entry_date": t.get("entry_date"),
+                "entry_price": t.get("entry_price"),
+                "exit_date": t.get("exit_date"),
+                "exit_price": t.get("exit_price"),
+                "quantity": int(t.get("quantity") or 0),
+                "entry_commission": t.get("entry_commission"),
+                "exit_commission": t.get("exit_commission"),
+                "gross_pnl": t.get("gross_pnl"),
+                "net_pnl": t.get("net_pnl"),
+                "return_pct": t.get("return_pct"),
+                "holding_days": t.get("holding_days"),
+                "signal_confidence": t.get("signal_confidence"),
+                "predicted_return": t.get("predicted_return"),
+                "entry_reason": t.get("entry_reason"),
+                "exit_reason": t.get("exit_reason"),
+                "is_closed": bool(t.get("is_closed")),
+                "updated_at": datetime.now(timezone.utc).isoformat(),
+            })
+
+        for s in (snap.get("signals") or []):
+            sym = s.get("symbol")
+            dt = s.get("date")
+            if not sym or not dt:
+                continue
+            signals_rows.append({
+                "market_id": market_id,
+                "date": dt,
+                "symbol": sym,
+                "signal": s.get("signal"),
+                "confidence": s.get("confidence"),
+                "predicted_return": s.get("predicted_return"),
+                "ml_signal": s.get("ml_signal"),
+                "tech_signal": s.get("tech_signal"),
+                "action": s.get("action"),
+            })
+
+        for e in (snap.get("equityCurve") or []):
+            dt = e.get("date")
+            if not dt:
+                continue
+            equity_rows.append({
+                "market_id": market_id,
+                "date": dt,
+                "equity": e.get("equity"),
+                "cash": e.get("cash"),
+                "positions_count": e.get("positions_count") or e.get("positions") or 0,
+                "unrealized_pnl": e.get("unrealized_pnl"),
+                "realized_pnl_today": e.get("realized_pnl_today") or e.get("realizedPnl"),
+                "updated_at": datetime.now(timezone.utc).isoformat(),
+            })
+
+    _post_rows("worker_trades", trades_rows, on_conflict="market_id,source_id")
+    _post_rows("worker_signals", signals_rows, on_conflict="market_id,date,symbol")
+    _post_rows("worker_equity", equity_rows, on_conflict="market_id,date")
+
+
 class TradingWorker:
     """Production trading worker with full safety stack.