diff --git a/app/api/v1/endpoints/filings.py b/app/api/v1/endpoints/filings.py index 3de3100..7c6372f 100644 --- a/app/api/v1/endpoints/filings.py +++ b/app/api/v1/endpoints/filings.py @@ -2,15 +2,22 @@ SEC Filings endpoints: search, document listing, and exhibit extraction. """ +import asyncio import logging from datetime import datetime, timezone -from typing import Optional +from typing import List, Optional from fastapi import APIRouter, Depends, HTTPException, Query, Response from sqlalchemy.ext.asyncio import AsyncSession from app.core.database import get_db from app.schemas.filing import ( + BulkExhibitItem, + BulkExhibitRequest, + BulkExhibitResponse, + BulkFilingSearchItem, + BulkFilingSearchRequest, + BulkFilingSearchResponse, ExhibitContentResponse, FilingDocumentInfo, FilingDocumentListResponse, @@ -75,15 +82,20 @@ async def search_filings( raise HTTPException(status_code=502, detail=f"SEC indexing failed: {e}") try: - filings, total_count = await sec_filings_service.search_filings( - db, - ticker, - form_types=form_types_set, - start_date=start_dt, - end_date=end_dt, - limit=limit, - offset=offset, + filings, total_count = await asyncio.wait_for( + sec_filings_service.search_filings( + db, + ticker, + form_types=form_types_set, + start_date=start_dt, + end_date=end_dt, + limit=limit, + offset=offset, + ), + timeout=120, ) + except asyncio.TimeoutError: + raise HTTPException(status_code=504, detail="Filing search timed out after 120s.") except ValueError as e: raise HTTPException(status_code=404, detail=str(e)) except Exception as e: @@ -118,13 +130,20 @@ async def search_filings( @router.get("/documents/{accession_number}", response_model=FilingDocumentListResponse) +@with_cache(namespace="filings:documents", ttl=86400, key_params=["accession_number"]) async def get_filing_documents( accession_number: str, + response: Response, db: AsyncSession = Depends(get_db), ): """List all documents within a SEC filing.""" 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: raise HTTPException(status_code=404, detail=str(e)) except Exception as e: @@ -140,16 +159,21 @@ async def get_filing_documents( @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( accession_number: str, + response: Response, exhibit_type: str = Query("EX-99.1", description="Exhibit type (e.g. EX-99.1)"), db: AsyncSession = Depends(get_db), ): """Extract exhibit content (e.g., press release EX-99.1) from a filing.""" try: - result = await sec_filings_service.get_exhibit_content( - db, accession_number, exhibit_type + result = await asyncio.wait_for( + 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: raise HTTPException(status_code=404, detail=str(e)) except Exception as e: @@ -164,3 +188,150 @@ async def get_exhibit_content( filename=result.get("filename"), 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), + ) diff --git a/app/api/v1/endpoints/financial.py b/app/api/v1/endpoints/financial.py index 0e8fabd..718aa12 100644 --- a/app/api/v1/endpoints/financial.py +++ b/app/api/v1/endpoints/financial.py @@ -620,9 +620,15 @@ async def get_bulk_financial_data( async with semaphore: 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] - 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 successful_count = 0 diff --git a/app/api/v1/endpoints/price.py b/app/api/v1/endpoints/price.py index 16a34a9..161815f 100644 --- a/app/api/v1/endpoints/price.py +++ b/app/api/v1/endpoints/price.py @@ -448,15 +448,22 @@ async def get_bulk_price_data( } ) - # Use optimized bulk processing method - results, successful_count, failed_count = await price_service.get_multiple_tickers_data_optimized( - db=db, - tickers=request.tickers, - start_date=start_date, - end_date=end_date, - interval=request.interval, - force_refresh=request.force_refresh - ) + # Use optimized bulk processing method (300s endpoint-level timeout) + import asyncio as _asyncio + try: + results, successful_count, failed_count = await _asyncio.wait_for( + price_service.get_multiple_tickers_data_optimized( + db=db, + tickers=request.tickers, + start_date=start_date, + end_date=end_date, + interval=request.interval, + 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( results=results, diff --git a/app/schemas/filing.py b/app/schemas/filing.py index b71e10e..b4c536b 100644 --- a/app/schemas/filing.py +++ b/app/schemas/filing.py @@ -3,7 +3,7 @@ Pydantic schemas for SEC filing endpoints. """ from datetime import datetime -from typing import List, Optional +from typing import Any, Dict, List, Optional from pydantic import BaseModel, Field @@ -45,3 +45,56 @@ class ExhibitContentResponse(BaseModel): content_type: Optional[str] = None filename: 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 diff --git a/app/services/financial_service.py b/app/services/financial_service.py index 795999a..f7e4857 100644 --- a/app/services/financial_service.py +++ b/app/services/financial_service.py @@ -5,6 +5,7 @@ Real financial data service that combines price data with calculations from datetime import datetime, timezone, timedelta, date from typing import Dict, List, Optional, Tuple, Union import logging +import asyncio import numpy as np from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select, and_, or_, desc @@ -23,54 +24,50 @@ class FinancialService: def __init__(self): 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 - - Args: - ticker: Stock ticker symbol - + 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. + Returns: Tuple of (listing_date, current_date) or fallback to 20 years if yfinance unavailable """ try: - # Try to get ticker info from yfinance_plus to find actual listing date import yfinance_plus as yf - + ticker_obj = yf.Ticker(ticker) - - # Get a small sample of historical data to find the earliest available date - # Use period="max" and interval="1mo" for faster query - hist = ticker_obj.history(period="max", interval="1mo") - + loop = asyncio.get_event_loop() + hist = await asyncio.wait_for( + loop.run_in_executor( + None, + lambda: ticker_obj.history(period="max", interval="1mo") + ), + timeout=20, + ) + if not hist.empty: - # Get the earliest date from the historical data earliest_date = hist.index[0].to_pydatetime() if earliest_date.tzinfo is None: 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) - - # Ensure we don't go beyond SEC data availability (1994) sec_start = datetime(settings.SEC_DATA_START_YEAR, 1, 1, tzinfo=timezone.utc) actual_start = max(earliest_date, sec_start) - + logger.info(f"Found actual listing date for {ticker}: {actual_start.date()}") return actual_start, end_date - + except Exception as e: logger.warning(f"Could not get ticker info for {ticker}: {e}") - + # Fallback to 20-year max if yfinance_plus fails logger.info(f"Using fallback 20-year range for {ticker}") 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 - - # Ensure we don't go beyond SEC data availability (1994) + sec_start = datetime(settings.SEC_DATA_START_YEAR, 1, 1, tzinfo=timezone.utc) actual_start = max(start_date, sec_start) - + return actual_start, end_date async def get_or_create_company_data( @@ -87,12 +84,16 @@ class FinancialService: Get company data with real price-based calculations """ ticker = ticker.upper() - - # Resolve time parameters to standard datetime range - resolved_start, resolved_end = resolve_time_parameters( - start_date, end_date, quarters, period, ticker, - ticker_max_range_fn=self._get_ticker_max_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( + start_date, end_date, quarters, period, ticker + ) # Get or create company company = await self._get_or_create_company(db, ticker) diff --git a/app/services/price_data_service.py b/app/services/price_data_service.py index 8625bf3..5972a71 100644 --- a/app/services/price_data_service.py +++ b/app/services/price_data_service.py @@ -21,6 +21,14 @@ from app.core.config import settings 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 try: import yfinance_plus as yf @@ -184,16 +192,20 @@ class PriceDataService: # Run yfinance-plus in executor to avoid blocking loop = asyncio.get_event_loop() - hist_data = await loop.run_in_executor( - None, - lambda: yf_ticker.history( - start=start_str, - end=end_str, - interval=interval, - auto_adjust=True, - prepost=False, - period=None # Explicitly set period to None when using start/end dates - ) + hist_data = await _run_with_timeout( + loop.run_in_executor( + None, + lambda: yf_ticker.history( + start=start_str, + end=end_str, + interval=interval, + auto_adjust=True, + prepost=False, + 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: @@ -219,7 +231,11 @@ class PriceDataService: try: yf_ticker = yf.Ticker(ticker) 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 regular = info.get("regularMarketPrice") post = info.get("postMarketPrice") if use_prepost else None @@ -260,9 +276,13 @@ class PriceDataService: try: yf_ticker = yf.Ticker(ticker) loop = asyncio.get_event_loop() - df = await loop.run_in_executor( - None, - lambda: yf_ticker.history(period=period, interval=interval, auto_adjust=True, prepost=True) + df = await _run_with_timeout( + loop.run_in_executor( + None, + lambda: yf_ticker.history(period=period, interval=interval, auto_adjust=True, prepost=True) + ), + timeout_seconds=30, + description=f"intraday {ticker} {period}/{interval}" ) candles = [] if not df.empty: @@ -291,8 +311,12 @@ class PriceDataService: # First try daily with period=1d yf_ticker = yf.Ticker(ticker) loop = asyncio.get_event_loop() - daily = await loop.run_in_executor( - None, lambda: yf_ticker.history(period="1d", interval="1d", auto_adjust=True, prepost=False) + daily = await _run_with_timeout( + loop.run_in_executor( + 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: ts, row = list(daily.iterrows())[-1] @@ -422,11 +446,15 @@ class PriceDataService: try: yf_ticker = yf.Ticker(ticker) - + # Run in executor to avoid blocking 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 except Exception as e: @@ -716,18 +744,23 @@ class PriceDataService: logger.info(f"Processing chunk {i//chunk_size + 1}: {len(chunk_tickers)} tickers") # Use yfinance-plus bulk download - bulk_data = await loop.run_in_executor( - None, - lambda: yf.download( - tickers=' '.join(chunk_tickers), - start=start_str, - end=end_str, - interval=interval, - auto_adjust=True, - prepost=False, - group_by='ticker', - threads=True # Enable multi-threading - ) + _chunk_str = ' '.join(chunk_tickers) + bulk_data = await _run_with_timeout( + loop.run_in_executor( + None, + lambda: yf.download( + tickers=_chunk_str, + start=start_str, + end=end_str, + interval=interval, + auto_adjust=True, + prepost=False, + group_by='ticker', + 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 diff --git a/app/services/sec_filings_service.py b/app/services/sec_filings_service.py index 4ff1998..2c8cc7f 100644 --- a/app/services/sec_filings_service.py +++ b/app/services/sec_filings_service.py @@ -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). """ +import asyncio import logging +import time from datetime import datetime, timezone from typing import Dict, List, Optional, Set, Tuple @@ -45,127 +47,131 @@ class SECFilingsService: Returns the number of filings indexed (inserted or updated). """ ticker = ticker.upper() - cik = await self._http.get_company_cik(ticker) - if not cik: - raise ValueError(f"Could not find CIK for ticker {ticker}") - - cik_int = int(cik) - url = f"{self._http.sec_base_data}/submissions/CIK{cik_int:010d}.json" - data = await self._http.fetch_json(url) - - # Determine which form types to index - target_forms = form_types or self.SUPPORTED_FORM_TYPES - - # Collect filings from recent block - raw_filings: List[Dict] = [] - - def add_from_block(block: dict) -> None: - forms = block.get("form", []) - dates = block.get("filingDate", []) - accessions = block.get("accessionNumber", []) - primary_docs = block.get("primaryDocument", []) - descriptions = block.get("primaryDocDescription", []) - accepted_dates = block.get("acceptanceDateTime", []) - for i, (form, dt_str, acc) in enumerate(zip(forms, dates, accessions)): - if form not in target_forms: + self._http.set_deadline(60.0) + try: + cik = await self._http.get_company_cik(ticker) + if not cik: + raise ValueError(f"Could not find CIK for ticker {ticker}") + + cik_int = int(cik) + url = f"{self._http.sec_base_data}/submissions/CIK{cik_int:010d}.json" + data = await self._http.fetch_json(url) + + # Determine which form types to index + target_forms = form_types or self.SUPPORTED_FORM_TYPES + + # Collect filings from recent block + raw_filings: List[Dict] = [] + + def add_from_block(block: dict) -> None: + forms = block.get("form", []) + dates = block.get("filingDate", []) + accessions = block.get("accessionNumber", []) + primary_docs = block.get("primaryDocument", []) + descriptions = block.get("primaryDocDescription", []) + accepted_dates = block.get("acceptanceDateTime", []) + for i, (form, dt_str, acc) in enumerate(zip(forms, dates, accessions)): + if form not in target_forms: + continue + pri_doc = primary_docs[i] if i < len(primary_docs) else None + desc = descriptions[i] if i < len(descriptions) else None + try: + filing_date = datetime.strptime(dt_str, "%Y-%m-%d").replace( + tzinfo=timezone.utc + ) + except Exception: + continue + accepted_at = None + if i < len(accepted_dates) and accepted_dates[i]: + try: + accepted_at = datetime.fromisoformat( + accepted_dates[i].replace("Z", "+00:00") + ) + except Exception: + pass + acc_clean = acc.replace("-", "") + pri_doc_url = None + if pri_doc: + pri_doc_url = ( + f"https://www.sec.gov/Archives/edgar/data/{cik_int}/{acc_clean}/{pri_doc}" + ) + raw_filings.append({ + "ticker": ticker, + "cik": cik, + "accession_number": acc, + "form_type": form, + "filing_date": filing_date, + "accepted_at": accepted_at, + "primary_document": pri_doc, + "primary_document_url": pri_doc_url, + "filing_description": desc, + }) + + recent = data.get("filings", {}).get("recent", {}) + add_from_block(recent) + + # Fetch older yearly submission files + files_meta = data.get("filings", {}).get("files", []) or [] + for meta in files_meta[:6]: + name = meta.get("name") + if not name: continue - pri_doc = primary_docs[i] if i < len(primary_docs) else None - desc = descriptions[i] if i < len(descriptions) else None + older_url = f"{self._http.sec_base_data}/submissions/{name}" try: - filing_date = datetime.strptime(dt_str, "%Y-%m-%d").replace( - tzinfo=timezone.utc - ) + older = await self._http.fetch_json(older_url) + add_from_block(older) except Exception: continue - accepted_at = None - if i < len(accepted_dates) and accepted_dates[i]: - try: - accepted_at = datetime.fromisoformat( - accepted_dates[i].replace("Z", "+00:00") - ) - except Exception: - pass - acc_clean = acc.replace("-", "") - pri_doc_url = None - if pri_doc: - pri_doc_url = ( - f"https://www.sec.gov/Archives/edgar/data/{cik_int}/{acc_clean}/{pri_doc}" - ) - raw_filings.append({ - "ticker": ticker, - "cik": cik, - "accession_number": acc, - "form_type": form, - "filing_date": filing_date, - "accepted_at": accepted_at, - "primary_document": pri_doc, - "primary_document_url": pri_doc_url, - "filing_description": desc, - }) - recent = data.get("filings", {}).get("recent", {}) - add_from_block(recent) + if not raw_filings: + return 0 - # Fetch older yearly submission files - files_meta = data.get("filings", {}).get("files", []) or [] - for meta in files_meta[:6]: - name = meta.get("name") - if not name: - continue - older_url = f"{self._http.sec_base_data}/submissions/{name}" - try: - older = await self._http.fetch_json(older_url) - add_from_block(older) - except Exception: - continue - - if not raw_filings: - return 0 - - # Upsert into DB - indexed_count = 0 - 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) - ) - 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: - 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"], - indexed_at=now, - created_at=now, - updated_at=now, + # Upsert into DB + indexed_count = 0 + 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) ) - db.add(filing) - indexed_count += 1 + 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: + 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"], + indexed_at=now, + created_at=now, + updated_at=now, + ) + db.add(filing) + indexed_count += 1 - if indexed_count: - await db.commit() - logger.info(f"Indexed {indexed_count} filings for {ticker}") - return indexed_count + if indexed_count: + await db.commit() + logger.info(f"Indexed {indexed_count} filings for {ticker}") + return indexed_count + finally: + self._http.clear_deadline() # ------------------------------------------------------------------ # search_filings: query DB with filters @@ -235,69 +241,76 @@ class SECFilingsService: self, db: AsyncSession, accession_number: str ) -> List[Dict]: """Get list of documents for a filing. Caches in documents_json column.""" - result = await db.execute( - select(SECFiling).where(SECFiling.accession_number == accession_number) - ) - filing = result.scalar_one_or_none() - if not filing: - raise ValueError(f"Filing {accession_number} not found in database") - - # Return cached documents if available - if filing.documents_json: - return filing.documents_json - - # Fetch and parse index page - cik_int = int("".join(ch for ch in filing.cik if ch.isdigit())) - acc_clean = accession_number.replace("-", "") - index_url = ( - f"https://www.sec.gov/Archives/edgar/data/{cik_int}" - f"/{acc_clean}/{accession_number}-index.htm" - ) + _owned_deadline = self._http.remaining_time() is None + if _owned_deadline: + self._http.set_deadline(30.0) + try: + result = await db.execute( + select(SECFiling).where(SECFiling.accession_number == accession_number) + ) + filing = result.scalar_one_or_none() + if not filing: + raise ValueError(f"Filing {accession_number} not found in database") + + # Return cached documents if available + if filing.documents_json: + return filing.documents_json + + # Fetch and parse index page + cik_int = int("".join(ch for ch in filing.cik if ch.isdigit())) + acc_clean = accession_number.replace("-", "") + index_url = ( + f"https://www.sec.gov/Archives/edgar/data/{cik_int}" + f"/{acc_clean}/{accession_number}-index.htm" + ) - html = await self._http.fetch_text(index_url) - soup = BeautifulSoup(html, "html.parser") - - base_url = index_url.rsplit("/", 1)[0] - - def mk_abs(href: str) -> str: - if href.startswith("http"): - return href - if href.startswith("/"): - return f"https://www.sec.gov{href}" - return f"{base_url}/{href}" - - documents: List[Dict] = [] - for row in soup.find_all("tr"): - cells = row.find_all(["td", "th"]) - if len(cells) < 4: - continue - # Typical columns: Seq, Description, Document, Type, Size - desc = cells[1].get_text(strip=True) if len(cells) > 1 else "" - doc_cell = cells[2] - doc_type = cells[3].get_text(strip=True) if len(cells) > 3 else "" - size_text = cells[4].get_text(strip=True) if len(cells) > 4 else "" - - a_tag = doc_cell.find("a") - if not a_tag or not a_tag.get("href"): - continue - href = a_tag["href"] - filename = a_tag.get_text(strip=True) or doc_cell.get_text(strip=True) - - documents.append({ - "type": doc_type, - "description": desc, - "filename": filename, - "url": mk_abs(href), - "size": size_text, - }) + html = await self._http.fetch_text(index_url) + soup = BeautifulSoup(html, "html.parser") - # Cache to DB - if documents: - filing.documents_json = documents - filing.updated_at = datetime.now(timezone.utc) - await db.commit() + base_url = index_url.rsplit("/", 1)[0] - return documents + def mk_abs(href: str) -> str: + if href.startswith("http"): + return href + if href.startswith("/"): + return f"https://www.sec.gov{href}" + return f"{base_url}/{href}" + + documents: List[Dict] = [] + for row in soup.find_all("tr"): + cells = row.find_all(["td", "th"]) + if len(cells) < 4: + continue + # Typical columns: Seq, Description, Document, Type, Size + desc = cells[1].get_text(strip=True) if len(cells) > 1 else "" + doc_cell = cells[2] + doc_type = cells[3].get_text(strip=True) if len(cells) > 3 else "" + size_text = cells[4].get_text(strip=True) if len(cells) > 4 else "" + + a_tag = doc_cell.find("a") + if not a_tag or not a_tag.get("href"): + continue + href = a_tag["href"] + filename = a_tag.get_text(strip=True) or doc_cell.get_text(strip=True) + + documents.append({ + "type": doc_type, + "description": desc, + "filename": filename, + "url": mk_abs(href), + "size": size_text, + }) + + # Cache to DB + if documents: + filing.documents_json = documents + filing.updated_at = datetime.now(timezone.utc) + await db.commit() + + return documents + finally: + if _owned_deadline: + self._http.clear_deadline() # ------------------------------------------------------------------ # get_exhibit_content: extract exhibit text @@ -310,56 +323,151 @@ class SECFilingsService: exhibit_type: str = "EX-99.1", ) -> Dict: """Download and return the content of a specific exhibit.""" - documents = await self.get_filing_documents(db, accession_number) - if not documents: - raise ValueError(f"No documents found for filing {accession_number}") - - exhibit_type_upper = exhibit_type.upper() - # Find matching document by type - target = None - for doc in documents: - doc_type = (doc.get("type") or "").upper() - if doc_type == exhibit_type_upper: - target = doc - break - - # Fallback: try matching description - if not target: + self._http.set_deadline(30.0) + try: + documents = await self.get_filing_documents(db, accession_number) + if not documents: + raise ValueError(f"No documents found for filing {accession_number}") + + exhibit_type_upper = exhibit_type.upper() + # Find matching document by type + target = None for doc in documents: - desc = (doc.get("description") or "").upper() - if exhibit_type_upper in desc: + doc_type = (doc.get("type") or "").upper() + if doc_type == exhibit_type_upper: target = doc break - if not target: - raise ValueError( - f"Exhibit {exhibit_type} not found in filing {accession_number}. " - f"Available types: {[d.get('type') for d in documents]}" - ) + # Fallback: try matching description + if not target: + for doc in documents: + desc = (doc.get("description") or "").upper() + if exhibit_type_upper in desc: + target = doc + break + + if not target: + raise ValueError( + f"Exhibit {exhibit_type} not found in filing {accession_number}. " + f"Available types: {[d.get('type') for d in documents]}" + ) - url = target["url"] - content = await self._http.fetch_text(url) + url = target["url"] + content = await self._http.fetch_text(url) - if len(content) > MAX_EXHIBIT_SIZE: - raise ValueError( - f"Exhibit content exceeds size limit ({len(content)} bytes > {MAX_EXHIBIT_SIZE} bytes)" - ) + if len(content) > MAX_EXHIBIT_SIZE: + raise ValueError( + f"Exhibit content exceeds size limit ({len(content)} bytes > {MAX_EXHIBIT_SIZE} bytes)" + ) + + # Determine content type + filename = target.get("filename", "") + if filename.lower().endswith(".htm") or filename.lower().endswith(".html"): + content_type = "text/html" + elif filename.lower().endswith(".xml"): + content_type = "application/xml" + else: + content_type = "text/plain" + + return { + "content": content, + "content_type": content_type, + "filename": filename, + "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), + }) - # Determine content type - filename = target.get("filename", "") - if filename.lower().endswith(".htm") or filename.lower().endswith(".html"): - content_type = "text/html" - elif filename.lower().endswith(".xml"): - content_type = "application/xml" - else: - content_type = "text/plain" - - return { - "content": content, - "content_type": content_type, - "filename": filename, - "url": url, - } + elapsed = time.monotonic() - t0 + return results, elapsed # Module-level singleton