yirugi
/
stock-oracle
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix: SEC filings 동시 insert 경쟁 조건 수정 (ON CONFLICT + 백그라운드 재인덱스)

Browse Source 
동시 요청 시 둘 다 빈 existing_accs 읽고 중복 INSERT → UniqueViolationError 발생.
pg_insert().on_conflict_do_nothing()으로 원자적 upsert 교체,
staleness 재인덱스를 background task로 이동해 요청 차단 제거.
auto-parse는 최근 20개로 제한.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
main
I Luk Kim 4 months ago
parent b4d8c97ba2
commit be8dce201d
1 changed files with 113 additions and 100 deletions
Show Stats Download Patch File Download Diff File

169
app/services/sec_filings_service.py
Unescape Escape View File

@ -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
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:
# Snapshot pre-existing accessions to identify truly new inserts later
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 = {}
_pre_existing_accs: Set[str] = {row[0] for row in acc_result.fetchall()}
# Snapshot pre-existing accessions to identify truly new inserts later
_pre_existing_accs: Set[str] = set(existing_accs)
# 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
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
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
]
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,
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:
result = await db.execute(stmt)
await db.commit()
indexed_count += chunk_count
indexed_count += result.rowcount
except Exception as e:
logger.error(
f"Chunk commit failed at offset {chunk_start} for {ticker}: {e}"
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,21 +320,12 @@ 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
)
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.
# Touch indexed_at immediately to prevent re-trigger from
# concurrent requests while background task runs.
await db.execute(
SECFiling.__table__.update()
.where(
@ -323,14 +339,11 @@ class SECFilingsService:
.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)
# Fire background re-index with its own DB session
asyncio.ensure_future(
self._background_reindex(ticker, form_types, lock)
)
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
# Query with pagination
result = await db.execute(

Write Preview
Loading…
Cancel
Save