Greg Pomerantz
d1e85026bf
fundlab/cef_universe.py: SEC 'Closed-End Fund Information' report (973 active CEFs) + company_tickers CIK join -> 295 listed common classes (preferreds and 6(c)-converted companies flagged). Supersedes the N-2 full-index approach (N-2/A annual updates + 404ing index paths). Prices via goget. fundlab/cef_screen.py: 290 screened - t5/t12, vol5, maxdd5, the 5 crash episodes, 12m payout proxy. Energy/midstream CEFs (EMO +305%, SRV +234%, NML +232%) top return AND crash resilience; Voya Dividend-Premium series (IGD maxDD -16%); EM CEFs volatile + 14-24% dist; long-dur munis -26..-42%. fundlab/cef_character.py: 50-fund shortlist, 35-sleeve character + crude tax_arb = character x (upside + 0.4 x vol). Tests: 6 new cef checks (105 total). RESEARCH.md: CEF form facts (N-2ASR, N-PX, N-23C-3A, BDC caveats) + remaining stage 2b work.
117 lines
4.1 KiB
Python
117 lines
4.1 KiB
Python
"""CEF stage 2a: tax-character proxy for the shortlist.
|
|
|
|
Takes the cef_screen.json shortlist (top 5y return + crash-resilient +
|
|
high-payout CEFs), runs the 35-sleeve factor screen on each, and scores
|
|
the sleeve mix with taxplan.sleeve_score (same proxy the open-end
|
|
candidates use). Then ranks by a crude "tax-arb per dollar" heuristic
|
|
for a constrained taxable account:
|
|
|
|
tax_arb ~= character x (upside + 0.4 x vol)
|
|
|
|
character share of the return mix that is tax-favorable (0..1)
|
|
upside max(t5,0)/5 - the LTCG-deferral term
|
|
0.4 x vol5 ~ expected |annual loss| (normal approx) - the
|
|
harvest term (0.4 = 1/sqrt(2*pi))
|
|
|
|
Both terms are per dollar of balance; the heuristic is a SCREENER,
|
|
the EDGAR deep dive (N-2ASR/N-CSR distribution character + NPORT NAV
|
|
for the discount) is what decides.
|
|
|
|
Output: fundlab/cef_character.json
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
|
|
from fundlab import factors, taxplan
|
|
|
|
HERE = Path(__file__).parent
|
|
SCREEN = HERE / "cef_screen.json"
|
|
OUT = HERE / "cef_character.json"
|
|
|
|
# 6(c)-converted / non-fund tickers that show up in the SEC CEF report
|
|
# (organized as CEFs, later converted to operating companies)
|
|
NON_FUNDS = {"cfnd", "pwrl", "bot", "fxby"}
|
|
SPECULATIVE = re.compile(r"corp\.?$|tech100", re.I)
|
|
|
|
|
|
def select(res: dict, n_ret: int = 40, n_pay: int = 15) -> list[str]:
|
|
rows = [v for v in res.values()
|
|
if v.get("t5") is not None and v["sym"] not in NON_FUNDS]
|
|
by_ret = sorted(rows, key=lambda x: -x["t5"])[:n_ret]
|
|
by_res = [v for v in rows if v.get("n_pos_scen", 0) >= 3]
|
|
by_pay = sorted(rows, key=lambda x: -(x.get("payout_12m") or 0))[:n_pay]
|
|
out: dict[str, dict] = {}
|
|
for v in by_ret + by_res + by_pay:
|
|
out.setdefault(v["sym"], v)
|
|
return list(out)
|
|
|
|
|
|
def _betas(sym: str) -> dict:
|
|
fs = factors.factor_screen(sym)
|
|
if not fs:
|
|
return {}
|
|
return (fs.get("full") or fs.get("rec5") or {}).get("betas") or {}
|
|
|
|
|
|
def run() -> dict:
|
|
res = json.loads(SCREEN.read_text())
|
|
syms = select(res)
|
|
print(f"{len(syms)} shortlisted CEFs", flush=True)
|
|
out: dict = {}
|
|
for i, s in enumerate(syms, 1):
|
|
v = res[s.upper()]
|
|
comps = _betas(s.lower())
|
|
char = taxplan.sleeve_score(comps) if comps else None
|
|
t5 = v.get("t5") or 0.0
|
|
vol = v.get("vol5") or 0.0
|
|
upside = max(t5, 0.0) / 5.0
|
|
harvest = 0.4 * vol
|
|
arb = (char if char is not None else 0.3) * (upside + harvest)
|
|
out[s] = {
|
|
**{k: v.get(k) for k in ("name", "bdc", "days", "last")},
|
|
"sym": s, "t5": t5, "t12": v.get("t12"), "vol5": vol,
|
|
"maxdd5": v.get("maxdd5"),
|
|
"2022": v.get("2022 bear mkt"), "2023": v.get("2023 rate shock"),
|
|
"2025t": v.get("2025 tariff crash"),
|
|
"2026": v.get("2026 Q1 drawdown"),
|
|
"n_pos_scen": v.get("n_pos_scen"),
|
|
"payout_12m": v.get("payout_12m"),
|
|
"character": round(char, 2) if char is not None else None,
|
|
"sleeves": {k: round(b, 2) for k, b in comps.items()
|
|
if abs(b) > 0.15},
|
|
"tax_arb": round(arb, 4),
|
|
}
|
|
print(f"{i:2}/{len(syms)} {s:7} char={out[s]['character']} "
|
|
f"arb={out[s]['tax_arb']}", flush=True)
|
|
OUT.write_text(json.dumps(out, indent=1, default=str))
|
|
print(f"wrote {OUT}")
|
|
return out
|
|
|
|
|
|
def _pc(v: float | None) -> str:
|
|
return "" if v is None else f"{v:+.1%}"
|
|
|
|
|
|
def _print(out: dict, top: int = 35) -> None:
|
|
rows = sorted(out.values(), key=lambda x: -x["tax_arb"])
|
|
print(f"{'fund':7} {'name':40} {'char':>5} {'5yTR':>8} {'vol':>6}"
|
|
f" {'maxDD':>8} {'2022':>8} {'2026':>8} {'dist12':>8} {'tax_arb':>8}")
|
|
for v in rows[:top]:
|
|
c = v["character"]
|
|
print(f"{v['sym']:7} {v.get('name','')[:40]:40} "
|
|
f"{('' if c is None else f'{c:.2f}'):>5} "
|
|
f"{v['t5']:+8.1%} {v['vol5']:6.2f} "
|
|
f"{_pc(v.get('maxdd5')):>8} {_pc(v.get('2022')):>8} "
|
|
f"{_pc(v.get('2026')):>8} {_pc(v.get('payout_12m')):>8} "
|
|
f"{v['tax_arb']:8.3f}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
o = run()
|
|
_print(o)
|