diff --git a/README.md b/README.md index 13e69d606..9e302cbc2 100644 --- a/README.md +++ b/README.md @@ -324,6 +324,8 @@ The memory log is always on. Each completed run appends its decision to `~/.trad Override the path with `TRADINGAGENTS_MEMORY_LOG_PATH`. +A scheduled job whose tickers change can settle the decisions of tickers it no longer analyses with `ta.settle_all_pending()`, which returns the decisions it settled and any it could not. + ### Checkpoint resume Checkpoint resume is opt-in via `--checkpoint`. When enabled, LangGraph saves state after each node so a crashed or interrupted run resumes from the last successful step instead of starting over. The run view says whether it resumed a saved run or started fresh. Checkpoints are cleared automatically on successful completion. diff --git a/tests/test_memory_log.py b/tests/test_memory_log.py index b3516b3fb..9d1ae0f1f 100644 --- a/tests/test_memory_log.py +++ b/tests/test_memory_log.py @@ -1133,3 +1133,58 @@ def test_a_close_that_is_not_a_price_is_skipped_not_scored(monkeypatch): raw, alpha, _, _ = settlement.fetch_returns("NVDA", "2026-01-05", 5, "SPY") assert raw == pytest.approx(105.0 / 100.0 - 1) # the window opens on the first real close + + +@pytest.mark.unit +class TestSettleAll: + """A scheduler that stops analysing a ticker can still settle its decisions (#1445).""" + + def _setup(self, tmp_path, monkeypatch): + log = make_log(tmp_path) + log.store_decision("AAPL", "2026-01-05", DECISION_BUY) + log.store_decision("MSFT", "2026-01-05", DECISION_BUY) + idx = pd.bdate_range("2026-01-02", periods=12) + series = pd.Series([100.0 + i for i in range(12)], index=idx) + monkeypatch.setattr(settlement, "get_closes", _closes({"AAPL": series, "MSFT": series, "SPY": series})) + + class _Reflector: + def reflect_on_final_decision(self, **kwargs): + return "Held up." + return log, _Reflector() + + def test_every_ticker_with_a_due_decision_is_settled(self, tmp_path, monkeypatch): + log, reflector = self._setup(tmp_path, monkeypatch) + + result = settlement.settle_all_pending(log, reflector, {"holding_period_days": 5}) + + assert sorted(result.settled) == [("AAPL", "2026-01-05"), ("MSFT", "2026-01-05")] + assert result.failed == [] + assert not log.get_pending_entries() + + def test_a_failed_reflection_is_reported_and_left_pending(self, tmp_path, monkeypatch): + log, _ = self._setup(tmp_path, monkeypatch) + + class _Down: + def reflect_on_final_decision(self, **kwargs): + raise RuntimeError("provider down") + + result = settlement.settle_pending("AAPL", log, _Down(), {"holding_period_days": 5}) + + assert result.settled == [] and [f[:2] for f in result.failed] == [("AAPL", "2026-01-05")] + assert {e["ticker"] for e in log.get_pending_entries()} == {"AAPL", "MSFT"} + + +@pytest.mark.unit +def test_the_graph_settles_every_ticker_under_its_own_config(tmp_path, monkeypatch): + from tradingagents.dataflows.config import get_config + + graph = object.__new__(TradingAgentsGraph) + graph.config = {"holding_period_days": 5, "output_language": "Deutsch"} + graph.memory_log, graph.reflector = make_log(tmp_path), object() + seen = [] + monkeypatch.setattr(settlement, "settle_all_pending", + lambda log, reflector, config: seen.append(get_config()["output_language"]) + or settlement.Settlement(settled=[("AAPL", "2026-01-05")])) + + assert graph.settle_all_pending().settled == [("AAPL", "2026-01-05")] + assert seen == ["Deutsch"] diff --git a/tradingagents/graph/trading_graph.py b/tradingagents/graph/trading_graph.py index 180ab65b5..c24cfe7a3 100644 --- a/tradingagents/graph/trading_graph.py +++ b/tradingagents/graph/trading_graph.py @@ -322,16 +322,25 @@ class TradingAgentsGraph: portfolio_context=portfolio.render(company_name) if portfolio is not None else "", ) - def settle_pending(self, company_name): + def settle_pending(self, company_name) -> settlement.Settlement: """Settle this ticker's decisions whose holding window has now traded. - A run settles the ticker's earlier decisions on its way in, so the most - recent one stays pending until the next run for that ticker. A caller - that is done analyzing a ticker (a backtest sweep, a scheduled job) calls - this to settle it now. + A run settles its ticker's earlier decisions as it starts, so the most + recent one stays pending until a later run. A caller that is done + analyzing a ticker (a backtest sweep, a scheduled job) calls this to + settle it now. Returns what was settled and what failed. """ with run_config(self.config): - settlement.settle_pending(company_name, self.memory_log, self.reflector, self.config) + return settlement.settle_pending(company_name, self.memory_log, self.reflector, self.config) + + def settle_all_pending(self) -> settlement.Settlement: + """Settle every ticker's decisions whose holding window has now traded (#1445). + + For a scheduler whose tickers rotate: a ticker it stops analysing would + otherwise keep its decisions pending, and their lessons out of later runs. + """ + with run_config(self.config): + return settlement.settle_all_pending(self.memory_log, self.reflector, self.config) def record_decision(self, company_name, trade_date, final_state): """Record a finished run: its state log, and its decision in the memory log diff --git a/tradingagents/memory/settlement.py b/tradingagents/memory/settlement.py index de14c9b11..69e39c1fe 100644 --- a/tradingagents/memory/settlement.py +++ b/tradingagents/memory/settlement.py @@ -2,6 +2,7 @@ it against its benchmark and record a reflection on it in the memory log.""" import logging +from dataclasses import dataclass, field from datetime import datetime, timedelta from tradingagents.dataflows.symbols import normalize_symbol @@ -104,19 +105,26 @@ def fetch_returns( return None, None, None, None -def settle_pending(ticker: str, memory_log, reflector, config: dict) -> None: +@dataclass +class Settlement: + """What a settlement pass did: the decisions it settled and those it could not.""" + + settled: list[tuple[str, str]] = field(default_factory=list) # (ticker, date) + failed: list[tuple[str, str, str]] = field(default_factory=list) # (ticker, date, reason) + + +def settle_pending(ticker: str, memory_log, reflector, config: dict) -> Settlement: """Settle ``ticker``'s pending decisions whose holding window has traded. Fetches returns for each same-ticker pending entry, generates reflections, then writes all updates in a single atomic batch write to avoid redundant I/O. - Skips entries whose price data is not yet available (too recent or delisted). - - Trade-off: only same-ticker entries are resolved per run. Entries for - other tickers accumulate until that ticker is run again. + Skips entries whose price data is not yet available (too recent or delisted); + an entry whose reflection fails stays pending and is reported as failed. """ + result = Settlement() pending = [e for e in memory_log.get_pending_entries() if e["ticker"] == ticker] if not pending: - return + return result benchmark = resolve_benchmark(ticker, config) updates = [] @@ -136,10 +144,10 @@ def settle_pending(ticker: str, memory_log, reflector, config: dict) -> None: holding_days=days, ) except Exception as exc: - # Reflection calls a provider, and this runs on the way into a - # new run: a transient failure leaves the entry pending for the - # next one rather than stopping the analysis that was asked for. + # Reflection calls a provider: a transient failure leaves the entry + # pending for the next pass rather than stopping the analysis. logger.warning("Reflection failed for %s on %s: %s", ticker, entry["date"], exc) + result.failed.append((ticker, entry["date"], f"reflection failed: {exc}")) continue updates.append({ "ticker": ticker, @@ -153,3 +161,19 @@ def settle_pending(ticker: str, memory_log, reflector, config: dict) -> None: if updates: memory_log.batch_update_with_outcomes(updates) + result.settled.extend((ticker, u["trade_date"]) for u in updates) + return result + + +def settle_all_pending(memory_log, reflector, config: dict) -> Settlement: + """Settle every ticker's pending decisions whose holding window has traded (#1445). + + A run settles its own ticker; a ticker no longer analysed would otherwise + keep its decisions pending, and their lessons out of later runs. + """ + result = Settlement() + for ticker in dict.fromkeys(e["ticker"] for e in memory_log.get_pending_entries()): + done = settle_pending(ticker, memory_log, reflector, config) + result.settled.extend(done.settled) + result.failed.extend(done.failed) + return result