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

#!/usr/bin/env python3
"""
DualDirection 策略 + 择时过滤
保持原策略所有参数不变，只增加日线趋势和30分钟状态过滤
"""

import csv
from datetime import datetime
import math

class DualDirectionWithTiming:
    def __init__(self, initial_capital=1000000):
        self.initial_capital = initial_capital
        self.position_size_pct = 1.0
        self.stop_loss_pct = 0.008
        self.take_profit_pct = 0.02
        self.max_hold_bars = 16
        self.min_trend_prob = 0.3
        self.require_daily_uptrend = True
        
        self.long_signal_count = 0
        self.filtered_count = 0
        self.trades = []
        self.capital = initial_capital
    
    def load_daily_data(self, filepath):
        daily_data = {}
        with open(filepath, 'r', encoding='utf-8-sig') as f:
            reader = csv.DictReader(f)
            for row in reader:
                try:
                    dt = datetime.strptime(row['datetime'], '%Y-%m-%d %H:%M:%S')
                    date_str = dt.strftime('%Y-%m-%d')
                    daily_data[date_str] = {
                        'open': float(row['open']),
                        'high': float(row['high']),
                        'low': float(row['low']),
                        'close': float(row['close'])
                    }
                except:
                    continue
        
        dates = sorted(daily_data.keys())
        closes = [daily_data[d]['close'] for d in dates]
        
        for i, date in enumerate(dates):
            if i < 19:
                daily_data[date]['ma20'] = None
                daily_data[date]['trend'] = 0
            else:
                ma20 = sum(closes[i-19:i+1]) / 20
                daily_data[date]['ma20'] = ma20
                close = closes[i]
                if close > ma20 * 1.02:
                    daily_data[date]['trend'] = 1
                elif close < ma20 * 0.98:
                    daily_data[date]['trend'] = -1
                else:
                    daily_data[date]['trend'] = 0
        
        return daily_data
    
    def detect_market_regime(self, data, current_idx):
        if current_idx < 16:
            return 0, 0.0
        
        window = data[current_idx-16:current_idx]
        closes = [row['Close'] for row in window]
        highs = [row['High'] for row in window]
        lows = [row['Low'] for row in window]
        
        start_price = closes[0]
        end_price = closes[-1]
        period_return = (end_price / start_price - 1) * 100
        
        max_price = max(highs)
        min_price = min(lows)
        price_range = (max_price - min_price) / start_price * 100
        
        gains = []
        losses = []
        for i in range(1, len(closes)):
            change = closes[i] - closes[i-1]
            gains.append(max(0, change))
            losses.append(max(0, -change))
        
        avg_gain = sum(gains[-14:]) / 14 if len(gains) >= 14 else sum(gains) / len(gains)
        avg_loss = sum(losses[-14:]) / 14 if len(losses) >= 14 else sum(losses) / len(losses)
        
        if avg_loss == 0:
            rsi = 100
        else:
            rs = avg_gain / avg_loss
            rsi = 100 - (100 / (1 + rs))
        
        returns = [(closes[i] - closes[i-1]) / closes[i-1] * 100 for i in range(1, len(closes))]
        volatility = math.sqrt(sum(r**2 for r in returns) / len(returns)) if returns else 0
        
        reversal_score = 0
        if rsi > 70 or rsi < 30:
            reversal_score += 2
        elif rsi > 65 or rsi < 35:
            reversal_score += 1
        
        if price_range > 4 and abs(period_return) < 1.5:
            reversal_score += 1
        
        if reversal_score >= 3:
            return 2, 0.3
        
        trend_score = 0
        if abs(period_return) >= 2.0:
            trend_score += 3
        elif abs(period_return) >= 1.0:
            trend_score += 2
        elif abs(period_return) >= 0.5:
            trend_score += 1
        
        if 0.5 < volatility < 2.0:
            trend_score += 1
        
        first_half = closes[:len(closes)//2]
        second_half = closes[len(closes)//2:]
        first_avg = sum(first_half) / len(first_half)
        second_avg = sum(second_half) / len(second_half)
        
        if (period_return > 0 and second_avg > first_avg) or (period_return < 0 and second_avg < first_avg):
            trend_score += 1
        
        if trend_score >= 4:
            prob = min(0.95, 0.5 + abs(period_return) / 10)
            return 1, prob
        
        return 0, 0.2
    
    def calculate_indicators(self, data):
        for i, row in enumerate(data):
            if i < 24:
                row['RSI'] = 50
                row['MACD_hist'] = 0
                row['BB_lower'] = row['Close'] * 0.98
                row['Volume_Ratio'] = 1.0
                row['Price_Momentum'] = 0
                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)]
            
            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_lower'] = bb_middle - bb_std * 2
            
            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']
            
            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
        
        return data
    
    def calculate_long_score(self, row, prev_rows):
        long_score = 0
        long_signals = []
        
        if row['RSI'] < 30:
            long_score += 2
            long_signals.append("RSI超卖")
        elif row['RSI'] < 35:
            long_score += 1
            long_signals.append("RSI偏弱")
        
        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("接近下轨")
        
        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改善")
        
        if row['Price_Momentum'] > 0.005:
            long_score += 1
            long_signals.append("动量向上")
        
        if row['Volume_Ratio'] > 1.5:
            long_score += 1
            long_signals.append("放量")
        
        return long_score, long_signals
    
    def run_backtest(self, data_file, daily_file):
        print("="*70)
        print("DualDirection + 择时过滤 回测")
        print("="*70)
        print(f"\nDualDirection参数: 止损0.8% 止盈2% 最大持仓16周期")
        print(f"择时过滤: 日线向上 + 30分钟趋势概率>=0.3")
        
        print(f"\n[1/4] 加载30分钟数据...")
        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"\n[2/4] 加载日线数据...")
        daily_data = self.load_daily_data(daily_file)
        print(f"      {len(daily_data)}条")
        
        print(f"\n[3/4] 计算技术指标...")
        data = self.calculate_indicators(data)
        
        print(f"\n[4/4] 执行回测...")
        position = 0
        entry_price = 0
        entry_idx = 0
        
        for i in range(24, len(data)):
            row = data[i]
            current_time = row['DateTime']
            current_price = row['Close']
            date_str = current_time[:10]
            
            daily_info = daily_data.get(date_str, {'trend': 0})
            daily_trend = daily_info['trend']
            
            regime_state, trend_prob = self.detect_market_regime(data, i)
            
            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
                if pnl_pct <= -self.stop_loss_pct:
                    exit_reason = f"止损({current_price:.2f})"
                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}周期)"
                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,
                        'pnl': pnl, 'pnl_pct': pnl_pct * 100, 'reason': exit_reason
                    })
                    position = 0
                    entry_price = 0
            
            elif long_score >= 4 and position == 0:
                self.long_signal_count += 1
                
                can_trade = True
                if self.require_daily_uptrend and daily_trend != 1:
                    can_trade = False
                if regime_state != 1 or trend_prob < self.min_trend_prob:
                    can_trade = False
                
                if can_trade:
                    position_value = self.capital * self.position_size_pct
                    position = position_value / current_price
                    entry_price = current_price
                    entry_idx = i
                    self.trades.append({
                        'action': 'OPEN', 'time': current_time, 'price': current_price,
                        'value': position_value,
                        'reason': f"信号{long_score}分|日线向上|趋势{trend_prob:.2f}"
                    })
                else:
                    self.filtered_count += 1
        
        closed = [t for t in self.trades if t['action']=='CLOSE']
        print(f"      信号: {self.long_signal_count} 过滤: {self.filtered_count} 交易: {len(closed)}")
        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
        
        print("\n" + "="*70)
        print("回测报告 - DualDirection + 择时过滤")
        print("="*70)
        print(f"  收益率: {total_return:+.2f}%")
        print(f"  信号: {self.long_signal_count} 过滤: {self.filtered_count} 交易: {len(closed_trades)}")
        print(f"  胜率: {win_rate:.2f}% 盈亏比: {profit_factor:.2f}")
        print(f"\n最近5笔:")
        for t in closed_trades[-5:]:
            print(f"  {t['time']} | {t['pnl']:+10,.2f} | {t['reason']}")
        print("="*70)
        
        return {'total_return': total_return, 'win_rate': win_rate}

if __name__ == '__main__':
    backtest = DualDirectionWithTiming()
    backtest.run_backtest('cyb50_30min_2023_to_20260325.csv', '../data-fetch/data/399673_SZ_day_20150101_20260325.csv')