diff --git a/tests/test_backtest.py b/tests/test_backtest.py new file mode 100644 index 000000000..a81ce7778 --- /dev/null +++ b/tests/test_backtest.py @@ -0,0 +1,181 @@ +"""Backtesting: many single-shot decisions, scored by the decision log. + +A run already records its rating and later settles it with realized and alpha +return against the regional benchmark. A backtest is that machinery over a grid +of tickers and dates, aggregated. It evaluates decision quality; it does not +simulate a portfolio, so there is no execution, no fees and no equity curve. +""" + +from __future__ import annotations + +import pytest + +from tradingagents.agents.utils.memory import TradingMemoryLog +from tradingagents.backtest import iter_grid, run_backtest, summarize + +DECISION = "Rating: Buy\n\nbuy it" + + +@pytest.mark.unit +def test_grid_spacing_and_canonical_dates(): + assert iter_grid("2026-01-05", "2026-01-20", every_n_days=7) == ["2026-01-05", "2026-01-12", "2026-01-19"] + + +@pytest.mark.unit +def test_grid_stops_at_today(monkeypatch): + import tradingagents.backtest as bt + + monkeypatch.setattr(bt, "get_current_date", lambda: "2026-01-10") + assert iter_grid("2026-01-05", "2026-02-20", every_n_days=5) == ["2026-01-05", "2026-01-10"] + + +@pytest.mark.unit +def test_grid_rejects_a_non_canonical_date(): + with pytest.raises(ValueError, match="YYYY-MM-DD"): + iter_grid("2026-1-5", "2026-01-20") + + +class _FakeGraph: + """Stands in for TradingAgentsGraph, writing to the log the harness gave it.""" + + instances: list = [] + fail_on: set = set() + + def __init__(self, selected_analysts=None, config=None, **kw): + self.analysts = list(selected_analysts) if selected_analysts else None + self.config = config + self.memory_log = TradingMemoryLog(config) + self.calls = [] + self.settled = [] + _FakeGraph.instances.append(self) + + def propagate(self, ticker, trade_date, asset_type="stock", portfolio=None): + self.calls.append((ticker, trade_date)) + if (ticker, trade_date) in _FakeGraph.fail_on: + raise RuntimeError("vendor exploded") + self.memory_log.store_decision(ticker, trade_date, DECISION) + return {"final_trade_decision": DECISION}, "Buy" + + def settle_pending(self, ticker): + self.settled.append(ticker) + + +@pytest.fixture(autouse=True) +def _fake_graph(monkeypatch, tmp_path): + import tradingagents.backtest as bt + + _FakeGraph.instances = [] + _FakeGraph.fail_on = set() + monkeypatch.setattr(bt, "TradingAgentsGraph", _FakeGraph) + return _FakeGraph + + +def _config(tmp_path): + return {"results_dir": str(tmp_path / "results"), + "memory_log_path": str(tmp_path / "live_trading_memory.md")} + + +@pytest.mark.unit +def test_the_live_decision_log_is_never_written(tmp_path): + config = _config(tmp_path) + result = run_backtest(["NVDA"], ["2026-01-05", "2026-01-12"], config) + + assert not (tmp_path / "live_trading_memory.md").exists() + assert result.log_path.exists() and result.cells_run == 2 + + +@pytest.mark.unit +def test_a_cell_already_in_the_log_is_not_run_again(tmp_path): + config = _config(tmp_path) + first = run_backtest(["NVDA"], ["2026-01-05"], config) + + again = run_backtest(["NVDA"], ["2026-01-05", "2026-01-12"], config, run_id=first.run_id) + + assert again.cells_run == 1 and again.skipped == 1 + assert _FakeGraph.instances[-1].calls == [("NVDA", "2026-01-12")] + + +@pytest.mark.unit +def test_every_ticker_is_settled_after_the_grid(tmp_path): + """Settlement runs at the start of the next same-ticker run, so the last + date of each ticker would stay pending without an explicit pass.""" + run_backtest(["NVDA", "AAPL"], ["2026-01-05", "2026-01-12"], _config(tmp_path)) + assert sorted(_FakeGraph.instances[-1].settled) == ["AAPL", "NVDA"] + + +@pytest.mark.unit +def test_a_failed_cell_does_not_abort_the_sweep(tmp_path): + _FakeGraph.fail_on = {("NVDA", "2026-01-05")} + result = run_backtest(["NVDA"], ["2026-01-05", "2026-01-12"], _config(tmp_path)) + + assert result.cells_run == 1 + assert result.failures == [("NVDA", "2026-01-05", "vendor exploded")] + + +# --- reading the result ------------------------------------------------------ + +def _log_with(tmp_path, rows): + log = TradingMemoryLog({"memory_log_path": str(tmp_path / "m.md")}) + for ticker, date, decision, outcome in rows: + log.store_decision(ticker, date, decision) + if outcome is not None: + log.update_with_outcome(ticker, date, outcome[0], outcome[1], 5, "note", "2026-02-01") + return log + + +@pytest.mark.unit +def test_summary_scores_resolved_cells_and_keeps_pending_out_of_the_average(tmp_path): + log = _log_with(tmp_path, [ + ("NVDA", "2026-01-05", "Rating: Buy\n\nx", (0.10, 0.04)), + ("NVDA", "2026-01-12", "Rating: Buy\n\nx", (-0.02, -0.02)), + ("AAPL", "2026-01-05", "Rating: Sell\n\nx", None), + ]) + + summary = summarize(log) + + assert summary.resolved == 2 and summary.pending == 1 + buys = summary.by_rating["Buy"] + assert buys.count == 2 and buys.hit_rate == 0.5 and round(buys.mean_alpha, 4) == 0.01 + assert "Sell" not in summary.by_rating # unsettled: nothing to score yet + + +@pytest.mark.unit +def test_summary_states_what_it_cannot_prove(tmp_path): + text = summarize(_log_with(tmp_path, [("NVDA", "2026-01-05", DECISION, (0.1, 0.05))])).render() + assert "not archived" in text + assert "one" in text.lower() and "sampl" in text.lower() + + +@pytest.mark.unit +def test_the_analyst_set_under_test_is_the_one_that_runs(tmp_path): + """A backtest of a two-analyst setup must not silently run four.""" + run_backtest(["NVDA"], ["2026-01-05"], _config(tmp_path), selected_analysts=["market", "news"]) + assert _FakeGraph.instances[-1].analysts == ["market", "news"] + + +@pytest.mark.unit +def test_a_run_id_cannot_escape_the_results_directory(tmp_path): + """run_id becomes a path segment, so it is validated like a ticker is.""" + with pytest.raises(ValueError): + run_backtest(["NVDA"], ["2026-01-05"], _config(tmp_path), run_id="../../escaped") + with pytest.raises(ValueError): + run_backtest(["NVDA"], ["2026-01-05"], _config(tmp_path), run_id="/etc/cron.d/x") + + +@pytest.mark.unit +def test_a_failed_settlement_does_not_lose_the_remaining_tickers(tmp_path, monkeypatch): + """Settlement reflects with an LLM, so it can fail; the sweep still returns + its result and every other ticker still gets settled.""" + settled = [] + + def _settle(self, ticker): + if ticker == "NVDA": + raise RuntimeError("reflector timed out") + settled.append(ticker) + + monkeypatch.setattr(_FakeGraph, "settle_pending", _settle, raising=False) + result = run_backtest(["NVDA", "AAPL"], ["2026-01-05"], _config(tmp_path)) + + assert result.cells_run == 2 + assert settled == ["AAPL"] + assert result.settlement_failures == [("NVDA", "reflector timed out")] diff --git a/tradingagents/backtest.py b/tradingagents/backtest.py new file mode 100644 index 000000000..6557f3f96 --- /dev/null +++ b/tradingagents/backtest.py @@ -0,0 +1,176 @@ +"""Run the graph over a grid of tickers and dates, and score what came back. + +One run yields one decision, so it cannot say whether the system decides well. +This runs the same machinery over many (ticker, date) cells and reads the +aggregate. The decision log is the results table: every run already records its +rating and later settles it with realized and alpha return against the +instrument's regional benchmark, so there is nothing to record separately. + +Scope: this evaluates decision quality. It is not a portfolio simulator, and +must not grow one. Turning a rating into a filled order needs a quantity, a fill +price and a cash ledger, none of which the system has; inventing them here would +put an execution model behind an evaluation tool. Cells are therefore +independent, and a portfolio, when given, is the same standing book for every +cell rather than a position carried forward. +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass, field +from datetime import datetime, timedelta +from pathlib import Path + +from tradingagents.agents.utils.memory import TradingMemoryLog +from tradingagents.dataflows.utils import get_current_date, safe_ticker_component +from tradingagents.graph.trading_graph import TradingAgentsGraph + +logger = logging.getLogger(__name__) + + +def iter_grid(start_date: str, end_date: str, every_n_days: int = 1) -> list[str]: + """Analysis dates from ``start_date``, never past today. + + A future date has no outcome to settle against, and the graph rejects one, so + the grid stops at the present rather than producing cells that cannot score. + """ + start, end = _canonical(start_date), _canonical(end_date) + if every_n_days < 1: + raise ValueError("every_n_days must be at least 1") + + last = min(end, datetime.strptime(get_current_date(), "%Y-%m-%d")) + dates, cursor = [], start + while cursor <= last: + dates.append(cursor.strftime("%Y-%m-%d")) + cursor += timedelta(days=every_n_days) + return dates + + +def _canonical(date: str) -> datetime: + """Parse a grid bound, rejecting anything the run date would also reject.""" + try: + parsed = datetime.strptime(str(date), "%Y-%m-%d") + except (TypeError, ValueError) as exc: + raise ValueError(f"grid dates must be in YYYY-MM-DD format, got {date!r}") from exc + if parsed.strftime("%Y-%m-%d") != str(date): + raise ValueError(f"grid dates must be in YYYY-MM-DD format, got {date!r}") + return parsed + + +def _alpha(entry: dict) -> float | None: + """Alpha return of a settled entry, or None when it has not settled. + + The log stores it as a percentage rounded to one decimal, so aggregates here + are accurate to 0.1 of a percentage point, not to the raw quote. + """ + text = (entry.get("alpha") or "").strip().rstrip("%") + try: + return float(text) / 100 + except ValueError: + return None + + +@dataclass +class BacktestResult: + run_id: str + log_path: Path + cells_run: int = 0 + skipped: int = 0 + failures: list[tuple[str, str, str]] = field(default_factory=list) + settlement_failures: list[tuple[str, str]] = field(default_factory=list) + + +@dataclass +class RatingScore: + count: int + hit_rate: float + mean_alpha: float + + +@dataclass +class BacktestSummary: + resolved: int + pending: int + by_rating: dict[str, RatingScore] + + def render(self) -> str: + lines = [f"Resolved cells: {self.resolved} ยท pending: {self.pending}"] + for rating, score in self.by_rating.items(): + lines.append( + f"- {rating}: n={score.count}, beat the benchmark " + f"{score.hit_rate:.0%}, mean alpha {score.mean_alpha:+.2%}" + ) + lines += [ + "", + "Pending cells are not scored above; re-run to settle them.", + "One model sampling per cell, and text feeds are not archived, so " + "these figures are indicative rather than repeatable.", + ] + return "\n".join(lines) + + +def run_backtest( + tickers: list[str], + dates: list[str], + config: dict, + asset_type: str = "stock", + portfolio=None, + selected_analysts=("market", "social", "news", "fundamentals"), + run_id: str | None = None, +) -> BacktestResult: + """Analyze every ticker on every date, into a decision log of this run's own. + + The live log stays untouched: a sweep would otherwise flood the context that + real runs read back. Cells already in this run's log are skipped, so an + interrupted sweep resumes by being run again. + """ + # run_id becomes a path segment, so it is validated like a ticker: an + # absolute or dotted value would otherwise place the run outside results_dir. + run_id = safe_ticker_component(run_id or datetime.now().strftime("%Y%m%d_%H%M%S")) + run_dir = Path(config["results_dir"]) / "backtest" / run_id + run_dir.mkdir(parents=True, exist_ok=True) + run_config = {**config, "results_dir": str(run_dir), + "memory_log_path": str(run_dir / "trading_memory.md")} + + graph = TradingAgentsGraph(selected_analysts, config=run_config) + result = BacktestResult(run_id=run_id, log_path=Path(run_config["memory_log_path"])) + done = {(e["ticker"], e["date"]) for e in graph.memory_log.load_entries()} + + for ticker in tickers: + for date in dates: + if (ticker, date) in done: + result.skipped += 1 + continue + try: + graph.propagate(ticker, date, asset_type, portfolio=portfolio) + result.cells_run += 1 + except Exception as exc: # one unreachable vendor must not end the sweep + logger.warning("Backtest cell %s %s failed: %s", ticker, date, exc) + result.failures.append((ticker, date, str(exc))) + + # Settlement runs at the start of the next run for a ticker, so each ticker's + # last cell would stay pending without this pass. + for ticker in tickers: + try: + graph.settle_pending(ticker) + except Exception as exc: # reflection calls an LLM; one failure is not the sweep's + logger.warning("Settling %s failed: %s", ticker, exc) + result.settlement_failures.append((ticker, str(exc))) + return result + + +def summarize(memory_log: TradingMemoryLog) -> BacktestSummary: + """Score the settled decisions in a log, by rating.""" + entries = memory_log.load_entries() + resolved = [(e, _alpha(e)) for e in entries if not e["pending"]] + resolved = [(e, a) for e, a in resolved if a is not None] + by_rating: dict[str, RatingScore] = {} + for rating in dict.fromkeys(e["rating"] for e, _ in resolved): + alphas = [a for e, a in resolved if e["rating"] == rating] + by_rating[rating] = RatingScore( + count=len(alphas), + hit_rate=sum(a > 0 for a in alphas) / len(alphas), + mean_alpha=sum(alphas) / len(alphas), + ) + return BacktestSummary(resolved=len(resolved), pending=len(entries) - len(resolved), + by_rating=by_rating)