gahusb
7ea1a21487
Atomic mv of root V1 assets (main_server.py + modules/ + data/ + tests/ + entry scripts + docs + logs) into signal_v1/ subdirectory. load_dotenv() updated to load web-ai/.env explicitly via Path. Adds web-ai/CLAUDE.md (workspace guide) and web-ai/start.bat (signal_v1 entry wrapper). Prepares for signal_v2/ Phase 2. Tests: signal_v1/tests/unit baseline preserved (no regression). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
422 lines
19 KiB
Python
422 lines
19 KiB
Python
"""
|
|
성과 평가 엔진 - PerformanceEvaluator
|
|
|
|
기능:
|
|
1. compute_metrics() - 핵심 성과 지표 계산
|
|
2. get_grade() - 지표별 S/A/B/C/D/F 등급 산출
|
|
3. generate_expert_panel() - Ollama LLM 5명 전문가 의견
|
|
4. generate_weekly_report() - 텔레그램 HTML 주간 보고서
|
|
"""
|
|
import json
|
|
import math
|
|
from datetime import datetime, timedelta
|
|
|
|
from modules.utils.performance_db import PerformanceDB
|
|
|
|
|
|
class PerformanceEvaluator:
|
|
def __init__(self):
|
|
self.perf_db = PerformanceDB()
|
|
|
|
# ─────────────────────────────────────────
|
|
# 1. 핵심 지표 계산
|
|
# ─────────────────────────────────────────
|
|
|
|
def compute_metrics(self, snapshots, trades):
|
|
"""성과 지표를 딕셔너리로 반환.
|
|
|
|
Args:
|
|
snapshots (list): daily_snapshots 리스트
|
|
trades (list): trade_records 리스트
|
|
|
|
Returns:
|
|
dict: 지표 딕셔너리 (또는 {"error": ...})
|
|
"""
|
|
if not snapshots:
|
|
return {"error": "스냅샷 데이터 없음 (운영 시작 후 첫 영업일까지 대기)"}
|
|
|
|
metrics = {}
|
|
|
|
# ── 수익률 ──────────────────────────────
|
|
initial = snapshots[0].get("total_eval", 0)
|
|
current = snapshots[-1].get("total_eval", 0)
|
|
metrics["total_return_pct"] = round(
|
|
(current - initial) / initial * 100, 2) if initial > 0 else 0.0
|
|
|
|
cutoff_7 = (datetime.now() - timedelta(days=7)).strftime("%Y-%m-%d")
|
|
recent_snaps = [s for s in snapshots if s.get("date", "") >= cutoff_7]
|
|
if len(recent_snaps) >= 2:
|
|
w_init = recent_snaps[0].get("total_eval", 0)
|
|
w_curr = recent_snaps[-1].get("total_eval", 0)
|
|
metrics["weekly_return_pct"] = round(
|
|
(w_curr - w_init) / w_init * 100, 2) if w_init > 0 else 0.0
|
|
else:
|
|
metrics["weekly_return_pct"] = 0.0
|
|
|
|
# ── 리스크 지표 ──────────────────────────
|
|
daily_returns = [s.get("daily_return_pct", 0.0) / 100.0 for s in snapshots]
|
|
|
|
if len(daily_returns) >= 2:
|
|
mean_daily = sum(daily_returns) / len(daily_returns)
|
|
variance = sum((r - mean_daily) ** 2 for r in daily_returns) / len(daily_returns)
|
|
std_daily = math.sqrt(variance) if variance > 0 else 0.0
|
|
annual_return = mean_daily * 252
|
|
|
|
# Sharpe Ratio
|
|
if std_daily > 0:
|
|
metrics["sharpe_ratio"] = round(
|
|
annual_return / (std_daily * math.sqrt(252)), 3)
|
|
else:
|
|
metrics["sharpe_ratio"] = 0.0
|
|
|
|
# Sortino Ratio (하방 편차만 사용)
|
|
downside = [r for r in daily_returns if r < 0]
|
|
if downside:
|
|
dv = sum(r ** 2 for r in downside) / len(downside)
|
|
ds = math.sqrt(dv)
|
|
metrics["sortino_ratio"] = round(
|
|
annual_return / (ds * math.sqrt(252)), 3) if ds > 0 else 0.0
|
|
else:
|
|
metrics["sortino_ratio"] = 10.0 # 손실 없음
|
|
|
|
# Max Drawdown
|
|
peak = snapshots[0].get("total_eval", 0)
|
|
max_dd = 0.0
|
|
for snap in snapshots:
|
|
ev = snap.get("total_eval", 0)
|
|
if ev > peak:
|
|
peak = ev
|
|
if peak > 0:
|
|
dd = (peak - ev) / peak * 100
|
|
if dd > max_dd:
|
|
max_dd = dd
|
|
metrics["max_drawdown_pct"] = round(max_dd, 2)
|
|
|
|
# Calmar Ratio
|
|
ann_pct = annual_return * 100
|
|
metrics["calmar_ratio"] = round(
|
|
ann_pct / max_dd, 3) if max_dd > 0 else 0.0
|
|
else:
|
|
metrics["sharpe_ratio"] = 0.0
|
|
metrics["sortino_ratio"] = 0.0
|
|
metrics["max_drawdown_pct"] = 0.0
|
|
metrics["calmar_ratio"] = 0.0
|
|
|
|
# ── 매매 지표 ─────────────────────────────
|
|
closed = [t for t in trades
|
|
if t.get("action") == "BUY" and t.get("outcome_return_pct") is not None]
|
|
|
|
if closed:
|
|
wins = [t for t in closed if t.get("outcome_return_pct", 0) > 0]
|
|
losses = [t for t in closed if t.get("outcome_return_pct", 0) <= 0]
|
|
|
|
metrics["win_rate_pct"] = round(
|
|
len(wins) / len(closed) * 100, 1)
|
|
|
|
total_profit = sum(t["outcome_return_pct"] for t in wins)
|
|
total_loss = abs(sum(t["outcome_return_pct"] for t in losses))
|
|
metrics["profit_factor"] = round(
|
|
total_profit / total_loss, 3) if total_loss > 0 else 10.0
|
|
|
|
hd_list = [t["holding_days"] for t in closed if t.get("holding_days") is not None]
|
|
metrics["avg_holding_days"] = round(
|
|
sum(hd_list) / len(hd_list), 1) if hd_list else 0.0
|
|
else:
|
|
metrics["win_rate_pct"] = 0.0
|
|
metrics["profit_factor"] = 0.0
|
|
metrics["avg_holding_days"] = 0.0
|
|
|
|
metrics["total_trades"] = len(closed)
|
|
|
|
# ── 벤치마크 Alpha ────────────────────────
|
|
kospi_vals = [s.get("benchmark_kospi_close") for s in snapshots]
|
|
kospi_valid = [k for k in kospi_vals if k is not None]
|
|
if len(kospi_valid) >= 2:
|
|
kospi_ret = (kospi_valid[-1] - kospi_valid[0]) / kospi_valid[0] * 100
|
|
metrics["alpha"] = round(metrics["total_return_pct"] - kospi_ret, 2)
|
|
metrics["kospi_return_pct"] = round(kospi_ret, 2)
|
|
else:
|
|
metrics["alpha"] = 0.0
|
|
metrics["kospi_return_pct"] = 0.0
|
|
|
|
# ── AI 품질 지표 ──────────────────────────
|
|
if closed:
|
|
# LSTM 방향 정확도
|
|
correct = 0
|
|
direction_n = 0
|
|
for t in closed:
|
|
pred = t.get("ai_prediction_change")
|
|
outcome = t.get("outcome_return_pct")
|
|
if pred is not None and outcome is not None:
|
|
if (pred > 0) == (outcome > 0):
|
|
correct += 1
|
|
direction_n += 1
|
|
metrics["lstm_direction_accuracy"] = round(
|
|
correct / direction_n * 100, 1) if direction_n > 0 else 0.0
|
|
|
|
# 신호별 수익 상관도
|
|
outcomes = [t.get("outcome_return_pct", 0) for t in closed]
|
|
|
|
def pearson(xs, ys):
|
|
n = len(xs)
|
|
if n < 2:
|
|
return 0.0
|
|
mx = sum(xs) / n
|
|
my = sum(ys) / n
|
|
num = sum((x - mx) * (y - my) for x, y in zip(xs, ys))
|
|
denom_x = sum((x - mx) ** 2 for x in xs)
|
|
denom_y = sum((y - my) ** 2 for y in ys)
|
|
denom = math.sqrt(denom_x * denom_y)
|
|
return num / denom if denom > 0 else 0.0
|
|
|
|
corr_tech = pearson([t.get("tech_score", 0) for t in closed], outcomes)
|
|
corr_sent = pearson([t.get("sentiment_score", 0) for t in closed], outcomes)
|
|
corr_lstm = pearson([t.get("lstm_score", 0) for t in closed], outcomes)
|
|
|
|
metrics["signal_correlation"] = {
|
|
"tech": round(corr_tech, 3),
|
|
"sentiment": round(corr_sent, 3),
|
|
"lstm": round(corr_lstm, 3)
|
|
}
|
|
metrics["best_signal_source"] = max(
|
|
["tech", "sentiment", "lstm"],
|
|
key=lambda k: abs(metrics["signal_correlation"][k])
|
|
)
|
|
else:
|
|
metrics["lstm_direction_accuracy"] = 0.0
|
|
metrics["signal_correlation"] = {"tech": 0.0, "sentiment": 0.0, "lstm": 0.0}
|
|
metrics["best_signal_source"] = "unknown"
|
|
|
|
metrics["snapshot_count"] = len(snapshots)
|
|
return metrics
|
|
|
|
# ─────────────────────────────────────────
|
|
# 2. 등급 산출
|
|
# ─────────────────────────────────────────
|
|
|
|
def get_grade(self, metric, value):
|
|
"""지표 이름과 값으로 S/A/B/C/D/F 등급 반환."""
|
|
# MDD는 낮을수록 좋음
|
|
if metric == "max_drawdown_pct":
|
|
thresholds = [(5, "S"), (10, "A"), (15, "B"), (20, "C"), (30, "D")]
|
|
for threshold, grade in thresholds:
|
|
if value < threshold:
|
|
return grade
|
|
return "F"
|
|
|
|
grade_rules = {
|
|
"sharpe_ratio": [(2.0, "S"), (1.5, "A"), (1.0, "B"), (0.5, "C"), (0.0, "D")],
|
|
"sortino_ratio": [(3.0, "S"), (2.0, "A"), (1.5, "B"), (1.0, "C"), (0.0, "D")],
|
|
"win_rate_pct": [(70, "S"), (60, "A"), (50, "B"), (40, "C"), (30, "D")],
|
|
"profit_factor": [(3.0, "S"), (2.0, "A"), (1.5, "B"), (1.0, "C"), (0.5, "D")],
|
|
"alpha": [(15, "S"), (10, "A"), (5, "B"), (0, "C"), (-5, "D")],
|
|
"total_return_pct": [(30, "S"), (20, "A"), (10, "B"), (0, "C"), (-10, "D")],
|
|
"weekly_return_pct": [(5, "S"), (3, "A"), (1, "B"), (0, "C"), (-1, "D")],
|
|
"lstm_direction_accuracy":[(70, "S"), (60, "A"), (55, "B"), (50, "C"), (40, "D")],
|
|
"calmar_ratio": [(3.0, "S"), (2.0, "A"), (1.0, "B"), (0.5, "C"), (0.0, "D")],
|
|
}
|
|
|
|
thresholds = grade_rules.get(metric, [])
|
|
for threshold, grade in thresholds:
|
|
if value >= threshold:
|
|
return grade
|
|
return "F"
|
|
|
|
# ─────────────────────────────────────────
|
|
# 3. 전문가 패널 (Ollama LLM)
|
|
# ─────────────────────────────────────────
|
|
|
|
def generate_expert_panel(self, metrics):
|
|
"""5명의 전문가 역할로 Ollama에 평가를 요청.
|
|
|
|
Returns:
|
|
list[dict]: [{role, grade, comment, suggestion}, ...]
|
|
"""
|
|
from modules.services.ollama import OllamaManager
|
|
ollama = OllamaManager()
|
|
|
|
sig_corr = metrics.get("signal_correlation", {})
|
|
experts = [
|
|
{
|
|
"role": "Risk Manager",
|
|
"focus": "risk level assessment and bankruptcy risk",
|
|
"data": (
|
|
f"Sharpe={metrics.get('sharpe_ratio', 0):.2f}, "
|
|
f"Sortino={metrics.get('sortino_ratio', 0):.2f}, "
|
|
f"MDD={metrics.get('max_drawdown_pct', 0):.1f}%, "
|
|
f"Calmar={metrics.get('calmar_ratio', 0):.2f}"
|
|
)
|
|
},
|
|
{
|
|
"role": "Fund Manager",
|
|
"focus": "alpha generation vs market benchmark",
|
|
"data": (
|
|
f"TotalReturn={metrics.get('total_return_pct', 0):.2f}%, "
|
|
f"Alpha={metrics.get('alpha', 0):.2f}%, "
|
|
f"KOSPI={metrics.get('kospi_return_pct', 0):.2f}%, "
|
|
f"WeeklyReturn={metrics.get('weekly_return_pct', 0):.2f}%"
|
|
)
|
|
},
|
|
{
|
|
"role": "Quant Analyst",
|
|
"focus": "AI model validity and signal quality",
|
|
"data": (
|
|
f"LSTM_Accuracy={metrics.get('lstm_direction_accuracy', 0):.1f}%, "
|
|
f"TechCorr={sig_corr.get('tech', 0):.3f}, "
|
|
f"SentCorr={sig_corr.get('sentiment', 0):.3f}, "
|
|
f"LSTMCorr={sig_corr.get('lstm', 0):.3f}, "
|
|
f"BestSignal={metrics.get('best_signal_source', 'N/A')}"
|
|
)
|
|
},
|
|
{
|
|
"role": "Trader",
|
|
"focus": "trading strategy effectiveness",
|
|
"data": (
|
|
f"WinRate={metrics.get('win_rate_pct', 0):.1f}%, "
|
|
f"ProfitFactor={metrics.get('profit_factor', 0):.2f}, "
|
|
f"AvgHolding={metrics.get('avg_holding_days', 0):.1f}days, "
|
|
f"TotalTrades={metrics.get('total_trades', 0)}"
|
|
)
|
|
},
|
|
{
|
|
"role": "Portfolio PM",
|
|
"focus": "overall strategy direction and sustainability",
|
|
"data": (
|
|
f"WeeklyReturn={metrics.get('weekly_return_pct', 0):.2f}%, "
|
|
f"Sharpe={metrics.get('sharpe_ratio', 0):.2f}, "
|
|
f"WinRate={metrics.get('win_rate_pct', 0):.1f}%, "
|
|
f"Alpha={metrics.get('alpha', 0):.2f}%, "
|
|
f"MDD={metrics.get('max_drawdown_pct', 0):.1f}%"
|
|
)
|
|
}
|
|
]
|
|
|
|
results = []
|
|
for exp in experts:
|
|
prompt = (
|
|
f"You are a professional {exp['role']} evaluating an AI stock trading bot. "
|
|
f"Your focus: {exp['focus']}. "
|
|
f"Performance data: {exp['data']}. "
|
|
f"Respond ONLY with valid JSON (no markdown, no extra text): "
|
|
f"{{\"grade\":\"S|A|B|C|D|F\","
|
|
f"\"comment\":\"1 sentence evaluation in Korean\","
|
|
f"\"suggestion\":\"1 sentence improvement tip in Korean\"}}"
|
|
)
|
|
try:
|
|
resp = ollama.request_inference(prompt)
|
|
if not resp:
|
|
raise ValueError("Empty response from Ollama")
|
|
data = json.loads(resp)
|
|
results.append({
|
|
"role": exp["role"],
|
|
"grade": data.get("grade", "C"),
|
|
"comment": data.get("comment", "(응답 없음)"),
|
|
"suggestion": data.get("suggestion", "데이터 축적 필요")
|
|
})
|
|
except Exception as e:
|
|
print(f"[Evaluator] Expert panel [{exp['role']}] error: {e}")
|
|
results.append({
|
|
"role": exp["role"],
|
|
"grade": "C",
|
|
"comment": "평가 데이터가 부족합니다.",
|
|
"suggestion": "더 많은 거래 데이터 축적 후 재평가를 권장합니다."
|
|
})
|
|
|
|
return results
|
|
|
|
# ─────────────────────────────────────────
|
|
# 4. 주간 보고서 생성
|
|
# ─────────────────────────────────────────
|
|
|
|
def generate_weekly_report(self):
|
|
"""주간 성과 보고서 (텔레그램 HTML 형식) 반환."""
|
|
snapshots = self.perf_db.load_snapshots(days=7)
|
|
# 매매 완료 건은 30일치 사용 (주간 거래 수가 적을 수 있음)
|
|
trades = self.perf_db.load_trades(days=30)
|
|
|
|
metrics = self.compute_metrics(snapshots, trades)
|
|
|
|
if "error" in metrics:
|
|
return (
|
|
f"<b>[주간 성과 평가 보고서]</b>\n"
|
|
f"⚠️ {metrics['error']}\n"
|
|
f"<i>매일 오전 09:05~09:15에 스냅샷이 저장됩니다.</i>"
|
|
)
|
|
|
|
# 등급 계산
|
|
g_sharpe = self.get_grade("sharpe_ratio", metrics.get("sharpe_ratio", 0))
|
|
g_win = self.get_grade("win_rate_pct", metrics.get("win_rate_pct", 0))
|
|
g_mdd = self.get_grade("max_drawdown_pct", metrics.get("max_drawdown_pct", 0))
|
|
g_alpha = self.get_grade("alpha", metrics.get("alpha", 0))
|
|
g_weekly = self.get_grade("weekly_return_pct", metrics.get("weekly_return_pct", 0))
|
|
g_lstm = self.get_grade("lstm_direction_accuracy",
|
|
metrics.get("lstm_direction_accuracy", 0))
|
|
|
|
# 종합 등급 (Sharpe/Win/MDD/Alpha 평균)
|
|
grade_map = {"S": 5, "A": 4, "B": 3, "C": 2, "D": 1, "F": 0}
|
|
grade_rev = {v: k for k, v in grade_map.items()}
|
|
key_grades = [grade_map[g] for g in [g_sharpe, g_win, g_mdd, g_alpha]]
|
|
overall_grade = grade_rev[round(sum(key_grades) / len(key_grades))]
|
|
|
|
# 전문가 패널 (Ollama 호출)
|
|
try:
|
|
experts = self.generate_expert_panel(metrics)
|
|
except Exception as e:
|
|
print(f"[Evaluator] Expert panel skipped: {e}")
|
|
experts = []
|
|
|
|
now_str = datetime.now().strftime("%Y/%m/%d %H:%M")
|
|
corr = metrics.get("signal_correlation", {})
|
|
|
|
report = (
|
|
f"📊 <b>[주간 성과 평가 보고서]</b> <code>{now_str}</code>\n"
|
|
f"━━━━━━━━━━━━━━━━━━━━━━\n"
|
|
f"\n<b>■ 수익률</b>\n"
|
|
f" 주간: <code>{metrics.get('weekly_return_pct', 0):+.2f}%</code> [{g_weekly}]"
|
|
f" 누적: <code>{metrics.get('total_return_pct', 0):+.2f}%</code>\n"
|
|
f" Alpha: <code>{metrics.get('alpha', 0):+.2f}%</code> [{g_alpha}]"
|
|
f" vs KOSPI <code>{metrics.get('kospi_return_pct', 0):+.2f}%</code>\n"
|
|
f"\n<b>■ 리스크</b>\n"
|
|
f" Sharpe: <code>{metrics.get('sharpe_ratio', 0):.2f}</code> [{g_sharpe}]"
|
|
f" Sortino: <code>{metrics.get('sortino_ratio', 0):.2f}</code>\n"
|
|
f" MDD: <code>{metrics.get('max_drawdown_pct', 0):.1f}%</code> [{g_mdd}]"
|
|
f" Calmar: <code>{metrics.get('calmar_ratio', 0):.2f}</code>\n"
|
|
f"\n<b>■ 매매 통계</b>\n"
|
|
f" 승률: <code>{metrics.get('win_rate_pct', 0):.1f}%</code> [{g_win}]"
|
|
f" PF: <code>{metrics.get('profit_factor', 0):.2f}</code>\n"
|
|
f" 평균보유: <code>{metrics.get('avg_holding_days', 0):.1f}일</code>"
|
|
f" 완료매매: <code>{metrics.get('total_trades', 0)}건</code>\n"
|
|
f"\n<b>■ AI 품질</b>\n"
|
|
f" LSTM 방향정확도: <code>{metrics.get('lstm_direction_accuracy', 0):.1f}%</code>"
|
|
f" [{g_lstm}]\n"
|
|
f" 신호 상관도 — Tech: <code>{corr.get('tech', 0):.3f}</code>"
|
|
f" Sent: <code>{corr.get('sentiment', 0):.3f}</code>"
|
|
f" LSTM: <code>{corr.get('lstm', 0):.3f}</code>\n"
|
|
f" 최고기여 신호: <code>{metrics.get('best_signal_source', 'N/A')}</code>\n"
|
|
)
|
|
|
|
if experts:
|
|
role_icons = {
|
|
"Risk Manager": "🛡",
|
|
"Fund Manager": "💼",
|
|
"Quant Analyst": "🧮",
|
|
"Trader": "📈",
|
|
"Portfolio PM": "🏦"
|
|
}
|
|
report += "\n<b>■ 전문가 패널 의견</b>\n"
|
|
for exp in experts:
|
|
icon = role_icons.get(exp["role"], "👤")
|
|
report += (
|
|
f"{icon} <b>{exp['role']}</b> [{exp['grade']}]\n"
|
|
f" {exp['comment']}\n"
|
|
f" 💡 {exp['suggestion']}\n"
|
|
)
|
|
|
|
report += (
|
|
f"\n━━━━━━━━━━━━━━━━━━━━━━\n"
|
|
f"🏆 <b>종합 등급: [{overall_grade}]</b>\n"
|
|
f"<i>스냅샷 {metrics.get('snapshot_count', 0)}일 | 완료매매 {metrics.get('total_trades', 0)}건 기준</i>"
|
|
)
|
|
|
|
return report
|