"""Tax-location plan: which funds belong in the TAXABLE account vs an IRA. The user's premise: current LTCG rate < ordinary-income rate expected after retirement. So a fund's right home depends on the CHARACTER of its distributions: qualified dividends + LTCG -> TAXABLE (the LTCG rate is the benefit) tax-exempt interest (munis) -> TAXABLE (wasted in an IRA) ordinary interest / STCG / non-qualified (REIT, K-1) -> IRA (deferring it is the benefit) cash -> FLEXIBLE (no character to place) We don't have 1099-DIV characterizations on file for 2,400 funds, so this is a STRUCTURAL estimate from what the fund actually holds: 1. 16-fund shortlist -> nport_cache/*.json buckets (keyword-based) 2. 22 cross-checked -> xcheck_report.json buckets (SEC assetCat/issuerCat codes, precise) 3. 250 candidates -> factor_results.json sleeve loadings (proxy: a fund that trades like 90% Treasuries earns ~90% ordinary income) plus name-based strategy overrides (merger arb gains are mostly SHORT-TERM, munis are tax-exempt, FOFs are pass-through, ...). The output says where a fund's income character points; the 1099-DIV for the most recent year is the final arbiter for MIXED funds. Run: python -m fundlab.taxplan Output: fundlab/taxplan_results.json """ from __future__ import annotations import json import re from pathlib import Path HERE = Path(__file__).parent EXTRA_BETAS = HERE / "universe_cache" / "betas_extra.json" SHORTLIST = HERE / "decompose_results.json" XCHECK = HERE / "xcheck_report.json" FACTORS = HERE / "factor_results.json" RESULTS = HERE / "taxplan_results.json" # ------------------------------------------------------------------------ # bucket -> (fraction of distributions that are tax-favorable, character) # tax-favorable = qualified dividends + LTCG + tax-exempt interest. # A None fraction means "pass-through / unclassified - unknown". # ------------------------------------------------------------------------ BUCKET_FRAC: dict[str, tuple[float | None, str]] = { # equity: qualified dividends + LTCG "Equity (US)": (1.0, "qualified div + LTCG"), "Equity (intl)": (1.0, "qualified div + LTCG"), "Equity (common)": (1.0, "qualified div + LTCG"), "EC": (1.0, "qualified div + LTCG"), # tax-exempt "Municipal bond": (1.0, "tax-exempt interest"), # mostly non-qualified "Real estate": (0.15, "REIT income (mostly ordinary)"), "RE": (0.15, "REIT income (mostly ordinary)"), "Preferred stock": (0.15, "preferred div (mostly ordinary)"), "EP": (0.15, "preferred div (mostly ordinary)"), # commodities: 60/40 LTCG if section 1256 futures "Commodities": (0.60, "commodity (60/40 if 1256)"), "Commodity": (0.60, "commodity (60/40 if 1256)"), "COMM": (0.60, "commodity (60/40 if 1256)"), # derivatives / structured "Futures": (0.40, "derivatives (mixed)"), "Derivative / hedge": (0.40, "derivatives (mixed)"), "Structured note": (0.30, "structured (mixed)"), "SN": (0.30, "structured (mixed)"), # debt: ordinary interest "US Treasury": (0.0, "ordinary interest"), "US agency": (0.0, "ordinary interest"), "US GSE (Fed/FF)": (0.0, "ordinary interest"), "Corporate bond": (0.05, "ordinary interest"), "Debt": (0.05, "ordinary interest"), "Foreign sovereign": (0.0, "ordinary interest"), "High yield": (0.0, "ordinary interest"), "Loan (leveraged/private credit)": (0.0, "ordinary / K-1"), "LON": (0.0, "ordinary / K-1"), "CLO (collateralized debt)": (0.0, "ordinary interest"), "ABS-CBDO": (0.0, "ordinary interest"), "ABS-O": (0.0, "ordinary interest"), "ABS-MBS": (0.0, "ordinary interest"), "ABS-APCP": (0.0, "ordinary interest"), "ABS other (CLO/CMBS/AB)": (0.0, "ordinary interest"), "ABS auto/personal loan": (0.0, "ordinary interest"), "MBS (mortgage-backed)": (0.0, "ordinary interest"), "Agency MBS": (0.0, "ordinary interest"), "DBT": (0.05, "debt (ordinary)"), # cash: ordinary, no placement value "Cash & T-bills": (0.0, "cash (ordinary)"), "Cash/MMF (short-term)": (0.0, "cash (ordinary)"), "STIV": (0.0, "cash (ordinary)"), "Repurchase agreement": (0.0, "ordinary interest"), "RA": (0.0, "ordinary interest"), # pass-through / unclassified "Fund holdings": (None, "FOF - pass-through"), "Fund/ETF holdings": (None, "FOF - pass-through"), "PF": (None, "private fund - pass-through"), "RF": (None, "fund - pass-through"), # nport.py lump: corporates + munis together "IG credit / munis": (0.45, "IG credit + munis (mixed)"), "Other": (None, "unclassified"), } # ------------------------------------------------------------------------ # sleeve -> same fraction (proxy for the candidate set, no N-PORT on # file). Equity sleeves = qualified div + LTCG; bond sleeves = ordinary; # commodity/CTA sleeves = 60/40 if section 1256. # ------------------------------------------------------------------------ SLEEVE_FRAC: dict[str, float] = { "qqq": 1.0, "ivv": 1.0, "iwm": 1.0, "vea": 1.0, "efa": 1.0, "vwo": 1.0, "vug": 1.0, "vtv": 1.0, "xlk": 1.0, "xlf": 1.0, "xle": 0.90, "xlv": 0.95, "xlp": 1.0, "xlu": 1.0, "xly": 0.95, "xlb": 0.95, "vnq": 0.15, "bil": 0.0, "shv": 0.0, "shy": 0.0, "ief": 0.0, "tlt": 0.0, "agg": 0.0, "lqd": 0.05, "hyg": 0.0, "pff": 0.15, "emb": 0.0, "tip": 0.10, "vweax": 0.0, "vmbix": 0.0, "finux": 0.0, "fxe": 0.0, "fxy": 0.0, "vblix": 0.0, "djp": 0.60, "gsg": 0.60, "gld": 0.60, "dbb": 0.60, "dbmf": 0.60, } # name-based strategy overrides: (regex, action, note) # action: "cap" -> score capped at the given number; None -> note only OVERRIDES: list[tuple[re.Pattern, float | None, str]] = [ (re.compile(r"\bmerger\b", re.I), 0.35, "merger arb: gains are largely SHORT-TERM (deals close <1 yr) - " "1099 will show STCG despite the equity book"), (re.compile(r"style premia|style and valuation", re.I), 0.50, "long/short factor strategy: gains mix STCG/LTCG - check 1099"), (re.compile(r"event[- ]?driven", re.I), 0.45, "event-driven: gains mostly short-term"), (re.compile(r"market neutral", re.I), 0.35, "market-neutral: gains from short-dated option/systematic trades " "- often STCG"), (re.compile(r"global macro", re.I), None, "macro: 60% LTCG if section-1256 futures; OTC swaps -> STCG - " "check 1099"), (re.compile(r"managed futures|\bCTA\b|systematic", re.I), None, "CTA/systematic: 60/40 if section-1256 regulated futures"), (re.compile(r"hedge (fund|strategy|strategies)", re.I), 0.50, "hedge fund: gains often short-term - check 1099"), (re.compile(r"absolute return", re.I), None, "absolute-return: character varies - check 1099"), (re.compile(r"multi[- ]?asset", re.I), None, "multi-asset: mixed qualified/LTCG + ordinary interest - " "check 1099"), (re.compile(r"fund of funds", re.I), None, "FOF: pass-through of underlying character - check 1099"), ] MUNI_RX = re.compile(r"municipal|tax[- ]?exempt|munis\b", re.I) MMF_RX = re.compile(r"money market", re.I) TAXABLE_AT = 0.60 # score >= this -> taxable IRA_AT = 0.35 # score <= this -> IRA FOF_UNKNOWN_AT = 0.40 # >40% pass-through -> can't clear either bar def bucket_score(buckets: list[dict]) -> tuple[float, float, list[str]]: """(score, unknown_frac, unknown_names) from N-PORT-style buckets.""" score = 0.0 known = 0.0 unknown = 0.0 unk_names: list[str] = [] for b in buckets: pct = (b.get("pct") or 0.0) / 100.0 frac, _ = BUCKET_FRAC.get(b["name"], (None, b["name"])) if frac is None: unknown += pct if pct >= 0.05: unk_names.append(f"{b['name']} {pct*100:.0f}%") else: score += frac * pct known += pct if known > 0: score /= known # renormalize over classified assets return score, unknown, unk_names def sleeve_score(betas: dict) -> float | None: """Score from sleeve loadings (proxy when no N-PORT is on file).""" num = 0.0 den = 0.0 for s, b in (betas or {}).items(): if not isinstance(b, (int, float)) or b <= 0: continue f = SLEEVE_FRAC.get(s) if f is None: continue num += f * b den += b if den < 0.05: return None return num / den def classify(name: str, buckets: list[dict] | None = None, betas: dict | None = None) -> dict: """Location decision for one fund. Returns a result dict.""" notes: list[str] = [] unknown = 0.0 used_sleeves = False if buckets: score, unknown, unk_names = bucket_score(buckets) basis = "N-PORT" if unk_names: notes.append("unclassified: " + ", ".join(unk_names)) # keyword parser left most of the book unclassified ("Other") - # trust the return sleeves instead if unknown > 0.5 and betas: s2 = sleeve_score(betas) if s2 is not None: score, basis, used_sleeves = s2, "N-PORT+sleeves", True notes.append("holdings mostly unclassified - used " "return sleeves") else: score = sleeve_score(betas) basis = "sleeves" if score is None: score = 0.5 notes.append("no clean sleeve match - treat as mixed") score = min(max(score, 0.0), 1.0) is_muni = bool(MUNI_RX.search(name or "")) is_mmf = bool(MMF_RX.search(name or "")) if is_muni: score = 1.0 # all distributions are tax-exempt = tax-favorable # strategy overrides cap: float | None = None for rx, act, note in OVERRIDES: if rx.search(name or ""): notes.append(note) if act is not None: cap = min(cap, act) if cap is not None else act if cap is not None: score = min(score, cap) # location if is_muni: loc = "TAXABLE (munis)" notes.append("tax-exempt interest - keep OUT of the IRA") elif is_mmf: loc = "FLEXIBLE (cash)" notes.append("ordinary interest, no placement value either way") elif not used_sleeves and unknown > FOF_UNKNOWN_AT: loc = "MIXED (check 1099)" notes.append(f"{unknown*100:.0f}% pass-through/unclassified - " "character is the underlying funds'") elif score >= TAXABLE_AT: loc = "TAXABLE" elif score <= IRA_AT: loc = "IRA" else: loc = "MIXED (check 1099)" return {"name": name, "basis": basis, "score": round(score, 2), "unknown": round(unknown, 2), "location": loc, "notes": "; ".join(dict.fromkeys(notes))} # Manual overrides for funds the structural data can't resolve - each # with the reason it's needed. (LCORX/LCRIX are new share classes with # no return history; the NPORT shows they are a 91.7% wrapper around the # Leuthold Core ETF, a US equity fund.) MANUAL: dict[str, tuple[str, str]] = { "LCORX": ("TAXABLE", "wrapper: 91.7% Leuthold Core ETF (US equity) " "+ 8% money market"), "LCRIX": ("TAXABLE", "wrapper: 91.7% Leuthold Core ETF (US equity) " "+ 8% money market"), } SPLIT = HERE / "taxsplit_results.json" APPR_SHARE_MIN = 0.50 # >= half of the 5y return is price appreciation def _splits() -> dict: """flattened taxsplit results: sym -> split dict (all groups).""" if not SPLIT.exists(): return {} d = json.loads(SPLIT.read_text()) out: dict = {} for grp in d.values(): for s, v in grp.items(): if v: out.setdefault(s.upper(), v) return out def finalize(sym: str, r: dict, splits: dict | None = None) -> dict: s = (splits or _splits()).get(sym.upper()) # an ACCUMULATOR: most of its return is price appreciation, realized # as the INVESTOR'S own LTCG on a >1y sale (the distributions are # small, so the annual ordinary/STCG drag is small too). That is a # taxable-account profile even when the distribution character is # murky. if (s and s.get("appr_share") is not None and s["appr_share"] >= APPR_SHARE_MIN and r["location"] in ("IRA", "MIXED (check 1099)")): r["location"] = "TAXABLE (accrues)" r["notes"] = (f"{s['appr_share']*100:.0f}% of 5y return is price " f"appreciation (only {s['payout_12m']*100:.1f}% " f"payout in the last 12m) - the gain is YOURS on a " ">1y sale, at the LTCG rate. " + r["notes"]).strip(" ;") if sym.upper() in MANUAL: loc, note = MANUAL[sym.upper()] r["location"] = loc r["score"] = 1.0 if loc == "TAXABLE" else 0.0 r["basis"] = (r["basis"] + "+manual").lstrip("+") r["notes"] = (note + "; " + r["notes"]).strip("; ") return r # ------------------------------------------------------------------------ def run() -> dict: # the appreciation split must exist before finalize() can use it if not SPLIT.exists(): from fundlab import taxsplit taxsplit.run() # sleeve loadings as a fallback for groups whose N-PORT parse left # the book mostly unclassified. The 16-fund shortlist was never in # the 2,384 screen, so compute + cache betas for those on demand. fac = json.loads(FACTORS.read_text()) extra = (json.loads(EXTRA_BETAS.read_text()) if EXTRA_BETAS.exists() else {}) def betas_of(sym: str) -> dict: v = fac.get(sym.lower()) or fac.get(sym.upper()) or {} b = (v.get("full") or {}).get("betas") if b: return b key = sym.lower() if key in extra: return extra[key] from fundlab import factors r = factors.factor_screen(sym) if r and r.get("full"): extra[key] = r["full"]["betas"] EXTRA_BETAS.parent.mkdir(exist_ok=True) EXTRA_BETAS.write_text(json.dumps(extra, indent=1)) return extra.get(key) or {} # 16-fund shortlist (N-PORT keyword buckets + names from funds.json) funds = json.loads((HERE.parent / "funds.json").read_text()) shortlist: dict = {} dr = json.loads(SHORTLIST.read_text()) for sym in dr: snap = json.loads((HERE / "nport_cache" / f"{sym}.json").read_text()) name = (funds.get(sym.lower()) or {}).get("name") or sym.upper() r = classify(name, buckets=snap.get("buckets") or None, betas=betas_of(sym)) r["as_of"] = snap.get("as_of", "") shortlist[sym.upper()] = finalize(sym, r) # 22 cross-checked (precise code-based buckets) xcheck: dict = {} xc = json.loads(XCHECK.read_text()) for sym, v in xc.items(): r = classify(v.get("name") or sym.upper(), buckets=v.get("buckets") or None, betas=betas_of(sym)) r["as_of"] = v.get("as_of", "") xcheck[sym.upper()] = finalize(sym, r) # 250 candidates (sleeve loadings + name) cands: dict = {} for sym, v in fac.items(): if not (v.get("verdict") or "").startswith("CANDIDATE"): continue betas = (v.get("full") or {}).get("betas") or {} cands[sym.upper()] = finalize(sym, classify(v.get("name") or "", betas=betas)) res = {"shortlist": shortlist, "xcheck": xcheck, "candidates": cands} RESULTS.write_text(json.dumps(res, indent=1)) _print(res) return res def _print(res: dict) -> None: from collections import Counter def row(s: str, r: dict) -> str: return (f" {s:<7} {r['location']:<20} {r['score']:<5.2f} " f"{r['basis']:<14} {r['name'][:34]:<34} {r['notes'][:64]}") hdr = (f" {'fund':<7} {'location':<20} score basis " f"{'name':<34} notes") for key, title in (("shortlist", "16-fund shortlist"), ("xcheck", "22 cross-checked")): funds = res[key] print(f"\n== {title} ({len(funds)}) ==") print(hdr) for s, r in sorted(funds.items()): print(row(s, r)) c = res["candidates"] locs = Counter(r["location"] for r in c.values()) print(f"\n== 250 candidates: {dict(locs)} ==") sections = [ ("TAXABLE - equity character (by score)", lambda r: r["location"] == "TAXABLE", -18, 18), ("TAXABLE - munis (sample)", lambda r: r["location"] == "TAXABLE (munis)", 0, 12), ("MIXED - check the 1099 (by score)", lambda r: r["location"] == "MIXED (check 1099)", -12, 12), ("IRA - most ordinary income (by score)", lambda r: r["location"] == "IRA", 12, 12), ("FLEXIBLE - cash", lambda r: r["location"] == "FLEXIBLE (cash)", 0, 8), ] for title, pick, sortkey, limit in sections: sel = [(s, r) for s, r in c.items() if pick(r)] if sortkey: sel.sort(key=lambda kv: (sortkey * kv[1]["score"], kv[0])) else: sel.sort(key=lambda kv: kv[0]) print(f"\n -- {title} ({len(sel)})") print(hdr) for s, r in sel[:limit]: print(row(s, r)) if __name__ == "__main__": run()