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.
69 lines
2.4 KiB
Python
69 lines
2.4 KiB
Python
"""Probe the per-share table region of the failed CEFs (debug aid)."""
|
|
import json
|
|
import re
|
|
import sys
|
|
|
|
from fundlab import edgar
|
|
|
|
HERE = __import__("pathlib").Path(__file__).parent
|
|
|
|
|
|
def probe(syms, tag):
|
|
uni = json.loads((HERE / "cef_universe.json").read_text())
|
|
out = {}
|
|
for sym in syms:
|
|
u = uni.get(sym)
|
|
if not u:
|
|
continue
|
|
cik = int(u["cik"])
|
|
try:
|
|
d = json.loads(edgar.sec_get(
|
|
f"https://data.sec.gov/submissions/CIK{cik:010d}.json"))
|
|
except Exception as e:
|
|
out[sym] = f"submissions fail: {e}"
|
|
continue
|
|
r = d["filings"]["recent"]
|
|
cands = [(r["filingDate"][i], r["accessionNumber"][i], r["form"][i])
|
|
for i in range(len(r["form"]))
|
|
if r["form"][i] in ("N-CSR", "N-CSRS")][:2]
|
|
notes = []
|
|
for fd, acc, form in cands:
|
|
try:
|
|
t = re.sub(r"\s+", " ", edgar.to_text(
|
|
edgar.sec_get(
|
|
f"https://www.sec.gov/Archives/edgar/data/{cik}/"
|
|
f"{acc}.txt")))
|
|
except Exception as e:
|
|
notes.append(f"{fd}: fetch fail {e}")
|
|
continue
|
|
# find every candidate anchor region
|
|
pats = [
|
|
r"net asset value[^.]{0,40}beginning of (?:year|period)",
|
|
r"nav per (?:common )?share, beginning",
|
|
r"per share (?:operating )?performance",
|
|
r"selected per share data",
|
|
]
|
|
hit = None
|
|
for p in pats:
|
|
m = re.search(p, t, re.I)
|
|
if m:
|
|
hit = (p, m)
|
|
break
|
|
if hit:
|
|
p, m = hit
|
|
pre = t[max(0, m.start() - 200): m.start()]
|
|
notes.append(f"{fd} [{form}] ANCHOR '{p[:30]}' share-in-pre="
|
|
f"{bool(re.search(r'share', pre, re.I))} :: "
|
|
f"{t[m.start():m.start()+260]!r}")
|
|
else:
|
|
notes.append(f"{fd} [{form}] NO ANCHOR (len={len(t)})")
|
|
out[sym] = "\n".join(notes)
|
|
print(f"== {sym}\n{out[sym]}", flush=True)
|
|
(HERE / f"probe_{tag}.json").write_text(json.dumps(out, indent=1))
|
|
return out
|
|
|
|
|
|
if __name__ == "__main__":
|
|
syms = [a.upper() for a in sys.argv[1:]]
|
|
probe(syms, "batch")
|