live: keep new sessions on subset refreshes; replay books from a backfilled day

A refresh of a few tickers judged a new date's coverage against every
ticker in the stored panel, so the date was dropped as sparse; it also
re-ran the sparse filter over the whole history. Coverage is now checked
only inside the refresh window, with unrequested tickers counted where they
already have a bar.

When a session that was skipped earlier appears in the panel, the trader
rewinds the ledger to that day (undoing later trades, or restoring a legacy
snapshot) and replays every session from it in order. State files are
written atomically, and the monitor refreshes volume with close/open.

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

data_manager.py

@@ -195,7 +195,7 @@ def _save_unavailable(market: str, unavailable: dict[str, str]) -> None:
         json.dump(unavailable, f, indent=0, sort_keys=True)
 
 
-def _drop_sparse_rows(df: pd.DataFrame) -> pd.DataFrame:
+def _drop_sparse_rows(df: pd.DataFrame, refreshed_columns=None) -> pd.DataFrame:
     """Drop rows where too few of the tickers alive on that date have a bar."""
     df = df.dropna(how="all")
     if df.empty:
@@ -207,6 +207,9 @@ def _drop_sparse_rows(df: pd.DataFrame) -> pd.DataFrame:
     last = np.where(has_any, n - 1 - notna[::-1].argmax(axis=0), -1)
     pos = np.arange(n)[:, None]
     alive = (pos >= first) & (pos <= last + ALIVE_GRACE_ROWS)
+    if refreshed_columns is not None:
+        unrequested = ~df.columns.isin(refreshed_columns)
+        alive[:, unrequested] = notna[:, unrequested]
     coverage = notna.sum(axis=1) / np.maximum(alive.sum(axis=1), 1)
     keep = coverage >= MIN_ROW_COVERAGE
     if not keep.all():
@@ -234,6 +237,7 @@ def _adjustment_ratio(old: pd.Series, new: pd.Series) -> float | None:
 def _merge_refresh(existing: pd.DataFrame | None, new: pd.DataFrame,
                    window_start: pd.Timestamp | None) -> pd.DataFrame:
     """Overwrite requested tickers from ``window_start`` on with freshly downloaded bars."""
+    refreshed_columns = new.columns
     new = new.dropna(axis=1, how="all")
     if existing is None or existing.empty:
         return _drop_sparse_rows(new.sort_index())
@@ -254,7 +258,10 @@ def _merge_refresh(existing: pd.DataFrame | None, new: pd.DataFrame,
                       columns=list(out.columns) + [c for c in new.columns if c not in out.columns])
     window = out.index >= window_start
     out.loc[window, new.columns] = new.reindex(out.index[window]).to_numpy()
-    return _drop_sparse_rows(out.sort_index())
+    # Unrequested columns count only where an actual bar already exists.
+    # They cannot make a newly fetched date look sparse; older history stays intact.
+    valid = _drop_sparse_rows(out.loc[window], refreshed_columns=refreshed_columns).index
+    return out.loc[~window | out.index.isin(valid)].sort_index()
 
 
 def _fill_open(close: pd.DataFrame, open_raw: pd.DataFrame | None) -> pd.DataFrame:

tests/test_live_pipeline.py

@@ -1,6 +1,7 @@
 """Contracts of the live pipeline: price panel refresh, session gating,
 order sizing, and how results are reported."""
 
+import copy
 import unittest
 from datetime import datetime, timezone
 from types import SimpleNamespace
@@ -62,17 +63,47 @@ class PanelRefreshTests(unittest.TestCase):
         self.assertNotIn(panel.index[2], cleaned.index)
         self.assertEqual(len(cleaned), 4)
 
+    def test_subset_refresh_keeps_trading_days_without_erasing_older_history(self):
+        stored = _panel({"A": [np.nan, 10, 11], "B": [np.nan, 20, 21],
+                         "C": [30, 31, 32], "D": [40, 41, 42], "E": [50, 51, 52]})
+        fresh = pd.DataFrame({"A": [11, 12], "B": [21, 22]},
+                             index=pd.to_datetime(["2026-09-16", "2026-09-17"]))
+        merged = data_manager._merge_refresh(stored, fresh, fresh.index[0])
+        self.assertIn(fresh.index[-1], merged.index)
+        self.assertEqual(merged.loc[fresh.index[-1], "A"], 12)
+        self.assertTrue(pd.isna(merged.loc[fresh.index[-1], "C"]))
+        pd.testing.assert_frame_equal(merged.loc[stored.index[:2]], stored.iloc[:2], check_freq=False)
+
+    def test_subset_refresh_still_drops_partial_rows_in_the_requested_universe(self):
+        stored = _panel({c: [10, 10, 10] for c in "ABCDEFGH"})
+        fresh = pd.DataFrame({"A": [10, np.nan], "B": [10, np.nan],
+                              "C": [10, np.nan], "D": [10, 11]},
+                             index=pd.to_datetime(["2026-09-16", "2026-09-17"]))
+        merged = data_manager._merge_refresh(stored, fresh, fresh.index[0])
+        self.assertNotIn(fresh.index[-1], merged.index)
+        self.assertIn(stored.index[0], merged.index)
+
+    def test_failed_requested_columns_still_count_toward_freshness(self):
+        stored = _panel({c: [10, 10, 10] for c in "ABCDEFGH"})
+        fresh = pd.DataFrame({"A": [10, 11], "B": [np.nan, np.nan],
+                              "C": [np.nan, np.nan], "D": [np.nan, np.nan]},
+                             index=pd.to_datetime(["2026-09-16", "2026-09-17"]))
+        merged = data_manager._merge_refresh(stored, fresh, fresh.index[0])
+        self.assertNotIn(fresh.index[-1], merged.index)
+        self.assertEqual(merged.loc[stored.index[-1], "B"], 10)
+
 
 class LiveBookTests(unittest.TestCase):
     BOOK = {"label": "test", "start": {"us": "2026-09-16"}, "capital": {"us": 5_000.0},
             "fee_per_order": {"us": 1.0}, "strategies": {"us": ["momentum"]}}
 
