f/fundlab/edgar_universe.py

"""Find candidate funds NOT in the local database - exhaustive pass over
the SEC's full filing index.

Source: the EDGAR full-index (https://www.sec.gov/Archives/edgar/
full-index/YYYY/QTRn/company.gz) lists every filing. Unique CIKs that
filed a 497-family form in the quarter = currently registered US
open-end funds (1,732 in 2026 Q2). One fetch per CIK - the full
submission .txt (~10-30KB) contains the SGML prospectus cover with
<SERIES-NAME> (fund name) and <CLASS-CONTRACT-TICKER-SYMBOL> (ticker)
tags, often several series per filing.

Pass:
  1. full-index -> latest 497 filing per CIK
  2. fetch the .txt (small), extract series names + tickers from the
     SGML cover
  3. keep alpha-leaning series, drop local-DB and shortlist tickers
  4. verify on Yahoo chart metadata (OTC fund class), require >=5y
     history, goget-download, screen with the same engine.

Politeness: 0.15s between requests; resumable (per-CIK cache).
"""
from __future__ import annotations

import gzip
import json
import re
import subprocess
import time
import urllib.request
from concurrent.futures import as_completed
from pathlib import Path

from fundlab import edgar, search
from fundlab.dbmine import PATTERN as ALPHA_KW, SKIP

# A fund files its base 497 (and 497K) about once a year, in whatever
# quarter its re-filing lands - so the FULL universe is the union over
# ~4 consecutive quarters.
QUARTERS = [("2026", "2"), ("2026", "1"), ("2025", "4"), ("2025", "3")]


def _index_url(year: str, qtr: str) -> str:
    return (f"https://www.sec.gov/Archives/edgar/full-index/{year}"
            f"/QTR{qtr}/company.gz")
# index paths are relative to /Archives/ (edgar/data/<cik>/<acc>.txt)
ARCH = "https://www.sec.gov/Archives/"
CACHE = Path(__file__).parent / "universe_cache"
COVERS = CACHE / "covers.json"
RESULTS = Path(__file__).parent / "search_external.json"
DATA = Path.home() / "prog/fin/stocks"
GOGET = Path.home() / "go/bin/goget"
UA = {"User-Agent": "research test@example.com"}

# The prospectus cover is line-based SGML with UNCLOSED tags:
#   <SERIES-NAME>1290 GAMCO Small/Mid Cap Value Fund
#   <CLASS-CONTRACT-TICKER-SYMBOL>TNVAX
# so parse line by line, not with closed-tag regexes.


def fetch_index(quarters: list[tuple[str, str]] = QUARTERS) -> dict:
    """cik -> path of its LATEST 497/497K filing's .txt across quarters."""
    rx = re.compile(r"^(.*?)\s+(497\w*)\s+(\d+)\s+(\d{4}-\d{2}-\d{2})\s+"
                    r"(edgar/data/\d+/[0-9\-]+)\.txt\s*$")
    best: dict[str, tuple[tuple, str]] = {}
    for year, qtr in quarters:
        req = urllib.request.Request(_index_url(year, qtr), headers=UA)
        text = gzip.decompress(urllib.request.urlopen(req, timeout=120)
                               .read()).decode("utf-8", "ignore")
        for line in text.splitlines():
            m = rx.match(line)
            if not m:
                continue
            _name, form, cik, date, path = m.groups()
            if form not in ("497", "497K"):   # base/annual carry tickers
                continue
            key = (year, qtr, date)
            if cik not in best or key > best[cik][0]:
                best[cik] = (key, path)
    return {c: p + ".txt" for c, (_k, p) in best.items()}


def parse_cover(text: str) -> list[dict]:
    """Line-based SGML cover -> [{name, tickers[]}].

    The 497 cover uses UNCLOSED tags, one per line:
      <SERIES-NAME>1290 Multi-Alternative Strategies Fund
      <CLASS-CONTRACT-TICKER-SYMBOL>TNMAX
    Each ticker attaches to the most recent <SERIES-NAME>.
    """
    series: list[dict] = []
    cur: dict | None = None
    for line in text.splitlines():
        if line.startswith("<SERIES-NAME>"):
            cur = {"name": line[len("<SERIES-NAME>"):].strip(),
                   "tickers": []}
            series.append(cur)
        elif line.startswith("<CLASS-CONTRACT-TICKER-SYMBOL>") \
                and cur is not None:
            t = line[len("<CLASS-CONTRACT-TICKER-SYMBOL>"):]
            t = t.split("<")[0].strip()
            if re.fullmatch(r"[A-Z][A-Z0-9]{3,8}", t):
                cur["tickers"].append(t)
    return series


