diff --git a/app/api/v1/endpoints/attention.py b/app/api/v1/endpoints/attention.py index 2f1ccac..f30bb31 100644 --- a/app/api/v1/endpoints/attention.py +++ b/app/api/v1/endpoints/attention.py @@ -13,6 +13,7 @@ Admin endpoints: POST /api/v1/attention/admin/collect/gdelt/{ticker}?event_date=... """ +import asyncio import logging from datetime import date, timedelta @@ -40,6 +41,11 @@ from app.services.attention.wiki_collector import collect_wiki_pageviews logger = logging.getLogger(__name__) router = APIRouter() +# Fast-fail gate for /event/{ticker} — blocks bulk scan floods that triggered +# wiki 429 bursts and contributed to container memory pressure. +_EVENT_SEMAPHORE = asyncio.Semaphore(8) +_EVENT_SEMAPHORE_WAIT = 10 + def _entity_to_info(entity: CompanyEntityMap) -> EntityInfo: return EntityInfo( @@ -309,6 +315,24 @@ async def get_event_attention( ) -> EventAttentionResponse: ticker = ticker.upper() + try: + await asyncio.wait_for(_EVENT_SEMAPHORE.acquire(), timeout=_EVENT_SEMAPHORE_WAIT) + except asyncio.TimeoutError: + raise HTTPException( + status_code=429, + detail="서버가 바빠서 요청을 처리할 수 없습니다. 잠시 후 다시 시도하세요.", + ) + try: + return await _get_event_attention_impl(ticker, event_date, db) + finally: + _EVENT_SEMAPHORE.release() + + +async def _get_event_attention_impl( + ticker: str, + event_date: date, + db: AsyncSession, +) -> EventAttentionResponse: # 1. Get or resolve entity entity_result = await db.execute( select(CompanyEntityMap).where(CompanyEntityMap.ticker == ticker) diff --git a/app/api/v1/endpoints/filings.py b/app/api/v1/endpoints/filings.py index cb3176f..e05d3eb 100644 --- a/app/api/v1/endpoints/filings.py +++ b/app/api/v1/endpoints/filings.py @@ -43,6 +43,11 @@ logger = logging.getLogger("app.api.v1.filings") _SEARCH_SEMAPHORE = asyncio.Semaphore(8) _SEARCH_SEMAPHORE_TIMEOUT = 10 +# Lazy 8-K parsing in /events/{ticker} downloads up to 20 SEC HTML documents +# per request. A-Z bulk scans without this gate have OOM-killed the container. +_EVENTS_SEMAPHORE = asyncio.Semaphore(4) +_EVENTS_SEMAPHORE_TIMEOUT = 10 + @router.get( "/search/{ticker}", @@ -500,14 +505,28 @@ async def get_filing_events( except ValueError: raise HTTPException(status_code=400, detail="Invalid date format. Use YYYY-MM-DD") - # Lazy-parse any pending 8-Ks for this ticker (limit 20 to avoid long waits) + # Fast-fail when server is overloaded (bulk A-Z scans trigger SEC HTML downloads) try: await asyncio.wait_for( - sec_8k_parser.parse_bulk(db, tickers=[ticker_upper], limit=20), - timeout=60, + _EVENTS_SEMAPHORE.acquire(), timeout=_EVENTS_SEMAPHORE_TIMEOUT + ) + except asyncio.TimeoutError: + raise HTTPException( + status_code=429, + detail="서버가 바빠서 요청을 처리할 수 없습니다. 잠시 후 다시 시도하세요.", ) - except (asyncio.TimeoutError, Exception) as e: - logger.warning(f"Lazy parse timed out/failed for {ticker_upper}: {e}") + + try: + # Lazy-parse any pending 8-Ks for this ticker (limit 20 to avoid long waits) + try: + await asyncio.wait_for( + sec_8k_parser.parse_bulk(db, tickers=[ticker_upper], limit=20), + timeout=60, + ) + except (asyncio.TimeoutError, Exception) as e: + logger.warning(f"Lazy parse timed out/failed for {ticker_upper}: {e}") + finally: + _EVENTS_SEMAPHORE.release() # Query events conditions = [SECFilingEvent.ticker == ticker_upper] diff --git a/app/services/attention/wiki_collector.py b/app/services/attention/wiki_collector.py index e1fdeb7..4ef1f06 100644 --- a/app/services/attention/wiki_collector.py +++ b/app/services/attention/wiki_collector.py @@ -6,7 +6,9 @@ Only fetches dates not already in the database. Uses Wikimedia REST API range endpoint for efficiency. """ +import asyncio import logging +import time from datetime import date, timedelta import httpx @@ -25,6 +27,15 @@ _WIKI_HEADERS = { _LOOKBACK_DAYS = 20 _LOOKAHEAD_DAYS = 2 +# Process-wide concurrency + rate limiting for Wikimedia REST API. +# Prevents burst-induced 429 floods that correlate with container OOM. +_WIKI_SEMAPHORE = asyncio.Semaphore(2) +_WIKI_MIN_INTERVAL = 0.25 # seconds between requests (≈ 4 req/s ceiling) +_WIKI_RATE_LOCK = asyncio.Lock() +_WIKI_LAST_CALL_TS: float = 0.0 +_WIKI_429_LAST_LOG_TS: float = 0.0 +_WIKI_429_LOG_WINDOW = 30.0 # log at most once per 30 s during a burst + def _date_range(start: date, end: date) -> list[date]: """Return list of dates from start to end inclusive.""" @@ -91,15 +102,30 @@ async def collect_wiki_pageviews( logger.info("Fetching wiki pageviews for %r (%s → %s)", wiki_title, start_str, end_str) + global _WIKI_LAST_CALL_TS, _WIKI_429_LAST_LOG_TS try: - async with httpx.AsyncClient(timeout=15.0, headers=_WIKI_HEADERS) as client: - resp = await client.get(url) - resp.raise_for_status() - data = resp.json() + async with _WIKI_SEMAPHORE: + async with _WIKI_RATE_LOCK: + now = time.monotonic() + wait = _WIKI_MIN_INTERVAL - (now - _WIKI_LAST_CALL_TS) + if wait > 0: + await asyncio.sleep(wait) + _WIKI_LAST_CALL_TS = time.monotonic() + + async with httpx.AsyncClient(timeout=15.0, headers=_WIKI_HEADERS) as client: + resp = await client.get(url) + resp.raise_for_status() + data = resp.json() except httpx.HTTPStatusError as exc: - if exc.response.status_code == 404: + status = exc.response.status_code + if status == 404: logger.warning("Wikipedia page not found: %r", wiki_title) return 0 + if status == 429: + now = time.monotonic() + if now - _WIKI_429_LAST_LOG_TS > _WIKI_429_LOG_WINDOW: + logger.warning("Wikimedia rate limit 429 for %r (burst suppressed %ds)", wiki_title, int(_WIKI_429_LOG_WINDOW)) + _WIKI_429_LAST_LOG_TS = now raise except Exception as exc: logger.error("Wiki API request failed: %s", exc) diff --git a/app/services/sec_http_client.py b/app/services/sec_http_client.py index 494160c..62bb949 100644 --- a/app/services/sec_http_client.py +++ b/app/services/sec_http_client.py @@ -95,6 +95,10 @@ class SECHttpClient: self._pending: int = 0 self._text_cache: Dict[str, str] = {} self._json_cache: Dict[str, dict] = {} + # Negative cache for tickers with no SEC CIK — prevents bulk A-Z scans + # from scanning the 10MB tickers map hundreds of times per second. + self._cik_negative_cache: Dict[str, float] = {} + self._cik_negative_ttl: float = 3600.0 # 1 hour self._cache_dir = "/tmp/stock_oracle_sec_cache" try: os.makedirs(self._cache_dir, exist_ok=True) @@ -162,15 +166,25 @@ class SECHttpClient: async def get_company_cik(self, ticker: str) -> Optional[str]: """Look up zero-padded 10-digit CIK for a ticker.""" + tkr = ticker.upper() + now = _time.monotonic() + + # Negative-cache hit: don't scan 10MB dict for known-missing tickers + expiry = self._cik_negative_cache.get(tkr) + if expiry is not None and now < expiry: + return None + url = f"{self.sec_base_www}/files/company_tickers.json" try: data = await self.fetch_json(url) for _key, company_info in data.items(): - if company_info.get("ticker", "").upper() == ticker.upper(): + if company_info.get("ticker", "").upper() == tkr: cik_str = str(company_info.get("cik_str", "")).zfill(10) logger.info(f"Found CIK {cik_str} for ticker {ticker}") + self._cik_negative_cache.pop(tkr, None) return cik_str logger.warning(f"Ticker {ticker} not found in SEC mapping") + self._cik_negative_cache[tkr] = now + self._cik_negative_ttl return None except Exception as e: logger.error(f"Error fetching CIK for {ticker}: {e}") diff --git a/docker-compose.yml b/docker-compose.yml index 016fd4f..25948e0 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -65,10 +65,10 @@ services: - ./API_DOCUMENTATION.md:/app/API_DOCUMENTATION.md # API documentation - ./yfinance_plus:/app/yfinance_plus # Mount yfinance_plus for development - ./data:/app/data # For data files - mem_limit: 2g - memswap_limit: 2g + mem_limit: 3g + memswap_limit: 3g restart: unless-stopped - command: ["python", "-m", "uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "18000", "--reload", "--limit-concurrency", "25"] + command: ["python", "-m", "uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "18000", "--limit-concurrency", "25"] # Frontend Application frontend: