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 os20import time2122import httpx2324from src.utils.constants import SEC_ARCHIVES_URL, SEC_SUBMISSIONS_URL25from src.utils.logging import get_logger2627logger = get_logger("acquisition.sec_client")2829MAX_RETRIES = 530RETRY_BACKOFF_BASE = 2.031MIN_REQUEST_INTERVAL = 0.1323334class SECClient:3536    def __init__(self) -> None:37        self._user_agent = os.environ.get(38            "SEC_USER_AGENT", "Research Agent research@example.com"39        )40        self._last_request_time: float = 0.041        self._client = httpx.Client(42            timeout=30,43            headers={44                "User-Agent": self._user_agent,45                "Accept": "application/json",46            },47        )4849    def close(self) -> None:50        self._client.close()5152    def __enter__(self) -> SECClient:53        return self5455    def __exit__(self, *exc) -> None:56        self.close()5758    def get_company_filings(59        self,60        cik: str,61        form_type: str = "10-K",62        start_date: str = "2010-01-01",63    ) -> list[dict]:64        padded_cik = cik.zfill(10)65        url = f"{SEC_SUBMISSIONS_URL}/CIK{padded_cik}.json"6667        data = self._request_with_retry(url)68        if data is None:69            return []7071        filings = self._extract_filings(data, padded_cik, form_type, start_date)7273        older_files = data.get("filings", {}).get("files", [])74        for file_entry in older_files:75            file_name = file_entry.get("name", "")76            if not file_name:77                continue78            older_url = f"{SEC_SUBMISSIONS_URL}/{file_name}"79            older_data = self._request_with_retry(older_url)80            if older_data is not None:81                filings.extend(82                    self._extract_filings_from_block(83                        older_data, padded_cik, form_type, start_date84                    )85                )8687        logger.info(88            "Found %d %s filings for CIK %s since %s",89            len(filings), form_type, padded_cik, start_date,90        )91        return filings9293    def _extract_filings(94        self,95        data: dict,96        cik: str,97        form_type: str,98        start_date: str,99    ) -> list[dict]:100        recent = data.get("filings", {}).get("recent", {})101        return self._extract_filings_from_block(recent, cik, form_type, start_date)102103    def _extract_filings_from_block(104        self,105        block: dict,106        cik: str,107        form_type: str,108        start_date: str,109    ) -> list[dict]:110        forms = block.get("form", [])111        accessions = block.get("accessionNumber", [])112        filing_dates = block.get("filingDate", [])113        report_dates = block.get("reportDate", [])114        primary_docs = block.get("primaryDocument", [])115116        results: list[dict] = []117        for i in range(len(forms)):118            if forms[i] != form_type:119                continue120            filing_date = filing_dates[i] if i < len(filing_dates) else ""121            if filing_date < start_date:122                continue123124            accession = accessions[i] if i < len(accessions) else ""125            accession_clean = accession.replace("-", "")126            primary_doc = primary_docs[i] if i < len(primary_docs) else ""127            report_date = report_dates[i] if i < len(report_dates) else ""128129            filing_url = f"{SEC_ARCHIVES_URL}/{cik.lstrip('0')}/{accession_clean}/{primary_doc}"130131            results.append({132                "accession_number": accession,133                "form_type": forms[i],134                "filing_date": filing_date,135                "report_date": report_date,136                "primary_document": primary_doc,137                "filing_url": filing_url,138            })139140        return results141142    def _request_with_retry(self, url: str) -> dict | None:143        for attempt in range(MAX_RETRIES):144            self._rate_limit()145            try:146                response = self._client.get(url)147                if response.status_code == 404:148                    logger.warning("Resource not found: %s", url)149                    return None150                response.raise_for_status()151                return response.json()152            except httpx.HTTPStatusError as exc:153                wait = RETRY_BACKOFF_BASE ** attempt154                logger.warning(155                    "HTTP %d on %s (attempt %d/%d), retrying in %.1fs",156                    exc.response.status_code, url, attempt + 1, MAX_RETRIES, wait,157                )158                time.sleep(wait)159            except httpx.RequestError as exc:160                wait = RETRY_BACKOFF_BASE ** attempt161                logger.warning(162                    "Request error on %s: %s (attempt %d/%d), retrying in %.1fs",163                    url, exc, attempt + 1, MAX_RETRIES, wait,164                )165                time.sleep(wait)166167        logger.error("All %d retries exhausted for %s", MAX_RETRIES, url)168        return None169170    def _rate_limit(self) -> None:171        elapsed = time.monotonic() - self._last_request_time172        if elapsed < MIN_REQUEST_INTERVAL:173            time.sleep(MIN_REQUEST_INTERVAL - elapsed)174        self._last_request_time = time.monotonic()175