fix: OOM kill 방지 — Phase 4 경량화 + Semaphore 3으로 제한

Worker가 OOM kill로 죽는 것을 확인 (docker inspect: OOMKilled=true).
호스트 메모리: 물리 RAM ~68MB 여유, 스왑 97% 사용 상태.

원인: Phase 4에서 SELECT AlpacaPriceData (전체 ORM 객체) 로드.
- 75 ticker × 78 5분봉 = 5,850개 ORM 객체/요청
- SQLAlchemy ORM instrumentation으로 dict 대비 ~10x 메모리 소비
- Semaphore(10) → 최대 10개 요청 동시 처리 = 피크 수백 MB ~ 1GB

수정:
- Phase 4: SELECT * → SELECT 7 columns only (ticker, date, OHLCV)
  → result.all()로 경량 Row namedtuple 반환, ORM 객체 생성 없음
  → SQLAlchemy Row는 속성 접근(row.date, row.open 등) 지원 — 엔드포인트 호환
- Semaphore(10) → Semaphore(3): 동시 처리 3개로 제한
  → 피크 메모리 ~70% 감소

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
main
I Luk Kim 4 months ago
parent ffaed4fc9d
commit abfab0fa57

@ -15,7 +15,7 @@ _MARKET_CLOSE_HOUR = 16 # 4:00 PM ET
# Limit concurrent Alpaca intraday processing to prevent event-loop saturation # Limit concurrent Alpaca intraday processing to prevent event-loop saturation
# under bulk backfill workloads. Callers beyond this limit wait on the semaphore # under bulk backfill workloads. Callers beyond this limit wait on the semaphore
# (cheap asyncio wait) rather than flooding httpx connections and DB sessions. # (cheap asyncio wait) rather than flooding httpx connections and DB sessions.
_INTRADAY_SEMAPHORE = asyncio.Semaphore(10) _INTRADAY_SEMAPHORE = asyncio.Semaphore(3)
def _market_closed_for(d: date) -> bool: def _market_closed_for(d: date) -> bool:

@ -191,10 +191,20 @@ class AlpacaPriceService:
f"Alpaca multi-bars: stored {len(rows)} rows for {len(need_fetch)} tickers" f"Alpaca multi-bars: stored {len(rows)} rows for {len(need_fetch)} tickers"
) )
# Phase 4: read back (short session) — filtered by interval to avoid mixing 1d/5m/etc. # Phase 4: read back (short session, lightweight column query).
# Select only the 7 columns the endpoints need instead of loading full ORM
# objects — avoids SQLAlchemy instrumentation overhead (~10x lighter per row).
async with AsyncSessionLocal() as db: async with AsyncSessionLocal() as db:
result = await db.execute( result = await db.execute(
select(AlpacaPriceData) select(
AlpacaPriceData.ticker,
AlpacaPriceData.date,
AlpacaPriceData.open,
AlpacaPriceData.high,
AlpacaPriceData.low,
AlpacaPriceData.close,
AlpacaPriceData.volume,
)
.where( .where(
and_( and_(
AlpacaPriceData.ticker.in_(upper_tickers), AlpacaPriceData.ticker.in_(upper_tickers),
@ -205,10 +215,10 @@ class AlpacaPriceService:
) )
.order_by(AlpacaPriceData.ticker, AlpacaPriceData.date) .order_by(AlpacaPriceData.ticker, AlpacaPriceData.date)
) )
db_rows = result.scalars().all() db_rows = result.all() # lightweight Row namedtuples, not ORM objects
await asyncio.sleep(0) # yield after bulk ORM object creation await asyncio.sleep(0) # yield after bulk row creation
data: Dict[str, List[AlpacaPriceData]] = {t: [] for t in upper_tickers} data: Dict[str, List] = {t: [] for t in upper_tickers}
for row in db_rows: for row in db_rows:
data[row.ticker].append(row) data[row.ticker].append(row)
return data return data

Loading…
Cancel
Save