cyb50-quant/cat-fly/t1_converter.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
CYB50 T+1 转换器 - 基于多空版本的做多交易转换为T+1规则
规则：
1. 提取多空版本中的所有做多交易
2. 买入当天不能卖出（T+1限制）
3. 如果原交易是T0（当天买卖），则延期到T+1开盘卖出
4. 重新计算延期后的盈亏（基于实际价格变化）
"""

import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import sys
sys.path.insert(0, '/root/.openclaw/workspace/cat-fly')

from cyb50_30min_dual_direction import (
    ConfigManager, IntradayDataFetcher, 
    DualDirectionSignalGenerator, DualDirectionExecutor
)


def get_next_trading_session_open(data_df, current_time):
    """获取下一个交易日的第一个开盘时间（注意：不是当天，是下一天）"""
    current_date = current_time.date()
    
    # 查找当前日期之后的所有数据
    future_data = data_df[data_df.index > current_time]
    
    if future_data.empty:
        return None, None
    
    # 获取所有日期，找到第一个不同于current_date的日期
    future_dates = future_data.index.date
    next_date = None
    
    for d in future_dates:
        if d != current_date:
            next_date = d
            break
    
    if next_date is None:
        return None, None
    
    # 获取下一个交易日的所有数据
    next_day_data = data_df[data_df.index.date == next_date]
    
    if next_day_data.empty:
        return None, None
    
    open_time = next_day_data.index[0]
    open_price = next_day_data.iloc[0]['Open']  # 使用开盘价
    
    return open_time, open_price


def simulate_t1_trades(data_df, long_trades_df, initial_capital=1000000):
    """模拟T+1规则下的交易
    
    规则：
    - 买入当天不能卖出
    - 如果原T0交易（当天买卖），延期到T+1开盘卖出
    - 卖出后当天可以再买（这是关键特性）
    """
    print("\n" + "="*80)
    print("T+1规则转换 - 基于多空版本做多交易")
    print("="*80)
    
    if len(long_trades_df) == 0:
        print("没有做多交易记录")
        return pd.DataFrame()
    
    # 按开仓时间排序
    long_trades_df = long_trades_df.sort_values('开仓时间').reset_index(drop=True)
    
    t1_trades = []
    capital = initial_capital
    
    for idx, trade in long_trades_df.iterrows():
        entry_time = trade['开仓时间']
        entry_price = trade['开仓价格']
        original_exit_time = trade['平仓时间']
        original_exit_price = trade['平仓价格']
        position_size = int(trade['仓位'])
        entry_signals = trade.get('入场信号', '')
        
        entry_date = entry_time.date()
        exit_date = original_exit_time.date()
        
        # 判断是否是T0交易
        is_t0 = (entry_date == exit_date)
        
        if is_t0:
            # T0交易需要延期到T+1开盘
            new_exit_time, new_exit_price = get_next_trading_session_open(data_df, entry_time)
            
            if new_exit_time is None:
                print(f"⚠️  交易 #{idx+1}: 无法找到T+1开盘时间，使用原平仓价格")
                new_exit_time = original_exit_time
                new_exit_price = original_exit_price
                t1_adjusted = False
            else:
                # 计算新的盈亏（无手续费）
                original_pnl = trade['盈亏金额']
                
                # 新的盈亏（毛盈亏，无手续费扣除）
                new_pnl = (new_exit_price - entry_price) * position_size
                new_pnl_pct = (new_exit_price - entry_price) / entry_price * 100
                
                # 判断新的退出原因
                stop_loss = entry_price * 0.992  # 0.8%止损
                take_profit = entry_price * 1.02  # 2%止盈
                
                if new_exit_price <= stop_loss:
                    exit_reason = f"T+1延期止损(价格{new_exit_price:.2f}触及止损线{stop_loss:.2f},亏损{abs(new_pnl_pct):.2f}%)"
                elif new_exit_price >= take_profit:
                    exit_reason = f"T+1延期止盈(价格{new_exit_price:.2f}触及止盈线{take_profit:.2f},盈利{new_pnl_pct:.2f}%)"
                else:
                    exit_reason = f"T+1延期平仓(价格{new_exit_price:.2f},盈亏{new_pnl_pct:+.2f}%)"
                
                # 计算持仓时长（小时）
                hold_hours = (new_exit_time - entry_time).total_seconds() / 3600
                
                print(f"\n[T0→T1调整] 交易 #{idx+1}")
                print(f"  原交易: {entry_time.strftime('%m-%d %H:%M')} 买 → {original_exit_time.strftime('%m-%d %H:%M')} 卖")
                print(f"  新交易: {entry_time.strftime('%m-%d %H:%M')} 买 → {new_exit_time.strftime('%m-%d %H:%M')} 卖")
                print(f"  原盈亏: {original_pnl:+.2f}元")
                print(f"  新盈亏: {new_pnl:+.2f}元 (基于T+1开盘{new_exit_price:.2f})")
                print(f"  盈亏变化: {(new_pnl - original_pnl):+.2f}元")
                
                t1_adjusted = True
                
                # 更新交易记录
                trade_record = {
                    '交易方向': '做多',
                    '开仓时间': entry_time,
                    '平仓时间': new_exit_time,
                    '开仓价格': entry_price,
                    '平仓价格': new_exit_price,
                    '仓位': position_size,
                    '盈亏金额': new_pnl,
                    '盈亏百分比': new_pnl_pct,
                    '退出原因': exit_reason,
                    '持仓周期数': int(hold_hours * 2),  # 30分钟周期数
                    '持仓小时数': hold_hours,
                    'T+1调整': '是(T0→T1)',
                    '原平仓时间': original_exit_time,
                    '原平仓价格': original_exit_price,
                    '原盈亏': original_pnl,
                    '盈亏变化': new_pnl - original_pnl,
                    '入场信号': entry_signals,
                    '开仓市值': position_size * entry_price,
                }
                
                capital += new_pnl
                trade_record['平仓时资金'] = capital
                t1_trades.append(trade_record)
                continue
        
        # 非T0交易，使用原始盈亏（已包含手续费，无需重复计算）
        hold_hours = trade['持仓小时数']
        
        # 直接使用原始盈亏（原始交易已计算好手续费）
        pnl = trade['盈亏金额']
        pnl_pct = trade['盈亏百分比']
        
        capital += pnl
        
        trade_record = {
            '交易方向': '做多',
            '开仓时间': entry_time,
            '平仓时间': original_exit_time,
            '开仓价格': entry_price,
            '平仓价格': original_exit_price,
            '仓位': position_size,
            '盈亏金额': pnl,
            '盈亏百分比': pnl_pct,
            '退出原因': trade['退出原因'],
            '持仓周期数': trade['持仓周期数'],
            '持仓小时数': hold_hours,
            'T+1调整': '否',
            '原平仓时间': original_exit_time,
            '原平仓价格': original_exit_price,
            '原盈亏': trade['盈亏金额'],
            '盈亏变化': 0,
            '入场信号': entry_signals,
            '开仓市值': position_size * entry_price,
            '平仓时资金': capital,
        }
        t1_trades.append(trade_record)
    
    t1_trades_df = pd.DataFrame(t1_trades)
    
    return t1_trades_df


def compare_results(original_trades, t1_trades, initial_capital=1000000):
    """对比原始交易和T+1转换后的结果"""
    print("\n" + "="*80)
    print("T+1转换前后对比")
    print("="*80)
    
    # 原始统计
    original_total_pnl = original_trades['盈亏金额'].sum()
    original_final = initial_capital + original_total_pnl
    original_return = (original_final / initial_capital - 1) * 100
    original_win_rate = (original_trades['盈亏金额'] > 0).sum() / len(original_trades) * 100
    
    # T+1统计
    t1_total_pnl = t1_trades['盈亏金额'].sum()
    t1_final = initial_capital + t1_total_pnl
    t1_return = (t1_final / initial_capital - 1) * 100
    t1_win_rate = (t1_trades['盈亏金额'] > 0).sum() / len(t1_trades) * 100
    
    # T0交易统计
    t0_adjusted = t1_trades[t1_trades['T+1调整'] == '是(T0→T1)']
    
    print(f"\n【原始交易（T0规则）】")
    print(f"  交易次数: {len(original_trades)}")
    print(f"  总盈亏: {original_total_pnl:+,.2f}元")
    print(f"  最终资金: {original_final:,.2f}元")
    print(f"  收益率: {original_return:+.2f}%")
    print(f"  胜率: {original_win_rate:.1f}%")
    
    print(f"\n【T+1转换后】")
    print(f"  交易次数: {len(t1_trades)}")
    print(f"  总盈亏: {t1_total_pnl:+,.2f}元")
    print(f"  最终资金: {t1_final:,.2f}元")
    print(f"  收益率: {t1_return:+.2f}%")
    print(f"  胜率: {t1_win_rate:.1f}%")
    
    print(f"\n【T+1调整统计】")
    print(f"  T0→T1调整交易数: {len(t0_adjusted)}笔")
    if len(t0_adjusted) > 0:
        print(f"  调整后盈亏变化: {t0_adjusted['盈亏变化'].sum():+,.2f}元")
        print(f"  平均每笔变化: {t0_adjusted['盈亏变化'].mean():+,.2f}元")
        print(f"  调整交易明细:")
        for _, row in t0_adjusted.iterrows():
            print(f"    {row['开仓时间'].strftime('%m-%d %H:%M')} - "
                  f"原盈亏{row['原盈亏']:+.0f} → 新盈亏{row['盈亏金额']:+.0f} "
                  f"({row['盈亏变化']:+.0f})")
    
    print(f"\n【收益差异】")
    print(f"  收益率变化: {(t1_return - original_return):+.2f}%")
    print(f"  绝对盈亏差: {(t1_total_pnl - original_total_pnl):+,.2f}元")


def main():
    """主程序"""
    print("="*80)
    print("CYB50 T+1 交易转换器")
    print("基于多空版本的做多交易，应用T+1规则")
    print("="*80)
    
    initial_capital = 1000000
    
    # 1. 运行多空版本获取原始交易数据
    print("\n【步骤1】运行多空版本获取原始交易...")
    config_manager = ConfigManager('config.json')
    fetcher = IntradayDataFetcher(config_manager)
    
    end_date = datetime.now()
    start_date = end_date - timedelta(days=70)
    
    raw_data = fetcher.fetch_30min_data(start_date, end_date)
    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()
    print(f"✅ 获取到 {len(long_trades)} 笔做多交易")
    
    # 2. 应用T+1规则
    print("\n【步骤2】应用T+1规则转换...")
    t1_trades = simulate_t1_trades(data_with_indicators, long_trades, initial_capital)
    
    # 3. 对比结果
    print("\n【步骤3】对比分析...")
    compare_results(long_trades, t1_trades, initial_capital)
    
    # 4. 导出结果
    if len(t1_trades) > 0:
        print("\n【步骤4】导出T+1交易记录...")
        
        # 格式化时间
        export_df = t1_trades.copy()
        for col in ['开仓时间', '平仓时间', '原平仓时间']:
            if col in export_df.columns:
                export_df[col] = export_df[col].dt.strftime('%Y-%m-%d %H:%M:%S')
        
        timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
        output_file = f'cyb50_t1_converted_trades_{timestamp}.csv'
        export_df.to_csv(output_file, index=False, encoding='utf-8-sig')
        print(f"✅ T+1交易记录已保存: {output_file}")
    
    print("\n" + "="*80)
    print("转换完成!")
    print("="*80)


if __name__ == "__main__":
    main()