""" Point-in-time S&P 500 membership and historical market-cap proxies for backtests. Uses the ``index-constitution`` package for membership (add/remove dates) and yfinance ``get_shares_full`` for quarterly shares outstanding (cached locally). Daily market cap ≈ ``close × shares_outstanding`` (shares forward-filled). """ from __future__ import annotations import json import time from pathlib import Path import numpy as np import pandas as pd _REPO = Path(__file__).resolve().parents[2] UNIVERSE_DIR = _REPO / "RenTech" / "data" / "universe" SHARES_CACHE = UNIVERSE_DIR / "sp500_shares_outstanding.parquet" SHARES_META = UNIVERSE_DIR / "sp500_shares_outstanding_meta.json" try: import index_constitution as ic except ImportError as e: # pragma: no cover ic = None # type: ignore[assignment] _IC_IMPORT_ERROR = e else: _IC_IMPORT_ERROR = None def _require_index_constitution() -> None: if ic is None: raise ImportError( "index-constitution is required for point-in-time S&P 500 membership. " "Install with: pip install index-constitution\n" f"Original error: {_IC_IMPORT_ERROR}" ) def _to_yahoo(symbol: str) -> str: return str(symbol).strip().upper().replace(".", "-") def sp500_members_at(asof: str | pd.Timestamp) -> set[str]: """Return Yahoo-style tickers in the S&P 500 on ``asof`` (date string or Timestamp).""" _require_index_constitution() dt = pd.Timestamp(asof).strftime("%Y-%m-%d") df = ic.constituents_at("sp500", dt) sym_col = "symbol" if "symbol" in df.columns else df.columns[0] return {_to_yahoo(s) for s in df[sym_col].astype(str)} def sp500_union_tickers(start: str | pd.Timestamp, end: str | pd.Timestamp) -> list[str]: """ All canonical tickers that were S&P 500 members at any rebalance between ``start`` and ``end``. """ _require_index_constitution() hist = ic.history("sp500") sym_col = "symbol" if "symbol" in hist.columns else hist.columns[0] opt_in = pd.to_datetime(hist["opt-in"]).dt.tz_localize(None) opt_out = pd.to_datetime(hist["opt-out"]).dt.tz_localize(None) t0 = pd.Timestamp(start).normalize() t1 = pd.Timestamp(end).normalize() keep = (opt_in <= t1) & (opt_out.isna() | (opt_out >= t0)) syms = sorted({_to_yahoo(s) for s in hist.loc[keep, sym_col].astype(str)}) return syms def sp500_membership_mask( tickers: list[str], rebalance_dates: pd.DatetimeIndex, ) -> pd.DataFrame: """ Boolean panel: index = rebalance dates, columns = ``tickers``. True when the ticker was an S&P 500 member on that reference date. """ _require_index_constitution() hist = ic.history("sp500") sym_col = "symbol" if "symbol" in hist.columns else hist.columns[0] hist = hist.copy() hist["symbol"] = hist[sym_col].astype(str).map(_to_yahoo) hist["opt-in"] = pd.to_datetime(hist["opt-in"]).dt.tz_localize(None) hist["opt-out"] = pd.to_datetime(hist["opt-out"]).dt.tz_localize(None) out = pd.DataFrame(False, index=rebalance_dates, columns=tickers, dtype=bool) for _, row in hist.iterrows(): sym = str(row["symbol"]) if sym not in out.columns: continue tin = pd.Timestamp(row["opt-in"]).tz_localize(None) if pd.notna(row["opt-in"]) else pd.Timestamp.min tout = pd.Timestamp(row["opt-out"]).tz_localize(None) if pd.notna(row["opt-out"]) else pd.Timestamp.max active = (rebalance_dates >= tin) & (rebalance_dates <= tout) out.loc[active, sym] = True return out def _fetch_shares_series(ticker: str, start: str) -> pd.Series: import yfinance as yf t = yf.Ticker(ticker) try: sh = t.get_shares_full(start=start, end=None) except Exception: sh = None if sh is None or len(sh) == 0: info = getattr(t, "fast_info", None) or {} so = None try: so = t.info.get("sharesOutstanding") except Exception: so = None if so is None and hasattr(info, "shares"): so = info.shares if so is None or not np.isfinite(float(so)): return pd.Series(dtype=np.float64, name=ticker) return pd.Series({pd.Timestamp.utcnow().normalize(): float(so)}, name=ticker) idx = pd.to_datetime(sh.index).tz_localize(None) s = pd.Series(sh.to_numpy(dtype=np.float64), index=idx, name=ticker).sort_index() s = s[~s.index.duplicated(keep="last")] return s def load_shares_outstanding_dict( tickers: list[str], *, start: str = "1995-01-01", refresh: bool = False, sleep_sec: float = 0.15, ) -> dict[str, pd.Series]: """ Quarterly shares-outstanding series per ticker (cached on disk). """ UNIVERSE_DIR.mkdir(parents=True, exist_ok=True) cached: dict[str, pd.Series] = {} if SHARES_CACHE.is_file() and not refresh: df = pd.read_parquet(SHARES_CACHE) for col in df.columns: cached[str(col)] = df[col].dropna().astype(np.float64) missing = [t for t in tickers if t not in cached or cached[t].empty] if missing: print(f" Fetching shares outstanding for {len(missing)} tickers …", flush=True) for i, t in enumerate(missing): cached[t] = _fetch_shares_series(t, start) if sleep_sec > 0 and i + 1 < len(missing): time.sleep(sleep_sec) wide = pd.DataFrame({t: cached.get(t, pd.Series(dtype=float)) for t in tickers}, dtype=float) wide = wide.loc[:, ~wide.columns.duplicated()] wide.to_parquet(SHARES_CACHE) meta = { "n_tickers": len(tickers), "updated": pd.Timestamp.utcnow().isoformat(), "source": "yfinance get_shares_full", } SHARES_META.write_text(json.dumps(meta, indent=2) + "\n", encoding="utf-8") return {t: cached.get(t, pd.Series(dtype=np.float64)) for t in tickers} def build_market_cap_panel( close_df: pd.DataFrame, shares_dict: dict[str, pd.Series], ) -> pd.DataFrame: """Wide daily market-cap panel aligned to ``close_df`` index/columns.""" mcap_pan: list[pd.Series] = [] master = pd.DatetimeIndex(close_df.index).sort_values() for t in close_df.columns: close = close_df[t].astype(np.float64) sh = shares_dict.get(str(t), pd.Series(dtype=np.float64)) if sh is None or sh.empty: mcap_pan.append((close * np.nan).rename(t)) continue sh = sh.dropna().sort_index() sh_d = sh.reindex(master.union(sh.index).sort_values()).ffill().reindex(master) mcap_pan.append((close * sh_d).rename(t)) out = pd.concat(mcap_pan, axis=1) out.columns = close_df.columns return out.reindex(index=master, columns=close_df.columns)