cyb50-quant/cat-fly/diagnose_strategy.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
诊断脚本 - 检查交易策略的bug
"""

import pandas as pd
import numpy as np
import sys
sys.path.insert(0, '/root/.openclaw/workspace/cat-fly')
sys.path.insert(0, '/root/.openclaw/workspace/quant')

def diagnose_strategy():
    """诊断策略问题"""
    print("="*60)
    print("🔍 策略诊断报告")
    print("="*60)
    
    # 加载数据
    df = pd.read_csv('/root/.openclaw/workspace/quant/cyb50_baostock.csv')
    df['date'] = pd.to_datetime(df['date'])
    df = df.set_index('date').sort_index()
    
    for col in ['open', 'high', 'low', 'close', 'volume']:
        df[col] = pd.to_numeric(df[col], errors='coerce')
    
    # 计算指标
    df['ma10'] = df['close'].rolling(window=10).mean()
    df['ma30'] = df['close'].rolling(window=30).mean()
    df['high_20'] = df['high'].rolling(window=20).max()
    df['low_20'] = df['low'].rolling(window=20).min()
    df['ret_10'] = df['close'].pct_change(periods=10)
    
    # 买入条件
    buy_cond = (
        (df['close'] > df['ma10']) & 
        (df['ma10'] > df['ma30']) & 
        (df['close'] >= df['high_20'] * 0.995) & 
        (df['ret_10'] > 0.02)
    )
    
    # 卖出条件
    sell_cond = (
        (df['close'] < df['ma30']) | 
        (df['close'] <= df['low_20'] * 1.005)
    )
    
    print("\n📊 信号统计:")
    print(f"  买入信号天数: {buy_cond.sum()}")
    print(f"  卖出信号天数: {sell_cond.sum()}")
    
    # 检查同一天买卖冲突
    both_cond = buy_cond & sell_cond
    print(f"  ⚠️ 同一天买卖冲突: {both_cond.sum()} 天")
    
    if both_cond.sum() > 0:
        print("\n  冲突日期示例:")
        for date in df[both_cond].index[:5]:
            print(f"    {date.date()}: 买={buy_cond.loc[date]}, 卖={sell_cond.loc[date]}")
    
    # 模拟回测，检查交易
    print("\n📈 交易明细检查:")
    
    df['signal'] = 0
    df.loc[buy_cond, 'signal'] = 1
    df.loc[sell_cond, 'signal'] = -1
    
    position = 0
    entry_price = 0
    peak_price = 0
    capital = 1000000
    trades = []
    issues = []
    
    for i in range(30, len(df)):
        date = df.index[i]
        price = df['close'].iloc[i]
        signal = df['signal'].iloc[i]
        
        # 检查移动止损
        stop_loss_triggered = False
        if position > 0:
            if price > peak_price:
                peak_price = price
            if price < peak_price * 0.90:
                signal = -1
                stop_loss_triggered = True
        
        # 买入
        if signal == 1 and position == 0:
            position = 1
            entry_price = price
            peak_price = price
            trades.append({
                'date': date,
                'action': 'BUY',
                'price': price,
                'capital': capital,
                'stop_loss': peak_price * 0.90
            })
        
        # 卖出
        elif signal == -1 and position == 1:
            pnl = (price / entry_price - 1) * capital
            capital += pnl
            
            exit_reason = '止损' if stop_loss_triggered else '信号'
            hold_days = (date - trades[-1]['date']).days if trades else 0
            
            trades.append({
                'date': date,
                'action': 'SELL',
                'price': price,
                'capital': capital,
                'pnl': pnl,
                'return_pct': (price / entry_price - 1) * 100,
                'reason': exit_reason,
                'hold_days': hold_days
            })
            
            # 检查问题
            if hold_days == 0:
                issues.append(f"⚠️ {date.date()}: 同日买卖 (T+0交易)")
            elif hold_days == 1:
                issues.append(f"⚠️ {date.date()}: 隔日交易 (T+1卖出)")
            
            position = 0
    
    print(f"\n  总交易次数: {len([t for t in trades if t['action'] == 'SELL'])}")
    print(f"  发现问题: {len(issues)}")
    
    if issues:
        print("\n  问题列表:")
        for issue in issues[:10]:
            print(f"    {issue}")
    
    # 检查连续信号
    print("\n🔍 连续信号检查:")
    signal_changes = df['signal'].diff().abs()
    consecutive_days = 0
    max_consecutive = 0
    
    for i in range(30, len(df)):
        if df['signal'].iloc[i] == 1:
            consecutive_days += 1
            max_consecutive = max(max_consecutive, consecutive_days)
        else:
            consecutive_days = 0
    
    print(f"  最大连续买入信号: {max_consecutive} 天")
    
    # 显示最近5次交易
    print("\n📋 最近5次完整交易:")
    complete_trades = []
    for i in range(0, len(trades)-1, 2):
        if i+1 < len(trades):
            buy = trades[i]
            sell = trades[i+1]
            complete_trades.append({
                'entry': buy['date'].strftime('%Y-%m-%d'),
                'exit': sell['date'].strftime('%Y-%m-%d'),
                'entry_price': buy['price'],
                'exit_price': sell['price'],
                'return': sell['return_pct'],
                'hold_days': sell['hold_days']
            })
    
    for t in complete_trades[-5:]:
        print(f"  买入: {t['entry']} @ {t['entry_price']:.2f}")
        print(f"  卖出: {t['exit']} @ {t['exit_price']:.2f}")
        print(f"  收益: {t['return']:+.2f}%, 持仓: {t['hold_days']}天")
        print()
    
    print("="*60)
    print(f"最终资产: {capital:,.0f}元, 收益率: {(capital/1000000-1)*100:+.2f}%")
    print("="*60)

if __name__ == "__main__":
    diagnose_strategy()