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

#!/usr/bin/env python3
"""
基于 cyb50_30min_dual_direction.py 的只做多T+1回测
使用与 auto_report_long_only_t1.py 完全相同的策略逻辑
"""

import csv
import json
from datetime import datetime, timedelta
from collections import deque
import os

class DualDirectionLongOnlyBacktest:
    """
    只做多回测 - 基于 cyb50_30min_dual_direction 策略
    """
    
    def __init__(self, initial_capital=1000000):
        self.initial_capital = initial_capital
        self.position_size_pct = 1.0  # 满仓
        self.stop_loss_pct = 0.008    # 0.8%止损
        self.take_profit_pct = 0.02   # 2%止盈
        self.max_hold_bars = 16       # 最大8小时
        
        # 统计数据
        self.long_signal_count = 0
        self.trades = []
        self.capital = initial_capital
        
    def calculate_indicators(self, data):
        """计算技术指标"""
        print("计算技术指标...")
        
        # 为每行数据添加指标
        for i, row in enumerate(data):
            if i < 24:  # 需要至少24个周期的历史数据
                row['RSI'] = 50
                row['MACD'] = 0
                row['MACD_hist'] = 0
                row['K'] = 50
                row['D'] = 50
                row['J'] = 50
                row['BB_middle'] = row['Close']
                row['BB_upper'] = row['Close'] * 1.02
                row['BB_lower'] = row['Close'] * 0.98
                row['Volume_Ratio'] = 1.0
                row['Price_Momentum'] = 0
                row['Close_Open_Pct'] = 0
                row['MA6'] = row['Close']
                row['MA12'] = row['Close']
                continue
            
            closes = [data[j]['Close'] for j in range(i-23, i+1)]
            highs = [data[j]['High'] for j in range(i-23, i+1)]
            lows = [data[j]['Low'] for j in range(i-23, i+1)]
            volumes = [data[j]['Volume'] for j in range(i-23, i+1)]
            
            # MA
            row['MA6'] = sum(closes[-6:]) / 6
            row['MA12'] = sum(closes[-12:]) / 12
            
            # RSI
            gains = []
            losses = []
            for j in range(1, 15):
                change = closes[-j] - closes[-j-1]
                gains.append(max(0, change))
                losses.append(max(0, -change))
            avg_gain = sum(gains) / 14
            avg_loss = sum(losses) / 14
            if avg_loss == 0:
                row['RSI'] = 100
            else:
                rs = avg_gain / avg_loss
                row['RSI'] = 100 - (100 / (1 + rs))
            
            # 布林带
            bb_middle = sum(closes[-20:]) / 20
            variance = sum((c - bb_middle) ** 2 for c in closes[-20:]) / 20
            bb_std = variance ** 0.5
            row['BB_middle'] = bb_middle
            row['BB_upper'] = bb_middle + bb_std * 2
            row['BB_lower'] = bb_middle - bb_std * 2
            
            # 简化MACD
            ema12 = sum(closes[-12:]) / 12
            ema26 = sum(closes[-26:]) / 26 if len(closes) >= 26 else sum(closes) / len(closes)
            row['MACD'] = ema12 - ema26
            row['MACD_hist'] = row['MACD']  # 简化
            
            # KDJ (简化)
            low_9 = min(lows[-9:])
            high_9 = max(highs[-9:])
            if high_9 == low_9:
                rsv = 50
            else:
                rsv = (row['Close'] - low_9) / (high_9 - low_9) * 100
            row['K'] = rsv
            row['D'] = rsv
            row['J'] = 3 * rsv - 2 * rsv
            
            # 成交量比率
            vol_ma = sum(volumes[-12:]) / 12
            row['Volume_Ratio'] = row['Volume'] / vol_ma if vol_ma > 0 else 1
            
            # 价格动量
            row['Price_Momentum'] = (row['Close'] - closes[-6]) / closes[-6] if closes[-6] > 0 else 0
            row['Close_Open_Pct'] = (row['Close'] - row['Open']) / row['Open'] if row['Open'] > 0 else 0
        
        print(f"  指标计算完成，共{len(data)}条")
        return data
    
    def calculate_long_score(self, row, prev_rows):
        """计算做多信号强度 - 完全按照 DualDirection 逻辑"""
        long_score = 0
        long_signals = []
        
        # 1. RSI超卖做多
        if row['RSI'] < 30:
            long_score += 2
            long_signals.append("RSI超卖")
        elif row['RSI'] < 35:
            long_score += 1
            long_signals.append("RSI偏弱")
        
        # 2. 价格触及布林带下轨
        if row['Close'] <= row['BB_lower'] * 1.01:
            long_score += 2
            long_signals.append("触及下轨")
        elif row['Close'] <= row['BB_lower'] * 1.03:
            long_score += 1
            long_signals.append("接近下轨")
        
        # 3. MACD金叉或柱状图转正
        if len(prev_rows) > 0:
            prev_macd_hist = prev_rows[-1]['MACD_hist']
            if row['MACD_hist'] > 0 and prev_macd_hist <= 0:
                long_score += 2
                long_signals.append("MACD金叉")
            elif row['MACD_hist'] > prev_macd_hist:
                long_score += 1
                long_signals.append("MACD改善")
        
        # 4. 价格动量向上
        if row['Price_Momentum'] > 0.005:
            long_score += 1
            long_signals.append("动量向上")
        
        # 5. 成交量放大
        if row['Volume_Ratio'] > 1.5:
            long_score += 1
            long_signals.append("放量")
        
        return long_score, long_signals
    
    def run_backtest(self, data_file):
        """运行回测"""
        print("="*70)
        print("DualDirection 策略 - 只做多T+1回测")
        print("="*70)
        print(f"\n参数设置:")
        print(f"  初始资金: {self.initial_capital:,.0f}元")
        print(f"  仓位比例: {self.position_size_pct*100:.0f}%")
        print(f"  止损: {self.stop_loss_pct*100:.1f}%")
        print(f"  止盈: {self.take_profit_pct*100:.1f}%")
        print(f"  最大持仓: {self.max_hold_bars}周期(8小时)")
        
        # 1. 加载数据
        print(f"\n[1/3] 加载数据: {data_file}")
        data = []
        with open(data_file, 'r', encoding='utf-8-sig') as f:
            reader = csv.DictReader(f)
            for row in reader:
                data.append({
                    'DateTime': row['DateTime'],
                    'Open': float(row['Open']),
                    'High': float(row['High']),
                    'Low': float(row['Low']),
                    'Close': float(row['Close']),
                    'Volume': float(row['Volume'])
                })
        print(f"      加载完成: {len(data)}条")
        print(f"      时间范围: {data[0]['DateTime']} ~ {data[-1]['DateTime']}")
        
        # 2. 计算指标
        print("\n[2/3] 计算技术指标...")
        data = self.calculate_indicators(data)
        
        # 3. 生成信号并执行回测
        print("\n[3/3] 生成信号并执行回测...")
        
        position = 0
        entry_price = 0
        entry_time = None
        entry_idx = 0
        
        for i in range(24, len(data)):
            row = data[i]
            current_time = row['DateTime']
            current_price = row['Close']
            
            # 计算做多信号
            prev_rows = data[max(0, i-5):i]
            long_score, long_signals = self.calculate_long_score(row, prev_rows)
            
            # 持仓管理
            if position > 0:
                holding_bars = i - entry_idx
                pnl_pct = (current_price - entry_price) / entry_price
                
                exit_reason = None
                
                # 止损 -0.8%
                if pnl_pct <= -self.stop_loss_pct:
                    exit_reason = f"止损({current_price:.2f})"
                # 止盈 +2%
                elif pnl_pct >= self.take_profit_pct:
                    exit_reason = f"止盈({current_price:.2f})"
                # 最大持仓时间
                elif holding_bars >= self.max_hold_bars:
                    exit_reason = f"时间平仓({holding_bars}周期)"
                # RSI超买
                elif row['RSI'] > 75:
                    exit_reason = f"RSI超买({row['RSI']:.1f})"
                
                if exit_reason:
                    pnl = (current_price - entry_price) * position
                    self.capital += pnl
                    
                    self.trades.append({
                        'action': 'CLOSE',
                        'time': current_time,
                        'price': current_price,
                        'shares': position,
                        'pnl': pnl,
                        'pnl_pct': pnl_pct * 100,
                        'reason': exit_reason
                    })
                    
                    position = 0
                    entry_price = 0
                    entry_time = None
            
            # 开仓判断 - 信号强度>=4
            elif long_score >= 4 and position == 0:
                position_value = self.capital * self.position_size_pct
                position = position_value / current_price
                entry_price = current_price
                entry_time = current_time
                entry_idx = i
                self.long_signal_count += 1
                
                self.trades.append({
                    'action': 'OPEN',
                    'time': current_time,
                    'price': current_price,
                    'shares': position,
                    'value': position_value,
                    'reason': f"做多信号(强度{long_score}): {'+'.join(long_signals[:3])}"
                })
        
        print(f"      做多信号: {self.long_signal_count}个")
        print(f"      实际交易: {len([t for t in self.trades if t['action']=='CLOSE'])}笔")
        
        return self.generate_report()
    
    def generate_report(self):
        """生成报告"""
        closed_trades = [t for t in self.trades if t['action'] == 'CLOSE']
        
        if not closed_trades:
            print("\n无交易记录")
            return None
        
        # 计算统计
        wins = [t for t in closed_trades if t['pnl'] > 0]
        losses = [t for t in closed_trades if t['pnl'] <= 0]
        
        total_pnl = sum(t['pnl'] for t in closed_trades)
        final_capital = self.initial_capital + total_pnl
        total_return = (final_capital / self.initial_capital - 1) * 100
        
        win_rate = len(wins) / len(closed_trades) * 100
        
        total_profit = sum(t['pnl'] for t in wins) if wins else 0
        total_loss = abs(sum(t['pnl'] for t in losses)) if losses else 0
        profit_factor = total_profit / total_loss if total_loss > 0 else 0
        
        # 计算最大回撤
        peak = self.initial_capital
        max_dd = 0
        current = self.initial_capital
        
        for t in closed_trades:
            current += t['pnl']
            if current > peak:
                peak = current
            dd = (peak - current) / peak * 100
            if dd > max_dd:
                max_dd = dd
        
        print("\n" + "="*70)
        print("回测报告 - DualDirection 只做多T+1")
        print("="*70)
        print(f"\n【整体表现】")
        print(f"  初始资金: {self.initial_capital:,.2f}元")
        print(f"  最终资金: {final_capital:,.2f}元")
        print(f"  净盈亏: {total_pnl:+,.2f}元")
        print(f"  总收益率: {total_return:+.2f}%")
        print(f"  最大回撤: {max_dd:.2f}%")
        
        print(f"\n【交易统计】")
        print(f"  总交易: {len(closed_trades)}笔")
        print(f"  盈利: {len(wins)}笔")
        print(f"  亏损: {len(losses)}笔")
        print(f"  胜率: {win_rate:.2f}%")
        print(f"  盈亏比: {profit_factor:.2f}")
        print(f"  总盈利: {total_profit:,.2f}元")
        print(f"  总亏损: {total_loss:,.2f}元")
        
        print(f"\n【最近10笔交易】")
        for t in closed_trades[-10:]:
            print(f"  {t['time']} | {t['pnl']:+10,.2f}元 ({t['pnl_pct']:+.2f}%) | {t['reason']}")
        
        print("="*70)
        
        return {
            'total_return': total_return,
            'win_rate': win_rate,
            'profit_factor': profit_factor,
            'trade_count': len(closed_trades),
            'max_drawdown': max_dd
        }

if __name__ == '__main__':
    backtest = DualDirectionLongOnlyBacktest(initial_capital=1000000)
    result = backtest.run_backtest('cyb50_30min_2023_to_20260325.csv')