"""N-PORT holdings cross-check for the screen's top candidates. For each candidate: 1. ticker -> the fund's OWN registrant CIK (browse-edgar; the 497-cover CIK is the family/trust and is the fallback) — cached in nport_cache/cik_map.json; 2. exact series name for the ticker from the 497 covers (disambiguates sibling funds like "... Absolute Return Fund" vs "... Advantage Fund", which share 6 of 7 name words); 3. walk the fund's recent NPORT-Ps (submissions API): - interactive XML primaries -> parse_interactive(), match on the exact seriesName (this is also the only route for wrapper funds whose whole book is one managed portfolio, e.g. egrix); - standalone SOI htm exhibits -> nport.build() section matching; 4. keep the best result and record what the fund actually holds. Usage: python -m fundlab.xcheck [sym ...] (default: built-in list) Output: fundlab/xcheck_report.json + console table. """ from __future__ import annotations import json import re import time import xml.etree.ElementTree as ET from pathlib import Path from . import edgar, nport HERE = Path(__file__).parent COVERS = HERE / "universe_cache" / "covers.json" REPORT = HERE / "xcheck_report.json" MANIFEST = HERE / "nport_manifest.json" RAW = nport.CACHE / "raw" CIKMAP = nport.CACHE / "cik_map.json" MAX_FILE_MB = 20 # skip pathologically large exhibits MAX_FILINGS = 30 # top candidates from the v2 screen + named missing-factor funds DEFAULT = [ "scfzx", "hmezx", "coiax", "wmnux", "fhcox", "rctix", "egrix", "etsix", "aflix", "fhmix", "hicox", "dultx", "usmsx", "btmix", "safex", "qcmmrx", "mervx", "aguax", "dmszx", "anglx", "femdx", "lpxax", ] _STOP = {"fund", "funds", "inc", "ltd", "llc", "company", "class", "series", "the", "and", "of", "for", "trust", "portfolio", "account", "shares", "share"} # ------------------------------------------------------------------ CIKs def ticker_to_cik() -> dict[str, str]: """ticker -> 497-cover CIK (the family/trust registrant).""" cov = json.loads(COVERS.read_text()) out: dict[str, str] = {} for cik, info in cov.items(): for ser in info.get("series", []): for tk in ser.get("tickers", []): out.setdefault(tk.lower(), cik) return out def resolve_cik(ticker: str, cover_cik: str | None) -> str | None: """The fund's OWN registrant CIK (its NPORT filer), cached. The 497-cover CIK is the family/trust; the fund's NPORTs are often filed under the fund's own registrant (e.g. Eaton Vance: trust CIK 1552324 vs the fund's 745463). browse-edgar resolves the ticker to its direct registrant. """ m = json.loads(CIKMAP.read_text()) if CIKMAP.exists() else {} tk = ticker.upper() if tk in m: return m[tk] own: str | None = None try: own = edgar.ticker_to_company(tk)[0] except Exception: pass cik = own or cover_cik m[tk] = cik CIKMAP.write_text(json.dumps(m, indent=1)) return cik def series_name_for(ticker: str, own_cik: str | None, cover_cik: str | None) -> str | None: """Exact series name for the ticker, from the 497 covers. This is the reliable disambiguator between sibling funds whose Yahoo names differ by one word. """ cov = json.loads(COVERS.read_text()) for cik in dict.fromkeys(filter(None, (own_cik, cover_cik))): for ser in cov.get(cik, {}).get("series", []): if ticker.upper() in [t.upper() for t in ser.get("tickers", [])]: return ser["name"] return None def _tokens(name: str) -> list[str]: """Section-finder tokens: the fund's own identifying words.""" words = re.sub(r"[^A-Za-z ]", " ", name.lower()).split() words = [w for w in words if w not in _STOP and len(w) > 3] return words[-3:] or words def _norm_name(s: str) -> str: s = s.lower().replace("&", " and ") s = re.sub(r"[^a-z0-9 ]", "", s) return re.sub(r"\s+", "", s) def _name_match(series: str | None, target: str) -> int: """0 = no, 1 = containment/token-overlap, 2 = exact (normalized). Normalization absorbs the common cover-vs-filing drift: "&" vs "and", punctuation, spacing ("Fund,Inc." vs "Fund, Inc."). Token overlap catches one-word drift ("...Return Fund" vs "...Return Advantage Fund"); it only matters when no exact match exists, and xml_exact always outranks it. """ if not series or not target: return 0 a, b = _norm_name(series), _norm_name(target) if a == b: return 2 if min(len(a), len(b)) >= 12 and (a in b or b in a): return 1 wa = set(re.sub(r"[^a-z0-9]+", " ", series.lower()).split()) wb = set(re.sub(r"[^a-z0-9]+", " ", target.lower()).split()) if wa and wb and len(wa & wb) / min(len(wa), len(wb)) >= 0.7: return 1 return 0 # ------------------------------------------------------------------ XML def parse_interactive(text: str) -> dict | None: """Parse an interactive NPORT-P XML (primary_doc.xml). Returns {name, net_assets, positions:[{name, val_usd, pct, asset_cat, issuer_cat, country, profile}]} or None if the document is not an investment-level NPORT form. """ try: root = ET.fromstring(text) except ET.ParseError: return None def first(name: str) -> str | None: for e in root.iter(): if e.tag.split("}", 1)[-1] == name and (e.text or "").strip(): return e.text.strip() return None name = first("seriesName") invs = [e for e in root.iter() if e.tag.split("}", 1)[-1] == "invstOrSec"] if not name or not invs: return None positions = [] for inv in invs: def f(n: str, _inv=inv) -> str | None: for e in _inv.iter(): if e.tag.split("}", 1)[-1] == n: return (e.text or "").strip() return None positions.append({ "name": f("name"), "title": f("title"), "val_usd": f("valUSD"), "pct": f("pctVal"), "asset_cat": f("assetCat"), "issuer_cat": f("issuerCat"), "country": f("invCountry"), "profile": f("payoffProfile"), }) return {"name": name, "net_assets": first("netAssets"), "positions": positions} def _fnum(s: str | None) -> float | None: try: return float(s) except (TypeError, ValueError): return None # NPORT Part C asset-category (C.4.a) + issuer-category (C.4.b) codes # are authoritative - they classify the instrument, unlike position-name # keywords (a bank note's "inc/corp" issuer name reads as "equity"). _ASSET = { "STIV": "Cash/MMF (short-term)", "RA": "Repurchase agreement", "EC": "Equity (common)", "EP": "Preferred stock", "DBT": None, # see issuer "LON": "Loan (leveraged/private credit)", "COMM": "Commodity", "RE": "Real estate", "SN": "Structured note", "ABS-CBDO": "CLO (collateralized debt)", "ABS-MBS": "MBS (mortgage-backed)", "ABS-CP": "ABS commercial paper", "ABS-O": "ABS other (CLO/CMBS/AB)", "ABS-APCP": "ABS auto/personal loan", "ABS-HE": "ABS home equity", "ABS-CO": "ABS credit card", } _DEBT_BY_ISSUER = { "UST": "US Treasury", "USGA": "US agency", "USGSE": "US GSE (Fed/FF)", "MUN": "Municipal bond", "NUSS": "Foreign sovereign", "CORP": "Corporate bond", "PF": "Private fund", "RF": "Fund/ETF", } def xml_bucket(p: dict) -> str: a = (p.get("asset_cat") or "").upper() i = (p.get("issuer_cat") or "").upper() # open-end fund / ETF shares file as EC (equity) with issuer RF if a in ("EC", "EP") and i == "RF": return "Fund/ETF holdings" if a in _ASSET and _ASSET[a] is not None: return _ASSET[a] if a == "DBT": return _DEBT_BY_ISSUER.get(i, "Debt") if a.startswith("D"): # DCO DCR DE DFE DIR DO = derivatives return "Derivative / hedge" return f"{a or 'Other'}" def xml_snapshot(d: dict) -> dict: """Shape an XML parse like nport.build's snapshot (for the report).""" buckets: dict[str, float] = {} for p in d["positions"]: v = _fnum(p["val_usd"]) or 0.0 buckets[xml_bucket(p)] = buckets.get(xml_bucket(p), 0.0) + v total = sum(buckets.values()) or 1.0 pos = sorted(d["positions"], key=lambda p: -abs(_fnum(p["val_usd"]) or 0.0)) return { "n_positions": len(d["positions"]), "net_assets": _fnum(d["net_assets"]), "buckets": [{"name": k, "value": v, "pct": 100.0 * v / total} for k, v in sorted(buckets.items(), key=lambda kv: -abs(kv[1]))], "top": pos[:15], } # ------------------------------------------------------------------ fetch def _soi_ok(text: str) -> bool: return bool(re.search(r"(?i)schedule of (portfolio )?investments", text) or re.search(r"(?i)portfolio of investments", text) or re.search(r"(?i)investment portfolio", text)) def _accession_docs(cik: str, f: dict) -> list[tuple[str, str, str, str]]: """(url, text, kind) for each candidate document in the accession. kind is 'xml' (interactive NPORT primary) or 'htm' (SOI exhibit). Everything is cached under nport_cache/raw/. """ base = f"https://www.sec.gov/Archives/edgar/data/{int(cik)}" acc = f["accession"].replace("-", "") try: idx = json.loads(edgar.sec_get(f"{base}/{acc}/index.json", timeout=60)) items = idx["directory"]["item"] if isinstance(items, dict): items = [items] cands = [(it["name"], int(it.get("size") or 0)) for it in items] htms = [(n, s) for n, s in cands if n.lower().endswith((".htm", ".html")) and "index" not in n.lower()] pdoc = f["doc"] # NPORT interactive filings: the submissions API points at the # XSL-rendered view (xslFormNPORT-P_X01/primary_doc.xml - a huge # HTML page); the raw schema data sits at the accession root if pdoc.lower().endswith(".xml") and "/" in pdoc: pdoc = pdoc.split("/")[-1] doc = (pdoc, 0) if any(n == pdoc for n, _ in cands) \ or pdoc.lower().endswith((".htm", ".html", ".xml")) else None except Exception: htms, doc = [], None out: list[tuple[str, str, str]] = [] def take(fn: str, size: int) -> None: if size and size > MAX_FILE_MB * 1e6: return url = f"{base}/{acc}/{fn}" c = RAW / f"{acc}_{fn.replace('/', '__')}" try: if c.exists(): text = c.read_text() else: text = edgar.sec_get(url, timeout=180).decode( "utf-8", "ignore") RAW.mkdir(parents=True, exist_ok=True) c.write_text(text) except Exception: return kind = "xml" if fn.lower().endswith(".xml") else "htm" if kind == "xml" or _soi_ok(text): out.append((url, text, kind, f["filed"])) for n, s in htms: take(n, s) if doc: take(*doc) return out def fetch_docs(cik: str, filings: list[dict]) -> list[tuple[str, str, str, str]]: out: list[tuple[str, str, str, str]] = [] for f in filings: out.extend(_accession_docs(cik, f)) return out # ------------------------------------------------------------------ run def run(syms: list[str] | None = None, force: bool = False) -> dict: syms = [s.lower() for s in (syms or DEFAULT)] t2c = ticker_to_cik() fr = json.loads((HERE / "factor_results.json").read_text()) man = json.loads(MANIFEST.read_text()) if MANIFEST.exists() else {} report: dict[str, dict] = {} if not force and REPORT.exists(): try: report = json.loads(REPORT.read_text()) except Exception: report = {} def _done(r: dict | None) -> bool: return bool(r) and not r.get("error") and r.get("n_positions") for sym in syms: if not force and _done(report.get(sym)): print(f" [{len(report)}/{len(syms)}] {sym.upper()} cached, skip", flush=True) continue meta = fr.get(sym, {}) ticker = (meta.get("ticker") or sym).upper() cover_cik = t2c.get(sym) cik = resolve_cik(ticker, cover_cik) target = series_name_for(ticker, cik, cover_cik) \ or meta.get("name", "") rec = {"sym": sym, "name": target, "t5": meta.get("alpha_t_5y"), "r2": (meta.get("full") or {}).get("r2")} if not cik: rec["error"] = "no CIK (covers.json or browse-edgar)" report[sym] = rec continue rec["cik"] = cik try: all_f = edgar.cik_recent_filings(cik, "NPORT-P", 200) except Exception as e: rec["error"] = f"filings: {e}" report[sym] = rec continue if not all_f: rec["error"] = "no NPORT-P filings found" report[sym] = rec continue # large trusts file dozens of NPORTs per quarter (one per fund); # take every filing on the 4 most recent distinct dates instead # of a flat newest-30 window (which misses a fund's own filing) dates: list[str] = [] for f in all_f: if f["filed"] not in dates: dates.append(f["filed"]) if len(dates) == 4: break keep = set(dates) filings = [f for f in all_f if f["filed"] in keep] docs = fetch_docs(cik, filings) # --- pick: exact seriesName match in an XML > best HTM parse > # containment XML match xml_exact = xml_contain = None htm_best: tuple[int, dict, str, str] | None = None for url, text, kind, filed in docs: if kind == "xml": d = parse_interactive(text) if not d: continue m = _name_match(d["name"], target) if m == 2 and xml_exact is None: xml_exact = (d, url, filed) elif m == 1 and xml_contain is None: xml_contain = (d, url, filed) else: tokens = _tokens(target) (nport.CACHE / f"{sym}.html").write_text(text) for tok in ([tokens, [tokens[-1]]] if len(tokens) > 1 else [tokens]): snap = nport.build(sym, tok, force=True, frac=0.66) if snap and (htm_best is None or snap["n_positions"] > htm_best[0]): htm_best = (snap["n_positions"], snap, url, filed) chosen = None if xml_exact: d, url, filed = xml_exact snap = xml_snapshot(d) rec["src"] = "interactive NPORT XML (exact series match)" if (len(d["positions"]) == 1 and (_fnum(d["positions"][0]["pct"]) or 0) >= 95): rec["note"] = ("100% of NAV in ONE managed portfolio - " "underlying positions not disclosed in " "the fund's NPORT") chosen = (snap, url, filed) elif htm_best and htm_best[0] >= 5: chosen = (htm_best[1], htm_best[2], htm_best[3]) rec["src"] = "SOI htm exhibit" elif xml_contain: d, url, filed = xml_contain rec["src"] = "interactive NPORT XML (series name containment)" chosen = (xml_snapshot(d), url, filed) if not chosen: rec["error"] = ("no matching NPORT document parsed" if not docs else f"weak parse ({htm_best[0] if htm_best else 0} " "positions)") report[sym] = rec REPORT.write_text(json.dumps(report, indent=1)) print(f" [{len(report)}/{len(syms)}] {sym.upper()} ERROR " f"{rec['error']}", flush=True) continue snap, url, filed = chosen rec["soi_url"] = url rec["as_of"] = snap.get("as_of") or filed man[sym] = {"url": url, "filed": filed} rec["net_assets"] = snap.get("net_assets") rec["n_positions"] = snap["n_positions"] rec["categories"] = snap.get("categories", [])[:15] rec["buckets"] = snap["buckets"][:12] rec["top"] = snap["top"][:10] report[sym] = rec REPORT.write_text(json.dumps(report, indent=1)) print(f" [{len(report)}/{len(syms)}] {sym.upper()} ok " f"(n={rec['n_positions']}, {rec.get('src', '')[:40]})", flush=True) time.sleep(0.15) MANIFEST.write_text(json.dumps(man, indent=1)) REPORT.write_text(json.dumps(report, indent=1)) return report def _print(rpt: dict) -> None: for sym, r in rpt.items(): print(f"\n{'=' * 74}\n{sym.upper()} {r.get('name', '')}") t5, r2 = r.get("t5"), r.get("r2") stats = f" screen: t5={t5:+.1f} R2={r2:.2f}" if t5 else "" print(f" as-of {r.get('as_of')} n={r.get('n_positions', 0)}{stats}") if r.get("src"): print(f" source: {r['src']}") if r.get("note"): print(f" NOTE: {r['note']}") if r.get("error"): print(f" ERROR: {r['error']}") continue if r.get("categories"): print(" categories (as reported):") for c in r["categories"][:12]: print(f" {c['pct']:6.2f}% {c['name'][:58]}") print(" keyword buckets (positions by $):") for b in r.get("buckets", [])[:8]: print(f" {b['pct']:6.1f}% {b['name']}") print(" top positions (by |value|):") for p in r.get("top", [])[:10]: if "pct" in p: # xml pc = _fnum(p.get("pct")) pcs = f"{pc:+.2f}%" if pc is not None else "?" nm = (p.get("name") or p.get("title") or "")[:52] cat = p.get("asset_cat") or "" print(f" {pcs:>9} {nm} [{cat}]") else: # htm print(f" ${p['value']:>14,.0f} {p['text'][:58]}") if __name__ == "__main__": import sys args = sys.argv[1:] force = "-f" in args or "--force" in args syms = [a for a in args if not a.startswith("-")] _print(run(syms or None, force=force))