feat(memory): settle the pending decisions of every ticker on request (#1445)

- TradingAgentsGraph.settle_all_pending() settles each ticker with a due decision; settle_pending and it return the decisions settled and those that failed
This commit is contained in:
Yijia-Xiao
2026-10-03 21:27:43 +00:00
parent 97b79e08c3
commit 5d8de36f6d
4 changed files with 105 additions and 15 deletions
+2
View File
@@ -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`. 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
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. 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.
+55
View File
@@ -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") 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 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"]
+15 -6
View File
@@ -322,16 +322,25 @@ class TradingAgentsGraph:
portfolio_context=portfolio.render(company_name) if portfolio is not None else "", 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. """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 A run settles its ticker's earlier decisions as it starts, so the most
recent one stays pending until the next run for that ticker. A caller recent one stays pending until a later run. A caller that is done
that is done analyzing a ticker (a backtest sweep, a scheduled job) calls analyzing a ticker (a backtest sweep, a scheduled job) calls this to
this to settle it now. settle it now. Returns what was settled and what failed.
""" """
with run_config(self.config): 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): def record_decision(self, company_name, trade_date, final_state):
"""Record a finished run: its state log, and its decision in the memory log """Record a finished run: its state log, and its decision in the memory log
+33 -9
View File
@@ -2,6 +2,7 @@
it against its benchmark and record a reflection on it in the memory log.""" it against its benchmark and record a reflection on it in the memory log."""
import logging import logging
from dataclasses import dataclass, field
from datetime import datetime, timedelta from datetime import datetime, timedelta
from tradingagents.dataflows.symbols import normalize_symbol from tradingagents.dataflows.symbols import normalize_symbol
@@ -104,19 +105,26 @@ def fetch_returns(
return None, None, None, None 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. """Settle ``ticker``'s pending decisions whose holding window has traded.
Fetches returns for each same-ticker pending entry, generates reflections, 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. 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). 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.
Trade-off: only same-ticker entries are resolved per run. Entries for
other tickers accumulate until that ticker is run again.
""" """
result = Settlement()
pending = [e for e in memory_log.get_pending_entries() if e["ticker"] == ticker] pending = [e for e in memory_log.get_pending_entries() if e["ticker"] == ticker]
if not pending: if not pending:
return return result
benchmark = resolve_benchmark(ticker, config) benchmark = resolve_benchmark(ticker, config)
updates = [] updates = []
@@ -136,10 +144,10 @@ def settle_pending(ticker: str, memory_log, reflector, config: dict) -> None:
holding_days=days, holding_days=days,
) )
except Exception as exc: except Exception as exc:
# Reflection calls a provider, and this runs on the way into a # Reflection calls a provider: a transient failure leaves the entry
# new run: a transient failure leaves the entry pending for the # pending for the next pass rather than stopping the analysis.
# next one rather than stopping the analysis that was asked for.
logger.warning("Reflection failed for %s on %s: %s", ticker, entry["date"], exc) logger.warning("Reflection failed for %s on %s: %s", ticker, entry["date"], exc)
result.failed.append((ticker, entry["date"], f"reflection failed: {exc}"))
continue continue
updates.append({ updates.append({
"ticker": ticker, "ticker": ticker,
@@ -153,3 +161,19 @@ def settle_pending(ticker: str, memory_log, reflector, config: dict) -> None:
if updates: if updates:
memory_log.batch_update_with_outcomes(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