fix: 로그 분석으로 찾은 버그 4건 (과다매도·수량 0 오염·LSTM 무력화·sync 충돌)

1) DT 분할매도가 ActiveTrade.quantity를 갱신하지 않아 재기동 복원 시 원래 수량이 되살아남
   → 잔량 4주인데 7주 매도되어 같은 종목 스윙 보유분 3주까지 청산됨(2026-09-23 LG씨엔에스).
2) reflect_sell이 '전략 포지션 잔량'을 Portfolio 행 잔량으로 사용 → DT 전량청산이 스윙
   보유분까지 0으로 만듦. 행 잔량은 항상 (행 수량 - 매도 수량)으로 계산.
3) LSTM 체크포인트에 numpy 배열이 들어가 weights_only=True 로드가 매번 실패 → 앙상블에서
   LSTM이 조용히 빠진 채 예측 중이었음. 저장을 list로 바꾸고, 기존 파일은 MODEL_DIR 내부일
   때만 완화 재시도.
4) 수동 매도(행 삭제)와 sync 동시 실행 시 StaleDataError로 sync 전체 실패 → 전량 매도는
   행 삭제 대신 수량 0, sync는 충돌 시 실패 기록 없이 다음 주기 재시도.
   추가로 비강제 sync는 락 점유 중이면 30초 대기 없이 캐시 반환.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-09-23 10:24:45 +09:00
parent d8550a2241
commit f32bf412f3
5 changed files with 153 additions and 11 deletions

View File

