SWLC MCP Server

""" 回测引擎模块 提供历史数据回测功能，评估预测算法的准确性 """ import logging from typing import List, Dict, Any, Optional from dataclasses import dataclass from datetime import datetime from .predictor import PredictionManager logger = logging.getLogger(__name__) @dataclass class BacktestResult: """回测结果""" period: str actual_numbers: List[str] actual_special_numbers: Optional[List[str]] = None predictions: List[Dict[str, Any]] = None accuracy: float = 0.0 method: str = "" @dataclass class BacktestSummary: """回测摘要""" total_periods: int average_accuracy: float best_strategy: str strategy_performance: Dict[str, float] chart_data: Dict[str, Any] timestamp: str class BacktestEngine: """回测引擎""" def __init__(self): self.prediction_manager = PredictionManager() self.methods = ['rule'] async def run_backtest(self, lottery_type: str, historical_data: List[Dict], window_size: int = 100, step: int = 50) -> BacktestSummary: """运行回测""" try: if len(historical_data) < window_size: raise ValueError(f"历史数据不足，需要至少{window_size}期数据") results = [] chart_data = { 'periods': [], 'accuracy': [], 'precision': [] } # 滑动窗口回测 for start_idx in range(0, len(historical_data) - window_size, step): end_idx = start_idx + window_size test_period = start_idx + window_size if test_period >= len(historical_data): break # 训练数据（历史数据） train_data = historical_data[start_idx:end_idx] # 测试数据（目标期） test_data = historical_data[test_period] # 为每个方法生成预测 method_results = {} for method in self.methods: try: predictions = await self.prediction_manager.predict( lottery_type, train_data, method=method, count=5 ) # 计算预测准确性 accuracy = self._calculate_prediction_accuracy( predictions, test_data, lottery_type ) method_results[method] = { 'predictions': predictions, 'accuracy': accuracy } except Exception as e: logger.error(f"方法{method}预测失败: {e}") method_results[method] = { 'predictions': [], 'accuracy': 0.0 } # 记录结果 period_result = BacktestResult( period=test_data.get('period', f'Period_{test_period}'), actual_numbers=test_data.get('numbers', []), actual_special_numbers=test_data.get('special_numbers'), predictions=method_results, accuracy=max([r['accuracy'] for r in method_results.values()]), method=max(method_results.items(), key=lambda x: x[1]['accuracy'])[0] ) results.append(period_result) # 更新图表数据 chart_data['periods'].append(period_result.period) chart_data['accuracy'].append(period_result.accuracy) chart_data['precision'].append(self._calculate_precision(method_results)) # 生成摘要 summary = self._generate_summary(results, chart_data) return summary except Exception as e: logger.error(f"回测失败: {e}") raise def _calculate_prediction_accuracy(self, predictions: List, actual_data: Dict, lottery_type: str) -> float: """计算预测准确性""" if not predictions: return 0.0 actual_numbers = actual_data.get('numbers', []) actual_special_numbers = actual_data.get('special_numbers', []) best_accuracy = 0.0 for prediction in predictions: pred_numbers = prediction.numbers pred_special_numbers = prediction.special_numbers or [] # 计算红球命中数 red_hits = len(set(pred_numbers) & set(actual_numbers)) # 计算蓝球命中数 blue_hits = 0 if pred_special_numbers and actual_special_numbers: blue_hits = len(set(pred_special_numbers) & set(actual_special_numbers)) # 根据彩票类型计算准确性（支持中英文类型） if lottery_type in ('ssq', '双色球'): accuracy = self._calculate_ssq_accuracy(red_hits, blue_hits) elif lottery_type in ('3d', '福彩3D'): accuracy = self._calculate_3d_accuracy(pred_numbers, actual_numbers) elif lottery_type in ('qlc', '七乐彩'): accuracy = self._calculate_qlc_accuracy(red_hits, blue_hits) elif lottery_type in ('kl8', '快乐8'): accuracy = self._calculate_kl8_accuracy(pred_numbers, actual_numbers) else: accuracy = red_hits / len(actual_numbers) if actual_numbers else 0.0 best_accuracy = max(best_accuracy, accuracy) return round(best_accuracy, 4) def _calculate_ssq_accuracy(self, red_hits: int, blue_hits: int) -> float: """计算双色球准确性""" red_accuracy = red_hits / 6.0 blue_accuracy = blue_hits / 1.0 return red_accuracy * 0.8 + blue_accuracy * 0.2 def _calculate_3d_accuracy(self, pred_numbers: List[str], actual_numbers: List[str]) -> float: """计算福彩3D准确性""" if len(pred_numbers) != 3 or len(actual_numbers) != 3: return 0.0 hits = 0 for i in range(3): if pred_numbers[i] == actual_numbers[i]: hits += 1 return hits / 3.0 def _calculate_qlc_accuracy(self, red_hits: int, blue_hits: int) -> float: """计算七乐彩准确性""" red_accuracy = red_hits / 7.0 blue_accuracy = blue_hits / 1.0 return red_accuracy * 0.8 + blue_accuracy * 0.2 def _calculate_kl8_accuracy(self, pred_numbers: List[str], actual_numbers: List[str]) -> float: """计算快乐8准确性""" hits = len(set(pred_numbers) & set(actual_numbers)) return hits / 20.0 def _calculate_precision(self, method_results: Dict[str, Dict]) -> float: """计算精确度""" if not method_results: return 0.0 accuracies = [result['accuracy'] for result in method_results.values()] return sum(accuracies) / len(accuracies) def _generate_summary(self, results: List[BacktestResult], chart_data: Dict[str, Any]) -> BacktestSummary: """生成回测摘要""" if not results: return BacktestSummary( total_periods=0, average_accuracy=0.0, best_strategy="", strategy_performance={}, chart_data=chart_data, timestamp=datetime.now().isoformat() ) total_periods = len(results) average_accuracy = sum(r.accuracy for r in results) / total_periods strategy_performance = {} for method in self.methods: method_results = [r for r in results if r.method == method] if method_results: strategy_performance[method] = sum(r.accuracy for r in method_results) / len(method_results) else: strategy_performance[method] = 0.0 best_strategy = max(strategy_performance.items(), key=lambda x: x[1])[0] return BacktestSummary( total_periods=total_periods, average_accuracy=round(average_accuracy, 4), best_strategy=best_strategy, strategy_performance=strategy_performance, chart_data=chart_data, timestamp=datetime.now().isoformat() )