f/fundlab/taxsplit.py

"""Price appreciation vs. payout split, for tax placement.

The fund price files carry two series:

  Close     = raw NAV with distributions PAID OUT (price appreciation)
  Adj Close = total return with distributions REINVESTED

So for each fund we can decompose its total return into:

  price appreciation (realized as LTCG when you SELL the shares)
  payout component    (taxed every year when it is DISTRIBUTED)

That is exactly the trade-off behind the taxable-vs-IRA placement:
the payout is the recurring tax drag in a taxable account, the
appreciation is deferred (LTCG if held >1y).  A fund whose returns
come mostly as payouts is the one that hurts in a taxable account;
a fund that accumulates (appreciates) defers the tax.

Caveats:
- "payout component" includes dividends, interest AND capital-gain
  distributions; it does not tell you their character (that needs the
  1099-DIV).
- arithmetic annualization: dist_a = tr_a - pr_a is an approximation
  (fine while yields are <~15%/yr).
- NAV gaps: funds with <250 observations in the window are skipped.

Run:  python -m fundlab.taxsplit
Output: fundlab/taxsplit_results.json
"""
from __future__ import annotations

import json
from pathlib import Path

import pandas as pd

HERE = Path(__file__).parent
RESULTS = HERE / "taxsplit_results.json"
STOCKS = Path.home() / "prog" / "fin" / "stocks"

W_5Y = "2021-01-01"


def fund_split(sym: str, start: str | None) -> dict | None:
    """(price- and total-return-based split) over a window."""
    p = STOCKS / f"{sym.lower()}-history.csv"
    if not p.exists():
        return None
    d = pd.read_csv(p, parse_dates=["Date"])
    if start:
        d = d[d["Date"] >= start]
    d = d.dropna(subset=["Close", "Adj Close"]).sort_values("Date")
    if len(d) < 250:
        return None
    yrs = (d["Date"].iloc[-1] - d["Date"].iloc[0]).days / 365.25
    pr = float(d["Close"].iloc[-1] / d["Close"].iloc[0] - 1)
    tr = float(d["Adj Close"].iloc[-1] / d["Adj Close"].iloc[0] - 1)
    pr_a = (1 + pr) ** (1 / yrs) - 1
    tr_a = (1 + tr) ** (1 / yrs) - 1
    dist_a = tr_a - pr_a                      # annualized payout, approx
    # share of the ANNUALIZED total return that is price appreciation
    appr_share = pr_a / tr_a if tr_a > 0.005 else None
    # most-recent-12m payout: the CURRENT distribution behavior is what
    # matters for placement - the 5y average can be skewed by one-time
    # events (special distributions, share-class reorganizations)
    d12 = d[d["Date"] >= d["Date"].iloc[-1] - pd.DateOffset(months=12)]
    p12 = float(d12["Close"].iloc[-1] / d12["Close"].iloc[0] - 1)
    t12 = float(d12["Adj Close"].iloc[-1] / d12["Adj Close"].iloc[0] - 1)
    payout_12m = t12 - p12
    return {"yrs": round(yrs, 2),
            "tot": round(tr, 4), "price": round(pr, 4),
            "tot_a": round(tr_a, 4), "price_a": round(pr_a, 4),
            "payout_a": round(dist_a, 4),
            "payout_12m": round(payout_12m, 4),
            "appr_share": (round(appr_share, 3) if appr_share is not None
                           else None)}


def _groups() -> dict[str, list[str]]:
    """The three fund groups, from the primary sources (not from
    taxplan's output - taxplan depends on THIS file)."""
    dr = json.loads((HERE / "decompose_results.json").read_text())
    xc = json.loads((HERE / "xcheck_report.json").read_text())
    fac = json.loads((HERE / "factor_results.json").read_text())
    return {
        "shortlist": sorted(s.upper() for s in dr),
        "xcheck": sorted(s.upper() for s in xc),
        "candidates": sorted(s.upper() for s, v in fac.items()
                             if (v.get("verdict") or "").startswith(
                                 "CANDIDATE")),
    }


def run() -> dict:
    out: dict = {}
    for grp, syms in _groups().items():
        out[grp] = {}
        for s in syms:
            r5 = fund_split(s, W_5Y)
            if r5:
                out[grp][s] = r5
    RESULTS.write_text(json.dumps(out, indent=1))
    _print(out)
    return out


def _print(out: dict) -> None:
    def fmt(r: dict | None) -> str:
        if r is None:
            return "  (no data)"
        sh = "  -  " if r["appr_share"] is None else f"{r['appr_share']*100:3.0f}%"
        return (f"5y tot {r['tot']*100:+7.1f}%  price {r['price']*100:+7.1f}%  "
                f"payout {r['payout_a']*100:4.1f}%/yr  appr-share {sh}")

    for grp in ("shortlist", "xcheck"):
        print(f"\n== {grp} ==")
        for s in sorted(out[grp]):
            print(f"  {s:<7} {fmt(out[grp][s])}")

    c = out["candidates"]
    have = [s for s in c if c[s] is not None]
    # rank by appreciation share: who ACCUMULATES vs who PAYS OUT
    ranked = sorted((s for s in have),
                    key=lambda s: -(c[s]["appr_share"] or -1))
    print(f"\n== candidates: most appreciation (accumulate) vs "
          f"most payout (pay out) ==")
    print("  top 15 by appreciation share of total return:")
    for s in ranked[:15]:
        print(f"  {s:<7} {fmt(c[s])}")
    print("  bottom 15 (return comes almost entirely as payouts):")
    for s in ranked[-15:]:
        print(f"  {s:<7} {fmt(c[s])}")


if __name__ == "__main__":
    run()