quant-trading/research/lab/fetch_us_fundamentals.py

"""
Fetch point-in-time US fundamentals from SEC EDGAR (XBRL companyfacts) for
every S&P 500 ever-member in the research price panel, plus Yahoo split
history (needed to turn split-adjusted prices back into market caps).

    uv run python -m research.lab.fetch_us_fundamentals
    uv run python -m research.lab.fetch_us_fundamentals --universe mid   # S&P 400 (IJH) ever-members

Writes data/research/us_fund_raw.pkl (us_mid_fund_raw.pkl for --universe mid):
    {"facts": DataFrame[ticker, tag, start, end, val, filed, form],
     "splits": DataFrame (dates x tickers, split ratio on split days)}

Every fact keeps its FIRST filing date: a value restated in a later filing is
not visible before that later date, so factors built from "filed" are
point-in-time. XBRL coverage starts in 2009-2011 depending on filer size.
"""

import os
import pickle
import time
from concurrent.futures import ThreadPoolExecutor

import pandas as pd
import requests
import yfinance as yf

from research.lab.fetch_us import _proxy_env

OUT = os.path.join("data", "research", "us_fund_raw.pkl")
MID_OUT = os.path.join("data", "research", "us_mid_fund_raw.pkl")
UA = {"User-Agent": "quant-research/0.1 gahow@example.com", "Accept-Encoding": "gzip"}
FORMS = {"10-K", "10-Q", "10-K/A", "10-Q/A", "10-KT", "20-F", "40-F"}

TAGS = [
    # income statement (durations)
    "Revenues", "RevenueFromContractWithCustomerExcludingAssessedTax",
    "RevenueFromContractWithCustomerIncludingAssessedTax", "SalesRevenueNet",
    "SalesRevenueGoodsNet", "SalesRevenueServicesNet", "RevenuesNetOfInterestExpense",
    "CostOfRevenue", "CostOfGoodsAndServicesSold", "CostOfGoodsSold", "CostOfServices",
    "GrossProfit", "OperatingIncomeLoss", "NetIncomeLoss", "ProfitLoss",
    "NetIncomeLossAvailableToCommonStockholdersBasic",
    "IncomeLossFromContinuingOperationsBeforeIncomeTaxesExtraordinaryItemsNoncontrollingInterest",
    "ResearchAndDevelopmentExpense", "SellingGeneralAndAdministrativeExpense",
    "InterestExpense", "IncomeTaxExpenseBenefit", "EarningsPerShareDiluted",
    "WeightedAverageNumberOfDilutedSharesOutstanding",
    "DepreciationDepletionAndAmortization", "DepreciationAndAmortization",
    # cash flow (year-to-date durations in 10-Qs)
    "NetCashProvidedByUsedInOperatingActivities",
    "NetCashProvidedByUsedInOperatingActivitiesContinuingOperations",
    "PaymentsToAcquirePropertyPlantAndEquipment", "PaymentsForRepurchaseOfCommonStock",
    "PaymentsOfDividends", "PaymentsOfDividendsCommonStock",
    # balance sheet (instants)
    "Assets", "AssetsCurrent", "Liabilities", "LiabilitiesCurrent",
    "LiabilitiesAndStockholdersEquity", "StockholdersEquity",
    "StockholdersEquityIncludingPortionAttributableToNoncontrollingInterest",
    "CashAndCashEquivalentsAtCarryingValue", "LongTermDebt", "LongTermDebtNoncurrent",
    "LongTermDebtCurrent", "DebtCurrent", "ShortTermBorrowings", "InventoryNet",
    "AccountsReceivableNetCurrent", "AccountsPayableCurrent", "PropertyPlantAndEquipmentNet",
    "Goodwill", "RetainedEarningsAccumulatedDeficit", "CommonStockSharesOutstanding",
]
DEI_TAGS = ["EntityCommonStockSharesOutstanding"]


def cik_map() -> dict[str, int]:
    r = requests.get("https://www.sec.gov/files/company_tickers.json", headers=UA, timeout=60)
    r.raise_for_status()
    return {v["ticker"].upper().replace(".", "-"): int(v["cik_str"]) for v in r.json().values()}


