brunoandClaude
b28e9aa308
- 移动调试脚本到 debug_scripts/ 目录 - 清理 Python 缓存文件和编译文件 - 添加综合数据调度器和相关文档 - 整理项目结构，提高代码可维护性 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <[email protected]>
359 lines
14 KiB
Python
359 lines
14 KiB
Python
#!/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() |