-    def _run(self, state: dict):
+    def _run(self, state: dict, close=None, targets=None):
-        close = _panel({"AAA": [10, 11, 12, 13, 14], "BBB": [20, 20, 21, 21, 22],
+        if close is None:
-                        "SPY": [1, 1, 1, 1, 1]})
+            close = _panel({"AAA": [10, 11, 12, 13, 14], "BBB": [20, 20, 21, 21, 22],
+                            "SPY": [1, 1, 1, 1, 1]})
         args = SimpleNamespace(market="us", strategy="momentum", book="test", capital=None,
                                tx_cost=0.001, fixed_fee=2.0, integer_shares=True)
-        equal = pd.DataFrame(0.5, index=close.index, columns=["AAA", "BBB"])
+        equal = targets if targets is not None else pd.DataFrame(0.5, index=close.index, columns=["AAA", "BBB"])
         with mock.patch.dict(trader.LIVE_BOOKS, {"test": self.BOOK}), \
                 mock.patch.object(trader, "strategy_universe", lambda m, n: (["AAA", "BBB"], "SPY")), \
                 mock.patch.object(trader, "pit_candidates", lambda m, n, s: []), \
@@ -92,6 +123,46 @@ class LiveBookTests(unittest.TestCase):
         with mock.patch.dict(self.BOOK, {"start": {"us": "2026-10-01"}}):
             self.assertIsNone(self._run({}))
 
+    def test_backfilled_session_replays_later_trades_and_is_idempotent(self):
+        close = _panel({"AAA": [10, 11, 12, 13, 14], "BBB": [20, 20, 21, 21, 22],
+                        "SPY": [1, 1, 1, 1, 1]})
+        targets = pd.DataFrame({"AAA": [1, 1, 1, 0, 1], "BBB": [0, 0, 0, 1, 0]}, index=close.index)
+        expected = self._run({}, close, targets)
+        missing = close.index[3]
+        state = self._run({}, close.drop(missing), targets)
+        self.assertNotEqual(state["cash"], expected["cash"])
+        repaired = self._run(state, close, targets)
+        self.assertEqual(repaired["holdings"], expected["holdings"])
+        self.assertAlmostEqual(repaired["cash"], expected["cash"])
+        self.assertEqual(repaired["daily_equity"], expected["daily_equity"])
+        self.assertEqual(repaired["daily_log"], expected["daily_log"])
+        self.assertEqual(repaired["trade_log"], expected["trade_log"])
+        self.assertEqual(repaired["last_evening"], "2026-09-18")
+        before = copy.deepcopy(repaired)
+        self.assertEqual(self._run(repaired, close, targets), before)
+
+    def test_backfilled_first_book_session_restores_original_capital(self):
+        close = _panel({"AAA": [10, 11, 12, 13, 14], "BBB": [20, 20, 21, 21, 22],
+                        "SPY": [1, 1, 1, 1, 1]})
+        expected = self._run({}, close)
+        state = self._run({}, close.drop(close.index[2]))
+        repaired = self._run(state, close)
+        self.assertEqual(repaired["daily_equity"], expected["daily_equity"])
+        self.assertEqual(repaired["holdings"], expected["holdings"])
+        self.assertAlmostEqual(repaired["cash"], expected["cash"])
+
+    def test_legacy_checkpoint_restores_cash_when_old_trade_values_were_rounded(self):
+        state = trader.init_state("us", "momentum", 50)
+        state.update(cash=9.98, holdings={"AAA": 4}, last_evening="2026-09-18",
+                     daily_equity={"2026-09-16": 50, "2026-09-18": 49.98},
+                     daily_log=[{"date": "2026-09-16", "cash": 50, "holdings": {}}],
+                     trade_log=[{"date": "2026-09-18", "action": "BUY", "ticker": "AAA",
+                                 "shares": 4, "value": 40, "commission": 0}])
+        trader._rewind_state(state, pd.Timestamp("2026-09-17"))
+        self.assertEqual(state["cash"], 50)
+        self.assertEqual(state["holdings"], {})
+        self.assertEqual(state["last_evening"], "2026-09-16")
+
     def test_book_strategies_are_live_strategies_of_their_market(self):
         for book, cfg in trader.LIVE_BOOKS.items():
             for market, names in cfg["strategies"].items():

trader.py

@@ -676,8 +676,10 @@ def load_state(market: str, strategy_name: str, book: str | None = None) -> dict
 def save_state(state: dict, market: str, strategy_name: str, book: str | None = None) -> None:
     path = _state_path(market, strategy_name, book)
     os.makedirs(os.path.dirname(path), exist_ok=True)
-    with open(path, "w") as f:
+    tmp = f"{path}.tmp"
+    with open(tmp, "w") as f:
         json.dump(state, f, indent=2, default=str)
+    os.replace(tmp, path)
 
 
 def init_state(market: str, strategy_name: str, capital: float) -> dict:
@@ -1830,12 +1832,44 @@ def refresh_panels(market: str, strategy_names: list[str], label: str) -> None:
         data_market = strategy_data_market(market, name)
         requests.setdefault(data_market, set()).update(
             price_request(market, list(tickers) + [benchmark]))
-        fields.setdefault(data_market, {"close", "open"}).update(strategy_fields(name))
+        fields.setdefault(data_market, {"close", "open", "volume"}).update(strategy_fields(name))
     for data_market, request in requests.items():
         print(f"[{label}] Refreshing {data_market} panel for {len(request)} tickers")
         data_manager.update(data_market, sorted(request), fields=sorted(fields[data_market]))
 
 
+def _rewind_state(state: dict, before: pd.Timestamp) -> None:
+    cutoff = str(before.date())
+    holdings = dict(state["holdings"])
+    cash = state["cash"]
+    # Trade values and commissions retain full precision; daily snapshots round cash.
+    for trade in reversed(state["trade_log"]):
+        if trade["date"] < cutoff:
+            continue
+        ticker, shares = trade["ticker"], trade["shares"]
+        if trade["action"] == "BUY":
+            holdings[ticker] = holdings.get(ticker, 0) - shares
+            cash += trade["value"] + trade["commission"]
+        else:
+            holdings[ticker] = holdings.get(ticker, 0) + shares
+            cash -= trade["value"] - trade["commission"]
+    state["holdings"] = {t: q for t, q in holdings.items() if abs(q) > 1e-8}
+    state["cash"] = cash
+    state["trade_log"] = [t for t in state["trade_log"] if t["date"] < cutoff]
+    state["daily_log"] = [d for d in state.get("daily_log", []) if d["date"] < cutoff]
+    state["daily_equity"] = {d: v for d, v in state["daily_equity"].items() if d < cutoff}
+    state["engine_changes"] = [e for e in state.get("engine_changes", []) if e["first_session"] < cutoff]
+    last = max(state["daily_equity"], default=None)
+    state["last_morning"] = state["last_evening"] = last
+    state["pending_trades"] = None
+    state["engine_version"] = state["daily_log"][-1].get("engine_version") if state["daily_log"] else None
+    if state["daily_log"] and not state["engine_version"]:
+        # The legacy ledger rounded trades, so its recorded snapshot is authoritative.
+        snapshot = state["daily_log"][-1]
+        state["cash"] = snapshot["cash"]
+        state["holdings"] = {t: h["shares"] for t, h in snapshot["holdings"].items()}
+
+
 def _process_sessions(args, state: dict | None, start: pd.Timestamp | None = None,
                       label: str = "auto", refresh: bool = True) -> dict | None:
     """Run every completed session not yet recorded in ``state`` (catch-up safe).
@@ -1876,6 +1910,16 @@ def _process_sessions(args, state: dict | None, start: pd.Timestamp | None = Non
     else:
         last = state.get("last_evening")
         first = book_start(getattr(args, "book", None), market)
+        if last and state.get("daily_equity"):
+            beginning = first if first is not None else pd.Timestamp(min(state["daily_equity"]))
+            recorded = pd.to_datetime(list(state["daily_equity"]))
+            gaps = sessions[(sessions >= beginning) & (sessions <= pd.Timestamp(last))
+                            & ~sessions.isin(recorded)]
+            if len(gaps):
+                print(f"[{label}] {strategy_name}: backfilled {gaps[0].date()}; replaying from that session")
+                _rewind_state(state, gaps[0])
+                last = state.get("last_evening")
+                first = beginning
         if last:
             todo = sessions[sessions > pd.Timestamp(last)]
         else:  # created by a morning run: start with the book's first session
@@ -1910,7 +1954,8 @@ def cmd_auto(args):
 
     Processes every completed session since the last recorded one, so missed
     days (monitor downtime, late data) are caught up in order. A session is
-    only processed after the exchange has closed.
+    only processed after the exchange has closed. Backfilled historical days
+    rewind the ledger and replay the affected sessions in order.
 
     Usage with cron (US market, run at 5pm ET weekdays):
         0 17 * * 1-5  cd /path/to/quant && uv run python trader.py auto --market us