quant/gogogo/result_visualizer.py

# 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, debug_signal_consistency, debug_signal_logic

# 设置中文显示
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()
    
    # 调试信号生成逻辑
    debug_signal_logic(strategy_data)

    # 创建基于比率的投资组合（用于分析）
    ratio_portfolio = create_portfolio(strategy_data)
    
    # 创建基于真实股票的投资组合（用于真实收益计算）
    stock_portfolio = create_real_stock_portfolio(strategy_data)
    
    # 调试信号一致性
    debug_signal_consistency(strategy_data, stock_portfolio)
    
    # 打印基于比率的统计
    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()