spb/uqo-research-db
Public
1# ============================================================2# Project: EDGAR Research Database3# Description: SEC EDGAR Research Infrastructure4#5# Authors:6# - Manel Kammoun — manel.kammoun@uqo.ca7# - Djerrycharli Tandjambianda — djerrycharli.tandjambianda@uqo.ca8# - Simon-Pierre Boucher — simon-pierre.boucher@uqo.ca9#10# Institution:11# Université du Québec en Outaouais (UQO)12#13# License:14# Research / Academic Use15# ============================================================1617"""Company universes beyond the S&P 500.1819``Russell3000Universe`` reads the public holdings file of the iShares Russell203000 ETF (IWV), the standard investable proxy for the index, and keeps the21equity lines that can be mapped to an SEC CIK. ``build_universe`` merges the22cached S&P 500 constituents with the Russell 3000 list into a single23``data/metadata/universe.json`` used by every downstream script.24"""2526from __future__ import annotations2728import csv29import io30import json31import re32from datetime import datetime, timezone33from pathlib import Path3435import httpx3637from src.acquisition.cik_mapper import CIKMapper38from src.utils.constants import DATA_DIRS39from src.utils.logging import get_logger4041logger = get_logger("acquisition.universe")4243ISHARES_IWV_URL = (44 "https://www.ishares.com/us/products/239714/ishares-russell-3000-etf/latest-holdings.csv"45)46BROWSER_UA = (47 "Mozilla/5.0 (Macintosh; Intel Mac OS X 14_0) AppleWebKit/537.36 "48 "(KHTML, like Gecko) Chrome/124.0 Safari/537.36"49)5051# iShares sector labels -> GICS sector names used by the S&P 500 cache52SECTOR_MAP = {53 "Communication": "Communication Services",54 "Information Technology": "Information Technology",55 "Financials": "Financials",56 "Health Care": "Health Care",57 "Industrials": "Industrials",58 "Consumer Discretionary": "Consumer Discretionary",59 "Consumer Staples": "Consumer Staples",60 "Energy": "Energy",61 "Materials": "Materials",62 "Real Estate": "Real Estate",63 "Utilities": "Utilities",64}6566_NON_COMPANY = re.compile(r"\b(CVR|ESCROW|RIGHTS?|WHEN ISSUED|WARRANT)\b", re.I)676869def normalize_ticker(ticker: str) -> str:70 """'BRK B' / 'BRK.B' -> 'BRK-B' (SEC convention)."""71 return re.sub(r"[ .]", "-", ticker.strip().upper())727374class Russell3000Universe:75 def __init__(self, cache_dir: Path | None = None) -> None:76 self._cache_dir = cache_dir or DATA_DIRS["metadata"]77 self._cache_path = self._cache_dir / "russell3000_constituents.json"7879 def fetch_holdings_csv(self) -> str:80 resp = httpx.get(81 ISHARES_IWV_URL,82 headers={"User-Agent": BROWSER_UA, "Accept": "text/csv,*/*"},83 follow_redirects=True,84 timeout=60,85 )86 resp.raise_for_status()87 text = resp.text88 if "Ticker,Name" not in text:89 raise RuntimeError("iShares holdings file did not contain a holdings table")90 return text9192 @staticmethod93 def parse_holdings(text: str) -> tuple[str, list[dict]]:94 lines = text.splitlines()95 as_of = ""96 for line in lines[:10]:97 if line.startswith("Fund Holdings as of"):98 as_of = line.split(",", 1)[1].strip().strip('"')99 start = next(i for i, l in enumerate(lines) if l.startswith("Ticker,Name"))100 reader = csv.DictReader(io.StringIO("\n".join(lines[start:])))101 rows: list[dict] = []102 for r in reader:103 ticker = (r.get("Ticker") or "").strip()104 if r.get("Asset Class") != "Equity" or ticker in ("", "-"):105 continue106 if _NON_COMPANY.search(r.get("Name") or ""):107 continue108 rows.append({109 "ticker_raw": ticker,110 "ticker": normalize_ticker(ticker),111 "company_name": (r.get("Name") or "").strip(),112 "sector": SECTOR_MAP.get((r.get("Sector") or "").strip(), (r.get("Sector") or "").strip()),113 "location": (r.get("Location") or "").strip(),114 "exchange": (r.get("Exchange") or "").strip(),115 "weight_pct": float((r.get("Weight (%)") or "0").replace(",", "") or 0),116 "market_value_usd": float((r.get("Market Value") or "0").replace(",", "") or 0),117 })118 return as_of, rows119120 def get_constituents(self, cik_mapper: CIKMapper | None = None) -> list[dict]:121 text = self.fetch_holdings_csv()122 as_of, rows = self.parse_holdings(text)123 mapper = cik_mapper or CIKMapper()124 if not mapper._ticker_to_cik: # noqa: SLF001 - force a fresh SEC map125 mapper.build_mapping()126127 out: list[dict] = []128 unmatched: list[str] = []129 for r in rows:130 cik = mapper.get_cik(r["ticker"]) or mapper.get_cik(r["ticker_raw"])131 if not cik:132 unmatched.append(f"{r['ticker_raw']} ({r['company_name']})")133 continue134 out.append({**r, "cik": cik})135 logger.info(136 "Russell 3000 (IWV as of %s): %d equity lines, %d mapped to a CIK, %d unmatched",137 as_of, len(rows), len(out), len(unmatched),138 )139 if unmatched:140 logger.info("Unmatched tickers: %s", "; ".join(unmatched))141 self._save_cache(as_of, out, unmatched)142 return out143144 def load_cached(self) -> list[dict]:145 if not self._cache_path.exists():146 return self.get_constituents()147 with open(self._cache_path) as f:148 return json.load(f)["constituents"]149150 def _save_cache(self, as_of: str, constituents: list[dict], unmatched: list[str]) -> None:151 self._cache_dir.mkdir(parents=True, exist_ok=True)152 payload = {153 "source": ISHARES_IWV_URL,154 "holdings_as_of": as_of,155 "fetched_at": datetime.now(timezone.utc).isoformat(),156 "count": len(constituents),157 "unmatched": unmatched,158 "constituents": constituents,159 }160 with open(self._cache_path, "w") as f:161 json.dump(payload, f, indent=2)162 logger.info("Cached %d Russell 3000 constituents to %s", len(constituents), self._cache_path)163164165def build_universe(166 sp500: list[dict],167 russell: list[dict],168 out_path: Path | None = None,169) -> list[dict]:170 """Merge the two lists on CIK (10 digits). S&P 500 names/sectors win."""171 by_cik: dict[str, dict] = {}172 for r in russell:173 cik = str(r["cik"]).zfill(10)174 entry = by_cik.setdefault(cik, {175 "cik": cik, "ticker": r["ticker"], "company_name": r["company_name"],176 "sector": r.get("sector", ""), "sub_industry": "", "date_added": "",177 "in_sp500": False, "in_russell3000": True,178 "russell_weight_pct": 0.0, "tickers": [],179 })180 entry["russell_weight_pct"] += r.get("weight_pct", 0.0)181 if r["ticker"] not in entry["tickers"]:182 entry["tickers"].append(r["ticker"])183 # prefer the most liquid share class as the primary ticker184 if r.get("weight_pct", 0.0) > max(185 (x.get("weight_pct", 0.0) for x in russell if x["ticker"] == entry["ticker"]), default=0.0186 ):187 entry["ticker"] = r["ticker"]188189 for c in sp500:190 if not c.get("cik"):191 continue192 cik = str(c["cik"]).zfill(10)193 entry = by_cik.get(cik)194 if entry is None:195 entry = by_cik[cik] = {196 "cik": cik, "ticker": c["ticker"], "company_name": c["company_name"],197 "sector": c.get("sector", ""), "sub_industry": c.get("sub_industry", ""),198 "date_added": c.get("date_added", ""), "in_sp500": True,199 "in_russell3000": False, "russell_weight_pct": 0.0, "tickers": [c["ticker"]],200 }201 else:202 entry["in_sp500"] = True203 entry["ticker"] = c["ticker"]204 entry["company_name"] = c["company_name"]205 entry["sector"] = c.get("sector") or entry["sector"]206 entry["sub_industry"] = c.get("sub_industry", "")207 entry["date_added"] = c.get("date_added", "")208 if c["ticker"] not in entry["tickers"]:209 entry["tickers"].append(c["ticker"])210211 universe = sorted(by_cik.values(), key=lambda e: e["cik"])212 out_path = out_path or DATA_DIRS["metadata"] / "universe.json"213 payload = {214 "built_at": datetime.now(timezone.utc).isoformat(),215 "count": len(universe),216 "n_sp500": sum(e["in_sp500"] for e in universe),217 "n_russell3000": sum(e["in_russell3000"] for e in universe),218 "constituents": universe,219 }220 with open(out_path, "w") as f:221 json.dump(payload, f, indent=2)222 logger.info(223 "Universe: %d companies (%d S&P 500, %d Russell 3000) -> %s",224 len(universe), payload["n_sp500"], payload["n_russell3000"], out_path,225 )226 return universe227228229def load_universe(path: Path | None = None) -> list[dict]:230 path = path or DATA_DIRS["metadata"] / "universe.json"231 with open(path) as f:232 return json.load(f)["constituents"]233