# ============================================================ # 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 # ============================================================ from __future__ import annotations import json import os import time from datetime import datetime, timezone from pathlib import Path import httpx from tqdm import tqdm from src.acquisition.sec_client import SECClient from src.utils.constants import DATA_DIRS from src.utils.logging import get_logger logger = get_logger("acquisition.downloader") MIN_REQUEST_INTERVAL = 0.1 RETRY_BACKOFF_BASE = 2.0 MAX_RETRIES = 5 class FilingDownloader: def __init__(self, output_dir: Path | None = None) -> None: self._output_dir = output_dir or DATA_DIRS["raw"] self._metadata_dir = DATA_DIRS["metadata"] self._download_log_path = self._metadata_dir / "download_log.json" self._user_agent = os.environ.get( "SEC_USER_AGENT", "Research Agent research@example.com" ) self._last_request_time: float = 0.0 self._downloaded: set[str] = set() self._client = httpx.Client( timeout=30, headers={"User-Agent": self._user_agent}, follow_redirects=True, ) self._load_download_log() def close(self) -> None: self._client.close() def __enter__(self): return self def __exit__(self, *args): self.close() def download_filing( self, cik: str, accession: str, primary_doc: str, form_type: str = "10-K", filing_url: str | None = None, ) -> Path: padded_cik = cik.zfill(10) accession_clean = accession.replace("-", "") filing_date_year = "" if filing_url: url = filing_url else: raw_cik = cik.lstrip("0") or "0" url = ( f"https://www.sec.gov/Archives/edgar/data/" f"{raw_cik}/{accession_clean}/{primary_doc}" ) save_dir = ( self._output_dir / padded_cik / form_type / accession_clean ) save_dir.mkdir(parents=True, exist_ok=True) save_path = save_dir / primary_doc if save_path.exists(): logger.debug("File already exists: %s", save_path) return save_path content = self._download_with_retry(url) if content is None: raise RuntimeError(f"Failed to download {url} after {MAX_RETRIES} retries") save_path.write_bytes(content) logger.debug("Downloaded %s -> %s", url, save_path) return save_path def download_company_filings( self, cik: str, ticker: str, form_type: str = "10-K", start_date: str = "2010-01-01", sec_client: SECClient | None = None, ) -> list[dict]: own_client = sec_client is None client = sec_client or SECClient() try: filings = client.get_company_filings( cik=cik, form_type=form_type, start_date=start_date ) finally: if own_client: client.close() results: list[dict] = [] for filing in filings: accession = filing["accession_number"] download_key = f"{cik}|{accession}|{filing['primary_document']}" if download_key in self._downloaded: logger.debug("Skipping already-downloaded: %s", download_key) filing["local_path"] = str( self._output_dir / cik.zfill(10) / form_type / accession.replace("-", "") / filing["primary_document"] ) filing["ticker"] = ticker results.append(filing) continue try: local_path = self.download_filing( cik=cik, accession=accession, primary_doc=filing["primary_document"], form_type=form_type, filing_url=filing.get("filing_url"), ) filing["local_path"] = str(local_path) filing["ticker"] = ticker results.append(filing) self._record_download(download_key, filing) except Exception: logger.error( "Failed to download filing %s for %s (%s)", accession, ticker, cik, exc_info=True, ) logger.info( "Downloaded %d/%d %s filings for %s (CIK %s)", len(results), len(filings), form_type, ticker, cik, ) return results def download_universe( self, tickers: list[str], form_types: list[str], start_date: str = "2010-01-01", resume: bool = True, cik_mapper=None, ) -> None: if cik_mapper is None: from src.acquisition.cik_mapper import CIKMapper cik_mapper = CIKMapper() if not resume: self._downloaded.clear() total_tasks = len(tickers) * len(form_types) progress = tqdm(total=total_tasks, desc="Downloading universe", unit="task") completed = 0 failed = 0 sec_client = SECClient() try: for ticker in tickers: cik = cik_mapper.get_cik(ticker) if not cik: logger.warning("No CIK found for ticker %s, skipping", ticker) progress.update(len(form_types)) failed += len(form_types) continue for form_type in form_types: progress.set_postfix_str(f"{ticker} {form_type}") try: self.download_company_filings( cik=cik, ticker=ticker, form_type=form_type, start_date=start_date, sec_client=sec_client, ) completed += 1 except Exception: logger.error( "Failed universe download for %s %s", ticker, form_type, exc_info=True, ) failed += 1 progress.update(1) finally: sec_client.close() progress.close() logger.info( "Universe download complete: %d succeeded, %d failed out of %d tasks", completed, failed, total_tasks, ) def _download_with_retry(self, url: str) -> bytes | None: for attempt in range(MAX_RETRIES): self._rate_limit() try: response = self._client.get(url) if response.status_code == 404: logger.warning("Filing not found (404): %s", url) return None response.raise_for_status() return response.content except httpx.HTTPStatusError as exc: wait = RETRY_BACKOFF_BASE ** attempt logger.warning( "HTTP %d downloading %s (attempt %d/%d), retrying in %.1fs", exc.response.status_code, url, attempt + 1, MAX_RETRIES, wait, ) time.sleep(wait) except httpx.RequestError as exc: wait = RETRY_BACKOFF_BASE ** attempt logger.warning( "Request error downloading %s: %s (attempt %d/%d), retrying in %.1fs", url, exc, attempt + 1, MAX_RETRIES, wait, ) time.sleep(wait) logger.error("All %d retries exhausted for %s", MAX_RETRIES, url) return None def _rate_limit(self) -> None: elapsed = time.monotonic() - self._last_request_time if elapsed < MIN_REQUEST_INTERVAL: time.sleep(MIN_REQUEST_INTERVAL - elapsed) self._last_request_time = time.monotonic() def _load_download_log(self) -> None: if not self._download_log_path.exists(): return try: with open(self._download_log_path) as f: for line in f: line = line.strip() if not line: continue entry = json.loads(line) key = entry.get("key", "") if key: self._downloaded.add(key) logger.info("Loaded %d entries from download log", len(self._downloaded)) except Exception: logger.warning("Could not load download log, starting fresh", exc_info=True) def _record_download(self, key: str, filing: dict) -> None: self._downloaded.add(key) self._metadata_dir.mkdir(parents=True, exist_ok=True) entry = { "key": key, "timestamp": datetime.now(timezone.utc).isoformat(), "ticker": filing.get("ticker", ""), "form_type": filing.get("form_type", ""), "accession_number": filing.get("accession_number", ""), "filing_date": filing.get("filing_date", ""), "local_path": filing.get("local_path", ""), } with open(self._download_log_path, "a") as f: f.write(json.dumps(entry) + "\n")