def _facts_for(ticker: str, cik: int) -> pd.DataFrame | None:
    url = f"https://data.sec.gov/api/xbrl/companyfacts/CIK{cik:010d}.json"
    for attempt in range(4):
        try:
            r = requests.get(url, headers=UA, timeout=120)
            if r.status_code == 404:
                return None
            r.raise_for_status()
            js = r.json()
            break
        except Exception as exc:  # noqa: BLE001 — retry transient SEC errors
            print(f"  {ticker}: {exc}; retry {attempt + 1}", flush=True)
            time.sleep(5 * (attempt + 1))
    else:
        return None
    rows = []
    for ns, tags in (("us-gaap", TAGS), ("dei", DEI_TAGS)):
        facts = js.get("facts", {}).get(ns, {})
        for tag in tags:
            if tag not in facts:
                continue
            for unit, xs in facts[tag]["units"].items():
                if unit not in ("USD", "shares", "USD/shares"):
                    continue
                for x in xs:
                    if x.get("form") not in FORMS:
                        continue
                    rows.append((ticker, tag, x.get("start"), x["end"], float(x["val"]),
                                 x["filed"], x["form"]))
    if not rows:
        return None
    df = pd.DataFrame(rows, columns=["ticker", "tag", "start", "end", "val", "filed", "form"])
    # first filing of each (tag, period) wins: later restatements are not
    # knowable on the original date
    df = df.sort_values("filed").drop_duplicates(["tag", "start", "end"], keep="first")
    return df


def fetch_splits(tickers: list[str], chunk: int = 60) -> tuple[pd.DataFrame, pd.DataFrame]:
    """Split ratios (on split days, 0 elsewhere) and split-adjusted closes
    that are NOT dividend-adjusted; market cap = shares reported at c x
    close_sa[t] / (product of split ratios after c)."""
    splits, closes = [], []

    def run(batch):
        raw = yf.download(batch, start="2004-01-01", actions=True, auto_adjust=False,
                          progress=False, group_by="column", threads=True)
        lv = raw.columns.get_level_values(0)
        if "Stock Splits" in lv:
            splits.append(raw["Stock Splits"])
        if "Close" in lv:
            closes.append(raw["Close"].dropna(axis=1, how="all"))
        time.sleep(2)

    def cat(parts):
        s = pd.concat(parts, axis=1)
        return s.loc[:, ~s.columns.duplicated(keep="last")]

    for i in range(0, len(tickers), chunk):
        run(tickers[i:i + chunk])
    # Yahoo fails whole batches now and then; retry the missing names in small batches
    for attempt in range(3):
        have = set(cat(closes).columns)
        missing = [t for t in tickers if t not in have]
        if not missing:
            break
        print(f"  retrying {len(missing)} tickers", flush=True)
        for i in range(0, len(missing), 15):
            run(missing[i:i + 15])
    return cat(splits).fillna(0.0), cat(closes)


_SUFFIXES = {"INC", "CORP", "CORPORATION", "CO", "COMPANY", "LTD", "PLC", "HOLDINGS", "HOLDING",
             "GROUP", "THE", "CLASS", "CL", "A", "B", "NEW", "LLC", "LP", "SA", "NV", "TRUST"}


def _name_tokens(name: str) -> set[str]:
    words = "".join(ch if ch.isalnum() else " " for ch in str(name).upper()).split()
    return {w for w in words if w not in _SUFFIXES and len(w) > 1}


