f/scripts/scan_adj_misalign.py

#!/usr/bin/env python3
"""Scan for Yahoo adj-close/event-date misalignment.

Yahoo's Adj Close column applies a distribution's adjustment on the EVENT
date from its event feed. When that date precedes the actual ex-div date
(the market price only drops later), the adjusted series shows a bogus
spike on the event date and a snap-back a day or two later.

Signature per symbol: a date D with |adj return| >= --min-adj (default 5%)
while |close return| on D is < --max-close (default 2%), AND the close
drops by more than half the implied adjustment within --horizon following
sessions (the price finally going ex-div).

Usage:  python3 scripts/scan_adj_misalign.py [--data DIR] [--out PATH]
Prints a summary and a table; writes markdown to --out (default
reports/adj_misalign/scan.md).
"""
import argparse
import json
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

ROOT_DEFAULT = Path.home() / "prog/fin/stocks"


def scan(root: Path, min_adj: float, max_close: float, horizon: int) -> list[dict]:
    import data as D
    hits = []
    syms = sorted({f.name[:-len("-history.csv")] for f in root.glob("*-history.csv")})
    for i, sym in enumerate(syms):
        fr = D.FROZEN_DIR / f"{sym}-history.csv"
        f = fr if (D.FROZEN_DIR.is_dir() and fr.exists()) else root / f"{sym}-history.csv"
        if not f.exists():
            continue
        rows = []
        for l in f.read_text(errors="replace").splitlines()[1:]:
            if not l.strip():
                continue
            r = l.split(",")
            if len(r) < 6:
                continue
            try:
                cl, aj = float(r[4]), float(r[5])
            except ValueError:      # e.g. '%!f(<nil>)' from an old goget bug
                continue
            rows.append((r[0], cl, aj))
        if len(rows) < 3:
            continue
        closes = [x[1] for x in rows]
        adjs = [x[2] for x in rows]
        for i in range(1, len(rows)):
            if closes[i - 1] <= 0 or adjs[i - 1] <= 0 or closes[i] <= 0:
                continue
            adj_ret = adjs[i] / adjs[i - 1] - 1
            cl_ret = closes[i] / closes[i - 1] - 1
            if abs(adj_ret) < min_adj or abs(cl_ret) >= max_close:
                continue
            # the drop has to land within the next `horizon` sessions
            j = i + horizon
            if j >= len(rows) or closes[j] <= 0:
                continue
            later_drop = (closes[i - 1] - closes[j]) / closes[i - 1]
            implied = min(abs(adj_ret), 0.6)   # cap: events rarely exceed 60%
            if later_drop < 0.5 * implied:
                continue
            hits.append({
                "symbol": sym,
                "date": rows[i][0],
                "adj_ret": round(adj_ret, 4),
                "close_ret": round(cl_ret, 4),
                "close_before": closes[i - 1],
                "close_after_horizon": closes[j],
                "drop_later": round(later_drop, 4),
            })
        if i % 500 == 0:
            print(f"{i}/{len(syms)} ...", file=sys.stderr)
    return hits


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--data", type=Path, default=ROOT_DEFAULT)
    ap.add_argument("--min-adj", type=float, default=0.05)
    ap.add_argument("--max-close", type=float, default=0.02)
    ap.add_argument("--horizon", type=int, default=3)
    ap.add_argument("--out", type=Path, default=None)
    args = ap.parse_args()
    hits = scan(args.data, args.min_adj, args.max_close, args.horizon)
    hits.sort(key=lambda h: -abs(h["adj_ret"]))
    print(json.dumps({"hits": len(hits),
                      "symbols": len({h["symbol"] for h in hits})}, indent=1))
    for h in hits[:30]:
        print(f"{h['symbol']:8s} {h['date']} adj {h['adj_ret']:+.1%} "
              f"(close {h['close_ret']:+.2%}), drop later {h['drop_later']:.1%}")
    if args.out:
        args.out.parent.mkdir(parents=True, exist_ok=True)
        lines = ["# Adj-close / event-date misalignment scan",
                 "",
                 f"params: min_adj={args.min_adj} max_close={args.max_close} "
                 f"horizon={args.horizon}; {len(hits)} hits on "
                 f"{len({h['symbol'] for h in hits})} symbols",
                 "",
                 "| symbol | date | adj ret | close ret | later drop |",
                 "|---|---|---|---|---|"]
        lines += [f"| {h['symbol']} | {h['date']} | {h['adj_ret']:+.2%} | "
                  f"{h['close_ret']:+.2%} | {h['drop_later']:+.2%} |" for h in hits]
        args.out.write_text("\n".join(lines) + "\n")
        print(f"-> {args.out}")
    return 0


if __name__ == "__main__":
    main()