mirror of
https://github.com/TauricResearch/TradingAgents.git
synced 2026-09-19 11:15:24 +03:00
fix(dataflows): keep one OHLCV cache file per symbol (#1330)
- the cache file is keyed by symbol and serves only on the day it was written
This commit is contained in:
@@ -1,80 +1,72 @@
|
|||||||
"""Same-day OHLCV cache must not serve a stale snapshot all day (#1150).
|
"""The OHLCV cache: one file per symbol, fresh only on the day it was written.
|
||||||
|
|
||||||
The cache file is keyed per day, so a run started before the day's bar was final
|
A current-day request also refetches past a TTL, so a run started before the
|
||||||
would be reused by every later run, feeding a stale close into technical
|
day's bar was final is not served that snapshot all day (#1150). Keying the file
|
||||||
analysis. Two cases matter for a current-day request: the bar may be missing, or
|
by symbol rather than by day keeps the cache from growing a file per symbol per
|
||||||
present but still in progress (Yahoo publishes a partial daily candle intraday).
|
day (#1330).
|
||||||
Refresh is bounded by a TTL so repeated runs cannot hammer the vendor.
|
|
||||||
"""
|
"""
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
import os
|
import os
|
||||||
import time
|
|
||||||
|
|
||||||
import pandas as pd
|
import pandas as pd
|
||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
import tradingagents.dataflows.stockstats_utils as su
|
import tradingagents.dataflows.stockstats_utils as su
|
||||||
|
|
||||||
TODAY = pd.Timestamp("2026-07-18")
|
NOW = pd.Timestamp("2026-07-18 12:00")
|
||||||
STALE = su.OHLCV_CACHE_TTL_SECONDS + 60
|
STALE = su.OHLCV_CACHE_TTL_SECONDS + 60
|
||||||
|
|
||||||
|
|
||||||
def _write(tmp_path, name="cache.csv", age_seconds=0.0, last_date="2026-07-17"):
|
def _write(tmp_path, name="AAPL-YFin-data.csv", age_seconds=0.0, last_date="2026-07-17"):
|
||||||
f = tmp_path / name
|
f = tmp_path / name
|
||||||
pd.DataFrame({"Date": [last_date], "Close": [1.0]}).to_csv(f, index=False)
|
pd.DataFrame({"Date": [last_date], "Close": [100.0]}).to_csv(f, index=False)
|
||||||
if age_seconds:
|
written = NOW.timestamp() - age_seconds
|
||||||
old = time.time() - age_seconds
|
os.utime(f, (written, written))
|
||||||
os.utime(f, (old, old))
|
return f
|
||||||
return str(f)
|
|
||||||
|
|
||||||
|
def _load(tmp_path, monkeypatch, curr_date, download):
|
||||||
|
monkeypatch.setattr(su, "get_config", lambda: {"data_cache_dir": str(tmp_path)})
|
||||||
|
monkeypatch.setattr(su.pd.Timestamp, "today", staticmethod(lambda: NOW))
|
||||||
|
monkeypatch.setattr(su.yf, "download", download)
|
||||||
|
return su.load_ohlcv("AAPL", curr_date)
|
||||||
|
|
||||||
|
|
||||||
|
def _fail_download(*a, **k):
|
||||||
|
raise AssertionError("fresh cache must not refetch")
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.unit
|
@pytest.mark.unit
|
||||||
def test_current_day_cache_past_ttl_is_refreshed(tmp_path):
|
def test_current_day_cache_past_ttl_is_not_fresh(tmp_path):
|
||||||
# Bar missing (rows stop at yesterday) and file older than the TTL -> refetch.
|
# Today's bar missing or still in progress: row inspection can't tell, so the TTL governs.
|
||||||
assert su._needs_same_day_refresh(_write(tmp_path, age_seconds=STALE), TODAY, TODAY) is True
|
assert su._cache_is_fresh(_write(tmp_path, age_seconds=STALE), NOW.normalize(), NOW) is False
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.unit
|
|
||||||
def test_partial_current_day_bar_is_still_refreshed(tmp_path):
|
|
||||||
# Today's row is present but may be an in-progress candle whose Close is not
|
|
||||||
# the closing price. Row inspection can't distinguish it, so the TTL governs.
|
|
||||||
f = _write(tmp_path, age_seconds=STALE, last_date="2026-07-18")
|
f = _write(tmp_path, age_seconds=STALE, last_date="2026-07-18")
|
||||||
assert su._needs_same_day_refresh(f, TODAY, TODAY) is True
|
assert su._cache_is_fresh(f, NOW.normalize(), NOW) is False
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.unit
|
@pytest.mark.unit
|
||||||
def test_recent_cache_is_not_refetched(tmp_path):
|
def test_recent_cache_is_fresh(tmp_path):
|
||||||
# Written moments ago: don't hammer the vendor (weekend/holiday guard).
|
# Written moments ago: don't hammer the vendor (weekend/holiday guard).
|
||||||
assert su._needs_same_day_refresh(_write(tmp_path), TODAY, TODAY) is False
|
assert su._cache_is_fresh(_write(tmp_path), NOW.normalize(), NOW) is True
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.unit
|
@pytest.mark.unit
|
||||||
def test_historical_request_always_uses_cache(tmp_path):
|
def test_historical_request_uses_todays_cache_past_the_ttl(tmp_path):
|
||||||
# Past dates are immutable: never refetch, however old the file is.
|
|
||||||
past = pd.Timestamp("2026-05-01")
|
|
||||||
f = _write(tmp_path, age_seconds=STALE, last_date="2026-04-30")
|
f = _write(tmp_path, age_seconds=STALE, last_date="2026-04-30")
|
||||||
assert su._needs_same_day_refresh(f, past, TODAY) is False
|
assert su._cache_is_fresh(f, pd.Timestamp("2026-05-01"), NOW) is True
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.unit
|
||||||
|
def test_a_download_from_an_earlier_day_is_not_fresh(tmp_path):
|
||||||
|
f = _write(tmp_path, age_seconds=13 * 3600) # yesterday 23:00
|
||||||
|
assert su._cache_is_fresh(f, pd.Timestamp("2026-05-01"), NOW) is False
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.unit
|
@pytest.mark.unit
|
||||||
def test_load_ohlcv_refetches_stale_same_day_cache(tmp_path, monkeypatch):
|
def test_load_ohlcv_refetches_stale_same_day_cache(tmp_path, monkeypatch):
|
||||||
"""End-to-end: the helper is actually wired into load_ohlcv's cache branch.
|
"""End-to-end: the freshness check is wired into load_ohlcv's cache branch."""
|
||||||
|
_write(tmp_path, age_seconds=STALE)
|
||||||
Without this, the unit tests above would still pass if the helper were never
|
|
||||||
called from the real code path.
|
|
||||||
"""
|
|
||||||
monkeypatch.setattr(su, "get_config", lambda: {"data_cache_dir": str(tmp_path)})
|
|
||||||
monkeypatch.setattr(su.pd.Timestamp, "today", staticmethod(lambda: TODAY))
|
|
||||||
|
|
||||||
# Pre-seed the cache file load_ohlcv will look for, aged past the TTL.
|
|
||||||
start = (TODAY - pd.DateOffset(years=5)).strftime("%Y-%m-%d")
|
|
||||||
end = (TODAY + pd.Timedelta(days=1)).strftime("%Y-%m-%d")
|
|
||||||
cache_file = tmp_path / f"AAPL-YFin-data-{start}-{end}.csv"
|
|
||||||
pd.DataFrame({"Date": ["2026-07-17"], "Close": [100.0]}).to_csv(cache_file, index=False)
|
|
||||||
old = time.time() - STALE
|
|
||||||
os.utime(cache_file, (old, old))
|
|
||||||
|
|
||||||
calls = []
|
calls = []
|
||||||
|
|
||||||
def _fake_download(*a, **k):
|
def _fake_download(*a, **k):
|
||||||
@@ -83,27 +75,31 @@ def test_load_ohlcv_refetches_stale_same_day_cache(tmp_path, monkeypatch):
|
|||||||
{"Date": pd.to_datetime(["2026-07-17", "2026-07-18"]), "Close": [100.0, 222.0]}
|
{"Date": pd.to_datetime(["2026-07-17", "2026-07-18"]), "Close": [100.0, 222.0]}
|
||||||
).set_index("Date")
|
).set_index("Date")
|
||||||
|
|
||||||
monkeypatch.setattr(su.yf, "download", _fake_download)
|
out = _load(tmp_path, monkeypatch, "2026-07-18", _fake_download)
|
||||||
|
|
||||||
out = su.load_ohlcv("AAPL", TODAY.strftime("%Y-%m-%d"))
|
|
||||||
|
|
||||||
assert calls, "stale same-day cache must trigger a refetch"
|
assert calls, "stale same-day cache must trigger a refetch"
|
||||||
assert 222.0 in out["Close"].values, "refreshed close must reach the caller"
|
assert 222.0 in out["Close"].values, "refreshed close must reach the caller"
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.unit
|
@pytest.mark.unit
|
||||||
def test_load_ohlcv_reuses_fresh_same_day_cache(tmp_path, monkeypatch):
|
def test_load_ohlcv_reuses_fresh_same_day_cache(tmp_path, monkeypatch):
|
||||||
# Mirror image: a fresh cache must NOT trigger a download.
|
_write(tmp_path, last_date="2026-07-18")
|
||||||
|
_load(tmp_path, monkeypatch, "2026-07-18", _fail_download)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.unit
|
||||||
|
def test_one_cache_file_per_symbol_across_days(tmp_path, monkeypatch):
|
||||||
|
"""A later day's download replaces the symbol's file instead of adding one (#1330)."""
|
||||||
monkeypatch.setattr(su, "get_config", lambda: {"data_cache_dir": str(tmp_path)})
|
monkeypatch.setattr(su, "get_config", lambda: {"data_cache_dir": str(tmp_path)})
|
||||||
monkeypatch.setattr(su.pd.Timestamp, "today", staticmethod(lambda: TODAY))
|
frame = pd.DataFrame({"Date": pd.to_datetime(["2026-07-16", "2026-07-17"]), "Close": [1.0, 2.0]})
|
||||||
|
downloads = []
|
||||||
|
monkeypatch.setattr(su.yf, "download", lambda *a, **k: downloads.append(1) or frame.set_index("Date"))
|
||||||
|
|
||||||
start = (TODAY - pd.DateOffset(years=5)).strftime("%Y-%m-%d")
|
for day in ("2026-07-18 10:00", "2026-07-19 10:00", "2026-07-20 10:00"):
|
||||||
end = (TODAY + pd.Timedelta(days=1)).strftime("%Y-%m-%d")
|
now = pd.Timestamp(day)
|
||||||
cache_file = tmp_path / f"AAPL-YFin-data-{start}-{end}.csv"
|
monkeypatch.setattr(su.pd.Timestamp, "today", staticmethod(lambda now=now: now))
|
||||||
pd.DataFrame({"Date": ["2026-07-18"], "Close": [100.0]}).to_csv(cache_file, index=False)
|
su.load_ohlcv("AAPL", "2026-07-17")
|
||||||
|
written = list(tmp_path.glob("AAPL-*.csv"))
|
||||||
|
os.utime(written[0], (now.timestamp(), now.timestamp()))
|
||||||
|
|
||||||
def _fail_download(*a, **k):
|
assert len(downloads) == 3, "each new day refetches"
|
||||||
raise AssertionError("fresh cache must not refetch")
|
assert [p.name for p in tmp_path.iterdir()] == ["AAPL-YFin-data.csv"]
|
||||||
|
|
||||||
monkeypatch.setattr(su.yf, "download", _fail_download)
|
|
||||||
su.load_ohlcv("AAPL", TODAY.strftime("%Y-%m-%d"))
|
|
||||||
|
|||||||
@@ -13,6 +13,8 @@ anywhere counts as no data and the staleness check judges the rest.
|
|||||||
"""
|
"""
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import os
|
||||||
|
|
||||||
import pandas as pd
|
import pandas as pd
|
||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
@@ -89,9 +91,9 @@ def _run_load(monkeypatch, tmp_path, frame, curr_date):
|
|||||||
monkeypatch.setattr(su, "get_config", lambda: {"data_cache_dir": str(tmp_path)})
|
monkeypatch.setattr(su, "get_config", lambda: {"data_cache_dir": str(tmp_path)})
|
||||||
today = pd.Timestamp(curr_date)
|
today = pd.Timestamp(curr_date)
|
||||||
monkeypatch.setattr(su.pd.Timestamp, "today", staticmethod(lambda: today))
|
monkeypatch.setattr(su.pd.Timestamp, "today", staticmethod(lambda: today))
|
||||||
start = (today - pd.DateOffset(years=5)).strftime("%Y-%m-%d")
|
cache_file = tmp_path / "AAPL-YFin-data.csv"
|
||||||
end = (today + pd.Timedelta(days=1)).strftime("%Y-%m-%d")
|
cache_file.write_text(frame.to_csv(index=False))
|
||||||
(tmp_path / f"AAPL-YFin-data-{start}-{end}.csv").write_text(frame.to_csv(index=False))
|
os.utime(cache_file, (today.timestamp(), today.timestamp()))
|
||||||
|
|
||||||
def _fail_download(*a, **k):
|
def _fail_download(*a, **k):
|
||||||
raise AssertionError("should use the seeded cache, not download")
|
raise AssertionError("should use the seeded cache, not download")
|
||||||
|
|||||||
@@ -164,21 +164,19 @@ def _assert_ohlcv_not_stale(
|
|||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
def _needs_same_day_refresh(data_file, curr_date_dt, today_date) -> bool:
|
def _cache_is_fresh(data_file, curr_date_dt, now) -> bool:
|
||||||
"""Whether a cached frame must be refetched to reflect the requested day.
|
"""Whether the symbol's cached download can serve this request.
|
||||||
|
|
||||||
The cache file is keyed per day, so without this a run started before the
|
The file holds the download made on the day it was written, so it serves
|
||||||
day's bar was final keeps serving that snapshot to every later run (#1150).
|
only that day. A current-day request also refetches once the file is older
|
||||||
Two distinct staleness cases exist for a current-day request: the bar may be
|
than the TTL: Yahoo publishes a partial daily candle during market hours,
|
||||||
missing entirely, or present but still in progress — Yahoo publishes a
|
whose ``Close`` is not the closing price, and row inspection cannot tell it
|
||||||
partial daily candle during market hours, whose ``Close`` is not the closing
|
from a final one (#1150).
|
||||||
price. Row inspection cannot tell a partial bar from a final one, so the TTL
|
|
||||||
governs every current-day cache. Historical requests always reuse the cache,
|
|
||||||
since those rows are immutable.
|
|
||||||
"""
|
"""
|
||||||
if curr_date_dt.date() < today_date.date():
|
written = pd.Timestamp.fromtimestamp(os.path.getmtime(data_file))
|
||||||
|
if written.date() != now.date():
|
||||||
return False
|
return False
|
||||||
return time.time() - os.path.getmtime(data_file) > OHLCV_CACHE_TTL_SECONDS
|
return curr_date_dt.date() < now.date() or (now - written).total_seconds() <= OHLCV_CACHE_TTL_SECONDS
|
||||||
|
|
||||||
|
|
||||||
def load_ohlcv(symbol: str, curr_date: str) -> pd.DataFrame:
|
def load_ohlcv(symbol: str, curr_date: str) -> pd.DataFrame:
|
||||||
@@ -197,19 +195,19 @@ def load_ohlcv(symbol: str, curr_date: str) -> pd.DataFrame:
|
|||||||
config = get_config()
|
config = get_config()
|
||||||
curr_date_dt = pd.to_datetime(curr_date).normalize()
|
curr_date_dt = pd.to_datetime(curr_date).normalize()
|
||||||
|
|
||||||
# Cache uses a fixed window (5y to today) so one file per symbol.
|
# One cache file per symbol, holding the latest 5y-to-today download.
|
||||||
today_date = pd.Timestamp.today()
|
now = pd.Timestamp.today()
|
||||||
start_date = today_date - pd.DateOffset(years=5)
|
start_date = now - pd.DateOffset(years=5)
|
||||||
start_str = start_date.strftime("%Y-%m-%d")
|
start_str = start_date.strftime("%Y-%m-%d")
|
||||||
# yfinance ``end`` is EXCLUSIVE; request tomorrow so today's row is included
|
# yfinance ``end`` is EXCLUSIVE; request tomorrow so today's row is included
|
||||||
# when curr_date is the current day (#986). Look-ahead is still prevented by
|
# when curr_date is the current day (#986). Look-ahead is still prevented by
|
||||||
# the curr_date filter below.
|
# the curr_date filter below.
|
||||||
end_str = (today_date + pd.Timedelta(days=1)).strftime("%Y-%m-%d")
|
end_str = (now + pd.Timedelta(days=1)).strftime("%Y-%m-%d")
|
||||||
|
|
||||||
os.makedirs(config["data_cache_dir"], exist_ok=True)
|
os.makedirs(config["data_cache_dir"], exist_ok=True)
|
||||||
data_file = os.path.join(
|
data_file = os.path.join(
|
||||||
config["data_cache_dir"],
|
config["data_cache_dir"],
|
||||||
f"{safe_symbol}-YFin-data-{start_str}-{end_str}.csv",
|
f"{safe_symbol}-YFin-data.csv",
|
||||||
)
|
)
|
||||||
|
|
||||||
# A cached file may be empty if a prior fetch failed (unknown symbol,
|
# A cached file may be empty if a prior fetch failed (unknown symbol,
|
||||||
@@ -218,12 +216,10 @@ def load_ohlcv(symbol: str, curr_date: str) -> pd.DataFrame:
|
|||||||
data = None
|
data = None
|
||||||
if os.path.exists(data_file):
|
if os.path.exists(data_file):
|
||||||
cached = pd.read_csv(data_file, on_bad_lines="skip", encoding="utf-8")
|
cached = pd.read_csv(data_file, on_bad_lines="skip", encoding="utf-8")
|
||||||
# Serve the cache only when it is usable and not a stale snapshot of the
|
|
||||||
# day being requested (#1150); otherwise fall through and refetch.
|
|
||||||
if (
|
if (
|
||||||
not cached.empty
|
not cached.empty
|
||||||
and "Close" in cached.columns
|
and "Close" in cached.columns
|
||||||
and not _needs_same_day_refresh(data_file, curr_date_dt, today_date)
|
and _cache_is_fresh(data_file, curr_date_dt, now)
|
||||||
):
|
):
|
||||||
data = cached
|
data = cached
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user