f/scripts/scan_double_listing.py

#!/usr/bin/env python3
"""Comprehensive scan of the local dump for the Yahoo double-listing pattern.

Yahoo's event feed has been observed to list one distribution in BOTH the
dividend and capitalGains event maps (same date; sometimes the same amount,
sometimes two different amounts where the dividend-file value is the true
one — confirmed against official filings for 10 funds). This scans every
symbol's effective event files (frozen copies shadow the data root, matching
data.py) and reports:

  A  same date + same amount in dividend AND capitalGain
     -> near-certain double listing (one copy is spurious)
  B  same date, differing amounts in dividend AND capitalGain
     -> verified rule: the dividend amount is true, the capitalGain row is
        the spurious copy (10/10 officially-verified cases)
  C  two rows on the same date within ONE file
     -> same mechanism, same file (or a rare legitimate same-day pair)

Output: reports/double_listing/scan.json + scan.md (human summary).
Read-only: it proposes nothing, it just reports. Pair the report with
overrides/corrections/ entries (dedup / drop_capg_copy ops) to fix.

Usage:  python3 scripts/scan_double_listing.py [--data DIR] [--limit N]
"""
import argparse
import json
import sys
from collections import Counter
from pathlib import Path

ROOT_DEFAULT = Path.home() / "prog/fin/stocks"
FROZEN = Path(__file__).resolve().parent.parent / "overrides" / "frozen"
OUT = Path(__file__).resolve().parent.parent / "reports" / "double_listing"


def load_events(p: Path) -> dict:
    """date -> [amounts] from a 2-col event csv."""
    out = {}
    if not p.exists():
        return out
    for line in p.read_text(errors="replace").splitlines()[1:]:
        a = line.split(",")
        if len(a) >= 2:
            try:
                out.setdefault(a[0][:10], []).append(float(a[1]))
            except ValueError:
                pass
    return out


def effective(sym: str, sfx: str, root: Path) -> Path:
    f = FROZEN / f"{sym}-{sfx}.csv"
    return f if f.exists() else root / f"{sym}-{sfx}.csv"


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--data", type=Path, default=ROOT_DEFAULT)
    ap.add_argument("--limit", type=int, default=0)
    args = ap.parse_args()

    syms = sorted({f.name[: -len("-dividend.csv")]
                   for f in args.data.glob("*-dividend.csv")}
                  | {f.name[: -len("-capitalGain.csv")]
                     for f in args.data.glob("*-capitalGain.csv")})
    if args.limit:
        syms = syms[: args.limit]

    OUT.mkdir(parents=True, exist_ok=True)
    results = {}
    cat_counts = Counter()
    pair_counts = Counter()   # category -> number of pairs

    for n, sym in enumerate(syms, 1):
        div = load_events(effective(sym, "dividend", args.data))
        capg = load_events(effective(sym, "capitalGain", args.data))
        if not div and not capg:
            continue
        a, b, c = [], [], []
        for date in sorted(set(div) & set(capg)):
            for v in div[date]:
                if v in capg[date]:
                    a.append({"date": date, "amount": v})
                    pair_counts["A"] += 1
                else:
                    b.append({"date": date, "div": v, "capg": capg[date]})
                    pair_counts["B"] += 1
        for name, m in (("div", div), ("capg", capg)):
            for date, v in sorted(m.items()):
                if len(v) > 1:
                    c.append({"file": name, "date": date, "amounts": v})
                    pair_counts["C"] += len(v) - 1
        if a or b or c:
            if a:
                cat_counts["A_syms"] += 1
            if b:
                cat_counts["B_syms"] += 1
            if c:
                cat_counts["C_syms"] += 1
            results[sym] = {"same_amt": a, "diff_amt": b, "within_file": c}
        if n % 5000 == 0:
            print(f"  {n}/{len(syms)} ...", file=sys.stderr, flush=True)

    # report
    total = len(results)
    summary = {
        "symbols_scanned": len(syms),
        "symbols_with_any_pattern": total,
        "symbols_with_same_amt_pair": cat_counts["A_syms"],
        "symbols_with_diff_amt_pair": cat_counts["B_syms"],
        "symbols_with_within_file_dup": cat_counts["C_syms"],
        "pairs": dict(pair_counts),
    }
    (OUT / "scan.json").write_text(
        json.dumps({"summary": summary, "symbols": results}, indent=1))

    lines = ["# Double-listing scan", "",
             f"scanned {summary['symbols_scanned']} symbols; "
             f"{total} show the pattern", "",
             f"- same-date/same-amount (A): {cat_counts['A_syms']} symbols, "
             f"{pair_counts['A']} pairs",
             f"- same-date/diff-amount (B): {cat_counts['B_syms']} symbols, "
             f"{pair_counts['B']} pairs",
             f"- within-file same-date (C): {cat_counts['C_syms']} symbols, "
             f"{pair_counts['C']} extra rows", "",
             "| symbol | A | B | C | detail (first 3 of each) |",
             "|---|---|---|---|---|"]
    for sym, r in sorted(results.items(),
                         key=lambda kv: -(len(kv[1]["same_amt"])
                                          + len(kv[1]["diff_amt"]))):
        da = ", ".join("%s %s" % (x["date"], x["amount"]) for x in r["same_amt"][:3])
        db = ", ".join("%s div=%s capg=%s" % (x["date"], x["div"], x["capg"])
                       for x in r["diff_amt"][:3])
        dc = ", ".join("%s %s %s" % (x["file"], x["date"], x["amounts"])
                       for x in r["within_file"][:3])
        det = f"A: {da}; B: {db}; C: {dc}"
        lines.append(f"| {sym} | {len(r['same_amt'])} | {len(r['diff_amt'])} "
                     f"| {len(r['within_file'])} | {det} |")
    (OUT / "scan.md").write_text("\n".join(lines) + "\n")
    print(json.dumps(summary, indent=1))
    print(f"-> {OUT/'scan.md'}")
    return 0


if __name__ == "__main__":
    sys.exit(main())