From fc1ab1db07f1b4056bc3f62118b0bae96b3132c6 Mon Sep 17 00:00:00 2001 From: Yijia-Xiao Date: Thu, 24 Sep 2026 20:49:13 +0000 Subject: [PATCH] fix(dataflows): write cache files whole, through a temp file of their own - the OHLCV cache and the SEC EDGAR cache go through dataflows.files.replace_file: a reader sees the old file or the new one, and concurrent writers never share a temp file --- tests/test_cache_writes.py | 84 +++++++++++++++++++ tradingagents/dataflows/files.py | 31 +++++++ tradingagents/dataflows/vendors/sec_edgar.py | 5 +- .../dataflows/vendors/yahoo/ohlcv.py | 3 +- 4 files changed, 119 insertions(+), 4 deletions(-) create mode 100644 tests/test_cache_writes.py create mode 100644 tradingagents/dataflows/files.py diff --git a/tests/test_cache_writes.py b/tests/test_cache_writes.py new file mode 100644 index 000000000..195809038 --- /dev/null +++ b/tests/test_cache_writes.py @@ -0,0 +1,84 @@ +"""Cache files are replaced whole: a reader never sees a half-written file, and +concurrent writers (tool calls run in parallel) never share a temp file.""" + +import os +from pathlib import Path + +import pandas as pd +import pytest + +from tradingagents.dataflows.vendors import sec_edgar +from tradingagents.dataflows.vendors.yahoo import ohlcv + + +def _recording_replace(monkeypatch): + moves = [] + real = os.replace + + def replace(src, dst): + moves.append((Path(src), Path(dst))) + real(src, dst) + + monkeypatch.setattr(os, "replace", replace) + return moves + + +@pytest.mark.unit +def test_sec_edgar_writers_never_share_a_temp_file(monkeypatch, tmp_path): + monkeypatch.setattr(sec_edgar, "get_config", lambda: {"data_cache_dir": str(tmp_path)}) + monkeypatch.setattr(sec_edgar, "_fetch_json", lambda url: {"url": url}) + moves = _recording_replace(monkeypatch) + + sec_edgar._cached_json("https://example/a", "facts.json") + (tmp_path / "sec_edgar" / "facts.json").unlink() + sec_edgar._cached_json("https://example/a", "facts.json") + + temps = [src for src, _ in moves] + assert len(set(temps)) == 2 + assert all(src.parent == dst.parent for src, dst in moves) + + +@pytest.mark.unit +def test_the_price_cache_is_replaced_whole(monkeypatch, tmp_path): + frame = pd.DataFrame({"Date": pd.bdate_range("2026-01-02", periods=3), "Open": 1.0, "High": 1.0, + "Low": 1.0, "Close": 1.0, "Volume": 1}) + monkeypatch.setattr(ohlcv, "get_config", lambda: {"data_cache_dir": str(tmp_path)}) + monkeypatch.setattr(ohlcv.yf, "Ticker", lambda s: type("T", (), {"history": lambda self, **k: frame.set_index("Date")})()) + moves = _recording_replace(monkeypatch) + + ohlcv.load_ohlcv("AAPL", "2026-01-06") + + assert [dst.name for _, dst in moves] == ["AAPL-YFin-data.csv"] + assert moves[0][0].parent == tmp_path and moves[0][0] != moves[0][1] + assert not list(tmp_path.glob("*.tmp*")) + + +@pytest.mark.unit +def test_a_replaced_file_keeps_the_usual_permissions(tmp_path): + from tradingagents.dataflows.files import replace_file + + plain = tmp_path / "plain.txt" + plain.write_text("x") + replaced = tmp_path / "replaced.txt" + replace_file(replaced, lambda temp: Path(temp).write_text("x")) + + assert replaced.stat().st_mode & 0o777 == plain.stat().st_mode & 0o777 + + +@pytest.mark.unit +def test_a_file_held_open_elsewhere_keeps_its_old_content(monkeypatch, tmp_path): + """On Windows a file another reader has open cannot be replaced; the cache + write is skipped rather than failing the call that produced the data.""" + from tradingagents.dataflows import files + + target = tmp_path / "cache.csv" + target.write_text("old") + + def locked(src, dst): + raise PermissionError("in use") + + monkeypatch.setattr(files.os, "replace", locked) + files.replace_file(target, lambda temp: Path(temp).write_text("new")) + + assert target.read_text() == "old" + assert list(tmp_path.iterdir()) == [target] diff --git a/tradingagents/dataflows/files.py b/tradingagents/dataflows/files.py new file mode 100644 index 000000000..98e32afbd --- /dev/null +++ b/tradingagents/dataflows/files.py @@ -0,0 +1,31 @@ +"""Cache files written whole, safe under concurrent writers.""" + +import logging +import os +import uuid +from collections.abc import Callable +from pathlib import Path + +logger = logging.getLogger(__name__) + + +def replace_file(path, write: Callable[[str], None]) -> None: + """Write ``path`` through a uniquely named temp file beside it, then move it into place. + + A reader sees the old file or the new one, never a partial write, and two + writers of the same path (tool calls run concurrently) never share a temp file. + ``write`` receives the temp path and creates the file, so it gets the usual + permissions. The file is a cache: where another reader holds it open (Windows), + the old one stays and the write is skipped. + """ + path = Path(path) + temp = path.with_name(f"{path.name}.{uuid.uuid4().hex}.tmp") + try: + write(str(temp)) + os.replace(temp, path) + except PermissionError as exc: + temp.unlink(missing_ok=True) + logger.warning("Kept the cached %s; it is in use (%s)", path.name, exc) + except BaseException: + temp.unlink(missing_ok=True) + raise diff --git a/tradingagents/dataflows/vendors/sec_edgar.py b/tradingagents/dataflows/vendors/sec_edgar.py index 76a9ce2ab..f9ccbadf2 100644 --- a/tradingagents/dataflows/vendors/sec_edgar.py +++ b/tradingagents/dataflows/vendors/sec_edgar.py @@ -29,6 +29,7 @@ import requests from tradingagents import __version__ from tradingagents.dataflows.config import get_config from tradingagents.dataflows.errors import NoMarketDataError, VendorRateLimitError +from tradingagents.dataflows.files import replace_file logger = logging.getLogger(__name__) @@ -122,9 +123,7 @@ def _cached_json(url: str, name: str) -> dict: pass # a truncated file is a miss, not a failure data = _fetch_json(url) path.parent.mkdir(parents=True, exist_ok=True) - temp = path.with_suffix(".tmp") - temp.write_text(json.dumps(data), encoding="utf-8") - os.replace(temp, path) + replace_file(path, lambda temp: Path(temp).write_text(json.dumps(data), encoding="utf-8")) return data diff --git a/tradingagents/dataflows/vendors/yahoo/ohlcv.py b/tradingagents/dataflows/vendors/yahoo/ohlcv.py index 87029e777..ddcd58609 100644 --- a/tradingagents/dataflows/vendors/yahoo/ohlcv.py +++ b/tradingagents/dataflows/vendors/yahoo/ohlcv.py @@ -8,6 +8,7 @@ from yfinance.exceptions import YFRateLimitError from tradingagents.dataflows.config import get_config from tradingagents.dataflows.errors import NoMarketDataError, VendorRateLimitError +from tradingagents.dataflows.files import replace_file from tradingagents.dataflows.net import vendor_reachable from tradingagents.dataflows.symbols import normalize_symbol, safe_ticker_component @@ -266,7 +267,7 @@ def load_ohlcv(symbol: str, as_of_date: str, fill_gaps: bool = True) -> pd.DataF # Only cache real data — never persist an empty frame. if downloaded.empty or "Close" not in downloaded.columns: raise_for_empty(symbol, canonical, "price rows") - downloaded.to_csv(data_file, index=False, encoding="utf-8") + replace_file(data_file, lambda temp: downloaded.to_csv(temp, index=False, encoding="utf-8")) data = downloaded data = _clean_dataframe(data)