232 lines
10 KiB
Python
232 lines
10 KiB
Python
# result_visualizer.py
|
|
import vectorbt as vbt
|
|
import pandas as pd
|
|
import matplotlib.pyplot as plt
|
|
from strategy_executor import execute_strategy
|
|
from strategy_executor import *
|
|
|
|
# 设置中文显示
|
|
plt.rcParams['font.sans-serif'] = ['SimHei']
|
|
plt.rcParams['axes.unicode_minus'] = False
|
|
|
|
|
|
def plot_results(strategy_data, portfolio):
|
|
"""绘制结果图表"""
|
|
price_ratio = strategy_data['price_ratio']
|
|
signals = strategy_data['signals']
|
|
ratio_ma = strategy_data['ratio_ma']
|
|
upper_band = strategy_data['upper_band']
|
|
lower_band = strategy_data['lower_band']
|
|
initial_cash = strategy_data['initial_cash']
|
|
|
|
# ========== 绘制基于比率的技术分析图表 ==========
|
|
print("\n绘制基于比率的技术分析图表...")
|
|
|
|
# 创建详细的技术分析图表
|
|
fig, axes = plt.subplots(4, 1, figsize=(15, 16))
|
|
|
|
# 1. 价格比率和布林带 + 交易信号
|
|
axes[0].plot(price_ratio.index, price_ratio, label='价格比率(中芯/华虹)', linewidth=1.5, color='blue')
|
|
axes[0].plot(price_ratio.index, ratio_ma, label='移动平均', linewidth=1, alpha=0.7, color='orange')
|
|
axes[0].plot(price_ratio.index, upper_band, label='上轨', linewidth=1, alpha=0.7, linestyle='--', color='red')
|
|
axes[0].plot(price_ratio.index, lower_band, label='下轨', linewidth=1, alpha=0.7, linestyle='--', color='green')
|
|
axes[0].set_title('中芯国际-华虹半导体价格比率 (配对交易标的)')
|
|
axes[0].set_ylabel('价格比率')
|
|
axes[0].legend()
|
|
axes[0].grid(True, alpha=0.3)
|
|
|
|
# 标记交易信号
|
|
long_signals = signals[signals == 1]
|
|
short_signals = signals[signals == -1]
|
|
|
|
if len(long_signals) > 0:
|
|
axes[0].scatter(long_signals.index, price_ratio[long_signals.index],
|
|
color='green', marker='^', s=80, label='买入比率(做多中芯/做空华虹)', zorder=5)
|
|
if len(short_signals) > 0:
|
|
axes[0].scatter(short_signals.index, price_ratio[short_signals.index],
|
|
color='red', marker='v', s=80, label='卖空比率(做空中芯/做多华虹)', zorder=5)
|
|
|
|
# 2. 交易信号
|
|
axes[1].plot(signals.index, signals, label='交易信号', linewidth=2, color='purple', drawstyle='steps-post')
|
|
axes[1].axhline(y=1, color='green', linestyle='--', alpha=0.5, label='买入信号')
|
|
axes[1].axhline(y=-1, color='red', linestyle='--', alpha=0.5, label='卖出信号')
|
|
axes[1].axhline(y=0, color='gray', linestyle='-', alpha=0.3)
|
|
axes[1].set_title('交易信号时序')
|
|
axes[1].set_ylabel('信号')
|
|
axes[1].set_ylim(-1.5, 1.5)
|
|
axes[1].legend()
|
|
axes[1].grid(True, alpha=0.3)
|
|
|
|
# 3. 仓位变化
|
|
positions = signals.replace({1: '做多', -1: '做空', 0: '空仓'})
|
|
axes[2].plot(positions.index, positions, label='仓位状态', linewidth=2, color='darkorange', drawstyle='steps-post')
|
|
axes[2].set_title('仓位变化')
|
|
axes[2].set_ylabel('仓位')
|
|
axes[2].legend()
|
|
axes[2].grid(True, alpha=0.3)
|
|
|
|
# 4. 组合净值
|
|
if portfolio is not None:
|
|
portfolio_value = portfolio.value()
|
|
if len(portfolio_value) > 0:
|
|
axes[3].plot(portfolio_value.index, portfolio_value, label='组合净值', linewidth=2, color='darkblue')
|
|
# 标记初始资金线
|
|
axes[3].axhline(y=initial_cash, color='red', linestyle='--', alpha=0.7, label=f'初始资金({initial_cash})')
|
|
axes[3].set_title('基于价格比率的配对交易组合净值')
|
|
axes[3].set_ylabel('组合价值')
|
|
axes[3].set_xlabel('日期')
|
|
axes[3].legend()
|
|
axes[3].grid(True, alpha=0.3)
|
|
|
|
plt.tight_layout()
|
|
plt.show()
|
|
|
|
def print_statistics(strategy_data, portfolio):
|
|
"""打印统计结果"""
|
|
price_ratio = strategy_data['price_ratio']
|
|
signals = strategy_data['signals']
|
|
|
|
print("\n=== 基于价格比率的配对交易策略表现 ===")
|
|
|
|
if portfolio is not None:
|
|
# ========== 使用vectorbt进行专业分析 ==========
|
|
print("\n=== VectorBT 专业分析（基于价格比率） ===")
|
|
|
|
try:
|
|
# 选择第一列（也是唯一的一列）
|
|
portfolio_single = portfolio['RATIO']
|
|
|
|
print(portfolio_single.stats())
|
|
|
|
# 绘制vectorbt图表
|
|
fig = portfolio_single.plot(subplots=[
|
|
'orders', # 订单
|
|
'trade_pnl', # 交易盈亏
|
|
'cum_returns', # 累积收益
|
|
'drawdowns' # 回撤
|
|
])
|
|
fig.update_layout(
|
|
title='基于价格比率的配对交易详细分析',
|
|
height=800
|
|
)
|
|
fig.show()
|
|
except Exception as e:
|
|
print(f"详细分析绘制失败: {e}")
|
|
|
|
# ========== 打印详细统计 ==========
|
|
print("\n=== 详细统计 ===")
|
|
try:
|
|
# 使用单列统计
|
|
stats = portfolio['RATIO'].stats()
|
|
|
|
def safe_get_stat(stat_dict, key, default="N/A"):
|
|
value = stat_dict.get(key, default)
|
|
if hasattr(value, 'iloc'):
|
|
return value.iloc[0] if len(value) == 1 else value
|
|
return value
|
|
|
|
print(f"开始日期: {safe_get_stat(stats, 'Start')}")
|
|
print(f"结束日期: {safe_get_stat(stats, 'End')}")
|
|
print(f"期间: {safe_get_stat(stats, 'Period')}")
|
|
print(f"总收益率: {safe_get_stat(stats, 'Total Return [%]', 'N/A')}%")
|
|
print(f"年化收益率: {safe_get_stat(stats, 'Annual Return [%]', 'N/A')}%")
|
|
print(f"年化波动率: {safe_get_stat(stats, 'Annual Volatility [%]', 'N/A')}%")
|
|
print(f"夏普比率: {safe_get_stat(stats, 'Sharpe Ratio', 'N/A')}")
|
|
print(f"最大回撤: {safe_get_stat(stats, 'Max Drawdown [%]', 'N/A')}%")
|
|
print(f"总交易次数: {safe_get_stat(stats, 'Total Trades', 'N/A')}")
|
|
print(f"胜率: {safe_get_stat(stats, 'Win Rate [%]', 'N/A')}%")
|
|
print(f"盈亏比: {safe_get_stat(stats, 'Profit Factor', 'N/A')}")
|
|
|
|
except Exception as e:
|
|
print(f"获取详细统计时出错: {e}")
|
|
|
|
# 分析每笔交易
|
|
try:
|
|
trades_df = portfolio['RATIO'].trades.records_readable
|
|
if len(trades_df) > 0:
|
|
print(f"\n交易分析:")
|
|
print(f"总交易次数: {len(trades_df)}")
|
|
if 'Duration' in trades_df.columns:
|
|
print(f"平均持仓时间: {trades_df['Duration'].mean():.1f} 天")
|
|
if 'PnL' in trades_df.columns:
|
|
print(f"最大单笔盈利: {trades_df['PnL'].max():.2f}")
|
|
print(f"最大单笔亏损: {trades_df['PnL'].min():.2f}")
|
|
winning_trades = trades_df[trades_df['PnL'] > 0]
|
|
losing_trades = trades_df[trades_df['PnL'] < 0]
|
|
if len(winning_trades) > 0:
|
|
print(f"平均盈利: {winning_trades['PnL'].mean():.2f}")
|
|
if len(losing_trades) > 0:
|
|
print(f"平均亏损: {losing_trades['PnL'].mean():.2f}")
|
|
except Exception as e:
|
|
print(f"分析交易时出错: {e}")
|
|
|
|
# ========== 比率数据统计摘要 ==========
|
|
print("\n=== 价格比率统计摘要 ===")
|
|
print(f"数据期间: {price_ratio.index.min()} 到 {price_ratio.index.max()}")
|
|
print(f"数据点数: {len(price_ratio)}")
|
|
print(f"比率均值: {price_ratio.mean():.4f}")
|
|
print(f"比率标准差: {price_ratio.std():.4f}")
|
|
print(f"比率变异系数: {price_ratio.std()/price_ratio.mean():.4f}")
|
|
|
|
# 计算交易信号统计
|
|
long_count = (signals == 1).sum()
|
|
short_count = (signals == -1).sum()
|
|
total_signals = long_count + short_count
|
|
print(f"\n交易信号统计:")
|
|
print(f"做多信号次数: {long_count}")
|
|
print(f"做空信号次数: {short_count}")
|
|
print(f"总信号次数: {total_signals}")
|
|
|
|
def main():
|
|
"""主函数"""
|
|
# 执行策略
|
|
strategy_data = execute_strategy()
|
|
|
|
# 创建基于比率的投资组合（用于分析）
|
|
ratio_portfolio = create_portfolio(strategy_data)
|
|
|
|
# 创建基于真实股票的投资组合（用于真实收益计算）
|
|
stock_portfolio = create_real_stock_portfolio(strategy_data)
|
|
|
|
# 绘制基于比率的结果
|
|
plot_results(strategy_data, ratio_portfolio)
|
|
|
|
# 打印基于比率的统计
|
|
print_statistics(strategy_data, ratio_portfolio)
|
|
|
|
# 打印真实股票组合的统计
|
|
if stock_portfolio is not None:
|
|
print("\n" + "="*60)
|
|
print("=== 真实股票投资组合表现 ===")
|
|
print("="*60)
|
|
|
|
try:
|
|
stats = stock_portfolio.stats()
|
|
print(f"总收益率: {stats.get('Total Return [%]', 'N/A')}%")
|
|
print(f"年化收益率: {stats.get('Annual Return [%]', 'N/A')}%")
|
|
print(f"夏普比率: {stats.get('Sharpe Ratio', 'N/A')}")
|
|
print(f"最大回撤: {stats.get('Max Drawdown [%]', 'N/A')}%")
|
|
print(f"最终组合价值: {stock_portfolio.value().iloc[-1]:.2f}")
|
|
print(f"总盈亏: {stock_portfolio.value().iloc[-1] - strategy_data['initial_cash']:.2f}")
|
|
|
|
# 绘制真实股票组合的净值曲线
|
|
plt.figure(figsize=(12, 6))
|
|
plt.plot(stock_portfolio.value().index, stock_portfolio.value(),
|
|
label='真实股票组合净值', linewidth=2, color='darkgreen')
|
|
plt.axhline(y=strategy_data['initial_cash'], color='red', linestyle='--',
|
|
alpha=0.7, label=f'初始资金({strategy_data["initial_cash"]})')
|
|
plt.title('真实股票配对交易组合净值')
|
|
plt.ylabel('组合价值')
|
|
plt.xlabel('日期')
|
|
plt.legend()
|
|
plt.grid(True, alpha=0.3)
|
|
plt.tight_layout()
|
|
plt.show()
|
|
|
|
except Exception as e:
|
|
print(f"分析真实股票组合时出错: {e}")
|
|
|
|
print("程序执行完成!")
|
|
|
|
if __name__ == "__main__":
|
|
main() |