perf: Phase 1-3 Stock Oracle API 성능 개선

Phase 1A - yfinance hang 제거
- _run_with_timeout() 헬퍼 추가 (asyncio.wait_for 래퍼)
- run_in_executor 6곳에 timeout 적용: history(30s), .info(20s), bulk download(60s)

Phase 1B - SEC Filing deadline 설정
- index_filings: 60s deadline + try/finally
- get_filing_documents: 30s deadline (중첩 호출 시 기존 deadline 유지)
- get_exhibit_content: 30s deadline + try/finally

Phase 1C - _get_ticker_max_range async 전환
- sync → async def + run_in_executor + wait_for(20s)
- get_or_create_company_data에서 period=="max" 사전 체크 → await 직접 호출

Phase 1D - Endpoint 레벨 timeout
- POST /price/data/bulk: 300s → 504
- POST /financial/data/bulk: 300s → 504
- GET /filings/search/{ticker}: 120s → 504
- GET /filings/documents/{accession}: 30s → 504
- GET /filings/exhibit/{accession}: 30s → 504

Phase 2 - Filing 캐시 추가
- GET /filings/documents: @with_cache(ttl=86400)
- GET /filings/exhibit: @with_cache(ttl=86400)

Phase 3A - POST /filings/search/bulk 추가
- BulkFilingSearchRequest/Item/Response 스키마
- search_filings_bulk(): 배치 DB 조회 → 미인덱싱 ticker 병렬 인덱싱(Semaphore 4)
- @with_cache(ttl=3600), 600s endpoint timeout

Phase 3B - POST /filings/exhibit/bulk 추가
- BulkExhibitRequest/Item/Response 스키마
- asyncio.gather + 개별 30s timeout, 최대 50건

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
main
I Luk Kim 5 months ago
parent b88835cded
commit 45d832ba5c