def _fetch_one(cik: str, path: str) -> tuple[str, dict]:
    url = ARCH + path
    try:
        req = urllib.request.Request(url, headers=UA)
        raw = urllib.request.urlopen(req, timeout=90).read()
        text = raw.decode("utf-8", "ignore")
        return cik, {"series": parse_cover(text)}
    except Exception as e:
        return cik, {"error": str(e)}


def fetch_covers(max_ciks: int | None = None,
                 workers: int = 4) -> dict:
    from concurrent.futures import ThreadPoolExecutor
    ciks = fetch_index()
    if max_ciks:
        ciks = dict(list(ciks.items())[:max_ciks])
    CACHE.mkdir(exist_ok=True)
    cache = (json.loads(COVERS.read_text()) if COVERS.exists() else {})
    todo = sorted(c for c, p in ciks.items()
                  if c not in cache or "error" in cache.get(c, {}))
    print(f"{len(ciks)} registered funds, {len(todo)} covers to fetch",
          flush=True)
    done = 0
    with ThreadPoolExecutor(max_workers=workers) as ex:
        futs = {ex.submit(_fetch_one, c, ciks[c]): c for c in todo}
        for fut in as_completed(futs):
            cik, res = fut.result()
            cache[cik] = res
            done += 1
            if done % 100 == 0:
                COVERS.write_text(json.dumps(cache))
                print(f"  {done}/{len(todo)} covers", flush=True)
    COVERS.write_text(json.dumps(cache))
    n_err = sum(1 for v in cache.values() if "error" in v)
    print(f"covers done: {len(cache)} ({n_err} errors)", flush=True)
    return cache


def run(max_ciks: int | None = None) -> dict:
    covers = fetch_covers(max_ciks)

    # candidate (name, ticker) pairs, alpha-leaning, not already known
    local = {p.name[:-5].lower() for p in DATA.glob("*.json")}
    shortlist = set(json.loads(
        (Path(__file__).parent.parent / "funds.json").read_text()))
    known = local | shortlist
    cands: dict[str, str] = {}
    for v in covers.values():
        for s in v.get("series", []):
            name = s.get("name", "")
            if not name or ALPHA_KW.search(name) is None:
                continue
            if SKIP.search(name):
                continue
            for t in s.get("tickers", []):
                if t.lower() not in known:
                    cands.setdefault(t, name)
    print(f"{len(cands)} alpha-leaning tickers not in local DB", flush=True)

    # Yahoo verify: OTC fund class + >=5y history
    verified = []
    for t, name in sorted(cands.items()):
        meta = search.chart_meta(t)
        if not meta:
            continue
        # keep OTC open-end fund classes; drop exchange-listed ETFs
        # (Yahoo shows OTC funds as exchange "Nasdaq" - instrumentType
        # is the real discriminator)
        if (meta.get("instrumentType") or "").upper() not in ("MUTUALFUND", ""): 
            continue
        n = 0
        try:
            req = urllib.request.Request(
                f"https://query1.finance.yahoo.com/v8/finance/chart/"
                f"{t}?range=20y&interval=1d", headers=search.UA)
            d = json.load(urllib.request.urlopen(req, timeout=30))
            res = (d.get("chart") or {}).get("result")
            n = len(res[0].get("timestamp", [])) if res else 0
        except Exception:
            pass
        if n >= 1250:
            verified.append({"ticker": t, "name": name, "days": n})
        time.sleep(0.15)
    print(f"{len(verified)} with >=5y daily history", flush=True)

    # one class per fund (share classes share the series name): keep the
    # longest-history class
    by_name: dict[str, dict] = {}
    for v in verified:
        cur = by_name.get(v["name"])
        if cur is None or v["days"] > cur["days"]:
            by_name[v["name"]] = v
    verified = list(by_name.values())
    print(f"{len(verified)} after share-class dedupe", flush=True)

    missing = [v["ticker"].lower() for v in verified
               if not (DATA / f"{v['ticker'].lower()}-history.csv").exists()]
    if missing and GOGET.exists():
        print(f"goget downloading {len(missing)} symbols...", flush=True)
        subprocess.run([str(GOGET), *missing], cwd=DATA,
                       capture_output=True, timeout=3600)

    results = {}
    for v in verified:
        sym = v["ticker"].lower()
        if not (DATA / f"{sym}-history.csv").exists():
            results[sym] = {"sym": sym, "name": v["name"],
                            "error": "no history after download"}
            continue
        row = search.screen_fund(sym, v["name"], "external")
        results[sym] = row
        print(f"{sym:7} {v['name'][:44]:44} "
              f"{row.get('verdict', row.get('error'))[:40]}", flush=True)
    RESULTS.write_text(json.dumps(results, indent=1, default=str))
    print(f"wrote {RESULTS}")
    return results


if __name__ == "__main__":
    import sys
    run(max_ciks=int(sys.argv[1]) if len(sys.argv) > 1 else None)