From 969861e8be80d8b04ac7e88e37c8007e8c69fea5 Mon Sep 17 00:00:00 2001 From: Yijia-Xiao Date: Wed, 23 Sep 2026 23:28:47 +0000 Subject: [PATCH] refactor(graph): settle past decisions in graph/settlement.py - resolve_benchmark, fetch_returns and settle_pending are module functions; the graph's settle_pending runs them under its config - create_run_state settles through settle_pending, so the CLI path also settles under the graph's config --- tests/test_cli_decision_log.py | 2 +- tests/test_dataflows_config.py | 9 +- tests/test_memory_log.py | 134 +++++++++-------------- tests/test_portfolio_context.py | 2 +- tests/test_symbol_normalization_paths.py | 7 +- tradingagents/graph/settlement.py | 132 ++++++++++++++++++++++ tradingagents/graph/trading_graph.py | 128 +--------------------- 7 files changed, 201 insertions(+), 213 deletions(-) create mode 100644 tradingagents/graph/settlement.py diff --git a/tests/test_cli_decision_log.py b/tests/test_cli_decision_log.py index d58c0d962..f9f291bd7 100644 --- a/tests/test_cli_decision_log.py +++ b/tests/test_cli_decision_log.py @@ -30,7 +30,7 @@ def test_create_run_state_settles_pending_and_carries_context(tmp_path, monkeypa graph = _bare_graph(tmp_path) graph.propagator = Propagator() settled = [] - monkeypatch.setattr(graph, "_resolve_pending_entries", settled.append, raising=False) + monkeypatch.setattr(graph, "settle_pending", settled.append, raising=False) monkeypatch.setattr(graph, "resolve_instrument_context", lambda t, a="stock", d=None: f"id:{t}", raising=False) monkeypatch.setattr(graph, "_memory_as_of", lambda d: d, raising=False) graph.memory_log.store_decision("NVDA", "2026-01-05", "Rating: Buy\nold call") diff --git a/tests/test_dataflows_config.py b/tests/test_dataflows_config.py index e4a8113c6..fd1ef37e5 100644 --- a/tests/test_dataflows_config.py +++ b/tests/test_dataflows_config.py @@ -140,15 +140,18 @@ def test_concurrent_runs_each_read_their_own_config(): @pytest.mark.unit -def test_settling_reads_the_graphs_own_config(): +def test_settling_reads_the_graphs_own_config(monkeypatch): from tradingagents.dataflows.router import get_vendor config = copy.deepcopy(default_config.DEFAULT_CONFIG) config["tool_vendors"] = {"get_stock_data": "alpha_vantage"} graph = _graph(config) + graph.memory_log = graph.reflector = None # the settlement below is a stand-in seen = [] - graph._resolve_pending_entries = lambda ticker: seen.append( - get_vendor("core_stock_apis", "get_stock_data")) + from tradingagents.graph import settlement + + monkeypatch.setattr(settlement, "settle_pending", + lambda *a: seen.append(get_vendor("core_stock_apis", "get_stock_data"))) graph.settle_pending("AAPL") diff --git a/tests/test_memory_log.py b/tests/test_memory_log.py index a92cd10bb..42dc34b11 100644 --- a/tests/test_memory_log.py +++ b/tests/test_memory_log.py @@ -8,6 +8,7 @@ import pytest from tradingagents.agents.managers.portfolio_manager import create_portfolio_manager from tradingagents.agents.schemas import PortfolioDecision, PortfolioRating from tradingagents.decision_log import TradingMemoryLog +from tradingagents.graph import settlement from tradingagents.graph.propagation import Propagator from tradingagents.graph.reflection import Reflector from tradingagents.graph.trading_graph import TradingAgentsGraph @@ -407,7 +408,7 @@ class TestTradingMemoryLogCore: # --------------------------------------------------------------------------- -# Deferred reflection: update_with_outcome, Reflector, _fetch_returns +# Deferred reflection: update_with_outcome, Reflector, fetch_returns # --------------------------------------------------------------------------- class TestDeferredReflection: @@ -510,19 +511,18 @@ class TestDeferredReflection: assert "-5.0%" in human_content assert "Exit position immediately." in human_content - # TradingAgentsGraph._fetch_returns + # settlement.fetch_returns def test_fetch_returns_valid_ticker(self): stock_prices = [100.0, 102.0, 104.0, 103.0, 105.0, 106.0] spy_prices = [400.0, 402.0, 404.0, 403.0, 405.0, 406.0] - mock_graph = MagicMock(spec=TradingAgentsGraph) with patch("yfinance.Ticker") as mock_ticker_cls: def _make_ticker(sym): m = MagicMock() m.history.return_value = _price_df(spy_prices if sym == "SPY" else stock_prices) return m mock_ticker_cls.side_effect = _make_ticker - raw, alpha, days, resolved = TradingAgentsGraph._fetch_returns(mock_graph, "NVDA", "2026-01-05") + raw, alpha, days, resolved = settlement.fetch_returns("NVDA", "2026-01-05") assert raw is not None and alpha is not None and days is not None assert isinstance(raw, float) and isinstance(alpha, float) and isinstance(days, int) assert days == 5 @@ -531,22 +531,20 @@ class TestDeferredReflection: def test_fetch_returns_too_recent(self): """Only 1 data point available → returns all-None, no crash.""" - mock_graph = MagicMock(spec=TradingAgentsGraph) with patch("yfinance.Ticker") as mock_ticker_cls: m = MagicMock() m.history.return_value = _price_df([100.0]) mock_ticker_cls.return_value = m - raw, alpha, days, resolved = TradingAgentsGraph._fetch_returns(mock_graph, "NVDA", "2026-04-19") + raw, alpha, days, resolved = settlement.fetch_returns("NVDA", "2026-04-19") assert (raw, alpha, days, resolved) == (None, None, None, None) def test_fetch_returns_delisted(self): """Empty DataFrame → returns all-None, no crash.""" - mock_graph = MagicMock(spec=TradingAgentsGraph) with patch("yfinance.Ticker") as mock_ticker_cls: m = MagicMock() m.history.return_value = pd.DataFrame({"Close": []}) mock_ticker_cls.return_value = m - raw, alpha, days, resolved = TradingAgentsGraph._fetch_returns(mock_graph, "XXXXXFAKE", "2026-01-10") + raw, alpha, days, resolved = settlement.fetch_returns("XXXXXFAKE", "2026-01-10") assert (raw, alpha, days, resolved) == (None, None, None, None) def test_fetch_returns_spy_shorter_than_stock(self): @@ -554,14 +552,13 @@ class TestDeferredReflection: not raise IndexError.""" stock_prices = [100.0, 102.0, 104.0, 103.0, 105.0, 106.0, 107.0, 108.0] # 8 rows spy_prices = [400.0, 402.0, 403.0, 405.0, 406.0, 407.0] # 6 rows - mock_graph = MagicMock(spec=TradingAgentsGraph) with patch("yfinance.Ticker") as mock_ticker_cls: def _make_ticker(sym): m = MagicMock() m.history.return_value = _price_df(spy_prices if sym == "SPY" else stock_prices) return m mock_ticker_cls.side_effect = _make_ticker - raw, alpha, days, resolved = TradingAgentsGraph._fetch_returns(mock_graph, "NVDA", "2026-01-05") + raw, alpha, days, resolved = settlement.fetch_returns("NVDA", "2026-01-05") assert raw is not None and alpha is not None assert days == 5 # full holding window used for both series assert resolved == "2026-01-10" @@ -572,32 +569,29 @@ class TestDeferredReflection: on a premature partial return.""" stock_prices = [100.0, 102.0, 104.0] # only 3 rows; holding window is 5 spy_prices = [400.0, 402.0, 404.0] - mock_graph = MagicMock(spec=TradingAgentsGraph) with patch("yfinance.Ticker") as mock_ticker_cls: def _make_ticker(sym): m = MagicMock() m.history.return_value = _price_df(spy_prices if sym == "SPY" else stock_prices) return m mock_ticker_cls.side_effect = _make_ticker - result = TradingAgentsGraph._fetch_returns(mock_graph, "NVDA", "2026-01-05") + result = settlement.fetch_returns("NVDA", "2026-01-05") assert result == (None, None, None, None) - # TradingAgentsGraph._resolve_benchmark — picks index for alpha calc + # settlement.resolve_benchmark picks the index for the alpha calculation def test_resolve_benchmark_explicit_override(self): """config['benchmark_ticker'] wins for every ticker.""" - mock_graph = MagicMock(spec=TradingAgentsGraph) - mock_graph.config = { + config = { "benchmark_ticker": "QQQ", "benchmark_map": {"": "SPY", ".T": "^N225"}, } - assert TradingAgentsGraph._resolve_benchmark(mock_graph, "7203.T") == "QQQ" - assert TradingAgentsGraph._resolve_benchmark(mock_graph, "NVDA") == "QQQ" + assert settlement.resolve_benchmark("7203.T", config) == "QQQ" + assert settlement.resolve_benchmark("NVDA", config) == "QQQ" def test_resolve_benchmark_suffix_map(self): """Known suffixes route to their regional index.""" - mock_graph = MagicMock(spec=TradingAgentsGraph) - mock_graph.config = { + config = { "benchmark_ticker": None, "benchmark_map": { ".T": "^N225", ".HK": "^HSI", ".NS": "^NSEI", @@ -605,66 +599,60 @@ class TestDeferredReflection: ".BO": "^BSESN", "": "SPY", }, } - assert TradingAgentsGraph._resolve_benchmark(mock_graph, "7203.T") == "^N225" - assert TradingAgentsGraph._resolve_benchmark(mock_graph, "0700.HK") == "^HSI" - assert TradingAgentsGraph._resolve_benchmark(mock_graph, "RELIANCE.NS") == "^NSEI" - assert TradingAgentsGraph._resolve_benchmark(mock_graph, "AZN.L") == "^FTSE" + assert settlement.resolve_benchmark("7203.T", config) == "^N225" + assert settlement.resolve_benchmark("0700.HK", config) == "^HSI" + assert settlement.resolve_benchmark("RELIANCE.NS", config) == "^NSEI" + assert settlement.resolve_benchmark("AZN.L", config) == "^FTSE" def test_explicit_benchmark_is_resolved_like_any_other_symbol(self): """A configured benchmark takes the same alias mapping as the ticker, or the return lookup finds nothing and the decision never settles.""" - mock_graph = MagicMock(spec=TradingAgentsGraph) - mock_graph.config = {"benchmark_ticker": "SPX500", "benchmark_map": {"": "SPY"}} - assert TradingAgentsGraph._resolve_benchmark(mock_graph, "NVDA") == "^GSPC" + config = {"benchmark_ticker": "SPX500", "benchmark_map": {"": "SPY"}} + assert settlement.resolve_benchmark("NVDA", config) == "^GSPC" def test_resolve_benchmark_china_a_shares(self): """A-share tickers route to their exchange composite (uses the real default benchmark_map, since A-share support relies on it).""" from tradingagents.default_config import DEFAULT_CONFIG - mock_graph = MagicMock(spec=TradingAgentsGraph) - mock_graph.config = {"benchmark_ticker": None, + config = {"benchmark_ticker": None, "benchmark_map": DEFAULT_CONFIG["benchmark_map"]} - assert TradingAgentsGraph._resolve_benchmark(mock_graph, "600519.SS") == "000001.SS" - assert TradingAgentsGraph._resolve_benchmark(mock_graph, "000001.SZ") == "399001.SZ" + assert settlement.resolve_benchmark("600519.SS", config) == "000001.SS" + assert settlement.resolve_benchmark("000001.SZ", config) == "399001.SZ" # .SH is the exchange's own suffix; Yahoo spells Shanghai .SS (#1260) - assert TradingAgentsGraph._resolve_benchmark(mock_graph, "600519.SH") == "000001.SS" + assert settlement.resolve_benchmark("600519.SH", config) == "000001.SS" def test_resolve_benchmark_brazil(self): """B3 tickers were measured against SPY.""" from tradingagents.default_config import DEFAULT_CONFIG - mock_graph = MagicMock(spec=TradingAgentsGraph) - mock_graph.config = {"benchmark_ticker": None, + config = {"benchmark_ticker": None, "benchmark_map": DEFAULT_CONFIG["benchmark_map"]} - assert TradingAgentsGraph._resolve_benchmark(mock_graph, "PETR4.SA") == "^BVSP" + assert settlement.resolve_benchmark("PETR4.SA", config) == "^BVSP" def test_resolve_benchmark_us_ticker_defaults_to_spy(self): """US tickers (no dotted suffix) take the empty-suffix entry.""" - mock_graph = MagicMock(spec=TradingAgentsGraph) - mock_graph.config = { + config = { "benchmark_ticker": None, "benchmark_map": {"": "SPY", ".T": "^N225"}, } - assert TradingAgentsGraph._resolve_benchmark(mock_graph, "NVDA") == "SPY" - assert TradingAgentsGraph._resolve_benchmark(mock_graph, "AAPL") == "SPY" + assert settlement.resolve_benchmark("NVDA", config) == "SPY" + assert settlement.resolve_benchmark("AAPL", config) == "SPY" def test_resolve_benchmark_unknown_suffix_falls_back(self): """Unrecognised suffix (BRK.B, FAKE.XX) falls back to SPY.""" - mock_graph = MagicMock(spec=TradingAgentsGraph) - mock_graph.config = { + config = { "benchmark_ticker": None, "benchmark_map": {"": "SPY", ".T": "^N225"}, } - assert TradingAgentsGraph._resolve_benchmark(mock_graph, "FAKE.XX") == "SPY" - assert TradingAgentsGraph._resolve_benchmark(mock_graph, "BRK.B") == "SPY" + assert settlement.resolve_benchmark("FAKE.XX", config) == "SPY" + assert settlement.resolve_benchmark("BRK.B", config) == "SPY" def test_resolve_benchmark_case_insensitive(self): """Suffix matching is case-insensitive so 7203.t resolves like 7203.T.""" - mock_graph = MagicMock(spec=TradingAgentsGraph) - mock_graph.config = { + config = { "benchmark_ticker": None, "benchmark_map": {".T": "^N225", "": "SPY"}, } - assert TradingAgentsGraph._resolve_benchmark(mock_graph, "7203.t") == "^N225" + assert settlement.resolve_benchmark("7203.t", config) == "^N225" def test_reflector_includes_benchmark_in_label(self): """benchmark_name appears in the prompt label, not 'SPY' hardcoded.""" @@ -696,32 +684,25 @@ class TestDeferredReflection: human_content = next(content for role, content in messages if role == "human") assert "Alpha vs SPY:" in human_content - # TradingAgentsGraph._resolve_pending_entries + # settlement.settle_pending def test_resolve_skips_other_tickers(self, tmp_path): """Pending AAPL entry is not resolved when the run is for NVDA.""" log = make_log(tmp_path) log.store_decision("AAPL", "2026-01-10", DECISION_BUY) - mock_graph = MagicMock(spec=TradingAgentsGraph) - mock_graph.config = {} - mock_graph.memory_log = log - mock_graph._fetch_returns = MagicMock(return_value=(0.05, 0.02, 5, "2026-01-12")) - TradingAgentsGraph._resolve_pending_entries(mock_graph, "NVDA") - mock_graph._fetch_returns.assert_not_called() + with patch.object(settlement, "fetch_returns", return_value=(0.05, 0.02, 5, "2026-01-12")) as fetch: + settlement.settle_pending("NVDA", log, MagicMock(), {}) + fetch.assert_not_called() assert len(log.get_pending_entries()) == 1 def test_resolve_marks_entry_completed(self, tmp_path): """After resolve, get_pending_entries() is empty and the entry has a REFLECTION.""" log = make_log(tmp_path) log.store_decision("NVDA", "2026-01-05", DECISION_BUY) - mock_reflector = MagicMock() - mock_reflector.reflect_on_final_decision.return_value = "Momentum confirmed." - mock_graph = MagicMock(spec=TradingAgentsGraph) - mock_graph.config = {} - mock_graph.memory_log = log - mock_graph.reflector = mock_reflector - mock_graph._fetch_returns = MagicMock(return_value=(0.05, 0.02, 5, "2026-01-12")) - TradingAgentsGraph._resolve_pending_entries(mock_graph, "NVDA") + reflector = MagicMock() + reflector.reflect_on_final_decision.return_value = "Momentum confirmed." + with patch.object(settlement, "fetch_returns", return_value=(0.05, 0.02, 5, "2026-01-12")): + settlement.settle_pending("NVDA", log, reflector, {}) assert log.get_pending_entries() == [] entries = log.load_entries() assert len(entries) == 1 @@ -731,19 +712,15 @@ class TestDeferredReflection: assert "+2.0%" in entries[0]["alpha"] def test_resolve_leaves_premature_entry_pending(self, tmp_path): - """#1169: when the outcome can't be settled yet (_fetch_returns None), + """#1169: when the outcome can't be settled yet (fetch_returns None), the entry stays pending and the reflector is never called.""" log = make_log(tmp_path) log.store_decision("NVDA", "2026-01-05", DECISION_BUY) - mock_reflector = MagicMock() - mock_graph = MagicMock(spec=TradingAgentsGraph) - mock_graph.config = {} - mock_graph.memory_log = log - mock_graph.reflector = mock_reflector - mock_graph._fetch_returns = MagicMock(return_value=(None, None, None, None)) - TradingAgentsGraph._resolve_pending_entries(mock_graph, "NVDA") + reflector = MagicMock() + with patch.object(settlement, "fetch_returns", return_value=(None, None, None, None)): + settlement.settle_pending("NVDA", log, reflector, {}) assert len(log.get_pending_entries()) == 1 # still pending - mock_reflector.reflect_on_final_decision.assert_not_called() + reflector.reflect_on_final_decision.assert_not_called() # --------------------------------------------------------------------------- @@ -965,9 +942,9 @@ def test_a_failed_reflection_leaves_the_entry_pending_and_lets_the_run_start(tmp graph.memory_log = TradingMemoryLog(graph.config) graph.memory_log.store_decision("NVDA", "2026-01-05", "Rating: Buy\n\nx") graph.memory_log.store_decision("NVDA", "2026-01-12", "Rating: Sell\n\ny") - monkeypatch.setattr(graph, "_resolve_benchmark", lambda t: "SPY", raising=False) - monkeypatch.setattr(graph, "_fetch_returns", - lambda t, d, holding_days=5, benchmark=None: (0.01, 0.005, holding_days, "2026-01-19"), raising=False) + monkeypatch.setattr(settlement, "resolve_benchmark", lambda t, c: "SPY") + monkeypatch.setattr(settlement, "fetch_returns", + lambda t, d, holding_days=5, benchmark=None: (0.01, 0.005, holding_days, "2026-01-19")) class _Reflector: calls = 0 @@ -980,7 +957,7 @@ def test_a_failed_reflection_leaves_the_entry_pending_and_lets_the_run_start(tmp graph.reflector = _Reflector() - graph._resolve_pending_entries("NVDA") # must not raise + graph.settle_pending("NVDA") # must not raise entries = graph.memory_log.load_entries() assert [e["pending"] for e in entries] == [True, False] # the failed one waits for next time @@ -997,17 +974,17 @@ def test_the_holding_window_is_configurable(tmp_path, monkeypatch): graph.config = {"memory_log_path": str(tmp_path / "m.md"), "holding_period_days": 21} graph.memory_log = TradingMemoryLog(graph.config) graph.memory_log.store_decision("NVDA", "2026-01-05", "**Rating**: Buy\n\nx") - monkeypatch.setattr(graph, "_resolve_benchmark", lambda t: "SPY", raising=False) + monkeypatch.setattr(settlement, "resolve_benchmark", lambda t, c: "SPY") asked = {} def _returns(ticker, date, holding_days=5, benchmark=None): asked["holding_days"] = holding_days return 0.05, 0.02, holding_days, "2026-02-02" - monkeypatch.setattr(graph, "_fetch_returns", _returns, raising=False) + monkeypatch.setattr(settlement, "fetch_returns", _returns) graph.reflector = type("R", (), {"reflect_on_final_decision": lambda self, **kw: "lesson"})() - graph._resolve_pending_entries("NVDA") + graph.settle_pending("NVDA") assert asked["holding_days"] == 21 assert graph.memory_log.load_entries()[0]["holding"] == "21d" @@ -1027,9 +1004,6 @@ def test_the_reflection_states_the_window_it_judges(): def test_a_longer_window_asks_for_enough_price_history(monkeypatch): """Trading days are not calendar days: a 21-day window needs about a month of bars, and asking for 28 days left every outcome unsettled.""" - from tradingagents.graph.trading_graph import TradingAgentsGraph - - graph = object.__new__(TradingAgentsGraph) asked = {} class _Ticker: @@ -1044,6 +1018,6 @@ def test_a_longer_window_asks_for_enough_price_history(monkeypatch): monkeypatch.setattr("tradingagents.dataflows.vendors.yahoo.market.yf.Ticker", _Ticker) - raw, alpha, days, resolved = graph._fetch_returns("NVDA", "2026-06-01", 21, benchmark="SPY") + raw, alpha, days, resolved = settlement.fetch_returns("NVDA", "2026-06-01", 21, benchmark="SPY") assert days == 21 and resolved is not None, (raw, alpha, days, resolved) diff --git a/tests/test_portfolio_context.py b/tests/test_portfolio_context.py index f467fe441..b48481e39 100644 --- a/tests/test_portfolio_context.py +++ b/tests/test_portfolio_context.py @@ -86,7 +86,7 @@ def _bare_graph(tmp_path): graph.memory_log = TradingMemoryLog(graph.config) graph.propagator = Propagator() graph.selected_analysts = ["market"] - graph._resolve_pending_entries = lambda t: None + graph.settle_pending = lambda t: None graph.resolve_instrument_context = lambda t, a="stock", d=None: "" graph._memory_as_of = lambda d: None return graph diff --git a/tests/test_symbol_normalization_paths.py b/tests/test_symbol_normalization_paths.py index d29974c3b..f802ae53b 100644 --- a/tests/test_symbol_normalization_paths.py +++ b/tests/test_symbol_normalization_paths.py @@ -10,7 +10,7 @@ import pandas as pd import tradingagents.agents.context as au import tradingagents.dataflows.vendors.yahoo.market as yahoo_market import tradingagents.dataflows.vendors.yahoo.news as ynews -from tradingagents.graph.trading_graph import TradingAgentsGraph +from tradingagents.graph import settlement def test_identity_lookup_normalizes_symbol(monkeypatch): @@ -47,9 +47,8 @@ def test_fetch_returns_normalizes_symbol(monkeypatch): monkeypatch.setattr(yahoo_market.yf, "Ticker", FakeTicker) - # _fetch_returns does not use ``self``; call unbound to avoid building the graph. - raw, alpha, days, resolved = TradingAgentsGraph._fetch_returns( - None, "XAUUSD", "2025-01-02", holding_days=5, benchmark="SPY" + raw, alpha, days, resolved = settlement.fetch_returns( + "XAUUSD", "2025-01-02", holding_days=5, benchmark="SPY" ) assert queried[0] == "GC=F" # stock symbol normalized (#984) diff --git a/tradingagents/graph/settlement.py b/tradingagents/graph/settlement.py new file mode 100644 index 000000000..794b987ee --- /dev/null +++ b/tradingagents/graph/settlement.py @@ -0,0 +1,132 @@ +"""Settling past decisions: once a decision's holding window has traded, score +it against its benchmark and record a reflection on it in the decision log.""" + +import logging +from datetime import datetime, timedelta + +from tradingagents.dataflows.symbols import normalize_symbol +from tradingagents.dataflows.vendors.yahoo.market import get_closes + +logger = logging.getLogger(__name__) + + +def resolve_benchmark(ticker: str, config: dict) -> str: + """Pick the benchmark ticker for alpha calculation against ``ticker``. + + ``config["benchmark_ticker"]`` overrides everything when set; otherwise + the suffix map matches the ticker's exchange suffix (e.g. ``.T`` for + Tokyo). US-listed tickers without a dotted suffix fall through to the + empty-suffix entry (SPY by default). Unrecognised suffixes (including + US tickers with dots like ``BRK.B``) also fall back to the empty-suffix + entry, which is the right default because the alpha calculation works + in USD. + """ + + explicit = config.get("benchmark_ticker") + if explicit: + # Same alias mapping as the analyzed ticker; an unmapped alias finds + # no prices, and the decision would stay pending for good. + return normalize_symbol(explicit) + benchmark_map = config.get("benchmark_map", {}) + ticker_upper = normalize_symbol(ticker) + for suffix, benchmark in benchmark_map.items(): + if suffix and ticker_upper.endswith(suffix.upper()): + return benchmark + return benchmark_map.get("", "SPY") + + +def fetch_returns( + ticker: str, trade_date: str, holding_days: int = 5, + benchmark: str = "SPY", +) -> tuple[float | None, float | None, int | None, str | None]: + """Fetch raw and alpha return for ticker over holding_days from trade_date. + + ``benchmark`` is the index used as the alpha baseline (resolved by the + caller via ``resolve_benchmark``). Returns ``(raw_return, alpha_return, + holding_days, resolution_date)`` — where ``resolution_date`` is the date + of the last price bar used, i.e. when the outcome became known (#1251) — + or ``(None, None, None, None)`` when the outcome cannot be settled yet: + the full holding window has not traded (#1169), or the symbol is delisted + or unreachable. + """ + try: + start = datetime.strptime(trade_date, "%Y-%m-%d") + # holding_days counts trading days, so ask for the calendar span they + # occupy (about 7 for every 5) plus a week for holidays. + end = start + timedelta(days=round(holding_days * 7 / 5) + 7) + end_str = end.strftime("%Y-%m-%d") + + # Closes for the instrument the analysis priced (XAUUSD -> GC=F, #984). + stock = get_closes(ticker, trade_date, end_str) + bench = get_closes(benchmark, trade_date, end_str) + + # Require the full holding window in both series. A rerun before it + # has traded leaves the entry pending to retry next run, rather than + # settling on a premature partial return (#1169). + if len(stock) <= holding_days or len(bench) <= holding_days: + return None, None, None, None + + raw = float((stock.iloc[holding_days] - stock.iloc[0]) / stock.iloc[0]) + bench_ret = float((bench.iloc[holding_days] - bench.iloc[0]) / bench.iloc[0]) + alpha = raw - bench_ret + # The date of the last price bar used is when this outcome became + # known — the point-in-time cutoff for injecting the lesson (#1251). + resolution_date = stock.index[holding_days].strftime("%Y-%m-%d") + return raw, alpha, holding_days, resolution_date + except Exception as e: + logger.warning( + "Could not resolve outcome for %s on %s vs %s (will retry next run): %s", + ticker, trade_date, benchmark, e, + ) + return None, None, None, None + + +def settle_pending(ticker: str, memory_log, reflector, config: dict) -> None: + """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. + """ + pending = [e for e in memory_log.get_pending_entries() if e["ticker"] == ticker] + if not pending: + return + + benchmark = resolve_benchmark(ticker, config) + updates = [] + for entry in pending: + raw, alpha, days, resolution_date = fetch_returns( + ticker, entry["date"], config.get("holding_period_days", 5), + benchmark=benchmark, + ) + if raw is None: + continue # price not available yet — try again next run + try: + reflection = reflector.reflect_on_final_decision( + final_decision=entry.get("decision", ""), + raw_return=raw, + alpha_return=alpha, + benchmark_name=benchmark, + 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. + logger.warning("Reflection failed for %s on %s: %s", ticker, entry["date"], exc) + continue + updates.append({ + "ticker": ticker, + "trade_date": entry["date"], + "raw_return": raw, + "alpha_return": alpha, + "holding_days": days, + "reflection": reflection, + "resolution_date": resolution_date, + }) + + if updates: + memory_log.batch_update_with_outcomes(updates) diff --git a/tradingagents/graph/trading_graph.py b/tradingagents/graph/trading_graph.py index e896b9152..9ba2bc496 100644 --- a/tradingagents/graph/trading_graph.py +++ b/tradingagents/graph/trading_graph.py @@ -4,7 +4,7 @@ import json import logging import os from contextlib import contextmanager -from datetime import datetime, timedelta +from datetime import datetime from pathlib import Path from typing import Any @@ -13,12 +13,12 @@ from tradingagents.agents.rating import parse_rating from tradingagents.dataflows.config import run_config, set_config from tradingagents.dataflows.date_window import get_current_date from tradingagents.dataflows.symbols import safe_ticker_component -from tradingagents.dataflows.vendors.yahoo.market import get_closes from tradingagents.decision_log import TradingMemoryLog from tradingagents.default_config import DEFAULT_CONFIG from tradingagents.llm_clients import build_llm_kwargs, create_llm_client from tradingagents.reporting import write_report_tree +from . import settlement from .checkpointer import checkpoint_step, clear_checkpoint, get_checkpointer, thread_id from .conditional_logic import ConditionalLogic from .propagation import Propagator @@ -121,126 +121,6 @@ class TradingAgentsGraph: self._checkpointer_ctx = None self._resuming = False - def _resolve_benchmark(self, ticker: str) -> str: - """Pick the benchmark ticker for alpha calculation against ``ticker``. - - ``config["benchmark_ticker"]`` overrides everything when set; otherwise - the suffix map matches the ticker's exchange suffix (e.g. ``.T`` for - Tokyo). US-listed tickers without a dotted suffix fall through to the - empty-suffix entry (SPY by default). Unrecognised suffixes (including - US tickers with dots like ``BRK.B``) also fall back to the empty-suffix - entry, which is the right default because the alpha calculation works - in USD. - """ - from tradingagents.dataflows.symbols import normalize_symbol - - explicit = self.config.get("benchmark_ticker") - if explicit: - # Same alias mapping as the analyzed ticker; an unmapped alias finds - # no prices, and the decision would stay pending for good. - return normalize_symbol(explicit) - benchmark_map = self.config.get("benchmark_map", {}) - ticker_upper = normalize_symbol(ticker) - for suffix, benchmark in benchmark_map.items(): - if suffix and ticker_upper.endswith(suffix.upper()): - return benchmark - return benchmark_map.get("", "SPY") - - def _fetch_returns( - self, ticker: str, trade_date: str, holding_days: int = 5, - benchmark: str = "SPY", - ) -> tuple[float | None, float | None, int | None, str | None]: - """Fetch raw and alpha return for ticker over holding_days from trade_date. - - ``benchmark`` is the index used as the alpha baseline (resolved by the - caller via ``_resolve_benchmark``). Returns ``(raw_return, alpha_return, - holding_days, resolution_date)`` — where ``resolution_date`` is the date - of the last price bar used, i.e. when the outcome became known (#1251) — - or ``(None, None, None, None)`` when the outcome cannot be settled yet: - the full holding window has not traded (#1169), or the symbol is delisted - or unreachable. - """ - try: - start = datetime.strptime(trade_date, "%Y-%m-%d") - # holding_days counts trading days, so ask for the calendar span they - # occupy (about 7 for every 5) plus a week for holidays. - end = start + timedelta(days=round(holding_days * 7 / 5) + 7) - end_str = end.strftime("%Y-%m-%d") - - # Closes for the instrument the analysis priced (XAUUSD -> GC=F, #984). - stock = get_closes(ticker, trade_date, end_str) - bench = get_closes(benchmark, trade_date, end_str) - - # Require the full holding window in both series. A rerun before it - # has traded leaves the entry pending to retry next run, rather than - # settling on a premature partial return (#1169). - if len(stock) <= holding_days or len(bench) <= holding_days: - return None, None, None, None - - raw = float((stock.iloc[holding_days] - stock.iloc[0]) / stock.iloc[0]) - bench_ret = float((bench.iloc[holding_days] - bench.iloc[0]) / bench.iloc[0]) - alpha = raw - bench_ret - # The date of the last price bar used is when this outcome became - # known — the point-in-time cutoff for injecting the lesson (#1251). - resolution_date = stock.index[holding_days].strftime("%Y-%m-%d") - return raw, alpha, holding_days, resolution_date - except Exception as e: - logger.warning( - "Could not resolve outcome for %s on %s vs %s (will retry next run): %s", - ticker, trade_date, benchmark, e, - ) - return None, None, None, None - - def _resolve_pending_entries(self, ticker: str) -> None: - """Resolve pending log entries for ticker at the start of a new run. - - 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. - """ - pending = [e for e in self.memory_log.get_pending_entries() if e["ticker"] == ticker] - if not pending: - return - - benchmark = self._resolve_benchmark(ticker) - updates = [] - for entry in pending: - raw, alpha, days, resolution_date = self._fetch_returns( - ticker, entry["date"], self.config.get("holding_period_days", 5), - benchmark=benchmark, - ) - if raw is None: - continue # price not available yet — try again next run - try: - reflection = self.reflector.reflect_on_final_decision( - final_decision=entry.get("decision", ""), - raw_return=raw, - alpha_return=alpha, - benchmark_name=benchmark, - 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. - logger.warning("Reflection failed for %s on %s: %s", ticker, entry["date"], exc) - continue - updates.append({ - "ticker": ticker, - "trade_date": entry["date"], - "raw_return": raw, - "alpha_return": alpha, - "holding_days": days, - "reflection": reflection, - "resolution_date": resolution_date, - }) - - if updates: - self.memory_log.batch_update_with_outcomes(updates) - def resolve_instrument_context(self, ticker: str, asset_type: str = "stock", curr_date: str | None = None) -> str: """Resolve ticker identity once and return the full instrument context. @@ -392,7 +272,7 @@ class TradingAgentsGraph: resolved instrument identity for every agent (#814). An entry point that assembled the state itself would skip the decision log. """ - self._resolve_pending_entries(company_name) + self.settle_pending(company_name) return self.propagator.create_initial_state( company_name, trade_date, @@ -413,7 +293,7 @@ class TradingAgentsGraph: this to settle it now. """ with run_config(self.config): - self._resolve_pending_entries(company_name) + settlement.settle_pending(company_name, self.memory_log, self.reflector, self.config) def record_decision(self, company_name, trade_date, final_state): """Log a finished run's decision for reflection on the next same-ticker run."""