@ -2,15 +2,22 @@
SEC Filings endpoints: search, document listing, and exhibit extraction. SEC Filings endpoints: search, document listing, and exhibit extraction.
""" """
import asyncio
import logging import logging
from datetime import datetime, timezone from datetime import datetime, timezone
from typing import Optional from typing import List, Optional
from fastapi import APIRouter, Depends, HTTPException, Query, Response from fastapi import APIRouter, Depends, HTTPException, Query, Response
from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.ext.asyncio import AsyncSession
from app.core.database import get_db from app.core.database import get_db
from app.schemas.filing import ( from app.schemas.filing import (
BulkExhibitItem,
BulkExhibitRequest,
BulkExhibitResponse,
BulkFilingSearchItem,
BulkFilingSearchRequest,
BulkFilingSearchResponse,
ExhibitContentResponse, ExhibitContentResponse,
FilingDocumentInfo, FilingDocumentInfo,
FilingDocumentListResponse, FilingDocumentListResponse,
@ -75,7 +82,8 @@ async def search_filings(
raise HTTPException(status_code=502, detail=f"SEC indexing failed: {e}") raise HTTPException(status_code=502, detail=f"SEC indexing failed: {e}")
try: try:
filings, total_count = await sec_filings_service.search_filings( filings, total_count = await asyncio.wait_for(
sec_filings_service.search_filings(
db, db,
ticker, ticker,
form_types=form_types_set, form_types=form_types_set,
@ -83,7 +91,11 @@ async def search_filings(
end_date=end_dt, end_date=end_dt,
limit=limit, limit=limit,
offset=offset, offset=offset,
),
timeout=120,
) )
except asyncio.TimeoutError:
raise HTTPException(status_code=504, detail="Filing search timed out after 120s.")
except ValueError as e: except ValueError as e:
raise HTTPException(status_code=404, detail=str(e)) raise HTTPException(status_code=404, detail=str(e))
except Exception as e: except Exception as e:
@ -118,13 +130,20 @@ async def search_filings(
@router.get("/documents/{accession_number}", response_model=FilingDocumentListResponse) @router.get("/documents/{accession_number}", response_model=FilingDocumentListResponse)
@with_cache(namespace="filings:documents", ttl=86400, key_params=["accession_number"])
async def get_filing_documents( async def get_filing_documents(
accession_number: str, accession_number: str,
response: Response,
db: AsyncSession = Depends(get_db), db: AsyncSession = Depends(get_db),
): ):
"""List all documents within a SEC filing.""" """List all documents within a SEC filing."""
try: try:
documents = await sec_filings_service.get_filing_documents(db, accession_number) documents = await asyncio.wait_for(
sec_filings_service.get_filing_documents(db, accession_number),
timeout=30,
)
except asyncio.TimeoutError:
raise HTTPException(status_code=504, detail="Document listing timed out after 30s.")
except ValueError as e: except ValueError as e:
raise HTTPException(status_code=404, detail=str(e)) raise HTTPException(status_code=404, detail=str(e))
except Exception as e: except Exception as e:
@ -140,16 +159,21 @@ async def get_filing_documents(
@router.get("/exhibit/{accession_number}", response_model=ExhibitContentResponse) @router.get("/exhibit/{accession_number}", response_model=ExhibitContentResponse)
@with_cache(namespace="filings:exhibit", ttl=86400, key_params=["accession_number", "exhibit_type"])
async def get_exhibit_content( async def get_exhibit_content(
accession_number: str, accession_number: str,
response: Response,
exhibit_type: str = Query("EX-99.1", description="Exhibit type (e.g. EX-99.1)"), exhibit_type: str = Query("EX-99.1", description="Exhibit type (e.g. EX-99.1)"),
db: AsyncSession = Depends(get_db), db: AsyncSession = Depends(get_db),
): ):
"""Extract exhibit content (e.g., press release EX-99.1) from a filing.""" """Extract exhibit content (e.g., press release EX-99.1) from a filing."""
try: try:
result = await sec_filings_service.get_exhibit_content( result = await asyncio.wait_for(
db, accession_number, exhibit_type sec_filings_service.get_exhibit_content(db, accession_number, exhibit_type),
timeout=30,
) )
except asyncio.TimeoutError:
raise HTTPException(status_code=504, detail="Exhibit extraction timed out after 30s.")
except ValueError as e: except ValueError as e:
raise HTTPException(status_code=404, detail=str(e)) raise HTTPException(status_code=404, detail=str(e))
except Exception as e: except Exception as e:
@ -164,3 +188,150 @@ async def get_exhibit_content(
filename=result.get("filename"), filename=result.get("filename"),
url=result.get("url"), url=result.get("url"),
) )
@router.post("/search/bulk", response_model=BulkFilingSearchResponse)
@with_cache(namespace="filings:search:bulk", ttl=3600, key_params=["request"])
async def search_filings_bulk(
request: BulkFilingSearchRequest,
response: Response,
db: AsyncSession = Depends(get_db),
):
"""Bulk search SEC filings for multiple tickers. Auto-indexes from SEC if not in DB."""
form_types_set = None
if request.form_type:
form_types_set = {ft.strip().upper() for ft in request.form_type.split(",") if ft.strip()}
unsupported = form_types_set - sec_filings_service.SUPPORTED_FORM_TYPES
if unsupported:
raise HTTPException(
status_code=400,
detail=f"Unsupported form types: {unsupported}.",
)
start_dt = None
end_dt = None
try:
if request.start_date:
start_dt = datetime.strptime(request.start_date, "%Y-%m-%d").replace(tzinfo=timezone.utc)
if request.end_date:
end_dt = datetime.strptime(request.end_date, "%Y-%m-%d").replace(
hour=23, minute=59, second=59, tzinfo=timezone.utc
)
except ValueError:
raise HTTPException(status_code=400, detail="Invalid date format. Use YYYY-MM-DD")
try:
raw_results, elapsed = await asyncio.wait_for(
sec_filings_service.search_filings_bulk(
db,
request.tickers,
form_types=form_types_set,
start_date=start_dt,
end_date=end_dt,
limit_per_ticker=request.limit_per_ticker,
),
timeout=600,
)
except asyncio.TimeoutError:
raise HTTPException(status_code=504, detail="Bulk filing search timed out after 600s.")
except Exception as e:
logger.error(f"Bulk filing search failed: {e}")
raise HTTPException(status_code=502, detail=f"Bulk filing search failed: {e}")
items = []
successful = 0
failed = 0
for r in raw_results:
summaries = []
for f in r["filings"]:
docs_count = len(f.documents_json) if f.documents_json else None
summaries.append(
FilingSummary(
accession_number=f.accession_number,
form_type=f.form_type,
filing_date=f.filing_date.strftime("%Y-%m-%d"),
accepted_at=f.accepted_at.isoformat() if f.accepted_at else None,
primary_document=f.primary_document,
filing_description=f.filing_description,
documents_count=docs_count,
)
)
items.append(BulkFilingSearchItem(
ticker=r["ticker"],
success=r["success"],
filings=summaries,
total_count=r["total_count"],
error=r["error"],
))
if r["success"]:
successful += 1
else:
failed += 1
return BulkFilingSearchResponse(
results=items,
total_tickers=len(items),
successful_count=successful,
failed_count=failed,
query_time_seconds=round(elapsed, 3),
metadata={"form_types": list(form_types_set) if form_types_set else None},
)
@router.post("/exhibit/bulk", response_model=BulkExhibitResponse)
async def get_exhibit_bulk(
request: BulkExhibitRequest,
db: AsyncSession = Depends(get_db),
):
"""Bulk fetch exhibit content for multiple accession numbers."""
import time as _time
t0 = _time.monotonic()
async def _fetch_one(item: dict) -> BulkExhibitItem:
accession_number = item.get("accession_number", "")
exhibit_type = item.get("exhibit_type", "EX-99.1")
try:
result = await asyncio.wait_for(
sec_filings_service.get_exhibit_content(db, accession_number, exhibit_type),
timeout=30,
)
return BulkExhibitItem(
accession_number=accession_number,
exhibit_type=exhibit_type,
success=True,
content=result["content"],
content_type=result.get("content_type"),
filename=result.get("filename"),
url=result.get("url"),
)
except Exception as e:
return BulkExhibitItem(
accession_number=accession_number,
exhibit_type=exhibit_type,
success=False,
error=str(e),
)
raw = await asyncio.gather(*[_fetch_one(item) for item in request.items], return_exceptions=True)
results: List[BulkExhibitItem] = []
for r in raw:
if isinstance(r, Exception):
results.append(BulkExhibitItem(
accession_number="unknown",
exhibit_type="unknown",
success=False,
error=str(r),
))
else:
results.append(r)
successful = sum(1 for r in results if r.success)
elapsed = _time.monotonic() - t0
return BulkExhibitResponse(
results=results,
total_items=len(results),
successful_count=successful,
failed_count=len(results) - successful,
query_time_seconds=round(elapsed, 3),
)

@ -620,9 +620,15 @@ async def get_bulk_financial_data(
async with semaphore: async with semaphore:
return await process_ticker(ticker) return await process_ticker(ticker)
# Execute all tickers in parallel # Execute all tickers in parallel (300s endpoint-level timeout)
tasks = [process_with_limit(ticker) for ticker in request.tickers] tasks = [process_with_limit(ticker) for ticker in request.tickers]
results = await asyncio.gather(*tasks, return_exceptions=True) try:
results = await asyncio.wait_for(
asyncio.gather(*tasks, return_exceptions=True),
timeout=300,
)
except asyncio.TimeoutError:
raise HTTPException(status_code=504, detail="Bulk financial data request timed out after 300s.")
# Count successful and failed results # Count successful and failed results
successful_count = 0 successful_count = 0

@ -448,15 +448,22 @@ async def get_bulk_price_data(
} }
) )
# Use optimized bulk processing method # Use optimized bulk processing method (300s endpoint-level timeout)
results, successful_count, failed_count = await price_service.get_multiple_tickers_data_optimized( import asyncio as _asyncio
try:
results, successful_count, failed_count = await _asyncio.wait_for(
price_service.get_multiple_tickers_data_optimized(
db=db, db=db,
tickers=request.tickers, tickers=request.tickers,
start_date=start_date, start_date=start_date,
end_date=end_date, end_date=end_date,
interval=request.interval, interval=request.interval,
force_refresh=request.force_refresh force_refresh=request.force_refresh
),
timeout=300,
) )
except _asyncio.TimeoutError:
raise HTTPException(status_code=504, detail="Bulk price data request timed out after 300s.")
return BulkPriceDataResponse( return BulkPriceDataResponse(
results=results, results=results,

@ -3,7 +3,7 @@ Pydantic schemas for SEC filing endpoints.
""" """
from datetime import datetime from datetime import datetime
from typing import List, Optional from typing import Any, Dict, List, Optional
from pydantic import BaseModel, Field from pydantic import BaseModel, Field
@ -45,3 +45,56 @@ class ExhibitContentResponse(BaseModel):
content_type: Optional[str] = None content_type: Optional[str] = None
filename: Optional[str] = None filename: Optional[str] = None
url: Optional[str] = None url: Optional[str] = None
# ── Bulk filing search ──────────────────────────────────────────────────────
class BulkFilingSearchRequest(BaseModel):
tickers: List[str] = Field(..., min_length=1, max_length=200)
form_type: Optional[str] = None
start_date: Optional[str] = None
end_date: Optional[str] = None
limit_per_ticker: int = Field(default=20, ge=1, le=100)
class BulkFilingSearchItem(BaseModel):
ticker: str
success: bool
filings: List[FilingSummary] = []
total_count: int = 0
error: Optional[str] = None
class BulkFilingSearchResponse(BaseModel):
results: List[BulkFilingSearchItem]
total_tickers: int
successful_count: int
failed_count: int
query_time_seconds: float
metadata: Dict[str, Any] = Field(default_factory=dict)
# ── Bulk exhibit ────────────────────────────────────────────────────────────
class BulkExhibitRequest(BaseModel):
items: List[Dict[str, str]] = Field(..., min_length=1, max_length=50)
# Each item: {"accession_number": "...", "exhibit_type": "EX-99.1"}
class BulkExhibitItem(BaseModel):
accession_number: str
exhibit_type: str
success: bool
content: Optional[str] = None
content_type: Optional[str] = None
filename: Optional[str] = None
url: Optional[str] = None
error: Optional[str] = None
class BulkExhibitResponse(BaseModel):
results: List[BulkExhibitItem]
total_items: int
successful_count: int
failed_count: int
query_time_seconds: float

@ -5,6 +5,7 @@ Real financial data service that combines price data with calculations
from datetime import datetime, timezone, timedelta, date from datetime import datetime, timezone, timedelta, date
from typing import Dict, List, Optional, Tuple, Union from typing import Dict, List, Optional, Tuple, Union
import logging import logging
import asyncio
import numpy as np import numpy as np
from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, and_, or_, desc from sqlalchemy import select, and_, or_, desc
@ -23,36 +24,33 @@ class FinancialService:
def __init__(self): def __init__(self):
self.price_service = PriceDataService() self.price_service = PriceDataService()
def _get_ticker_max_range(self, ticker: str) -> Tuple[datetime, datetime]: async def _get_ticker_max_range(self, ticker: str) -> Tuple[datetime, datetime]:
""" """
Get maximum date range for a ticker by checking its listing date via yfinance_plus Get maximum date range for a ticker by checking its listing date via yfinance_plus.
Runs the blocking yfinance call in a thread pool with a 20s timeout.
Args:
ticker: Stock ticker symbol
Returns: Returns:
Tuple of (listing_date, current_date) or fallback to 20 years if yfinance unavailable Tuple of (listing_date, current_date) or fallback to 20 years if yfinance unavailable
""" """
try: try:
# Try to get ticker info from yfinance_plus to find actual listing date
import yfinance_plus as yf import yfinance_plus as yf
ticker_obj = yf.Ticker(ticker) ticker_obj = yf.Ticker(ticker)
loop = asyncio.get_event_loop()
# Get a small sample of historical data to find the earliest available date hist = await asyncio.wait_for(
# Use period="max" and interval="1mo" for faster query loop.run_in_executor(
hist = ticker_obj.history(period="max", interval="1mo") None,
lambda: ticker_obj.history(period="max", interval="1mo")
),
timeout=20,
)
if not hist.empty: if not hist.empty:
# Get the earliest date from the historical data
earliest_date = hist.index[0].to_pydatetime() earliest_date = hist.index[0].to_pydatetime()
if earliest_date.tzinfo is None: if earliest_date.tzinfo is None:
earliest_date = earliest_date.replace(tzinfo=timezone.utc) earliest_date = earliest_date.replace(tzinfo=timezone.utc)
# Current date as end
end_date = datetime.now(timezone.utc).replace(hour=23, minute=59, second=59, microsecond=0) end_date = datetime.now(timezone.utc).replace(hour=23, minute=59, second=59, microsecond=0)
# Ensure we don't go beyond SEC data availability (1994)
sec_start = datetime(settings.SEC_DATA_START_YEAR, 1, 1, tzinfo=timezone.utc) sec_start = datetime(settings.SEC_DATA_START_YEAR, 1, 1, tzinfo=timezone.utc)
actual_start = max(earliest_date, sec_start) actual_start = max(earliest_date, sec_start)
@ -67,7 +65,6 @@ class FinancialService:
end_date = datetime.now(timezone.utc).replace(hour=23, minute=59, second=59, microsecond=0) end_date = datetime.now(timezone.utc).replace(hour=23, minute=59, second=59, microsecond=0)
start_date = end_date - timedelta(days=20 * 365.25) # 20 years start_date = end_date - timedelta(days=20 * 365.25) # 20 years
# Ensure we don't go beyond SEC data availability (1994)
sec_start = datetime(settings.SEC_DATA_START_YEAR, 1, 1, tzinfo=timezone.utc) sec_start = datetime(settings.SEC_DATA_START_YEAR, 1, 1, tzinfo=timezone.utc)
actual_start = max(start_date, sec_start) actual_start = max(start_date, sec_start)
@ -88,10 +85,14 @@ class FinancialService:
""" """
ticker = ticker.upper() ticker = ticker.upper()
# Resolve time parameters to standard datetime range # Resolve time parameters to standard datetime range.
# _get_ticker_max_range is async, so handle "max" period before calling
# the sync resolve_time_parameters helper.
if period and period.lower() == "max":
resolved_start, resolved_end = await self._get_ticker_max_range(ticker)
else:
resolved_start, resolved_end = resolve_time_parameters( resolved_start, resolved_end = resolve_time_parameters(
start_date, end_date, quarters, period, ticker, start_date, end_date, quarters, period, ticker
ticker_max_range_fn=self._get_ticker_max_range
) )
# Get or create company # Get or create company

@ -21,6 +21,14 @@ from app.core.config import settings
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
async def _run_with_timeout(coro, timeout_seconds: float, description: str):
try:
return await asyncio.wait_for(coro, timeout=timeout_seconds)
except asyncio.TimeoutError:
raise TimeoutError(f"yfinance timed out after {timeout_seconds}s: {description}")
# Import yfinance-plus for price data only # Import yfinance-plus for price data only
try: try:
import yfinance_plus as yf import yfinance_plus as yf
@ -184,7 +192,8 @@ class PriceDataService:
# Run yfinance-plus in executor to avoid blocking # Run yfinance-plus in executor to avoid blocking
loop = asyncio.get_event_loop() loop = asyncio.get_event_loop()
hist_data = await loop.run_in_executor( hist_data = await _run_with_timeout(
loop.run_in_executor(
None, None,
lambda: yf_ticker.history( lambda: yf_ticker.history(
start=start_str, start=start_str,
@ -194,6 +203,9 @@ class PriceDataService:
prepost=False, prepost=False,
period=None # Explicitly set period to None when using start/end dates period=None # Explicitly set period to None when using start/end dates
) )
),
timeout_seconds=30,
description=f"history {ticker} {start_str}:{end_str}"
) )
if hist_data.empty: if hist_data.empty:
@ -219,7 +231,11 @@ class PriceDataService:
try: try:
yf_ticker = yf.Ticker(ticker) yf_ticker = yf.Ticker(ticker)
loop = asyncio.get_event_loop() loop = asyncio.get_event_loop()
info = await loop.run_in_executor(None, lambda: yf_ticker.info) info = await _run_with_timeout(
loop.run_in_executor(None, lambda: yf_ticker.info),
timeout_seconds=20,
description=f"info {ticker}"
)
# Prefer regular/post/pre values # Prefer regular/post/pre values
regular = info.get("regularMarketPrice") regular = info.get("regularMarketPrice")
post = info.get("postMarketPrice") if use_prepost else None post = info.get("postMarketPrice") if use_prepost else None
@ -260,9 +276,13 @@ class PriceDataService:
try: try:
yf_ticker = yf.Ticker(ticker) yf_ticker = yf.Ticker(ticker)
loop = asyncio.get_event_loop() loop = asyncio.get_event_loop()
df = await loop.run_in_executor( df = await _run_with_timeout(
loop.run_in_executor(
None, None,
lambda: yf_ticker.history(period=period, interval=interval, auto_adjust=True, prepost=True) lambda: yf_ticker.history(period=period, interval=interval, auto_adjust=True, prepost=True)
),
timeout_seconds=30,
description=f"intraday {ticker} {period}/{interval}"
) )
candles = [] candles = []
if not df.empty: if not df.empty:
@ -291,8 +311,12 @@ class PriceDataService:
# First try daily with period=1d # First try daily with period=1d
yf_ticker = yf.Ticker(ticker) yf_ticker = yf.Ticker(ticker)
loop = asyncio.get_event_loop() loop = asyncio.get_event_loop()
daily = await loop.run_in_executor( daily = await _run_with_timeout(
loop.run_in_executor(
None, lambda: yf_ticker.history(period="1d", interval="1d", auto_adjust=True, prepost=False) None, lambda: yf_ticker.history(period="1d", interval="1d", auto_adjust=True, prepost=False)
),
timeout_seconds=30,
description=f"today_ohlc {ticker}"
) )
if daily is not None and not daily.empty: if daily is not None and not daily.empty:
ts, row = list(daily.iterrows())[-1] ts, row = list(daily.iterrows())[-1]
@ -425,7 +449,11 @@ class PriceDataService:
# Run in executor to avoid blocking # Run in executor to avoid blocking
loop = asyncio.get_event_loop() loop = asyncio.get_event_loop()
info = await loop.run_in_executor(None, lambda: yf_ticker.info) info = await _run_with_timeout(
loop.run_in_executor(None, lambda: yf_ticker.info),
timeout_seconds=20,
description=f"ticker_info {ticker}"
)
return info return info
@ -716,10 +744,12 @@ class PriceDataService:
logger.info(f"Processing chunk {i//chunk_size + 1}: {len(chunk_tickers)} tickers") logger.info(f"Processing chunk {i//chunk_size + 1}: {len(chunk_tickers)} tickers")
# Use yfinance-plus bulk download # Use yfinance-plus bulk download
bulk_data = await loop.run_in_executor( _chunk_str = ' '.join(chunk_tickers)
bulk_data = await _run_with_timeout(
loop.run_in_executor(
None, None,
lambda: yf.download( lambda: yf.download(
tickers=' '.join(chunk_tickers), tickers=_chunk_str,
start=start_str, start=start_str,
end=end_str, end=end_str,
interval=interval, interval=interval,
@ -728,6 +758,9 @@ class PriceDataService:
group_by='ticker', group_by='ticker',
threads=True # Enable multi-threading threads=True # Enable multi-threading
) )
),
timeout_seconds=60,
description=f"bulk_download {len(chunk_tickers)} tickers"
) )
# Process and store data for each ticker in the chunk # Process and store data for each ticker in the chunk

@ -3,7 +3,9 @@ SEC Filings Service: index, search, and retrieve SEC filings (8-K, 6-K, 20-F, 40
and extract exhibit content (e.g., EX-99.1 press releases). and extract exhibit content (e.g., EX-99.1 press releases).
""" """
import asyncio
import logging import logging
import time
from datetime import datetime, timezone from datetime import datetime, timezone
from typing import Dict, List, Optional, Set, Tuple from typing import Dict, List, Optional, Set, Tuple
@ -45,6 +47,8 @@ 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)
try:
cik = await self._http.get_company_cik(ticker) cik = await self._http.get_company_cik(ticker)
if not cik: if not cik:
raise ValueError(f"Could not find CIK for ticker {ticker}") raise ValueError(f"Could not find CIK for ticker {ticker}")
@ -166,6 +170,8 @@ class SECFilingsService:
await db.commit() await db.commit()
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:
self._http.clear_deadline()
# ------------------------------------------------------------------ # ------------------------------------------------------------------
# search_filings: query DB with filters # search_filings: query DB with filters
@ -235,6 +241,10 @@ class SECFilingsService:
self, db: AsyncSession, accession_number: str self, db: AsyncSession, accession_number: str
) -> List[Dict]: ) -> List[Dict]:
"""Get list of documents for a filing. Caches in documents_json column.""" """Get list of documents for a filing. Caches in documents_json column."""
_owned_deadline = self._http.remaining_time() is None
if _owned_deadline:
self._http.set_deadline(30.0)
try:
result = await db.execute( result = await db.execute(
select(SECFiling).where(SECFiling.accession_number == accession_number) select(SECFiling).where(SECFiling.accession_number == accession_number)
) )
@ -298,6 +308,9 @@ class SECFilingsService:
await db.commit() await db.commit()
return documents return documents
finally:
if _owned_deadline:
self._http.clear_deadline()
# ------------------------------------------------------------------ # ------------------------------------------------------------------
# get_exhibit_content: extract exhibit text # get_exhibit_content: extract exhibit text
@ -310,6 +323,8 @@ class SECFilingsService:
exhibit_type: str = "EX-99.1", exhibit_type: str = "EX-99.1",
) -> Dict: ) -> Dict:
"""Download and return the content of a specific exhibit.""" """Download and return the content of a specific exhibit."""
self._http.set_deadline(30.0)
try:
documents = await self.get_filing_documents(db, accession_number) documents = await self.get_filing_documents(db, accession_number)
if not documents: if not documents:
raise ValueError(f"No documents found for filing {accession_number}") raise ValueError(f"No documents found for filing {accession_number}")
@ -360,6 +375,99 @@ class SECFilingsService:
"filename": filename, "filename": filename,
"url": url, "url": url,
} }
finally:
self._http.clear_deadline()
# ------------------------------------------------------------------
# search_filings_bulk: bulk filing search for multiple tickers
# ------------------------------------------------------------------
async def search_filings_bulk(
self,
db: AsyncSession,
tickers: List[str],
form_types: Optional[Set[str]] = None,
start_date: Optional[datetime] = None,
end_date: Optional[datetime] = None,
limit_per_ticker: int = 20,
) -> List[Dict]:
"""Bulk search filings for multiple tickers. Auto-indexes missing tickers in parallel."""
t0 = time.monotonic()
tickers = [t.upper() for t in tickers]
# Identify which tickers already have indexed filings
conditions = [SECFiling.ticker.in_(tickers)]
if form_types:
conditions.append(SECFiling.form_type.in_(form_types))
count_result = await db.execute(
select(SECFiling.ticker, func.count().label("cnt"))
.where(and_(*conditions))
.group_by(SECFiling.ticker)
)
indexed_counts = {row.ticker: row.cnt for row in count_result}
missing = [t for t in tickers if indexed_counts.get(t, 0) == 0]
# Index missing tickers in parallel (Semaphore(4) to not overwhelm SEC)
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,
)
except Exception as e:
logger.warning(f"Bulk index failed for {ticker}: {e}")
await asyncio.gather(*[_index_one(t) for t in missing])
# Fetch results for all tickers from DB
results = []
for ticker in tickers:
ticker_conditions = [SECFiling.ticker == ticker]
if form_types:
ticker_conditions.append(SECFiling.form_type.in_(form_types))
if start_date:
ticker_conditions.append(SECFiling.filing_date >= start_date)
if end_date:
ticker_conditions.append(SECFiling.filing_date <= end_date)
try:
count_res = await db.execute(
select(func.count())
.select_from(SECFiling)
.where(and_(*ticker_conditions))
)
total = count_res.scalar() or 0
rows = await db.execute(
select(SECFiling)
.where(and_(*ticker_conditions))
.order_by(SECFiling.filing_date.desc())
.limit(limit_per_ticker)
)
filings = rows.scalars().all()
results.append({
"ticker": ticker,
"success": True,
"filings": list(filings),
"total_count": total,
"error": None,
})
except Exception as e:
results.append({
"ticker": ticker,
"success": False,
"filings": [],
"total_count": 0,
"error": str(e),
})
elapsed = time.monotonic() - t0
return results, elapsed
# Module-level singleton # Module-level singleton

Loading…
Cancel
Save