cyb50-quant/cat-fly/t1/show_summary.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import pandas as pd
import numpy as np
from datetime import datetime
import warnings
import sys
warnings.filterwarnings('ignore')

# Redirect stdout to suppress verbose output
from io import StringIO
old_stdout = sys.stdout
sys.stdout = StringIO()

from cyb50_30min_dual_direction import ConfigManager, IntradayDataFetcher, DualDirectionSignalGenerator, DualDirectionExecutor
from t1_converter import simulate_t1_trades

def load_local_data(csv_file='cyb50_30min_2023_to_20260325.csv'):
    df = pd.read_csv(csv_file)
    df['DateTime'] = pd.to_datetime(df['DateTime'])
    df.set_index('DateTime', inplace=True)
    df.sort_index(inplace=True)
    if 'Open' not in df.columns and 'o' in df.columns:
        df.rename(columns={'o':'Open','h':'High','l':'Low','c':'Close','v':'Volume','a':'Amount'}, inplace=True)
    for col in ['Open', 'High', 'Low', 'Close', 'Volume']:
        if col in df.columns:
            df[col] = pd.to_numeric(df[col], errors='coerce')
    df['Returns'] = df['Close'].pct_change()
    df['High_Low_Pct'] = (df['High'] - df['Low']) / df['Close'].shift(1)
    df['Close_Open_Pct'] = (df['Close'] - df['Open']) / df['Open']
    df.ffill(inplace=True)
    df.dropna(inplace=True)
    return df

initial_capital = 1000000
raw_data = load_local_data('cyb50_30min_2023_to_20260325.csv')

config_manager = ConfigManager('config.json')
fetcher = IntradayDataFetcher(config_manager)
data_with_indicators = fetcher.calculate_intraday_indicators(raw_data)

signal_generator = DualDirectionSignalGenerator()
signals_df = signal_generator.generate_dual_direction_signals(data_with_indicators)

executor = DualDirectionExecutor(initial_capital=initial_capital)
results_df, trades_df = executor.execute_dual_direction_trades(signals_df)

long_trades = trades_df[trades_df['交易方向'] == '做多'].copy()
t1_trades = simulate_t1_trades(data_with_indicators, long_trades, initial_capital)
t1_adjusted = t1_trades[t1_trades['T+1调整'] == '是(T0→T1)']

final_capital = t1_trades['平仓时资金'].iloc[-1] if len(t1_trades) > 0 else initial_capital
total_return = (final_capital - initial_capital) / initial_capital * 100
total_trades = len(t1_trades)
winning_trades = t1_trades[t1_trades['盈亏金额'] > 0]
losing_trades = t1_trades[t1_trades['盈亏金额'] < 0]
win_rate = len(winning_trades) / total_trades * 100 if total_trades > 0 else 0
total_profit = winning_trades['盈亏金额'].sum() if len(winning_trades) > 0 else 0
total_loss = abs(losing_trades['盈亏金额'].sum()) if len(losing_trades) > 0 else 0
profit_factor = total_profit / total_loss if total_loss > 0 else 0
normal_trades = t1_trades[t1_trades['T+1调整'] != '是(T0→T1)']
normal_pnl = normal_trades['盈亏金额'].sum() if len(normal_trades) > 0 else 0
t1_pnl = t1_adjusted['盈亏金额'].sum() if len(t1_adjusted) > 0 else 0

# Restore stdout
sys.stdout = old_stdout

print('='*60)
print('创业板50 T+1 回测结果汇总')
print('='*60)
print(f'数据加载: {len(raw_data)}条K线')
print(f'数据区间: {raw_data.index[0]} ~ {raw_data.index[-1]}')
print(f'多空策略: 共{len(trades_df)}笔交易, 做多{len(long_trades)}笔')
print(f'T+1转换: {len(t1_trades)}笔交易')
print(f'T+1调整: {len(t1_adjusted)}笔')
print()
print('='*60)
print('总体绩效')
print('='*60)
print(f'初始资金: {initial_capital:,.0f}元')
print(f'最终资金: {final_capital:,.0f}元')
print(f'总收益率: {total_return:+.2f}%')
print(f'总交易次数: {total_trades}笔')
print(f'盈利次数: {len(winning_trades)}笔')
print(f'亏损次数: {len(losing_trades)}笔')
print(f'胜率: {win_rate:.1f}%')
print(f'盈亏比: {profit_factor:.2f}')
print()
print('='*60)
print('T+1调整统计')
print('='*60)
print(f'正常交易: {len(normal_trades)}笔, 盈亏{normal_pnl:+,.0f}元')
print(f'T+1调整(T0→T1): {len(t1_adjusted)}笔, 盈亏{t1_pnl:+,.0f}元')
print()
print('='*60)
print('最近5笔交易')
print('='*60)
print(t1_trades.tail(5)[['开仓时间', '平仓时间', '盈亏金额', 'T+1调整']].to_string(index=False))