From d2829a47a72a9101cd4b7ac9bc6313fcaf4f2b7f Mon Sep 17 00:00:00 2001 From: shuim Date: Mon, 23 Feb 2026 08:37:24 +0900 Subject: [PATCH] =?UTF-8?q?=EB=82=A0=EC=A7=9C=20=EC=88=98=EC=A0=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/api/endpoints/day_trading.py | 119 +++++++++++++++++-- backend/app/main.py | 29 ++++- backend/app/services/korea_investment_api.py | 33 ++--- backend/app/services/loss_tracker.py | 2 +- backend/app/services/mdd_manager.py | 4 +- backend/app/services/ml_ensemble.py | 5 +- backend/app/services/watchlist.py | 3 +- 7 files changed, 163 insertions(+), 32 deletions(-) diff --git a/backend/app/api/endpoints/day_trading.py b/backend/app/api/endpoints/day_trading.py index 0efb586..f1dba70 100644 --- a/backend/app/api/endpoints/day_trading.py +++ b/backend/app/api/endpoints/day_trading.py @@ -45,8 +45,9 @@ from app.core.config import settings from app.core.json_helpers import _dumps from app.services.websocket_manager import websocket_manager from app.core.database import get_db, AsyncSessionLocal -from app.models.models import TradeHistory +from app.models.models import TradeHistory, ActiveTrade from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy import select, update as sa_update router = APIRouter() logger = logging.getLogger(__name__) @@ -263,8 +264,11 @@ async def run_day_trading_cycle() -> Dict[str, Any]: result["phase"] = "active" + # 사이클 단위 분봉 캐시 — 청산 체크와 신규 진입에서 동일 종목 중복 fetch 방지 + _cycle_candle_cache: Dict[str, List[MinuteCandle]] = {} + # ── 4. 활성 포지션 청산 체크 ── - exits = await _check_exits() + exits = await _check_exits(candle_cache=_cycle_candle_cache) result["exits"] = exits # ── 5. 신규 진입 신호 체크 ── @@ -281,6 +285,7 @@ async def run_day_trading_cycle() -> Dict[str, Any]: total_assets=total_assets, available_cash=available_cash, current_day_trade_amount=current_day_trade_amount, + candle_cache=_cycle_candle_cache, ) result["signals"] = [ { @@ -329,8 +334,14 @@ async def _collect_opening_range() -> None: await asyncio.sleep(0.6) -async def _check_exits() -> List[Dict[str, Any]]: - """활성 포지션 청산 조건 체크 및 실행""" +async def _check_exits( + candle_cache: Optional[Dict[str, List[MinuteCandle]]] = None, +) -> List[Dict[str, Any]]: + """활성 포지션 청산 조건 체크 및 실행 + + Args: + candle_cache: 사이클 단위 분봉 캐시 {stock_code: candles}. None이면 직접 fetch. + """ global _daily_pnl exits = [] now_time = _kst_time() @@ -345,7 +356,12 @@ async def _check_exits() -> List[Dict[str, Any]]: if current_price > pos["highest_price"]: _positions[code]["highest_price"] = current_price - candles = await _fetch_candles(code) + if candle_cache is not None and code in candle_cache: + candles = candle_cache[code] + else: + candles = await _fetch_candles(code) + if candle_cache is not None: + candle_cache[code] = candles candles_5m = aggregate_to_5min(candles) exit_signal = check_day_trade_exit( @@ -371,6 +387,12 @@ async def _check_exits() -> List[Dict[str, Any]]: db, code, pos["stock_name"], "SELL", current_price, quantity, exit_signal.reason, pnl, ) + await db.execute( + sa_update(ActiveTrade) + .where(ActiveTrade.stock_code == code, ActiveTrade.strategy_name == "day_trade", ActiveTrade.status == "active") + .values(status="closed") + ) + await db.commit() exits.append({ "stock_code": code, @@ -399,14 +421,24 @@ async def _scan_signals( total_assets: float, available_cash: float, current_day_trade_amount: float, + candle_cache: Optional[Dict[str, List[MinuteCandle]]] = None, ) -> List[Dict[str, Any]]: - """감시 종목에서 진입 신호 스캔""" + """감시 종목에서 진입 신호 스캔 + + Args: + candle_cache: 사이클 단위 분봉 캐시 {stock_code: candles}. None이면 직접 fetch. + """ results = [] for code, name in list(_watchlist.items()): if code in _positions: continue try: - candles_1m = await _fetch_candles(code) + if candle_cache is not None and code in candle_cache: + candles_1m = candle_cache[code] + else: + candles_1m = await _fetch_candles(code) + if candle_cache is not None: + candle_cache[code] = candles_1m if not candles_1m: continue candles_5m = aggregate_to_5min(candles_1m) @@ -486,6 +518,21 @@ async def _execute_entry(signal: DayTradeSignal, quantity: int) -> Optional[Dict db, signal.stock_code, signal.stock_name, "BUY", signal.entry_price, quantity, signal.reason, ) + # ActiveTrade 저장 — 재시작 시 포지션 복구를 위해 + active = ActiveTrade( + stock_code=signal.stock_code, + stock_name=signal.stock_name, + quantity=quantity, + entry_price=signal.entry_price, + stop_loss=signal.stop_loss, + take_profit=signal.target_price, + highest_since_entry=signal.entry_price, + strategy_name="day_trade", + entry_reason=signal.strategy, + status="active", + ) + db.add(active) + await db.commit() logger.info( f"🛒 데이트레이딩 진입 [{signal.stock_code}] " @@ -519,11 +566,69 @@ async def _execute_eod_flat() -> None: _daily_pnl += pnl del _positions[code] logger.info(f"🔔 EOD 강제 청산 [{code}]: PNL={pnl:+,.0f}원") + + async with AsyncSessionLocal() as db: + await _record_trade( + db, code, pos["stock_name"], "SELL", + current_price, pos["quantity"], "EOD_flat", pnl, + ) + await db.execute( + sa_update(ActiveTrade) + .where(ActiveTrade.stock_code == code, ActiveTrade.strategy_name == "day_trade", ActiveTrade.status == "active") + .values(status="closed") + ) + await db.commit() + + await websocket_manager.broadcast(_dumps({ + "type": "day_trade_exit", + "stock_code": code, + "reason": "EOD_flat", + "pnl": pnl, + })) except Exception as e: logger.error(f"EOD 청산 오류 [{code}]: {e}") await asyncio.sleep(0.6) +async def restore_positions_from_db() -> None: + """앱 재시작 후 오늘 데이트레이딩 포지션 DB 복구. + + strategy_name='day_trade', status='active'인 당일 ActiveTrade 레코드로 + 인메모리 _positions를 재구성한다. + """ + if _positions: + return # 이미 포지션 있으면 스킵 + today = _today_kst() + try: + async with AsyncSessionLocal() as db: + result = await db.execute( + select(ActiveTrade).where( + ActiveTrade.strategy_name == "day_trade", + ActiveTrade.status == "active", + ) + ) + rows = result.scalars().all() + for row in rows: + if row.created_at and row.created_at.date() != today: + continue + entry_price = float(row.entry_price) + _positions[row.stock_code] = { + "stock_name": row.stock_name or row.stock_code, + "strategy": row.entry_reason or "unknown", + "entry_price": entry_price, + "stop_loss": float(row.stop_loss) if row.stop_loss else entry_price * 0.97, + "target_price": float(row.take_profit) if row.take_profit else entry_price * 1.03, + "quantity": row.quantity, + "highest_price": float(row.highest_since_entry or row.entry_price), + "entry_time": row.created_at.strftime("%H:%M:%S") if row.created_at else "??:??:??", + "reason": row.entry_reason or "", + } + if _positions: + logger.info(f"📋 데이트레이딩 포지션 복구: {list(_positions.keys())}") + except Exception as e: + logger.error(f"데이트레이딩 포지션 복구 실패: {e}") + + # ── API 엔드포인트 ──────────────────────────────────────────────────────────── @router.get("/watchlist") diff --git a/backend/app/main.py b/backend/app/main.py index d5aea45..d5d4256 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -13,6 +13,7 @@ import asyncio import json import logging import time +import pytz from app.core.config import settings from app.api import api_router @@ -207,16 +208,18 @@ async def lifespan(app: FastAPI): from datetime import date, timedelta async with AsyncSessionLocal() as session: + _kst = pytz.timezone("Asia/Seoul") + _today_kst = datetime.now(_kst).date() # 오래된 stock_analysis 삭제 (7일 이전) - cutoff_date = date.today() - timedelta(days=7) + cutoff_date = _today_kst - timedelta(days=7) result = await session.execute( delete(StockAnalysis).where(StockAnalysis.analysis_date < cutoff_date) ) if result.rowcount > 0: logger.info(f"🗑️ Deleted {result.rowcount} old stock_analysis records") - + # PENDING/FAILED 오래된 거래 삭제 (1일 이전) - cutoff_datetime = date.today() - timedelta(days=1) + cutoff_datetime = _today_kst - timedelta(days=1) result = await session.execute( delete(TradeHistory).where( TradeHistory.status.in_(['PENDING', 'FAILED']), @@ -248,6 +251,13 @@ async def lifespan(app: FastAPI): logger.error(traceback.format_exc()) raise + # 데이트레이딩 포지션 복구 (재시작 시 인메모리 유실 방지) + try: + from app.api.endpoints.day_trading import restore_positions_from_db + await restore_positions_from_db() + except Exception as e: + logger.warning(f"⚠️ 데이트레이딩 포지션 복구 실패 (non-critical): {e}") + # 스케줄러 시작 app.state.auto_trade_task = asyncio.create_task(auto_trade_scheduler()) app.state.day_trade_task = asyncio.create_task(day_trade_scheduler()) @@ -279,9 +289,20 @@ app = FastAPI( ) # CORS 설정 (반드시 라우터 등록 전에) +# allow_origins=["*"] + allow_credentials=True 조합은 CORS 스펙 위반으로 브라우저 차단됨 +_cors_origins = getattr(settings, "CORS_ORIGINS", None) +if _cors_origins: + _allowed_origins = [o.strip() for o in _cors_origins.split(",") if o.strip()] +else: + _allowed_origins = [ + "http://localhost:5173", + "http://localhost:3000", + "http://127.0.0.1:5173", + "http://127.0.0.1:3000", + ] app.add_middleware( CORSMiddleware, - allow_origins=["*"], + allow_origins=_allowed_origins, allow_credentials=True, allow_methods=["*"], allow_headers=["*"], diff --git a/backend/app/services/korea_investment_api.py b/backend/app/services/korea_investment_api.py index 9625d54..ce65d4b 100644 --- a/backend/app/services/korea_investment_api.py +++ b/backend/app/services/korea_investment_api.py @@ -181,34 +181,35 @@ class KoreaInvestmentAPI: logger.info("🔑 새로운 토큰 발급 요청...") self._last_token_request_time = datetime.now() - - async with httpx.AsyncClient() as client: + + # HTTP 요청도 Lock 안에서 실행 — 중복 발급 방지 try: - response = await client.post( - f"{self.base_url}/oauth2/tokenP", - json={ - "grant_type": "client_credentials", - "appkey": self.app_key, - "appsecret": self.app_secret - }, - timeout=30.0 - ) - + async with httpx.AsyncClient() as client: + response = await client.post( + f"{self.base_url}/oauth2/tokenP", + json={ + "grant_type": "client_credentials", + "appkey": self.app_key, + "appsecret": self.app_secret + }, + timeout=30.0 + ) + if response.status_code == 200: data = response.json() self.access_token = data["access_token"] # 토큰 만료 시간 설정 (일반적으로 24시간) self.token_expires_at = datetime.now() + timedelta(hours=23) - - # 파일에 저장 (비동기) + + # 파일에 저장 (비동기) — Lock 안에서 호출하되 내부는 to_thread이므로 OK await self._save_token_to_cache() - + logger.info(f"✅ 새 토큰 발급 완료 (만료: {self.token_expires_at.strftime('%Y-%m-%d %H:%M')})") return self.access_token else: error_msg = response.text logger.error(f"❌ 토큰 발급 실패: {response.status_code} - {error_msg}") - + # 토큰 발급 실패 시에도 기존 토큰이 있다면 계속 사용 시도 if self.access_token: logger.warning(f"⚠️ 토큰 발급 실패했지만 기존 토큰 재사용 시도") diff --git a/backend/app/services/loss_tracker.py b/backend/app/services/loss_tracker.py index c58406f..e3687c3 100644 --- a/backend/app/services/loss_tracker.py +++ b/backend/app/services/loss_tracker.py @@ -71,7 +71,7 @@ class ConsecutiveLossTracker: else: record.consecutive_losses = entry["consecutive_losses"] record.ban_until = entry["ban_until"] - record.updated_at = datetime.utcnow() + record.updated_at = datetime.now() await db.commit() logger.debug(f"loss_tracker DB 저장 완료: {stock_code}") diff --git a/backend/app/services/mdd_manager.py b/backend/app/services/mdd_manager.py index fdfbb93..a8dd047 100644 --- a/backend/app/services/mdd_manager.py +++ b/backend/app/services/mdd_manager.py @@ -7,7 +7,7 @@ MDD(Maximum Drawdown) 관리 - CLAUDE.md 섹션 6.3 """ import asyncio import logging -from typing import Dict +from typing import Dict, Optional logger = logging.getLogger(__name__) @@ -18,7 +18,7 @@ class MDDManager: def __init__(self): self.peak_balance: float = 0.0 self.is_paused: bool = False - self.reduction_applied: str = None # None, "10pct", "20pct" + self.reduction_applied: Optional[str] = None # None, "10pct", "20pct" self._lock = asyncio.Lock() # 스케줄러-sync 동시 접근 방지 async def update(self, current_balance: float) -> Dict: diff --git a/backend/app/services/ml_ensemble.py b/backend/app/services/ml_ensemble.py index d0a6bad..5dac4a1 100644 --- a/backend/app/services/ml_ensemble.py +++ b/backend/app/services/ml_ensemble.py @@ -56,6 +56,9 @@ class StockLSTM: self.dropout = dropout self.model = None self.device = None + # 정규화 파라미터 — train_model() 또는 load() 전까지 None + self.seq_mean: Optional[np.ndarray] = None + self.seq_std: Optional[np.ndarray] = None self._build() def _build(self): @@ -148,7 +151,7 @@ class StockLSTM: def predict_proba(self, X_seq: np.ndarray) -> np.ndarray: """예측 확률 반환 (N, 3)""" torch, nn = _get_torch() - if torch is None or self.model is None: + if torch is None or self.model is None or self.seq_mean is None or self.seq_std is None: return np.full((len(X_seq), 3), 1.0 / 3) self.model.eval() diff --git a/backend/app/services/watchlist.py b/backend/app/services/watchlist.py index ba9df38..f3b4ce2 100644 --- a/backend/app/services/watchlist.py +++ b/backend/app/services/watchlist.py @@ -12,6 +12,7 @@ from datetime import datetime, date from typing import Optional import logging +import pytz from sqlalchemy import select, update from sqlalchemy.ext.asyncio import AsyncSession @@ -53,7 +54,7 @@ async def initialize_watchlist( Returns: 추가된 종목 수 """ scanner_tickers = scanner_tickers or set() - now = datetime.now() + now = datetime.now(pytz.timezone("Asia/Seoul")).replace(tzinfo=None) # 1. 기존 활성 종목 전부 비활성화 await session.execute(