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

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
创业板50 T+1 高级优化策略 - 简化版
聚焦: 移动止损、分批止盈效果验证
"""
import pandas as pd
import numpy as np
from datetime import datetime
import warnings
warnings.filterwarnings('ignore')

from cyb50_30min_dual_direction import ConfigManager, IntradayDataFetcher, DualDirectionSignalGenerator, DualDirectionExecutor
from t1_converter import simulate_t1_trades

def load_local_data(csv_file='cyb50_30min_2023_to_20260325.csv'):
    df = pd.read_csv(csv_file)
    df['DateTime'] = pd.to_datetime(df['DateTime'])
    df.set_index('DateTime', inplace=True)
    df.sort_index(inplace=True)
    if 'Open' not in df.columns and 'o' in df.columns:
        df.rename(columns={'o':'Open','h':'High','l':'Low','c':'Close','v':'Volume','a':'Amount'}, inplace=True)
    for col in ['Open', 'High', 'Low', 'Close', 'Volume']:
        if col in df.columns:
            df[col] = pd.to_numeric(df[col], errors='coerce')
    df['Returns'] = df['Close'].pct_change()
    df.ffill(inplace=True)
    df.dropna(inplace=True)
    return df

def calculate_indicators(df):
    """计算指标"""
    # 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_14'] = 100 - (100 / (1 + rs))

    # 动量
    df['Momentum_5'] = (df['Close'] / df['Close'].shift(5) - 1) * 100

    # 均线
    df['EMA_5'] = df['Close'].ewm(span=5, adjust=False).mean()
    df['EMA_20'] = df['Close'].ewm(span=20, adjust=False).mean()
    df['Trend_Score'] = 0
    df.loc[df['Close'] > df['EMA_5'], 'Trend_Score'] += 1
    df.loc[df['Close'] > df['EMA_20'], 'Trend_Score'] += 1

    # 添加Close_Open_Pct（原策略需要）
    df['Close_Open_Pct'] = (df['Close'] - df['Open']) / df['Open']

    df.dropna(inplace=True)
    return df

# 加载数据
print("="*60)
print("创业板50 T+1 高级优化策略 - 简化版")
print("="*60)

raw_data = load_local_data('cyb50_30min_2023_to_20260325.csv')
data = calculate_indicators(raw_data)

print(f"\n📊 数据: {len(data)}条K线")

# 获取原策略交易
config_manager = ConfigManager('config.json')
fetcher = IntradayDataFetcher(config_manager)
signal_generator = DualDirectionSignalGenerator()
executor = DualDirectionExecutor(initial_capital=1000000)

data_with_indicators = fetcher.calculate_intraday_indicators(data)
signals_df = signal_generator.generate_dual_direction_signals(data_with_indicators)
results_df, trades_df = executor.execute_dual_direction_trades(signals_df)

long_trades = trades_df[trades_df['交易方向'] == '做多'].copy()
t1_trades = simulate_t1_trades(data_with_indicators, long_trades, 1000000)

print(f"\n📈 原策略交易: {len(t1_trades)}笔")

# 准备数据
t1_trades['开仓时间'] = pd.to_datetime(t1_trades['开仓时间'])
t1_trades['平仓时间'] = pd.to_datetime(t1_trades['平仓时间'])
t1_trades['是否盈利'] = t1_trades['盈亏金额'] > 0

# 计算期间高低点（用于移动止损）
for idx, row in t1_trades.iterrows():
    mask = (data.index >= row['开仓时间']) & (data.index <= row['平仓时间'])
    if mask.any():
        period_data = data.loc[mask]
        t1_trades.loc[idx, '期间最高'] = period_data['High'].max()
        t1_trades.loc[idx, '期间最低'] = period_data['Low'].min()
    else:
        t1_trades.loc[idx, '期间最高'] = row['平仓价格']
        t1_trades.loc[idx, '期间最低'] = row['平仓价格']

# 策略对比
print("\n" + "="*60)
print("📊 策略对比分析")
print("="*60)

strategies = []

# 策略1: 原策略
original_pnl = t1_trades['盈亏金额'].sum()
original_winrate = (t1_trades['盈亏金额'] > 0).mean() * 100
strategies.append({
    '策略': '原策略(止损0.8%止盈2%)',
    '交易数': len(t1_trades),
    '胜率': f"{original_winrate:.1f}%",
    '总盈亏': f"{original_pnl:+.0f}",
    '改善': '-'
})

# 策略2: 固定止损0.8%止盈2.5%
def simulate_fixed_sl_tp(trades, sl, tp, position=0.7):
    total = 0
    for _, row in trades.iterrows():
        entry = row['开仓价格']
        exit_p = row['平仓价格']
        pnl_pct = (exit_p - entry) / entry

        if pnl_pct <= -sl:
            actual = -sl
        elif pnl_pct >= tp:
            actual = tp
        else:
            actual = pnl_pct

        total += actual * 1000000 * position
    return total

pnl_v1 = simulate_fixed_sl_tp(t1_trades, 0.008, 0.025)
strategies.append({
    '策略': '优化V1(固定SL0.8% TP2.5%)',
    '交易数': len(t1_trades),
    '胜率': f"{original_winrate:.1f}%",
    '总盈亏': f"{pnl_v1:+.0f}",
    '改善': f"{pnl_v1 - original_pnl:+.0f}"
})

# 策略3: 移动止损
def simulate_trailing_stop(trades, sl, tp, trail_activate, trail_dist, position=0.7):
    total = 0
    wins = 0
    losses = 0

    for _, row in trades.iterrows():
        entry = row['开仓价格']
        exit_p = row['平仓价格']
        high = row['期间最高']

        pnl_pct = (exit_p - entry) / entry
        max_profit_pct = (high - entry) / entry

        # 硬止损
        if pnl_pct <= -sl:
            actual = -sl
            losses += 1
        # 移动止损激活
        elif max_profit_pct >= trail_activate:
            trail_stop_price = high * (1 - trail_dist)
            trail_stop_pct = (trail_stop_price - entry) / entry

            if pnl_pct <= trail_stop_pct:
                actual = trail_stop_pct
            else:
                actual = pnl_pct

            if actual > 0:
                wins += 1
            else:
                losses += 1
        # 目标止盈
        elif pnl_pct >= tp:
            actual = tp
            wins += 1
        else:
            actual = pnl_pct
            if actual > 0:
                wins += 1
            else:
                losses += 1

        total += actual * 1000000 * position

    win_rate = wins / (wins + losses) * 100 if (wins + losses) > 0 else 0
    return total, win_rate

pnl_v2, wr_v2 = simulate_trailing_stop(t1_trades, 0.008, 0.025, 0.01, 0.005)
strategies.append({
    '策略': '优化V2(移动止损)',
    '交易数': len(t1_trades),
    '胜率': f"{wr_v2:.1f}%",
    '总盈亏': f"{pnl_v2:+.0f}",
    '改善': f"{pnl_v2 - original_pnl:+.0f}"
})

# 策略4: 分批止盈
def simulate_partial_exit(trades, sl, tp1, tp2, ratio, position=0.7):
    total = 0
    for _, row in trades.iterrows():
        entry = row['开仓价格']
        exit_p = row['平仓价格']
        high = row['期间最高']

        max_profit_pct = (high - entry) / entry

        # 第一批止盈
        if max_profit_pct >= tp1:
            # 假设50%仓位在tp1止盈，剩余按实际或tp2
            if max_profit_pct >= tp2:
                actual = tp1 * ratio + tp2 * (1 - ratio)
            else:
                actual_pnl = (exit_p - entry) / entry
                actual = tp1 * ratio + actual_pnl * (1 - ratio)
        elif (exit_p - entry) / entry <= -sl:
            actual = -sl
        else:
            actual = (exit_p - entry) / entry

        total += actual * 1000000 * position
    return total

pnl_v3 = simulate_partial_exit(t1_trades, 0.008, 0.015, 0.03, 0.5)
strategies.append({
    '策略': '优化V3(分批止盈 1.5%+3%)',
    '交易数': len(t1_trades),
    '胜率': f"{original_winrate:.1f}%",
    '总盈亏': f"{pnl_v3:+.0f}",
    '改善': f"{pnl_v3 - original_pnl:+.0f}"
})

# 策略5: 最优组合
def simulate_optimal(trades, position=0.7):
    total = 0
    for _, row in trades.iterrows():
        entry = row['开仓价格']
        exit_p = row['平仓价格']
        high = row['期间最高']
        rsi = 50

        # 获取RSI
        mask = data.index <= row['开仓时间']
        if mask.any():
            try:
                rsi = data.loc[data.index[mask][-1], 'RSI_14']
            except:
                pass

        # 只在RSI 35-55交易
        if not (35 <= rsi <= 55):
            continue

        pnl_pct = (exit_p - entry) / entry
        max_profit_pct = (high - entry) / entry

        # 移动止损 + 分批止盈组合
        if pnl_pct <= -0.008:
            actual = -0.008
        elif max_profit_pct >= 0.015:
            # 部分止盈
            trail_stop = max_profit_pct - 0.005
            if pnl_pct <= trail_stop and trail_stop > 0.01:
                actual = 0.015 * 0.5 + trail_stop * 0.5
            elif pnl_pct >= 0.03:
                actual = 0.015 * 0.5 + 0.03 * 0.5
            else:
                actual = pnl_pct
        elif pnl_pct >= 0.025:
            actual = 0.025
        else:
            actual = pnl_pct

        total += actual * 1000000 * position
    return total

pnl_v4 = simulate_optimal(t1_trades)

# 统计优化V4的交易数
count_v4 = 0
for _, row in t1_trades.iterrows():
    mask = data.index <= row['开仓时间']
    if mask.any():
        try:
            rsi = data.loc[data.index[mask][-1], 'RSI_14']
            if 35 <= rsi <= 55:
                count_v4 += 1
        except:
            pass

strategies.append({
    '策略': '优化V4(RSI35-55+移动+分批)',
    '交易数': count_v4,
    '胜率': '-',
    '总盈亏': f"{pnl_v4:+.0f}",
    '改善': f"{pnl_v4 - original_pnl:+.0f}"
})

# 打印结果
results_df = pd.DataFrame(strategies)
print("\n" + results_df.to_string(index=False))

# 最优参数推荐
print("\n" + "="*60)
print("🏆 最优策略推荐")
print("="*60)

best_pnl = max([original_pnl, pnl_v1, pnl_v2, pnl_v3, pnl_v4])
if best_pnl == pnl_v2:
    print("【推荐】优化V2: 移动止损策略")
    print("  参数: 止损0.8%, 止盈2.5%, 移动激活1%, 移动距离0.5%")
    print(f"  预期收益: {pnl_v2:+.0f}元")
elif best_pnl == pnl_v4:
    print("【推荐】优化V4: 综合策略")
    print("  参数: RSI35-55过滤 + 移动止损 + 分批止盈")
    print(f"  预期收益: {pnl_v4:+.0f}元")
elif best_pnl == pnl_v3:
    print("【推荐】优化V3: 分批止盈策略")
    print("  参数: 第一批1.5%止盈50%, 第二批3%止盈剩余")
    print(f"  预期收益: {pnl_v3:+.0f}元")
else:
    print("【推荐】优化V1: 固定止盈2.5%")
    print(f"  预期收益: {pnl_v1:+.0f}元")

print("\n" + "="*60)
print("✅ 分析完成")
print("="*60)