"""Data-cache tests: incremental refresh from changed data files. Uses a small synthetic data root in a temp dir — no real data, no server. Run: .venv/bin/python tests/test_data.py """ from __future__ import annotations import json import os import pathlib import shutil import sys import tempfile import time sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent)) import data as D # noqa: E402 import pandas as pd # noqa: E402 PASS, FAIL = 0, 0 def check(name: str, cond: bool, extra: str = "") -> None: global PASS, FAIL if cond: PASS += 1 print(f" ok {name}", flush=True) else: FAIL += 1 print(f" FAIL {name} {extra}", flush=True) def write_history(root: pathlib.Path, sym: str, last_price: float | None = None, days: int = 10) -> pathlib.Path: """Prices 100.00, 101.00, ..., 109.00 (last day overridable).""" f = root / f"{sym}-history.csv" lines = ["Date,Open,High,Low,Close,Adj Close,Volume"] for i in range(days): p = last_price if (last_price is not None and i == days - 1) \ else 100.0 + i lines.append(f"2024-01-{i + 1:02d},{p:.2f},{p:.2f},{p:.2f},{p:.2f},{p:.2f},1000") f.write_text("\n".join(lines) + "\n") return f def touch(p: pathlib.Path) -> None: """Bump mtime so the manifest sees the file as changed.""" st = p.stat() os.utime(p, ns=(st.st_atime_ns + 10**9, st.st_mtime_ns + 10**9)) def reload(d: pathlib.Path) -> D.Bundle: # refresh() is the synchronous path: deterministic for these assertions return D.refresh(root=d["root"], cache=d["cache"]) def main() -> int: # fast watcher for the background-refresh test below D.WATCH_INTERVAL, D.REFRESH_CADENCE = 0.2, 0.0 # synthetic root: no frozen overrides tmp = pathlib.Path(tempfile.mkdtemp(prefix="fdatatest-")) D.FROZEN_DIR = tmp / "frozen" D.FROZEN_DIR.mkdir() root, cache = tmp / "stocks", tmp / "cache" root.mkdir() d = {"root": root, "cache": cache} try: write_history(root, "aaa") write_history(root, "bbb") write_history(root, "ccc") (root / "aaa-dividend.csv").write_text("Date,Dividends\n2024-01-05,0.5\n") (root / "bbb.json").write_text( json.dumps({"chart": {"result": [{"meta": {"longName": "Bee Bee Corp"}}]}})) b = reload(d) check("initial full build", list(b.adj.columns) == ["aaa", "bbb", "ccc"] and b.adj["aaa"].iloc[-1] == 109.0, str(list(b.adj.columns))) check("names from json", b.names.get("bbb") == "Bee Bee Corp") check("dividend panel", b.div["aaa"].loc[pd.Timestamp("2024-01-05")] == 0.5) # --- modified file: last price changes f = write_history(root, "aaa", last_price=200.0) touch(f) b = reload(d) check("changed file re-read incrementally", b.adj["aaa"].iloc[-1] == 200.0) check("untouched symbols intact", b.adj["bbb"].iloc[-1] == 109.0) # --- new file: symbol added write_history(root, "ddd") b = reload(d) check("new symbol added", "ddd" in b.adj.columns) # --- deleted file: symbol dropped (root / "ccc-history.csv").unlink() b = reload(d) check("deleted symbol dropped", "ccc" not in b.adj.columns and "ccc" not in b.div.columns) # --- name file added for a new symbol (root / "ddd.json").write_text( json.dumps({"chart": {"result": [{"meta": {"longName": "Dee Dee"}}]}})) b = reload(d) check("name added incrementally", b.names.get("ddd") == "Dee Dee") # --- dividend file modified (root / "aaa-dividend.csv").write_text( "Date,Dividends\n2024-01-05,0.5\n2024-01-06,1.25\n") b = reload(d) check("dividend change picked up", b.div["aaa"].loc[pd.Timestamp("2024-01-06")] == 1.25) # --- no-change reload: cheap, no rewrite before = (cache / "panel_adj.parquet").stat().st_mtime_ns b = reload(d) after = (cache / "panel_adj.parquet").stat().st_mtime_ns check("no-op reload does not rewrite parquet", before == after) # --- background refresh: load_bundle never blocks on changed data; # it serves the current bundle and a background thread catches up f = write_history(root, "aaa", last_price=300.0) touch(f) t0 = time.monotonic() b = D.load_bundle(root=root, cache=cache) check("load_bundle returns immediately on changed data", time.monotonic() - t0 < 2.0) deadline = time.monotonic() + 15.0 while time.monotonic() < deadline: b = D.load_bundle(root=root, cache=cache) if b.adj["aaa"].iloc[-1] == 300.0: break time.sleep(0.2) check("background thread picks up the change", b.adj["aaa"].iloc[-1] == 300.0) check("generation counter advances", D.generation(root=root, cache=cache) >= 1) # --- forced full rebuild still works b = D.load_bundle(root=root, cache=cache, rebuild=True) check("forced full rebuild", list(b.adj.columns) == ["aaa", "bbb", "ddd"]) finally: shutil.rmtree(tmp, ignore_errors=True) # --- cross-file correction invariants (Yahoo double-listing) --- tmp = pathlib.Path(tempfile.mkdtemp(prefix="fd-")) try: root = tmp / "data"; cache = tmp / "cache"; corrdir = tmp / "corr" root.mkdir(parents=True); cache.mkdir(); corrdir.mkdir() (root / "eee-dividend.csv").write_text( "Date,Dividends\n2020-01-02,1.5\n2020-12-15,4.0\n") (root / "eee-capitalGain.csv").write_text( "Date,Capital Gains\n2020-12-15,4.0\n2021-12-14,2.5\n") (root / "fff-dividend.csv").write_text( "Date,Dividends\n2020-12-15,1.0\n") (root / "fff-capitalGain.csv").write_text( "Date,Capital Gains\n2020-12-15,0.9\n") # eee: same-amount pair (dedup -> keep capg) ; fff: differing amounts # (drop_capg_copy -> keep the dividend row) (corrdir / "EEE.json").write_text(json.dumps({"dedup": [["2020-12-15", 4.0]]})) (corrdir / "FFF.json").write_text(json.dumps({"drop_capg_copy": ["2020-12-15"]})) real_corr = D.CORRECTIONS_DIR D.CORRECTIONS_DIR = corrdir try: # pure list-level function, both layouts d, c = D.dedupe_event_rows({"dedup": [["2020-12-15", 4.0]]}, [("2020-12-15", 4.0)], [("2020-12-15", 4.0)]) check("dedup: both copies -> keep capg", d == [] and c == [("2020-12-15", 4.0)]) d, c = D.dedupe_event_rows({"dedup": [["2020-12-15", 4.0]]}, [("2020-12-15", 4.0)], []) check("dedup: lone div copy (new layout) -> keep", d == [("2020-12-15", 4.0)] and c == []) d, c = D.dedupe_event_rows({"drop_capg_copy": ["2020-12-15"]}, [("2020-12-15", 1.0)], [("2020-12-15", 0.9)]) check("drop_capg_copy: old layout", d == [("2020-12-15", 1.0)] and c == []) d, c = D.dedupe_event_rows({"drop_capg_copy": ["2020-12-15"]}, [("2020-12-15", 1.0)], []) check("drop_capg_copy: new layout (no capg) no-op", d == [("2020-12-15", 1.0)] and c == []) # bundle-level application b = D.load_bundle(root=root, cache=cache) ts = pd.Timestamp("2020-12-15") check("bundle: eee div copy dropped, capg kept", b.div.at[ts, "eee"] == 0.0 and b.capg.at[ts, "eee"] == 4.0) check("bundle: eee non-pair rows untouched", b.div["eee"].loc[pd.Timestamp("2020-01-02")] == 1.5 and b.capg["eee"].loc[pd.Timestamp("2021-12-14")] == 2.5) check("bundle: fff capg copy dropped, div kept", b.div.at[ts, "fff"] == 1.0 and b.capg.at[ts, "fff"] == 0.0) # idempotent: reloading does not double-apply b2 = D.load_bundle(root=root, cache=cache) check("invariants idempotent on reload", b2.div["eee"][ts] == 0.0 and b2.capg["eee"][ts] == 4.0) finally: D.CORRECTIONS_DIR = real_corr finally: shutil.rmtree(tmp, ignore_errors=True) # --- history cell corrections (misaligned Yahoo adj-close spike) --- tmp = pathlib.Path(tempfile.mkdtemp(prefix="fd-")) try: root = tmp / "data"; cache = tmp / "cache"; corrdir = tmp / "corr" root.mkdir(parents=True); cache.mkdir(); corrdir.mkdir() (root / "ghh-history.csv").write_text( "Date,Open,High,Low,Close,Adj Close,Volume\n" "2020-12-10,30.10,30.10,30.10,30.10,11.247889,1000\n" "2020-12-11,30.10,30.10,30.10,30.10,14.545517,1000\n" "2020-12-14,23.22,23.22,23.22,23.22,11.220826,1000\n") (corrdir / "GHH.json").write_text(json.dumps( {"history": {"2020-12-11": {"Adj Close": 11.247889}}})) real_corr = D.CORRECTIONS_DIR D.CORRECTIONS_DIR = corrdir try: b = D.load_bundle(root=root, cache=cache) t10, t11, t14 = (pd.Timestamp(x) for x in ("2020-12-10", "2020-12-11", "2020-12-14")) check("history: adj cell patched", b.adj["ghh"][t11] == 11.247889) check("history: other adj cells untouched", b.adj["ghh"][t10] == 11.247889 and b.adj["ghh"][t14] == 11.220826) check("history: raw close untouched", b.close["ghh"][t11] == 30.10 and b.close["ghh"][t14] == 23.22) b2 = D.load_bundle(root=root, cache=cache) check("history: idempotent on reload", b2.adj["ghh"][t11] == 11.247889) finally: D.CORRECTIONS_DIR = real_corr finally: shutil.rmtree(tmp, ignore_errors=True) print(f"\n{PASS} passed, {FAIL} failed") return 1 if FAIL else 0 if __name__ == "__main__": sys.exit(main())