@@ -7,6 +7,7 @@
- POST /sell: 수동 매도 (청산) - POST /sell: 수동 매도 (청산)
""" """
from fastapi import APIRouter, Depends, HTTPException from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm.exc import StaleDataError
from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, func from sqlalchemy import select, func
from typing import List, Optional from typing import List, Optional
@@ -477,6 +478,11 @@ async def sync_portfolio(db: AsyncSession = Depends(get_db), force: bool = True)
if not force and not portfolio_cache.should_sync(): if not force and not portfolio_cache.should_sync():
logger.debug("Skip sync: cache fresh & no change flag") logger.debug("Skip sync: cache fresh & no change flag")
return {"message": "Skipped (cache fresh)", "cached": True} return {"message": "Skipped (cache fresh)", "cached": True}
# 다른 sync가 진행 중이면 비강제 호출은 기다리지 않고 캐시 반환
# (KIS 지연 시 요청이 30초씩 쌓여 화면 전체가 느려지던 문제, 2026-09-23)
if not force and portfolio_cache.sync_lock.locked():
return {"message": "Sync in progress (using cache)", "cached": True}
# 락 획득에 30초 타임아웃 적용 (이전 sync hang 시 데드락 방지) # 락 획득에 30초 타임아웃 적용 (이전 sync hang 시 데드락 방지)
try: try:
await asyncio.wait_for(portfolio_cache.sync_lock.acquire(), timeout=30) await asyncio.wait_for(portfolio_cache.sync_lock.acquire(), timeout=30)
@@ -798,6 +804,13 @@ async def sync_portfolio(db: AsyncSession = Depends(get_db), force: bool = True)
portfolio_cache.last_trade_id = await _get_latest_executed_trade_id(db) portfolio_cache.last_trade_id = await _get_latest_executed_trade_id(db)
logger.info(f"Portfolio synced: {len(portfolio_cache.holdings)} holdings") logger.info(f"Portfolio synced: {len(portfolio_cache.holdings)} holdings")
return {"message": "Portfolio synchronized", "count": len(portfolio_cache.holdings), "cached": False} return {"message": "Portfolio synchronized", "count": len(portfolio_cache.holdings), "cached": False}
except StaleDataError as e:
# 수동 매도(행 삭제)와 동시 실행 시 발생 — 실패로 기록하지 않고 다음 주기에 재시도
# (2026-09-22 14:33/14:42 수동 청산 중 sync 전체가 실패하던 문제)
await db.rollback()
portfolio_cache.mark_changed()
logger.warning(f"Portfolio sync 충돌(동시 매도) — 다음 주기 재시도: {e}")
return {"message": "Sync skipped (concurrent update)", "cached": True}
except Exception as e: except Exception as e:
portfolio_cache.record_failure(e) portfolio_cache.record_failure(e)
logger.error(f"Portfolio sync failed: {e}") logger.error(f"Portfolio sync failed: {e}")
@@ -1433,8 +1446,10 @@ async def sell_stock(
# ✅ Portfolio DB 즉시 반영 (매도 후 보유종목에서 즉시 제거) # ✅ Portfolio DB 즉시 반영 (매도 후 보유종목에서 즉시 제거)
if quantity >= original_holding_qty: if quantity >= original_holding_qty:
# 전량 매도 → 레코드 삭제 # 전량 매도 → 수량 0 (행 삭제는 동시 실행 중인 sync 와 StaleDataError 충돌을
await db.delete(holding) # 일으킨다. 수량 0 행은 조회에서 제외되고 다음 sync 가 KIS 기준으로 정리한다)
holding.quantity = 0
holding.current_price = Decimal(str(executed_price))
else: else:
# 분할 매도 → 수량 차감 # 분할 매도 → 수량 차감
holding.quantity -= quantity holding.quantity -= quantity

View File

@@ -284,6 +284,15 @@ async def _execute_exit(
}) })
if is_full_exit: if is_full_exit:
await enqueue_close_active_trade(stock_code, strategy_name="day_trade") await enqueue_close_active_trade(stock_code, strategy_name="day_trade")
else:
# 분할 매도 후 ActiveTrade 잔량 갱신 — 누락 시 재기동 복원에서 원래 수량이 되살아나
# 다음 청산이 보유량보다 많이 매도한다(2026-09-23 LG씨엔에스 4주 보유 중 7주 매도).
from app.services.db_worker import enqueue_active_trade_upsert
await enqueue_active_trade_upsert({
"stock_code": stock_code,
"strategy_name": "day_trade",
"quantity": total_qty - quantity,
})
# 포트폴리오 화면에서 즉시 제거/차감 # 포트폴리오 화면에서 즉시 제거/차감
from app.services.portfolio_cache_bridge import reflect_sell from app.services.portfolio_cache_bridge import reflect_sell
await reflect_sell(stock_code, quantity, price, remaining_qty=max(0, int(total_qty) - int(quantity))) await reflect_sell(stock_code, quantity, price, remaining_qty=max(0, int(total_qty) - int(quantity)))

View File

