shuim/auto-stock
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

수정본
Some checks failed
CI/CD Pipeline / test-backend (push) Has been cancelled
Details
CI/CD Pipeline / test-frontend (push) Has been cancelled
Details
CI/CD Pipeline / build-and-push (push) Has been cancelled
Details

Browse Source
This commit is contained in:
shuim
2026-03-27 14:47:49 +09:00
parent 844ee8944f
commit c818810667
25 changed files with 2708 additions and 266 deletions
Download Patch File Download Diff File Expand all files Collapse all files

314
backend/app/services/trade_analyzer.py
View File

@@ -4,10 +4,14 @@
[역할]
- 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
@@ -70,6 +74,276 @@ def _parse_exit_type(exit_reason: Optional[str]) -> str:
return "unknown"
# ─────────────────────────────────────────────────────────────────────────────
# 리스크 조정 수익률 지표 헬퍼
# ─────────────────────────────────────────────────────────────────────────────
# 무위험이자율 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
# ─────────────────────────────────────────────────────────────────────────────
# 핵심 분석 함수
# ─────────────────────────────────────────────────────────────────────────────
@@ -86,6 +360,11 @@ async def analyze_performance(db: AsyncSession, days: int = 30) -> Dict[str, Any
"by_exit_type": {청산 유형별 통계},
"worst_stocks": [최대 손실 종목 Top5],
"best_stocks": [최대 수익 종목 Top5],
"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 문자열
}
@@ -111,6 +390,11 @@ async def analyze_performance(db: AsyncSession, days: int = 30) -> Dict[str, Any
"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": datetime.utcnow().isoformat(),
@@ -210,6 +494,12 @@ async def analyze_performance(db: AsyncSession, days: int = 30) -> Dict[str, Any
worst_stocks = [{"stock_code": c, "net_pnl": round(p, 2)} for c, p in sorted_stocks[:5]]
best_stocks = [{"stock_code": c, "net_pnl": round(p, 2)} for c, p in sorted_stocks[-5:][::-1]]
# ── 리스크 조정 수익률 지표 ────────────────────────────────────────────
risk_metrics = _calc_risk_adjusted_metrics(trades)
# ── 전략 드리프트 감지 ─────────────────────────────────────────────────
drift_alerts = _detect_drift_alerts(trades)
return {
"summary": summary,
"by_strategy": strategy_stats,
@@ -217,6 +507,11 @@ async def analyze_performance(db: AsyncSession, days: int = 30) -> Dict[str, Any
"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": datetime.utcnow().isoformat(),
@@ -302,6 +597,25 @@ def build_insights(perf: Dict[str, Any]) -> List[Dict[str, str]]:
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