diff --git a/app/services/sec_filings_service.py b/app/services/sec_filings_service.py index d70fd6e..2d41737 100644 --- a/app/services/sec_filings_service.py +++ b/app/services/sec_filings_service.py @@ -6,6 +6,7 @@ and extract exhibit content (e.g., EX-99.1 press releases). import asyncio import logging import time +import uuid from datetime import datetime, timezone from typing import Dict, List, Optional, Set, Tuple @@ -13,6 +14,7 @@ from app.core.config import settings from bs4 import BeautifulSoup from sqlalchemy import select, and_, func +from sqlalchemy.dialects.postgresql import insert as pg_insert from sqlalchemy.ext.asyncio import AsyncSession from app.models.filing import SECFiling @@ -139,83 +141,85 @@ class SECFilingsService: if not raw_filings: return 0 - # Pre-fetch existing accession numbers in one query (eliminates N+1) - now = datetime.now(timezone.utc) - indexed_count = 0 + # Snapshot pre-existing accessions to identify truly new inserts later + acc_result = await db.execute( + select(SECFiling.accession_number).where(SECFiling.ticker == ticker) + ) + _pre_existing_accs: Set[str] = {row[0] for row in acc_result.fetchall()} - if force_refresh: - existing_result = await db.execute( - select(SECFiling).where(SECFiling.ticker == ticker) - ) - existing_map: Dict[str, SECFiling] = { - f.accession_number: f for f in existing_result.scalars().all() - } - existing_accs: Set[str] = set(existing_map.keys()) - else: - acc_result = await db.execute( - select(SECFiling.accession_number).where(SECFiling.ticker == ticker) - ) - existing_accs = {row[0] for row in acc_result.fetchall()} - existing_map = {} + # Deduplicate within the same SEC response (accession_number may repeat) + seen: Set[str] = set() + deduped: List[Dict] = [] + for rf in raw_filings: + if rf["accession_number"] not in seen: + seen.add(rf["accession_number"]) + deduped.append(rf) + raw_filings = deduped - # Snapshot pre-existing accessions to identify truly new inserts later - _pre_existing_accs: Set[str] = set(existing_accs) + now = datetime.now(timezone.utc) + indexed_count = 0 - # Upsert in chunks to avoid all-or-nothing transaction failures + # Upsert in chunks (ON CONFLICT eliminates race conditions) for chunk_start in range(0, len(raw_filings), self.CHUNK_SIZE): chunk = raw_filings[chunk_start: chunk_start + self.CHUNK_SIZE] - chunk_count = 0 - - for rf in chunk: - acc = rf["accession_number"] - if acc in existing_accs: - if force_refresh: - existing = existing_map[acc] - existing.ticker = rf["ticker"] - existing.cik = rf["cik"] - existing.form_type = rf["form_type"] - existing.filing_date = rf["filing_date"] - existing.accepted_at = rf["accepted_at"] - existing.primary_document = rf["primary_document"] - existing.primary_document_url = rf["primary_document_url"] - existing.filing_description = rf["filing_description"] - existing.indexed_at = now - existing.updated_at = now - chunk_count += 1 - continue - filing = SECFiling( - ticker=rf["ticker"], - cik=rf["cik"], - accession_number=rf["accession_number"], - form_type=rf["form_type"], - filing_date=rf["filing_date"], - accepted_at=rf["accepted_at"], - primary_document=rf["primary_document"], - primary_document_url=rf["primary_document_url"], - filing_description=rf["filing_description"], - parsed_status="pending" if rf["form_type"] in ("8-K", "8-K/A") else None, - indexed_at=now, - created_at=now, - updated_at=now, + rows = [ + { + "id": uuid.uuid4(), + "ticker": rf["ticker"], + "cik": rf["cik"], + "accession_number": rf["accession_number"], + "form_type": rf["form_type"], + "filing_date": rf["filing_date"], + "accepted_at": rf["accepted_at"], + "primary_document": rf["primary_document"], + "primary_document_url": rf["primary_document_url"], + "filing_description": rf["filing_description"], + "parsed_status": "pending" if rf["form_type"] in ("8-K", "8-K/A") else None, + "indexed_at": now, + "created_at": now, + "updated_at": now, + } + for rf in chunk + ] + + stmt = pg_insert(SECFiling).values(rows) + if force_refresh: + stmt = stmt.on_conflict_do_update( + index_elements=["accession_number"], + set_={ + "ticker": stmt.excluded.ticker, + "cik": stmt.excluded.cik, + "form_type": stmt.excluded.form_type, + "filing_date": stmt.excluded.filing_date, + "accepted_at": stmt.excluded.accepted_at, + "primary_document": stmt.excluded.primary_document, + "primary_document_url": stmt.excluded.primary_document_url, + "filing_description": stmt.excluded.filing_description, + "indexed_at": stmt.excluded.indexed_at, + "updated_at": stmt.excluded.updated_at, + }, + ) + else: + stmt = stmt.on_conflict_do_nothing( + index_elements=["accession_number"], ) - db.add(filing) - existing_accs.add(acc) # prevent duplicates within same run - chunk_count += 1 - if chunk_count: - try: - await db.commit() - indexed_count += chunk_count - except Exception as e: - logger.error( - f"Chunk commit failed at offset {chunk_start} for {ticker}: {e}" - ) - await db.rollback() + try: + result = await db.execute(stmt) + await db.commit() + indexed_count += result.rowcount + except Exception as e: + logger.error( + f"Chunk upsert failed at offset {chunk_start} for {ticker}: {e}" + ) + await db.rollback() logger.info(f"Indexed {indexed_count} filings for {ticker}") - # Auto-parse newly inserted 8-K filings (non-blocking; failures are logged) + # Auto-parse newly inserted 8-K filings — cap at 20 most recent + # to avoid blocking for tickers with 1000+ historical 8-Ks. + # Remaining pending filings are lazy-parsed via /events endpoint. truly_new_8k = [ rf["accession_number"] for rf in raw_filings @@ -223,9 +227,11 @@ class SECFilingsService: and rf["accession_number"] not in _pre_existing_accs ] if truly_new_8k: + # raw_filings is ordered newest-first from SEC, so slice keeps recent ones + parse_batch = truly_new_8k[:20] from app.services.sec_8k_parser import sec_8k_parser try: - await sec_8k_parser.parse_filings_by_accessions(db, truly_new_8k) + await sec_8k_parser.parse_filings_by_accessions(db, parse_batch) except Exception as e: logger.warning(f"Auto-parse failed for {ticker} 8-Ks: {e}") @@ -233,6 +239,25 @@ class SECFilingsService: finally: self._http.clear_deadline() + # ------------------------------------------------------------------ + # Background staleness re-index (non-blocking) + # ------------------------------------------------------------------ + + async def _background_reindex( + self, ticker: str, form_types: Optional[Set[str]], lock: asyncio.Lock + ) -> None: + """Re-index a stale ticker in the background with its own DB session.""" + async with lock: + try: + from app.core.database import AsyncSessionLocal + async with AsyncSessionLocal() as session: + new_count = await self.index_filings( + session, ticker, form_types=form_types, skip_cache=True + ) + logger.info(f"Background reindex for {ticker}: {new_count} new filings") + except Exception as e: + logger.warning(f"Background reindex failed for {ticker}: {e}") + # ------------------------------------------------------------------ # search_filings: query DB with filters # ------------------------------------------------------------------ @@ -283,8 +308,8 @@ class SECFilingsService: return [], 0 else: - # Ticker already has filings — check for staleness and incrementally re-index - # if the data is older than SEC_DATA_REFRESH_HOURS. + # Ticker already has filings — check for staleness and fire background re-index. + # Returns stale data immediately; the background task updates for next requests. staleness_result = await db.execute( select(func.max(SECFiling.indexed_at)).where(SECFiling.ticker == ticker) ) @@ -295,42 +320,30 @@ class SECFilingsService: (now_utc - last_indexed).total_seconds() > refresh_seconds ) if is_stale: - # Use per-ticker lock to prevent thundering herd. - # If another coroutine is already re-indexing this ticker, skip and - # serve stale data rather than stacking up duplicate SEC requests. if ticker not in self._reindex_locks: self._reindex_locks[ticker] = asyncio.Lock() lock = self._reindex_locks[ticker] if not lock.locked(): - async with lock: - try: - new_count = await self.index_filings( - db, ticker, form_types=form_types, skip_cache=True + # Touch indexed_at immediately to prevent re-trigger from + # concurrent requests while background task runs. + await db.execute( + SECFiling.__table__.update() + .where( + SECFiling.id.in_( + select(SECFiling.id) + .where(SECFiling.ticker == ticker) + .order_by(SECFiling.filing_date.desc()) + .limit(1) ) - if new_count == 0: - # No new filings from SEC — touch indexed_at on one row so - # the next query doesn't immediately re-trigger staleness. - await db.execute( - SECFiling.__table__.update() - .where( - SECFiling.id.in_( - select(SECFiling.id) - .where(SECFiling.ticker == ticker) - .order_by(SECFiling.filing_date.desc()) - .limit(1) - ) - ) - .values(indexed_at=now_utc) - ) - await db.commit() - # Re-count to include any newly inserted filings - count_result = await db.execute( - select(func.count()).select_from(SECFiling).where(where_clause) - ) - total_count = count_result.scalar() or 0 - except Exception as e: - logger.warning(f"Staleness re-index failed for {ticker}: {e}") - # Serve existing stale data rather than returning empty + ) + .values(indexed_at=now_utc) + ) + await db.commit() + + # Fire background re-index with its own DB session + asyncio.ensure_future( + self._background_reindex(ticker, form_types, lock) + ) # Query with pagination result = await db.execute(