@@ -170,8 +170,10 @@ class StockLSTM:
return return
torch.save({ torch.save({
"model_state": self.model.state_dict(), "model_state": self.model.state_dict(),
"seq_mean": self.seq_mean, # numpy 배열을 그대로 저장하면 weights_only=True 로드가 거부된다
"seq_std": self.seq_std, # (2026-09-23: LSTM이 매번 로드 실패해 앙상블에서 조용히 빠져 있었음) → list 로 저장
"seq_mean": self.seq_mean.tolist() if self.seq_mean is not None else None,
"seq_std": self.seq_std.tolist() if self.seq_std is not None else None,
"config": { "config": {
"input_size": self.input_size, "input_size": self.input_size,
"hidden_size": self.hidden_size, "hidden_size": self.hidden_size,
@@ -193,9 +195,23 @@ class StockLSTM:
except TypeError: except TypeError:
# PyTorch < 2.0: weights_only 파라미터 미지원 # PyTorch < 2.0: weights_only 파라미터 미지원
checkpoint = torch.load(path, map_location=self.device) # noqa: S614 checkpoint = torch.load(path, map_location=self.device) # noqa: S614
except Exception as _wo_err:
# 구 체크포인트(numpy 배열 포함)는 weights_only=True 로 못 읽는다.
# 우리 학습 산출물(MODEL_DIR 내부)일 때만 완화해서 재시도한다.
try:
_inside = Path(path).resolve().is_relative_to(MODEL_DIR.resolve())
except Exception:
_inside = False
if not _inside:
raise
logger.warning(
f"LSTM 체크포인트 weights_only 로드 실패 → 자체 학습 산출물로 재시도: {_wo_err}"
)
checkpoint = torch.load(path, map_location=self.device, weights_only=False) # noqa: S614
self.model.load_state_dict(checkpoint["model_state"]) self.model.load_state_dict(checkpoint["model_state"])
self.seq_mean = checkpoint["seq_mean"] _mean, _std = checkpoint["seq_mean"], checkpoint["seq_std"]
self.seq_std = checkpoint["seq_std"] self.seq_mean = None if _mean is None else np.asarray(_mean, dtype=np.float32)
self.seq_std = None if _std is None else np.asarray(_std, dtype=np.float32)
self.model.eval() self.model.eval()
return True return True
except Exception as e: except Exception as e:

View File

@@ -77,8 +77,10 @@ async def reflect_sell(
) -> None: ) -> None:
"""매도 체결 즉시 Portfolio(큐) 차감 + 캐시 제거/차감 + 최근매도 기록. """매도 체결 즉시 Portfolio(큐) 차감 + 캐시 제거/차감 + 최근매도 기록.
remaining_qty: 호출측이 아는 매도 후 잔량(권장). None이면 캐시 보유량 기준으로 계산 remaining_qty: 호출측(전략 엔진)이 아는 '그 전략 포지션'의 잔량. Portfolio 행은
(캐시에 없으면 전량 매도로 간주하지 않고 DB 갱신을 건너뜀). 같은 종목의 스윙+데이트레이딩 수량이 합쳐진 값이므로, 행 잔량은 항상
`행 수량 - 매도 수량`으로 계산한다. remaining_qty는 캐시에 행이 없을 때의 폴백으로만 쓴다.
(2026-09-23: DT 전량청산이 스윙 보유분까지 0으로 만들던 버그)
""" """
try: try:
from app.services.db_worker import enqueue_portfolio_upsert from app.services.db_worker import enqueue_portfolio_upsert
@@ -86,10 +88,10 @@ async def reflect_sell(
cache = _cache() cache = _cache()
existing = next((h for h in cache.holdings if h.stock_code == stock_code), None) existing = next((h for h in cache.holdings if h.stock_code == stock_code), None)
avg = float(existing.average_price or price) if existing is not None else price avg = float(existing.average_price or price) if existing is not None else price
if remaining_qty is not None: if existing is not None:
remaining = max(0, int(remaining_qty))
elif existing is not None:
remaining = max(0, int(existing.quantity or 0) - quantity) remaining = max(0, int(existing.quantity or 0) - quantity)
elif remaining_qty is not None:
remaining = max(0, int(remaining_qty))
else: else:
remaining = None # 알 수 없음 → DB 수량은 건드리지 않음 remaining = None # 알 수 없음 → DB 수량은 건드리지 않음
if existing is not None and remaining is not None: if existing is not None and remaining is not None:

View File

@@ -0,0 +1,100 @@
"""2026-09-23 로그 기반 버그 수정 회귀 테스트."""
from decimal import Decimal
from pathlib import Path
from unittest.mock import AsyncMock, patch
import numpy as np
import pytest
from app.api.endpoints.portfolio import portfolio_cache
from app.models.models import Portfolio
from app.services import portfolio_cache_bridge as bridge
@pytest.fixture(autouse=True)
def _reset():
portfolio_cache.holdings = []
portfolio_cache.summary = {"available_cash": 0, "total_value": 0, "total_investment": 0,
"total_assets": 0, "total_profit_loss": 0,
"total_profit_loss_rate": 0, "total_stocks": 0}
portfolio_cache.recent_sells = {}
yield
@pytest.mark.asyncio
async def test_dt_full_exit_does_not_zero_swing_shares():
"""DT 전량청산(잔량 0)이 같은 종목의 스윙 보유분까지 0으로 만들면 안 된다.
2026-09-23: LG씨엔에스 스윙 8 + DT 7 = 15주. DT 전량청산 7주에
remaining_qty=0이 전달돼 Portfolio 행 전체가 0이 됐음(실제 잔고 8주).
"""
portfolio_cache.holdings = [Portfolio(
stock_code="064400", stock_name="LG씨엔에스", quantity=15,
average_price=Decimal("70400"), current_price=Decimal("70300"),
strategy_type="swing",
)]
q = AsyncMock()
with patch("app.services.db_worker.enqueue_portfolio_upsert", q):
await bridge.reflect_sell("064400", 7, 70300.0, remaining_qty=0)
assert portfolio_cache.holdings[0].quantity == 8, "스윙 보유분이 남아야 함"
assert q.await_args.args[0]["quantity"] == 8
def test_lstm_checkpoint_roundtrip_with_weights_only():
"""LSTM 체크포인트가 weights_only=True 로 로드 가능해야 한다(앙상블 무력화 방지)."""
torch = pytest.importorskip("torch")
from app.services.ml_ensemble import LSTMPredictor
p = LSTMPredictor(input_size=4)
if p.model is None:
pytest.skip("torch 모델 초기화 불가")
p.seq_mean = np.array([1.0, 2.0, 3.0, 4.0], dtype=np.float32)
p.seq_std = np.array([1.0, 1.0, 1.0, 1.0], dtype=np.float32)
import tempfile
with tempfile.TemporaryDirectory() as d:
path = Path(d) / "lstm.pt"
p.save(path)
ckpt = torch.load(path, map_location="cpu", weights_only=True) # 여기서 예외 없어야 함
assert ckpt["seq_mean"] == [1.0, 2.0, 3.0, 4.0]
p2 = LSTMPredictor(input_size=4)
assert p2.load(path) is True
assert isinstance(p2.seq_mean, np.ndarray)
np.testing.assert_allclose(p2.seq_mean, p.seq_mean)
@pytest.mark.asyncio
async def test_dt_partial_exit_updates_active_trade_quantity():
"""DT 분할매도 후 ActiveTrade.quantity가 줄어야 재기동 복원 시 과다 매도가 없다.
2026-09-23: 7주 중 3주 분할매도 후에도 DB는 7주 → 재기동 복원이 7주로 되살아나
잔량 4주인데 7주 매도(스윙 3주 침범).
"""
from app.services.daytrading import exit_service as es
from app.services.daytrading.position_manager import day_trading_state
day_trading_state.positions["064400"] = {
"stock_name": "LG씨엔에스", "strategy": "day_trade", "entry_price": 70800.0,
"stop_loss": 69000.0, "target_price": 73000.0, "quantity": 7,
"highest_price": 71200.0, "entry_time": es._kst_now(), "reason": "", "order_id": "",
}
upserts = []
async def _capture_at(data):
upserts.append(data)
with patch("app.services.trade_executor.safe_place_order",
AsyncMock(return_value={"rt_cd": "0", "output": {"ODNO": "1"}})), \
patch.object(es, "enqueue_trade_history", AsyncMock()), \
patch("app.services.db_worker.enqueue_active_trade_upsert", _capture_at), \
patch("app.services.portfolio_cache_bridge.reflect_sell", AsyncMock()):
out = await es._execute_exit("064400", day_trading_state.positions["064400"],
71200.0, reason="목표가 달성", sell_pct=0.5)
assert out is not None
assert day_trading_state.positions["064400"]["quantity"] == 4 # 인메모리
at = [u for u in upserts if u.get("stock_code") == "064400"]
assert at and at[-1]["quantity"] == 4, f"ActiveTrade 잔량 갱신 누락: {upserts}"
day_trading_state.positions.clear()