From 9263283995e717f0631ceb39415413e96ea9dc76 Mon Sep 17 00:00:00 2001 From: Greg Pomerantz Date: Mon, 31 Aug 2026 14:49:39 -0400 Subject: [PATCH] Corrections overlay: per-symbol distribution fixes survive re-downloads overrides/corrections/{SYM}.json (git-tracked, with as_of/source/note) holds remove/replace/add ops on the dividend and capital-gain series. data.py applies them on top of whatever the data root (or the frozen snapshot) provides, in both the full build and the incremental refresh path, and the corrections dir joins the cache manifest so a change invalidates the cache. A goget re-download of the base CSV can never clobber a confirmed correction. Format and usage documented in data.py. --- data.py | 92 ++++++++++++++++++++++++++++++++-- overrides/corrections/.gitkeep | 0 2 files changed, 89 insertions(+), 3 deletions(-) create mode 100644 overrides/corrections/.gitkeep diff --git a/data.py b/data.py index afefa7a..24d9b23 100644 --- a/data.py +++ b/data.py @@ -33,6 +33,51 @@ CACHE_DIR = Path(__file__).parent / ".cache" # frozen copy, the data-root files are ignored so a future goget # re-download can never clobber the last known-good series. FROZEN_DIR = Path(__file__).parent / "overrides" / "frozen" +# Corrections: per-symbol distribution fixes confirmed against official +# sources (see scripts/verify_official.py). Applied on top of whatever the +# data root / frozen dir says, so re-downloads can never clobber them. +# Format, one file per symbol in overrides/corrections/: +# {"dividends": {"remove": [["2008-12-18", 0.292], ...], +# "replace": {"2008-12-18": 0.0}, +# "add": [["2007-12-18", 0.292], ...]}, +# "capitalGains": {...same ops...}, +# "as_of": "2026-08-31", "source": "", "note": "..."} +CORRECTIONS_DIR = Path(__file__).parent / "overrides" / "corrections" +_corr_cache: dict = {} + + +def _load_corrections(sym: str) -> dict: + f = CORRECTIONS_DIR / f"{sym.upper()}.json" + if f.exists(): + try: + if f not in _corr_cache or _corr_cache[f][0] != f.stat().st_mtime_ns: + _corr_cache[f] = (f.stat().st_mtime_ns, + json.loads(f.read_text())) + return _corr_cache[f][1] + except Exception: + return {} + return {} + + +def _apply_corrections(sym: str, suffix: str, ser: pd.Series) -> pd.Series: + key = {"dividend": "dividends", "capitalGain": "capitalGains"}.get(suffix) + ops = _load_corrections(sym).get(key) if key else None + if not ops: + return ser + d = {ts: float(v) for ts, v in ser.items()} + for r in ops.get("remove", []): + ts = pd.Timestamp(r[0]) + amt = r[1] if len(r) > 1 else None + if ts in d and (amt is None or abs(d[ts] - amt) < 1e-9): + del d[ts] + for ts_s, amt in ops.get("replace", {}).items(): + d[pd.Timestamp(ts_s)] = float(amt) + for r in ops.get("add", []): + ts = pd.Timestamp(r[0]) + d[ts] = d.get(ts, 0.0) + float(r[1]) + if not d: + return pd.Series(dtype=float) + return pd.Series(d).sort_index() _SUFFIXES = ("history", "dividend", "capitalGain") _PANELS = ("adj", "close", "div", "capg") @@ -84,7 +129,7 @@ def _read_panel(root: Path, suffix: str, col: str) -> pd.DataFrame: except Exception: continue if col in df.columns: - cols.append(df[col]) + cols.append(_apply_corrections(sym, suffix, df[col])) syms.append(sym) if not cols: return pd.DataFrame() @@ -136,6 +181,12 @@ def _scan_files(root: Path) -> dict: if e.is_file() and e.name.endswith((".csv", ".json")): st = e.stat() out["frozen/" + e.name] = (st.st_mtime_ns, st.st_size) + if CORRECTIONS_DIR.is_dir(): + with os.scandir(CORRECTIONS_DIR) as it: + for e in it: + if e.is_file() and e.name.endswith(".json"): + st = e.stat() + out["corrections/" + e.name] = (st.st_mtime_ns, st.st_size) return out @@ -196,7 +247,37 @@ def _apply_changes(root: Path, cache: Path, old: dict, new: dict) -> None: # data-root files of frozen symbols are ignored: their canonical series # lives in the overrides dir, which is tracked in the same scan frozen = frozen_syms() + + def _recompute_corrected(symfile: str) -> None: + """corrections/{SYM}.json changed: re-derive the symbol's + dividend/capital-gain series from the base file + corrections.""" + for sfx, colname, key in (("dividend", "Dividends", "div"), + ("capitalGain", "Capital Gains", "capg")): + sym = None + p = None + for case in (symfile[:-5].lower(), symfile[:-5].upper()): + for base in (FROZEN_DIR, root): + q = base / f"{case}-{sfx}.csv" + if q.exists(): + sym, p = case, q + break + if sym: + break + if p is None: + continue + try: + df = _read_csv_clean(p) + except Exception: + continue + if colname not in df.columns: + continue + updates[key][sym] = _apply_corrections(sym, sfx, df[colname]) \ + .fillna(0.0) + for fname in changed: + if fname.startswith("corrections/"): + _recompute_corrected(fname[12:]) + continue sfx, sym = _suffix(fname) if sfx is None: continue @@ -212,11 +293,16 @@ def _apply_changes(root: Path, cache: Path, old: dict, new: dict) -> None: if "Close" in df.columns: updates["close"][sym] = df["Close"] elif sfx == "dividend" and "Dividends" in df.columns: - updates["div"][sym] = df["Dividends"].fillna(0.0) + updates["div"][sym] = _apply_corrections(sym, sfx, df["Dividends"]) \ + .fillna(0.0) elif sfx == "capitalGain" and "Capital Gains" in df.columns: - updates["capg"][sym] = df["Capital Gains"].fillna(0.0) + updates["capg"][sym] = _apply_corrections(sym, sfx, df["Capital Gains"]) \ + .fillna(0.0) for fname in removed: sfx, sym = _suffix(fname) + if fname.startswith("corrections/"): + _recompute_corrected(fname[12:]) + continue if fname.startswith("frozen/"): continue # frozen files are not data-root symbols if sym in frozen: diff --git a/overrides/corrections/.gitkeep b/overrides/corrections/.gitkeep new file mode 100644 index 0000000..e69de29