spb/uqo-research-db
Public
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# ============================================================1617"""Concurrent, rate-limited, resumable EDGAR downloader (2026-10-27).1819Same on-disk layout and download-log format as ``FilingDownloader`` so that20``scripts/build_database.py`` works unchanged, but:2122* one global token bucket shared by all threads (SEC allows 10 req/s per IP;23 several cluster nodes behind the same IP each run with ``rate = 9 / n``);24* a thread pool keeps the pipe full (latency no longer caps throughput);25* HTTP 403/429 (rate-limit ban) triggers a cool-down instead of a hard failure;26* the trimmed company header of every submissions JSON (SIC, state, exchanges)27 is saved to ``metadata/submissions/<cik>.json`` for the companies table.28"""2930from __future__ import annotations3132import json33import os34import threading35import time36from concurrent.futures import ThreadPoolExecutor, as_completed37from datetime import datetime, timezone38from pathlib import Path3940import httpx4142from src.utils.constants import DATA_DIRS, SEC_ARCHIVES_URL, SEC_SUBMISSIONS_URL43from src.utils.logging import get_logger4445logger = get_logger("acquisition.bulk")4647MAX_RETRIES = 648COOLDOWN_SECONDS = 65.0495051class TokenBucket:52 def __init__(self, rate: float, capacity: float | None = None) -> None:53 self.rate = max(rate, 0.05)54 self.capacity = capacity or max(1.0, self.rate)55 self._tokens = self.capacity56 self._last = time.monotonic()57 self._lock = threading.Lock()5859 def acquire(self) -> None:60 while True:61 with self._lock:62 now = time.monotonic()63 self._tokens = min(self.capacity, self._tokens + (now - self._last) * self.rate)64 self._last = now65 if self._tokens >= 1.0:66 self._tokens -= 1.067 return68 wait = (1.0 - self._tokens) / self.rate69 time.sleep(wait)707172class BulkDownloader:73 def __init__(74 self,75 raw_dir: Path | None = None,76 log_path: Path | None = None,77 submissions_dir: Path | None = None,78 rate: float = 9.0,79 workers: int = 8,80 user_agent: str | None = None,81 trusted_logs: list[Path] | None = None,82 ) -> None:83 self.raw_dir = Path(raw_dir) if raw_dir else DATA_DIRS["raw"]84 self.log_path = Path(log_path) if log_path else DATA_DIRS["metadata"] / "download_log.json"85 self.submissions_dir = Path(submissions_dir) if submissions_dir else DATA_DIRS["metadata"] / "submissions"86 self.bucket = TokenBucket(rate)87 self.workers = workers88 self.user_agent = user_agent or os.environ.get("SEC_USER_AGENT", "UQO-Research research@uqo.ca")89 self._client = httpx.Client(90 timeout=60,91 headers={"User-Agent": self.user_agent, "Accept-Encoding": "gzip, deflate"},92 follow_redirects=True,93 limits=httpx.Limits(max_connections=workers + 4, max_keepalive_connections=workers + 4),94 )95 self._log_lock = threading.Lock()96 self._cooldown_until = 0.097 self._cooldown_lock = threading.Lock()98 self._done_keys: set[str] = set()99 self._trusted_keys: set[str] = set()100 self._load_log()101 for tl in trusted_logs or []:102 self._load_trusted(Path(tl))103 self.stats = {"ciks": 0, "filings_listed": 0, "downloaded": 0, "skipped": 0, "failed": 0, "bytes": 0}104 self._stats_lock = threading.Lock()105106 # ---------------------------------------------------------------- HTTP107 def _get(self, url: str, expect_json: bool = False):108 for attempt in range(MAX_RETRIES):109 with self._cooldown_lock:110 until = self._cooldown_until111 now = time.monotonic()112 if until > now:113 time.sleep(until - now)114 self.bucket.acquire()115 try:116 r = self._client.get(url)117 except httpx.RequestError as exc:118 wait = min(60.0, 2.0 ** attempt)119 logger.warning("request error %s (%s) attempt %d, sleeping %.0fs", url, exc, attempt + 1, wait)120 time.sleep(wait)121 continue122 if r.status_code == 404:123 return None124 if r.status_code in (403, 429, 503):125 with self._cooldown_lock:126 self._cooldown_until = time.monotonic() + COOLDOWN_SECONDS127 logger.warning("HTTP %d on %s -> cooling down %.0fs (attempt %d)",128 r.status_code, url, COOLDOWN_SECONDS, attempt + 1)129 continue130 if r.status_code >= 400:131 wait = min(60.0, 2.0 ** attempt)132 logger.warning("HTTP %d on %s attempt %d, sleeping %.0fs", r.status_code, url, attempt + 1, wait)133 time.sleep(wait)134 continue135 return r.json() if expect_json else r.content136 logger.error("giving up on %s", url)137 return None138139 # ------------------------------------------------------------- listing140 def list_filings(self, cik: str, forms: list[str], start_date: str) -> tuple[dict | None, list[dict]]:141 padded = cik.zfill(10)142 data = self._get(f"{SEC_SUBMISSIONS_URL}/CIK{padded}.json", expect_json=True)143 if data is None:144 return None, []145 header = {146 "cik": padded,147 "name": data.get("name"),148 "sic": data.get("sic"),149 "sic_description": data.get("sicDescription"),150 "state_of_incorporation": data.get("stateOfIncorporation"),151 "state_of_business": (data.get("addresses", {}).get("business", {}) or {}).get("stateOrCountry"),152 "fiscal_year_end": data.get("fiscalYearEnd"),153 "exchanges": data.get("exchanges"),154 "tickers": data.get("tickers"),155 "entity_type": data.get("entityType"),156 "former_names": [n.get("name") for n in data.get("formerNames", [])],157 "fetched_at": datetime.now(timezone.utc).isoformat(),158 }159 blocks = [data.get("filings", {}).get("recent", {})]160 for f in data.get("filings", {}).get("files", []):161 name = f.get("name")162 if not name:163 continue164 # older blocks are needed only if they can contain filings after start_date165 if f.get("filingTo", "9999") < start_date:166 continue167 older = self._get(f"{SEC_SUBMISSIONS_URL}/{name}", expect_json=True)168 if older:169 blocks.append(older)170 filings: list[dict] = []171 wanted = set(forms)172 for b in blocks:173 fm, acc, fd, rd, pd_ = (b.get("form", []), b.get("accessionNumber", []), b.get("filingDate", []),174 b.get("reportDate", []), b.get("primaryDocument", []))175 for i, form in enumerate(fm):176 if form not in wanted:177 continue178 if i >= len(fd) or fd[i] < start_date:179 continue180 primary = pd_[i] if i < len(pd_) else ""181 if not primary:182 continue183 accession = acc[i]184 filings.append({185 "accession_number": accession,186 "form_type": form,187 "filing_date": fd[i],188 "report_date": rd[i] if i < len(rd) else "",189 "primary_document": primary,190 "filing_url": f"{SEC_ARCHIVES_URL}/{padded.lstrip('0')}/{accession.replace('-', '')}/{primary}",191 })192 return header, filings193194 # ------------------------------------------------------------ download195 def _local_path(self, cik: str, filing: dict) -> Path:196 return (self.raw_dir / cik.zfill(10) / filing["form_type"]197 / filing["accession_number"].replace("-", "") / filing["primary_document"])198199 def download_cik(self, cik: str, ticker: str, forms: list[str], start_date: str) -> dict:200 cik = cik.zfill(10)201 header, filings = self.list_filings(cik, forms, start_date)202 res = {"cik": cik, "listed": len(filings), "downloaded": 0, "skipped": 0, "failed": 0}203 if header is None:204 logger.warning("no submissions for CIK %s (%s)", cik, ticker)205 res["failed"] = 1206 return res207 self.submissions_dir.mkdir(parents=True, exist_ok=True)208 (self.submissions_dir / f"{cik}.json").write_text(json.dumps(header, indent=1))209 if not ticker and header.get("tickers"):210 ticker = header["tickers"][0]211212 for f in filings:213 key = f"{cik}|{f['accession_number']}|{f['primary_document']}"214 path = self._local_path(cik, f)215 if key in self._trusted_keys:216 res["skipped"] += 1217 continue218 if key in self._done_keys and path.exists():219 res["skipped"] += 1220 continue221 if path.exists() and path.stat().st_size > 0:222 self._record(key, cik, ticker, f, path, path.stat().st_size)223 res["skipped"] += 1224 continue225 content = self._get(f["filing_url"])226 if content is None:227 res["failed"] += 1228 continue229 path.parent.mkdir(parents=True, exist_ok=True)230 tmp = path.with_suffix(path.suffix + ".part")231 tmp.write_bytes(content)232 os.replace(tmp, path)233 self._record(key, cik, ticker, f, path, len(content))234 res["downloaded"] += 1235 with self._stats_lock:236 self.stats["bytes"] += len(content)237 return res238239 def download_many(self, companies: list[dict], forms: list[str], start_date: str) -> dict:240 t0 = time.time()241 total = len(companies)242 logger.info("bulk download: %d companies, forms=%s, since %s, rate=%.2f/s, workers=%d, raw=%s",243 total, forms, start_date, self.bucket.rate, self.workers, self.raw_dir)244 done = 0245 with ThreadPoolExecutor(max_workers=self.workers) as ex:246 futs = {ex.submit(self.download_cik, c["cik"], c.get("ticker", ""), forms, start_date): c247 for c in companies}248 for fut in as_completed(futs):249 c = futs[fut]250 done += 1251 try:252 r = fut.result()253 except Exception: # noqa: BLE001254 logger.exception("CIK %s failed", c["cik"])255 with self._stats_lock:256 self.stats["failed"] += 1257 continue258 with self._stats_lock:259 self.stats["ciks"] += 1260 self.stats["filings_listed"] += r["listed"]261 self.stats["downloaded"] += r["downloaded"]262 self.stats["skipped"] += r["skipped"]263 self.stats["failed"] += r["failed"]264 s = dict(self.stats)265 el = time.time() - t0266 logger.info("[%d/%d] %s %s listed=%d dl=%d skip=%d fail=%d | total dl=%d skip=%d fail=%d %.1f GB %.0fs",267 done, total, c["cik"], c.get("ticker", ""), r["listed"], r["downloaded"],268 r["skipped"], r["failed"], s["downloaded"], s["skipped"], s["failed"],269 s["bytes"] / 1e9, el)270 self._client.close()271 logger.info("bulk download complete: %s in %.0fs", self.stats, time.time() - t0)272 return dict(self.stats)273274 # ------------------------------------------------------------------ log275 def _load_log(self) -> None:276 if not self.log_path.exists():277 return278 with open(self.log_path) as f:279 for line in f:280 line = line.strip()281 if not line:282 continue283 try:284 key = json.loads(line).get("key", "")285 except json.JSONDecodeError:286 continue287 if key:288 self._done_keys.add(key)289 logger.info("download log: %d entries already done (%s)", len(self._done_keys), self.log_path)290291 def _load_trusted(self, path: Path) -> None:292 """Keys from an earlier corpus (e.g. the S&P 500 log) are skipped even if the file is not here yet."""293 if not path.exists():294 logger.warning("trusted log %s not found", path)295 return296 n = 0297 with open(path) as f:298 for line in f:299 line = line.strip()300 if not line:301 continue302 try:303 key = json.loads(line).get("key", "")304 except json.JSONDecodeError:305 continue306 if key:307 self._trusted_keys.add(key)308 n += 1309 logger.info("trusted log: %d keys from %s", n, path)310311 def _record(self, key: str, cik: str, ticker: str, filing: dict, path: Path, size: int) -> None:312 entry = {313 "key": key,314 "timestamp": datetime.now(timezone.utc).isoformat(),315 "ticker": ticker,316 "form_type": filing["form_type"],317 "accession_number": filing["accession_number"],318 "filing_date": filing["filing_date"],319 "report_date": filing.get("report_date", ""),320 "primary_document": filing["primary_document"],321 "filing_url": filing["filing_url"],322 "file_size_bytes": size,323 "local_path": str(path),324 }325 with self._log_lock:326 self._done_keys.add(key)327 self.log_path.parent.mkdir(parents=True, exist_ok=True)328 with open(self.log_path, "a") as f:329 f.write(json.dumps(entry) + "\n")330