From f17644e5e8aca5f67c8aece19a0866b3e3758f04 Mon Sep 17 00:00:00 2001 From: I Luk Kim Date: Sun, 15 Mar 2026 00:57:53 -0700 Subject: [PATCH] =?UTF-8?q?fix(sec):=20N+1=20=EC=BF=BC=EB=A6=AC=20?= =?UTF-8?q?=EC=A0=9C=EA=B1=B0,=20=EC=B2=AD=ED=81=AC=20=EC=BB=A4=EB=B0=8B,?= =?UTF-8?q?=20=EC=84=B8=EC=85=98=20=EB=B6=84=EB=A6=AC,=20rate=20limiter=20?= =?UTF-8?q?=EC=B6=94=EA=B0=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - sec_filings_service: 루프 전 accession_number 일괄 pre-fetch로 N+1 제거 - sec_filings_service: 500건 단위 청크 커밋으로 all-or-nothing 트랜잭션 방지 - sec_filings_service: bulk 인덱싱 시 코루틴별 독립 세션 생성으로 동시 세션 충돌 해결 - sec_filings_service: index_filings 데드라인 60s → 120s, bulk timeout 동일 적용 - sec_http_client: _TokenBucket(10 req/sec) 추가로 SEC EDGAR 과부하 방지 Co-Authored-By: Claude Sonnet 4.6 --- app/services/sec_filings_service.py | 90 +++++++++++++++++++---------- app/services/sec_http_client.py | 31 ++++++++++ 2 files changed, 92 insertions(+), 29 deletions(-) diff --git a/app/services/sec_filings_service.py b/app/services/sec_filings_service.py index 2c8cc7f..de9efaf 100644 --- a/app/services/sec_filings_service.py +++ b/app/services/sec_filings_service.py @@ -27,6 +27,7 @@ class SECFilingsService: "8-K", "6-K", "20-F", "40-F", "10-K", "10-Q", "8-K/A", "6-K/A", "20-F/A", "40-F/A", } + CHUNK_SIZE = 500 def __init__(self): self._http = SECHttpClient("Stock Oracle Filings Service") @@ -47,7 +48,7 @@ class SECFilingsService: Returns the number of filings indexed (inserted or updated). """ ticker = ticker.upper() - self._http.set_deadline(60.0) + self._http.set_deadline(120.0) try: cik = await self._http.get_company_cik(ticker) if not cik: @@ -126,29 +127,48 @@ class SECFilingsService: if not raw_filings: return 0 - # Upsert into DB - indexed_count = 0 + # Pre-fetch existing accession numbers in one query (eliminates N+1) now = datetime.now(timezone.utc) - for rf in raw_filings: - acc = rf["accession_number"] - result = await db.execute( - select(SECFiling).where(SECFiling.accession_number == acc) + indexed_count = 0 + + if force_refresh: + existing_result = await db.execute( + select(SECFiling).where(SECFiling.ticker == ticker) ) - existing = result.scalar_one_or_none() - if existing: - if force_refresh: - 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 - indexed_count += 1 - else: + 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 = {} + + # Upsert in chunks to avoid all-or-nothing transaction failures + 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"], @@ -164,10 +184,19 @@ class SECFilingsService: updated_at=now, ) db.add(filing) - indexed_count += 1 + 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() - if indexed_count: - await db.commit() logger.info(f"Indexed {indexed_count} filings for {ticker}") return indexed_count finally: @@ -409,16 +438,19 @@ class SECFilingsService: missing = [t for t in tickers if indexed_counts.get(t, 0) == 0] # Index missing tickers in parallel (Semaphore(4) to not overwhelm SEC) + # Each coroutine uses its own session to avoid concurrent-session conflicts if missing: sem = asyncio.Semaphore(4) async def _index_one(ticker: str) -> None: async with sem: try: - await asyncio.wait_for( - self.index_filings(db, ticker, form_types=form_types), - timeout=60, - ) + from app.core.database import AsyncSessionLocal + async with AsyncSessionLocal() as session: + await asyncio.wait_for( + self.index_filings(session, ticker, form_types=form_types), + timeout=120, + ) except Exception as e: logger.warning(f"Bulk index failed for {ticker}: {e}") diff --git a/app/services/sec_http_client.py b/app/services/sec_http_client.py index 8fdb20a..b52f2ef 100644 --- a/app/services/sec_http_client.py +++ b/app/services/sec_http_client.py @@ -32,11 +32,39 @@ def _is_sec_block_page(text: str) -> bool: return False +class _TokenBucket: + """Async token bucket rate limiter. + + Allows up to `rate` requests per second with a burst capacity of `capacity`. + """ + + def __init__(self, rate: float, capacity: float) -> None: + self._rate = rate + self._capacity = capacity + self._tokens = capacity + self._last_refill = _time.monotonic() + self._lock = asyncio.Lock() + + async def acquire(self) -> None: + while True: + async with self._lock: + now = _time.monotonic() + elapsed = now - self._last_refill + self._tokens = min(self._capacity, self._tokens + elapsed * self._rate) + self._last_refill = now + if self._tokens >= 1.0: + self._tokens -= 1.0 + return + wait_time = (1.0 - self._tokens) / self._rate + await asyncio.sleep(wait_time) + + class SECHttpClient: """Shared HTTP client for SEC EDGAR API requests. Features: - asyncio.Semaphore(2) concurrent request limit + - Token bucket rate limiter (10 req/sec) - 6 retries with 1.8x exponential backoff + random jitter - 429 Retry-After header respect - SEC block page detection @@ -50,6 +78,7 @@ class SECHttpClient: self.sec_base_www = "https://www.sec.gov" self.http_timeout = aiohttp.ClientTimeout(total=12) self._req_sem = asyncio.Semaphore(2) + self._rate_limiter = _TokenBucket(rate=10.0, capacity=10.0) self._text_cache: Dict[str, str] = {} self._json_cache: Dict[str, dict] = {} self._cache_dir = "/tmp/stock_oracle_sec_cache" @@ -170,6 +199,7 @@ class SECHttpClient: total=min(remaining, getattr(self.http_timeout, "total", 12)) ) async with self._req_sem: + await self._rate_limiter.acquire() try: session = await self._get_session() async with session.get(url, timeout=req_timeout, headers={"Accept": "application/json"}) as resp: @@ -264,6 +294,7 @@ class SECHttpClient: total=min(remaining, getattr(self.http_timeout, "total", 12)) ) async with self._req_sem: + await self._rate_limiter.acquire() try: session = await self._get_session() async with session.get(url, timeout=req_timeout, headers={"Accept": accept}) as resp: