From 11dab3b0fea5fc4a691bc08b94453020163051ec Mon Sep 17 00:00:00 2001 From: Yijia-Xiao Date: Thu, 1 Oct 2026 05:56:16 +0000 Subject: [PATCH] fix(memory): settle each decision once, and keep each outcome as soon as it is known - a settlement pass holds the log's settlement lock, so runs sharing a log never pay for the same reflection twice; only outcomes actually written count as settled, and a run's Memory Log step does not wait for another run's pass - each outcome is written when its reflection is done, so an interrupted pass keeps what it finished - any failure to price a decision is reported as a failed settlement, not taken for a window that has yet to trade --- tests/test_cache_writes.py | 40 ++++++- tests/test_memory_log.py | 112 +++++++++++++++++- tradingagents/dataflows/files.py | 29 +++-- tradingagents/graph/trading_graph.py | 8 +- tradingagents/memory/log.py | 23 +++- tradingagents/memory/settlement.py | 164 +++++++++++++-------------- 6 files changed, 270 insertions(+), 106 deletions(-) diff --git a/tests/test_cache_writes.py b/tests/test_cache_writes.py index a9bf7d94e..b4800faeb 100644 --- a/tests/test_cache_writes.py +++ b/tests/test_cache_writes.py @@ -93,12 +93,12 @@ def _fake_msvcrt(results): def locking(fd, mode, nbytes): calls.append(mode) - if mode == 2 and results: # LK_LOCK; unlocking always succeeds + if mode in (1, 2) and results: # LK_NBLCK, LK_LOCK; unlocking always succeeds error = results.pop(0) if error is not None: raise error - return types.SimpleNamespace(LK_LOCK=2, LK_UNLCK=0, locking=locking), calls + return types.SimpleNamespace(LK_LOCK=2, LK_NBLCK=1, LK_UNLCK=0, locking=locking), calls @pytest.mark.unit @@ -126,3 +126,39 @@ def test_on_windows_a_lock_that_cannot_be_taken_is_reported_not_retried_forever( with pytest.raises(OSError, match="bad handle"), open(tmp_path / "log.md.lock", "a+b") as handle, \ files._hold_windows(handle): pass + + +@pytest.mark.unit +def test_a_lock_held_elsewhere_is_reported_when_not_waiting(tmp_path): + import threading + + path = tmp_path / "log.md" + held, release = threading.Event(), threading.Event() + + def holder(): + with files.locked(path): + held.set() + release.wait(5) + + worker = threading.Thread(target=holder) + worker.start() + held.wait(5) + with files.locked(path, wait=False) as acquired: + assert acquired is False + release.set() + worker.join() + with files.locked(path, wait=False) as acquired: + assert acquired is True + + +@pytest.mark.unit +def test_on_windows_a_held_lock_is_reported_when_not_waiting(tmp_path, monkeypatch): + import errno + import sys + + fake, calls = _fake_msvcrt([OSError(errno.EACCES, "held")]) + monkeypatch.setitem(sys.modules, "msvcrt", fake) + + with open(tmp_path / "log.md.lock", "a+b") as handle, files._hold_windows(handle, wait=False) as acquired: + assert acquired is False + assert calls == [1] # one try, no unlock of a lock it never held diff --git a/tests/test_memory_log.py b/tests/test_memory_log.py index 9d1ae0f1f..375be979e 100644 --- a/tests/test_memory_log.py +++ b/tests/test_memory_log.py @@ -1183,8 +1183,118 @@ def test_the_graph_settles_every_ticker_under_its_own_config(tmp_path, monkeypat 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"]) + lambda log, reflector, config, wait=True: 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"] + + +@pytest.mark.unit +class TestSettlementOnceAndDurably: + """Settlement is paid for in reflection calls, so it runs once per decision and + keeps what it finished; what it could not do is reported (#1428, #1445).""" + + def _log(self, tmp_path, monkeypatch, *dates): + log = make_log(tmp_path) + for date in dates: + log.store_decision("AAPL", date, DECISION_BUY) + series = pd.Series([100.0 + i for i in range(40)], index=pd.bdate_range("2026-01-02", periods=40)) + monkeypatch.setattr(settlement, "get_closes", _closes({"AAPL": series, "SPY": series})) + return log + + def test_two_runs_settling_at_once_reflect_once(self, tmp_path, monkeypatch): + import threading + import time + + self._log(tmp_path, monkeypatch, "2026-01-05") + calls = [] + + class _Slow: + def reflect_on_final_decision(self, **kwargs): + calls.append(1) + time.sleep(0.2) + return "Held up." + + results = [] + threads = [threading.Thread(target=lambda: results.append( + settlement.settle_pending("AAPL", make_log(tmp_path), _Slow(), {"holding_period_days": 5}))) + for _ in range(2)] + [t.start() for t in threads] + [t.join() for t in threads] + + assert len(calls) == 1 + assert sorted(len(r.settled) for r in results) == [0, 1] + + def test_an_interrupted_pass_keeps_the_outcomes_it_finished(self, tmp_path, monkeypatch): + log = self._log(tmp_path, monkeypatch, "2026-01-05", "2026-01-12") + + class _Interrupted: + calls = 0 + + def reflect_on_final_decision(self, **kwargs): + _Interrupted.calls += 1 + if _Interrupted.calls == 2: + raise KeyboardInterrupt + return "Held up." + + with pytest.raises(KeyboardInterrupt): + settlement.settle_pending("AAPL", log, _Interrupted(), {"holding_period_days": 5}) + + assert [e["date"] for e in log.get_pending_entries()] == ["2026-01-12"] + + def test_a_price_request_that_fails_is_reported_not_taken_for_not_due(self, tmp_path, monkeypatch): + from tradingagents.dataflows.errors import VendorUnavailableError + + log = self._log(tmp_path, monkeypatch, "2026-01-05") + + def down(*args, **kwargs): + raise VendorUnavailableError("Yahoo Finance request failed: TimeoutError") + + monkeypatch.setattr(settlement, "get_closes", down) + + result = settlement.settle_pending("AAPL", log, object(), {"holding_period_days": 5}) + + assert result.settled == [] and "prices unavailable" in result.failed[0][2] + + +@pytest.mark.unit +def test_any_failure_to_price_a_decision_is_reported(tmp_path, monkeypatch): + log = make_log(tmp_path) + log.store_decision("AAPL", "2026-01-05", DECISION_BUY) + + def broken(*args, **kwargs): + raise RuntimeError("vendor outage") + + monkeypatch.setattr(settlement, "get_closes", broken) + + result = settlement.settle_pending("AAPL", log, object(), {"holding_period_days": 5}) + + assert result.settled == [] and "vendor outage" in result.failed[0][2] + + +@pytest.mark.unit +def test_a_run_does_not_wait_on_another_settlement_pass(tmp_path): + """The other pass settles the same log; a run's step goes on without waiting.""" + import threading + import time + + log = make_log(tmp_path) + log.store_decision("AAPL", "2026-01-05", DECISION_BUY) + held, release = threading.Event(), threading.Event() + + def other_pass(): + with make_log(tmp_path).settling(): + held.set() + release.wait(5) + + worker = threading.Thread(target=other_pass) + worker.start() + held.wait(5) + started = time.monotonic() + result = settlement.settle_all_pending(log, object(), {"holding_period_days": 5}, wait=False) + elapsed = time.monotonic() - started + release.set() + worker.join() + + assert elapsed < 1 and result.settled == [] and result.failed == [] diff --git a/tradingagents/dataflows/files.py b/tradingagents/dataflows/files.py index 0c7408919..3c77fe07e 100644 --- a/tradingagents/dataflows/files.py +++ b/tradingagents/dataflows/files.py @@ -34,46 +34,55 @@ def replace_file(path, write: Callable[[str], None]) -> None: @contextmanager -def locked(path) -> Iterator[None]: +def locked(path, wait: bool = True) -> Iterator[bool]: """Hold ``path`` for one writer at a time, across threads and processes. For a file read, changed and written back: without the lock, two writers each read the same text and the second write drops the first's change. The - lock is taken on a ``.lock`` file beside ``path``. + lock is taken on a ``.lock`` file beside ``path``. Yields whether it is + held: with ``wait=False`` a lock someone else holds is not waited for, and + the block runs holding nothing. """ with open(f"{path}.lock", "a+b") as handle: hold = _hold_windows if os.name == "nt" else _hold_posix - with hold(handle): - yield + with hold(handle, wait) as acquired: + yield acquired @contextmanager -def _hold_posix(handle) -> Iterator[None]: +def _hold_posix(handle, wait: bool = True) -> Iterator[bool]: import fcntl - fcntl.flock(handle.fileno(), fcntl.LOCK_EX) try: - yield + fcntl.flock(handle.fileno(), fcntl.LOCK_EX if wait else fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError: + yield False + return + try: + yield True finally: fcntl.flock(handle.fileno(), fcntl.LOCK_UN) @contextmanager -def _hold_windows(handle) -> Iterator[None]: +def _hold_windows(handle, wait: bool = True) -> Iterator[bool]: import msvcrt handle.seek(0) while True: try: - msvcrt.locking(handle.fileno(), msvcrt.LK_LOCK, 1) + msvcrt.locking(handle.fileno(), msvcrt.LK_LOCK if wait else msvcrt.LK_NBLCK, 1) break except OSError as exc: + if not wait and exc.errno in (errno.EACCES, errno.EDEADLOCK): + yield False # another writer holds it + return # LK_LOCK gives up after about ten seconds of another writer's hold; # any other failure is not a wait. if exc.errno != errno.EDEADLOCK: raise try: - yield + yield True finally: handle.seek(0) msvcrt.locking(handle.fileno(), msvcrt.LK_UNLCK, 1) diff --git a/tradingagents/graph/trading_graph.py b/tradingagents/graph/trading_graph.py index 3ee417aa1..a265ff88f 100644 --- a/tradingagents/graph/trading_graph.py +++ b/tradingagents/graph/trading_graph.py @@ -317,7 +317,8 @@ class TradingAgentsGraph: """ note = "" try: - done = self.settle_all_pending() + # Another run settling the same log does the work; this one goes on. + done = self.settle_all_pending(wait=False) if done.failed: note = (f"{len(done.failed)} past decision(s) could not be settled this run " "and stay pending.") @@ -344,14 +345,15 @@ class TradingAgentsGraph: with run_config(self.config): return settlement.settle_pending(company_name, self.memory_log, self.reflector, self.config) - def settle_all_pending(self) -> settlement.Settlement: + def settle_all_pending(self, wait: bool = True) -> 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 ``wait=False`` a pass already running on the same log is not waited for. """ with run_config(self.config): - return settlement.settle_all_pending(self.memory_log, self.reflector, self.config) + return settlement.settle_all_pending(self.memory_log, self.reflector, self.config, wait=wait) 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/log.py b/tradingagents/memory/log.py index dc1622ccf..a8ce85a8a 100644 --- a/tradingagents/memory/log.py +++ b/tradingagents/memory/log.py @@ -1,5 +1,6 @@ """The memory log: an append-only markdown record of each decision and, once settled, its outcome.""" +import contextlib import re from pathlib import Path @@ -58,6 +59,18 @@ class TradingMemoryLog: with open(self._log_path, "a", encoding="utf-8") as f: f.write(entry) + def settling(self, wait: bool = True): + """Hold the log for one settlement pass at a time, across threads and processes. + + A pass reads the pending entries, asks the model for a reflection on + each and writes it; two passes at once would pay for the same + reflections. A separate lock from the one each write takes. Yields + whether it is held; with ``wait=False`` another pass is not waited for. + """ + if not self._log_path: + return contextlib.nullcontext(True) + return locked(f"{self._log_path}.settle", wait=wait) + # --- Read --- def load_entries(self) -> list[dict]: @@ -127,16 +140,19 @@ class TradingMemoryLog: holding_days: int, reflection: str, resolution_date: str | None = None, - ) -> None: + ) -> bool: """Replace pending tag and append REFLECTION section using atomic write. + Returns whether an entry was updated: False when the decision was no + longer pending (another pass settled it first). + Finds the first pending entry matching (trade_date, ticker), updates its tag with return figures (and the ``resolution_date`` the outcome became known), and appends a REFLECTION section. Uses a temp-file + os.replace() so a crash mid-write never corrupts the log. """ if not self._log_path or not self._log_path.exists(): - return + return False with locked(self._log_path): text = self._log_path.read_text(encoding="utf-8") blocks = text.split(self._SEPARATOR) @@ -175,13 +191,14 @@ class TradingMemoryLog: new_blocks.append(block) if not updated: - return + return False new_blocks = self._apply_rotation(new_blocks) new_text = self._SEPARATOR.join(new_blocks) tmp_path = self._log_path.with_suffix(".tmp") tmp_path.write_text(new_text, encoding="utf-8") tmp_path.replace(self._log_path) + return True def batch_update_with_outcomes(self, updates: list[dict]) -> None: """Apply multiple outcome updates in a single read + atomic write. diff --git a/tradingagents/memory/settlement.py b/tradingagents/memory/settlement.py index 69e39c1fe..032b3c68b 100644 --- a/tradingagents/memory/settlement.py +++ b/tradingagents/memory/settlement.py @@ -59,50 +59,43 @@ def fetch_returns( 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. + or ``(None, None, None, None)`` when the full holding window has not traded + yet (#1169). A failure to get or read the prices is raised, for the caller + to report. """ - 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") + 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 = _by_day(get_closes(ticker, trade_date, end_str)) - # From a week earlier, so the benchmark has a close on or before entry. - bench_start = (start - timedelta(days=7)).strftime("%Y-%m-%d") - bench = _by_day(get_closes(benchmark, bench_start, end_str)) + # Closes for the instrument the analysis priced (XAUUSD -> GC=F, #984). + stock = _by_day(get_closes(ticker, trade_date, end_str)) + # From a week earlier, so the benchmark has a close on or before entry. + bench_start = (start - timedelta(days=7)).strftime("%Y-%m-%d") + bench = _by_day(get_closes(benchmark, bench_start, end_str)) - # Require the full holding window to have traded. A rerun before it has - # leaves the entry pending to retry next run, rather than settling on a - # premature partial return (#1169). - if len(stock) <= holding_days: - return None, None, None, None - entry, exit_ = stock.index[0], stock.index[holding_days] - # The window is the stock's own sessions; the benchmark is valued at its - # last close by each end, so both returns span the same dates even when - # the calendars differ (a coin trades at weekends, an index does not). - # It must have traded through the exit, or its close there may yet move. - if bench.empty or bench.index[0] > entry or bench.index[-1] < exit_: - return None, None, None, None - - raw = float((stock.iloc[holding_days] - stock.iloc[0]) / stock.iloc[0]) - bench_entry, bench_exit = bench.asof(entry), bench.asof(exit_) - bench_ret = float((bench_exit - bench_entry) / bench_entry) - alpha = raw - bench_ret - # Every close used is known by the exit: the point-in-time cutoff for - # injecting the lesson (#1251). - resolution_date = exit_.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, - ) + # Require the full holding window to have traded. A rerun before it has + # leaves the entry pending to retry next run, rather than settling on a + # premature partial return (#1169). + if len(stock) <= holding_days: return None, None, None, None + entry, exit_ = stock.index[0], stock.index[holding_days] + # The window is the stock's own sessions; the benchmark is valued at its + # last close by each end, so both returns span the same dates even when + # the calendars differ (a coin trades at weekends, an index does not). + # It must have traded through the exit, or its close there may yet move. + if bench.empty or bench.index[0] > entry or bench.index[-1] < exit_: + return None, None, None, None + + raw = float((stock.iloc[holding_days] - stock.iloc[0]) / stock.iloc[0]) + bench_entry, bench_exit = bench.asof(entry), bench.asof(exit_) + bench_ret = float((bench_exit - bench_entry) / bench_entry) + alpha = raw - bench_ret + # Every close used is known by the exit: the point-in-time cutoff for + # injecting the lesson (#1251). + resolution_date = exit_.strftime("%Y-%m-%d") + return raw, alpha, holding_days, resolution_date @dataclass @@ -113,59 +106,56 @@ class Settlement: failed: list[tuple[str, str, str]] = field(default_factory=list) # (ticker, date, reason) -def settle_pending(ticker: str, memory_log, reflector, config: dict) -> Settlement: +def settle_pending(ticker: str, memory_log, reflector, config: dict, wait: bool = True) -> 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); - an entry whose reflection fails stays pending and is reported as failed. + One pass at a time per log (``memory_log.settling()``), so runs sharing a log + never pay for the same reflection twice. Each outcome is written as soon as + its reflection is done, so an interrupted pass keeps what it finished. + Entries whose window has not traded yet are skipped; an entry whose prices + or reflection could not be had stays pending and is reported as failed. + With ``wait=False`` a pass already running elsewhere is left to do the work. """ result = Settlement() - pending = [e for e in memory_log.get_pending_entries() if e["ticker"] == ticker] - if not pending: - return result - - 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: 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, - "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) - result.settled.extend((ticker, u["trade_date"]) for u in updates) + with memory_log.settling(wait=wait) as held: + if not held: + return result + pending = [e for e in memory_log.get_pending_entries() if e["ticker"] == ticker] + if not pending: + return result + benchmark = resolve_benchmark(ticker, config) + for entry in pending: + try: + raw, alpha, days, resolution_date = fetch_returns( + ticker, entry["date"], config.get("holding_period_days", 5), benchmark=benchmark, + ) + except Exception as exc: + logger.warning("Prices for settling %s on %s were unavailable: %s", ticker, entry["date"], exc) + result.failed.append((ticker, entry["date"], f"prices unavailable: {exc}")) + continue + if raw is None: + continue # the holding window has not traded 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: 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 + if memory_log.update_with_outcome(ticker, entry["date"], raw, alpha, days, reflection, + resolution_date=resolution_date): + result.settled.append((ticker, entry["date"])) return result -def settle_all_pending(memory_log, reflector, config: dict) -> Settlement: +def settle_all_pending(memory_log, reflector, config: dict, wait: bool = True) -> 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 @@ -173,7 +163,7 @@ def settle_all_pending(memory_log, reflector, config: dict) -> Settlement: """ 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) + done = settle_pending(ticker, memory_log, reflector, config, wait=wait) result.settled.extend(done.settled) result.failed.extend(done.failed) return result