spb/uqo-research-db

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# ============================================================1617from __future__ import annotations1819import json20import os21import time22from datetime import datetime, timezone23from pathlib import Path2425import httpx2627from src.utils.constants import DATA_DIRS, SEC_COMPANY_TICKERS_URL28from src.utils.logging import get_logger2930logger = get_logger("acquisition.cik_mapper")3132CIK_DIGITS = 10333435class CIKMapper:3637    def __init__(self, cache_dir: Path | None = None) -> None:38        self._cache_dir = cache_dir or DATA_DIRS["metadata"]39        self._cache_path = self._cache_dir / "cik_mapping.json"40        self._user_agent = os.environ.get(41            "SEC_USER_AGENT", "Research Agent research@example.com"42        )43        self._ticker_to_cik: dict[str, str] = {}44        self._cik_to_ticker: dict[str, str] = {}45        self._last_request_time: float = 0.04647    def build_mapping(self) -> dict[str, str]:48        self._rate_limit()49        headers = {"User-Agent": self._user_agent, "Accept": "application/json"}5051        with httpx.Client(timeout=30, headers=headers) as client:52            response = client.get(SEC_COMPANY_TICKERS_URL)53            response.raise_for_status()54            data = response.json()5556        ticker_to_cik: dict[str, str] = {}57        cik_to_ticker: dict[str, str] = {}5859        for entry in data.values():60            ticker = str(entry.get("ticker", "")).upper().strip()61            raw_cik = entry.get("cik_str", "")62            cik = str(raw_cik).zfill(CIK_DIGITS)6364            if ticker and cik:65                ticker_to_cik[ticker] = cik66                cik_to_ticker[cik] = ticker6768        self._ticker_to_cik = ticker_to_cik69        self._cik_to_ticker = cik_to_ticker7071        logger.info("Built CIK mapping: %d tickers", len(ticker_to_cik))72        self._save_cache()73        return dict(ticker_to_cik)7475    def get_cik(self, ticker: str) -> str | None:76        if not self._ticker_to_cik:77            self._load_or_build()78        return self._ticker_to_cik.get(ticker.upper().strip())7980    def get_ticker(self, cik: str) -> str | None:81        if not self._cik_to_ticker:82            self._load_or_build()83        padded = cik.zfill(CIK_DIGITS)84        return self._cik_to_ticker.get(padded)8586    def _load_or_build(self) -> None:87        if self._cache_path.exists():88            self._load_cache()89        else:90            self.build_mapping()9192    def _save_cache(self) -> None:93        self._cache_dir.mkdir(parents=True, exist_ok=True)94        payload = {95            "fetched_at": datetime.now(timezone.utc).isoformat(),96            "count": len(self._ticker_to_cik),97            "ticker_to_cik": self._ticker_to_cik,98            "cik_to_ticker": self._cik_to_ticker,99        }100        with open(self._cache_path, "w") as f:101            json.dump(payload, f, indent=2)102        logger.info("Saved CIK mapping cache to %s", self._cache_path)103104    def _load_cache(self) -> None:105        with open(self._cache_path) as f:106            data = json.load(f)107        self._ticker_to_cik = data.get("ticker_to_cik", {})108        self._cik_to_ticker = data.get("cik_to_ticker", {})109        logger.info(110            "Loaded CIK mapping cache: %d tickers (fetched %s)",111            len(self._ticker_to_cik),112            data.get("fetched_at", "unknown"),113        )114115    def _rate_limit(self) -> None:116        elapsed = time.monotonic() - self._last_request_time117        min_interval = 0.1118        if elapsed < min_interval:119            time.sleep(min_interval - elapsed)120        self._last_request_time = time.monotonic()121