Greg Pomerantz
5e5a10d725
Yahoo's dividend endpoint returns the fund's TOTAL per-share distribution
(dividend + capital gain) for many share classes while the capitalGains
endpoint returns the cap-gain portion separately; summing both double-counted
cap gains and broke pre/post-tax comparability (after-tax engine could beat
the adj-based pre-tax return: impossible).
Add _fix_total_distributions to the bundle-assembly invariants
(layout-agnostic, survives re-downloads): on same-date div>capg events, the
exact Yahoo-adj implied distribution (d = P - Q*A_{t-1}/A_t) must match the
div-file amount (not the sum) on >=3 dates before rewriting div -= capg.
Self-validating: genuine separate same-date distributions are untouched.
Rewrote 17,900 cells on 1,957 symbols.
Pool check (scripts/check_adj_consistency.py): all 14 symbols now
TR(close+events) == TR(adj) to 0.01pt and post < pre with plausible drags.
Tests: rewrite case, no-rewrite case, idempotency (tests/test_data.py).
289 lines
13 KiB
Python
289 lines
13 KiB
Python
"""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)
|
|
test_total_distributions()
|
|
print(f"\n{PASS} passed, {FAIL} failed")
|
|
return 1 if FAIL else 0
|
|
|
|
|
|
def test_total_distributions() -> None:
|
|
"""Yahoo 'dividend file = total distribution' invariant (unit level)."""
|
|
n = 8
|
|
idx = pd.DatetimeIndex([f"2024-01-{i + 1:02d}" for i in range(n)])
|
|
|
|
def close_from(start: float, total_by_i: dict[int, float]) -> list[float]:
|
|
# price drops by the distribution on ex-dates (+0.05 drift)
|
|
c = [start]
|
|
for i in range(1, n):
|
|
d = total_by_i.get(i, 0.0)
|
|
c.append(c[-1] - d + 0.05)
|
|
return c
|
|
|
|
def adj_from(closes: list[float], total_by_i: dict[int, float]) -> list[float]:
|
|
# Yahoo convention: A_i = A_{i-1} * Q_i / (P_{i-1} - d)
|
|
a = [closes[0]]
|
|
for i in range(1, n):
|
|
d = total_by_i.get(i, 0.0)
|
|
a.append(a[-1] * closes[i] / (closes[i - 1] - d) if d else a[-1])
|
|
return a
|
|
|
|
def mk(vals: dict[int, float]) -> pd.Series:
|
|
s = pd.Series({idx[i]: v for i, v in vals.items()}, dtype=float)
|
|
return s.reindex(idx).fillna(0.0)
|
|
|
|
# tot: div file carries the TOTAL, capg file the cap-gain part
|
|
tot_div = {1: 1.0, 3: 2.0, 5: 1.5}
|
|
tot_cap = {1: 0.4, 3: 0.8, 5: 0.5}
|
|
# sep: genuine separate same-date distributions (adj matches the SUM)
|
|
sep_div = {1: 0.9, 3: 1.4, 5: 1.2}
|
|
sep_cap = {1: 0.4, 3: 0.3, 5: 0.2}
|
|
tot_close = close_from(100.0, tot_div)
|
|
sep_close = close_from(100.0, {i: sep_div[i] + sep_cap[i] for i in sep_div})
|
|
close_df = pd.DataFrame({"tot": tot_close, "sep": sep_close}, index=idx)
|
|
adj_df = pd.DataFrame({
|
|
"tot": pd.Series(adj_from(tot_close, tot_div), index=idx),
|
|
"sep": pd.Series(adj_from(sep_close, {i: sep_div[i] + sep_cap[i]
|
|
for i in sep_div}), index=idx),
|
|
})
|
|
div_df = pd.DataFrame({"tot": mk(tot_div), "sep": mk(sep_div)}, index=idx)
|
|
capg_df = pd.DataFrame({"tot": mk(tot_cap), "sep": mk(sep_cap)}, index=idx)
|
|
|
|
fixed = D._fix_total_distributions(div_df, capg_df, close_df, adj_df)
|
|
check("total-dist: only 'tot' rewritten", fixed == ["tot"])
|
|
check("total-dist: tot div -= capg (sum preserved)",
|
|
all(abs(div_df.at[idx[i], "tot"] - (tot_div[i] - tot_cap[i])) < 1e-12
|
|
for i in tot_div)
|
|
and all(abs(div_df.at[idx[i], "tot"] + capg_df.at[idx[i], "tot"]
|
|
- tot_div[i]) < 1e-12 for i in tot_div))
|
|
check("total-dist: 'sep' (adj matches sum) untouched",
|
|
all(div_df.at[idx[i], "sep"] == sep_div[i] for i in sep_div))
|
|
# idempotent: second pass changes nothing (mask now div>cap still true but
|
|
# adj already reflects the total -> rewrite would break the match)
|
|
div2 = div_df.copy()
|
|
fixed2 = D._fix_total_distributions(div2, capg_df, close_df, adj_df)
|
|
check("total-dist: idempotent (no double subtraction)",
|
|
fixed2 == [] and (div2["tot"] == div_df["tot"]).all())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|