#!/usr/bin/env python3 """ 策略验证系统调试器 系统性地分析和修复策略验证中的问题 """ import sys import os sys.path.append(os.path.dirname(os.path.abspath(__file__))) import pandas as pd import numpy as np from datetime import datetime, timedelta import json import logging # 设置详细日志 logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) class StrategyValidatorDebugger: """策略验证调试器""" def __init__(self): self.debug_data = {} def debug_data_preparation(self): """调试数据准备阶段""" logger.info("🔍 开始调试数据准备...") try: from strategy_validation.advanced_validator import AdvancedStrategyValidator, StrategyConfig config = StrategyConfig( name="调试策略", description="用于调试", parameters={'ma5': 5, 'ma10': 10, 'ma20': 20, 'volume_threshold': 1.2}, risk_params={'max_position_size': 0.20, 'stop_loss': 0.05, 'take_profit': 0.10} ) validator = AdvancedStrategyValidator(config) # 准备数据 all_data = validator._prepare_data(['000001.SZ'], '2024-08-01', '2024-12-31') if not all_data: logger.error("❌ 数据准备失败：没有获取到数据") return False for stock_code, data in all_data.items(): logger.info(f"📊 {stock_code} 数据信息:") logger.info(f" - 数据形状: {data.shape}") logger.info(f" - 列名: {list(data.columns)}") logger.info(f" - 日期范围: {data.index[0]} 到 {data.index[-1]}") logger.info(f" - 缺失值: {data.isnull().sum().sum()}") # 检查关键字段 required_fields = ['open', 'high', 'low', 'close', 'volume'] for field in required_fields: if field in data.columns: logger.info(f" - ✅ {field}: 范围 {data[field].min():.2f} - {data[field].max():.2f}") else: logger.error(f" - ❌ 缺少字段: {field}") self.debug_data[stock_code] = data return True except Exception as e: logger.error(f"❌ 数据准备调试失败: {e}") import traceback traceback.print_exc() return False def debug_signal_generation(self): """调试信号生成阶段""" logger.info("🔍 开始调试信号生成...") if not self.debug_data: logger.error("❌ 没有可用的调试数据") return False try: from api.strategy_validation_api import triple_ma_strategy for stock_code, data in list(self.debug_data.items()): logger.info(f"🎯 生成 {stock_code} 的信号...") # 调试策略函数调用 signals_df = triple_ma_strategy(data, ma5=5, ma10=10, ma20=20, volume_threshold=1.2) if isinstance(signals_df, pd.DataFrame): logger.info(f" - ✅ 信号DataFrame形状: {signals_df.shape}") logger.info(f" - 信号统计: {dict(signals_df['signal'].value_counts())}") # 检查非零信号 non_zero_signals = signals_df[signals_df['signal'] != 0] logger.info(f" - 非零信号数量: {len(non_zero_signals)}") if len(non_zero_signals) > 0: logger.info(f" - 信号强度范围: {non_zero_signals['signal_strength'].min():.3f} - {non_zero_signals['signal_strength'].max():.3f}") # 转换为字典格式 signals_list = [] for idx, row in non_zero_signals.iterrows(): signal_dict = { 'date': idx.strftime('%Y-%m-%d') if hasattr(idx, 'strftime') else str(idx), 'signal': int(row['signal']), 'strength': float(row['signal_strength']), 'price': float(data.loc[idx, 'close']) if idx in data.index else 0 } signals_list.append(signal_dict) logger.info(f" - ✅ 转换后信号列表长度: {len(signals_list)}") # 显示前几个信号示例 for i, signal in enumerate(signals_list[:3]): logger.info(f" - 信号 {i+1}: {signal}") self.debug_data[f"{stock_code}_signals"] = signals_list else: logger.warning(f" - ⚠️ {stock_code} 没有生成非零信号") else: logger.error(f" - ❌ 策略函数返回类型错误: {type(signals_df)}") return True except Exception as e: logger.error(f"❌ 信号生成调试失败: {e}") import traceback traceback.print_exc() return False def debug_trading_simulation(self): """调试交易模拟阶段""" logger.info("🔍 开始调试交易模拟...") try: # 创建简化的交易模拟器 portfolio = { 'equity_curve': [], 'trades': [], 'positions': {}, 'cash': 100000, 'total_value': 100000 } # 获取所有交易日期 stock_data = None signals_list = None for key, value in self.debug_data.items(): if not key.endswith('_signals'): stock_data = value stock_code = key else: signals_list = value if stock_data is None or signals_list is None: logger.error("❌ 缺少股票数据或信号数据") return False logger.info(f"📈 股票数据: {stock_data.shape}") logger.info(f"🎯 信号数量: {len(signals_list)}") # 按日期模拟交易 trade_count = 0 for signal in signals_list: signal_date = signal['date'] signal_type = signal['signal'] signal_price = signal['price'] logger.info(f"🔄 处理信号: {signal_date} | 类型:{signal_type} | 价格:{signal_price:.2f}") if signal_type == 1: # 买入信号 # 计算可买入股数 max_investment = portfolio['cash'] * 0.2 # 最大20%仓位 shares = int(max_investment / signal_price) if shares > 0 and portfolio['cash'] >= shares * signal_price: cost = shares * signal_price portfolio['cash'] -= cost portfolio['positions'][stock_code] = { 'shares': shares, 'cost_price': signal_price, 'entry_date': signal_date } trade_record = { 'date': signal_date, 'action': 'BUY', 'shares': shares, 'price': signal_price, 'amount': cost } portfolio['trades'].append(trade_record) trade_count += 1 logger.info(f" ✅ 买入成功: {shares}股 @ {signal_price:.2f} = {cost:.2f}") elif signal_type == -1 and stock_code in portfolio['positions']: # 卖出信号 position = portfolio['positions'][stock_code] shares = position['shares'] revenue = shares * signal_price portfolio['cash'] += revenue trade_record = { 'date': signal_date, 'action': 'SELL', 'shares': shares, 'price': signal_price, 'amount': revenue, 'profit': revenue - (shares * position['cost_price']) } portfolio['trades'].append(trade_record) trade_count += 1 logger.info(f" ✅ 卖出成功: {shares}股 @ {signal_price:.2f} = {revenue:.2f}") logger.info(f" 💰 盈亏: {trade_record['profit']:.2f}") del portfolio['positions'][stock_code] logger.info(f"📊 交易模拟完成:") logger.info(f" - 总交易次数: {trade_count}") logger.info(f" - 剩余现金: {portfolio['cash']:.2f}") logger.info(f" - 持仓数量: {len(portfolio['positions'])}") # 计算总资产 total_value = portfolio['cash'] for stock_code, position in portfolio['positions'].items(): # 使用最新价格 latest_price = stock_data['close'].iloc[-1] position_value = position['shares'] * latest_price total_value += position_value logger.info(f" - 持仓 {stock_code}: {position['shares']}股 @ {latest_price:.2f} = {position_value:.2f}") logger.info(f" - 总资产: {total_value:.2f}") logger.info(f" - 总收益: {total_value - 100000:.2f} ({(total_value/100000-1)*100:.2f}%)") self.debug_data['portfolio'] = portfolio self.debug_data['final_value'] = total_value return trade_count > 0 except Exception as e: logger.error(f"❌ 交易模拟调试失败: {e}") import traceback traceback.print_exc() return False def debug_metrics_calculation(self): """调试指标计算阶段""" logger.info("🔍 开始调试指标计算...") try: portfolio = self.debug_data.get('portfolio') final_value = self.debug_data.get('final_value', 100000) if not portfolio: logger.error("❌ 没有交易组合数据") return False trades = portfolio['trades'] if not trades: logger.warning("⚠️ 没有交易记录，收益率为0") return True # 计算基本指标 total_return = (final_value - 100000) / 100000 logger.info(f"📊 总收益率: {total_return:.4f} ({total_return*100:.2f}%)") # 计算交易级别的收益率 returns = [] for trade in trades: if trade['action'] == 'SELL' and 'profit' in trade: returns.append(trade['profit'] / trade['amount']) if returns: avg_return = np.mean(returns) volatility = np.std(returns) if len(returns) > 1 else 0 logger.info(f"📊 平均单次收益率: {avg_return:.4f}") logger.info(f"📊 收益率波动率: {volatility:.4f}") # 计算夏普比率 (简化版) if volatility > 0: sharpe_ratio = avg_return / volatility logger.info(f"📊 夏普比率: {sharpe_ratio:.4f}") else: logger.info("📊 夏普比率: N/A (波动率为0)") # 胜率 profitable_trades = [r for r in returns if r > 0] win_rate = len(profitable_trades) / len(returns) logger.info(f"📊 胜率: {win_rate:.4f} ({win_rate*100:.1f}%)") else: logger.warning("⚠️ 没有完成的交易对，无法计算详细指标") return True except Exception as e: logger.error(f"❌ 指标计算调试失败: {e}") import traceback traceback.print_exc() return False def run_complete_debug(self): """运行完整调试流程""" logger.info("🚀 开始完整的策略验证系统调试...") logger.info("=" * 80) success_count = 0 # 步骤1: 数据准备 if self.debug_data_preparation(): success_count += 1 logger.info("=" * 80) # 步骤2: 信号生成 if self.debug_signal_generation(): success_count += 1 logger.info("=" * 80) # 步骤3: 交易模拟 if self.debug_trading_simulation(): success_count += 1 logger.info("=" * 80) # 步骤4: 指标计算 if self.debug_metrics_calculation(): success_count += 1 logger.info("=" * 80) logger.info(f"🎯 调试完成: {success_count}/4 个阶段成功") if success_count == 4: logger.info("✅ 所有调试阶段成功完成!") return True else: logger.error(f"❌ 有 {4-success_count} 个阶段失败") return False def main(): debugger = StrategyValidatorDebugger() success = debugger.run_complete_debug() if success: print("\n🎉 调试成功! 策略验证系统的各个环节都在正常工作。") else: print("\n⚠️ 发现问题，需要进一步修复。") if __name__ == "__main__": main()