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

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

import sys
sys.path.insert(0, '/home/erwin/.openclaw/workspace/cyb50-quant/cat-fly/t1')

import pandas as pd
import numpy as np
from datetime import datetime, timedelta

# 导入策略模块
from cyb50_30min_dual_direction import (
    DualDirectionSignalGenerator
)

def calculate_indicators(data):
    """计算技术指标"""
    df = data.copy()
    
    # 移动平均线
    df['MA6'] = df['Close'].rolling(window=6).mean()
    df['MA12'] = df['Close'].rolling(window=12).mean()
    df['MA24'] = df['Close'].rolling(window=24).mean()
    
    # RSI
    delta = df['Close'].diff()
    gain = (delta.where(delta > 0, 0)).rolling(window=14).mean()
    loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean()
    rs = gain / loss
    df['RSI'] = 100 - (100 / (1 + rs))
    
    # 布林带
    df['BB_middle'] = df['Close'].rolling(window=20).mean()
    bb_std = df['Close'].rolling(window=20).std()
    df['BB_upper'] = df['BB_middle'] + (bb_std * 2)
    df['BB_lower'] = df['BB_middle'] - (bb_std * 2)
    
    # MACD
    exp1 = df['Close'].ewm(span=12, adjust=False).mean()
    exp2 = df['Close'].ewm(span=26, adjust=False).mean()
    df['MACD'] = exp1 - exp2
    df['MACD_signal'] = df['MACD'].ewm(span=9, adjust=False).mean()
    df['MACD_hist'] = df['MACD'] - df['MACD_signal']
    
    # KDJ
    low_9 = df['Low'].rolling(window=9).min()
    high_9 = df['High'].rolling(window=9).max()
    rsv = (df['Close'] - low_9) / (high_9 - low_9) * 100
    df['K'] = rsv.ewm(com=2, adjust=False).mean()
    df['D'] = df['K'].ewm(com=2, adjust=False).mean()
    df['J'] = 3 * df['K'] - 2 * df['D']
    
    df['Volume_MA'] = df['Volume'].rolling(window=12).mean()
    df['Volume_Ratio'] = df['Volume'] / df['Volume_MA']
    df['Price_Momentum'] = (df['Close'] - df['Close'].shift(6)) / df['Close'].shift(6)
    df['Close_Open_Pct'] = (df['Close'] - df['Open']) / df['Open']
    
    return df

def run_backtest():
    """运行回测"""
    print("="*70)
    print("基于 DualDirection 策略的只做多回测")
    print("="*70)
    
    # 加载数据
    data_file = 'cyb50_30min_2023_to_20260325.csv'
    print(f"\n[1/3] 加载数据: {data_file}")
    data = pd.read_csv(data_file)
    data['DateTime'] = pd.to_datetime(data['DateTime'])
    data.set_index('DateTime', inplace=True)
    data.sort_index(inplace=True)
    print(f"      数据条数: {len(data)}")
    print(f"      范围: {data.index[0]} ~ {data.index[-1]}")
    
    # 计算技术指标
    print("\n[2/3] 计算技术指标...")
    df = calculate_indicators(data)
    print("      完成")
    
    # 生成信号
    print("\n[3/3] 生成多空信号...")
    signal_generator = DualDirectionSignalGenerator()
    signals_df = signal_generator.generate_dual_direction_signals(df)
    
    print(f"\n信号统计:")
    print(f"  做多信号: {signal_generator.long_signal_count}个")
    print(f"  做空信号: {signal_generator.short_signal_count}个")
    print(f"  总信号: {signal_generator.total_signal_count}个")
    
    # 提取做多信号
    long_signals = signals_df[signals_df['Signal'] == 1]
    print(f"\n做多信号详情:")
    print(long_signals[['Close', 'Long_Score', 'Long_Signals']].head(10))

if __name__ == '__main__':
    run_backtest()