""" 거래 성과 분석 서비스 [역할] - TradeHistory 레코드를 분석해 전략 성과 통계를 생성 - 전략별 / 요일별 승률, profit factor, 패턴 분석 - 리스크 조정 수익률 지표 (Sharpe, Sortino, Calmar) - 전략 드리프트 감지 (롤링 승률, 롤링 Sharpe, 연속 손실, MDD 가속) - 완료 거래 대상 매매일지 템플릿 자동 생성 - journals.py / analysis.py 엔드포인트에서 사용 """ import logging import math from collections import defaultdict from datetime import datetime, timedelta, date from typing import Any, Dict, List, Optional from decimal import Decimal from sqlalchemy import select, and_, func from sqlalchemy.ext.asyncio import AsyncSession from app.models.models import TradeHistory, TradeJournal, StockAnalysis, Portfolio from app.core.timeutil import now_kst logger = logging.getLogger(__name__) # ───────────────────────────────────────────────────────────────────────────── # 내부 헬퍼 # ───────────────────────────────────────────────────────────────────────────── def _to_float(val) -> float: if val is None: return 0.0 try: return float(val) except (TypeError, ValueError): return 0.0 def _parse_strategy(entry_reason: Optional[str], strategy_type: Optional[str] = None) -> str: """strategy_type 컬럼 우선, 없으면 entry_reason 문자열에서 전략 이름 추출.""" # strategy_type 컬럼이 있으면 우선 사용 if strategy_type and strategy_type.strip(): st = strategy_type.strip().lower() _type_map = { "swing": "mean_reversion", "daytrading": "day_trade", "kosdaq_scan": "kosdaq_scanner", "short_term": "short_term", } return _type_map.get(st, st) if not entry_reason: return "unknown" reason = entry_reason.lower() if "mean_reversion" in reason or "평균회귀" in reason: return "mean_reversion" if "short_term" in reason or "단기" in reason: return "short_term" if "kosdaq" in reason or "코스닥" in reason: return "kosdaq_scanner" if "day_trade" in reason or "데이트레이딩" in reason: return "day_trade" if "rule" in reason or "룰" in reason: return "rule_based" return "unknown" def _parse_exit_type(exit_reason: Optional[str]) -> str: """exit_reason 문자열에서 청산 유형 추출.""" if not exit_reason: return "unknown" reason = exit_reason.lower() # 시간손절(DT "시간 손절: N분 경과…", 스윙 "시간손절 …")은 손실 한도 도달이 아니라 # 보유 시간 초과 후 조기 정리다 — "손절" 글자만 보고 stop_loss 로 묶지 않는다(2026-10-07). if "time_stop" in reason or "시간 손절" in reason or "시간손절" in reason: return "time_stop" if "stop_loss" in reason or "손절" in reason: return "stop_loss" if "trailing" in reason or "트레일링" in reason: return "trailing_stop" if "take_profit" in reason or "익절" in reason: return "take_profit" if "eod" in reason or "장마감" in reason or "강제" in reason: return "eod_forced" if "manual" in reason or "수동" in reason: return "manual" if "rsi" in reason: return "rsi_exit" # [2026-10-09 R1] 기존 분류는 그대로 두고, 그동안 unknown 이던 사유만 새 키워드로 분리한다. if "목표가" in reason or "target" in reason: return "take_profit" if "미청산" in reason or "stale" in reason: return "stale_close" if "bb상단" in reason or "bb_upper" in reason or "과매수" in reason: return "bb_upper_exit" return "unknown" # ───────────────────────────────────────────────────────────────────────────── # 비용 추정 (2026-10-09 R1) # 운영 trade_history 의 commission IS NULL 인 SELL 은 net_profit_loss 가 비용 미차감(총손익)이다. # 정의는 trade_helpers.calculate_trade_costs 와 동일(수수료×2 + 매도세). 무거운 import 를 피하려고 # settings 요율을 직접 참조한다(테스트로 동일성 고정). # ───────────────────────────────────────────────────────────────────────────── def _is_gross_basis_v2(t) -> bool: """commission 이 NULL 이면 비용이 net_profit_loss 에 반영되지 않은 행으로 본다.""" return getattr(t, "commission", None) is None def _estimated_cost_v2(t) -> float: """commission 이 없는 SELL 의 왕복 비용 추정(원). commission 이 있으면 0(이미 차감됨).""" if not _is_gross_basis_v2(t): return 0.0 from app.core.config import settings amount = _to_float(getattr(t, "total_amount", None)) if amount <= 0: amount = _to_float(getattr(t, "executed_price", None)) * _trade_qty_v2(t) if amount <= 0: return 0.0 market = str(getattr(t, "market", None) or settings.DAY_TRADE_DEFAULT_TAX_MARKET) tax = (settings.KR_SELL_TAX_RATE_KOSDAQ if market.upper() == "KOSDAQ" else settings.KR_SELL_TAX_RATE_KOSPI) return round(amount * (2 * settings.KR_COMMISSION_RATE + tax), 0) def _adjusted_net_v2(t) -> float: return _to_float(getattr(t, "net_profit_loss", None)) - _estimated_cost_v2(t) def _cost_adjustment_v2(trades: List[Any]) -> Dict[str, Any]: gross_rows = [t for t in trades if _is_gross_basis_v2(t)] est = sum(_estimated_cost_v2(t) for t in gross_rows) net = sum(_to_float(getattr(t, "net_profit_loss", None)) for t in trades) return { "gross_basis_trades": len(gross_rows), "estimated_cost": round(est, 2), "estimated_cost_adjusted_net": round(net - est, 2), "cost_basis_note": "commission NULL 행은 비용 미차감(총손익)으로 보고 추정 비용을 차감한 값을 병기", } # ───────────────────────────────────────────────────────────────────────────── # 리스크 조정 수익률 지표 헬퍼 # ───────────────────────────────────────────────────────────────────────────── # 무위험이자율 3% (연), 일별 = 0.03/252 _RISK_FREE_ANNUAL = 0.03 _RISK_FREE_DAILY = _RISK_FREE_ANNUAL / 252.0 # 연환산 계수 _ANNUALIZE_FACTOR = math.sqrt(252) # 최소 거래일 수 (통계적 유의성) _MIN_TRADING_DAYS = 20 def _build_daily_pnl_series(trades: List["TradeHistory"]) -> Dict[date, float]: """SELL 거래의 created_at 날짜별 net_profit_loss를 집계하여 일별 PnL 딕셔너리 반환. Returns: {date(2026,3,1): 15000.0, date(2026,3,2): -3200.0, ...} """ daily_pnl: Dict[date, float] = defaultdict(float) for t in trades: if t.created_at is None: continue day = t.created_at.date() if isinstance(t.created_at, datetime) else t.created_at daily_pnl[day] += _to_float(t.net_profit_loss) return dict(daily_pnl) def _daily_pnl_to_returns(daily_pnl: Dict[date, float], initial_capital: float = 10_000_000.0) -> List[float]: """일별 PnL을 누적 equity curve로 변환 후 일별 수익률(%) 시계열 반환. Args: daily_pnl: 날짜별 PnL 금액 initial_capital: 기준 자본금 (기본 1000만원, equity curve 계산용) Returns: 일별 수익률 리스트 (소수 형태, 예: 0.01 = 1%) """ if not daily_pnl: return [] sorted_days = sorted(daily_pnl.keys()) equity = initial_capital daily_returns = [] for day in sorted_days: pnl = daily_pnl[day] if equity <= 0: # 파산 방지 — 잔여 자본이 0 이하이면 중단 break ret = pnl / equity daily_returns.append(ret) equity += pnl return daily_returns def _calc_sharpe_ratio(daily_returns: List[float]) -> Optional[float]: """Sharpe Ratio = (R - Rf) / sigma(R) * sqrt(252) 일별 수익률 시계열로부터 연환산 Sharpe Ratio를 계산. 거래일 _MIN_TRADING_DAYS 미만이면 None 반환. """ if len(daily_returns) < _MIN_TRADING_DAYS: return None n = len(daily_returns) mean_r = sum(daily_returns) / n excess = mean_r - _RISK_FREE_DAILY # 표본 표준편차 variance = sum((r - mean_r) ** 2 for r in daily_returns) / (n - 1) std = math.sqrt(variance) if variance > 0 else 0.0 if std == 0: return None return round((excess / std) * _ANNUALIZE_FACTOR, 4) def _calc_sortino_ratio(daily_returns: List[float]) -> Optional[float]: """Sortino Ratio = (R - Rf) / sigma_down(R) * sqrt(252) 하방 편차(downside deviation)만 사용 — 수익률이 무위험이자율 미만인 경우만 패널티. """ if len(daily_returns) < _MIN_TRADING_DAYS: return None n = len(daily_returns) mean_r = sum(daily_returns) / n excess = mean_r - _RISK_FREE_DAILY # 하방 편차: Rf 미만인 수익률에 대해서만 편차 계산 downside_diffs = [(r - _RISK_FREE_DAILY) ** 2 for r in daily_returns if r < _RISK_FREE_DAILY] if not downside_diffs: # 하방이 전혀 없으면 Sortino 무한대 — None 대신 높은 값 표시 return 99.0 if excess > 0 else None downside_variance = sum(downside_diffs) / n # 전체 N으로 나눔 (하방 편차 관행) downside_std = math.sqrt(downside_variance) if downside_std == 0: return None return round((excess / downside_std) * _ANNUALIZE_FACTOR, 4) def _calc_calmar_ratio(daily_returns: List[float]) -> Optional[float]: """Calmar Ratio = CAGR / |MDD| 일별 수익률로부터 CAGR과 MDD를 계산. """ if len(daily_returns) < _MIN_TRADING_DAYS: return None # Equity curve 구성 (1.0 기준) equity = 1.0 equities = [equity] for r in daily_returns: equity *= (1.0 + r) equities.append(equity) final_equity = equities[-1] n_days = len(daily_returns) # CAGR (연환산 복합 성장률) if final_equity <= 0 or n_days == 0: return None cagr = (final_equity ** (252.0 / n_days)) - 1.0 # MDD 계산 peak = equities[0] max_dd = 0.0 for eq in equities: if eq > peak: peak = eq dd = (eq - peak) / peak if peak > 0 else 0.0 if dd < max_dd: max_dd = dd if max_dd == 0: # drawdown이 없으면 — 전체 수익만 발생 return 99.0 if cagr > 0 else None return round(cagr / abs(max_dd), 4) def _calc_risk_adjusted_metrics(trades: List["TradeHistory"]) -> Dict[str, Any]: """Sharpe, Sortino, Calmar Ratio를 한 번에 계산하여 딕셔너리 반환. Returns: { "sharpe_ratio": float | None, "sortino_ratio": float | None, "calmar_ratio": float | None, "daily_returns_count": int } """ daily_pnl = _build_daily_pnl_series(trades) daily_returns = _daily_pnl_to_returns(daily_pnl) return { "sharpe_ratio": _calc_sharpe_ratio(daily_returns), "sortino_ratio": _calc_sortino_ratio(daily_returns), "calmar_ratio": _calc_calmar_ratio(daily_returns), "daily_returns_count": len(daily_returns), } # ───────────────────────────────────────────────────────────────────────────── # 전략 드리프트 감지 헬퍼 # ───────────────────────────────────────────────────────────────────────────── _DRIFT_ROLLING_WINDOW = 20 # 최근 20거래 / 20거래일 _DRIFT_WIN_RATE_THRESHOLD = 0.30 # 30% 미만 경고 _DRIFT_SHARPE_THRESHOLD = 0.0 # 0 이하 경고 _DRIFT_LOSS_STREAK_THRESHOLD = 5 # 5연패 이상 경고 _DRIFT_DRAWDOWN_THRESHOLD = -0.05 # -5% 이하 경고 def _detect_drift_alerts(trades: List["TradeHistory"]) -> List[Dict[str, Any]]: """전략 드리프트 감지 — 4가지 항목을 점검하여 경고 리스트 반환. 감지 항목: 1. rolling_win_rate: 최근 20거래 승률 < 30% 2. rolling_sharpe: 최근 20거래일 Sharpe <= 0 3. loss_streak: 최근 거래 중 5연패 이상 4. mdd_acceleration: 최근 일별 drawdown이 -5% 이하 Returns: [{"type": str, "value": float, "threshold": float, "message": str}, ...] """ alerts: List[Dict[str, Any]] = [] if not trades: return alerts pl_rates = [_to_float(t.profit_loss_rate) for t in trades] # ── 1. 롤링 승률 하락 ───────────────────────────────────────────────── recent_rates = pl_rates[-_DRIFT_ROLLING_WINDOW:] if len(recent_rates) >= _DRIFT_ROLLING_WINDOW: recent_wins = sum(1 for r in recent_rates if r > 0) rolling_wr = recent_wins / len(recent_rates) if rolling_wr < _DRIFT_WIN_RATE_THRESHOLD: alerts.append({ "type": "rolling_win_rate", "value": round(rolling_wr, 4), "threshold": _DRIFT_WIN_RATE_THRESHOLD, "message": f"최근 {_DRIFT_ROLLING_WINDOW}거래 승률 {rolling_wr*100:.1f}% (임계값 {_DRIFT_WIN_RATE_THRESHOLD*100:.0f}%)", }) # ── 2. 롤링 Sharpe 하락 ─────────────────────────────────────────────── daily_pnl = _build_daily_pnl_series(trades) daily_returns = _daily_pnl_to_returns(daily_pnl) recent_returns = daily_returns[-_DRIFT_ROLLING_WINDOW:] if len(recent_returns) >= _DRIFT_ROLLING_WINDOW: rolling_sharpe = _calc_sharpe_ratio(recent_returns) if rolling_sharpe is not None and rolling_sharpe <= _DRIFT_SHARPE_THRESHOLD: alerts.append({ "type": "rolling_sharpe", "value": rolling_sharpe, "threshold": _DRIFT_SHARPE_THRESHOLD, "message": f"최근 {_DRIFT_ROLLING_WINDOW}거래일 Sharpe {rolling_sharpe:.2f} (임계값 {_DRIFT_SHARPE_THRESHOLD:.1f})", }) # ── 3. 연속 손실 스트릭 ──────────────────────────────────────────────── # 최근 거래부터 역순으로 현재 연속 손실 카운트 current_streak = 0 for r in reversed(pl_rates): if r < 0: current_streak += 1 else: break if current_streak >= _DRIFT_LOSS_STREAK_THRESHOLD: alerts.append({ "type": "loss_streak", "value": current_streak, "threshold": _DRIFT_LOSS_STREAK_THRESHOLD, "message": f"현재 {current_streak}연패 진행 중 (임계값 {_DRIFT_LOSS_STREAK_THRESHOLD}연패)", }) # ── 4. MDD 가속 (최근 drawdown 체크) ────────────────────────────────── if len(daily_returns) >= 5: # 최근 equity curve에서 drawdown 계산 equity = 1.0 equities = [equity] for r in daily_returns: equity *= (1.0 + r) equities.append(equity) # 최근 5거래일의 drawdown 확인 recent_eq = equities[-6:] # 최근 5일분 + 직전 1일 if len(recent_eq) >= 2: recent_peak = max(recent_eq) recent_low = min(recent_eq) if recent_peak > 0: recent_dd = (recent_low - recent_peak) / recent_peak if recent_dd <= _DRIFT_DRAWDOWN_THRESHOLD: alerts.append({ "type": "mdd_acceleration", "value": round(recent_dd, 4), "threshold": _DRIFT_DRAWDOWN_THRESHOLD, "message": f"최근 drawdown {recent_dd*100:.2f}% (임계값 {_DRIFT_DRAWDOWN_THRESHOLD*100:.1f}%)", }) return alerts # ───────────────────────────────────────────────────────────────────────────── # 핵심 분석 함수 # ───────────────────────────────────────────────────────────────────────────── async def _resolve_stock_names(db: AsyncSession, codes: List[str]) -> Dict[str, str]: """trade_history.stock_name 이 비어 있는 종목의 종목명을 보강 조회. stock_analysis → portfolio 순으로 조회하며, 없으면 키를 만들지 않는다(호출부에서 코드로 대체). """ names: Dict[str, str] = {} if not codes: return names try: rows = await db.execute( select(StockAnalysis.stock_code, StockAnalysis.stock_name) .where(StockAnalysis.stock_code.in_(codes)) ) for code, name in rows.all(): if name and code not in names: names[code] = name remaining = [c for c in codes if c not in names] if remaining: rows = await db.execute( select(Portfolio.stock_code, Portfolio.stock_name) .where(Portfolio.stock_code.in_(remaining)) ) for code, name in rows.all(): if name and code not in names: names[code] = name except Exception as e: # 종목명은 부가 정보이므로 실패해도 분석은 계속 logger.warning(f"종목명 보강 조회 실패: {e}") return names async def analyze_performance(db: AsyncSession, days: int = 30) -> Dict[str, Any]: """ 최근 N일 완료 거래 성과 분석. Returns: { "summary": {전체 통계}, "by_strategy": {전략별 성과}, "by_weekday": {요일별 성과}, "by_exit_type": {청산 유형별 통계}, "worst_stocks": [최대 손실 종목 Top5 {stock_code, stock_name, net_pnl, trade_count}], "best_stocks": [최대 수익 종목 Top5 {stock_code, stock_name, net_pnl, trade_count}], "sharpe_ratio": float | None (연환산, Rf=3%), "sortino_ratio": float | None (하방편차만), "calmar_ratio": float | None (CAGR / |MDD|), "daily_returns_count": int (일별 수익률 데이터 수), "drift_alerts": [{type, value, threshold, message}, ...], "period_days": N, "generated_at": ISO 문자열 } """ since = now_kst() - timedelta(days=days) stmt = select(TradeHistory).where( and_( TradeHistory.order_type == "SELL", TradeHistory.status.in_(["COMPLETED", "FILLED"]), TradeHistory.created_at >= since, ) ).order_by(TradeHistory.created_at) result = await db.execute(stmt) trades: List[TradeHistory] = result.scalars().all() if not trades: return { "summary": _empty_summary(), "by_strategy": {}, "by_weekday": {}, "by_exit_type": {}, "worst_stocks": [], "best_stocks": [], "sharpe_ratio": None, "sortino_ratio": None, "calmar_ratio": None, "daily_returns_count": 0, "drift_alerts": [], "period_days": days, "total_trades": 0, "generated_at": now_kst().isoformat(), } # ── 전체 통계 ────────────────────────────────────────────────────────── pl_rates = [_to_float(t.profit_loss_rate) for t in trades] pl_amounts = [_to_float(t.net_profit_loss) for t in trades] wins = [r for r in pl_rates if r > 0] losses = [r for r in pl_rates if r < 0] win_rate = len(wins) / len(pl_rates) if pl_rates else 0.0 avg_win = sum(wins) / len(wins) if wins else 0.0 avg_loss = sum(losses) / len(losses) if losses else 0.0 gross_profit = sum(a for a in pl_amounts if a > 0) gross_loss = sum(a for a in pl_amounts if a < 0) net_pnl = gross_profit + gross_loss # [2026-10-01] PF 는 **금액** 기준이다(gross_profit / |gross_loss|). 이전에는 손익률(%) # 합계로 계산해 gross_profit 266,370 / gross_loss -430,619 인데 PF 1.2165 가 나왔다 # (소액 거래의 큰 %가 대금 큰 거래의 작은 %를 압도). 손실 없음 → inf(아래서 99 로 캡), # 수익도 손실도 없음 → 0. if gross_loss < 0: profit_factor = gross_profit / abs(gross_loss) elif gross_profit > 0: profit_factor = float("inf") else: profit_factor = 0.0 # 최대 연속 손실 max_consecutive_loss = _calc_max_consecutive_loss(pl_rates) summary = { "total_trades": len(trades), "win_trades": len(wins), "loss_trades": len(losses), "win_rate": round(win_rate * 100, 2), "avg_win_pct": round(avg_win, 4), "avg_loss_pct": round(avg_loss, 4), "profit_factor": round(min(profit_factor, 99.0), 4), "gross_profit": round(gross_profit, 2), "gross_loss": round(gross_loss, 2), "net_pnl": round(net_pnl, 2), "max_consecutive_loss": max_consecutive_loss, "risk_reward_ratio": round(abs(avg_win / avg_loss), 4) if avg_loss != 0 else 0.0, } summary.update(_cost_adjustment_v2(trades)) # ── 전략별 성과 ────────────────────────────────────────────────────── by_strategy: Dict[str, list] = {} for t in trades: strat = _parse_strategy(t.entry_reason, getattr(t, "strategy_type", None)) by_strategy.setdefault(strat, []).append(_to_float(t.profit_loss_rate)) strategy_stats = {} for strat, rates in by_strategy.items(): w = [r for r in rates if r > 0] strategy_stats[strat] = { "total": len(rates), "win_rate": round(len(w) / len(rates) * 100, 2) if rates else 0.0, "avg_return": round(sum(rates) / len(rates), 4) if rates else 0.0, } # ── 요일별 성과 ────────────────────────────────────────────────────── weekday_names = ["월", "화", "수", "목", "금", "토", "일"] by_weekday: Dict[str, list] = {} for t in trades: wday = weekday_names[t.created_at.weekday()] if t.created_at else "알수없음" by_weekday.setdefault(wday, []).append(_to_float(t.profit_loss_rate)) weekday_stats = { d: { "total": len(rates), "win_rate": round(len([r for r in rates if r > 0]) / len(rates) * 100, 2), "avg_return": round(sum(rates) / len(rates), 4), } for d, rates in by_weekday.items() if rates } # ── 청산 유형별 ────────────────────────────────────────────────────── by_exit: Dict[str, list] = {} for t in trades: etype = _parse_exit_type(t.exit_reason) by_exit.setdefault(etype, []).append(_to_float(t.profit_loss_rate)) exit_stats = { etype: { "count": len(rates), "avg_return": round(sum(rates) / len(rates), 4), "win_rate": round(len([r for r in rates if r > 0]) / len(rates) * 100, 2), } for etype, rates in by_exit.items() if rates } # ── 종목별 집계 ────────────────────────────────────────────────────── stock_pnl: Dict[str, float] = {} stock_names: Dict[str, str] = {} stock_trade_count: Dict[str, int] = {} for t in trades: code = t.stock_code stock_pnl[code] = stock_pnl.get(code, 0.0) + _to_float(t.net_profit_loss) stock_trade_count[code] = stock_trade_count.get(code, 0) + 1 # 종목명은 최신 거래 기준(trades는 created_at 오름차순)으로 갱신 if t.stock_name: stock_names[code] = t.stock_name # trade_history.stock_name 이 비어 있는 종목은 stock_analysis 에서 보강 missing = [c for c in stock_pnl if c not in stock_names] if missing: stock_names.update(await _resolve_stock_names(db, missing)) def _stock_row(code: str, pnl: float) -> Dict[str, Any]: return { "stock_code": code, "stock_name": stock_names.get(code) or code, "net_pnl": round(pnl, 2), "trade_count": stock_trade_count.get(code, 0), } sorted_stocks = sorted(stock_pnl.items(), key=lambda x: x[1]) # 손실 종목만 (음수), 수익 종목만 (양수) 필터링하여 색상 혼동 방지 worst_stocks = [_stock_row(c, p) for c, p in sorted_stocks if p < 0][:5] best_stocks = [_stock_row(c, p) for c, p in sorted_stocks[::-1] if p > 0][:5] # ── 리스크 조정 수익률 지표 ──────────────────────────────────────────── risk_metrics = _calc_risk_adjusted_metrics(trades) # ── 전략 드리프트 감지 ───────────────────────────────────────────────── drift_alerts = _detect_drift_alerts(trades) return { "summary": summary, "by_strategy": strategy_stats, "by_weekday": weekday_stats, "by_exit_type": exit_stats, "worst_stocks": worst_stocks, "best_stocks": best_stocks, "sharpe_ratio": risk_metrics["sharpe_ratio"], "sortino_ratio": risk_metrics["sortino_ratio"], "calmar_ratio": risk_metrics["calmar_ratio"], "daily_returns_count": risk_metrics["daily_returns_count"], "drift_alerts": drift_alerts, "period_days": days, "total_trades": len(trades), "generated_at": now_kst().isoformat(), } # ───────────────────────────────────────────────────────────────────────────── # v2 성과 분석 (2026-10-09 하드닝 R1) # - 라운드트립(포지션) 단위 귀인: entry_trade_id 가 NULL 이어도 종목별 FIFO 로 BUY 와 연결 # - 시스템 성과(manual/external 제외)와 전체를 분리 # - Sharpe/Sortino/Calmar: KRX 거래일 전체에 0 수익을 채우고 실제 자본(세션 starting_balance) 분모 # 기존 analyze_performance 및 헬퍼 시그니처는 그대로 둔다(호출처 호환). # ───────────────────────────────────────────────────────────────────────────── # 시스템 매매가 아닌(수동 판단·외부 유입) strategy_type. 전략 성과에서 분리한다. NON_SYSTEM_STRATEGY_TYPES = frozenset({"manual", "external"}) # 라운드트립 매칭용 BUY 조회 룩백(일) — 윈도우 이전에 진입한 포지션의 청산을 연결하기 위함 _ROUNDTRIP_BUY_LOOKBACK_DAYS = 90 # 표본이 이보다 적으면 low_sample 로 표시(신뢰구간 병기) _LOW_SAMPLE_ACTIVE_DAYS = 30 _FALLBACK_CAPITAL_KRW = 10_000_000.0 def _is_system_trade_v2(t) -> bool: st = (getattr(t, "strategy_type", None) or "").strip().lower() return st not in NON_SYSTEM_STRATEGY_TYPES def _trade_qty_v2(t) -> int: q = getattr(t, "executed_quantity", None) or getattr(t, "order_quantity", None) or 0 try: return int(q) except (TypeError, ValueError): return 0 def _trade_sort_key_v2(t): ts = getattr(t, "created_at", None) or datetime.min return (ts, getattr(t, "id", 0) or 0) def _match_roundtrips_v2(buys: List[Any], sells: List[Any]) -> List[Dict[str, Any]]: """BUY/SELL TradeHistory 를 종목별로 시간순 매칭해 SELL 마다 진입 BUY 를 연결한다. - SELL.entry_trade_id 가 가리키는 BUY(남은 수량 있음)가 있으면 그것을 우선 사용 - 아니면 같은 종목의 가장 오래된 BUY lot(동일 strategy_type 우선) FIFO - 수량이 여러 lot 에 걸치면 첫 소진 lot 에 귀속 - 매칭 불가(윈도우 밖 진입 등)는 entry_trade_id=None 의 단독 포지션 반환: SELL 순서와 무관한 dict 리스트(각 SELL 1개). """ out: List[Dict[str, Any]] = [] # 종목별 이벤트 시간순 처리 events: Dict[str, List[Any]] = defaultdict(list) for b in buys: events[b.stock_code].append(("B", b)) for sl in sells: events[sl.stock_code].append(("S", sl)) for code, evs in events.items(): evs.sort(key=lambda e: (_trade_sort_key_v2(e[1]), 0 if e[0] == "B" else 1)) code_lots: List[List[Any]] = [] for kind, t in evs: if kind == "B": code_lots.append([t, _trade_qty_v2(t)]) continue need = _trade_qty_v2(t) chosen: Optional[List[Any]] = None linked_by: Optional[str] = None link_id = getattr(t, "entry_trade_id", None) if link_id: for lot in code_lots: if lot[0].id == link_id and lot[1] > 0: chosen = lot linked_by = "entry_trade_id" break if chosen is None: st = getattr(t, "strategy_type", None) cands = [l for l in code_lots if l[1] > 0] same = [l for l in cands if getattr(l[0], "strategy_type", None) == st] pool = same or cands chosen = pool[0] if pool else None if chosen is not None: linked_by = "fifo" first_lot = chosen remaining_need = need if chosen is not None: take = min(chosen[1], remaining_need) chosen[1] -= take remaining_need -= take # 초과분은 다음 lot FIFO 로 소진(귀속은 첫 lot) for lot in code_lots: if remaining_need <= 0: break if lot[1] > 0: tk = min(lot[1], remaining_need) lot[1] -= tk remaining_need -= tk out.append({"sell": t, "buy": first_lot[0] if first_lot else None, "lot": first_lot, "linked_by": linked_by}) # 포지션 종료 여부: lot 잔량 0 (lot 객체 공유) for r in out: r["lot_remaining"] = r["lot"][1] if r["lot"] is not None else 0 return out def _sell_cost_basis_v2(sell, buy) -> Optional[float]: """해당 SELL 이 청산한 수량의 원가. BUY 체결가 우선, 없으면 매도대금-순손익 역산.""" qty = _trade_qty_v2(sell) if buy is not None: bp = _to_float(getattr(buy, "executed_price", None)) or _to_float(getattr(buy, "order_price", None)) if bp > 0 and qty > 0: return bp * qty sp = _to_float(getattr(sell, "executed_price", None)) if sp > 0 and qty > 0: cost = sp * qty - _to_float(getattr(sell, "net_profit_loss", None)) if cost > 0: return cost return None def _holding_bucket_v2(days: Optional[float]) -> str: if days is None: return "unknown" if days < 1: return "intraday" if days <= 5: return "1-5d" if days <= 20: return "6-20d" return "20d+" def _build_positions_v2(matches: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """SELL 매칭 결과를 진입 BUY 단위 포지션으로 합친다(분할매도 = 1 포지션).""" groups: Dict[Any, List[Dict[str, Any]]] = {} for m in matches: key = m["buy"].id if m["buy"] is not None else ("orphan", m["sell"].id) groups.setdefault(key, []).append(m) positions: List[Dict[str, Any]] = [] for key, ms in groups.items(): buy = ms[0]["buy"] sells = sorted((m["sell"] for m in ms), key=_trade_sort_key_v2) net = sum(_to_float(s.net_profit_loss) for s in sells) costs = [_sell_cost_basis_v2(m["sell"], buy) for m in ms] cost = sum(c for c in costs if c) if any(costs) else None entry_at = getattr(buy, "created_at", None) if buy is not None else None last_exit = sells[-1].created_at holding = None if entry_at is not None and last_exit is not None: holding = max((last_exit - entry_at).total_seconds() / 86400.0, 0.0) ref = buy if buy is not None else sells[0] closed = buy is None or all(m["lot_remaining"] <= 0 for m in ms) positions.append({ "entry_trade_id": buy.id if buy is not None else None, "stock_code": sells[0].stock_code, "strategy_type": (getattr(ref, "strategy_type", None) or "").strip().lower() or "unknown", "strategy": _parse_strategy(getattr(ref, "entry_reason", None), getattr(ref, "strategy_type", None)), "entry_reason": (getattr(buy, "entry_reason", None) or "")[:80] if buy is not None else "", "exit_types": sorted({ ("manual_exit" if not _is_system_trade_v2(s) else _parse_exit_type(s.exit_reason)) for s in sells}), "manual_exits": sum(1 for s in sells if not _is_system_trade_v2(s)), "n_exits": len(sells), "net_pnl": net, "cost": cost, "return_pct": (net / cost * 100.0) if cost else None, "holding_days": holding, "closed": closed, "last_exit_at": last_exit, }) positions.sort(key=lambda p: (p["last_exit_at"] is None, p["last_exit_at"])) return positions def _position_stats_v2(positions: List[Dict[str, Any]]) -> Dict[str, Any]: """포지션 단위 승률·PF·금액가중 수익률. 부분청산 진행 중(closed=False)은 제외하고 개수만 보고.""" closed = [p for p in positions if p["closed"]] open_partial = len(positions) - len(closed) wins = [p for p in closed if p["net_pnl"] > 0] losses = [p for p in closed if p["net_pnl"] < 0] gp = sum(p["net_pnl"] for p in wins) gl = sum(p["net_pnl"] for p in losses) if gl < 0: pf = gp / abs(gl) else: pf = float("inf") if gp > 0 else 0.0 wr = [p["return_pct"] for p in wins if p["return_pct"] is not None] lr = [p["return_pct"] for p in losses if p["return_pct"] is not None] cost_total = sum(p["cost"] for p in closed if p["cost"]) net_total = sum(p["net_pnl"] for p in closed if p["cost"]) # [2026-10-09 R2] 금액 기준 평균손익·손익분기 승률. 레그(SELL 행) 단위 승률은 분할익절로 부풀어 # 주 지표로 쓰지 않는다. 손익분기 승률 = 평균손실/(평균이익+평균손실). avg_win_amt = (gp / len(wins)) if wins else 0.0 avg_loss_amt = (abs(gl) / len(losses)) if losses else 0.0 be_wr = (avg_loss_amt / (avg_win_amt + avg_loss_amt) * 100.0 if (avg_win_amt + avg_loss_amt) > 0 else None) # 수동 청산이 닫은(일부라도 수동 매도가 섞인) 봇 포지션은 별도 열로 귀속 man = [p for p in closed if p.get("manual_exits", 0) > 0] sysx = [p for p in closed if p.get("manual_exits", 0) == 0] return { "avg_win_amount": round(avg_win_amt, 2), "avg_loss_amount": round(-avg_loss_amt, 2), "payoff_ratio_amount": round(avg_win_amt / avg_loss_amt, 4) if avg_loss_amt > 0 else None, "breakeven_win_rate": round(be_wr, 2) if be_wr is not None else None, "manual_exit_positions": len(man), "manual_exit_net_pnl": round(sum(p["net_pnl"] for p in man), 2), "system_exit_positions": len(sysx), "system_exit_net_pnl": round(sum(p["net_pnl"] for p in sysx), 2), "positions": len(closed), "open_partial_positions": open_partial, "win_positions": len(wins), "loss_positions": len(losses), "win_rate": round(len(wins) / len(closed) * 100, 2) if closed else 0.0, "profit_factor": round(min(pf, 99.0), 4), "net_pnl": round(sum(p["net_pnl"] for p in closed), 2), "avg_win_pct": round(sum(wr) / len(wr), 4) if wr else 0.0, "avg_loss_pct": round(sum(lr) / len(lr), 4) if lr else 0.0, # 금액 가중: sum(순손익)/sum(원가). 단순 평균(%)은 소액 대박에 끌려가므로 병기. "amount_weighted_return_pct": round(net_total / cost_total * 100.0, 4) if cost_total > 0 else None, "avg_holding_days": ( round(sum(p["holding_days"] for p in closed if p["holding_days"] is not None) / max(1, sum(1 for p in closed if p["holding_days"] is not None)), 2) if closed else None ), } def _group_positions_v2(positions: List[Dict[str, Any]], keyfn) -> Dict[str, Dict[str, Any]]: groups: Dict[str, List[Dict[str, Any]]] = defaultdict(list) for p in positions: if not p["closed"]: continue groups[str(keyfn(p))].append(p) out = {} for k, ps in groups.items(): st = _position_stats_v2(ps) out[k] = { "positions": st["positions"], "win_rate": st["win_rate"], "net_pnl": st["net_pnl"], "profit_factor": st["profit_factor"], "amount_weighted_return_pct": st["amount_weighted_return_pct"], } return out def _summary_from_trades_v2(trades: List[Any]) -> Dict[str, Any]: """analyze_performance 의 summary 와 같은 정의(SELL 행 단위, 금액 기준 PF)를 임의 SELL 목록에 적용.""" if not trades: return _empty_summary() pl_rates = [_to_float(t.profit_loss_rate) for t in trades] pl_amounts = [_to_float(t.net_profit_loss) for t in trades] wins = [r for r in pl_rates if r > 0] losses = [r for r in pl_rates if r < 0] avg_win = sum(wins) / len(wins) if wins else 0.0 avg_loss = sum(losses) / len(losses) if losses else 0.0 gp = sum(a for a in pl_amounts if a > 0) gl = sum(a for a in pl_amounts if a < 0) if gl < 0: pf = gp / abs(gl) else: pf = float("inf") if gp > 0 else 0.0 return { "total_trades": len(trades), "win_trades": len(wins), "loss_trades": len(losses), "win_rate": round(len(wins) / len(pl_rates) * 100, 2), "avg_win_pct": round(avg_win, 4), "avg_loss_pct": round(avg_loss, 4), "profit_factor": round(min(pf, 99.0), 4), "gross_profit": round(gp, 2), "gross_loss": round(gl, 2), "net_pnl": round(gp + gl, 2), "max_consecutive_loss": _calc_max_consecutive_loss(pl_rates), "risk_reward_ratio": round(abs(avg_win / avg_loss), 4) if avg_loss != 0 else 0.0, **_cost_adjustment_v2(trades), } def _krx_trading_days_v2(start: date, end: date) -> List[date]: from app.core.market_schedule import is_krx_holiday days: List[date] = [] d = start while d <= end: if d.weekday() < 5 and not is_krx_holiday(d): days.append(d) d += timedelta(days=1) return days def _daily_returns_v2( daily_pnl: Dict[date, float], trading_days: List[date], capital_by_day: Optional[Dict[date, float]] = None, fallback_capital: float = _FALLBACK_CAPITAL_KRW, ) -> Dict[str, Any]: """KRX 거래일 전체(무거래일=0)를 인덱스로 일별 수익률을 만든다. 분모: 그날 이전(포함) 가장 최근 세션 starting_balance(실제 총자산) — 없으면 fallback_capital 에 누적 실현손익을 더한 값(구 동작). 거래일이 아닌 날의 PnL 이 있으면 그날도 인덱스에 포함. """ days = sorted(set(trading_days) | set(daily_pnl.keys())) cap_days = sorted((capital_by_day or {}).keys()) returns: List[float] = [] equity = fallback_capital real_cap_days = 0 ci = 0 cur_cap: Optional[float] = None for d in days: while ci < len(cap_days) and cap_days[ci] <= d: v = capital_by_day[cap_days[ci]] if v and v > 0: cur_cap = v ci += 1 pnl = daily_pnl.get(d, 0.0) base = cur_cap if cur_cap else equity if cur_cap: real_cap_days += 1 if base <= 0: break returns.append(pnl / base) equity += pnl n = len(days) return { "returns": returns, "days": n, "active_days": sum(1 for d in days if d in daily_pnl and daily_pnl[d] != 0.0), "capital_source": ( "session_starting_balance" if real_cap_days == n and n > 0 else ("mixed" if real_cap_days > 0 else "fallback_fixed_capital") ), } def _sharpe_ci95_v2(sharpe: Optional[float], n: int) -> Optional[List[float]]: """Lo(2002) 근사 표준오차 기반 95% 신뢰구간(연환산 Sharpe). 표본 부족이면 None.""" if sharpe is None or n < 2: return None sr_d = sharpe / _ANNUALIZE_FACTOR se = math.sqrt((1.0 + 0.5 * sr_d ** 2) / n) * _ANNUALIZE_FACTOR return [round(sharpe - 1.96 * se, 4), round(sharpe + 1.96 * se, 4)] def _risk_metrics_v2( trades: List[Any], trading_days: List[date], capital_by_day: Optional[Dict[date, float]] = None, ) -> Dict[str, Any]: daily_pnl = _build_daily_pnl_series(trades) dr = _daily_returns_v2(daily_pnl, trading_days, capital_by_day) r = dr["returns"] sharpe = _calc_sharpe_ratio(r) adj_daily: Dict[date, float] = defaultdict(float) for t in trades: ca = getattr(t, "created_at", None) if ca is None: continue adj_daily[ca.date() if isinstance(ca, datetime) else ca] += _adjusted_net_v2(t) adj_r = _daily_returns_v2(dict(adj_daily), trading_days, capital_by_day)["returns"] return { "sharpe_ratio_cost_adjusted": _calc_sharpe_ratio(adj_r), "sharpe_ratio": sharpe, "sharpe_ci95": _sharpe_ci95_v2(sharpe, len(r)), "sortino_ratio": _calc_sortino_ratio(r), "calmar_ratio": _calc_calmar_ratio(r), "daily_returns_count": len(r), "active_trading_days": dr["active_days"], "capital_source": dr["capital_source"], "low_sample": dr["active_days"] < _LOW_SAMPLE_ACTIVE_DAYS, "_returns": r, } def _detect_drift_alerts_v2(trades: List[Any], daily_returns: List[float]) -> List[Dict[str, Any]]: """_detect_drift_alerts 의 승률/연패 경보는 유지하고, Sharpe/MDD 경보는 0 채움 시계열로 재계산.""" alerts = [a for a in _detect_drift_alerts(trades) if a.get("type") not in ("rolling_sharpe", "mdd_acceleration")] recent = daily_returns[-_DRIFT_ROLLING_WINDOW:] if len(recent) >= _DRIFT_ROLLING_WINDOW: rs = _calc_sharpe_ratio(recent) if rs is not None and rs <= _DRIFT_SHARPE_THRESHOLD: alerts.append({ "type": "rolling_sharpe", "value": rs, "threshold": _DRIFT_SHARPE_THRESHOLD, "message": f"최근 {_DRIFT_ROLLING_WINDOW}거래일 Sharpe {rs:.2f} (임계값 {_DRIFT_SHARPE_THRESHOLD:.1f})", }) if len(daily_returns) >= 5: equity = 1.0 eqs = [equity] for r in daily_returns: equity *= (1.0 + r) eqs.append(equity) recent_eq = eqs[-6:] peak, low = max(recent_eq), min(recent_eq) if peak > 0: dd = (low - peak) / peak if dd <= _DRIFT_DRAWDOWN_THRESHOLD: alerts.append({ "type": "mdd_acceleration", "value": round(dd, 4), "threshold": _DRIFT_DRAWDOWN_THRESHOLD, "message": f"최근 drawdown {dd*100:.2f}% (임계값 {_DRIFT_DRAWDOWN_THRESHOLD*100:.1f}%)", }) return alerts def _hour_kst_v2(ts) -> Optional[str]: """DB 타임스탬프는 KST naive 로 통일돼 있다(timeutil). tz-aware 면 KST 로 변환.""" if ts is None or not isinstance(ts, datetime): return None if ts.tzinfo is not None: from app.core.timeutil import KST ts = ts.astimezone(KST) return f"{ts.hour:02d}" def _hour_stats_v2(groups: Dict[str, List[Any]]) -> Dict[str, Dict[str, Any]]: out: Dict[str, Dict[str, Any]] = {} for h in sorted(groups): ts = groups[h] nets = [_to_float(t.net_profit_loss) for t in ts] adj = [_adjusted_net_v2(t) for t in ts] out[h] = { "count": len(ts), "win_rate": round(sum(1 for a in adj if a > 0) / len(ts) * 100, 2), "net_pnl": round(sum(nets), 2), "estimated_cost_adjusted_net": round(sum(adj), 2), } return out def _compose_performance_v2( sells: List[Any], buys: List[Any], trading_days: List[date], capital_by_day: Optional[Dict[date, float]], days: int, ) -> Dict[str, Any]: """DB 비의존 순수 조립부(테스트 대상).""" system_sells = [t for t in sells if _is_system_trade_v2(t)] external_sells = [t for t in sells if not _is_system_trade_v2(t)] # [2026-10-09 R1] 수동 매도가 시스템 BUY lot 을 소진하면 그 포지션은 시스템 포지션에 'manual_exit' # 로 귀속한다(이전에는 시스템 SELL 만 매칭해 lot 이 영구 open 으로 남았다). 전체 SELL 을 한 번에 # 매칭해야 lot 소진 순서가 실제와 같다. all_matches = _match_roundtrips_v2(buys, sells) def _buy_is_system(m) -> bool: return m["buy"] is not None and _is_system_trade_v2(m["buy"]) sys_matches = [m for m in all_matches if _buy_is_system(m) or (m["buy"] is None and _is_system_trade_v2(m["sell"]))] ext_matches = [m for m in all_matches if m not in sys_matches] matches = sys_matches positions = _build_positions_v2(sys_matches) ext_positions = _build_positions_v2(ext_matches) manual_exit_sells = [m["sell"] for m in sys_matches if not _is_system_trade_v2(m["sell"])] sys_summary = _summary_from_trades_v2(system_sells) risk = _risk_metrics_v2(system_sells, trading_days, capital_by_day) daily_returns = risk.pop("_returns") strat_stats: Dict[str, Any] = {} by_strat_rows: Dict[str, List[float]] = defaultdict(list) for t in system_sells: by_strat_rows[_parse_strategy(t.entry_reason, getattr(t, "strategy_type", None))].append( _to_float(t.profit_loss_rate)) for k, rates in by_strat_rows.items(): strat_stats[k] = { "total": len(rates), "win_rate": round(len([r for r in rates if r > 0]) / len(rates) * 100, 2), "avg_return": round(sum(rates) / len(rates), 4), } exit_rows: Dict[str, List[float]] = defaultdict(list) for t in system_sells: exit_rows[_parse_exit_type(t.exit_reason)].append(_to_float(t.profit_loss_rate)) exit_stats = { k: {"count": len(v), "avg_return": round(sum(v) / len(v), 4), "win_rate": round(len([r for r in v if r > 0]) / len(v) * 100, 2)} for k, v in exit_rows.items() if v } sys_sell_matches = [m for m in sys_matches if _is_system_trade_v2(m["sell"])] linked = sum(1 for m in sys_sell_matches if m["buy"] is not None) explicit = sum(1 for m in sys_sell_matches if m.get("linked_by") == "entry_trade_id") fifo = sum(1 for m in sys_sell_matches if m.get("linked_by") == "fifo") n_sys = len(system_sells) # 시간대(KST naive) 귀인: 청산 시각 / 진입(매칭된 BUY) 시각, 비용 추정 차감 net 병기 by_exit_hour: Dict[str, List[Any]] = defaultdict(list) by_entry_hour: Dict[str, List[Any]] = defaultdict(list) for m in sys_sell_matches: sl = m["sell"] eh = _hour_kst_v2(getattr(sl, "created_at", None)) if eh is not None: by_exit_hour[eh].append(sl) if m["buy"] is not None: nh = _hour_kst_v2(getattr(m["buy"], "created_at", None)) if nh is not None: by_entry_hour[nh].append(sl) return { "summary": sys_summary, # 시스템(manual/external 제외) SELL(레그) 단위 — 주 지표 아님 "summary_basis": "sell_leg", # 분할익절 레그 단위. 주 지표는 position_summary "summary_all": _summary_from_trades_v2(sells), # 참고: 수동·외부 포함 "external_summary": _summary_from_trades_v2(external_sells), "position_summary": _position_stats_v2(positions), "external_position_summary": _position_stats_v2(ext_positions), "by_strategy": strat_stats, "by_strategy_position": _group_positions_v2(positions, lambda p: p["strategy"]), "by_entry_reason": _group_positions_v2(positions, lambda p: p["entry_reason"] or "unknown"), "by_holding_period": _group_positions_v2(positions, lambda p: _holding_bucket_v2(p["holding_days"])), "by_exit_type": exit_stats, "roundtrip": { "sells": n_sys, "matched_to_buy": linked, "unmatched_sells": n_sys - linked, "linked_by_entry_trade_id": explicit, "linked_by_fifo_estimate": fifo, # FIFO 추정 비중이 높으면 보유기간·진입사유 귀인은 추정치다(운영 SELL 의 entry_trade_id 는 전부 NULL). "attribution_confidence": ( "high" if n_sys and explicit / n_sys >= 0.8 else ("low_fifo_estimate" if n_sys else "n/a")), "method": "entry_trade_id 우선, 없으면 종목별 FIFO(동일 strategy_type 우선)", }, "manual_exit_summary": { **_summary_from_trades_v2(manual_exit_sells), "note": "시스템 BUY lot 을 청산한 수동 매도(시스템 SELL summary 에는 미포함, 포지션에는 manual_exit 로 귀속)", }, "by_exit_hour": _hour_stats_v2(by_exit_hour), "by_entry_hour": _hour_stats_v2(by_entry_hour), **risk, "drift_alerts": _detect_drift_alerts_v2(system_sells, daily_returns), "period_days": days, "total_trades": len(system_sells), } async def analyze_performance_v2(db: AsyncSession, days: int = 30) -> Dict[str, Any]: """analyze_performance 의 보정판. summary 는 manual/external 을 제외한 시스템 성과이며, 포지션(라운드트립) 단위 통계·보유기간/진입사유별 귀인·0 채움 Sharpe 를 추가로 반환한다. build_insights() 와 호환(summary/by_strategy/by_exit_type/drift_alerts/total_trades 키 유지).""" from app.models.models import TradingSession now = now_kst() since = now - timedelta(days=days) sells_res = await db.execute( select(TradeHistory).where(and_( TradeHistory.order_type == "SELL", TradeHistory.status.in_(["COMPLETED", "FILLED"]), TradeHistory.created_at >= since, )).order_by(TradeHistory.created_at) ) sells: List[TradeHistory] = sells_res.scalars().all() buys: List[TradeHistory] = [] if sells: buys_res = await db.execute( select(TradeHistory).where(and_( TradeHistory.order_type == "BUY", TradeHistory.status.in_(["COMPLETED", "FILLED", "PARTIAL"]), TradeHistory.created_at >= since - timedelta(days=_ROUNDTRIP_BUY_LOOKBACK_DAYS), )).order_by(TradeHistory.created_at) ) buys = buys_res.scalars().all() capital_by_day: Dict[date, float] = {} sess_res = await db.execute( select(TradingSession.session_date, TradingSession.starting_balance) .where(TradingSession.session_date >= since.date()) ) for d, bal in sess_res.all(): if d is not None and bal is not None and float(bal) > 0: capital_by_day[d] = float(bal) trading_days = _krx_trading_days_v2(since.date(), now.date()) perf = _compose_performance_v2(sells, buys, trading_days, capital_by_day, days) perf["generated_at"] = now.isoformat() return perf def _empty_summary() -> Dict[str, Any]: return { "total_trades": 0, "win_trades": 0, "loss_trades": 0, "win_rate": 0.0, "avg_win_pct": 0.0, "avg_loss_pct": 0.0, "profit_factor": 0.0, "gross_profit": 0.0, "gross_loss": 0.0, "net_pnl": 0.0, "max_consecutive_loss": 0, "risk_reward_ratio": 0.0, "gross_basis_trades": 0, "estimated_cost": 0.0, "estimated_cost_adjusted_net": 0.0, } def _calc_max_consecutive_loss(pl_rates: List[float]) -> int: max_streak = current_streak = 0 for r in pl_rates: if r < 0: current_streak += 1 max_streak = max(max_streak, current_streak) else: current_streak = 0 return max_streak # ───────────────────────────────────────────────────────────────────────────── # 전략 인사이트 # ───────────────────────────────────────────────────────────────────────────── def build_insights(perf: Dict[str, Any]) -> List[Dict[str, str]]: """ 성과 데이터를 기반으로 개선 인사이트 목록을 생성. Returns: [{"level": "warning"|"info"|"danger", "message": "..."}] """ insights = [] summary = perf.get("summary", {}) win_rate = summary.get("win_rate", 0.0) pf = summary.get("profit_factor", 0.0) avg_loss = summary.get("avg_loss_pct", 0.0) max_cons_loss = summary.get("max_consecutive_loss", 0) rr = summary.get("risk_reward_ratio", 0.0) # [2026-10-09 R2] 포지션(BUY→전량청산) 단위 집계가 있으면 그것을 주 지표로 쓴다. SELL 레그 단위 # 승률/손익비는 분할익절 레그가 '승'을 부풀려 손익분기 승률을 과대 평가하게 만든다. pos = perf.get("position_summary") or {} pos_based = bool(pos.get("positions")) if pos_based: win_rate = pos.get("win_rate", 0.0) pf = pos.get("profit_factor", 0.0) avg_loss = pos.get("avg_loss_pct", 0.0) _aw = pos.get("avg_win_pct", 0.0) rr = abs(_aw / avg_loss) if avg_loss else 0.0 _be = pos.get("breakeven_win_rate") insights.append({"level": "info", "message": ( f"[포지션 단위 {pos['positions']}건] 승률 {win_rate:.1f}%" + (f" / 손익분기 승률 {_be:.1f}%" if _be is not None else "") + f" / PF {pf:.2f} (레그 단위 승률 {summary.get('win_rate', 0.0):.1f}% 는 참고용)")}) if pos.get("manual_exit_positions"): insights.append({"level": "info", "message": ( f"수동 청산이 닫은 봇 포지션 {pos['manual_exit_positions']}건 순손익 " f"{pos.get('manual_exit_net_pnl', 0):,.0f}원 / 시스템 청산 {pos.get('system_exit_positions', 0)}건 " f"{pos.get('system_exit_net_pnl', 0):,.0f}원 — 봇 성과 판단 시 분리해서 볼 것")}) if win_rate < 40: insights.append({"level": "danger", "message": f"승률 {win_rate:.1f}% — RSI 과매도 기준을 강화하거나 진입 조건을 추가 필터링하세요."}) elif win_rate < 50: insights.append({"level": "warning", "message": f"승률 {win_rate:.1f}% — 50% 미만입니다. RSI_OVERSOLD_LEVEL 하향 조정을 검토하세요."}) else: insights.append({"level": "info", "message": f"승률 {win_rate:.1f}% — 양호합니다."}) # 거래 없음/수익·손실 모두 0 이면 PF=0 은 "손실>수익"이 아니라 판단 불가 — 인사이트 생략 _has_pnl = bool(summary.get("gross_profit") or summary.get("gross_loss")) if not _has_pnl: pass elif pf < 1.0: insights.append({"level": "danger", "message": f"Profit Factor {pf:.2f} — 손실이 수익보다 큽니다. 손절 기준 강화 필요."}) elif pf < 1.5: insights.append({"level": "warning", "message": f"Profit Factor {pf:.2f} — 개선 여지 있습니다."}) if rr < 1.0 and rr > 0: insights.append({"level": "warning", "message": f"손익비 {rr:.2f} — 1:1 미만입니다. ATR 익절 배수 또는 트레일링 스탑 완화를 검토하세요."}) if max_cons_loss >= 3: insights.append({"level": "warning", "message": f"최대 연속 손실 {max_cons_loss}회 — 연속 손절 발생 중. 진입 필터(추세 필터, 시장국면 필터) 점검 필요."}) if avg_loss < -8.0: insights.append({"level": "danger", "message": f"평균 손실 {avg_loss:.2f}% — 손절선({avg_loss:.2f}%)이 너무 넓습니다. STOP_LOSS_PERCENT 또는 ATR_STOP_MULTIPLIER 조정 권장."}) by_strategy = perf.get("by_strategy", {}) for strat, stats in by_strategy.items(): if stats.get("total", 0) >= 5 and stats.get("win_rate", 100) < 35: insights.append({"level": "warning", "message": f"'{strat}' 전략 승률 {stats['win_rate']:.1f}% — 해당 전략 파라미터 재검토 권장."}) by_exit = perf.get("by_exit_type", {}) stop_loss_info = by_exit.get("stop_loss", {}) if stop_loss_info.get("count", 0) > 0: sl_ratio = stop_loss_info["count"] / max(perf["total_trades"], 1) if sl_ratio > 0.5: insights.append({"level": "warning", "message": f"손절 청산 비율 {sl_ratio*100:.1f}% — 전체 청산의 절반 이상이 손절입니다. 진입 타이밍 재검토."}) # ── 리스크 조정 지표 인사이트 ────────────────────────────────────────── sharpe = perf.get("sharpe_ratio") if sharpe is not None: if sharpe < 0: insights.append({"level": "danger", "message": f"Sharpe Ratio {sharpe:.2f} — 무위험이자율(3%) 대비 음의 초과수익입니다. 전략 전면 재검토 필요."}) elif sharpe < 0.5: insights.append({"level": "warning", "message": f"Sharpe Ratio {sharpe:.2f} — 낮은 수준입니다. 리스크 대비 수익 개선이 필요합니다."}) elif sharpe >= 1.5: insights.append({"level": "info", "message": f"Sharpe Ratio {sharpe:.2f} — 우수한 리스크 조정 수익률입니다."}) calmar = perf.get("calmar_ratio") if calmar is not None and calmar < 1.0: insights.append({"level": "warning", "message": f"Calmar Ratio {calmar:.2f} — MDD 대비 수익률이 낮습니다. 낙폭 관리 강화 필요."}) # ── 드리프트 경고 인사이트 ───────────────────────────────────────────── drift_alerts = perf.get("drift_alerts", []) for alert in drift_alerts: insights.append({"level": "danger", "message": f"[드리프트 감지] {alert['message']}"}) return insights # ───────────────────────────────────────────────────────────────────────────── # 매매일지 생성 # ───────────────────────────────────────────────────────────────────────────── # 일지 대상 SELL 상태. 성과 집계(analyze_performance)의 COMPLETED/FILLED 와 **의도적으로 다르다**: # PARTIAL 은 실제 체결된 매도(부분 체결)라 기록할 가치가 있고, 이후 COMPLETED 로 갱신돼도 # 같은 TradeHistory.id 라 trade_id 커버 판정으로 중복되지 않는다. PENDING(체결 미확인)· # FAILED(거부)는 손익이 확정되지 않았거나 거래가 없으므로 제외한다. # 성과 집계에는 PARTIAL 을 넣지 않는다(손익 이중 계상 방지) — 이 상수는 일지 전용. JOURNAL_SELL_STATUSES = ("COMPLETED", "FILLED", "PARTIAL") async def get_uncovered_trades(db: AsyncSession, days: int = 7) -> List[TradeHistory]: """일지가 아직 없는 체결 SELL(COMPLETED/FILLED/PARTIAL) 거래 목록을 반환.""" return await get_uncovered_trades_v2(db, days=days, min_age_minutes=0) async def get_uncovered_trades_v2( db: AsyncSession, days: int = 7, min_age_minutes: int = 0 ) -> List[TradeHistory]: """get_uncovered_trades + 유예시간: created_at 이 min_age_minutes 이내인 SELL 은 제외. 매도 직후 `_create_journal_background`(MCP 분석 최대 30초 후 큐 등록)가 같은 매도의 일지를 만들 시간을 준다 — 장중 주기 태스크가 그 사이에 중복 생성하지 않도록. """ since = now_kst() - timedelta(days=days) cutoff = now_kst() - timedelta(minutes=max(0, int(min_age_minutes))) # 이미 일지가 있는 trade_id 목록 # [2026-09-29 B3] `> 0` 조건이 필수다. 과거 호출부가 trade_id 미확보 시 `0` 을 # 넣어서, 0 이 covered 집합에 섞이면 존재하지 않는 PK 를 커버로 세게 된다 # (역으로 실제 거래는 계속 uncovered → 중복 일지). NULL/0 은 둘 다 "커버 안 함". covered_stmt = select(TradeJournal.trade_id).where(TradeJournal.trade_id > 0) covered_result = await db.execute(covered_stmt) covered_ids = {row[0] for row in covered_result.fetchall()} stmt = select(TradeHistory).where( and_( TradeHistory.order_type == "SELL", TradeHistory.status.in_(JOURNAL_SELL_STATUSES), TradeHistory.created_at >= since, TradeHistory.created_at <= cutoff, TradeHistory.id.not_in(covered_ids) if covered_ids else True, ) ).order_by(TradeHistory.created_at.desc()) result = await db.execute(stmt) return result.scalars().all() def generate_journal_for_trade(trade: TradeHistory, perf_context: Optional[Dict] = None) -> str: """ 템플릿 기반 매매일지 내용 생성. Args: trade: 완료된 SELL TradeHistory 레코드 perf_context: 전체 성과 컨텍스트 (선택) Returns: 일지 본문 문자열 """ pl_rate = _to_float(trade.profit_loss_rate) pl_amount = _to_float(trade.net_profit_loss) exec_price = _to_float(trade.executed_price) exec_qty = trade.executed_quantity or trade.order_quantity or 0 strategy = _parse_strategy(trade.entry_reason, getattr(trade, "strategy_type", None)) exit_type = _parse_exit_type(trade.exit_reason) # 결과 평가 if pl_rate > 3.0: result_label = "✅ 큰 수익 달성" result_comment = "목표 수익을 충분히 달성했습니다. 진입 타이밍과 전략이 잘 맞았습니다." elif pl_rate > 0: result_label = "✅ 소폭 수익" result_comment = "소폭 수익 실현. 목표 대비 아쉬운 결과이나 플러스 마감입니다." elif pl_rate > -3.0: result_label = "⚠️ 소폭 손실" result_comment = "소폭 손실. 리스크 관리 규칙 내에서 처리되었습니다." else: result_label = "❌ 손절 처리" result_comment = "손절 처리되었습니다. 진입 조건 및 시장 국면을 재검토하세요." # 청산 유형별 코멘트 # 임계값은 하드코딩하지 않고 실효 설정값을 쓴다(CLAUDE.md §4). from app.core.config import settings as _s exit_comments = { "stop_loss": ( f"→ 손절 기준(스윙 -{abs(float(_s.STOP_LOSS_PERCENT)):g}% / 고변동 -{abs(float(_s.STOP_LOSS_HIGH_VOL)):g}%, " f"ATR×{float(_s.ATR_STOP_MULTIPLIER):g}, 데이트레이딩은 진입 시 손절가)에 도달하여 청산." ), "time_stop": ( f"→ 시간손절: 데이트레이딩은 진입 {int(_s.DAY_TRADE_TIME_STOP_MINUTES)}분 경과 후 수익률 " f"{float(_s.DAY_TRADE_TIME_STOP_MIN_GAIN) * 100:+.1f}% 미만, 스윙은 장기 보유 수익 미달로 조기 정리. " "손실 한도 도달이 아니다." ), "trailing_stop": f"→ 트레일링 스탑(고점-{float(_s.ATR_TRAILING_MULTIPLIER):g}×ATR)에 도달. 고점 대비 일정 비율 하락.", "take_profit": ( f"→ 익절 기준(RSI {float(_s.MEAN_REVERSION_RSI_EXIT_THRESHOLD):g}+ 1차 / " f"{float(_s.MEAN_REVERSION_RSI_FULL_EXIT):g}+ 전량, 데이트레이딩 목표가)에 도달하여 수익 확정." ), "eod_forced": "→ 장 마감 강제 청산(15:20 EOD). 포지션 오버나이트 방지.", "rsi_exit": "→ RSI 과매수 구간 진입으로 부분/전량 청산.", "manual": "→ 수동 청산 처리.", "unknown": "→ 청산 사유 미상.", } exit_comment = exit_comments.get(exit_type, "→ 청산 완료.") lines = [ f"# 매매일지 — {trade.stock_code} ({trade.stock_name or ''})", f"**날짜**: {(trade.executed_at or trade.created_at).strftime('%Y-%m-%d %H:%M') if (trade.executed_at or trade.created_at) else '-'}", f"**전략**: {strategy}", "", f"## 거래 결과: {result_label}", f"- 체결가: {exec_price:,.0f}원 | 수량: {exec_qty:,}주", f"- 손익률: **{pl_rate:+.2f}%** | 손익금액: {pl_amount:+,.0f}원", f"- 청산 유형: `{exit_type}`", f" {exit_comment}", "", "## 진입 근거", f"{trade.entry_reason or '기록 없음'}", "", "## 청산 근거", f"{trade.exit_reason or '기록 없음'}", "", "## 평가 및 교훈", result_comment, ] # 성과 컨텍스트 추가 (있을 경우) if perf_context: overall_wr = perf_context.get("summary", {}).get("win_rate", 0) lines += [ "", f"---", f"*최근 {perf_context.get('period_days', 30)}일 전체 승률: {overall_wr:.1f}% (참고)*", ] return "\n".join(lines) async def create_journal_record( db: AsyncSession, trade: TradeHistory, content: str, ) -> TradeJournal: """TradeJournal 레코드를 DB에 저장 후 반환.""" pl_rate = _to_float(trade.profit_loss_rate) strategy = _parse_strategy(trade.entry_reason, getattr(trade, "strategy_type", None)) exit_type = _parse_exit_type(trade.exit_reason) mood = "positive" if pl_rate > 0 else ("neutral" if pl_rate > -3.0 else "negative") tags = [strategy, exit_type] if abs(pl_rate) > 5.0: tags.append("significant") journal = TradeJournal( trade_id=trade.id, stock_code=trade.stock_code, stock_name=trade.stock_name, journal_date=trade.executed_at or trade.created_at or now_kst(), content=content, strategy_name=strategy, mood=mood, tags=tags, ) db.add(journal) await db.flush() return journal