Metafazer/finrag-backend
0
1"""Async client for SEC EDGAR API.2 3Handles ticker-to-CIK resolution, filing discovery, download,4and basic section parsing. Respects SEC rate limits (10 req/s)5and requires a valid User-Agent header.6 7Design decisions:8- httpx for async HTTP (I/O-bound calls benefit from async)9- asyncio.Semaphore for client-side rate limiting10- Structured error types for each failure mode11- BeautifulSoup for HTML parsing (regex alone is too fragile)12"""13 14import asyncio15import json16import re17import warnings18from dataclasses import dataclass, field19from datetime import datetime20from pathlib import Path21 22import httpx23import structlog24from bs4 import BeautifulSoup, XMLParsedAsHTMLWarning25 26from finrag.config import Settings27 28# Suppress BS4 warning when lxml encounters XML-structured EDGAR filings.29# The parser works correctly regardless — this is a false-positive warning.30warnings.filterwarnings("ignore", category=XMLParsedAsHTMLWarning)31 32logger = structlog.get_logger(__name__)33 34# --------------------------------------------------------------------------- #35# Custom exceptions: each failure mode gets its own type36# --------------------------------------------------------------------------- #37 38 39class EdgarError(Exception):40 """Base exception for EDGAR client errors."""41 42 43class TickerNotFoundError(EdgarError):44 """Raised when a ticker cannot be resolved to a CIK."""45 46 47class FilingNotFoundError(EdgarError):48 """Raised when no filings are found for a CIK + filing type."""49 50 51class EdgarUnavailableError(EdgarError):52 """Raised when EDGAR API is unreachable after retries."""53 54 55class EdgarRateLimitError(EdgarError):56 """Raised when EDGAR returns a 429 rate limit response."""57 58 59# --------------------------------------------------------------------------- #60# Data models61# --------------------------------------------------------------------------- #62 63 64@dataclass(frozen=True)65class FilingMetadata:66 """Metadata for a single SEC filing.67 68 Attributes:69 cik: SEC Central Index Key.70 ticker: Stock ticker symbol.71 company_name: Full company name from EDGAR.72 filing_type: Filing form type (10-K, 10-Q, 8-K).73 filing_date: Date the filing was submitted.74 accession_number: Unique filing identifier.75 primary_document_url: URL to the main filing document.76 """77 78 cik: str79 ticker: str80 company_name: str81 filing_type: str82 filing_date: str83 accession_number: str84 primary_document_url: str85 86 87@dataclass88class ParsedFiling:89 """A downloaded and parsed SEC filing.90 91 Attributes:92 metadata: Filing metadata from EDGAR.93 sections: Dict mapping section name to section text content.94 raw_content_length: Length of the raw HTML content in characters.95 """96 97 metadata: FilingMetadata98 sections: dict[str, str] = field(default_factory=dict)99 raw_content_length: int = 0100 101 102# --------------------------------------------------------------------------- #103# 10-K section patterns104# --------------------------------------------------------------------------- #105 106# Standard 10-K items. These appear as "Item 1", "Item 1A", etc.107# We look for these in headings and bold text within the filing HTML.108SECTION_10K_ITEMS: dict[str, str] = {109 "1": "Business",110 "1A": "Risk Factors",111 "1B": "Unresolved Staff Comments",112 "2": "Properties",113 "3": "Legal Proceedings",114 "4": "Mine Safety Disclosures",115 "5": "Market for Common Equity",116 "6": "Reserved",117 "7": "MD&A",118 "7A": "Quantitative and Qualitative Disclosures About Market Risk",119 "8": "Financial Statements",120 "9": "Changes in and Disagreements with Accountants",121 "9A": "Controls and Procedures",122 "9B": "Other Information",123 "10": "Directors and Corporate Governance",124 "11": "Executive Compensation",125 "12": "Security Ownership",126 "13": "Certain Relationships",127 "14": "Principal Accountant Fees",128 "15": "Exhibits and Financial Statement Schedules",129}130 131 132# --------------------------------------------------------------------------- #133# EDGAR Client134# --------------------------------------------------------------------------- #135 136 137class EdgarClient:138 """Async client for SEC EDGAR API.139 140 Handles all interactions with the SEC EDGAR system including141 ticker resolution, filing discovery, and content download.142 143 Args:144 settings: Application settings with EDGAR configuration.145 """146 147 # SEC company tickers endpoint (returns all ticker-to-CIK mappings)148 TICKERS_URL = "https://www.sec.gov/files/company_tickers.json"149 150 # EDGAR filing submissions endpoint151 SUBMISSIONS_URL = "https://data.sec.gov/submissions/CIK{cik}.json"152 153 # Max retries for transient failures154 MAX_RETRIES = 3155 156 # Backoff base in seconds157 BACKOFF_BASE = 1.0158 159 def __init__(self, settings: Settings) -> None:160 """Initialize the EDGAR client.161 162 Args:163 settings: Application settings containing EDGAR configuration.164 """165 self._settings = settings166 self._headers = {167 "User-Agent": settings.edgar_user_agent,168 "Accept-Encoding": "gzip, deflate",169 }170 # Client-side rate limiting to respect SEC's 10 req/s limit171 self._semaphore = asyncio.Semaphore(settings.edgar_max_rps)172 self._client: httpx.AsyncClient | None = None173 174 async def __aenter__(self) -> "EdgarClient":175 """Enter async context manager."""176 self._client = httpx.AsyncClient(177 headers=self._headers,178 timeout=httpx.Timeout(30.0),179 follow_redirects=True,180 )181 return self182 183 async def __aexit__(self, *args: object) -> None:184 """Exit async context manager."""185 if self._client:186 await self._client.aclose()187 self._client = None188 189 async def _request(self, url: str) -> httpx.Response:190 """Make a rate-limited HTTP request with retry logic.191 192 Args:193 url: The URL to request.194 195 Returns:196 The HTTP response.197 198 Raises:199 EdgarUnavailableError: If all retries are exhausted.200 EdgarRateLimitError: If we hit a 429 response.201 """202 if not self._client:203 msg = "Client not initialized. Use 'async with EdgarClient(...)' context manager."204 raise EdgarError(msg)205 206 for attempt in range(self.MAX_RETRIES):207 async with self._semaphore:208 try:209 response = await self._client.get(url)210 211 if response.status_code == 429:212 logger.warning(213 "rate_limit_hit",214 url=url,215 attempt=attempt + 1,216 )217 if attempt < self.MAX_RETRIES - 1:218 wait = self.BACKOFF_BASE * (2**attempt)219 await asyncio.sleep(wait)220 continue221 raise EdgarRateLimitError(f"Rate limited by EDGAR after {self.MAX_RETRIES} attempts")222 223 if response.status_code == 503:224 logger.warning(225 "edgar_unavailable",226 url=url,227 attempt=attempt + 1,228 )229 if attempt < self.MAX_RETRIES - 1:230 wait = self.BACKOFF_BASE * (2**attempt)231 await asyncio.sleep(wait)232 continue233 raise EdgarUnavailableError(f"EDGAR unavailable (503) after {self.MAX_RETRIES} attempts")234 235 response.raise_for_status()236 return response237 238 except httpx.HTTPStatusError:239 raise240 except httpx.HTTPError as e:241 logger.warning(242 "http_error",243 url=url,244 error=str(e),245 attempt=attempt + 1,246 )247 if attempt < self.MAX_RETRIES - 1:248 wait = self.BACKOFF_BASE * (2**attempt)249 await asyncio.sleep(wait)250 continue251 raise EdgarUnavailableError(f"EDGAR unreachable after {self.MAX_RETRIES} attempts: {e}") from e252 253 raise EdgarUnavailableError("Exhausted all retries")254 255 async def ticker_to_cik(self, ticker: str) -> tuple[str, str]:256 """Resolve a stock ticker to its SEC CIK number.257 258 Args:259 ticker: Stock ticker symbol (e.g., "AAPL").260 261 Returns:262 Tuple of (cik_padded, company_name). CIK is zero-padded to 10 digits.263 264 Raises:265 TickerNotFoundError: If the ticker cannot be found in SEC records.266 """267 ticker_upper = ticker.upper().strip()268 269 logger.info("resolving_ticker", ticker=ticker_upper)270 response = await self._request(self.TICKERS_URL)271 data = response.json()272 273 # EDGAR returns format: {"0": {"cik_str": 320193, "ticker": "AAPL", ...}}274 for entry in data.values():275 if entry.get("ticker", "").upper() == ticker_upper:276 cik = str(entry["cik_str"]).zfill(10)277 company_name = entry.get("title", "Unknown")278 logger.info(279 "ticker_resolved",280 ticker=ticker_upper,281 cik=cik,282 company=company_name,283 )284 return cik, company_name285 286 raise TickerNotFoundError(f"Ticker '{ticker_upper}' not found in SEC EDGAR records.")287 288 async def get_filing_urls(self, cik: str, filing_type: str, count: int = 5) -> list[FilingMetadata]:289 """Get recent filing URLs for a company.290 291 Args:292 cik: SEC CIK number (zero-padded to 10 digits).293 filing_type: Filing form type (e.g., "10-K", "10-Q", "8-K").294 count: Maximum number of filings to return.295 296 Returns:297 List of FilingMetadata objects for matching filings.298 299 Raises:300 FilingNotFoundError: If no filings match the criteria.301 """302 filing_type_upper = filing_type.upper().strip()303 304 logger.info(305 "fetching_filings",306 cik=cik,307 filing_type=filing_type_upper,308 count=count,309 )310 311 url = self.SUBMISSIONS_URL.format(cik=cik)312 response = await self._request(url)313 data = response.json()314 315 company_name = data.get("name", "Unknown")316 ticker = data.get("tickers", [""])[0] if data.get("tickers") else ""317 318 recent = data.get("filings", {}).get("recent", {})319 forms = recent.get("form", [])320 dates = recent.get("filingDate", [])321 accession_numbers = recent.get("accessionNumber", [])322 primary_docs = recent.get("primaryDocument", [])323 324 results: list[FilingMetadata] = []325 for i, form in enumerate(forms):326 if form.upper() == filing_type_upper and len(results) < count:327 acc_no = accession_numbers[i].replace("-", "")328 doc_url = f"https://www.sec.gov/Archives/edgar/data/{cik.lstrip('0')}/{acc_no}/{primary_docs[i]}"329 results.append(330 FilingMetadata(331 cik=cik,332 ticker=ticker,333 company_name=company_name,334 filing_type=filing_type_upper,335 filing_date=dates[i],336 accession_number=accession_numbers[i],337 primary_document_url=doc_url,338 )339 )340 341 if not results:342 raise FilingNotFoundError(f"No {filing_type_upper} filings found for CIK {cik} ({company_name})")343 344 logger.info(345 "filings_found",346 count=len(results),347 filing_type=filing_type_upper,348 company=company_name,349 )350 return results351 352 async def download_filing(self, url: str) -> str:353 """Download raw filing content from SEC.354 355 Args:356 url: URL to the filing document.357 358 Returns:359 Raw HTML/text content of the filing.360 361 Raises:362 EdgarUnavailableError: If the filing cannot be downloaded.363 """364 logger.info("downloading_filing", url=url)365 response = await self._request(url)366 content = response.text367 logger.info(368 "filing_downloaded",369 url=url,370 content_length=len(content),371 )372 return content373 374 def parse_sections(self, raw_content: str, filing_type: str) -> dict[str, str]:375 """Extract named sections from filing HTML.376 377 Uses BeautifulSoup to find section headings and extract text378 between them. This is a heuristic parser suitable for most379 modern 10-K filings. Older or unusual filings may not parse380 fully. Unparseable sections are logged as warnings, not errors.381 382 [DEMO-ONLY] This parser handles common patterns well but is383 not production-hardened for all filing variants. Day 2 will384 build a proper section-aware chunker on top of this.385 386 Args:387 raw_content: Raw HTML content of the filing.388 filing_type: Filing type (e.g., "10-K") to select section patterns.389 390 Returns:391 Dict mapping section name to extracted text content.392 """393 if filing_type.upper() not in ("10-K", "10-K/A"):394 # For non-10K filings, return the full text as a single section.395 # Note: for 8-K filings the exhibit (Exhibit 99.1) is fetched separately396 # in ingest_filing() and merged in before saving.397 soup = BeautifulSoup(raw_content, "lxml")398 text = soup.get_text(separator="\n", strip=True)399 return {"full_text": text}400 401 soup = BeautifulSoup(raw_content, "lxml")402 403 sections: dict[str, str] = {}404 full_text = soup.get_text(separator="\n", strip=True)405 406 # Build regex patterns for each 10-K item407 # Match patterns like "Item 1.", "Item 1A.", "ITEM 7" etc.408 item_patterns: list[tuple[str, str, re.Pattern[str]]] = []409 for item_num, item_name in SECTION_10K_ITEMS.items():410 pattern = re.compile(411 rf"(?:^|\n)\s*(?:item|ITEM)\s+{re.escape(item_num)}\.?\s*[\.\-\u2014]?\s*"412 rf"(?:{re.escape(item_name)})?",413 re.IGNORECASE | re.MULTILINE,414 )415 item_patterns.append((item_num, item_name, pattern))416 417 # Find all section start positions418 found_positions: list[tuple[int, str, str]] = []419 for item_num, item_name, pattern in item_patterns:420 matches = list(pattern.finditer(full_text))421 if matches:422 # Use the last match (table of contents often has first match)423 # The actual section content is at the last occurrence424 match = matches[-1] if len(matches) > 1 else matches[0]425 found_positions.append((match.start(), item_num, item_name))426 427 # Sort by position in document428 found_positions.sort(key=lambda x: x[0])429 430 # Extract text between consecutive section headings431 for i, (pos, item_num, item_name) in enumerate(found_positions):432 section_key = f"Item {item_num} - {item_name}"433 if i + 1 < len(found_positions):434 end_pos = found_positions[i + 1][0]435 else:436 end_pos = len(full_text)437 438 section_text = full_text[pos:end_pos].strip()439 440 # Skip very short sections (likely just the heading)441 min_section_length = 100442 if len(section_text) > min_section_length:443 sections[section_key] = section_text444 445 if not sections:446 logger.warning(447 "no_sections_parsed",448 content_length=len(full_text),449 msg="Could not identify standard 10-K sections. Returning full text.",450 )451 sections["full_text"] = full_text452 453 logger.info(454 "sections_parsed",455 section_count=len(sections),456 section_names=list(sections.keys()),457 )458 return sections459 460 async def get_exhibit_urls(461 self,462 cik: str,463 accession_number: str,464 exhibit_types: list[str] | None = None,465 ) -> list[tuple[str, str]]:466 """Fetch the filing index and return URLs of matching exhibits.467 468 8-K filings store actual content (earnings results, press releases)469 in Exhibit 99.1 rather than the main document body. This method470 fetches the filing index page and finds exhibit document URLs.471 472 Args:473 cik: SEC CIK number (zero-padded).474 accession_number: Filing accession number (with dashes).475 exhibit_types: List of exhibit type strings to find, e.g. ["EX-99.1"].476 If None, defaults to ["EX-99.1", "EX-99"].477 478 Returns:479 List of (exhibit_type, url) tuples for matching exhibits.480 """481 if exhibit_types is None:482 exhibit_types = ["EX-99.1", "EX-99", "EX-99.2"]483 484 acc_no_clean = accession_number.replace("-", "")485 index_url = (486 f"https://www.sec.gov/Archives/edgar/data/{cik.lstrip('0')}"487 f"/{acc_no_clean}/{accession_number}-index.htm"488 )489 490 logger.info("fetching_filing_index", url=index_url)491 try:492 response = await self._request(index_url)493 except Exception as e:494 logger.warning("filing_index_unavailable", url=index_url, error=str(e))495 return []496 497 soup = BeautifulSoup(response.text, "lxml")498 results: list[tuple[str, str]] = []499 500 # Parse the EDGAR filing index table501 for row in soup.find_all("tr"):502 cells = row.find_all("td")503 if len(cells) < 4:504 continue505 506 # Column 3 is the formal document Type (e.g., "EX-99.1", "10-K")507 doc_type = cells[3].get_text(strip=True).upper()508 509 if any(doc_type.startswith(ex.upper()) for ex in exhibit_types):510 link = cells[2].find("a")511 if link and link.get("href"):512 href = link["href"]513 if not href.startswith("http"):514 href = "https://www.sec.gov" + href515 results.append((doc_type, href))516 logger.info("exhibit_found", exhibit_type=doc_type, url=href)517 518 if not results:519 logger.warning(520 "no_exhibits_found",521 accession_number=accession_number,522 exhibit_types=exhibit_types,523 )524 return results525 526 async def save_filing(527 self,528 parsed: ParsedFiling,529 data_dir: Path,530 ) -> Path:531 """Save a parsed filing to disk with metadata sidecar.532 533 Creates a directory per company/filing and saves:534 - sections as individual text files535 - metadata as a JSON sidecar536 537 Args:538 parsed: The parsed filing to save.539 data_dir: Base directory for saved filings.540 541 Returns:542 Path to the directory where filing was saved.543 """544 meta = parsed.metadata545 safe_date = meta.filing_date.replace("-", "")546 dir_name = f"{meta.ticker}_{meta.filing_type}_{safe_date}"547 filing_dir = data_dir / dir_name548 filing_dir.mkdir(parents=True, exist_ok=True)549 550 # Save metadata sidecar551 metadata_dict = {552 "cik": meta.cik,553 "ticker": meta.ticker,554 "company_name": meta.company_name,555 "filing_type": meta.filing_type,556 "filing_date": meta.filing_date,557 "accession_number": meta.accession_number,558 "primary_document_url": meta.primary_document_url,559 "raw_content_length": parsed.raw_content_length,560 "sections_found": list(parsed.sections.keys()),561 "saved_at": datetime.now().isoformat(),562 }563 metadata_path = filing_dir / "metadata.json"564 metadata_path.write_text(json.dumps(metadata_dict, indent=2), encoding="utf-8")565 566 # Save each section as a separate text file567 for section_name, section_text in parsed.sections.items():568 safe_name = re.sub(r"[^\w\s-]", "", section_name).strip()569 safe_name = re.sub(r"[\s]+", "_", safe_name).lower()570 section_path = filing_dir / f"{safe_name}.txt"571 section_path.write_text(section_text, encoding="utf-8")572 573 logger.info(574 "filing_saved",575 directory=str(filing_dir),576 sections=len(parsed.sections),577 )578 return filing_dir579 580 581async def ingest_filing(582 ticker: str,583 filing_type: str,584 settings: Settings,585 count: int = 1,586) -> list[Path]:587 """High-level ingestion function: fetch, parse, and save filings.588 589 This is the main entry point for the ingestion pipeline.590 For 8-K filings, also fetches Exhibit 99.1 (press release / earnings591 update) which contains the actual event content.592 593 Args:594 ticker: Stock ticker symbol (e.g., "AAPL").595 filing_type: Filing form type (e.g., "10-K").596 settings: Application settings.597 count: Number of recent filings to fetch.598 599 Returns:600 List of paths to saved filing directories.601 """602 saved_paths: list[Path] = []603 604 async with EdgarClient(settings) as client:605 cik, company_name = await client.ticker_to_cik(ticker)606 filings = await client.get_filing_urls(cik, filing_type, count=count)607 608 for filing_meta in filings:609 logger.info(610 "processing_filing",611 ticker=ticker,612 filing_date=filing_meta.filing_date,613 filing_type=filing_meta.filing_type,614 )615 616 raw_content = await client.download_filing(filing_meta.primary_document_url)617 sections = client.parse_sections(raw_content, filing_type)618 619 # For 8-K filings: fetch Exhibit 99.1 which contains the actual620 # event content (press releases, earnings tables, announcements).621 # The main 8-K body is almost always just a wrapper that says622 # "see Exhibit 99.1" — without the exhibit, no useful chunks exist.623 if filing_type.upper() == "8-K":624 exhibits = await client.get_exhibit_urls(625 cik=cik,626 accession_number=filing_meta.accession_number,627 )628 for exhibit_type, exhibit_url in exhibits:629 try:630 exhibit_raw = await client.download_filing(exhibit_url)631 exhibit_soup = BeautifulSoup(exhibit_raw, "lxml")632 exhibit_text = exhibit_soup.get_text(separator="\n", strip=True)633 if len(exhibit_text) > 200: # Skip empty/boilerplate exhibits634 section_key = f"exhibit_{exhibit_type.lower().replace('-', '_').replace('.', '_')}"635 sections[section_key] = exhibit_text636 logger.info(637 "exhibit_ingested",638 exhibit_type=exhibit_type,639 text_length=len(exhibit_text),640 )641 except Exception as e:642 logger.warning(643 "exhibit_download_failed",644 exhibit_type=exhibit_type,645 url=exhibit_url,646 error=str(e),647 )648 649 parsed = ParsedFiling(650 metadata=filing_meta,651 sections=sections,652 raw_content_length=len(raw_content),653 )654 655 path = await client.save_filing(parsed, settings.data_dir)656 saved_paths.append(path)657 658 logger.info(659 "ingestion_complete",660 ticker=ticker,661 filings_saved=len(saved_paths),662 )663 return saved_paths664