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

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
创业板50 T+1 多维度深度分析
"""
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import warnings
import sys
warnings.filterwarnings('ignore')

# Redirect stdout
from io import StringIO
old_stdout = sys.stdout
sys.stdout = StringIO()

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['High_Low_Pct'] = (df['High'] - df['Low']) / df['Close'].shift(1)
    df['Close_Open_Pct'] = (df['Close'] - df['Open']) / df['Open']
    df.ffill(inplace=True)
    df.dropna(inplace=True)
    return df

# 运行回测
initial_capital = 1000000
raw_data = load_local_data('cyb50_30min_2023_to_20260325.csv')

config_manager = ConfigManager('config.json')
fetcher = IntradayDataFetcher(config_manager)
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()
t1_trades = simulate_t1_trades(data_with_indicators, long_trades, initial_capital)

# Restore stdout
sys.stdout = old_stdout

print('='*80)
print('创业板50 T+1 多维度深度分析')
print('='*80)

# 准备数据
t1_trades['开仓时间'] = pd.to_datetime(t1_trades['开仓时间'])
t1_trades['平仓时间'] = pd.to_datetime(t1_trades['平仓时间'])
t1_trades['持仓小时'] = (t1_trades['平仓时间'] - t1_trades['开仓时间']).dt.total_seconds() / 3600
t1_trades['持仓周期'] = ((t1_trades['平仓时间'] - t1_trades['开仓时间']).dt.total_seconds() / 1800).astype(int)
t1_trades['是否盈利'] = t1_trades['盈亏金额'] > 0
t1_trades['盈亏比例'] = t1_trades['盈亏金额'] / initial_capital * 100
t1_trades['开仓年份'] = t1_trades['开仓时间'].dt.year

# 将交易与原始数据合并，获取更多维度
for idx, trade in t1_trades.iterrows():
    try:
        # 获取开仓时的市场数据
        mask = data_with_indicators.index <= trade['开仓时间']
        if mask.any():
            current_idx = data_with_indicators.index[mask][-1]
            current_data = data_with_indicators.loc[current_idx]

            # 计算价格位置（过去20日高低点中的位置）
            past_20d = data_with_indicators.loc[:current_idx].tail(20*8)  # 20天，每天8个30分钟
            if len(past_20d) > 0:
                high_20d = past_20d['High'].max()
                low_20d = past_20d['Low'].min()
                current_price = current_data['Close']
                t1_trades.loc[idx, '价格位置_20d'] = (current_price - low_20d) / (high_20d - low_20d) * 100 if high_20d > low_20d else 50

            # 获取技术指标
            for col in ['RSI_14', 'MACD', 'MACD_signal', 'KDJ_K', 'KDJ_D', '布林带宽度']:
                if col in current_data.index:
                    t1_trades.loc[idx, col] = current_data[col]
    except:
        pass

# ========== 1. 持仓时长分析 ==========
print('\n' + '='*80)
print('【1】持仓时长深度分析')
print('='*80)

# 按持仓周期分组
period_stats = t1_trades.groupby('持仓周期').agg({
    '盈亏金额': ['count', 'sum', 'mean'],
    '是否盈利': 'sum'
}).round(2)
period_stats.columns = ['交易次数', '总盈亏', '平均盈亏', '盈利次数']
period_stats['胜率'] = (period_stats['盈利次数'] / period_stats['交易次数'] * 100).round(1)

print('\n按持仓周期统计(30分钟为单位):')
print(period_stats[period_stats['交易次数'] >= 5].to_string())

# 持仓时长分布
print('\n持仓时长分布:')
t1_trades['持仓时长分类'] = pd.cut(t1_trades['持仓小时'],
    bins=[0, 1, 2, 4, 8, 16, 100],
    labels=['<1小时', '1-2小时', '2-4小时', '4-8小时', '8-16小时', '>16小时'])

duration_stats = t1_trades.groupby('持仓时长分类').agg({
    '盈亏金额': ['count', 'sum', 'mean'],
    '是否盈利': 'sum'
}).round(2)
duration_stats.columns = ['交易次数', '总盈亏', '平均盈亏', '盈利次数']
duration_stats['胜率'] = (duration_stats['盈利次数'] / duration_stats['交易次数'] * 100).round(1)
print(duration_stats.to_string())

# ========== 2. 价格位置分析 ==========
print('\n' + '='*80)
print('【2】开仓价格位置分析')
print('='*80)

# 将价格位置分箱
t1_trades['价格位置分类'] = pd.cut(t1_trades['价格位置_20d'],
    bins=[0, 20, 40, 60, 80, 100],
    labels=['极低(0-20%)', '低位(20-40%)', '中位(40-60%)', '高位(60-80%)', '极高(80-100%)'])

position_stats = t1_trades.groupby('价格位置分类').agg({
    '盈亏金额': ['count', 'sum', 'mean'],
    '是否盈利': 'sum'
}).round(2)
position_stats.columns = ['交易次数', '总盈亏', '平均盈亏', '盈利次数']
position_stats['胜率'] = (position_stats['盈利次数'] / position_stats['交易次数'] * 100).round(1)

print('\n按20日价格位置统计:')
print(position_stats.to_string())

# T+1调整与价格位置的关系
print('\nT+1调整与价格位置的关系:')
position_t1 = pd.crosstab(t1_trades['价格位置分类'], t1_trades['T+1调整'],
                          values=t1_trades['盈亏金额'], aggfunc='mean')
print(position_t1.round(2).to_string())

# ========== 3. 止盈止损分析 ==========
print('\n' + '='*80)
print('【3】止盈止损触发分析')
print('='*80)

# 解析退出原因
exit_reasons = t1_trades['退出原因'].str.extract(r'(.*?)(?:触发|$)')
exit_reasons.columns = ['退出类型']
exit_reasons['退出类型'] = exit_reasons['退出类型'].str.strip()
t1_trades['退出类型'] = exit_reasons['退出类型']

exit_stats = t1_trades.groupby('退出类型').agg({
    '盈亏金额': ['count', 'sum', 'mean'],
    '是否盈利': 'sum'
}).round(2)
exit_stats.columns = ['交易次数', '总盈亏', '平均盈亏', '盈利次数']
exit_stats['胜率'] = (exit_stats['盈利次数'] / exit_stats['交易次数'] * 100).round(1)

print('\n按退出类型统计:')
print(exit_stats.to_string())

# ========== 4. 波动率分析 ==========
print('\n' + '='*80)
print('【4】市场波动率与交易表现')
print('='*80)

# 计算开仓时的波动率
t1_trades['开仓波动率'] = np.nan
for idx, trade in t1_trades.iterrows():
    try:
        mask = data_with_indicators.index <= trade['开仓时间']
        if mask.sum() >= 20:
            recent_data = data_with_indicators.loc[mask].tail(20)
            volatility = recent_data['Returns'].std() * np.sqrt(48)  # 年化波动率，每天48个30分钟
            t1_trades.loc[idx, '开仓波动率'] = volatility
    except:
        pass

# 波动率分箱
t1_trades['波动率分类'] = pd.cut(t1_trades['开仓波动率'],
    bins=[0, 0.15, 0.25, 0.35, 0.5, 2.0],
    labels=['低波动(<15%)', '中低波动(15-25%)', '中等波动(25-35%)', '高波动(35-50%)', '极高波动(>50%)'])

vol_stats = t1_trades.groupby('波动率分类').agg({
    '盈亏金额': ['count', 'sum', 'mean'],
    '是否盈利': 'sum'
}).round(2)
vol_stats.columns = ['交易次数', '总盈亏', '平均盈亏', '盈利次数']
vol_stats['胜率'] = (vol_stats['盈利次数'] / vol_stats['交易次数'] * 100).round(1)

print('\n按市场波动率统计:')
print(vol_stats.to_string())

# ========== 5. 交易量分析 ==========
print('\n' + '='*80)
print('【5】交易量分析')
print('='*80)

# 计算开仓时的相对成交量
t1_trades['相对成交量'] = np.nan
for idx, trade in t1_trades.iterrows():
    try:
        mask = data_with_indicators.index <= trade['开仓时间']
        if mask.sum() >= 20:
            recent_data = data_with_indicators.loc[mask].tail(20)
            current_vol = recent_data['Volume'].iloc[-1]
            avg_vol = recent_data['Volume'].mean()
            t1_trades.loc[idx, '相对成交量'] = current_vol / avg_vol if avg_vol > 0 else 1
    except:
        pass

t1_trades['成交量分类'] = pd.cut(t1_trades['相对成交量'],
    bins=[0, 0.5, 1.0, 1.5, 2.0, 10.0],
    labels=['缩量(<0.5x)', '正常(0.5-1x)', '放量(1-1.5x)', '大量(1.5-2x)', '巨量(>2x)'])

volume_stats = t1_trades.groupby('成交量分类').agg({
    '盈亏金额': ['count', 'sum', 'mean'],
    '是否盈利': 'sum'
}).round(2)
volume_stats.columns = ['交易次数', '总盈亏', '平均盈亏', '盈利次数']
volume_stats['胜率'] = (volume_stats['盈利次数'] / volume_stats['交易次数'] * 100).round(1)

print('\n按相对成交量统计:')
print(volume_stats.to_string())

# ========== 6. 入场信号分析 ==========
print('\n' + '='*80)
print('【6】入场信号分析')
print('='*80)

# 解析入场信号
if '入场信号' in t1_trades.columns:
    # 统计各类信号出现的频率
    signal_types = []
    for signals in t1_trades['入场信号'].dropna():
        if isinstance(signals, str):
            for s in signals.split(','):
                signal_types.append(s.strip())

    from collections import Counter
    signal_counter = Counter(signal_types)

    print('\n入场信号频率统计:')
    for signal, count in signal_counter.most_common(15):
        print(f'  {signal}: {count}次')

    # 分析特定信号的表现
    print('\n关键信号组合分析:')
    for signal in ['MACD金叉', 'RSI超卖', 'KDJ金叉', '放量突破']:
        mask = t1_trades['入场信号'].str.contains(signal, na=False)
        if mask.sum() > 0:
            subset = t1_trades[mask]
            win_rate = (subset['盈亏金额'] > 0).mean() * 100
            avg_pnl = subset['盈亏金额'].mean()
            print(f'  {signal}: {mask.sum()}笔, 胜率{win_rate:.1f}%, 平均盈亏{avg_pnl:+,.0f}元')

# ========== 7. 盈亏分布分析 ==========
print('\n' + '='*80)
print('【7】盈亏分布特征')
print('='*80)

profit_trades = t1_trades[t1_trades['盈亏金额'] > 0]
loss_trades = t1_trades[t1_trades['盈亏金额'] < 0]

print('\n盈利交易特征:')
print(f'  总盈利: {profit_trades["盈亏金额"].sum():,.0f}元')
print(f'  平均盈利: {profit_trades["盈亏金额"].mean():,.0f}元')
print(f'  最大单笔: {profit_trades["盈亏金额"].max():,.0f}元')
print(f'  中位数: {profit_trades["盈亏金额"].median():,.0f}元')
print(f'  25分位: {profit_trades["盈亏金额"].quantile(0.25):,.0f}元')
print(f'  75分位: {profit_trades["盈亏金额"].quantile(0.75):,.0f}元')

print('\n亏损交易特征:')
print(f'  总亏损: {loss_trades["盈亏金额"].sum():,.0f}元')
print(f'  平均亏损: {loss_trades["盈亏金额"].mean():,.0f}元')
print(f'  最大单笔: {loss_trades["盈亏金额"].min():,.0f}元')
print(f'  中位数: {loss_trades["盈亏金额"].median():,.0f}元')
print(f'  25分位: {loss_trades["盈亏金额"].quantile(0.25):,.0f}元')
print(f'  75分位: {loss_trades["盈亏金额"].quantile(0.75):,.0f}元')

# 盈亏比分析
total_profit = profit_trades['盈亏金额'].sum()
total_loss = abs(loss_trades['盈亏金额'].sum())
profit_factor = total_profit / total_loss if total_loss > 0 else 0

print(f'\n盈亏比: {profit_factor:.2f}')
print(f'盈亏比(单笔平均): {abs(profit_trades["盈亏金额"].mean() / loss_trades["盈亏金额"].mean()):.2f}')

# ========== 8. 资金曲线分析 ==========
print('\n' + '='*80)
print('【8】资金曲线与回撤分析')
print('='*80)

# 计算资金曲线
t1_trades_sorted = t1_trades.sort_values('平仓时间')
cumulative = [initial_capital]
for pnl in t1_trades_sorted['盈亏金额']:
    cumulative.append(cumulative[-1] + pnl)

# 计算最大回撤
max_drawdown = 0
max_drawdown_pct = 0
peak = initial_capital
peak_idx = 0
drawdown_start = 0
drawdown_end = 0

for i, capital in enumerate(cumulative):
    if capital > peak:
        peak = capital
        peak_idx = i
    else:
        dd = peak - capital
        dd_pct = dd / peak * 100
        if dd_pct > max_drawdown_pct:
            max_drawdown = dd
            max_drawdown_pct = dd_pct
            drawdown_start = peak_idx
            drawdown_end = i

print(f'\n资金曲线统计:')
print(f'  期初资金: {initial_capital:,.0f}元')
print(f'  期末资金: {cumulative[-1]:,.0f}元')
print(f'  峰值资金: {max(cumulative):,.0f}元')
print(f'  谷值资金: {min(cumulative):,.0f}元')
print(f'  最大回撤: {max_drawdown:,.0f}元 ({max_drawdown_pct:.2f}%)')
print(f'  回撤开始: 第{drawdown_start}笔交易')
print(f'  回撤结束: 第{drawdown_end}笔交易')
print(f'  恢复耗时: {drawdown_end - drawdown_start}笔交易')

# 回撤期间分析
if drawdown_start < len(t1_trades_sorted) and drawdown_end < len(t1_trades_sorted):
    dd_period = t1_trades_sorted.iloc[drawdown_start:drawdown_end+1]
    print(f'\n最大回撤期间特征:')
    print(f'  交易次数: {len(dd_period)}笔')
    print(f'  期间盈亏: {dd_period["盈亏金额"].sum():,.0f}元')
    print(f'  盈利笔数: {(dd_period["盈亏金额"] > 0).sum()}笔')
    print(f'  亏损笔数: {(dd_period["盈亏金额"] < 0).sum()}笔')
    print(f'  期间胜率: {(dd_period["盈亏金额"] > 0).mean()*100:.1f}%')

# ========== 9. 多维度交叉分析 ==========
print('\n' + '='*80)
print('【9】多维度交叉分析')
print('='*80)

# 时间 + T+1调整
cross_analysis = t1_trades.groupby(['开仓年份', 'T+1调整']).agg({
    '盈亏金额': ['count', 'sum', 'mean'],
    '持仓小时': 'mean'
}).round(2)
cross_analysis.columns = ['交易次数', '总盈亏', '平均盈亏', '平均持仓']

print('\n年度 x T+1调整:')
print(cross_analysis.to_string())

# 价格位置 + 持仓时长
if t1_trades['价格位置分类'].notna().sum() > 0 and t1_trades['持仓时长分类'].notna().sum() > 0:
    print('\n价格位置 x 持仓时长 (平均盈亏):')
    cross_pnl = pd.pivot_table(t1_trades, values='盈亏金额',
                                index='价格位置分类',
                                columns='持仓时长分类',
                                aggfunc='mean')
    print(cross_pnl.round(0).to_string())

# 波动率 + T+1调整
if t1_trades['波动率分类'].notna().sum() > 0:
    print('\n波动率 x T+1调整 (平均盈亏):')
    cross_vol = pd.pivot_table(t1_trades, values='盈亏金额',
                                index='波动率分类',
                                columns='T+1调整',
                                aggfunc='mean')
    print(cross_vol.round(0).to_string())

# ========== 10. 亏损根因分析 ==========
print('\n' + '='*80)
print('【10】亏损根因深度分析')
print('='*80)

# 找出最差的交易
worst_trades = t1_trades.nsmallest(10, '盈亏金额')
print('\n亏损TOP10交易特征:')
for idx, row in worst_trades.iterrows():
    print(f"\n  #{idx+1} 亏损{row['盈亏金额']:,.0f}元")
    print(f"    时间: {row['开仓时间']} → {row['平仓时间']}")
    print(f"    持仓: {row['持仓小时']:.1f}小时")
    print(f"    退出: {row['退出原因']}")
    print(f"    T+1调整: {row['T+1调整']}")

# 亏损交易共性分析
print('\n\n亏损交易共性:')
loss_trades = t1_trades[t1_trades['盈亏金额'] < 0]

# 退出类型分布
loss_exit = loss_trades['退出类型'].value_counts()
print('\n  退出类型分布:')
for exit_type, count in loss_exit.head(5).items():
    pct = count / len(loss_trades) * 100
    avg_loss = loss_trades[loss_trades['退出类型'] == exit_type]['盈亏金额'].mean()
    print(f'    {exit_type}: {count}笔 ({pct:.1f}%), 平均亏损{avg_loss:,.0f}元')

# T+1调整分布
loss_t1 = loss_trades['T+1调整'].value_counts()
print('\n  T+1调整分布:')
for t1_type, count in loss_t1.items():
    pct = count / len(loss_trades) * 100
    avg_loss = loss_trades[loss_trades['T+1调整'] == t1_type]['盈亏金额'].mean()
    print(f'    {t1_type}: {count}笔 ({pct:.1f}%), 平均亏损{avg_loss:,.0f}元')

# 持仓时长分布
loss_duration = loss_trades['持仓时长分类'].value_counts()
print('\n  亏损交易持仓时长分布:')
for duration, count in loss_duration.items():
    if pd.notna(duration):
        pct = count / len(loss_trades) * 100
        print(f'    {duration}: {count}笔 ({pct:.1f}%)')

# ========== 11. 盈利因子分析 ==========
print('\n' + '='*80)
print('【11】盈利交易成功因子')
print('='*80)

profit_trades = t1_trades[t1_trades['盈亏金额'] > 0]

print('\n盈利交易特征:')

# 退出类型分布
profit_exit = profit_trades['退出类型'].value_counts()
print('\n  退出类型分布:')
for exit_type, count in profit_exit.head(5).items():
    pct = count / len(profit_trades) * 100
    avg_profit = profit_trades[profit_trades['退出类型'] == exit_type]['盈亏金额'].mean()
    print(f'    {exit_type}: {count}笔 ({pct:.1f}%), 平均盈利{avg_profit:,.0f}元')

# T+1调整分布
profit_t1 = profit_trades['T+1调整'].value_counts()
print('\n  T+1调整分布:')
for t1_type, count in profit_t1.items():
    pct = count / len(profit_trades) * 100
    avg_profit = profit_trades[profit_trades['T+1调整'] == t1_type]['盈亏金额'].mean()
    print(f'    {t1_type}: {count}笔 ({pct:.1f}%), 平均盈利{avg_profit:,.0f}元')

# 持仓时长分布
profit_duration = profit_trades['持仓时长分类'].value_counts()
print('\n  盈利交易持仓时长分布:')
for duration, count in profit_duration.items():
    if pd.notna(duration):
        pct = count / len(profit_trades) * 100
        print(f'    {duration}: {count}笔 ({pct:.1f}%)')

# ========== 12. 综合改进建议 ==========
print('\n' + '='*80)
print('【12】综合改进建议')
print('='*80)

print("""
基于多维度分析，提出以下改进策略:

【A. 时间维度优化】
1. 避开13:00-14:00开仓 (胜率22.9%)
2. 周五减少或停止交易 (胜率32.8%)
3. 优选周二、周三交易

【B. 持仓管理优化】
1. 限制持仓时长 > 16小时的交易 (过夜风险)
2. T+1调整后增加风控: 若当日已盈利，尾盘不平仓
3. 高波动期(>35%)降低仓位50%

【C. 价格位置过滤】
1. 避免在极高位置(>80%)开仓做多
2. 优先在低位(<40%)开仓
3. 结合20日高低点判断趋势

【D. 止盈止损优化】
1. 盈利交易平均持仓较短，可考虑收紧止盈
2. 亏损交易多因止损触发，检查止损位置是否合理
3. 考虑移动止损保护利润

【E. 信号质量提升】
1. 过滤低成交量信号(<0.5倍均量)
2. 增加趋势确认信号，避免逆势交易
3. 连续亏损3笔后暂停交易

【F. 风险管理强化】
1. 单日亏损超过3万当日停止
2. 连续2天亏损后降低仓位至30%
3. 月度亏损超过10万暂停策略复盘
""")

print('\n' + '='*80)
print('分析完成')
print('='*80)