146 lines
5.2 KiB
Python
146 lines
5.2 KiB
Python
"""绩效计算模块。
|
||
|
||
根据资金曲线计算收益、年化收益、夏普比率、最大回撤等指标。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import numpy as np
|
||
import pandas as pd
|
||
|
||
from utils.logger import setup_logger
|
||
|
||
logger = setup_logger(__name__)
|
||
|
||
|
||
def calc_performance(
|
||
equity_df: pd.DataFrame,
|
||
trade_count: int = 0,
|
||
trade_history: list = None,
|
||
trading_days_per_year: int = 252,
|
||
) -> dict:
|
||
"""根据资金曲线计算常用绩效指标。
|
||
|
||
参数：
|
||
- equity_df: 包含列 ['trade_date', 'total_asset', 'cash', 'market_value'] 的 DataFrame；
|
||
- trade_count: 总交易次数（买入+卖出）；
|
||
- trade_history: 交易历史记录（用于计算胜率和盈亏比）；
|
||
- trading_days_per_year: 年化使用的交易日数，默认 252。
|
||
|
||
返回：
|
||
- dict，包含累积收益、年化收益、夏普比率、最大回撤、资金利用率、胜率、盈亏比等。
|
||
"""
|
||
if equity_df.empty:
|
||
logger.warning("资金曲线为空，无法计算绩效")
|
||
return {}
|
||
|
||
df = equity_df.copy()
|
||
df = df.sort_values("trade_date").reset_index(drop=True)
|
||
df["ret"] = df["total_asset"].pct_change().fillna(0.0)
|
||
|
||
# 累积收益
|
||
cum_return = df["total_asset"].iloc[-1] / df["total_asset"].iloc[0] - 1
|
||
|
||
# 年化收益
|
||
n = len(df)
|
||
if n <= 1:
|
||
ann_return = 0.0
|
||
years = 0.0
|
||
else:
|
||
ann_return = (1 + cum_return) ** (trading_days_per_year / n) - 1
|
||
years = n / trading_days_per_year
|
||
|
||
# 夏普比率（假设无无风险利率）
|
||
ret_mean = df["ret"].mean()
|
||
ret_std = df["ret"].std(ddof=1)
|
||
if ret_std == 0:
|
||
sharpe = 0.0
|
||
else:
|
||
sharpe = (ret_mean * trading_days_per_year) / (ret_std * (trading_days_per_year**0.5))
|
||
|
||
# 最大回撤
|
||
cummax = df["total_asset"].cummax()
|
||
drawdown = df["total_asset"] / cummax - 1
|
||
max_drawdown = float(drawdown.min())
|
||
|
||
# 资金利用率统计（每日持仓市值 / 总资产）
|
||
if "market_value" in df.columns and "total_asset" in df.columns:
|
||
df["capital_utilization"] = df["market_value"] / df["total_asset"]
|
||
avg_capital_utilization = df["capital_utilization"].mean()
|
||
else:
|
||
avg_capital_utilization = 0.0
|
||
|
||
# 交易次数统计
|
||
total_trades = trade_count
|
||
if years > 0:
|
||
avg_trades_per_year = total_trades / years
|
||
else:
|
||
avg_trades_per_year = 0.0
|
||
|
||
# 计算胜率和盈亏比（从交易历史中获取）
|
||
win_rate = 0.0
|
||
profit_loss_ratio = 0.0
|
||
win_count = 0
|
||
loss_count = 0
|
||
total_win_pct = 0.0
|
||
total_loss_pct = 0.0
|
||
|
||
if trade_history and len(trade_history) > 0:
|
||
for trade in trade_history:
|
||
if trade.get("is_win", False):
|
||
win_count += 1
|
||
total_win_pct += trade.get("profit_pct", 0.0)
|
||
else:
|
||
loss_count += 1
|
||
total_loss_pct += abs(trade.get("profit_pct", 0.0))
|
||
|
||
total_complete_trades = win_count + loss_count
|
||
if total_complete_trades > 0:
|
||
win_rate = win_count / total_complete_trades
|
||
|
||
# 计算平均盈亏比：平均盈利 / 平均亏损
|
||
avg_win = total_win_pct / win_count if win_count > 0 else 0.0
|
||
avg_loss = total_loss_pct / loss_count if loss_count > 0 else 0.0
|
||
|
||
if avg_loss > 0:
|
||
profit_loss_ratio = avg_win / avg_loss
|
||
else:
|
||
profit_loss_ratio = 0.0 if avg_win == 0 else float('inf')
|
||
|
||
res = {
|
||
"cum_return": float(cum_return),
|
||
"ann_return": float(ann_return),
|
||
"sharpe": float(sharpe),
|
||
"max_drawdown": max_drawdown,
|
||
"avg_capital_utilization": float(avg_capital_utilization),
|
||
"total_trades": int(total_trades),
|
||
"avg_trades_per_year": float(avg_trades_per_year),
|
||
"backtest_years": float(years),
|
||
"win_rate": float(win_rate),
|
||
"profit_loss_ratio": float(profit_loss_ratio),
|
||
"win_count": int(win_count),
|
||
"loss_count": int(loss_count),
|
||
}
|
||
|
||
# 格式化输出绩效指标（中文、分行、百分比）
|
||
logger.info("=" * 60)
|
||
logger.info("回测绩效指标汇总")
|
||
logger.info("=" * 60)
|
||
logger.info(f"回测年数: {years:.2f} 年")
|
||
logger.info(f"总交易次数: {total_trades} 次")
|
||
logger.info(f"年平均交易次数: {avg_trades_per_year:.2f} 次/年")
|
||
logger.info("-" * 60)
|
||
logger.info(f"累计收益率: {cum_return * 100:+.2f}%")
|
||
logger.info(f"年化收益率: {ann_return * 100:+.2f}%")
|
||
logger.info(f"夏普比率: {sharpe:.4f}")
|
||
logger.info(f"最大回撤: {max_drawdown * 100:.2f}%")
|
||
logger.info(f"平均资金利用率: {avg_capital_utilization * 100:.2f}%")
|
||
logger.info("-" * 60)
|
||
logger.info(f"胜率: {win_rate * 100:.2f}% ({win_count}胜 / {loss_count}败)")
|
||
if profit_loss_ratio == float('inf'):
|
||
logger.info(f"平均盈亏比: ∞ (无亏损交易)")
|
||
else:
|
||
logger.info(f"平均盈亏比: {profit_loss_ratio:.2f}")
|
||
logger.info("=" * 60)
|
||
|
||
return res
|