# ============================================================ # Project: EDGAR Research Database # Description: SEC EDGAR Research Infrastructure # # Authors: # - Manel Kammoun — manel.kammoun@uqo.ca # - Djerrycharli Tandjambianda — djerrycharli.tandjambianda@uqo.ca # - Simon-Pierre Boucher — simon-pierre.boucher@uqo.ca # # Institution: # Université du Québec en Outaouais (UQO) # # License: # Research / Academic Use # ============================================================ """Company universes beyond the S&P 500. ``Russell3000Universe`` reads the public holdings file of the iShares Russell 3000 ETF (IWV), the standard investable proxy for the index, and keeps the equity lines that can be mapped to an SEC CIK. ``build_universe`` merges the cached S&P 500 constituents with the Russell 3000 list into a single ``data/metadata/universe.json`` used by every downstream script. """ from __future__ import annotations import csv import io import json import re from datetime import datetime, timezone from pathlib import Path import httpx from src.acquisition.cik_mapper import CIKMapper from src.utils.constants import DATA_DIRS from src.utils.logging import get_logger logger = get_logger("acquisition.universe") ISHARES_IWV_URL = ( "https://www.ishares.com/us/products/239714/ishares-russell-3000-etf/latest-holdings.csv" ) BROWSER_UA = ( "Mozilla/5.0 (Macintosh; Intel Mac OS X 14_0) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/124.0 Safari/537.36" ) # iShares sector labels -> GICS sector names used by the S&P 500 cache SECTOR_MAP = { "Communication": "Communication Services", "Information Technology": "Information Technology", "Financials": "Financials", "Health Care": "Health Care", "Industrials": "Industrials", "Consumer Discretionary": "Consumer Discretionary", "Consumer Staples": "Consumer Staples", "Energy": "Energy", "Materials": "Materials", "Real Estate": "Real Estate", "Utilities": "Utilities", } _NON_COMPANY = re.compile(r"\b(CVR|ESCROW|RIGHTS?|WHEN ISSUED|WARRANT)\b", re.I) def normalize_ticker(ticker: str) -> str: """'BRK B' / 'BRK.B' -> 'BRK-B' (SEC convention).""" return re.sub(r"[ .]", "-", ticker.strip().upper()) class Russell3000Universe: def __init__(self, cache_dir: Path | None = None) -> None: self._cache_dir = cache_dir or DATA_DIRS["metadata"] self._cache_path = self._cache_dir / "russell3000_constituents.json" def fetch_holdings_csv(self) -> str: resp = httpx.get( ISHARES_IWV_URL, headers={"User-Agent": BROWSER_UA, "Accept": "text/csv,*/*"}, follow_redirects=True, timeout=60, ) resp.raise_for_status() text = resp.text if "Ticker,Name" not in text: raise RuntimeError("iShares holdings file did not contain a holdings table") return text @staticmethod def parse_holdings(text: str) -> tuple[str, list[dict]]: lines = text.splitlines() as_of = "" for line in lines[:10]: if line.startswith("Fund Holdings as of"): as_of = line.split(",", 1)[1].strip().strip('"') start = next(i for i, l in enumerate(lines) if l.startswith("Ticker,Name")) reader = csv.DictReader(io.StringIO("\n".join(lines[start:]))) rows: list[dict] = [] for r in reader: ticker = (r.get("Ticker") or "").strip() if r.get("Asset Class") != "Equity" or ticker in ("", "-"): continue if _NON_COMPANY.search(r.get("Name") or ""): continue rows.append({ "ticker_raw": ticker, "ticker": normalize_ticker(ticker), "company_name": (r.get("Name") or "").strip(), "sector": SECTOR_MAP.get((r.get("Sector") or "").strip(), (r.get("Sector") or "").strip()), "location": (r.get("Location") or "").strip(), "exchange": (r.get("Exchange") or "").strip(), "weight_pct": float((r.get("Weight (%)") or "0").replace(",", "") or 0), "market_value_usd": float((r.get("Market Value") or "0").replace(",", "") or 0), }) return as_of, rows def get_constituents(self, cik_mapper: CIKMapper | None = None) -> list[dict]: text = self.fetch_holdings_csv() as_of, rows = self.parse_holdings(text) mapper = cik_mapper or CIKMapper() if not mapper._ticker_to_cik: # noqa: SLF001 - force a fresh SEC map mapper.build_mapping() out: list[dict] = [] unmatched: list[str] = [] for r in rows: cik = mapper.get_cik(r["ticker"]) or mapper.get_cik(r["ticker_raw"]) if not cik: unmatched.append(f"{r['ticker_raw']} ({r['company_name']})") continue out.append({**r, "cik": cik}) logger.info( "Russell 3000 (IWV as of %s): %d equity lines, %d mapped to a CIK, %d unmatched", as_of, len(rows), len(out), len(unmatched), ) if unmatched: logger.info("Unmatched tickers: %s", "; ".join(unmatched)) self._save_cache(as_of, out, unmatched) return out def load_cached(self) -> list[dict]: if not self._cache_path.exists(): return self.get_constituents() with open(self._cache_path) as f: return json.load(f)["constituents"] def _save_cache(self, as_of: str, constituents: list[dict], unmatched: list[str]) -> None: self._cache_dir.mkdir(parents=True, exist_ok=True) payload = { "source": ISHARES_IWV_URL, "holdings_as_of": as_of, "fetched_at": datetime.now(timezone.utc).isoformat(), "count": len(constituents), "unmatched": unmatched, "constituents": constituents, } with open(self._cache_path, "w") as f: json.dump(payload, f, indent=2) logger.info("Cached %d Russell 3000 constituents to %s", len(constituents), self._cache_path) def build_universe( sp500: list[dict], russell: list[dict], out_path: Path | None = None, ) -> list[dict]: """Merge the two lists on CIK (10 digits). S&P 500 names/sectors win.""" by_cik: dict[str, dict] = {} for r in russell: cik = str(r["cik"]).zfill(10) entry = by_cik.setdefault(cik, { "cik": cik, "ticker": r["ticker"], "company_name": r["company_name"], "sector": r.get("sector", ""), "sub_industry": "", "date_added": "", "in_sp500": False, "in_russell3000": True, "russell_weight_pct": 0.0, "tickers": [], }) entry["russell_weight_pct"] += r.get("weight_pct", 0.0) if r["ticker"] not in entry["tickers"]: entry["tickers"].append(r["ticker"]) # prefer the most liquid share class as the primary ticker if r.get("weight_pct", 0.0) > max( (x.get("weight_pct", 0.0) for x in russell if x["ticker"] == entry["ticker"]), default=0.0 ): entry["ticker"] = r["ticker"] for c in sp500: if not c.get("cik"): continue cik = str(c["cik"]).zfill(10) entry = by_cik.get(cik) if entry is None: entry = by_cik[cik] = { "cik": cik, "ticker": c["ticker"], "company_name": c["company_name"], "sector": c.get("sector", ""), "sub_industry": c.get("sub_industry", ""), "date_added": c.get("date_added", ""), "in_sp500": True, "in_russell3000": False, "russell_weight_pct": 0.0, "tickers": [c["ticker"]], } else: entry["in_sp500"] = True entry["ticker"] = c["ticker"] entry["company_name"] = c["company_name"] entry["sector"] = c.get("sector") or entry["sector"] entry["sub_industry"] = c.get("sub_industry", "") entry["date_added"] = c.get("date_added", "") if c["ticker"] not in entry["tickers"]: entry["tickers"].append(c["ticker"]) universe = sorted(by_cik.values(), key=lambda e: e["cik"]) out_path = out_path or DATA_DIRS["metadata"] / "universe.json" payload = { "built_at": datetime.now(timezone.utc).isoformat(), "count": len(universe), "n_sp500": sum(e["in_sp500"] for e in universe), "n_russell3000": sum(e["in_russell3000"] for e in universe), "constituents": universe, } with open(out_path, "w") as f: json.dump(payload, f, indent=2) logger.info( "Universe: %d companies (%d S&P 500, %d Russell 3000) -> %s", len(universe), payload["n_sp500"], payload["n_russell3000"], out_path, ) return universe def load_universe(path: Path | None = None) -> list[dict]: path = path or DATA_DIRS["metadata"] / "universe.json" with open(path) as f: return json.load(f)["constituents"]