brunoandClaude
b28e9aa308
- 移动调试脚本到 debug_scripts/ 目录 - 清理 Python 缓存文件和编译文件 - 添加综合数据调度器和相关文档 - 整理项目结构，提高代码可维护性 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <[email protected]>
446 lines
17 KiB
Python
446 lines
17 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
|
|
from typing import Dict, List, Optional, Tuple, Any
|
|
import logging
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class FixedTradingSimulator:
|
|
"""修复后的交易模拟器"""
|
|
|
|
def __init__(self, initial_cash: float = 100000):
|
|
self.initial_cash = initial_cash
|
|
self.portfolio = {
|
|
'cash': initial_cash,
|
|
'positions': {},
|
|
'trades': [],
|
|
'equity_curve': []
|
|
}
|
|
|
|
def simulate_trading(self, all_data: Dict[str, pd.DataFrame],
|
|
all_signals: Dict[str, List[Dict]]) -> Dict:
|
|
"""执行交易模拟"""
|
|
logger.info("🔄 开始执行修复后的交易模拟...")
|
|
|
|
# 重置组合
|
|
self.portfolio = {
|
|
'cash': self.initial_cash,
|
|
'positions': {},
|
|
'trades': [],
|
|
'equity_curve': []
|
|
}
|
|
|
|
# 收集所有交易日期
|
|
all_dates = set()
|
|
date_to_data = {}
|
|
|
|
for stock_code, data in all_data.items():
|
|
if data is None or data.empty:
|
|
logger.warning(f"⚠️ {stock_code}: 数据为空，跳过")
|
|
continue
|
|
|
|
if 'trade_date' in data.columns:
|
|
dates = data['trade_date'].tolist()
|
|
else:
|
|
dates = data.index.tolist()
|
|
|
|
for i, date in enumerate(dates):
|
|
date_key = str(date)
|
|
all_dates.add(date_key)
|
|
if date_key not in date_to_data:
|
|
date_to_data[date_key] = {}
|
|
try:
|
|
date_to_data[date_key][stock_code] = {
|
|
'close': data.iloc[i]['close'],
|
|
'volume': data.iloc[i].get('volume', data.iloc[i].get('vol', 0))
|
|
}
|
|
except (IndexError, KeyError) as e:
|
|
logger.warning(f"⚠️ {stock_code} 第{i}行数据异常: {e}")
|
|
continue
|
|
|
|
if not all_dates:
|
|
logger.warning("⚠️ 没有找到有效的交易数据")
|
|
return {
|
|
'trades': [],
|
|
'total_return': 0.0,
|
|
'sharpe_ratio': 0.0,
|
|
'max_drawdown': 0.0,
|
|
'equity_curve': []
|
|
}
|
|
|
|
# 按日期排序
|
|
sorted_dates = sorted(all_dates)
|
|
logger.info(f"📅 交易日期范围: {sorted_dates[0]} 到 {sorted_dates[-1]}")
|
|
logger.info(f"📊 总交易日数: {len(sorted_dates)}")
|
|
|
|
# 整理信号
|
|
signals_by_date = {}
|
|
total_signals = 0
|
|
for stock_code, signals in all_signals.items():
|
|
for signal in signals:
|
|
signal_date = str(signal['date'])
|
|
if signal_date not in signals_by_date:
|
|
signals_by_date[signal_date] = []
|
|
|
|
signal_copy = signal.copy()
|
|
signal_copy['stock_code'] = stock_code
|
|
signals_by_date[signal_date].append(signal_copy)
|
|
total_signals += 1
|
|
|
|
logger.info(f"🎯 总信号数量: {total_signals}")
|
|
logger.info(f"📊 信号分布在 {len(signals_by_date)} 个交易日")
|
|
|
|
# 逐日执行交易
|
|
executed_trades = 0
|
|
for date in sorted_dates:
|
|
# 更新组合价值
|
|
self._update_portfolio_value(date, date_to_data)
|
|
|
|
# 处理当日信号
|
|
if date in signals_by_date:
|
|
for signal in signals_by_date[date]:
|
|
if self._execute_signal(signal, date, date_to_data):
|
|
executed_trades += 1
|
|
|
|
# 最终价值更新
|
|
if sorted_dates:
|
|
self._update_portfolio_value(sorted_dates[-1], date_to_data)
|
|
|
|
logger.info(f"✅ 交易模拟完成:")
|
|
logger.info(f" - 执行交易: {executed_trades}")
|
|
logger.info(f" - 总交易记录: {len(self.portfolio['trades'])}")
|
|
logger.info(f" - 持仓数量: {len(self.portfolio['positions'])}")
|
|
logger.info(f" - 现金余额: {self.portfolio['cash']:.2f}")
|
|
|
|
final_value = self.portfolio['equity_curve'][-1]['total_value'] if self.portfolio['equity_curve'] else self.initial_cash
|
|
logger.info(f" - 最终价值: {final_value:.2f}")
|
|
logger.info(f" - 总收益: {final_value - self.initial_cash:.2f} ({(final_value/self.initial_cash-1)*100:.2f}%)")
|
|
|
|
return self.portfolio
|
|
|
|
def _execute_signal(self, signal: Dict, date: str, date_to_data: Dict) -> bool:
|
|
"""执行单个信号"""
|
|
try:
|
|
stock_code = signal['stock_code']
|
|
signal_type = signal['signal']
|
|
|
|
# 获取当日价格
|
|
if date not in date_to_data or stock_code not in date_to_data[date]:
|
|
return False
|
|
|
|
price = date_to_data[date][stock_code]['close']
|
|
|
|
if signal_type == 1: # 买入信号
|
|
return self._execute_buy(stock_code, price, date, signal)
|
|
elif signal_type == -1: # 卖出信号
|
|
return self._execute_sell(stock_code, price, date, signal)
|
|
|
|
return False
|
|
|
|
except Exception as e:
|
|
logger.error(f"执行信号失败: {e}")
|
|
return False
|
|
|
|
def _execute_buy(self, stock_code: str, price: float, date: str, signal: Dict) -> bool:
|
|
"""执行买入"""
|
|
try:
|
|
# 计算可买入金额（20%仓位限制）
|
|
max_investment = self.portfolio['cash'] * 0.2
|
|
if max_investment < price:
|
|
return False # 资金不足
|
|
|
|
shares = int(max_investment / price)
|
|
if shares == 0:
|
|
return False
|
|
|
|
cost = shares * price
|
|
|
|
# 检查是否有足够现金
|
|
if self.portfolio['cash'] < cost:
|
|
return False
|
|
|
|
# 执行买入
|
|
self.portfolio['cash'] -= cost
|
|
|
|
if stock_code in self.portfolio['positions']:
|
|
# 加仓
|
|
pos = self.portfolio['positions'][stock_code]
|
|
total_shares = pos['shares'] + shares
|
|
total_cost = pos['shares'] * pos['avg_price'] + cost
|
|
pos['shares'] = total_shares
|
|
pos['avg_price'] = total_cost / total_shares
|
|
else:
|
|
# 新建仓位
|
|
self.portfolio['positions'][stock_code] = {
|
|
'shares': shares,
|
|
'avg_price': price,
|
|
'entry_date': date
|
|
}
|
|
|
|
# 记录交易
|
|
self.portfolio['trades'].append({
|
|
'date': date,
|
|
'stock_code': stock_code,
|
|
'action': 'BUY',
|
|
'shares': shares,
|
|
'price': price,
|
|
'amount': cost
|
|
})
|
|
|
|
logger.debug(f"买入: {stock_code} {shares}股 @ {price:.2f} = {cost:.2f}")
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.error(f"买入执行失败: {e}")
|
|
return False
|
|
|
|
def _execute_sell(self, stock_code: str, price: float, date: str, signal: Dict) -> bool:
|
|
"""执行卖出"""
|
|
try:
|
|
if stock_code not in self.portfolio['positions']:
|
|
return False # 没有持仓
|
|
|
|
position = self.portfolio['positions'][stock_code]
|
|
shares = position['shares']
|
|
avg_price = position['avg_price']
|
|
|
|
# 执行卖出
|
|
revenue = shares * price
|
|
self.portfolio['cash'] += revenue
|
|
|
|
# 计算盈亏
|
|
cost = shares * avg_price
|
|
profit = revenue - cost
|
|
profit_pct = profit / cost if cost > 0 else 0
|
|
|
|
# 记录交易
|
|
self.portfolio['trades'].append({
|
|
'date': date,
|
|
'stock_code': stock_code,
|
|
'action': 'SELL',
|
|
'shares': shares,
|
|
'price': price,
|
|
'amount': revenue,
|
|
'profit': profit,
|
|
'profit_pct': profit_pct
|
|
})
|
|
|
|
# 清除持仓
|
|
del self.portfolio['positions'][stock_code]
|
|
|
|
logger.debug(f"卖出: {stock_code} {shares}股 @ {price:.2f} = {revenue:.2f}, 盈亏: {profit:.2f}")
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.error(f"卖出执行失败: {e}")
|
|
return False
|
|
|
|
def _update_portfolio_value(self, date: str, date_to_data: Dict):
|
|
"""更新组合价值"""
|
|
try:
|
|
total_value = self.portfolio['cash']
|
|
positions_value = 0
|
|
|
|
for stock_code, position in self.portfolio['positions'].items():
|
|
if date in date_to_data and stock_code in date_to_data[date]:
|
|
current_price = date_to_data[date][stock_code]['close']
|
|
position_value = position['shares'] * current_price
|
|
positions_value += position_value
|
|
total_value += position_value
|
|
|
|
self.portfolio['equity_curve'].append({
|
|
'date': date,
|
|
'total_value': total_value,
|
|
'cash': self.portfolio['cash'],
|
|
'positions_value': positions_value
|
|
})
|
|
|
|
except Exception as e:
|
|
logger.error(f"更新组合价值失败: {e}")
|
|
|
|
class FixedMetricsCalculator:
|
|
"""修复后的指标计算器"""
|
|
|
|
@staticmethod
|
|
def calculate_metrics(portfolio: Dict) -> Dict:
|
|
"""计算修复后的性能指标"""
|
|
logger.info("📊 开始计算性能指标...")
|
|
|
|
try:
|
|
equity_curve = pd.DataFrame(portfolio['equity_curve'])
|
|
trades = portfolio['trades']
|
|
|
|
if equity_curve.empty:
|
|
return FixedMetricsCalculator._empty_metrics()
|
|
|
|
# 基础收益指标
|
|
initial_value = equity_curve['total_value'].iloc[0]
|
|
final_value = equity_curve['total_value'].iloc[-1]
|
|
total_return = (final_value - initial_value) / initial_value
|
|
|
|
# 计算日收益率
|
|
equity_curve['daily_return'] = equity_curve['total_value'].pct_change().fillna(0)
|
|
daily_returns = equity_curve['daily_return'].dropna()
|
|
|
|
# 年化收益率
|
|
trading_days = len(equity_curve)
|
|
if trading_days > 0:
|
|
annualized_return = (1 + total_return) ** (252 / trading_days) - 1
|
|
else:
|
|
annualized_return = 0
|
|
|
|
# 风险指标
|
|
volatility = daily_returns.std() * np.sqrt(252) if len(daily_returns) > 1 else 0
|
|
|
|
# 夏普比率（修复版）
|
|
if len(daily_returns) > 1 and volatility > 1e-8:
|
|
risk_free_rate = 0.03
|
|
excess_return = annualized_return - risk_free_rate
|
|
sharpe_ratio = excess_return / volatility
|
|
else:
|
|
sharpe_ratio = 0
|
|
|
|
# 最大回撤
|
|
cummax = equity_curve['total_value'].cummax()
|
|
drawdown = (equity_curve['total_value'] - cummax) / cummax
|
|
max_drawdown = abs(drawdown.min()) if len(drawdown) > 0 else 0
|
|
|
|
# 卡尔马比率
|
|
calmar_ratio = annualized_return / max_drawdown if max_drawdown > 0 else 0
|
|
|
|
# 交易指标
|
|
total_trades = len(trades)
|
|
if total_trades > 0:
|
|
profitable_trades = [t for t in trades if t.get('profit', 0) > 0]
|
|
win_rate = len(profitable_trades) / len([t for t in trades if 'profit' in t])
|
|
|
|
profits = [t.get('profit', 0) for t in trades if 'profit' in t]
|
|
if profits:
|
|
avg_profit = np.mean([p for p in profits if p > 0]) if any(p > 0 for p in profits) else 0
|
|
avg_loss = abs(np.mean([p for p in profits if p < 0])) if any(p < 0 for p in profits) else 1
|
|
profit_loss_ratio = avg_profit / avg_loss if avg_loss > 0 else 0
|
|
else:
|
|
profit_loss_ratio = 0
|
|
else:
|
|
win_rate = 0
|
|
profit_loss_ratio = 0
|
|
|
|
result = {
|
|
'total_return': total_return,
|
|
'annualized_return': annualized_return,
|
|
'volatility': volatility,
|
|
'sharpe_ratio': sharpe_ratio,
|
|
'max_drawdown': max_drawdown,
|
|
'calmar_ratio': calmar_ratio,
|
|
'total_trades': total_trades,
|
|
'win_rate': win_rate,
|
|
'profit_loss_ratio': profit_loss_ratio
|
|
}
|
|
|
|
logger.info("✅ 指标计算完成:")
|
|
logger.info(f" - 总收益率: {total_return:.4f} ({total_return*100:.2f}%)")
|
|
logger.info(f" - 年化收益: {annualized_return:.4f} ({annualized_return*100:.2f}%)")
|
|
logger.info(f" - 夏普比率: {sharpe_ratio:.4f}")
|
|
logger.info(f" - 最大回撤: {max_drawdown:.4f} ({max_drawdown*100:.2f}%)")
|
|
logger.info(f" - 胜率: {win_rate:.4f} ({win_rate*100:.1f}%)")
|
|
|
|
return result
|
|
|
|
except Exception as e:
|
|
logger.error(f"指标计算失败: {e}")
|
|
return FixedMetricsCalculator._empty_metrics()
|
|
|
|
@staticmethod
|
|
def _empty_metrics():
|
|
"""空指标"""
|
|
return {
|
|
'total_return': 0,
|
|
'annualized_return': 0,
|
|
'volatility': 0,
|
|
'sharpe_ratio': 0,
|
|
'max_drawdown': 0,
|
|
'calmar_ratio': 0,
|
|
'total_trades': 0,
|
|
'win_rate': 0,
|
|
'profit_loss_ratio': 0
|
|
}
|
|
|
|
def test_fixed_system():
|
|
"""测试修复后的系统"""
|
|
print("🧪 测试修复后的策略验证系统...")
|
|
|
|
# 导入需要的模块
|
|
from strategy_validation.advanced_validator import AdvancedStrategyValidator, StrategyConfig
|
|
from api.strategy_validation_api import triple_ma_strategy
|
|
|
|
# 配置
|
|
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')
|
|
|
|
# 生成信号
|
|
all_signals = {}
|
|
for stock_code, data in all_data.items():
|
|
signals_df = triple_ma_strategy(data, **config.parameters)
|
|
|
|
# 转换信号格式
|
|
signals_list = []
|
|
for idx, row in signals_df.iterrows():
|
|
if row['signal'] != 0:
|
|
if 'trade_date' in data.columns:
|
|
date_str = str(data.iloc[idx]['trade_date'])
|
|
else:
|
|
date_str = str(idx)
|
|
|
|
signals_list.append({
|
|
'date': date_str,
|
|
'signal': int(row['signal']),
|
|
'strength': float(row['signal_strength']),
|
|
'price': float(data.loc[idx, 'close'])
|
|
})
|
|
|
|
all_signals[stock_code] = signals_list
|
|
print(f"✅ {stock_code}: 生成 {len(signals_list)} 个有效信号")
|
|
|
|
# 使用修复后的交易模拟器
|
|
simulator = FixedTradingSimulator()
|
|
portfolio = simulator.simulate_trading(all_data, all_signals)
|
|
|
|
# 使用修复后的指标计算器
|
|
metrics = FixedMetricsCalculator.calculate_metrics(portfolio)
|
|
|
|
# 检查结果
|
|
if metrics['total_trades'] > 0:
|
|
print("🎉 修复成功! 系统现在能够正确执行交易和计算指标")
|
|
return True
|
|
else:
|
|
print("⚠️ 仍有问题，需要进一步分析")
|
|
return False
|
|
|
|
if __name__ == "__main__":
|
|
logging.basicConfig(level=logging.INFO)
|
|
success = test_fixed_system()
|
|
if success:
|
|
print("\n✅ 策略验证系统修复完成，可以应用到主系统中")
|
|
else:
|
|
print("\n❌ 修复未完成，需要继续调试") |