1"""Point-in-time, stratified instrument universe with an immutable manifest (docs/DATA_CARD.md §universe).23Selection information set = instrument master (first/last date, delisted status) + daily bars up to the selection4date only (average dollar volume, daily volatility). Strata: asset class (stock / ETF), dollar-volume tercile,5volatility tercile, SIC division (stocks, from SEC submissions when available; "unknown" otherwise) or ETF category6(keyword rule on the vendor name). Delisted instruments with 1-minute history in the sample window are eligible; the7only survival requirement is presence for >= `min_months_in_sample` months after the sample start (documented bias).8"""9from __future__ import annotations1011import hashlib12import json13import math14import random15from dataclasses import dataclass, asdict16from datetime import date, datetime, timezone17from pathlib import Path1819import numpy as np20import pandas as pd2122from .client import FinDataClient2324ETF_RULES = [25 ("leveraged_inverse", ["2X", "3X", "ULTRA", "INVERSE", "SHORT ", "BEAR", "BULL", "-1X", "-2X", "-3X", "PROSHARES"]),26 ("fixed_income", ["BOND", "TREASURY", "CREDIT", "MUNI", "AGGREGATE", "FIXED INCOME", "TIPS", "YIELD", "DEBT"]),27 ("commodity", ["GOLD", "SILVER", "OIL", "COMMODIT", "NATURAL GAS", "METAL", "ENERGY FUND", "CRUDE"]),28 ("international", ["EMERGING", "INTERNATIONAL", "EAFE", "CHINA", "JAPAN", "EUROPE", "GLOBAL", "ASIA", "BRAZIL",29 "INDIA", "KOREA", "MEXICO", "CANADA", "WORLD", "FRONTIER", "LATIN"]),30 ("sector", ["SECTOR", "TECHNOLOGY", "FINANCIAL", "HEALTH", "UTILITIES", "ENERGY", "MATERIALS", "INDUSTRIAL",31 "CONSUMER", "REAL ESTATE", "SEMICONDUCTOR", "BIOTECH", "BANK", "RETAIL", "HOMEBUILDER", "TRANSPORT"]),32 ("volatility_alt", ["VIX", "VOLATILITY", "CURRENCY", "DOLLAR", "YEN", "EURO ", "BITCOIN", "CRYPTO"]),33 ("broad_equity", ["S&P", "500", "RUSSELL", "NASDAQ", "DOW", "TOTAL", "MARKET", "VALUE", "GROWTH", "DIVIDEND",34 "MID-CAP", "SMALL-CAP", "LARGE-CAP", "EQUITY", "INDEX", "CORE", "QUALITY", "MOMENTUM"]),35]3637SIC_DIVISIONS = [(100, 999, "agriculture"), (1000, 1499, "mining"), (1500, 1799, "construction"),38 (2000, 3999, "manufacturing"), (4000, 4999, "transport_utilities"), (5000, 5199, "wholesale"),39 (5200, 5999, "retail"), (6000, 6799, "finance_insurance_realestate"), (7000, 8999, "services"),40 (9100, 9999, "public_admin")]414243def sic_division(sic) -> str:44 try:45 s = int(sic)46 except (TypeError, ValueError):47 return "unknown"48 for lo, hi, name in SIC_DIVISIONS:49 if lo <= s <= hi:50 return name51 return "unknown"525354def etf_category(name: str | None) -> str:55 n = (name if isinstance(name, str) else "").upper()56 for cat, kws in ETF_RULES:57 if any(k in n for k in kws):58 return cat59 return "other"606162@dataclass63class UniverseConfig:64 selection_date: str = "2017-12-29"65 adv_window_start: str = "2017-12-01"66 sample_start: str = "2018-01-02"67 sample_end: str = "2025-12-31"68 min_history_months: int = 6 # first_date <= selection_date - 6 months69 min_months_in_sample: int = 6 # last_date >= sample_start + 6 months (survival requirement, documented)70 min_adv_usd: float = 5e6 # average daily dollar volume floor at selection (minute bars need activity)71 n_target: int = 30072 etf_share: float = 0.2573 holdout_share: float = 0.2074 seed: int = 4175 min_coverage: float = 0.0 # regular-hours minute coverage over the sample window (P1 coverage audit gate)76 coverage_file: str | None = None # results/metrics/coverage_audit_candidates.csv777879def daily_cross_sections(client: FinDataClient, start: str, end: str, asset_class: str) -> pd.DataFrame:80 """Daily bars for all symbols on each session in [start, end] (one scoped SQL per session date)."""81 from .calendar import sessions82 view = f"market_bars_{asset_class}_1d_raw"83 frames = []84 for s in sessions(start, end):85 d = s.date.date().isoformat()86 cols, rows, meta = client.sql(f"SELECT symbol_as_reported AS symbol, open, high, low, close, volume "87 f"FROM {view} WHERE session_date = DATE '{d}'", max_rows=100_000)88 df = pd.DataFrame(rows, columns=cols)89 df["session_date"] = d90 frames.append(df)91 out = pd.concat(frames, ignore_index=True)92 out["asset_class"] = asset_class93 return out949596def selection_stats(cross: pd.DataFrame) -> pd.DataFrame:97 """ADV (USD), daily log-return volatility and number of sessions per symbol from the cross-sections."""98 cross = cross.sort_values(["symbol", "session_date"]).copy()99 cross["dollar_volume"] = cross["close"] * cross["volume"]100 cross["lr"] = np.log(cross["close"]).groupby(cross["symbol"]).diff()101 g = cross.groupby("symbol")102 out = pd.DataFrame({"adv_usd": g["dollar_volume"].mean(), "daily_vol": g["lr"].std(), "n_days": g.size(),103 "last_close": g["close"].last(), "asset_class": g["asset_class"].first()})104 return out.reset_index()105106107def load_sic_map(sec_tickers_json: Path | None, sec_submissions_dir: Path | None) -> dict[str, str]:108 """ticker -> SIC division from the SEC company_tickers file + cached submissions (public data, optional)."""109 if not sec_tickers_json or not Path(sec_tickers_json).exists():110 return {}111 data = json.loads(Path(sec_tickers_json).read_text())112 if isinstance(data, dict) and "data" in data and "cik_str" not in data.get("0", {}):113 data = data["data"]114 tick2cik = {}115 for v in (data.values() if isinstance(data, dict) else data):116 tick2cik[str(v["ticker"]).upper()] = int(v["cik_str"])117 out = {}118 if sec_submissions_dir and Path(sec_submissions_dir).exists():119 for t, cik in tick2cik.items():120 p = Path(sec_submissions_dir) / f"CIK{cik:010d}.json"121 if p.exists():122 try:123 out[t] = sic_division(json.loads(p.read_text()).get("sic"))124 except Exception:125 out[t] = "unknown"126 return out127128129def tercile(x: pd.Series) -> pd.Series:130 r = x.rank(pct=True, method="first")131 return pd.cut(r, [0, 1 / 3, 2 / 3, 1.0001], labels=["low", "mid", "high"], include_lowest=True).astype(str)132133134def build_candidates(client: FinDataClient, cfg: UniverseConfig, sic_map: dict[str, str]) -> pd.DataFrame:135 sel = pd.Timestamp(cfg.selection_date)136 first_max = sel - pd.DateOffset(months=cfg.min_history_months)137 last_min = pd.Timestamp(cfg.sample_start) + pd.DateOffset(months=cfg.min_months_in_sample)138 inst = []139 for ac in ("stock", "etf"):140 for r in client.instruments(asset_class=ac):141 inst.append(r)142 master = pd.DataFrame(inst)143 master = master[master["asset_class"].isin(["stock", "etf"])].copy()144 master["first_date"] = pd.to_datetime(master["first_date"])145 master["last_date"] = pd.to_datetime(master["last_date"])146 elig = master[(master["first_date"] <= first_max) & (master["last_date"] >= last_min)].copy()147 stats = pd.concat([selection_stats(daily_cross_sections(client, cfg.adv_window_start, cfg.selection_date, ac))148 for ac in ("stock", "etf")], ignore_index=True)149 stats = stats.drop(columns=["asset_class"])150 cand = elig.merge(stats, on="symbol", how="inner")151 cand = cand[(cand["adv_usd"] >= cfg.min_adv_usd) & (cand["n_days"] >= 15) & cand["daily_vol"].notna()].copy()152 if cfg.coverage_file:153 cov = pd.read_csv(cfg.coverage_file)[["symbol", "coverage", "sessions_present"]]154 cand = cand.merge(cov, on="symbol", how="left")155 cand["coverage"] = cand["coverage"].fillna(0.0)156 cand = cand[cand["coverage"] >= cfg.min_coverage].copy()157 cand["base_symbol"] = cand["symbol"].str.replace("-DELISTED", "", regex=False)158 cand["delisted"] = cand["status"].eq("delisted")159 cand["group"] = np.where(cand["asset_class"].eq("etf"), cand["name"].map(etf_category),160 cand["base_symbol"].map(lambda t: sic_map.get(t, "unknown")))161 cand["adv_tercile"] = cand.groupby("asset_class")["adv_usd"].transform(tercile)162 cand["vol_tercile"] = cand.groupby("asset_class")["daily_vol"].transform(tercile)163 cand["stratum"] = cand["asset_class"] + "|" + cand["adv_tercile"] + "|" + cand["vol_tercile"]164 return cand.sort_values("symbol").reset_index(drop=True)165166167def stratified_sample(cand: pd.DataFrame, cfg: UniverseConfig) -> pd.DataFrame:168 """Proportional allocation across strata (asset class × ADV tercile × vol tercile), delisted instruments kept with169 their natural weight, sector/ETF category balanced inside each stratum by round-robin."""170 rng = random.Random(cfg.seed)171 n_etf = int(round(cfg.n_target * cfg.etf_share))172 n_stock = cfg.n_target - n_etf173 chosen = []174 for ac, n_ac in (("stock", n_stock), ("etf", n_etf)):175 sub = cand[cand["asset_class"] == ac]176 strata = sorted(sub["stratum"].unique())177 if not strata:178 continue179 quota = {s: n_ac // len(strata) for s in strata}180 for s in rng.sample(strata, n_ac - sum(quota.values())):181 quota[s] += 1182 for s in strata:183 pool = sub[sub["stratum"] == s]184 groups = {}185 for sym, grp in zip(pool["symbol"], pool["group"]):186 groups.setdefault(grp, []).append(sym)187 for g in groups.values():188 rng.shuffle(g)189 order = sorted(groups.keys(), key=lambda k: rng.random())190 picked = []191 while len(picked) < min(quota[s], pool.shape[0]):192 for g in order:193 if groups[g] and len(picked) < quota[s]:194 picked.append(groups[g].pop())195 chosen.extend(picked)196 out = cand[cand["symbol"].isin(chosen)].copy().sort_values("symbol").reset_index(drop=True)197 syms = out["symbol"].tolist()198 rng2 = random.Random(cfg.seed + 1)199 rng2.shuffle(syms)200 n_hold = int(round(len(syms) * cfg.holdout_share))201 hold = set(syms[:n_hold])202 out["role"] = np.where(out["symbol"].isin(hold), "heldout_instrument", "pooled")203 return out204205206def manifest(sample: pd.DataFrame, cand: pd.DataFrame, cfg: UniverseConfig, dataset_version: str | None,207 extra: dict | None = None) -> dict:208 rows = sample.to_dict(orient="records")209 for r in rows:210 for k, v in list(r.items()):211 if isinstance(v, (pd.Timestamp, datetime, date)):212 r[k] = str(v)[:10]213 elif isinstance(v, float) and math.isnan(v):214 r[k] = None215 elif isinstance(v, (np.integer,)):216 r[k] = int(v)217 elif isinstance(v, (np.floating,)):218 r[k] = float(v)219 elif isinstance(v, (np.bool_,)):220 r[k] = bool(v)221 body = {"config": asdict(cfg), "created_at": datetime.now(timezone.utc).isoformat(),222 "n_candidates": int(cand.shape[0]), "n_selected": len(rows),223 "n_heldout": int((sample["role"] == "heldout_instrument").sum()),224 "n_delisted": int(sample["delisted"].sum()), "instrument_master_version": dataset_version,225 "selection_rules": [226 f"first_date <= {cfg.selection_date} - {cfg.min_history_months} months",227 f"last_date >= {cfg.sample_start} + {cfg.min_months_in_sample} months (documented survival requirement)",228 f"average daily dollar volume over [{cfg.adv_window_start}, {cfg.selection_date}] >= {cfg.min_adv_usd:,.0f} USD",229 "proportional allocation over asset class x ADV tercile x volatility tercile; round-robin over SIC division / ETF category inside strata",230 f"{cfg.holdout_share:.0%} of selected instruments held out entirely (zero-shot), seed {cfg.seed}",231 f"regular-hours 1-minute coverage over the sample window >= {cfg.min_coverage:.0%} (coverage audit, availability filter)",232 ], "instruments": rows}233 if extra:234 body.update(extra)235 body["sha256"] = hashlib.sha256(json.dumps(body, sort_keys=True, default=str).encode()).hexdigest()236 return body237