diff --git a/app/api/v1/endpoints/alpaca.py b/app/api/v1/endpoints/alpaca.py index 4b56722..72a9536 100644 --- a/app/api/v1/endpoints/alpaca.py +++ b/app/api/v1/endpoints/alpaca.py @@ -2,6 +2,7 @@ Alpaca Market Data endpoints — standalone price data via Alpaca API """ +import asyncio from datetime import date, datetime, timezone, timedelta from typing import Optional from zoneinfo import ZoneInfo @@ -11,6 +12,11 @@ from fastapi import APIRouter, HTTPException, Query _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. +_INTRADAY_SEMAPHORE = asyncio.Semaphore(10) + def _market_closed_for(d: date) -> bool: """Return True if the US equity market session for date d has ended.""" @@ -120,21 +126,22 @@ async def get_alpaca_intraday_multi( start_dt = datetime.combine(_start, datetime.min.time()).replace(tzinfo=timezone.utc) end_dt = datetime.combine(_end, datetime.max.time()).replace(tzinfo=timezone.utc) - try: - data = await svc.get_or_fetch_multi_bars( - symbols, start_dt, end_dt, interval, force_refresh, feed="sip" - ) - except Exception as e: - err = str(e) - detail = f"Alpaca API error: {err}" - if "502" in err or "Bad Gateway" in err: - detail = ( - f"Alpaca 502 Bad Gateway — 요청당 심볼 수 초과 가능성. " - f"내부 배치 크기: 100개/요청. 원인: {err}" + async with _INTRADAY_SEMAPHORE: + try: + data = await svc.get_or_fetch_multi_bars( + symbols, start_dt, end_dt, interval, force_refresh, feed="sip" ) - raise HTTPException(status_code=502, detail=detail) - finally: - await svc.client.close() + except Exception as e: + err = str(e) + detail = f"Alpaca API error: {err}" + if "502" in err or "Bad Gateway" in err: + detail = ( + f"Alpaca 502 Bad Gateway — 요청당 심볼 수 초과 가능성. " + f"내부 배치 크기: 100개/요청. 원인: {err}" + ) + raise HTTPException(status_code=502, detail=detail) + finally: + await svc.client.close() bars = { ticker: [ @@ -193,21 +200,22 @@ async def get_alpaca_intraday_today( start_dt = datetime.combine(today, datetime.min.time()).replace(tzinfo=timezone.utc) end_dt = datetime.combine(today, datetime.max.time()).replace(tzinfo=timezone.utc) - try: - data = await svc.get_or_fetch_multi_bars( - symbols, start_dt, end_dt, interval, force_refresh=True, feed="iex" - ) - except Exception as e: - err = str(e) - detail = f"Alpaca API error: {err}" - if "502" in err or "Bad Gateway" in err: - detail = ( - f"Alpaca 502 Bad Gateway — 요청당 심볼 수 초과 가능성. " - f"내부 배치 크기: 100개/요청. 원인: {err}" + async with _INTRADAY_SEMAPHORE: + try: + data = await svc.get_or_fetch_multi_bars( + symbols, start_dt, end_dt, interval, force_refresh=True, feed="iex" ) - raise HTTPException(status_code=502, detail=detail) - finally: - await svc.client.close() + except Exception as e: + err = str(e) + detail = f"Alpaca API error: {err}" + if "502" in err or "Bad Gateway" in err: + detail = ( + f"Alpaca 502 Bad Gateway — 요청당 심볼 수 초과 가능성. " + f"내부 배치 크기: 100개/요청. 원인: {err}" + ) + raise HTTPException(status_code=502, detail=detail) + finally: + await svc.client.close() bars = { ticker: [ diff --git a/docker-compose.yml b/docker-compose.yml index 94d0075..144b6b8 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -66,7 +66,7 @@ services: - ./yfinance_plus:/app/yfinance_plus # Mount yfinance_plus for development - ./data:/app/data # For data files restart: unless-stopped - command: ["python", "-m", "uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "18000", "--reload"] + command: ["python", "-m", "uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "18000", "--reload", "--limit-concurrency", "100"] # Frontend Application frontend: