"""Build a local ticker → GICS sector map for research residualization. Sources (in order): 1. Public S&P 500 constituents CSV (datasets/s-and-p-500-companies) — bulk, free. 2. Existing map file (resume). 3. FMP stable ``profile`` for still-missing symbols (budget ~250 req/day). Writes ``data/research/ticker_sector_map.json``. Never touches production Postgres. Example ------- python scripts/build_ticker_sector_map.py \\ --snapshot backtest_snapshots/prod.sqlite python scripts/build_ticker_sector_map.py --fmp-limit 50 """ from __future__ import annotations import argparse import asyncio import csv import io import json import sys import time from datetime import datetime, timezone from pathlib import Path import httpx from sqlalchemy import create_engine, text ROOT = Path(__file__).resolve().parents[1] if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) from app.services.sector_map import ( # noqa: E402 DEFAULT_SECTOR_MAP_PATH, coverage_stats, load_ticker_sector_map, normalise_symbol, save_ticker_sector_map, sector_to_etf, ) SP500_CSV_URL = ( "https://raw.githubusercontent.com/datasets/s-and-p-500-companies/" "master/data/constituents.csv" ) FMP_STABLE = "https://financialmodelingprep.com/stable" def _parse_args() -> argparse.Namespace: p = argparse.ArgumentParser(description=__doc__) p.add_argument( "--snapshot", default="backtest_snapshots/prod.sqlite", help="Snapshot whose tickers define the universe.", ) p.add_argument( "--out", default=str(DEFAULT_SECTOR_MAP_PATH), help="Output JSON path.", ) p.add_argument( "--fmp-limit", type=int, default=200, help="Max FMP profile requests this run (free-tier cushion).", ) p.add_argument( "--skip-fmp", action="store_true", help="Only use public SP500 CSV + existing map.", ) p.add_argument("--sleep", type=float, default=0.35, help="Pause between FMP calls.") return p.parse_args() def _snapshot_symbols(snapshot: Path) -> list[str]: engine = create_engine(f"sqlite:///{snapshot.resolve().as_posix()}", future=True) try: with engine.connect() as conn: rows = conn.execute(text("SELECT symbol FROM tickers ORDER BY symbol")).fetchall() finally: engine.dispose() return [normalise_symbol(r[0]) for r in rows if r[0]] def _fetch_sp500_map() -> dict[str, str]: with httpx.Client(timeout=60.0, follow_redirects=True) as client: resp = client.get(SP500_CSV_URL) resp.raise_for_status() reader = csv.DictReader(io.StringIO(resp.text)) out: dict[str, str] = {} for row in reader: sym = normalise_symbol(row.get("Symbol") or "") sector = (row.get("GICS Sector") or "").strip() if sym and sector: out[sym] = sector return out async def _fmp_profile_sector(client: httpx.AsyncClient, api_key: str, symbol: str) -> str | None: resp = await client.get( f"{FMP_STABLE}/profile", params={"symbol": symbol, "apikey": api_key}, ) if resp.status_code == 429: raise RuntimeError(f"FMP rate limited on {symbol}") if resp.status_code == 402: return None resp.raise_for_status() data = resp.json() if isinstance(data, list): data = data[0] if data else {} if not isinstance(data, dict): return None sector = (data.get("sector") or data.get("industry") or "").strip() # industry alone is not a GICS sector — only accept if we can map to an ETF if sector and sector_to_etf(sector): return sector # FMP sometimes returns industry under sector when sector missing; try sector field only sec = (data.get("sector") or "").strip() return sec or None async def _fill_from_fmp( missing: list[str], *, api_key: str, limit: int, sleep_s: float, ) -> tuple[dict[str, str], int]: filled: dict[str, str] = {} used = 0 async with httpx.AsyncClient(timeout=30.0) as client: for sym in missing: if used >= limit: break try: sector = await _fmp_profile_sector(client, api_key, sym) except Exception as exc: print(f" FMP fail {sym}: {exc}") used += 1 await asyncio.sleep(sleep_s) continue used += 1 if sector: filled[sym] = sector print(f" FMP {sym} → {sector}") else: print(f" FMP {sym} → (no sector)") if sleep_s > 0: await asyncio.sleep(sleep_s) return filled, used async def _main() -> None: args = _parse_args() snapshot = Path(args.snapshot) if not snapshot.exists(): raise SystemExit(f"Snapshot not found: {snapshot}") symbols = _snapshot_symbols(snapshot) print(f"Universe: {len(symbols)} symbols from {snapshot}") existing = load_ticker_sector_map(args.out) print(f"Existing map entries: {len(existing)}") print("Fetching public S&P 500 sector CSV…") sp500 = _fetch_sp500_map() print(f" SP500 CSV rows: {len(sp500)}") mapping = dict(existing) from_sp500 = 0 for sym in symbols: if sym in mapping: continue if sym in sp500: mapping[sym] = sp500[sym] from_sp500 += 1 print(f" Newly filled from SP500 CSV: {from_sp500}") missing = [s for s in symbols if s not in mapping] fmp_used = 0 from_fmp = 0 if missing and not args.skip_fmp: from app.config import settings if not settings.fmp_api_key: print("WARNING: FMP key missing; leaving gaps unfilled") else: print(f"FMP fill for {len(missing)} missing (limit={args.fmp_limit})…") filled, fmp_used = await _fill_from_fmp( missing, api_key=settings.fmp_api_key, limit=int(args.fmp_limit), sleep_s=float(args.sleep), ) mapping.update(filled) from_fmp = len(filled) still_missing = [s for s in symbols if s not in mapping] stats = coverage_stats(symbols, mapping) meta = { "built_at": datetime.now(timezone.utc).isoformat(), "snapshot": str(snapshot.resolve()), "from_existing": len(existing), "from_sp500_csv": from_sp500, "from_fmp": from_fmp, "fmp_requests": fmp_used, "still_missing": still_missing, "coverage": { k: stats[k] for k in ("universe", "mapped", "mapped_pct", "with_etf", "by_sector") }, } out_path = save_ticker_sector_map(mapping, args.out, meta=meta) print(f"Wrote {out_path}") print(json.dumps(meta["coverage"], indent=2)) if still_missing: print(f"Still missing ({len(still_missing)}): {still_missing[:40]}") if len(still_missing) > 40: print(f" … +{len(still_missing) - 40} more") if __name__ == "__main__": asyncio.run(_main())