f/fundlab/xcheck.py

"""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))