# result_visualizer.py import vectorbt as vbt import pandas as pd import matplotlib.pyplot as plt from strategy_executor import generate_strategy, create_portfolio, create_real_stock_portfolio # 设置中文显示 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 print_trade_orders(strategy_data, stock_portfolio): """打印所有交易订单信息""" if stock_portfolio is None: return print("\n" + "="*60) print("=== 详细交易订单信息 ===") print("="*60) try: # 获取订单记录 orders_df = stock_portfolio.orders.records_readable if len(orders_df) > 0: print(f"\n总订单数量: {len(orders_df)}") # 按时间排序 time_column = None possible_time_columns = ['Timestamp', 'Date', 'Time'] for col in possible_time_columns: if col in orders_df.columns: time_column = col break if time_column: orders_sorted = orders_df.sort_values(time_column) else: orders_sorted = orders_df # 显示所有订单（按时间排序） pd.set_option('display.max_rows', None) pd.set_option('display.width', None) print("\n【所有交易订单（按时间排序）】") print(orders_sorted.to_string(index=False)) # 按股票分组统计 print(f"\n【按股票统计】") for column in ['SMIC', 'HHIC']: column_orders = orders_sorted[orders_sorted['Column'] == column] if len(column_orders) > 0: buy_orders = column_orders[column_orders['Side'] == 'Buy'] sell_orders = column_orders[column_orders['Side'] == 'Sell'] print(f"\n{column} 订单统计:") print(f" 总订单数: {len(column_orders)}") print(f" 买入订单: {len(buy_orders)}") print(f" 卖出订单: {len(sell_orders)}") print(f" 总交易股数: {column_orders['Size'].abs().sum():.2f} 股") print(f" 总手续费: {column_orders['Fees'].sum():.2f}") # 重置pandas显示选项 pd.reset_option('display.max_rows') pd.reset_option('display.width') else: print("没有找到交易订单") except Exception as e: print(f"打印订单信息时出错: {e}") import traceback traceback.print_exc() def print_detailed_trade_analysis(strategy_data, stock_portfolio): """打印详细的交易分析""" if stock_portfolio is None: return print("\n" + "="*60) print("=== 详细交易分析 ===") print("="*60) try: # 获取交易记录 trades_df = stock_portfolio.trades.records_readable if len(trades_df) > 0: print(f"\n总交易次数: {len(trades_df)}") # 按开仓时间排序 trades_sorted = trades_df.sort_values('Entry Timestamp') # 显示所有交易（按时间排序） pd.set_option('display.max_rows', None) pd.set_option('display.width', None) print("\n【所有交易记录（按开仓时间排序）】") print(trades_sorted.to_string(index=False)) # 按股票分组分析 print(f"\n【按股票交易分析】") for column in ['SMIC', 'HHIC']: column_trades = trades_sorted[trades_sorted['Column'] == column] if len(column_trades) > 0: winning_trades = column_trades[column_trades['PnL'] > 0] losing_trades = column_trades[column_trades['PnL'] < 0] print(f"\n{column} 交易分析:") print(f" 总交易次数: {len(column_trades)}") print(f" 盈利交易: {len(winning_trades)}") print(f" 亏损交易: {len(losing_trades)}") print(f" 胜率: {len(winning_trades)/len(column_trades)*100:.1f}%") print(f" 总盈亏: {column_trades['PnL'].sum():.2f}") print(f" 平均每笔盈亏: {column_trades['PnL'].mean():.2f}") print(f" 最大盈利: {column_trades['PnL'].max():.2f}") print(f" 最大亏损: {column_trades['PnL'].min():.2f}") if len(winning_trades) > 0: print(f" 平均盈利: {winning_trades['PnL'].mean():.2f}") if len(losing_trades) > 0: print(f" 平均亏损: {losing_trades['PnL'].mean():.2f}") # 正确的配对交易分析 print(f"\n【配对交易分析（修正）】") # 按开仓时间分组，找出同一天开仓的SMIC和HHIC交易 entry_groups = trades_sorted.groupby('Entry Timestamp').agg({ 'Column': list, 'PnL': list, 'Size': list, 'Direction': list, 'Exit Timestamp': list }) pair_trades = [] for entry_date, group_data in entry_groups.iterrows(): columns = group_data['Column'] pnls = group_data['PnL'] directions = group_data['Direction'] exit_times = group_data['Exit Timestamp'] # 检查是否同时有SMIC和HHIC的交易 if 'SMIC' in columns and 'HHIC' in columns: smic_idx = columns.index('SMIC') hhic_idx = columns.index('HHIC') # 检查交易方向是否配对（一个做多，一个做空） if directions[smic_idx] != directions[hhic_idx]: pair_pnl = pnls[smic_idx] + pnls[hhic_idx] pair_trades.append({ 'entry_date': entry_date, 'exit_date_smic': exit_times[smic_idx], 'exit_date_hhic': exit_times[hhic_idx], 'smic_pnl': pnls[smic_idx], 'hhic_pnl': pnls[hhic_idx], 'total_pnl': pair_pnl, 'smic_direction': directions[smic_idx], 'hhic_direction': directions[hhic_idx] }) if pair_trades: # 按开仓时间排序配对交易 pair_trades_sorted = sorted(pair_trades, key=lambda x: x['entry_date']) print(f"配对交易次数: {len(pair_trades_sorted)}") total_pair_pnl = sum(trade['total_pnl'] for trade in pair_trades_sorted) avg_pair_pnl = total_pair_pnl / len(pair_trades_sorted) winning_pairs = [t for t in pair_trades_sorted if t['total_pnl'] > 0] losing_pairs = [t for t in pair_trades_sorted if t['total_pnl'] < 0] print(f"配对交易总盈亏: {total_pair_pnl:.2f}") print(f"平均每对盈亏: {avg_pair_pnl:.2f}") print(f"盈利配对: {len(winning_pairs)}") print(f"亏损配对: {len(losing_pairs)}") print(f"配对胜率: {len(winning_pairs)/len(pair_trades_sorted)*100:.1f}%") # 显示每对交易的详细信息（按时间排序） print(f"\n【各配对交易详情（按开仓时间排序）】") for i, trade in enumerate(pair_trades_sorted, 1): status = "盈利" if trade['total_pnl'] > 0 else "亏损" print(f"第{i}对 - 开仓: {trade['entry_date']}") print(f" SMIC({trade['smic_direction']}): {trade['smic_pnl']:.2f} (平仓: {trade['exit_date_smic']})") print(f" HHIC({trade['hhic_direction']}): {trade['hhic_pnl']:.2f} (平仓: {trade['exit_date_hhic']})") print(f" 总盈亏: {trade['total_pnl']:.2f} [{status}]") print() else: print("没有找到配对交易") # 重置pandas显示选项 pd.reset_option('display.max_rows') pd.reset_option('display.width') else: print("没有找到交易记录") except Exception as e: print(f"打印交易分析时出错: {e}") import traceback traceback.print_exc() def main(): """主函数""" # 生成策略 strategy_data = generate_strategy() # 创建基于比率的投资组合（用于分析） ratio_portfolio = create_portfolio(strategy_data) # 创建基于真实股票的投资组合（用于真实收益计算） stock_portfolio = create_real_stock_portfolio(strategy_data) # 打印基于比率的统计 print_statistics(strategy_data, ratio_portfolio) # 打印真实股票组合的统计 print(stock_portfolio['SMIC'].stats()) print(stock_portfolio['HHIC'].stats()) # 绘制基于比率的结果 plot_results(strategy_data, ratio_portfolio) # 打印详细交易订单信息 print_trade_orders(strategy_data, stock_portfolio) # 打印详细交易分析 print_detailed_trade_analysis(strategy_data, stock_portfolio) print("程序执行完成!") if __name__ == "__main__": main()