From 0c4446a6ec044de43875db258dbc24766ab5fbae Mon Sep 17 00:00:00 2001 From: I Luk Kim Date: Sat, 18 Apr 2026 00:53:46 -0700 Subject: [PATCH] =?UTF-8?q?fix:=20=EB=8C=80=EB=9F=89=20=EC=8A=A4=EC=BA=94?= =?UTF-8?q?=20=EC=8B=9C=20=EC=84=9C=EB=B2=84=20=EC=82=AC=EB=A7=9D=20?= =?UTF-8?q?=EB=B0=A9=EC=A7=80=20=E2=80=94=20=EC=84=B8=EB=A7=88=ED=8F=AC?= =?UTF-8?q?=EC=96=B4=20fast-fail=20+=20limit-concurrency=20=EC=A1=B0?= =?UTF-8?q?=EC=A0=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Alpaca intraday: 세마포어 대기 10초 제한, 초과 시 429 즉시 반환 (기존: 120초 대기하며 커넥션 슬롯 점유 → 이벤트 루프 포화) - Filing search: 세마포어(8) + 10초 fast-fail 추가 (기존: 동시성 제한 없이 A-Z 스캔 시 수백 개 요청 쌓임) - limit-concurrency 50→25 (실제 처리량은 세마포어가 제한하므로 여유 슬롯 불필요) Co-Authored-By: Claude Sonnet 4.6 --- app/api/v1/endpoints/alpaca.py | 62 ++++++++++++++++++---------- app/api/v1/endpoints/filings.py | 72 +++++++++++++++++++++------------ docker-compose.yml | 2 +- 3 files changed, 88 insertions(+), 48 deletions(-) diff --git a/app/api/v1/endpoints/alpaca.py b/app/api/v1/endpoints/alpaca.py index 3d91bd3..d193d0c 100644 --- a/app/api/v1/endpoints/alpaca.py +++ b/app/api/v1/endpoints/alpaca.py @@ -14,10 +14,12 @@ _ET = ZoneInfo("America/New_York") _MARKET_CLOSE_HOUR = 16 # 4:00 PM ET # Limit concurrent Alpaca intraday processing to prevent event-loop saturation -# under bulk backfill workloads. Callers beyond this limit wait on the semaphore -# (cheap asyncio wait) rather than flooding httpx connections and DB sessions. -# Semaphore(5): BaseHTTPMiddleware removed → task count halved → safe to allow 5. +# under bulk backfill workloads. _INTRADAY_SEMAPHORE = asyncio.Semaphore(5) +# Max seconds a request waits for a semaphore slot before returning 429. +# Prevents hundreds of requests piling up (each holding a connection slot) +# during A-Z full-ticker scans. +_SEMAPHORE_WAIT_TIMEOUT = 10 def _market_closed_for(d: date) -> bool: @@ -129,16 +131,25 @@ async def get_alpaca_intraday_multi( end_dt = datetime.combine(_end, datetime.max.time()).replace(tzinfo=timezone.utc) async def _do_fetch(): - async with _INTRADAY_SEMAPHORE: + try: + await asyncio.wait_for( + _INTRADAY_SEMAPHORE.acquire(), timeout=_SEMAPHORE_WAIT_TIMEOUT + ) + except asyncio.TimeoutError: + raise HTTPException( + status_code=429, + detail="서버가 바빠서 요청을 처리할 수 없습니다. 잠시 후 다시 시도하세요.", + ) + try: + return await svc.get_or_fetch_multi_bars( + symbols, start_dt, end_dt, interval, force_refresh, feed="sip" + ) + finally: + _INTRADAY_SEMAPHORE.release() try: - return await svc.get_or_fetch_multi_bars( - symbols, start_dt, end_dt, interval, force_refresh, feed="sip" - ) - finally: - try: - await svc.client.close() - except Exception: - pass + await svc.client.close() + except Exception: + pass try: data = await asyncio.wait_for(_do_fetch(), timeout=120) @@ -219,16 +230,25 @@ async def get_alpaca_intraday_today( end_dt = datetime.combine(today, datetime.max.time()).replace(tzinfo=timezone.utc) async def _do_fetch(): - async with _INTRADAY_SEMAPHORE: + try: + await asyncio.wait_for( + _INTRADAY_SEMAPHORE.acquire(), timeout=_SEMAPHORE_WAIT_TIMEOUT + ) + except asyncio.TimeoutError: + raise HTTPException( + status_code=429, + detail="서버가 바빠서 요청을 처리할 수 없습니다. 잠시 후 다시 시도하세요.", + ) + try: + return await svc.get_or_fetch_multi_bars( + symbols, start_dt, end_dt, interval, force_refresh=True, feed="iex" + ) + finally: + _INTRADAY_SEMAPHORE.release() try: - return await svc.get_or_fetch_multi_bars( - symbols, start_dt, end_dt, interval, force_refresh=True, feed="iex" - ) - finally: - try: - await svc.client.close() - except Exception: - pass + await svc.client.close() + except Exception: + pass try: data = await asyncio.wait_for(_do_fetch(), timeout=120) diff --git a/app/api/v1/endpoints/filings.py b/app/api/v1/endpoints/filings.py index e4647d8..cb3176f 100644 --- a/app/api/v1/endpoints/filings.py +++ b/app/api/v1/endpoints/filings.py @@ -37,6 +37,12 @@ from sqlalchemy import select, and_, func as sql_func router = APIRouter() logger = logging.getLogger("app.api.v1.filings") +# Limit concurrent filing search requests to prevent event-loop saturation +# during bulk A-Z ticker scans. Requests beyond this wait up to +# _SEARCH_SEMAPHORE_TIMEOUT seconds, then fail fast with 429. +_SEARCH_SEMAPHORE = asyncio.Semaphore(8) +_SEARCH_SEMAPHORE_TIMEOUT = 10 + @router.get( "/search/{ticker}", @@ -89,37 +95,51 @@ async def search_filings( except ValueError: raise HTTPException(status_code=400, detail="Invalid date format. Use YYYY-MM-DD") - if force_refresh: + # Fast-fail when server is overloaded (bulk A-Z scans) + try: + await asyncio.wait_for( + _SEARCH_SEMAPHORE.acquire(), timeout=_SEARCH_SEMAPHORE_TIMEOUT + ) + except asyncio.TimeoutError: + raise HTTPException( + status_code=429, + detail="서버가 바빠서 요청을 처리할 수 없습니다. 잠시 후 다시 시도하세요.", + ) + + try: + if force_refresh: + try: + await sec_filings_service.index_filings( + db, ticker, form_types=form_types_set, force_refresh=True + ) + except ValueError as e: + raise HTTPException(status_code=404, detail=str(e)) + except Exception as e: + logger.error(f"Force refresh indexing failed for {ticker}: {e}") + raise HTTPException(status_code=502, detail=f"SEC indexing failed: {e}") + try: - await sec_filings_service.index_filings( - db, ticker, form_types=form_types_set, force_refresh=True + 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: - logger.error(f"Force refresh indexing failed for {ticker}: {e}") - raise HTTPException(status_code=502, detail=f"SEC indexing failed: {e}") - - try: - 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: - logger.error(f"Filing search failed for {ticker}: {e}") - raise HTTPException(status_code=502, detail=f"Filing search failed: {e}") + logger.error(f"Filing search failed for {ticker}: {e}") + raise HTTPException(status_code=502, detail=f"Filing search failed: {e}") + finally: + _SEARCH_SEMAPHORE.release() summaries = [] for f in filings: diff --git a/docker-compose.yml b/docker-compose.yml index 606ef50..016fd4f 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -68,7 +68,7 @@ services: mem_limit: 2g memswap_limit: 2g restart: unless-stopped - command: ["python", "-m", "uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "18000", "--reload", "--limit-concurrency", "50"] + command: ["python", "-m", "uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "18000", "--reload", "--limit-concurrency", "25"] # Frontend Application frontend: