453 lines
15 KiB
Python
453 lines
15 KiB
Python
# -*- coding: utf-8 -*-
|
||
|
||
"""
|
||
绩效分析模块
|
||
计算各种回测绩效指标，如收益率、夏普比率、最大回撤等
|
||
"""
|
||
|
||
import pandas as pd
|
||
import numpy as np
|
||
import logging
|
||
from scipy import stats
|
||
|
||
|
||
class PerformanceAnalyzer:
|
||
"""
|
||
绩效分析器
|
||
提供全面的回测结果绩效分析功能
|
||
"""
|
||
|
||
def __init__(self):
|
||
"""
|
||
初始化绩效分析器
|
||
"""
|
||
self.logger = logging.getLogger('performance_analyzer')
|
||
# 设置无风险利率（年化）
|
||
self.risk_free_rate = 0.02 # 假设无风险利率为2%
|
||
|
||
def calculate_metrics(self, portfolio):
|
||
"""
|
||
计算绩效指标
|
||
|
||
Args:
|
||
portfolio: 投资组合DataFrame，包含收益率等数据
|
||
|
||
Returns:
|
||
dict: 包含各种绩效指标的字典
|
||
"""
|
||
self.logger.info('开始计算绩效指标...')
|
||
|
||
metrics = {}
|
||
|
||
try:
|
||
# 计算基本收益率指标
|
||
metrics['total_return'] = self._calculate_total_return(portfolio)
|
||
metrics['annual_return'] = self._calculate_annual_return(portfolio)
|
||
metrics['daily_returns'] = portfolio['return'].tolist()
|
||
metrics['cumulative_returns'] = portfolio['cumulative_return'].tolist()
|
||
|
||
# 计算风险指标
|
||
metrics['volatility'] = self._calculate_volatility(portfolio)
|
||
metrics['annual_volatility'] = self._calculate_annual_volatility(portfolio)
|
||
metrics['max_drawdown'] = self._calculate_max_drawdown(portfolio)
|
||
metrics['max_drawdown_duration'] = self._calculate_max_drawdown_duration(portfolio)
|
||
|
||
# 计算风险调整收益指标
|
||
metrics['sharpe_ratio'] = self._calculate_sharpe_ratio(portfolio)
|
||
metrics['sortino_ratio'] = self._calculate_sortino_ratio(portfolio)
|
||
metrics['calmar_ratio'] = self._calculate_calmar_ratio(portfolio)
|
||
|
||
# 计算交易相关指标（如果有交易记录）
|
||
if 'signal' in portfolio.columns:
|
||
trade_metrics = self._calculate_trade_metrics(portfolio)
|
||
metrics.update(trade_metrics)
|
||
|
||
# 计算其他统计指标
|
||
metrics['skewness'] = self._calculate_skewness(portfolio)
|
||
metrics['kurtosis'] = self._calculate_kurtosis(portfolio)
|
||
metrics['beta'] = self._calculate_beta(portfolio) # 需要基准数据
|
||
|
||
self.logger.info('绩效指标计算完成')
|
||
|
||
except Exception as e:
|
||
self.logger.error(f'计算绩效指标时出错: {str(e)}')
|
||
# 至少返回基本指标
|
||
metrics['total_return'] = self._calculate_total_return(portfolio)
|
||
metrics['max_drawdown'] = self._calculate_max_drawdown(portfolio)
|
||
|
||
return metrics
|
||
|
||
def _calculate_total_return(self, portfolio):
|
||
"""
|
||
计算总收益率
|
||
|
||
Args:
|
||
portfolio: 投资组合数据
|
||
|
||
Returns:
|
||
float: 总收益率
|
||
"""
|
||
initial_value = portfolio['total'].iloc[0]
|
||
final_value = portfolio['total'].iloc[-1]
|
||
return (final_value - initial_value) / initial_value
|
||
|
||
def _calculate_annual_return(self, portfolio):
|
||
"""
|
||
计算年化收益率
|
||
|
||
Args:
|
||
portfolio: 投资组合数据
|
||
|
||
Returns:
|
||
float: 年化收益率
|
||
"""
|
||
total_return = self._calculate_total_return(portfolio)
|
||
# 计算交易日数量
|
||
trading_days = len(portfolio)
|
||
# 假设一年252个交易日
|
||
annual_factor = 252.0 / trading_days
|
||
return (1 + total_return) ** annual_factor - 1
|
||
|
||
def _calculate_volatility(self, portfolio):
|
||
"""
|
||
计算收益率波动率
|
||
|
||
Args:
|
||
portfolio: 投资组合数据
|
||
|
||
Returns:
|
||
float: 波动率
|
||
"""
|
||
return portfolio['return'].std()
|
||
|
||
def _calculate_annual_volatility(self, portfolio):
|
||
"""
|
||
计算年化波动率
|
||
|
||
Args:
|
||
portfolio: 投资组合数据
|
||
|
||
Returns:
|
||
float: 年化波动率
|
||
"""
|
||
daily_vol = self._calculate_volatility(portfolio)
|
||
return daily_vol * np.sqrt(252) # 年化
|
||
|
||
def _calculate_max_drawdown(self, portfolio):
|
||
"""
|
||
计算最大回撤
|
||
|
||
Args:
|
||
portfolio: 投资组合数据
|
||
|
||
Returns:
|
||
float: 最大回撤（负值）
|
||
"""
|
||
# 计算累计最大值
|
||
cumulative_max = portfolio['total'].cummax()
|
||
# 计算回撤
|
||
drawdown = (portfolio['total'] - cumulative_max) / cumulative_max
|
||
# 返回最大回撤
|
||
return drawdown.min()
|
||
|
||
def _calculate_max_drawdown_duration(self, portfolio):
|
||
"""
|
||
计算最大回撤持续时间
|
||
|
||
Args:
|
||
portfolio: 投资组合数据
|
||
|
||
Returns:
|
||
int: 最大回撤持续天数
|
||
"""
|
||
cumulative_max = portfolio['total'].cummax()
|
||
drawdown = (portfolio['total'] - cumulative_max) / cumulative_max
|
||
|
||
# 找出回撤期间
|
||
in_drawdown = False
|
||
current_duration = 0
|
||
max_duration = 0
|
||
|
||
for i, dd in enumerate(drawdown):
|
||
if dd < 0 and not in_drawdown:
|
||
in_drawdown = True
|
||
current_duration = 1
|
||
elif dd < 0 and in_drawdown:
|
||
current_duration += 1
|
||
max_duration = max(max_duration, current_duration)
|
||
elif dd == 0 and in_drawdown:
|
||
in_drawdown = False
|
||
current_duration = 0
|
||
|
||
return max_duration
|
||
|
||
def _calculate_sharpe_ratio(self, portfolio):
|
||
"""
|
||
计算夏普比率
|
||
|
||
Args:
|
||
portfolio: 投资组合数据
|
||
|
||
Returns:
|
||
float: 夏普比率
|
||
"""
|
||
# 计算超额收益
|
||
excess_returns = portfolio['return'] - (self.risk_free_rate / 252)
|
||
# 计算夏普比率
|
||
if excess_returns.std() == 0:
|
||
return 0
|
||
sharpe = excess_returns.mean() / excess_returns.std() * np.sqrt(252)
|
||
return sharpe
|
||
|
||
def _calculate_sortino_ratio(self, portfolio):
|
||
"""
|
||
计算索提诺比率（只考虑下行风险）
|
||
|
||
Args:
|
||
portfolio: 投资组合数据
|
||
|
||
Returns:
|
||
float: 索提诺比率
|
||
"""
|
||
# 计算超额收益
|
||
excess_returns = portfolio['return'] - (self.risk_free_rate / 252)
|
||
# 计算下行风险（仅考虑负收益）
|
||
downside_returns = excess_returns[excess_returns < 0]
|
||
if len(downside_returns) == 0:
|
||
return float('inf')
|
||
|
||
downside_risk = np.sqrt((downside_returns ** 2).mean())
|
||
if downside_risk == 0:
|
||
return 0
|
||
|
||
sortino = excess_returns.mean() / downside_risk * np.sqrt(252)
|
||
return sortino
|
||
|
||
def _calculate_calmar_ratio(self, portfolio):
|
||
"""
|
||
计算卡尔马比率（年化收益/最大回撤）
|
||
|
||
Args:
|
||
portfolio: 投资组合数据
|
||
|
||
Returns:
|
||
float: 卡尔马比率
|
||
"""
|
||
annual_return = self._calculate_annual_return(portfolio)
|
||
max_drawdown = abs(self._calculate_max_drawdown(portfolio))
|
||
|
||
if max_drawdown == 0:
|
||
return float('inf')
|
||
|
||
return annual_return / max_drawdown
|
||
|
||
def _calculate_trade_metrics(self, portfolio):
|
||
"""
|
||
计算交易相关指标
|
||
|
||
Args:
|
||
portfolio: 包含信号的投资组合数据
|
||
|
||
Returns:
|
||
dict: 交易指标
|
||
"""
|
||
# 找出买入和卖出信号
|
||
buy_signals = portfolio[portfolio['signal'] == 1]
|
||
sell_signals = portfolio[portfolio['signal'] == -1]
|
||
|
||
# 计算交易次数
|
||
total_trades = len(buy_signals) + len(sell_signals)
|
||
|
||
# 计算胜率（简化版本，实际应基于完整交易）
|
||
win_rate = 0
|
||
avg_return_per_trade = 0
|
||
|
||
if total_trades > 0:
|
||
# 计算每次交易的收益（简化计算）
|
||
trade_returns = []
|
||
position = 0
|
||
entry_price = None
|
||
|
||
for date, row in portfolio.iterrows():
|
||
if row['signal'] == 1 and position == 0:
|
||
entry_price = row['price']
|
||
position = 1
|
||
elif row['signal'] == -1 and position == 1:
|
||
exit_price = row['price']
|
||
trade_return = (exit_price - entry_price) / entry_price
|
||
trade_returns.append(trade_return)
|
||
position = 0
|
||
|
||
if trade_returns:
|
||
winning_trades = sum(1 for r in trade_returns if r > 0)
|
||
win_rate = winning_trades / len(trade_returns)
|
||
avg_return_per_trade = np.mean(trade_returns)
|
||
|
||
return {
|
||
'total_trades': total_trades,
|
||
'buy_signals': len(buy_signals),
|
||
'sell_signals': len(sell_signals),
|
||
'win_rate': win_rate,
|
||
'avg_return_per_trade': avg_return_per_trade
|
||
}
|
||
|
||
def _calculate_skewness(self, portfolio):
|
||
"""
|
||
计算收益率偏度
|
||
|
||
Args:
|
||
portfolio: 投资组合数据
|
||
|
||
Returns:
|
||
float: 偏度
|
||
"""
|
||
return stats.skew(portfolio['return'].dropna())
|
||
|
||
def _calculate_kurtosis(self, portfolio):
|
||
"""
|
||
计算收益率峰度
|
||
|
||
Args:
|
||
portfolio: 投资组合数据
|
||
|
||
Returns:
|
||
float: 峰度
|
||
"""
|
||
# 使用Fisher峰度（正态分布的峰度为0）
|
||
return stats.kurtosis(portfolio['return'].dropna(), fisher=True)
|
||
|
||
def _calculate_beta(self, portfolio, benchmark_returns=None):
|
||
"""
|
||
计算贝塔系数
|
||
|
||
Args:
|
||
portfolio: 投资组合数据
|
||
benchmark_returns: 基准收益率序列（如果不提供则返回None）
|
||
|
||
Returns:
|
||
float: 贝塔系数
|
||
"""
|
||
if benchmark_returns is None:
|
||
# 如果没有基准数据，尝试使用自身作为基准（简化处理）
|
||
# 实际应用中应该传入市场基准数据
|
||
return 1.0
|
||
|
||
# 确保两个序列长度相同
|
||
min_len = min(len(portfolio['return']), len(benchmark_returns))
|
||
if min_len < 2:
|
||
return 1.0
|
||
|
||
# 计算协方差和方差
|
||
cov_matrix = np.cov(portfolio['return'][-min_len:], benchmark_returns[-min_len:])
|
||
beta = cov_matrix[0, 1] / cov_matrix[1, 1] if cov_matrix[1, 1] != 0 else 1.0
|
||
|
||
return beta
|
||
|
||
def calculate_rolling_metrics(self, portfolio, window=20):
|
||
"""
|
||
计算滚动绩效指标
|
||
|
||
Args:
|
||
portfolio: 投资组合数据
|
||
window: 滚动窗口大小
|
||
|
||
Returns:
|
||
pandas.DataFrame: 滚动绩效指标
|
||
"""
|
||
rolling_metrics = pd.DataFrame(index=portfolio.index)
|
||
|
||
# 计算滚动收益率
|
||
rolling_metrics['rolling_return'] = portfolio['return'].rolling(window=window).mean()
|
||
|
||
# 计算滚动波动率
|
||
rolling_metrics['rolling_volatility'] = portfolio['return'].rolling(window=window).std()
|
||
|
||
# 计算滚动夏普比率（简化版）
|
||
excess_returns = portfolio['return'] - (self.risk_free_rate / 252)
|
||
rolling_metrics['rolling_sharpe'] = (
|
||
excess_returns.rolling(window=window).mean() /
|
||
excess_returns.rolling(window=window).std() *
|
||
np.sqrt(252)
|
||
)
|
||
|
||
return rolling_metrics
|
||
|
||
def generate_performance_report(self, metrics, portfolio, output_file=None):
|
||
"""
|
||
生成绩效报告
|
||
|
||
Args:
|
||
metrics: 绩效指标字典
|
||
portfolio: 投资组合数据
|
||
output_file: 输出文件路径
|
||
|
||
Returns:
|
||
str: 绩效报告内容
|
||
"""
|
||
# 生成报告内容
|
||
report = []
|
||
report.append('=' * 60)
|
||
report.append('回测绩效报告')
|
||
report.append('=' * 60)
|
||
|
||
# 基本信息
|
||
report.append('\n1. 基本信息')
|
||
report.append('-' * 30)
|
||
report.append(f"回测期间: {portfolio.index[0].strftime('%Y-%m-%d')} 至 {portfolio.index[-1].strftime('%Y-%m-%d')}")
|
||
report.append(f"交易天数: {len(portfolio)} 天")
|
||
report.append(f"初始资金: {portfolio['total'].iloc[0]:,.2f}")
|
||
report.append(f"最终资金: {portfolio['total'].iloc[-1]:,.2f}")
|
||
|
||
# 收益率指标
|
||
report.append('\n2. 收益率指标')
|
||
report.append('-' * 30)
|
||
report.append(f"总收益率: {metrics.get('total_return', 0) * 100:.2f}%")
|
||
report.append(f"年化收益率: {metrics.get('annual_return', 0) * 100:.2f}%")
|
||
|
||
# 风险指标
|
||
report.append('\n3. 风险指标')
|
||
report.append('-' * 30)
|
||
report.append(f"年化波动率: {metrics.get('annual_volatility', 0) * 100:.2f}%")
|
||
report.append(f"最大回撤: {metrics.get('max_drawdown', 0) * 100:.2f}%")
|
||
report.append(f"最大回撤持续天数: {metrics.get('max_drawdown_duration', 0)} 天")
|
||
|
||
# 风险调整收益
|
||
report.append('\n4. 风险调整收益')
|
||
report.append('-' * 30)
|
||
report.append(f"夏普比率: {metrics.get('sharpe_ratio', 0):.4f}")
|
||
report.append(f"索提诺比率: {metrics.get('sortino_ratio', 0):.4f}")
|
||
report.append(f"卡尔马比率: {metrics.get('calmar_ratio', 0):.4f}")
|
||
|
||
# 交易统计
|
||
report.append('\n5. 交易统计')
|
||
report.append('-' * 30)
|
||
report.append(f"总交易次数: {metrics.get('total_trades', 0)}")
|
||
report.append(f"买入信号: {metrics.get('buy_signals', 0)}")
|
||
report.append(f"卖出信号: {metrics.get('sell_signals', 0)}")
|
||
report.append(f"胜率: {metrics.get('win_rate', 0) * 100:.2f}%")
|
||
report.append(f"平均每笔收益: {metrics.get('avg_return_per_trade', 0) * 100:.2f}%")
|
||
|
||
# 统计特征
|
||
report.append('\n6. 统计特征')
|
||
report.append('-' * 30)
|
||
report.append(f"收益率偏度: {metrics.get('skewness', 0):.4f}")
|
||
report.append(f"收益率峰度: {metrics.get('kurtosis', 0):.4f}")
|
||
report.append(f"贝塔系数: {metrics.get('beta', 0):.4f}")
|
||
|
||
report.append('\n' + '=' * 60)
|
||
report.append('报告生成完毕')
|
||
report.append('=' * 60)
|
||
|
||
report_content = '\n'.join(report)
|
||
|
||
# 保存报告到文件
|
||
if output_file:
|
||
try:
|
||
with open(output_file, 'w', encoding='utf-8') as f:
|
||
f.write(report_content)
|
||
self.logger.info(f'绩效报告已保存到: {output_file}')
|
||
except Exception as e:
|
||
self.logger.error(f'保存绩效报告时出错: {str(e)}')
|
||
|
||
return report_content
|