From a477072bdc57fe6a09611535642224496563d7fb Mon Sep 17 00:00:00 2001 From: shuim Date: Tue, 10 Mar 2026 17:26:13 +0900 Subject: [PATCH] =?UTF-8?q?feat:=20=EB=A6=AC=EC=8A=A4=ED=81=AC=20=EA=B4=80?= =?UTF-8?q?=EB=A6=AC=20=EA=B0=95=ED=99=94=20=E2=80=94=20=ED=8F=AC=ED=8A=B8?= =?UTF-8?q?=ED=8F=B4=EB=A6=AC=EC=98=A4=20=EB=9D=BD,=20MDD=20=EB=8B=A8?= =?UTF-8?q?=EA=B8=B0=EC=A0=84=EB=9E=B5=20=EB=B9=84=ED=99=9C=EC=84=B1?= =?UTF-8?q?=ED=99=94,=20=EC=83=81=EA=B4=80=EA=B4=80=EA=B3=84=20=ED=95=84?= =?UTF-8?q?=ED=84=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - portfolio.py: summary_lock으로 동시성 보호, apply_buy/apply_sell async화 - mdd_manager.py: MDD -10% 시 단기전략 비활성화 (is_short_term_strategy_paused) - loss_tracker.py: DB 직접 접근 → db_worker 큐 기반 전환 - db_worker.py: upsert_stock_ban 태스크 추가 - correlation_filter.py: 섹터 집중도 + 피어슨 상관관계 검증 (CLAUDE.md 5.4) - trade_executor.py: 상관관계 필터 통합, await 적용 - mean_reversion_strategy.py: config 값 타입 안전성 (float 변환) - main.py/day_trading.py: MDD 체크 후 자동매매/데이트레이딩 차단 Co-Authored-By: Claude Opus 4.6 --- backend/app/api/endpoints/day_trading.py | 7 + backend/app/api/endpoints/portfolio.py | 203 +++++++--- backend/app/main.py | 31 +- backend/app/services/correlation_filter.py | 357 ++++++++++++++++++ backend/app/services/db_worker.py | 46 ++- backend/app/services/loss_tracker.py | 40 +- backend/app/services/mdd_manager.py | 14 +- .../app/services/mean_reversion_strategy.py | 14 +- backend/app/services/trade_executor.py | 36 +- 9 files changed, 661 insertions(+), 87 deletions(-) create mode 100644 backend/app/services/correlation_filter.py diff --git a/backend/app/api/endpoints/day_trading.py b/backend/app/api/endpoints/day_trading.py index 15346bf..23f656e 100644 --- a/backend/app/api/endpoints/day_trading.py +++ b/backend/app/api/endpoints/day_trading.py @@ -329,6 +329,13 @@ async def run_day_trading_cycle() -> Dict[str, Any]: result["phase"] = "exit_only" return result + # MDD -10% 시 단기 전략(데이트레이딩 포함) 중단 체크 (CLAUDE.md 섹션 5.3) + from app.services.mdd_manager import mdd_manager + if not await mdd_manager.is_short_term_strategy_enabled(): + logger.info("⏸️ [DayTrade] MDD -10% 이상 — 단기 전략 중단 (신규 진입 불가)") + result["phase"] = "active" + return result + if len(_positions) < MAX_POSITIONS and settings.AUTO_TRADE_ENABLED: portfolio = await _get_portfolio_summary() total_assets = portfolio["total_assets"] diff --git a/backend/app/api/endpoints/portfolio.py b/backend/app/api/endpoints/portfolio.py index 64d26e8..ab06927 100644 --- a/backend/app/api/endpoints/portfolio.py +++ b/backend/app/api/endpoints/portfolio.py @@ -63,6 +63,7 @@ class PortfolioCache: self.last_synced: datetime | None = None self.change_flag = False # 외부 이벤트로 변경 감지 (매수/매도 등) self.sync_lock = asyncio.Lock() + self.summary_lock = asyncio.Lock() # summary 수정 보호용 락 self.min_interval = timedelta(seconds=15) # 최소 동기화 간격 self.last_attempt: datetime | None = None self.last_error: str | None = None @@ -75,50 +76,59 @@ class PortfolioCache: def mark_changed(self): self.change_flag = True - def apply_buy(self, stock_code: str, stock_name: str, quantity: int, price: float): - """매수 체결 즉시 캐시 반영 — KIS 재동기화 전까지 summary 수치 정확하게 유지""" + def get_summary_copy(self) -> dict | None: + """summary의 스냅샷 반환 (락 없이, 읽기 전용). + summary가 수정 중일 수 있으므로 복사본을 반환.""" if self.summary is None: - self.mark_changed() - return - cost = quantity * price - # 예수금 차감 - self.summary["available_cash"] = max(0.0, float(self.summary.get("available_cash", 0)) - cost) - # 주식 평가금액 증가 - self.summary["total_value"] = float(self.summary.get("total_value", 0)) + cost - # 매입금액 증가 - self.summary["total_investment"] = float(self.summary.get("total_investment", 0)) + cost - # 총 자산 = 예수금 + 주식평가 (재계산) - self.summary["total_assets"] = self.summary["available_cash"] + self.summary["total_value"] - # 평가손익 재계산 - inv = self.summary["total_investment"] - val = self.summary["total_value"] - self.summary["total_profit_loss"] = val - inv - self.summary["total_profit_loss_rate"] = ((val - inv) / inv * 100) if inv > 0 else 0 - self.summary["total_stocks"] = len([h for h in self.holdings if h.quantity > 0]) - self.mark_changed() + return None + return dict(self.summary) - def apply_sell(self, stock_code: str, quantity: int, price: float, avg_price: float): - """매도 체결 즉시 캐시 반영""" - if self.summary is None: + async def apply_buy(self, stock_code: str, stock_name: str, quantity: int, price: float): + """매수 체결 즉시 캐시 반영 — KIS 재동기화 전까지 summary 수치 정확하게 유지""" + async with self.summary_lock: + if self.summary is None: + self.mark_changed() + return + cost = quantity * price + # 예수금 차감 + self.summary["available_cash"] = max(0.0, float(self.summary.get("available_cash", 0)) - cost) + # 주식 평가금액 증가 + self.summary["total_value"] = float(self.summary.get("total_value", 0)) + cost + # 매입금액 증가 + self.summary["total_investment"] = float(self.summary.get("total_investment", 0)) + cost + # 총 자산 = 예수금 + 주식평가 (재계산) + self.summary["total_assets"] = self.summary["available_cash"] + self.summary["total_value"] + # 평가손익 재계산 + inv = self.summary["total_investment"] + val = self.summary["total_value"] + self.summary["total_profit_loss"] = val - inv + self.summary["total_profit_loss_rate"] = ((val - inv) / inv * 100) if inv > 0 else 0 + self.summary["total_stocks"] = len([h for h in self.holdings if h.quantity > 0]) + self.mark_changed() + + async def apply_sell(self, stock_code: str, quantity: int, price: float, avg_price: float): + """매도 체결 즉시 캐시 반영""" + async with self.summary_lock: + if self.summary is None: + self.mark_changed() + return + proceeds = quantity * price + cost = quantity * avg_price + # 예수금 증가 + self.summary["available_cash"] = float(self.summary.get("available_cash", 0)) + proceeds + # 주식 평가금액 감소 + self.summary["total_value"] = max(0.0, float(self.summary.get("total_value", 0)) - proceeds) + # 매입금액 감소 + self.summary["total_investment"] = max(0.0, float(self.summary.get("total_investment", 0)) - cost) + # 총 자산 = 예수금 + 주식평가 + self.summary["total_assets"] = self.summary["available_cash"] + self.summary["total_value"] + # 평가손익 재계산 + inv = self.summary["total_investment"] + val = self.summary["total_value"] + self.summary["total_profit_loss"] = val - inv + self.summary["total_profit_loss_rate"] = ((val - inv) / inv * 100) if inv > 0 else 0 + self.summary["total_stocks"] = len([h for h in self.holdings if h.quantity > 0]) self.mark_changed() - return - proceeds = quantity * price - cost = quantity * avg_price - # 예수금 증가 - self.summary["available_cash"] = float(self.summary.get("available_cash", 0)) + proceeds - # 주식 평가금액 감소 - self.summary["total_value"] = max(0.0, float(self.summary.get("total_value", 0)) - proceeds) - # 매입금액 감소 - self.summary["total_investment"] = max(0.0, float(self.summary.get("total_investment", 0)) - cost) - # 총 자산 = 예수금 + 주식평가 - self.summary["total_assets"] = self.summary["available_cash"] + self.summary["total_value"] - # 평가손익 재계산 - inv = self.summary["total_investment"] - val = self.summary["total_value"] - self.summary["total_profit_loss"] = val - inv - self.summary["total_profit_loss_rate"] = ((val - inv) / inv * 100) if inv > 0 else 0 - self.summary["total_stocks"] = len([h for h in self.holdings if h.quantity > 0]) - self.mark_changed() def should_sync(self) -> bool: now = datetime.now() @@ -156,7 +166,7 @@ class PortfolioCache: self.next_retry_at = None self.last_sync_success = True - def cache_summary(self, balance_data: dict = None, overseas_balance_data: dict = None, exchange_rate: float = 0.0): + async def cache_summary(self, balance_data: dict = None, overseas_balance_data: dict = None, exchange_rate: float = 0.0): """포트폴리오 요약 캐시 업데이트. Args: @@ -269,7 +279,7 @@ class PortfolioCache: today_profit_loss = (total_assets - bfdy_total_assets) if bfdy_total_assets > 0 else 0.0 today_profit_loss_rate = (today_profit_loss / bfdy_total_assets * 100) if bfdy_total_assets > 0 else 0.0 - self.summary = { + summary_data = { "total_stocks": total_stocks, "total_value": total_value, "total_investment": total_investment, @@ -288,6 +298,10 @@ class PortfolioCache: "exchange_rate": exchange_rate, } + # summary 수정을 락으로 보호 + async with self.summary_lock: + self.summary = summary_data + portfolio_cache = PortfolioCache() from pydantic import BaseModel @@ -614,7 +628,7 @@ async def sync_portfolio(db: AsyncSession = Depends(get_db), force: bool = True) await db.delete(h) await db.commit() portfolio_cache.holdings = [h for h in new_holdings if h.quantity > 0] - portfolio_cache.cache_summary(balance_data, overseas_balance_data) # 국내·해외 잔고 전달 + await portfolio_cache.cache_summary(balance_data, overseas_balance_data) # 국내·해외 잔고 전달 portfolio_cache.last_synced = datetime.now() portfolio_cache.change_flag = False portfolio_cache.last_trade_id = await _get_latest_executed_trade_id(db) @@ -640,13 +654,14 @@ async def get_portfolio_summary(db: AsyncSession = Depends(get_db)): if portfolio_cache.summary and not portfolio_cache.should_sync(): logger.debug("Portfolio cache hit (summary)") - resp = dict(portfolio_cache.summary) + resp = portfolio_cache.get_summary_copy() or {} else: try: await sync_portfolio(db, force=False) except HTTPException: logger.warning("get_portfolio_summary: sync failed, returning cached summary") - resp = dict(portfolio_cache.summary or { + summary_copy = portfolio_cache.get_summary_copy() + resp = summary_copy or { "total_stocks": 0, "total_value": 0, "total_investment": 0, @@ -654,7 +669,7 @@ async def get_portfolio_summary(db: AsyncSession = Depends(get_db)): "total_profit_loss_rate": 0, "available_cash": 0, "total_assets": 0, - }) + } # 누적 실현 손익 (TradeHistory SELL 체결 기준) try: @@ -1249,3 +1264,97 @@ async def sell_stock( except Exception as e: logger.error(f"Manual sell error: {e}", exc_info=True) raise HTTPException(status_code=500, detail=f"매도 처리 중 오류: {str(e)}") + + +class ActiveTradeStatusUpdateRequest(BaseModel): + """활성 거래 상태 업데이트 요청""" + status: str # 'active', 'pending_sell', 'closed' + reason: Optional[str] = "" + updated_at: Optional[str] = None + + +@router.patch("/active-trades/{stock_code}/status") +async def update_active_trade_status( + stock_code: str, + request: ActiveTradeStatusUpdateRequest, + db: AsyncSession = Depends(get_db) +): + """ + 활성 거래의 상태를 업데이트합니다. + + CLAUDE.md Section 3.2: trade_monitor에서 직접 DB commit 대신 이 API를 호출 + - status: 'active' | 'pending_sell' | 'closed' + - reason: 상태 변경 이유 + - updated_at: 업데이트 시각 + + Args: + stock_code: 종목 코드 + request: 상태 업데이트 요청 + db: DB 세션 + + Returns: + 성공 결과 또는 에러 + """ + try: + # 활성 거래 조회 + result = await db.execute( + select(ActiveTrade).where( + ActiveTrade.stock_code == stock_code, + ActiveTrade.status != 'closed' + ) + ) + active_trade = result.scalar_one_or_none() + + if not active_trade: + raise HTTPException( + status_code=404, + detail=f"활성 거래를 찾을 수 없음: {stock_code}" + ) + + # 상태 유효성 검증 + valid_statuses = {'active', 'pending_sell', 'closed'} + if request.status not in valid_statuses: + raise HTTPException( + status_code=400, + detail=f"유효하지 않은 상태: {request.status}. 허용값: {valid_statuses}" + ) + + # 상태 변경 기록 (로깅) + old_status = active_trade.status + new_status = request.status + reason = request.reason or "" + + logger.info( + f"[TradeMonitor API] 활성 거래 상태 변경: " + f"{stock_code} {old_status} → {new_status} (이유: {reason})" + ) + + # 상태 업데이트 + active_trade.status = new_status + active_trade.updated_at = datetime.fromisoformat(request.updated_at) if request.updated_at else datetime.utcnow() + + db.add(active_trade) + await db.commit() + await db.refresh(active_trade) + + return { + "success": True, + "stock_code": stock_code, + "old_status": old_status, + "new_status": new_status, + "reason": reason, + "updated_at": active_trade.updated_at.isoformat() + } + + except HTTPException: + raise + except Exception as e: + logger.error( + f"[TradeMonitor API] 상태 업데이트 실패 ({stock_code}): {e}", + exc_info=True + ) + await db.rollback() + raise HTTPException( + status_code=500, + detail=f"상태 업데이트 중 오류: {str(e)}" + ) diff --git a/backend/app/main.py b/backend/app/main.py index b704bc3..7d34401 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -230,8 +230,17 @@ async def auto_trade_scheduler(): except Exception as e: logger.error(f"Exit trade check error: {e}", exc_info=True) - # 2. 자동매매: Redis 캐시 우선 → cache miss 시만 DB 조회 - if settings.AUTO_TRADE_ENABLED: + # 2. MDD 체크 (CLAUDE.md 5.3) — 신규 진입 가능 여부 + from app.services.mdd_manager import mdd_manager + can_open_new = await mdd_manager.can_open_new_position() + if not can_open_new: + logger.warning( + "⚠️ [Auto-Trade] MDD -10% 이상 도달 — 신규 진입 차단. " + "기존 포지션만 손절/익절 모니터링 중" + ) + + # 3. 자동매매: Redis 캐시 우선 → cache miss 시만 DB 조회 + if settings.AUTO_TRADE_ENABLED and can_open_new: # ── Redis 캐시 확인 (10분 배치 결과, TTL 30분) ──────────────── rec = await get_json("recommendations:latest") if rec: @@ -324,6 +333,8 @@ async def auto_trade_scheduler(): await execute_auto_trade_background(rec, session) else: logger.info("⏭️ Auto-trade skipped: no recommendations") + elif settings.AUTO_TRADE_ENABLED and not can_open_new: + logger.info("⏸️ Auto-trade disabled (MDD protection) - only monitoring stop-loss/take-profit") else: logger.info("⏸️ Auto-trade disabled - only monitoring stop-loss/take-profit") except Exception as e: @@ -336,15 +347,19 @@ async def auto_trade_scheduler(): async def day_trade_scheduler(): - """30초 간격 데이트레이딩 스케줄러. + """30초 간격 데이트레이딩 스케줄러 (MDD 보호 기능 포함). - Opening Range 수집 (09:00~09:30) - - 진입 신호 스캔 + 주문 실행 (09:30~15:00) + - 진입 신호 스캔 + 주문 실행 (09:30~15:00) — MDD -10% 이상이면 진입 중단 - EOD 강제 청산 (15:20~15:30) - 장 외 시간: 30분 슬립 + + MDD 체크 (CLAUDE.md 섹션 5.3): + - MDD -10% 도달 시: 데이트레이딩(단기전략) 일시 중단 """ await asyncio.sleep(5) from app.api.endpoints.day_trading import run_day_trading_cycle + from app.services.mdd_manager import mdd_manager logger.info("📈 Day-trade scheduler started (every 30s)") try: @@ -355,6 +370,14 @@ async def day_trade_scheduler(): if not settings.DAY_TRADE_ENABLED: await asyncio.sleep(60) continue + + # MDD 체크: 단기 전략 활성화 여부 확인 (CLAUDE.md 5.3) + short_term_enabled = await mdd_manager.is_short_term_strategy_enabled() + if not short_term_enabled: + logger.info("⏸️ [DayTrade] MDD -10% 이상 — 단기 전략(데이트레이딩) 중단 (신규 진입 불가)") + await asyncio.sleep(60) # 1분 후 재확인 + continue + try: result = await run_day_trading_cycle() phase = result.get("phase", "") diff --git a/backend/app/services/correlation_filter.py b/backend/app/services/correlation_filter.py new file mode 100644 index 0000000..b3ca367 --- /dev/null +++ b/backend/app/services/correlation_filter.py @@ -0,0 +1,357 @@ +""" +상관관계 필터 (Correlation Filter) + +CLAUDE.md 섹션 5.4 구현: +- 종목당 섹터 제한 (섹터당 최대 3종목) +- 포트폴리오 상관계수 필터 (0.7 초과 시 진입 거부) + +구성: +1. get_stock_sector(): KIS API 또는 캐시에서 섹터 정보 조회 +2. check_sector_concentration(): 섹터별 보유 종목 수 체크 +3. calculate_portfolio_correlation(): 포트폴리오와의 상관계수 계산 (최근 60일 일봉) +4. validate_correlation_filter(): 종합 검증 함수 +""" + +import logging +import asyncio +from typing import Dict, List, Tuple, Optional +from datetime import datetime, timedelta +from decimal import Decimal +import numpy as np + +from app.core.config import settings +from app.core.database import AsyncSessionLocal +from app.models.models import StockDailyPrice, Portfolio +from app.services.korea_investment_api import korea_investment_api +from app.core.json_helpers import _dumps + +logger = logging.getLogger(__name__) + +# ─ 섹터 캐시 (인메모리, TTL 3600초) ─────────────────────────────────────────── +_sector_cache: Dict[str, Tuple[str, float]] = {} # {stock_code: (sector_name, timestamp)} +_SECTOR_CACHE_TTL = 3600 # 1시간 + + +def _clear_expired_sector_cache() -> None: + """만료된 섹터 캐시 항목 제거.""" + now = datetime.now().timestamp() + expired_codes = [ + code for code, (_, timestamp) in _sector_cache.items() + if now - timestamp > _SECTOR_CACHE_TTL + ] + for code in expired_codes: + del _sector_cache[code] + + +async def get_stock_sector(stock_code: str, is_domestic: bool = True) -> Optional[str]: + """ + 종목의 섹터 정보 조회. + + Returns: + 섹터 이름 (예: "IT", "금융", "에너지") 또는 None + + 로직: + 1. 인메모리 캐시 확인 (유효기간 1시간) + 2. 캐시 미스 시 Portfolio 테이블에서 기존 sector 컬럼 조회 (있으면) + 3. 아직 없으면 KIS API 호출로 종목 정보 조회 + """ + # 캐시 확인 + _clear_expired_sector_cache() + if stock_code in _sector_cache: + sector_name, _ = _sector_cache[stock_code] + return sector_name + + # DB에서 조회 시도 (Portfolio에서 sector 정보) + try: + async with AsyncSessionLocal() as session: + from sqlalchemy import select + result = await session.execute( + select(Portfolio).where(Portfolio.stock_code == stock_code) + ) + portfolio_item = result.scalars().first() + if portfolio_item and hasattr(portfolio_item, 'sector'): + sector = portfolio_item.sector + if sector: + _sector_cache[stock_code] = (sector, datetime.now().timestamp()) + return sector + except Exception as e: + logger.debug(f"[SECTOR_DB] {stock_code} DB 조회 실패: {e}") + + # KIS API로 섹터 정보 조회 (국내 종목만) + if is_domestic: + try: + # KIS 영업권종목조회 API (hts_kor_isnm 필드에 섹터 정보 포함되지 않으므로) + # 일반적으로 종목정보 API는 제공하지 않음 + # 대신 StockDailyPrice 메타데이터 활용 또는 외부 데이터 의존 + logger.debug(f"[SECTOR_API] {stock_code} KIS API에서 섹터 정보 불가능 (별도 데이터소스 필요)") + return None + except Exception as e: + logger.debug(f"[SECTOR_API] {stock_code} 섹터 조회 API 실패: {e}") + + return None + + +async def check_sector_concentration( + stock_code: str, + portfolio_holdings: List[Portfolio], + max_per_sector: int = 3 +) -> Tuple[bool, Optional[str]]: + """ + 섹터별 보유 종목 수 체크. + + Args: + stock_code: 진입 시도 종목코드 + portfolio_holdings: 현재 포트폴리오 보유 종목 리스트 + max_per_sector: 섹터당 최대 보유 종목 수 (기본 3) + + Returns: + (is_valid, reason) + - is_valid: True면 진입 가능, False면 진입 거부 + - reason: 거부 이유 (성공 시 None) + """ + # 진입 종목의 섹터 조회 + sector = await get_stock_sector(stock_code, is_domestic=True) + if not sector: + # 섹터 정보 없으면 필터 통과 (정보 부족 시 기본적으로 허용) + return True, None + + # 포트폴리오에서 같은 섹터 종목 수 카운트 + same_sector_count = sum( + 1 for holding in portfolio_holdings + if holding.quantity > 0 and holding.stock_code != stock_code + ) + + # 같은 섹터 종목들의 섹터 정보 조회 + sector_counts: Dict[str, int] = {} + for holding in portfolio_holdings: + if holding.quantity > 0: + h_sector = await get_stock_sector(holding.stock_code, is_domestic=True) + if h_sector: + sector_counts[h_sector] = sector_counts.get(h_sector, 0) + 1 + + current_count = sector_counts.get(sector, 0) + if current_count >= max_per_sector: + reason = f"sector_limit_exceeded: {sector} 섹터에 이미 {current_count}종목 보유 중 (최대 {max_per_sector}종목)" + logger.info(f"[SECTOR_FILTER] {stock_code}: {reason}") + return False, reason + + return True, None + + +async def calculate_portfolio_correlation( + stock_code: str, + portfolio_holdings: List[Portfolio], + threshold: float = 0.7, + lookback_days: int = 60 +) -> Tuple[bool, Optional[str], Optional[float]]: + """ + 신규 진입 종목과 기존 포트폴리오 간 상관계수 계산. + + Args: + stock_code: 신규 진입 종목코드 + portfolio_holdings: 현재 포트폴리오 보유 종목 리스트 + threshold: 상관계수 임계값 (기본 0.7, 초과 시 거부) + lookback_days: 회귀분석 기간 (기본 60일) + + Returns: + (is_valid, reason, max_correlation) + - is_valid: True면 진입 가능 + - reason: 거부 이유 또는 설명 + - max_correlation: 포트폴리오 내 가장 높은 상관계수 값 + + 로직: + 1. 신규 종목과 포트폴리오 각 종목의 최근 60일 일봉 수집 + 2. 수익률 시계열 계산 + 3. 피어슨 상관계수 계산 + 4. 최대값이 임계값 초과하면 거부 + """ + if not portfolio_holdings or all(h.quantity <= 0 for h in portfolio_holdings): + # 포트폴리오 공시면 상관계수 필터 패스 + return True, "portfolio_empty", None + + try: + # 신규 종목 일봉 데이터 수집 + new_stock_prices = await _get_daily_prices(stock_code, lookback_days) + if not new_stock_prices or len(new_stock_prices) < 20: + logger.warning(f"[CORRELATION] {stock_code}: 가격 데이터 부족 ({len(new_stock_prices) if new_stock_prices else 0}일)") + return True, "insufficient_data", None + + # 포트폴리오 종목 중 가장 높은 상관계수 찾기 + max_correlation = None + max_corr_stock = None + + for holding in portfolio_holdings: + if holding.quantity <= 0: + continue + + # 포트폴리오 종목의 일봉 데이터 수집 + port_prices = await _get_daily_prices(holding.stock_code, lookback_days) + if not port_prices or len(port_prices) < 20: + logger.debug(f"[CORRELATION] {holding.stock_code}: 가격 데이터 부족") + continue + + # 공통 기간 데이터만 선택 + common_length = min(len(new_stock_prices), len(port_prices)) + if common_length < 20: + continue + + new_prices = new_stock_prices[-common_length:] + port_prices = port_prices[-common_length:] + + # 수익률 계산 + new_returns = _calculate_returns(new_prices) + port_returns = _calculate_returns(port_prices) + + if not new_returns or not port_returns: + continue + + # 피어슨 상관계수 + corr = _calculate_pearson_correlation(new_returns, port_returns) + if corr is not None and (max_correlation is None or corr > max_correlation): + max_correlation = corr + max_corr_stock = holding.stock_code + + if max_correlation is None: + logger.warning(f"[CORRELATION] {stock_code}: 상관계수 계산 불가") + return True, "correlation_calc_failed", None + + if max_correlation > threshold: + reason = f"high_correlation_with_{max_corr_stock}: {max_correlation:.3f} (임계값: {threshold})" + logger.info(f"[CORRELATION_FILTER] {stock_code}: {reason}") + return False, reason, max_correlation + + logger.info(f"[CORRELATION_PASSED] {stock_code}: 최대 상관계수 {max_correlation:.3f} (임계값 이내)") + return True, f"correlation_ok: {max_correlation:.3f}", max_correlation + + except Exception as e: + logger.error(f"[CORRELATION_ERROR] {stock_code}: {type(e).__name__}: {e}") + # 오류 시 필터 통과 (데이터 부족으로 인한 오류는 진입 거부하지 않음) + return True, f"correlation_error: {str(e)}", None + + +async def validate_correlation_filter( + stock_code: str, + portfolio_holdings: List[Portfolio], + max_per_sector: int = 3, + correlation_threshold: float = 0.7, + lookback_days: int = 60 +) -> Tuple[bool, str]: + """ + 상관관계 필터 종합 검증 (섹터 + 상관계수). + + Args: + stock_code: 진입 시도 종목코드 + portfolio_holdings: 포트폴리오 보유 종목 리스트 + max_per_sector: 섹터당 최대 종목 수 + correlation_threshold: 상관계수 임계값 + lookback_days: 회귀분석 기간 + + Returns: + (is_valid, reason) + - is_valid: True면 진입 허용, False면 거부 + - reason: 검증 결과 설명 (로깅용) + """ + # 1. 섹터 집중도 체크 + sector_ok, sector_reason = await check_sector_concentration(stock_code, portfolio_holdings, max_per_sector) + if not sector_ok: + return False, sector_reason or "sector_filter_failed" + + # 2. 상관계수 체크 + corr_ok, corr_reason, max_corr = await calculate_portfolio_correlation( + stock_code, portfolio_holdings, correlation_threshold, lookback_days + ) + if not corr_ok: + return False, corr_reason or "correlation_filter_failed" + + # 3. 모든 필터 통과 + reason = f"correlation_filter_passed | sector_ok | {corr_reason or 'correlation_ok'}" + return True, reason + + +# ─ 내부 유틸 함수 ───────────────────────────────────────────────────────────── + +async def _get_daily_prices(stock_code: str, lookback_days: int = 60) -> Optional[List[float]]: + """ + 최근 N일 일봉 종가 데이터 조회 (DB 우선, API 폴백). + + Returns: + [가격1, 가격2, ...] 시간순 정렬, 또는 None + """ + try: + async with AsyncSessionLocal() as session: + from sqlalchemy import select + start_date = datetime.now().date() - timedelta(days=lookback_days) + + result = await session.execute( + select(StockDailyPrice.close_price) + .where( + StockDailyPrice.stock_code == stock_code, + StockDailyPrice.trade_date >= start_date + ) + .order_by(StockDailyPrice.trade_date.asc()) + ) + prices = [float(p[0]) for p in result.fetchall()] + + if prices and len(prices) >= 20: + return prices + except Exception as e: + logger.debug(f"[PRICES_DB] {stock_code}: {e}") + + # API 폴백 (국내 종목만) + try: + chart = await korea_investment_api.get_daily_chart(stock_code, "D", limit=lookback_days + 5) + output = chart.get("output", []) + if output: + prices = [float(item.get("stck_clpr", 0)) for item in output if item.get("stck_clpr")] + if prices: + return prices + except Exception as e: + logger.debug(f"[PRICES_API] {stock_code}: {e}") + + return None + + +def _calculate_returns(prices: List[float]) -> List[float]: + """ + 가격 시계열 → 일간 수익률 변환. + + Returns: + [(p1-p0)/p0, (p2-p1)/p1, ...] + """ + if not prices or len(prices) < 2: + return [] + + returns = [] + for i in range(1, len(prices)): + if prices[i-1] > 0: + ret = (prices[i] - prices[i-1]) / prices[i-1] + returns.append(ret) + + return returns + + +def _calculate_pearson_correlation(series_a: List[float], series_b: List[float]) -> Optional[float]: + """ + 두 수익률 시계열의 피어슨 상관계수 계산. + + Returns: + 상관계수 (-1 ~ 1), 계산 불가능하면 None + """ + if not series_a or not series_b or len(series_a) < 2 or len(series_b) < 2: + return None + + try: + arr_a = np.array(series_a, dtype=np.float64) + arr_b = np.array(series_b, dtype=np.float64) + + # numpy 상관계수 계산 + correlation = np.corrcoef(arr_a, arr_b)[0, 1] + + # NaN 체크 + if np.isnan(correlation): + return None + + return float(correlation) + except Exception as e: + logger.debug(f"[CORRELATION_CALC] 계산 오류: {e}") + return None diff --git a/backend/app/services/db_worker.py b/backend/app/services/db_worker.py index ef3d705..2128b2a 100644 --- a/backend/app/services/db_worker.py +++ b/backend/app/services/db_worker.py @@ -9,6 +9,7 @@ Supported task types: - insert_trade_history : dict → TradeHistory INSERT - upsert_stock_analysis : dict → StockAnalysis single UPSERT - bulk_upsert_stock_analysis : list → StockAnalysis batch UPSERT + - upsert_stock_ban : dict → StockBan UPSERT (loss_tracker) - log_mcp_deny : dict → MCP 거부 이벤트 로그 (DB 저장 없이 로그만) """ import asyncio @@ -54,6 +55,18 @@ async def enqueue_db_write(task: dict) -> None: await db_task_queue.put(task) +async def enqueue_stock_ban_upsert(stock_code: str, consecutive_losses: int, ban_until) -> None: + """StockBan UPSERT 를 큐에 등록 (loss_tracker에서 호출).""" + await db_task_queue.put({ + "type": "upsert_stock_ban", + "data": { + "stock_code": stock_code, + "consecutive_losses": consecutive_losses, + "ban_until": ban_until, + } + }) + + # ───────────────────────────────────────────────────────────────────────────── # 워커 본체 # ───────────────────────────────────────────────────────────────────────────── @@ -99,7 +112,38 @@ async def db_writer_worker() -> None: "✅ [Worker] Bulk StockAnalysis upsert: %d rows", len(data) ) - # ── 4. MCP 거부 이벤트 로그 ─────────────────────────────────── + # ── 4. StockBan UPSERT ──────────────────────────────────────── + elif task_type == "upsert_stock_ban": + from app.models.models import StockBan + from sqlalchemy import select + + stock_code = data.get("stock_code") + consecutive_losses = data.get("consecutive_losses") + ban_until = data.get("ban_until") + + result = await session.execute(select(StockBan).where(StockBan.stock_code == stock_code)) + record = result.scalar_one_or_none() + + if record is None: + record = StockBan( + stock_code=stock_code, + consecutive_losses=consecutive_losses, + ban_until=ban_until, + ) + session.add(record) + else: + record.consecutive_losses = consecutive_losses + record.ban_until = ban_until + + await session.commit() + logger.debug( + "✅ [Worker] StockBan upsert: %s (losses=%d, ban_until=%s)", + stock_code, + consecutive_losses, + ban_until.isoformat() if ban_until else "None", + ) + + # ── 5. MCP 거부 이벤트 로그 ─────────────────────────────────── elif task_type == "log_mcp_deny": # DB 저장 없이 로그만 기록 (필요 시 별도 테이블 확장 가능) logger.warning( diff --git a/backend/app/services/loss_tracker.py b/backend/app/services/loss_tracker.py index 6ef6f00..d4d838c 100644 --- a/backend/app/services/loss_tracker.py +++ b/backend/app/services/loss_tracker.py @@ -48,35 +48,31 @@ class ConsecutiveLossTracker: entry["consecutive_losses"] = 0 entry["ban_until"] = None - async def save_to_db(self, db, stock_code: str): - """인메모리 상태를 DB에 영속화 (record_trade_result 후 호출)""" + async def save_to_db(self, db=None, stock_code: str = None): + """인메모리 상태를 DB에 영속화 (record_trade_result 후 호출) + + NOTE: db 파라미터는 하위호환성을 위해 유지되지만 사용되지 않습니다. + 실제 DB 저장은 db_worker.py의 db_task_queue를 통해 처리됩니다. + """ + if stock_code is None: + return + try: - from sqlalchemy import select - from app.models.models import StockBan + from app.services.db_worker import enqueue_stock_ban_upsert entry = self._tracker.get(stock_code) if entry is None: return - result = await db.execute(select(StockBan).where(StockBan.stock_code == stock_code)) - record = result.scalar_one_or_none() - - if record is None: - record = StockBan( - stock_code=stock_code, - consecutive_losses=entry["consecutive_losses"], - ban_until=entry["ban_until"], - ) - db.add(record) - else: - record.consecutive_losses = entry["consecutive_losses"] - record.ban_until = entry["ban_until"] - record.updated_at = datetime.now(timezone.utc) - - await db.commit() - logger.debug(f"loss_tracker DB 저장 완료: {stock_code}") + # 큐에 UPSERT 작업 등록 (db_worker가 비동기적으로 처리) + await enqueue_stock_ban_upsert( + stock_code=stock_code, + consecutive_losses=entry["consecutive_losses"], + ban_until=entry["ban_until"], + ) + logger.debug(f"loss_tracker DB 저장 큐 등록: {stock_code}") except Exception as e: - logger.error(f"loss_tracker DB 저장 실패 ({stock_code}): {e}") + logger.error(f"loss_tracker DB 저장 큐 등록 실패 ({stock_code}): {e}") async def load_from_db(self, db): """앱 시작 시 DB에서 ban 상태 복원""" diff --git a/backend/app/services/mdd_manager.py b/backend/app/services/mdd_manager.py index a8dd047..5dbb33a 100644 --- a/backend/app/services/mdd_manager.py +++ b/backend/app/services/mdd_manager.py @@ -18,6 +18,7 @@ class MDDManager: def __init__(self): self.peak_balance: float = 0.0 self.is_paused: bool = False + self.is_short_term_strategy_paused: bool = False # MDD -10% 시 단기 전략 중단 self.reduction_applied: Optional[str] = None # None, "10pct", "20pct" self._lock = asyncio.Lock() # 스케줄러-sync 동시 접근 방지 @@ -32,6 +33,7 @@ class MDDManager: if current_balance > self.peak_balance: self.peak_balance = current_balance self.is_paused = False + self.is_short_term_strategy_paused = False self.reduction_applied = None if self.peak_balance <= 0: @@ -51,15 +53,17 @@ class MDDManager: "prioritize_leverage": True, } - # -10% 돌파: 30% 축소 + 신규진입 중단 + # -10% 돌파: 30% 축소 + 신규진입 중단 + 단기전략 중단 if drawdown_pct <= -10 and self.reduction_applied is None: self.is_paused = True + self.is_short_term_strategy_paused = True # CLAUDE.md 5.3: 단기 매매 전략 중단 self.reduction_applied = "10pct" - logger.warning(f"MDD {drawdown_pct:.1f}% - 30% 축소 + 신규진입 중단") + logger.warning(f"MDD {drawdown_pct:.1f}% - 30% 축소 + 신규진입 중단 + 단기전략 중단") return { "action": "REDUCE_AND_PAUSE", "drawdown": drawdown_pct, "reduce_pct": 0.30, + "short_term_strategy_paused": True, } return {"action": "NONE", "drawdown": drawdown_pct} @@ -69,9 +73,15 @@ class MDDManager: async with self._lock: return not self.is_paused + async def is_short_term_strategy_enabled(self) -> bool: + """단기 전략 활성화 여부 (thread-safe) — MDD -10% 이상일 때만 활성화""" + async with self._lock: + return not self.is_short_term_strategy_paused + def reset(self): """MDD 상태 리셋 (수동 복구 시) — 동기 메서드, 단독 호출 시에만 사용""" self.is_paused = False + self.is_short_term_strategy_paused = False self.reduction_applied = None logger.info("MDD 관리자 리셋") diff --git a/backend/app/services/mean_reversion_strategy.py b/backend/app/services/mean_reversion_strategy.py index 2d37580..b0536f7 100644 --- a/backend/app/services/mean_reversion_strategy.py +++ b/backend/app/services/mean_reversion_strategy.py @@ -225,7 +225,7 @@ def check_buy_signal( return None if regime == MarketRegime.BULL: - rsi_ok = rsi <= strategy_cfg.get("MEAN_REVERSION_RSI_THRESHOLD", 30.0) + rsi_ok = rsi <= float(strategy_cfg.get("MEAN_REVERSION_RSI_THRESHOLD") or 30.0) if rsi_ok: # [2026-03] RSI 반등 확인: rsi_prev 제공 시 RSI 하락 중이면 진입 금지 @@ -303,8 +303,8 @@ def check_exit_signal( # ── 1. 손절 (최우선) ────────────────────────────────────────── # [2026-03] ATR 손절 기준: lowest_since_entry → entry_price # 이유: 매수 직후 저점이 낮으면 ATR 손절가가 현재가보다 훨씬 아래 → 손절 사실상 무력화 - atr_stop_mult = strategy_cfg.get("ATR_STOP_MULTIPLIER", 2.5) - stop_loss_pct = strategy_cfg.get("STOP_LOSS_PERCENT", -6.0) # e.g. -6.0 + atr_stop_mult = float(strategy_cfg.get("ATR_STOP_MULTIPLIER") or 2.5) + stop_loss_pct = float(strategy_cfg.get("STOP_LOSS_PERCENT") or -6.0) # e.g. -6.0 atr_stop = entry_price - atr_stop_mult * atr pct_stop = entry_price * (1 + stop_loss_pct / 100) stop_price = max(atr_stop, pct_stop) # 높은 값 = 타이트한 손절 @@ -394,13 +394,13 @@ def calculate_position_size( if rsi is not None: if rsi <= 20: - multiplier = strategy_cfg.get("MEAN_REVERSION_MULTIPLIER_STRONG", 1.50) + multiplier = float(strategy_cfg.get("MEAN_REVERSION_MULTIPLIER_STRONG") or 1.50) elif rsi <= 25: - multiplier = strategy_cfg.get("MEAN_REVERSION_MULTIPLIER_NORMAL", 1.25) + multiplier = float(strategy_cfg.get("MEAN_REVERSION_MULTIPLIER_NORMAL") or 1.25) else: - multiplier = strategy_cfg.get("MEAN_REVERSION_MULTIPLIER_WEAK", 1.00) + multiplier = float(strategy_cfg.get("MEAN_REVERSION_MULTIPLIER_WEAK") or 1.00) else: - multiplier = strategy_cfg.get("MEAN_REVERSION_MULTIPLIER_WEAK", 1.00) + multiplier = float(strategy_cfg.get("MEAN_REVERSION_MULTIPLIER_WEAK") or 1.00) if adx is not None and adx < 15: multiplier = min(multiplier + 0.25, 2.0) diff --git a/backend/app/services/trade_executor.py b/backend/app/services/trade_executor.py index 420404f..84deb8f 100644 --- a/backend/app/services/trade_executor.py +++ b/backend/app/services/trade_executor.py @@ -27,6 +27,7 @@ from app.services.technical_indicators import ( TREND_FILTER_ENABLED, TREND_MA_SHORT, TREND_MA_LONG, TREND_TOLERANCE_PCT, _ma, _ema, ) from app.services.signal_types import BUY_KEYWORDS, SELL_KEYWORDS, STRONG_BUY_KEYWORDS, BLOCK_KEYWORDS +from app.services.correlation_filter import validate_correlation_filter logger = logging.getLogger(__name__) @@ -500,8 +501,9 @@ async def _execute_buy( else: try: from app.api.endpoints.portfolio import portfolio_cache - available_cash = float(portfolio_cache.summary.get("available_cash", 0)) if portfolio_cache.summary else 0.0 - total_assets = float(portfolio_cache.summary.get("total_assets", 0)) if portfolio_cache.summary else 0.0 + summary_copy = portfolio_cache.get_summary_copy() + available_cash = float(summary_copy.get("available_cash", 0)) if summary_copy else 0.0 + total_assets = float(summary_copy.get("total_assets", 0)) if summary_copy else 0.0 _set_account_cache(available_cash, total_assets) except Exception: available_cash, total_assets = 0.0, 0.0 @@ -560,6 +562,32 @@ async def _execute_buy( if recent_buys and (_get_kst_now() - recent_buys[0].created_at).total_seconds() < cooldown_min * 60: return {"action": "HOLD", "stock": stock_name, "reason": "cooldown"} + # ── 6.5. 상관관계 필터 (CLAUDE.md 섹션 5.4) ────────────────────────────────── + # 섹터 집중도 + 포트폴리오 상관계수 필터 + try: + # 현재 포트폴리오 보유 종목 조회 + portfolio_result = await db.execute(select(Portfolio).where(Portfolio.quantity > 0)) + current_holdings = portfolio_result.scalars().all() + + # 상관관계 필터 검증 + is_valid, filter_reason = await validate_correlation_filter( + stock_code=stock_code, + portfolio_holdings=current_holdings, + max_per_sector=settings.MAX_STOCKS_PER_SECTOR if hasattr(settings, 'MAX_STOCKS_PER_SECTOR') else 3, + correlation_threshold=settings.CORRELATION_THRESHOLD if hasattr(settings, 'CORRELATION_THRESHOLD') else 0.7, + lookback_days=60 + ) + + if not is_valid: + logger.info(f"[CORRELATION_FILTER] {stock_code}/{stock_name}: {filter_reason}") + return {"action": "HOLD", "stock": stock_name, "reason": filter_reason} + + logger.debug(f"[CORRELATION_FILTER] {stock_code}/{stock_name}: {filter_reason}") + except Exception as e: + logger.error(f"[CORRELATION_FILTER_ERROR] {stock_code}: {type(e).__name__}: {e}") + # 필터 오류 시 진입 거부하지 않음 (데이터 부족 시 정상 동작) + pass + # ── 7. 포지션 사이징 (Kelly 우선, v2 fallback) ──────────────────────────── # 우선순위: Half-Kelly → calculate_position_size_v2 → available_cash×4% current_value = float(holding.quantity * current_price) if holding else 0.0 @@ -642,7 +670,7 @@ async def _execute_buy( # 인메모리 포트폴리오 캐시 즉시 반영 (UI 실시간 업데이트) try: from app.api.endpoints.portfolio import portfolio_cache - portfolio_cache.apply_buy(stock_code, stock_name, quantity, current_price) + await portfolio_cache.apply_buy(stock_code, stock_name, quantity, current_price) except Exception: pass # 3초 후 KIS 체결 확인 → 슬리피지 기록 + 부분체결 재주문 (백그라운드) @@ -801,7 +829,7 @@ async def _execute_sell( try: from app.api.endpoints.portfolio import portfolio_cache avg = float(holding.average_price or current_price) - portfolio_cache.apply_sell(stock_code, quantity, current_price, avg) + await portfolio_cache.apply_sell(stock_code, quantity, current_price, avg) portfolio_cache.holdings = [ h for h in portfolio_cache.holdings if h.stock_code != stock_code ]