fix(sec): N+1 쿼리 제거, 청크 커밋, 세션 분리, rate limiter 추가

- 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 <noreply@anthropic.com>
main
I Luk Kim 7 months ago
parent af88f89f69
commit f17644e5e8

@ -27,6 +27,7 @@ class SECFilingsService:
"8-K", "6-K", "20-F", "40-F", "10-K", "10-Q", "8-K", "6-K", "20-F", "40-F", "10-K", "10-Q",
"8-K/A", "6-K/A", "20-F/A", "40-F/A", "8-K/A", "6-K/A", "20-F/A", "40-F/A",
} }
CHUNK_SIZE = 500
def __init__(self): def __init__(self):
self._http = SECHttpClient("Stock Oracle Filings Service") self._http = SECHttpClient("Stock Oracle Filings Service")
@ -47,7 +48,7 @@ class SECFilingsService:
Returns the number of filings indexed (inserted or updated). Returns the number of filings indexed (inserted or updated).
""" """
ticker = ticker.upper() ticker = ticker.upper()
self._http.set_deadline(60.0) self._http.set_deadline(120.0)
try: try:
cik = await self._http.get_company_cik(ticker) cik = await self._http.get_company_cik(ticker)
if not cik: if not cik:
@ -126,17 +127,35 @@ class SECFilingsService:
if not raw_filings: if not raw_filings:
return 0 return 0
# Upsert into DB # Pre-fetch existing accession numbers in one query (eliminates N+1)
indexed_count = 0
now = datetime.now(timezone.utc) now = datetime.now(timezone.utc)
for rf in raw_filings: indexed_count = 0
acc = rf["accession_number"]
result = await db.execute( if force_refresh:
select(SECFiling).where(SECFiling.accession_number == acc) existing_result = await db.execute(
select(SECFiling).where(SECFiling.ticker == ticker)
) )
existing = result.scalar_one_or_none() existing_map: Dict[str, SECFiling] = {
if existing: 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: if force_refresh:
existing = existing_map[acc]
existing.ticker = rf["ticker"] existing.ticker = rf["ticker"]
existing.cik = rf["cik"] existing.cik = rf["cik"]
existing.form_type = rf["form_type"] existing.form_type = rf["form_type"]
@ -147,8 +166,9 @@ class SECFilingsService:
existing.filing_description = rf["filing_description"] existing.filing_description = rf["filing_description"]
existing.indexed_at = now existing.indexed_at = now
existing.updated_at = now existing.updated_at = now
indexed_count += 1 chunk_count += 1
else: continue
filing = SECFiling( filing = SECFiling(
ticker=rf["ticker"], ticker=rf["ticker"],
cik=rf["cik"], cik=rf["cik"],
@ -164,10 +184,19 @@ class SECFilingsService:
updated_at=now, updated_at=now,
) )
db.add(filing) db.add(filing)
indexed_count += 1 existing_accs.add(acc) # prevent duplicates within same run
chunk_count += 1
if indexed_count: if chunk_count:
try:
await db.commit() 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()
logger.info(f"Indexed {indexed_count} filings for {ticker}") logger.info(f"Indexed {indexed_count} filings for {ticker}")
return indexed_count return indexed_count
finally: finally:
@ -409,15 +438,18 @@ class SECFilingsService:
missing = [t for t in tickers if indexed_counts.get(t, 0) == 0] missing = [t for t in tickers if indexed_counts.get(t, 0) == 0]
# Index missing tickers in parallel (Semaphore(4) to not overwhelm SEC) # Index missing tickers in parallel (Semaphore(4) to not overwhelm SEC)
# Each coroutine uses its own session to avoid concurrent-session conflicts
if missing: if missing:
sem = asyncio.Semaphore(4) sem = asyncio.Semaphore(4)
async def _index_one(ticker: str) -> None: async def _index_one(ticker: str) -> None:
async with sem: async with sem:
try: try:
from app.core.database import AsyncSessionLocal
async with AsyncSessionLocal() as session:
await asyncio.wait_for( await asyncio.wait_for(
self.index_filings(db, ticker, form_types=form_types), self.index_filings(session, ticker, form_types=form_types),
timeout=60, timeout=120,
) )
except Exception as e: except Exception as e:
logger.warning(f"Bulk index failed for {ticker}: {e}") logger.warning(f"Bulk index failed for {ticker}: {e}")

@ -32,11 +32,39 @@ def _is_sec_block_page(text: str) -> bool:
return False 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: class SECHttpClient:
"""Shared HTTP client for SEC EDGAR API requests. """Shared HTTP client for SEC EDGAR API requests.
Features: Features:
- asyncio.Semaphore(2) concurrent request limit - asyncio.Semaphore(2) concurrent request limit
- Token bucket rate limiter (10 req/sec)
- 6 retries with 1.8x exponential backoff + random jitter - 6 retries with 1.8x exponential backoff + random jitter
- 429 Retry-After header respect - 429 Retry-After header respect
- SEC block page detection - SEC block page detection
@ -50,6 +78,7 @@ class SECHttpClient:
self.sec_base_www = "https://www.sec.gov" self.sec_base_www = "https://www.sec.gov"
self.http_timeout = aiohttp.ClientTimeout(total=12) self.http_timeout = aiohttp.ClientTimeout(total=12)
self._req_sem = asyncio.Semaphore(2) self._req_sem = asyncio.Semaphore(2)
self._rate_limiter = _TokenBucket(rate=10.0, capacity=10.0)
self._text_cache: Dict[str, str] = {} self._text_cache: Dict[str, str] = {}
self._json_cache: Dict[str, dict] = {} self._json_cache: Dict[str, dict] = {}
self._cache_dir = "/tmp/stock_oracle_sec_cache" self._cache_dir = "/tmp/stock_oracle_sec_cache"
@ -170,6 +199,7 @@ class SECHttpClient:
total=min(remaining, getattr(self.http_timeout, "total", 12)) total=min(remaining, getattr(self.http_timeout, "total", 12))
) )
async with self._req_sem: async with self._req_sem:
await self._rate_limiter.acquire()
try: try:
session = await self._get_session() session = await self._get_session()
async with session.get(url, timeout=req_timeout, headers={"Accept": "application/json"}) as resp: 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)) total=min(remaining, getattr(self.http_timeout, "total", 12))
) )
async with self._req_sem: async with self._req_sem:
await self._rate_limiter.acquire()
try: try:
session = await self._get_session() session = await self._get_session()
async with session.get(url, timeout=req_timeout, headers={"Accept": accept}) as resp: async with session.get(url, timeout=req_timeout, headers={"Accept": accept}) as resp:

Loading…
Cancel
Save