diff --git a/data_manager.py b/data_manager.py index 825451e..37c6944 100644 --- a/data_manager.py +++ b/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: diff --git a/tests/test_live_pipeline.py b/tests/test_live_pipeline.py index 9ec4ecb..f5af6f5 100644 --- a/tests/test_live_pipeline.py +++ b/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): - close = _panel({"AAA": [10, 11, 12, 13, 14], "BBB": [20, 20, 21, 21, 22], - "SPY": [1, 1, 1, 1, 1]}) + def _run(self, state: dict, close=None, targets=None): + if close is None: + 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(): diff --git a/trader.py b/trader.py index 3456480..8a1c468 100644 --- a/trader.py +++ b/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