262 lines
9.4 KiB
Python
262 lines
9.4 KiB
Python
"""
|
||
评估与报告模块 — 绩效评估指标、风险分析、可视化报告
|
||
"""
|
||
from dataclasses import dataclass
|
||
from typing import Dict, List, Optional, Tuple
|
||
|
||
import numpy as np
|
||
import pandas as pd
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 绩效评估器
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class PerformanceEvaluator:
|
||
"""
|
||
绩效评估器 — 计算常见量化指标
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
equity_curve: pd.DataFrame,
|
||
benchmark_returns: Optional[pd.Series] = None,
|
||
risk_free_rate: float = 0.03,
|
||
):
|
||
"""
|
||
equity_curve : DataFrame, 必须包含 'return' 或 'nav' 列, index=date
|
||
benchmark_returns : Series, 基准日收益率, index=date
|
||
risk_free_rate : 年化无风险利率
|
||
"""
|
||
self.equity = equity_curve.copy()
|
||
self.rf = risk_free_rate
|
||
|
||
# 确保有日收益率列
|
||
if "daily_return" not in self.equity.columns:
|
||
if "return" in self.equity.columns:
|
||
self.equity["daily_return"] = self.equity["return"].diff().fillna(0)
|
||
elif "nav" in self.equity.columns:
|
||
self.equity["daily_return"] = self.equity["nav"].pct_change().fillna(0)
|
||
else:
|
||
raise ValueError("equity_curve 必须包含 'return'/'nav' 列")
|
||
|
||
self.benchmark_returns = benchmark_returns
|
||
self._metrics: Dict[str, float] = {}
|
||
|
||
# ----- 基础指标 -----
|
||
|
||
def total_return(self) -> float:
|
||
"""累计收益率"""
|
||
if "nav" in self.equity.columns:
|
||
return self.equity["nav"].iloc[-1] - 1.0
|
||
return (1 + self.equity["daily_return"]).prod() - 1.0
|
||
|
||
def annual_return(self, periods_per_year: int = 252) -> float:
|
||
"""年化收益率"""
|
||
total = self.total_return()
|
||
years = self._year_frac(periods_per_year)
|
||
return (1 + total) ** (1 / years) - 1.0 if years > 0 else 0.0
|
||
|
||
def annual_volatility(self, periods_per_year: int = 252) -> float:
|
||
"""年化波动率"""
|
||
return self.equity["daily_return"].std() * np.sqrt(periods_per_year)
|
||
|
||
def max_drawdown(self) -> float:
|
||
"""最大回撤"""
|
||
if "nav" in self.equity.columns:
|
||
nav = self.equity["nav"]
|
||
else:
|
||
nav = (1 + self.equity["daily_return"]).cumprod()
|
||
running_max = nav.cummax()
|
||
drawdown = (nav - running_max) / running_max
|
||
return drawdown.min()
|
||
|
||
def max_drawdown_duration(self) -> int:
|
||
"""最长回撤持续天数"""
|
||
if "nav" in self.equity.columns:
|
||
nav = self.equity["nav"]
|
||
else:
|
||
nav = (1 + self.equity["daily_return"]).cumprod()
|
||
running_max = nav.cummax()
|
||
is_dd = nav < running_max
|
||
# 统计最长连续 True 段
|
||
dd_groups = (is_dd != is_dd.shift()).cumsum()
|
||
dd_durations = (
|
||
is_dd.groupby(dd_groups).sum()
|
||
)
|
||
return int(dd_durations.max()) if not dd_durations.empty else 0
|
||
|
||
# ----- 风险调整收益 -----
|
||
|
||
def sharpe_ratio(self, periods_per_year: int = 252) -> float:
|
||
"""夏普比率"""
|
||
excess = self.equity["daily_return"] - self.rf / periods_per_year
|
||
return (
|
||
excess.mean() / (excess.std() + 1e-12) * np.sqrt(periods_per_year)
|
||
)
|
||
|
||
def calmar_ratio(self) -> float:
|
||
"""Calmar 比率 (年化收益 / 最大回撤绝对值)"""
|
||
mdd = abs(self.max_drawdown())
|
||
return self.annual_return() / mdd if mdd > 0 else 0.0
|
||
|
||
def sortino_ratio(self, periods_per_year: int = 252) -> float:
|
||
"""Sortino 比率 (只考虑下行波动)"""
|
||
returns = self.equity["daily_return"]
|
||
downside = returns[returns < 0]
|
||
downside_vol = downside.std() * np.sqrt(periods_per_year)
|
||
excess_annual = self.annual_return() - self.rf
|
||
return excess_annual / (downside_vol + 1e-12)
|
||
|
||
def win_rate(self) -> float:
|
||
"""胜率（日收益率 > 0 的比例）"""
|
||
returns = self.equity["daily_return"]
|
||
return (returns > 0).mean()
|
||
|
||
def profit_loss_ratio(self) -> float:
|
||
"""盈亏比 (平均盈利 / 平均亏损绝对值)"""
|
||
returns = self.equity["daily_return"]
|
||
avg_win = returns[returns > 0].mean()
|
||
avg_loss = abs(returns[returns < 0].mean())
|
||
return avg_win / (avg_loss + 1e-12)
|
||
|
||
# ----- 相对基准指标 -----
|
||
|
||
def alpha(self, periods_per_year: int = 252) -> float:
|
||
"""Jensen's Alpha 相对基准"""
|
||
if self.benchmark_returns is None:
|
||
return 0.0
|
||
aligned = self._align_benchmark()
|
||
strat_ret = aligned["strat"]
|
||
bench_ret = aligned["bench"]
|
||
excess_strat = strat_ret - self.rf / periods_per_year
|
||
excess_bench = bench_ret - self.rf / periods_per_year
|
||
# 线性回归
|
||
cov = np.cov(excess_strat, excess_bench)
|
||
beta = cov[0, 1] / (cov[1, 1] + 1e-12)
|
||
alpha_daily = excess_strat.mean() - beta * excess_bench.mean()
|
||
return alpha_daily * periods_per_year
|
||
|
||
def beta(self) -> float:
|
||
"""Beta"""
|
||
if self.benchmark_returns is None:
|
||
return 1.0
|
||
aligned = self._align_benchmark()
|
||
cov = np.cov(aligned["strat"], aligned["bench"])
|
||
return cov[0, 1] / (cov[1, 1] + 1e-12)
|
||
|
||
def information_ratio(self, periods_per_year: int = 252) -> float:
|
||
"""信息比率"""
|
||
if self.benchmark_returns is None:
|
||
return 0.0
|
||
aligned = self._align_benchmark()
|
||
tracking_error = (aligned["strat"] - aligned["bench"]).std()
|
||
return (
|
||
(aligned["strat"].mean() - aligned["bench"].mean())
|
||
/ (tracking_error + 1e-12)
|
||
* np.sqrt(periods_per_year)
|
||
)
|
||
|
||
# ----- 报告 -----
|
||
|
||
def full_report(self) -> Dict[str, float]:
|
||
"""生成完整绩效报告字典"""
|
||
self._metrics = {
|
||
"累计收益率(%)": self.total_return() * 100,
|
||
"年化收益率(%)": self.annual_return() * 100,
|
||
"年化波动率(%)": self.annual_volatility() * 100,
|
||
"最大回撤(%)": self.max_drawdown() * 100,
|
||
"最长回撤天数": self.max_drawdown_duration(),
|
||
"夏普比率": self.sharpe_ratio(),
|
||
"Calmar比率": self.calmar_ratio(),
|
||
"Sortino比率": self.sortino_ratio(),
|
||
"日胜率(%)": self.win_rate() * 100,
|
||
"盈亏比": self.profit_loss_ratio(),
|
||
"Alpha(%)": self.alpha() * 100,
|
||
"Beta": self.beta(),
|
||
"信息比率": self.information_ratio(),
|
||
}
|
||
return self._metrics
|
||
|
||
def summary(self) -> str:
|
||
"""打印可读报告"""
|
||
report = self.full_report()
|
||
lines = ["=" * 50, " 量化策略绩效评估报告", "=" * 50]
|
||
for key, value in report.items():
|
||
if isinstance(value, float):
|
||
lines.append(f" {key}: {value:.4f}")
|
||
else:
|
||
lines.append(f" {key}: {value}")
|
||
lines.append("=" * 50)
|
||
return "\n".join(lines)
|
||
|
||
# ----- 内部方法 -----
|
||
|
||
def _year_frac(self, periods_per_year: int) -> float:
|
||
n = len(self.equity)
|
||
return n / periods_per_year
|
||
|
||
def _align_benchmark(self) -> pd.DataFrame:
|
||
"""对齐策略与基准日收益率"""
|
||
strat = self.equity["daily_return"]
|
||
bench = self.benchmark_returns
|
||
merged = pd.concat([strat, bench], axis=1, join="inner").dropna()
|
||
merged.columns = ["strat", "bench"]
|
||
return merged
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 报告生成器
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class ReportGenerator:
|
||
"""生成交易分析报告（Markdown / Excel）"""
|
||
|
||
def __init__(self, evaluator: PerformanceEvaluator, trades: Optional[List] = None):
|
||
self.evaluator = evaluator
|
||
self.trades = trades or []
|
||
|
||
def to_markdown(self) -> str:
|
||
"""输出 Markdown 格式报告"""
|
||
metrics = self.evaluator.full_report()
|
||
lines = [
|
||
"# 量化策略回测报告",
|
||
"",
|
||
"## 绩效指标",
|
||
"",
|
||
"| 指标 | 数值 |",
|
||
"|------|------|",
|
||
]
|
||
for key, value in metrics.items():
|
||
lines.append(f"| {key} | {value:.4f} |")
|
||
|
||
# 交易统计
|
||
if self.trades:
|
||
lines.extend(["", "## 交易统计", ""])
|
||
trades_df = pd.DataFrame(
|
||
[
|
||
{
|
||
"date": t.date,
|
||
"stock": t.stock,
|
||
"side": t.side,
|
||
"quantity": t.quantity,
|
||
"price": t.price,
|
||
}
|
||
for t in self.trades
|
||
]
|
||
)
|
||
total_trades = len(trades_df)
|
||
buy_trades = len(trades_df[trades_df["side"] == "buy"])
|
||
sell_trades = len(trades_df[trades_df["side"] == "sell"])
|
||
lines.append(f"- 总交易笔数: {total_trades}")
|
||
lines.append(f"- 买入笔数: {buy_trades}")
|
||
lines.append(f"- 卖出笔数: {sell_trades}")
|
||
|
||
return "\n".join(lines)
|
||
|
||
def to_dataframe(self) -> pd.DataFrame:
|
||
"""将绩效指标导出为 DataFrame"""
|
||
metrics = self.evaluator.full_report()
|
||
return pd.DataFrame(
|
||
list(metrics.items()), columns=["指标", "数值"]
|
||
).set_index("指标") |