def mid_ciks(tickers: list[str]) -> tuple[dict[str, int], pd.DataFrame]:
    """CIKs for S&P 400 tickers. Candidates come from sec_sic.csv and
    mid_sec_sic.csv (earlier runs), then today's company_tickers.json; both
    map a ticker to whoever holds it NOW, so a candidate counts only if one of
    its SEC names (current or former) shares a word with an iShares holding
    name of that ticker. Returns the accepted map and a per-ticker audit."""
    import json
    from research.lab.fetch_us_mid import load_holdings
    base = os.path.join("data", "research")
    cand: dict[str, int] = {}
    for f in ("sec_sic.csv", "mid_sec_sic.csv"):
        d = pd.read_csv(os.path.join(base, f)).dropna(subset=["cik"])
        cand.update({t: int(c) for t, c in zip(d["ticker"], d["cik"]) if t not in cand})
    live = cik_map()
    for t in tickers:
        cand.setdefault(t, live.get(t, 0))
    names: dict[str, list[str]] = {}
    for f in ("sec_names.json", "mid_sec_names.json"):
        with open(os.path.join(base, f)) as fh:
            names.update(json.load(fh))
    for t in tickers:
        c = cand.get(t, 0)
        if c and str(c) not in names:
            r = requests.get(f"https://data.sec.gov/submissions/CIK{c:010d}.json", headers=UA, timeout=60)
            time.sleep(0.15)
            if r.ok:
                js = r.json()
                names[str(c)] = [js.get("name", "")] + [x.get("name", "") for x in js.get("formerNames", [])]
    hold = load_holdings()
    hnames = hold[hold["fund"] == "IJH"].groupby("ticker")["name"].apply(lambda s: set(s.dropna()))
    rows, ok = [], {}
    for t in tickers:
        c = cand.get(t, 0)
        sec = names.get(str(c), []) if c else []
        hw = set().union(*[_name_tokens(n) for n in hnames.get(t, set())]) if t in hnames else set()
        sw = set().union(*[_name_tokens(n) for n in sec]) if sec else set()
        # token overlap, or the same leading letters once spacing and
        # punctuation are gone (MACYS / Macy's, SIRIUSXM / Sirius XM, VF / V F)
        squash = lambda n: "".join(w for w in "".join(ch if ch.isalnum() else " " for ch in str(n).upper()).split()
                                   if w not in _SUFFIXES)
        hs = {squash(n)[:5] for n in hnames.get(t, set())} - {""}
        ss = {squash(n)[:5] for n in sec} - {""}
        match = bool(hw & sw) or bool(hs & ss)
        if c and match:
            ok[t] = c
        rows.append({"ticker": t, "cik": c or None, "ishares_names": " | ".join(sorted(hnames.get(t, set()))),
                     "sec_names": " | ".join(sec), "accepted": bool(c and match)})
    return ok, pd.DataFrame(rows)


def main() -> None:
    import sys
    global OUT
    if "--universe" in sys.argv and sys.argv[sys.argv.index("--universe") + 1] == "mid":
        from research.lab.panel_mid import load_us_mid
        OUT = MID_OUT
        tickers = list(load_us_mid("mid").close.columns)
        ciks, audit = mid_ciks(tickers)
        audit.to_csv(os.path.join("data", "research", "us_mid_fund_cik_audit.csv"), index=False)
        print(f"{len(ciks)}/{len(tickers)} S&P 400 tickers have a name-checked CIK", flush=True)
    else:
        with open(os.path.join("data", "research", "us_ohlcv.pkl"), "rb") as fh:
            tickers = [t for t in pickle.load(fh)["close"].columns]
        ciks = None
    if "--yahoo-only" in sys.argv:        # refresh splits/closes of an existing file
        with open(OUT, "rb") as fh:
            saved = pickle.load(fh)
        _proxy_env()
        saved["splits"], saved["close_sa"] = fetch_splits(tickers)
        with open(OUT, "wb") as fh:
            pickle.dump(saved, fh)
        print(f"splits: {int((saved['splits'] > 0).sum().sum())} events; saved {OUT}")
        return
    ciks = ciks if ciks is not None else cik_map()
    todo = [(t, ciks[t]) for t in tickers if t in ciks]
    print(f"{len(todo)}/{len(tickers)} tickers map to a CIK", flush=True)

    frames, t0 = [], time.time()

    def job(tc):
        time.sleep(0.25)            # stays well under SEC's 10 requests/second
        return _facts_for(*tc)

    with ThreadPoolExecutor(max_workers=3) as ex:
        for k, df in enumerate(ex.map(job, todo), 1):
            if df is not None:
                frames.append(df)
            if k % 50 == 0:
                print(f"  {k}/{len(todo)} companies [{time.time() - t0:.0f}s]", flush=True)
    facts = pd.concat(frames, ignore_index=True)
    for c in ("start", "end", "filed"):
        facts[c] = pd.to_datetime(facts[c])
    print(f"{len(facts):,} facts for {facts['ticker'].nunique()} tickers", flush=True)

    _proxy_env()
    splits, close_sa = fetch_splits(tickers)
    print(f"splits: {int((splits > 0).sum().sum())} events", flush=True)
    with open(OUT, "wb") as fh:
        pickle.dump({"facts": facts, "splits": splits, "close_sa": close_sa,
                     "unmapped": sorted(set(tickers) - set(ciks))}, fh)
    print(f"Saved {OUT}")


if __name__ == "__main__":
    main()