Greg Pomerantz
22faeef2ec
Per-share financial-highlights parser (cef_annual.py) now reconciles divs+gains+ROC==distributions and the full NAV chain per column for 48 of 50 shortlist funds, from three layout families (inline, transposed, + repair rules for footnotes/split-decimals/sign- magnitude/combined-rows). DXYZ is a correct zero-distribution result; STEW is a documented columnar-stream exception. cef_merged.json character column now uses actual data. 123/123 tests.
983 lines
39 KiB
Python
983 lines
39 KiB
Python
"""CEF stage 2b: per-share financial highlights from the annual report.
|
|
|
|
For each CEF, fetch the recent N-CSR/N-CSRS submissions and parse the
|
|
per-share "Financial Highlights" table. CEF tables come in two
|
|
families (verified on EMO / TWN):
|
|
|
|
Less distributions to common shareholders from: (Franklin style)
|
|
Dividends ...
|
|
Return of capital ...
|
|
Total distributions to common shareholders ...
|
|
Less Distributions to Stockholders from: (classic CEF)
|
|
Net investment income ...
|
|
Net realized gains ...
|
|
Total distributions to stockholders ...
|
|
|
|
Gotchas handled:
|
|
- Blanks are dropped when the table flattens: a short row's values can
|
|
sit in ANY columns. EMO's ROC row (3.42, 0.93) belongs to columns
|
|
0 and 4, not 0 and 1; TWN's total-distributions row has 4 values
|
|
for 5 columns (a zero year dropped).
|
|
- The robust total per column comes from the NAV IDENTITY, not the
|
|
printed total row: dist[c] = NAV_beg[c] + ops[c] + captch[c]
|
|
- NAV_end[c]. The printed sub-rows (NII / gains / ROC) are then
|
|
placed combinatorially against those derived totals.
|
|
- The first "Net asset value, beginning of year" anchor in the doc is
|
|
the most recent 5-year table (a continuation table for older years
|
|
follows it). A "share" proximity guard skips dollar-basis mentions.
|
|
- Annual vs semi-annual: try the 3 most recent shareholder reports and
|
|
keep the one that parses with the most columns (annual = 5).
|
|
|
|
Also captured (optional lines): market value end of year (-> discount
|
|
series), leverage (debt + preferred vs net assets), tax-paying flag.
|
|
|
|
Run: python -m fundlab.cef_annual [SYM ...] (default: stage-2a
|
|
shortlist)
|
|
Output: fundlab/cef_annual.json (cached per fund in cef_cache/)
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
|
|
from fundlab import edgar
|
|
|
|
HERE = Path(__file__).parent
|
|
CACHE = HERE / "cef_cache"
|
|
OUT = HERE / "cef_annual.json"
|
|
CHARACTER = HERE / "cef_character.json"
|
|
|
|
# distribution-character fractions for the placement score (1 = fully
|
|
# taxable-account favorable)
|
|
FRAC_NII = 0.10 # ordinary income (equity-fund NII is partly QD)
|
|
FRAC_GAINS = 0.70 # mix of STCG/LTCG - long-term equity funds skew LT
|
|
FRAC_ROC = 1.0 # defers to the investor's own LTCG on a >1y sale
|
|
FRAC_NII_MUNI = 1.0 # tax-exempt interest
|
|
|
|
|
|
NUMTOK = re.compile(r"\$?\(?(?:\d[\d,]*(?:\.\d+)?|\.\d+)\)?")
|
|
FOOTNOTE = re.compile(r"\([a-z0-9]{1,3}\)")
|
|
|
|
|
|
def _nums(tokens: list[str]) -> list[float | None]:
|
|
out = []
|
|
for t in tokens:
|
|
t = t.strip().rstrip(":")
|
|
if not re.fullmatch(r"\$?\(?(?:\d[\d,]*(?:\.\d+)?|\.\d+)\)?", t):
|
|
out.append(None)
|
|
continue
|
|
neg = t.startswith("(")
|
|
v = float(t.strip("$()").replace(",", ""))
|
|
out.append(-v if neg else v)
|
|
return out
|
|
|
|
|
|
def _row(block: str, label_rx: str, span: int = 700) -> list[float | None]:
|
|
"""Numeric row following the first label match in the block."""
|
|
rm = re.search(label_rx, block, re.I)
|
|
if not rm:
|
|
return []
|
|
tail = block[rm.end(): rm.end() + span]
|
|
tail = re.sub(r"\(\s+", "(", tail)
|
|
tail = re.sub(r"\s+\)", ")", tail)
|
|
tail = re.sub(r"(?<![\d.])(\d) \. ?(\d)", r"\1.\2", tail)
|
|
vals: list[str] = []
|
|
for t in re.findall(r"\S+", tail):
|
|
if t == "$":
|
|
continue
|
|
if (NUMTOK.fullmatch(t)
|
|
and not re.fullmatch(r"\(\d{1,2}\)", t)
|
|
and not re.fullmatch(r"\d{1,2}", t)):
|
|
vals.append(t)
|
|
elif FOOTNOTE.fullmatch(t):
|
|
continue # footnote markers sit between values
|
|
elif re.fullmatch(r"\(\d{1,2}\)", t):
|
|
continue # numeric footnote marker like (3)
|
|
elif re.fullmatch(r"\d{1,2}", t):
|
|
continue # bare footnote digit (EMO: "(2.16) 4 (3.42)")
|
|
elif re.fullmatch(r"[,.;:()]+", t):
|
|
continue # bare punctuation between values
|
|
elif vals:
|
|
break # first word after values = next label
|
|
return _nums(vals[:10])
|
|
|
|
|
|
|
|
def _fixlen(row: list, ncols: int) -> list:
|
|
"""Footnote integers (bare 1-9) can inflate a row (RMT '1 , 2',
|
|
NICHX/VCRDX leading '1'). Dropping them is only accepted when it
|
|
lands exactly on ncols; the per-column arithmetic re-validates."""
|
|
if len(row) <= ncols:
|
|
return row
|
|
kept = [v for v in row
|
|
if not (v is not None and v == int(v) and 0 < abs(v) < 10)]
|
|
return kept if len(kept) == ncols else row
|
|
|
|
|
|
def _perms4(k):
|
|
"""Ordered assignments of k values to distinct slots of 4."""
|
|
for a in range(4):
|
|
if k == 1:
|
|
yield (a,)
|
|
continue
|
|
for b in range(4):
|
|
if b == a:
|
|
continue
|
|
if k == 2:
|
|
yield (a, b)
|
|
continue
|
|
for c in range(4):
|
|
if c != a and c != b:
|
|
yield (a, b, c)
|
|
|
|
|
|
def _align_trans_row(vals: list, has_roc: bool, has_gains: bool) -> tuple | None:
|
|
"""Align one transposed data row (newest period first).
|
|
|
|
v0=nav_beg v1=nii v2=gains v3=ops, then the distribution sub-values
|
|
and total. Three label layouts:
|
|
I1 (default, Voya w/ dropped ROC): v4=divs v5=gdist v6=dtot
|
|
I3 (STEW: ROC label, no gains label): v4=divs v5=roc v6=dtot
|
|
I2 (Voya w/ non-zero ROC value): v4=divs v5=gdist v6=roc v7=dtot
|
|
Zero-dropped columns make this ambiguous - resolve by arithmetic:
|
|
divs+gdist+roc == dtot AND nav_end == nav_beg+ops+accretion-dtot.
|
|
Positive outflow convention (transposed tables print positives)."""
|
|
if len(vals) < 7:
|
|
return None
|
|
nav_b = abs(vals[0])
|
|
# NII and gains are SIGNED (loss years are negative) - the ops row
|
|
# is their signed sum; only NAV magnitudes are always positive
|
|
nii, gains, ops = vals[1], vals[2], vals[3]
|
|
if abs(nii + gains - ops) > 0.02 * max(1.0, abs(ops)) + 0.02:
|
|
return None
|
|
interps = []
|
|
if has_roc and not has_gains:
|
|
interps.append((4, None, 6, 5, 7)) # divs, gdist=0, dtot, roc, next
|
|
interps.append((4, 5, 6, None, 7))
|
|
if has_roc:
|
|
interps.append((4, 5, 7, 6, 8))
|
|
strict_fb = None
|
|
for di, gi, ti, ri, nxt in interps:
|
|
if len(vals) < max(ti, ri or 0, gi or 0) + 1:
|
|
continue
|
|
divs = abs(vals[di])
|
|
gdist = 0.0 if gi is None else abs(vals[gi])
|
|
dtot = abs(vals[ti])
|
|
roc = 0.0 if ri is None else abs(vals[ri])
|
|
if abs(divs + gdist + roc - dtot) > 0.03:
|
|
continue
|
|
rest = vals[nxt:]
|
|
if not rest:
|
|
# table ends at the total distributions row (PDX)
|
|
return (nav_b, divs, gdist, roc, dtot, None, None)
|
|
for acc in [0.0] + list(rest):
|
|
nav_e = nav_b + ops + abs(acc) - dtot
|
|
for i, x in enumerate(rest):
|
|
if abs(x - nav_e) <= 0.03:
|
|
mkt_e = rest[i + 1] if i + 1 < len(rest) else None
|
|
return (nav_b, divs, gdist, roc, dtot, nav_e, mkt_e)
|
|
# dist arithmetic is exact but no nav_end pin - keep as fallback
|
|
if strict_fb is None and abs(divs + gdist + roc - dtot) <= 0.011:
|
|
strict_fb = (nav_b, divs, gdist, roc, dtot, None, None)
|
|
|
|
# slot placement: some components dropped entirely (JCE: divs + dtot
|
|
# only, tail = nav_e/mkt_e). Assign the post-ops values to the
|
|
# (divs, gdist, dtot, roc) slots, zeros implied, and require
|
|
# divs+gdist+roc == dtot plus the nav identity on the tail.
|
|
for m in (2, 1, 0):
|
|
tail = vals[-m:] if m else []
|
|
dvals = vals[4: len(vals) - m]
|
|
if not (1 <= len(dvals) <= 3):
|
|
continue
|
|
for perm in _perms4(len(dvals)):
|
|
slots = [0.0, 0.0, 0.0, 0.0]
|
|
for s, v in zip(perm, dvals):
|
|
if slots[s]:
|
|
break
|
|
slots[s] = abs(v)
|
|
else:
|
|
sdivs, sgdist, sdtot, sroc = slots
|
|
if sdtot and abs(sdivs + sgdist + sroc - sdtot) \
|
|
<= 0.02 * max(1.0, sdtot) + 0.02:
|
|
nav_e = None
|
|
if tail:
|
|
pred = nav_b + ops - sdtot
|
|
nav_e = next(
|
|
(abs(x) for x in tail
|
|
if abs(abs(x) - pred) <= 0.05), None)
|
|
if nav_e is None:
|
|
continue
|
|
mkt_e = tail[-1] if tail else None
|
|
return (nav_b, sdivs, sgdist, sroc, sdtot,
|
|
nav_e, mkt_e)
|
|
return strict_fb
|
|
|
|
|
|
def _transposed_rows(after: str, header_end: int, span: int) -> list:
|
|
"""Data rows (list of value lists) after a transposed header.
|
|
Handles dash dates (Voya MM-DD-YY) and slash dates (PIMCO/Nuveen
|
|
MM/DD/YYYY)."""
|
|
dates = [dm for dm in re.finditer(r"\d{1,2}[-/]\d{1,2}[-/]\d{2,4}", after)]
|
|
dates = [dm for dm in dates if len(dm.group(0)) >= 8]
|
|
rows = []
|
|
for k, dm in enumerate(dates):
|
|
stop = (dates[k + 1].start() if k + 1 < len(dates)
|
|
else min(len(after), dm.end() + span))
|
|
seg = after[dm.end():stop]
|
|
seg = re.sub(r"\(\s+", "(", seg)
|
|
seg = re.sub(r"\s+\)", ")", seg)
|
|
seg = re.sub(r"(?<![\d.])(\d) \. ?(\d)", r"\1.\2", seg)
|
|
toks = seg.split()
|
|
vals = _nums([x for x in toks if NUMTOK.fullmatch(x)])
|
|
if len(vals) >= 7:
|
|
rows.append(vals)
|
|
return rows
|
|
|
|
|
|
def _section_split(after: str, span: int):
|
|
"""Split a multi-fund consolidated region into (header, rows)
|
|
sections. A section break is a text run of >= 3 words (a fund
|
|
name) followed by date rows. Returns [(header, rows), ...]."""
|
|
parts = []
|
|
pos = 0
|
|
cur_header = ""
|
|
cur = []
|
|
date_re = re.compile(r"\d{1,2}[-/]\d{1,2}[-/]\d{2,4}")
|
|
while pos < len(after):
|
|
dm = date_re.search(after, pos)
|
|
if not dm:
|
|
break
|
|
between = after[pos:dm.start()]
|
|
words = [w for w in between.split()
|
|
if not NUMTOK.fullmatch(w)
|
|
and not re.fullmatch(r"[()$%,.\-]+", w)
|
|
and re.search(r"[A-Za-z]", w)]
|
|
tickerish = (1 <= len(words) <= 2 and all(
|
|
re.fullmatch(r"[A-Za-z]{1,8}", w) for w in words))
|
|
if words and cur:
|
|
# text run before a date with existing rows: section break
|
|
if len(words) >= 3 or tickerish:
|
|
parts.append((cur_header, cur))
|
|
cur_header = " ".join(words)
|
|
cur = []
|
|
# gather this date row
|
|
stop_m = date_re.search(after, dm.end() + 1)
|
|
stop = stop_m.start() if stop_m else min(len(after), dm.end() + span)
|
|
seg = after[dm.end():stop]
|
|
seg = re.sub(r"\(\s+", "(", seg)
|
|
seg = re.sub(r"\s+\)", ")", seg)
|
|
seg = re.sub(r"(?<![\d.])(\d) \. ?(\d)", r"\1.\2", seg)
|
|
toks = seg.split()
|
|
vals = _nums([x for x in toks if NUMTOK.fullmatch(x)])
|
|
if len(vals) >= 7:
|
|
cur.append(vals)
|
|
pos = stop
|
|
if cur:
|
|
parts.append((cur_header, cur))
|
|
return parts
|
|
|
|
|
|
def _pick_section(parts, name: str, ticker: str):
|
|
"""Choose the section belonging to THIS fund in a consolidated
|
|
report (PIMCO/Nuveen report several funds in one table)."""
|
|
if not parts:
|
|
return None
|
|
if len(parts) == 1:
|
|
return parts[0][1]
|
|
name_words = {w.lower() for w in re.findall(r"[A-Za-z0-9]+", name)
|
|
if len(w) > 2
|
|
and w.lower() not in
|
|
("fund", "funds", "inc", "company", "llc", "corp",
|
|
"corporation", "series", "trust", "the", "of", "and")}
|
|
best, best_score = None, 0
|
|
for header, rows in parts:
|
|
hw = {w.lower() for w in re.findall(r"[A-Za-z0-9]+", header)}
|
|
score = len(hw & name_words)
|
|
if ticker and re.search(r"\b" + re.escape(ticker.upper())
|
|
+ r"\b", header, re.I):
|
|
score += 5
|
|
if score > best_score:
|
|
best, best_score = rows, score
|
|
if best is None or best_score < 2:
|
|
return parts[0][1] # no identifying header: first section
|
|
return best
|
|
|
|
|
|
|
|
|
|
def parse_transposed(t: str, name: str = "", ticker: str = "") -> dict | None:
|
|
def _trans_from(m):
|
|
# the header row ("Year(s) [or Period] Ended") may sit before the
|
|
# label list (PIMCO) or after it (Voya) - take the one closest
|
|
# before the first data row
|
|
window = t[max(0, m.start() - 500): m.start() + 9000]
|
|
datere = r"\d{1,2}[-/]\d{1,2}[-/]\d{2,4}"
|
|
first_date = re.search(datere, window)
|
|
if not first_date:
|
|
return None
|
|
hm = None
|
|
for h in re.finditer(r"years? (?:or period )?ended", window, re.I):
|
|
if h.start() < first_date.start():
|
|
hm = h
|
|
if hm is None:
|
|
# no "Year(s) ended" header (Nuveen JCE): start at the anchor -
|
|
# the label list before it carries no dates
|
|
after = window[m.start() - max(0, m.start() - 500):]
|
|
else:
|
|
after = window[hm.end():]
|
|
has_roc = bool(re.search(r"(?:tax )?return of capital", window, re.I))
|
|
has_gains = bool(re.search(r"from net realized gains", window, re.I))
|
|
parts = _section_split(after, 300)
|
|
rows = _pick_section(parts, name, ticker)
|
|
if not rows or len(rows) < 3:
|
|
return None
|
|
cols = [_align_trans_row(v, has_roc, has_gains) for v in rows]
|
|
# the oldest 1-2 periods are sometimes structurally different
|
|
# (reorganization years) - tolerate trailing failures
|
|
bad = [i for i, c in enumerate(cols) if c is None]
|
|
if bad and any(i < len(cols) - 2 for i in bad):
|
|
return None
|
|
if sum(c is None for c in cols) >= 3:
|
|
return None
|
|
rows = [v for v, c in zip(rows, cols) if c is not None]
|
|
cols = [c for c in cols if c is not None]
|
|
if len(cols) < 3:
|
|
return None
|
|
mkt = [c[6] for c in cols]
|
|
out = {
|
|
"ncols": len(cols),
|
|
"nav_beg": [c[0] for c in cols],
|
|
"nav_end": ([c[5] for c in cols]
|
|
if all(c[5] is not None for c in cols) else []),
|
|
"mkt_end": mkt if all(x is not None for x in mkt) else [],
|
|
"dist_tot": [c[4] for c in cols],
|
|
"divs": [c[1] for c in cols],
|
|
"gains": [c[2] for c in cols],
|
|
"roc": [c[3] for c in cols],
|
|
"ops": [abs(v[3]) for v in rows],
|
|
}
|
|
tot5 = sum(v for v in out["dist_tot"] if v > 0)
|
|
if tot5 > 0:
|
|
out["share_div"] = sum(out["divs"]) / tot5
|
|
out["share_gains"] = sum(out["gains"]) / tot5
|
|
out["share_roc"] = sum(out["roc"]) / tot5
|
|
return out
|
|
|
|
for m in re.finditer(
|
|
r"net asset value(?: per (?:common )?share)?,? beginning of "
|
|
r"(?:the )?(?:year|period)", t, re.I):
|
|
h = _trans_from(m)
|
|
if h:
|
|
return h
|
|
return None
|
|
|
|
|
|
|
|
|
|
def parse_columnar(t: str, name: str = "", ticker: str = "") -> dict | None:
|
|
"""Columnar value stream (SRH/STEW): a LABEL LIST (no values on
|
|
the label rows), then a period header, then one contiguous value run
|
|
per label. Rows are identified arithmetically, not by position."""
|
|
anchor_rx = (r"net asset value(?: per (?:common )?share)?,? beginning of "
|
|
r"(?:the )?(?:year|period)")
|
|
for m in re.finditer(anchor_rx, t, re.I):
|
|
region = t[m.start(): m.start() + 9000]
|
|
toks = re.findall(r"\S+", region)
|
|
# the NAV-begin row = the longest run of numeric tokens ("$" is
|
|
# transparent); period-header year runs (2025 2024 ...) are
|
|
# rejected by magnitude
|
|
run, run_toks, best = [], [], (None, 0)
|
|
for x in toks:
|
|
if x == "$":
|
|
if run:
|
|
run_toks.append(x)
|
|
continue
|
|
if NUMTOK.fullmatch(x) and not re.fullmatch(r"\(\d{1,2}\)", x):
|
|
run.append(x)
|
|
run_toks.append(x)
|
|
else:
|
|
vals = [float(v.strip("$()").replace(",", "")) for v in run]
|
|
if len(run) >= 4 and best[1] < len(run) \
|
|
and all(abs(v) < 100 for v in vals):
|
|
best = (run_toks[:], len(run))
|
|
run, run_toks = [], []
|
|
vals = [float(v.strip("$()").replace(",", "")) for v in run]
|
|
if len(run) >= 4 and best[1] < len(run) \
|
|
and all(abs(v) < 100 for v in vals):
|
|
best = (run_toks[:], len(run))
|
|
if best[0] is None or best[1] < 4:
|
|
continue
|
|
ncols = best[1]
|
|
stream = [x for x in best[0]
|
|
if x != "$" and NUMTOK.fullmatch(x)
|
|
and not re.fullmatch(r"\(\d{1,2}\)", x)]
|
|
stream = stream + [x for x in toks[toks.index(best[0][0]) + len(best[0]):]
|
|
if x != "$" and NUMTOK.fullmatch(x)
|
|
and not re.fullmatch(r"\(\d{1,2}\)", x)]
|
|
rows = [_nums(stream[k: k + ncols])
|
|
for k in range(0, len(stream) - ncols + 1, ncols)]
|
|
if len(rows) < 4:
|
|
continue
|
|
nav_b = rows[0]
|
|
# ops row: nii=rows[1], gains=rows[2] in the standard order;
|
|
# verify nii+gains==ops and locate ops (rows[3], or scan)
|
|
nii, gains = rows[1], rows[2]
|
|
ops = None
|
|
for cand in range(2, min(6, len(rows) - 2)):
|
|
r = rows[cand]
|
|
err = sum(abs((_get(nii, c) + _get(gains, c)) - _get(r, c))
|
|
for c in range(ncols))
|
|
if err <= 0.05 * ncols:
|
|
ops = r
|
|
ops_i = cand
|
|
break
|
|
if ops is None:
|
|
continue
|
|
# dtot row: some later row t with divs/roc before it satisfying
|
|
# divs (+roc) == dtot per column
|
|
best = None
|
|
for t_ in range(ops_i + 2, len(rows) - 1):
|
|
dtot = rows[t_]
|
|
window_rows = rows[ops_i + 1: t_]
|
|
for extra in (0, 1):
|
|
need_cols = len(window_rows) - extra
|
|
if need_cols < 1:
|
|
continue
|
|
# try every way to pick divs (+ optional roc) from the
|
|
# window rows so the column arithmetic holds
|
|
from itertools import combinations
|
|
for pick in combinations(range(len(window_rows)),
|
|
2 if extra else 1):
|
|
ok = True
|
|
dv = [0.0] * ncols
|
|
rc = [0.0] * ncols
|
|
for j, pi in enumerate(pick):
|
|
w = window_rows[pi]
|
|
vec = dv if j == 0 else rc
|
|
for c in range(ncols):
|
|
vec[c] += abs(_get(w, c))
|
|
for c in range(ncols):
|
|
d = abs(_get(dtot, c))
|
|
if abs(dv[c] + rc[c] - d) > 0.02 * max(1.0, d) + 0.02:
|
|
ok = False
|
|
break
|
|
if ok:
|
|
best = (t_, dv, rc, dtot)
|
|
break
|
|
if best:
|
|
break
|
|
if best:
|
|
break
|
|
if best is None:
|
|
continue
|
|
t_, dv, rc, dtot = best
|
|
# nav_e: first row after dtot close to nav_b + ops - dtot
|
|
# (capital transactions are usually small)
|
|
nav_e = mkt_e = None
|
|
pred = [_get(nav_b, c) + _get(ops, c) - abs(_get(dtot, c))
|
|
for c in range(ncols)]
|
|
for e_ in range(t_ + 1, len(rows)):
|
|
r = rows[e_]
|
|
close = sum(1 for c in range(ncols)
|
|
if abs(_get(r, c) - pred[c]) < 3.0)
|
|
if close >= ncols - 1:
|
|
nav_e = r
|
|
if e_ + 1 < len(rows):
|
|
mkt_e = rows[e_ + 1]
|
|
break
|
|
out = {
|
|
"ncols": ncols,
|
|
"nav_beg": list(nav_b),
|
|
"nav_end": [abs(_get(nav_e, c)) for c in range(ncols)]
|
|
if nav_e is not None else [],
|
|
"mkt_end": [abs(_get(mkt_e, c)) for c in range(ncols)]
|
|
if mkt_e is not None else [],
|
|
"dist_tot": [abs(_get(dtot, c)) for c in range(ncols)],
|
|
"divs": list(dv),
|
|
"gains": [0.0] * ncols,
|
|
"roc": list(rc),
|
|
"ops": [abs(_get(ops, c)) for c in range(ncols)],
|
|
}
|
|
tot5 = sum(v for v in out["dist_tot"] if v > 0)
|
|
if tot5 > 0:
|
|
out["share_div"] = sum(out["divs"]) / tot5
|
|
out["share_gains"] = sum(out["gains"]) / tot5
|
|
out["share_roc"] = sum(out["roc"]) / tot5
|
|
return out
|
|
return None
|
|
|
|
|
|
def parse_highlights(t: str, name: str = "", ticker: str = "") -> dict | None:
|
|
"""Parse the per-share highlights table. Returns None on failure.
|
|
name/ticker identify THIS fund in a consolidated report."""
|
|
trans_name = name
|
|
trans_ticker = ticker
|
|
# collapse whitespace: CEF HTML tables flatten with one <td> per line,
|
|
# splitting labels ("Net\nasset value, beginning of year")
|
|
t = re.sub(r"\s+", " ", t)
|
|
# anchor: first per-share "Net asset value[,]? beginning of
|
|
# [the] year|period" whose values are per-share scale. (The old
|
|
# "share within 400 chars" guard rejected real tables - ASA/DXYZ/
|
|
# NML/... - a dollar-basis statement has values in the millions,
|
|
# so the magnitude of the first value is the discriminator.)
|
|
cands = []
|
|
for m in re.finditer(
|
|
r"net asset value(?: per (?:common )?share)?,? beginning of "
|
|
r"(?:the )?(?:year|period)", t, re.I):
|
|
blk = t[m.start(): m.start() + 6000]
|
|
row = _row(blk,
|
|
r"net asset value(?: per (?:common )?share)?,? beginning "
|
|
r"of (?:the )?(?:year|period)")
|
|
if row and 0 < abs(row[0]) < 10000:
|
|
cands.append(m)
|
|
if not cands:
|
|
# no inline per-share row: transposed layout (Voya, PIMCO,
|
|
# Nuveen - label list, then one row per year)
|
|
return parse_transposed(t, trans_name, trans_ticker)
|
|
for m in cands:
|
|
h = _parse_inline(t[m.start(): m.start() + 6000])
|
|
if h:
|
|
return h
|
|
h = parse_columnar(t, trans_name, trans_ticker)
|
|
if h:
|
|
return h
|
|
return parse_transposed(t, trans_name, trans_ticker)
|
|
|
|
|
|
def _parse_inline(block: str) -> dict | None:
|
|
"""One inline (row-per-label) highlights table starting at its
|
|
NAV-beginning anchor."""
|
|
nav_beg = _row(block, r"net asset value(?: per (?:common )?share)?,? beginning of (?:the )?(?:year|period)")
|
|
nav_end = _fixlen(
|
|
_row(block, r"net asset value(?: per (?:common )?share)?,? end of (?:the )?(?:year|period)"),
|
|
len(nav_beg))
|
|
if len(nav_beg) < 2 or len(nav_end) < 2:
|
|
return None
|
|
ncols = len(nav_beg)
|
|
if len(nav_end) != ncols or ncols > 6:
|
|
return None
|
|
|
|
nii_ops = _fixlen(_row(block, r"net investment (?:income|loss)\b"), ncols)
|
|
gains_ops = _fixlen(
|
|
_row(block,
|
|
r"net (?:realized (?:and|&) (?:unrealized )?gain"
|
|
r"|realized gain \(loss\)|gain \(loss\))"), ncols)
|
|
ops_src = None
|
|
ops = _fixlen(_row(block,
|
|
r"total .{0,60}?from (?:the )?(?:investment )?operations\b"),
|
|
ncols)
|
|
if ops:
|
|
ops_src = "ops"
|
|
else:
|
|
ops = _fixlen(_row(
|
|
block,
|
|
r"net (?:increase|decrease)(?:\s*/?\(?\s*decrease\)?)?"
|
|
r" in net assets (?:resulting )?(?:from )?(?:the )?(?:investment )?"
|
|
r"operations"), ncols)
|
|
if ops:
|
|
ops_src = "ops"
|
|
if not ops:
|
|
# last resort: the printed net-change row (ops + captch). When
|
|
# used, captch must NOT be added again (DXYZ - no dist rows at
|
|
# all; identity is nav_beg + netchange - nav_end == 0)
|
|
ops = _fixlen(_row(
|
|
block,
|
|
r"total? increase.{0,3}?decrease.{0,3}?in net asset value"),
|
|
ncols)
|
|
if ops:
|
|
ops_src = "netchange"
|
|
if not ops:
|
|
ops = [(_get(nii_ops, i) + _get(gains_ops, i)) for i in range(ncols)]
|
|
ops_src = "sum"
|
|
_ = ops_src
|
|
|
|
# CEFs adjust NAV for capital transactions (repurchases, rights
|
|
# offerings, tenders). Collect every variant row present, deduping a
|
|
# physical row that matches two labels (e.g. "Capital Share
|
|
# Transactions: Accretion (dilution) ...").
|
|
captch_rows = []
|
|
seen = []
|
|
if ops_src != "netchange":
|
|
for rx in (r"accretion (?:\(dilution\) )?to net asset value",
|
|
r"capital share transactions",
|
|
r"anti-dilutive impact of repurchase",
|
|
r"dilutive impact of rights offering",
|
|
r"anti-dilutive impact of tender offer"
|
|
r"repurchase of shares",
|
|
r"at-the-market offering",
|
|
r"effect of shares issued",
|
|
r"offering expenses",
|
|
r"impact of capital share transactions",
|
|
r"total capital stock transactions",):
|
|
row = _fixlen(_row(block, rx), ncols)
|
|
if row and any(len(row) == len(r2) and
|
|
all(abs((a or 0) - (b or 0)) < 0.005
|
|
for a, b in zip(row, r2)) for r2 in seen):
|
|
continue
|
|
seen.append(row)
|
|
captch_rows.append(row)
|
|
mkt_end = _row(block, r"market (?:value|price), end of year")
|
|
|
|
# distribution sub-rows (character). The block ends at NAV-end:
|
|
# footnotes below it also contain the word "dividends".
|
|
danchor = re.search(
|
|
r"distributions to (?:common )?shareholders from|"
|
|
r"distributions to stockholders from|"
|
|
r"distributions declared (?:to shareholders)? from|"
|
|
r"distributions and dividends to|"
|
|
r"distributions to (?:common )?(?:share|stock)holders|"
|
|
r"less distributions|"
|
|
r"distributions \( ?[a-z0-9]{1,3} ?\)|"
|
|
r"distributions:|"
|
|
r"common dividends|"
|
|
r"dividends from|"
|
|
r"distributions on|"
|
|
r"distributions from",
|
|
block, re.I)
|
|
if danchor:
|
|
dblock = block[danchor.start():]
|
|
else:
|
|
# no distribution heading: start after the total-operations row
|
|
# (the distribution section, if any, follows it)
|
|
om = re.search(
|
|
r"total .{0,60}?from (?:the )?(?:investment )?operations\b|"
|
|
r"net (?:increase|decrease).{0,30}?in net assets.{0,30}?operations",
|
|
block, re.I)
|
|
dblock = block[om.end():] if om else block
|
|
em = re.search(r"net asset value(?: per (?:common )?share)?,? end of (?:the )?(?:year|period)", dblock, re.I)
|
|
if em:
|
|
dblock = dblock[: em.start()]
|
|
# the TOTAL row label also contains "dividends" in some funds
|
|
# ("Total dividends and distributions to stockholders") - scope the
|
|
# sub-row search to before it
|
|
tm = re.search(r"total (?:dividends and )?distributions|"
|
|
r"total dividends\b", dblock, re.I)
|
|
dwindow = dblock[: tm.start()] if tm else dblock
|
|
divs = _fixlen(_row(dwindow, r"\bdividends\b")
|
|
or _row(dwindow, r"net investment income"), ncols)
|
|
gains = _fixlen(_row(dwindow,
|
|
r"(?<!income and )(?:net )?realized (?:capital )?gains?|"
|
|
r"capital gains\b|capital gain distributions"), ncols)
|
|
roc = _fixlen(_row(dwindow, r"return[- ]of[- ]capital"), ncols)
|
|
|
|
def _same(a, b):
|
|
return (len(a) == len(b) and
|
|
all(abs((x or 0) - (y or 0)) < 0.005
|
|
for x, y in zip(a, b)))
|
|
|
|
# a "Net investment income and net realized gains" combined row is
|
|
# caught by BOTH the divs and the gains label - count it once
|
|
if gains and divs and _same(gains, divs):
|
|
gains = []
|
|
# an empty gains row resolves to the next row's values (NXG)
|
|
if gains and roc and _same(gains, roc):
|
|
gains = []
|
|
|
|
# total distributions: the PRINTED row is authoritative when it has
|
|
# one value per column; otherwise derive from the NAV identity
|
|
# (dist = NAV_beg + ops + captch - NAV_end)
|
|
dist_row = _fixlen(
|
|
_row(dblock, r"total (?:dividends and )?distributions|total dividends\b"),
|
|
ncols)
|
|
complete = len(dist_row) == ncols
|
|
if complete:
|
|
dist_tot = list(dist_row)
|
|
# some funds print distributions positive, others negative
|
|
if sum(1 for v in dist_tot if v < 0) > ncols // 2:
|
|
dist_tot = [-v for v in dist_tot]
|
|
else:
|
|
dist_tot = None
|
|
|
|
# JOINT PLACEMENT. CEF tables drop zero columns, so a short row can
|
|
# sit in ANY subset of columns (TWN: gains in 4 of 5 cols, NII-divs
|
|
# in the other two; EMO: ROC in cols 0 and 4; KF: accretion in cols
|
|
# 1-4, not 0-3). One backtracking search places every short row
|
|
# (captch first - they affect the NAV identity - then the
|
|
# distribution rows) so the per-column arithmetic holds:
|
|
# printed total complete: divs+gains+roc == total
|
|
# printed total short: divs+gains+roc == NAV_beg+ops+captch-NAV_end
|
|
from itertools import combinations
|
|
|
|
dist_rows = (("divs", divs), ("gains", gains), ("roc", roc))
|
|
place = [(i, "c", row) for i, row in enumerate(captch_rows)
|
|
if 0 < len(row) < ncols]
|
|
place += [(i, "d", row) for i, (name, row) in enumerate(dist_rows)
|
|
if 0 < len(row) < ncols]
|
|
place.sort(key=lambda t: -len(t[2]))
|
|
|
|
base_d = [0.0] * ncols
|
|
for name, row in dist_rows:
|
|
if len(row) == ncols:
|
|
for c in range(ncols):
|
|
if row[c]:
|
|
base_d[c] += abs(row[c])
|
|
base_c = [0.0] * ncols
|
|
for row in captch_rows:
|
|
if len(row) == ncols:
|
|
for c in range(ncols):
|
|
if row[c]:
|
|
base_c[c] += row[c]
|
|
|
|
dsum = [0.0] * ncols
|
|
csum = [0.0] * ncols
|
|
assign = []
|
|
|
|
def rec(i):
|
|
if i == len(place):
|
|
for c in range(ncols):
|
|
if complete:
|
|
need = abs(dist_tot[c])
|
|
else:
|
|
need = abs(_get(nav_beg, c) + _get(ops, c) + base_c[c]
|
|
+ csum[c] - _get(nav_end, c))
|
|
if abs(base_d[c] + dsum[c] - need) > 0.02 * max(1.0, need) + 0.02:
|
|
return False
|
|
return True
|
|
idx, kind, vals = place[i]
|
|
nz = [v for v in vals if v]
|
|
for cols in combinations(range(ncols), len(nz)):
|
|
if kind == "d":
|
|
for c, v in zip(cols, nz):
|
|
dsum[c] += abs(v)
|
|
if complete and any(base_d[c] + dsum[c] > abs(dist_tot[c]) + 0.02
|
|
for c in cols):
|
|
for c, v in zip(cols, nz):
|
|
dsum[c] -= abs(v)
|
|
continue
|
|
else:
|
|
for c, v in zip(cols, nz):
|
|
csum[c] += v
|
|
if rec(i + 1):
|
|
assign.append((idx, kind, nz, cols))
|
|
return True
|
|
if kind == "d":
|
|
for c, v in zip(cols, nz):
|
|
dsum[c] -= abs(v)
|
|
else:
|
|
for c, v in zip(cols, nz):
|
|
csum[c] -= v
|
|
return False
|
|
|
|
if place and not rec(0):
|
|
return None
|
|
|
|
def colvec(vals, cols):
|
|
vec = [0.0] * ncols
|
|
for ci, col in enumerate(cols):
|
|
vec[col] = vals[ci]
|
|
return vec
|
|
|
|
placed_d = {}
|
|
placed_c = {i: [0.0] * ncols for i in
|
|
(i for i, k, _ in place if k == "c")}
|
|
for idx, kind, nz, cols in assign:
|
|
if kind == "d":
|
|
placed_d[idx] = colvec(nz, cols)
|
|
else:
|
|
placed_c[idx] = colvec(nz, cols)
|
|
|
|
for i, (name, row) in enumerate(dist_rows):
|
|
if len(row) == ncols:
|
|
vec = list(row)
|
|
elif 0 < len(row) < ncols:
|
|
vec = placed_d.get(i, [0.0] * ncols)
|
|
else:
|
|
vec = [0.0] * ncols
|
|
if name == "divs":
|
|
divs = vec
|
|
elif name == "gains":
|
|
gains = vec
|
|
else:
|
|
roc = vec
|
|
captch = [base_c[c] for c in range(ncols)]
|
|
for i, vec in placed_c.items():
|
|
for c in range(ncols):
|
|
captch[c] += vec[c]
|
|
if not complete:
|
|
dist_tot = [(_get(nav_beg, c) + _get(ops, c) + captch[c]
|
|
- _get(nav_end, c)) for c in range(ncols)]
|
|
if sum(1 for v in dist_tot if v < 0) > ncols // 2:
|
|
dist_tot = [-v for v in dist_tot]
|
|
|
|
# some funds print distributions positive, others negative - work
|
|
# with magnitudes from here on (dist_tot is already normalized)
|
|
divs = [abs(v) for v in divs]
|
|
gains = [abs(v) for v in gains]
|
|
roc = [abs(v) for v in roc]
|
|
|
|
# sanity: components should (roughly) sum to the derived total
|
|
bad = sum(1 for c in range(ncols)
|
|
if abs(divs[c] + gains[c] + roc[c] - dist_tot[c])
|
|
> 0.02 * max(1.0, abs(dist_tot[c])) + 0.02)
|
|
if bad > max(1, ncols // 2):
|
|
return None
|
|
|
|
out = {
|
|
"ncols": ncols,
|
|
"nav_beg": nav_beg, "nav_end": nav_end,
|
|
"mkt_end": mkt_end[:ncols] if mkt_end else [],
|
|
"dist_tot": dist_tot, "divs": divs, "gains": gains, "roc": roc,
|
|
"ops": ops[:ncols],
|
|
}
|
|
tot5 = sum(v for v in dist_tot if v > 0)
|
|
if tot5 > 0:
|
|
out["share_div"] = sum(divs) / tot5
|
|
out["share_gains"] = sum(gains) / tot5
|
|
out["share_roc"] = sum(roc) / tot5
|
|
return out
|
|
|
|
|
|
|
|
def _get(row: list, i: int) -> float:
|
|
if i < len(row) and row[i] is not None:
|
|
return row[i]
|
|
return 0.0
|
|
|
|
|
|
def _submissions(cik: int) -> dict | None:
|
|
try:
|
|
raw = edgar.sec_get(f"https://data.sec.gov/submissions/CIK{cik:010d}.json")
|
|
return json.loads(raw)
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def analyze(sym: str, cik: int, force: bool = False) -> dict:
|
|
cf = CACHE / f"{sym}.json"
|
|
if cf.exists() and not force:
|
|
return json.loads(cf.read_text())
|
|
d = _submissions(cik)
|
|
if not d:
|
|
return {"sym": sym, "error": "submissions fetch failed"}
|
|
r = d["filings"]["recent"]
|
|
forms = r["form"]
|
|
out: dict = {"sym": sym, "cik": cik, "name": d.get("name"),
|
|
"bdc": "10-K" in set(forms),
|
|
"n_tender": sum(1 for f in forms if f.startswith("N-23C")),
|
|
"ticker": (d.get("tickers") or [sym])[0]}
|
|
# try the 3 most recent shareholder reports, keep the best parse
|
|
cands = [(r["filingDate"][i], r["accessionNumber"][i])
|
|
for i in range(len(forms))
|
|
if forms[i] in ("N-CSR", "N-CSRS")][:3]
|
|
def _score(h):
|
|
# the 5-column annual table is the target; a 6-col semiannual
|
|
# (period + 5 FYs) is a fallback
|
|
return 2 if h["ncols"] == 5 else 1
|
|
|
|
best = None
|
|
for fd, acc in cands:
|
|
try:
|
|
raw = edgar.sec_get(
|
|
f"https://www.sec.gov/Archives/edgar/data/{cik}/{acc}.txt")
|
|
t = edgar.to_text(raw)
|
|
except Exception:
|
|
continue
|
|
h = parse_highlights(t, d.get("name", ""), out.get("ticker", sym))
|
|
if h and (best is None or _score(h) > _score(best[1])
|
|
or (_score(h) == _score(best[1])
|
|
and h["ncols"] > best[1]["ncols"])):
|
|
best = (fd, h, t)
|
|
if not best:
|
|
out["error"] = "no per-share table found in the 3 latest reports"
|
|
else:
|
|
fd, h, t = best
|
|
out["report_filed"] = fd
|
|
out.update(h)
|
|
# leverage + tax-paying flags (optional lines)
|
|
lm = re.search(
|
|
r"Loan and Debt Issuance Outstanding, End of Year \(000s\)\s*"
|
|
r"([\d,]+)", t)
|
|
pm = re.search(
|
|
r"Preferred Stock at Liquidation Value, End of Year \(000s\)\s*"
|
|
r"([\d,]+)", t)
|
|
nm = re.search(
|
|
r"Net assets applicable to common shareholders, end of year "
|
|
r"\(millions\)\s*\$?([\d,]+)", t)
|
|
if lm and nm:
|
|
debt = float(lm.group(1).replace(",", "")) * 1e3
|
|
pref = float(pm.group(1).replace(",", "")) * 1e3 if pm else 0.0
|
|
nav = float(nm.group(1).replace(",", "")) * 1e6
|
|
out["leverage_pct"] = round((debt + pref) / nav * 100, 1)
|
|
if re.search(r"income tax expenses?\s+[\d.]+\s*%", t):
|
|
out["tax_paying"] = True
|
|
CACHE.mkdir(exist_ok=True)
|
|
cf.write_text(json.dumps(out, default=str))
|
|
return out
|
|
|
|
|
|
def current_nav_discount(h: dict) -> float | None:
|
|
"""FY-end discount (market vs NAV) from the table; used as a
|
|
rough estimate of the current discount (no live NAV locally)."""
|
|
nav_end = h.get("nav_end") or []
|
|
mkt_end = h.get("mkt_end") or []
|
|
if not nav_end or not mkt_end or not nav_end[0]:
|
|
return None
|
|
return float(mkt_end[0] / nav_end[0] - 1)
|
|
|
|
|
|
def character_score(h: dict, name: str) -> float | None:
|
|
if "share_div" not in h:
|
|
return None
|
|
f_nii = FRAC_NII_MUNI if re.search(r"muni|tax[- ]?exempt", name, re.I) \
|
|
else FRAC_NII
|
|
return (h["share_div"] * f_nii + h["share_gains"] * FRAC_GAINS
|
|
+ h["share_roc"] * FRAC_ROC)
|
|
|
|
|
|
def run(syms: list[str] | None = None) -> dict:
|
|
uni = json.loads((HERE / "cef_universe.json").read_text())
|
|
if syms is None:
|
|
ch = json.loads(CHARACTER.read_text())
|
|
syms = [s.upper() for s in
|
|
sorted(ch, key=lambda s: -ch[s]["tax_arb"])]
|
|
res: dict = {}
|
|
for i, s in enumerate(syms, 1):
|
|
u = uni.get(s) or {}
|
|
cik = int(u.get("cik", 0))
|
|
a = analyze(s, cik)
|
|
a["char_actual"] = round(character_score(a, u.get("name", "")), 2) \
|
|
if "share_div" in a else None
|
|
if "nav_end" in a:
|
|
disc = current_nav_discount(a)
|
|
a["disc_now_approx"] = round(disc, 4) if disc is not None else None
|
|
res[s] = a
|
|
if "error" in a:
|
|
print(f"{i:2}/{len(syms)} {s:7} ERR: {a['error'][:40]}", flush=True)
|
|
else:
|
|
print(f"{i:2}/{len(syms)} {s:7} "
|
|
f"div {a.get('share_div',0):.0%} gain {a.get('share_gains',0):.0%} "
|
|
f"ROC {a.get('share_roc',0):.0%} char={a.get('char_actual')} "
|
|
f"lev={a.get('leverage_pct')}", flush=True)
|
|
OUT.write_text(json.dumps(res, indent=1, default=str))
|
|
print(f"wrote {OUT}")
|
|
return res
|
|
|
|
|
|
def _print(res: dict) -> None:
|
|
print(f"{'fund':7} {'name':38} {'fdiv':>5} {'fgain':>6} {'froc':>5}"
|
|
f" {'char':>5} {'disc':>7} {'lev%':>6} {'taxp':>4} {'tend':>5} "
|
|
f"{'report':>10}")
|
|
for s, v in res.items():
|
|
if "share_div" not in v:
|
|
print(f"{s:7} {v.get('name','')[:38]:38} -- "
|
|
f"{v.get('error', 'no data')}")
|
|
continue
|
|
disc = v.get("disc_now_approx")
|
|
print(f"{s:7} {v.get('name','')[:38]:38} "
|
|
f"{v['share_div']:5.0%} {v['share_gains']:6.0%} "
|
|
f"{v['share_roc']:5.0%} {v.get('char_actual') or 0:5.2f} "
|
|
f"{('' if disc is None else f'{disc:+.1%}'):>7} "
|
|
f"{str(v.get('leverage_pct', '')):>6} "
|
|
f"{'Y' if v.get('tax_paying') else '':>4} "
|
|
f"{v.get('n_tender', 0):>5} {v.get('report_filed',''):>10}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import sys
|
|
args = [a.upper() for a in sys.argv[1:]]
|
|
r = run(args or None)
|
|
_print(r)
|