#!/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['盈亏金额'] # 计算手续费（万分之一） commission_rate = 0.0001 open_cost = position_size * entry_price * commission_rate close_cost = position_size * new_exit_price * commission_rate # 新的盈亏 gross_pnl = (new_exit_price - entry_price) * position_size new_pnl = gross_pnl - open_cost - close_cost 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()