# -*- coding: utf-8 -*- """ 蒸鱼融合评分回测验证引擎 v1.0 ================================ 基于FT-Transformer四维奖励机制的融合评分验证系统 包含: Walk-Forward滚动验证 | 四维奖励评分 | 反过拟合检测 | 参数学习优化 """ import numpy as np import pandas as pd import json, os, time, math from datetime import datetime, timedelta from collections import defaultdict # ============================================================ # 一、四维奖励评分器（沿用FT奖励机制） # ============================================================ class FusionRewardScorer: """ 四维奖励评分器 - 评估融合评分对未来走势的预测质量 维度: 趋势方向(30%) | 评分排序(25%) | 波动贴合(25%) | 信号稳定性(20%) """ def __init__(self): self.weights = { 'trend_direction': 0.30, 'score_ranking': 0.25, 'volatility_fit': 0.25, 'signal_stability': 0.20, } def score(self, fusion_scores, future_returns, future_volatilities=None): n = len(fusion_scores) if n < 5: return {'total_reward': 0, 'dimensions': {}, 'samples': n} scores = np.array(fusion_scores) returns = np.array(future_returns) vols = np.array(future_volatilities) if future_volatilities is not None else None # 1. 趋势方向得分 (30%) bullish = scores > 60 bearish = scores < 40 neutral = ~bullish & ~bearish direction_correct = np.zeros(n) direction_correct[bullish & (returns > 0)] = 1 direction_correct[bearish & (returns < 0)] = 1 direction_correct[neutral] = 0.5 trend_score = np.mean(direction_correct) * 100 confidence = np.abs(scores - 50) / 50 weighted_trend = np.sum(direction_correct * confidence) / (np.sum(confidence) + 1e-10) * 100 trend_dim = trend_score * 0.4 + weighted_trend * 0.6 # 2. 评分排序得分 (25%) if n >= 10: sorted_idx = np.argsort(scores) half = n // 2 top_returns = np.mean(returns[sorted_idx[-half:]]) bottom_returns = np.mean(returns[sorted_idx[:half]]) ranking_score = max(0, min(100, 50 + (top_returns - bottom_returns) * 500)) else: if np.std(scores) > 1 and np.std(returns) > 0.001: corr = np.corrcoef(scores, returns)[0, 1] ranking_score = max(0, min(100, 50 + corr * 50)) else: ranking_score = 50 # 3. 波动贴合得分 (25%) if vols is not None and len(vols) == n: high_conf = np.abs(scores - 50) / 50 > 0.3 if np.sum(high_conf) > 3: high_conf_vol = np.mean(vols[high_conf]) low_conf_vol = np.mean(vols[~high_conf]) if np.sum(~high_conf) > 0 else high_conf_vol vol_ratio = low_conf_vol / (high_conf_vol + 1e-10) vol_score = max(0, min(100, 50 + (vol_ratio - 1) * 30)) else: vol_score = 50 else: abs_returns = np.abs(returns) high_conf = np.abs(scores - 50) / 50 > 0.3 if np.sum(high_conf) > 3: high_conf_vol = np.mean(abs_returns[high_conf]) low_conf_vol = np.mean(abs_returns[~high_conf]) if np.sum(~high_conf) > 0 else high_conf_vol vol_ratio = low_conf_vol / (high_conf_vol + 1e-10) vol_score = max(0, min(100, 50 + (vol_ratio - 1) * 30)) else: vol_score = 50 # 4. 信号稳定性得分 (20%) if n >= 10: score_changes = np.abs(np.diff(scores)) avg_change = np.mean(score_changes) stability_score = max(0, min(100, 100 - avg_change * 4)) else: stability_score = 50 total = (trend_dim * self.weights['trend_direction'] + ranking_score * self.weights['score_ranking'] + vol_score * self.weights['volatility_fit'] + stability_score * self.weights['signal_stability']) return { 'total_reward': round(float(total), 2), 'dimensions': { 'trend_direction': round(float(trend_dim), 2), 'score_ranking': round(float(ranking_score), 2), 'volatility_fit': round(float(vol_score), 2), 'signal_stability': round(float(stability_score), 2), }, 'details': { 'trend_accuracy': round(float(trend_score), 2), 'weighted_trend_accuracy': round(float(weighted_trend), 2), 'avg_score_change': round(float(avg_change), 2) if n >= 10 else None, 'samples': n, }, 'weights': self.weights, } # ============================================================ # 二、反过拟合检测器 # ============================================================ class FusionAntiOverfit: """融合评分反过拟合检测 - 4项检验""" def __init__(self): self.results = {} def run_all(self, fusion_scores, future_returns, dates=None): scores = np.array(fusion_scores) returns = np.array(future_returns) n = len(scores) self.results = { 'ic_stability': self._test_ic_stability(scores, returns, n), 'subsample_stress': self._test_subsample_stress(scores, returns, n), 'placebo_test': self._test_placebo(scores, returns, n), 'parameter_sensitivity': self._test_parameter_sensitivity(scores, returns, n), } scores_list = [] for k, v in self.results.items(): if v.get('passed') is not None: scores_list.append(100 if v['passed'] else 0) elif v.get('score') is not None: scores_list.append(v['score']) overall = np.mean(scores_list) if scores_list else 0 self.results['overall'] = { 'score': round(float(overall), 2), 'passed': overall >= 60, 'judgment': '通过' if overall >= 60 else ('警告' if overall >= 40 else '失败'), } return self.results def _test_ic_stability(self, scores, returns, n, min_periods=3): if n < 60: return {'passed': None, 'score': 50, 'reason': '样本不足'} chunk_size = max(20, n // min_periods) periods = [] for i in range(0, n, chunk_size): end = min(i + chunk_size, n) if end - i < 10: continue s_seg = scores[i:end] r_seg = returns[i:end] if np.std(s_seg) > 0.1 and np.std(r_seg) > 0.0001: corr = np.corrcoef(s_seg, r_seg)[0, 1] else: corr = 0 periods.append(corr) if len(periods) < 2: return {'passed': None, 'score': 50, 'reason': '分段不足'} positive_rate = np.mean([1 if p > 0 else 0 for p in periods]) mean_ic = np.mean(np.abs(periods)) has_reversal = any(periods[i] * periods[i-1] < -0.1 for i in range(1, len(periods))) score = min(positive_rate / 0.6, 1) * 40 + min(mean_ic / 0.05, 1) * 40 score += 20 if not has_reversal else 0 passed = positive_rate >= 0.6 and mean_ic >= 0.03 and not has_reversal return {'passed': passed, 'score': round(float(score), 2), 'positive_rate': round(float(positive_rate), 4), 'mean_abs_ic': round(float(mean_ic), 4), 'has_reversal': has_reversal, 'period_ics': [round(float(p), 4) for p in periods], 'reason': f'正IC率={positive_rate:.0%}, 平均|IC|={mean_ic:.4f}' + (', 存在反转' if has_reversal else '')} def _test_subsample_stress(self, scores, returns, n): if n < 100: return {'passed': None, 'score': 50, 'reason': '样本不足'} abs_returns = np.abs(returns) vol_median = np.median(abs_returns) high_vol = abs_returns > vol_median low_vol = ~high_vol up_market = returns > 0 down_market = returns < 0 def _seg_ic(s, r): if len(s) < 10 or np.std(s) < 0.1 or np.std(r) < 0.0001: return 0 return np.corrcoef(s, r)[0, 1] ic_high_vol = _seg_ic(scores[high_vol], returns[high_vol]) ic_low_vol = _seg_ic(scores[low_vol], returns[low_vol]) ic_up = _seg_ic(scores[up_market], returns[up_market]) ic_down = _seg_ic(scores[down_market], returns[down_market]) ics = [ic_high_vol, ic_low_vol, ic_up, ic_down] positive_count = sum(1 for ic in ics if ic > 0) consistency = positive_count / len(ics) ic_std = np.std(ics) score = consistency * 50 + max(0, 50 - ic_std * 200) passed = consistency >= 0.75 and ic_std < 0.3 return {'passed': passed, 'score': round(float(score), 2), 'consistency': round(float(consistency), 4), 'ic_std': round(float(ic_std), 4), 'ic_high_vol': round(float(ic_high_vol), 4), 'ic_low_vol': round(float(ic_low_vol), 4), 'ic_up': round(float(ic_up), 4), 'ic_down': round(float(ic_down), 4)} def _test_placebo(self, scores, returns, n, n_shuffles=50): if n < 30: return {'passed': None, 'score': 50, 'reason': '样本不足'} real_corr = np.corrcoef(scores, returns)[0, 1] if np.std(scores) > 0.1 and np.std(returns) > 0.0001 else 0 shuffled_corrs = [] for _ in range(n_shuffles): shuffled = np.random.permutation(returns) if np.std(scores) > 0.1 and np.std(shuffled) > 0.0001: shuffled_corrs.append(np.corrcoef(scores, shuffled)[0, 1]) else: shuffled_corrs.append(0) shuffled_corrs = np.array(shuffled_corrs) rank = np.sum(shuffled_corrs >= real_corr) / len(shuffled_corrs) score = max(0, 100 - rank * 100) passed = rank < 0.05 return {'passed': passed, 'score': round(float(score), 2), 'real_corr': round(float(real_corr), 4), 'shuffle_mean': round(float(np.mean(shuffled_corrs)), 4), 'shuffle_std': round(float(np.std(shuffled_corrs)), 4), 'p_value': round(float(rank), 4), 'n_shuffles': n_shuffles} def _test_parameter_sensitivity(self, scores, returns, n): if n < 30: return {'passed': None, 'score': 50, 'reason': '样本不足'} base_signal = (scores > 60).astype(float) - (scores < 40).astype(float) perturbations = [-2, -1, 1, 2] signal_changes = [] for p in perturbations: perturbed = scores + p p_signal = (perturbed > 60).astype(float) - (perturbed < 40).astype(float) change_rate = np.mean(p_signal != base_signal) signal_changes.append(change_rate) avg_change = np.mean(signal_changes) score = max(0, 100 - avg_change * 200) passed = avg_change < 0.15 return {'passed': passed, 'score': round(float(score), 2), 'avg_signal_change': round(float(avg_change), 4), 'perturbation_results': {f'delta_{p}': round(float(c), 4) for p, c in zip(perturbations, signal_changes)}} # ============================================================ # 三、完整回测验证管道 # ============================================================ class FusionBacktestValidator: """融合评分回测验证器 - 一站式验证管道""" def __init__(self): self.reward_scorer = FusionRewardScorer() self.anti_overfit = FusionAntiOverfit() self.results = {} def validate(self, fusion_scores, future_returns, future_vols=None, dates=None): scores = np.array(fusion_scores) returns = np.array(future_returns) vols = np.array(future_vols) if future_vols is not None else None n = len(scores) if n < 10: return {'status': 'error', 'message': f'样本不足: {n}'} # 1. 四维奖励评分 reward_result = self.reward_scorer.score(scores, returns, vols) # 2. 反过拟合检测 anti_overfit_result = self.anti_overfit.run_all(scores, returns, dates) # 3. 综合评估 total_reward = reward_result['total_reward'] anti_score = anti_overfit_result['overall']['score'] composite_score = total_reward * 0.6 + anti_score * 0.4 # 4. 信号质量分析 bullish_signals = np.sum(scores > 60) bearish_signals = np.sum(scores < 40) bullish_win_rate = np.mean(returns[scores > 60] > 0) * 100 if bullish_signals > 0 else 0 bearish_win_rate = np.mean(returns[scores < 40] < 0) * 100 if bearish_signals > 0 else 0 # 评分分档收益 bins = [0, 32, 40, 46, 54, 60, 68, 100] labels = ['强烈卖出', '卖出', '偏空观望', '中性', '偏多观望', '买入', '强烈买入'] bin_returns = {} for i in range(len(bins) - 1): mask = (scores >= bins[i]) & (scores < bins[i + 1]) if np.sum(mask) > 0: bin_returns[labels[i]] = { 'count': int(np.sum(mask)), 'mean_return': round(float(np.mean(returns[mask]) * 100), 2), 'win_rate': round(float(np.mean(returns[mask] > 0) * 100), 1), } self.results = { 'status': 'success', 'samples': n, 'composite_score': round(float(composite_score), 2), 'reward': reward_result, 'anti_overfit': anti_overfit_result, 'signal_quality': { 'bullish_signals': int(bullish_signals), 'bearish_signals': int(bearish_signals), 'bullish_win_rate': round(float(bullish_win_rate), 1), 'bearish_win_rate': round(float(bearish_win_rate), 1), 'overall_win_rate': round(float(np.mean(returns > 0) * 100), 1), 'avg_return_bullish': round(float(np.mean(returns[scores > 60]) * 100), 2) if bullish_signals > 0 else 0, 'avg_return_bearish': round(float(np.mean(returns[scores < 40]) * 100), 2) if bearish_signals > 0 else 0, }, 'bin_returns': bin_returns, 'evaluation': self._generate_evaluation(composite_score, reward_result, anti_overfit_result), } return self.results def _generate_evaluation(self, composite_score, reward, anti): level = '优秀' if composite_score >= 75 else ('良好' if composite_score >= 60 else ('一般' if composite_score >= 45 else ('较差' if composite_score >= 30 else '失败'))) issues = [] if reward['total_reward'] < 50: issues.append('四维奖励得分偏低，融合评分预测能力不足') if anti['overall']['score'] < 60: issues.append('反过拟合检测未通过，存在过拟合风险') if anti.get('ic_stability', {}).get('has_reversal', False): issues.append('IC存在反转，评分稳定性不足') if anti.get('placebo_test', {}).get('passed') == False: issues.append('安慰剂检验未通过，评分可能无真实预测能力') suggestions = [] if reward['dimensions'].get('trend_direction', 0) < 50: suggestions.append('提高趋势方向权重或调整评分阈值') if reward['dimensions'].get('signal_stability', 0) < 50: suggestions.append('降低共识增强幅度，提高评分稳定性') if anti.get('parameter_sensitivity', {}).get('passed') == False: suggestions.append('降低参数敏感度，简化融合规则') return {'level': level, 'composite_score': round(float(composite_score), 2), 'issues': issues, 'suggestions': suggestions} # ============================================================ # 四、参数学习优化器 # ============================================================ class FusionParameterLearner: """融合参数学习优化器 - 基于回测奖励分数的贝叶斯风格搜索""" def __init__(self): self.best_params = {} self.search_history = [] def optimize_consensus_boost(self, fusion_func, data_loader, stock_codes, n_trials=20, verbose=True): param_grid = { 'boost_threshold': [2, 3, 4, 5], 'boost_base': [4, 6, 8, 10, 12], 'min_fusion_for_boost': [50, 55, 60, 65], 'temperature': [0.08, 0.10, 0.12, 0.15, 0.20], } best_score = -float('inf') best_params = {} validator = FusionBacktestValidator() for trial in range(n_trials): params = {k: np.random.choice(v) for k, v in param_grid.items()} try: all_scores, all_returns = [], [] for code in stock_codes[:5]: df = data_loader(code) if df is None or len(df) < 60: continue for i in range(30, len(df) - 5): slice_df = df.iloc[:i].copy() future_ret = df['close'].iloc[i+5] / df['close'].iloc[i] - 1 result = fusion_func(slice_df, params) if result and 'fusion_score' in result: all_scores.append(result['fusion_score']) all_returns.append(future_ret) if len(all_scores) < 20: continue v_result = validator.validate(all_scores, all_returns) reward = v_result['reward']['total_reward'] anti_score = v_result['anti_overfit']['overall']['score'] composite = reward * 0.5 + anti_score * 0.5 complexity_penalty = params['boost_base'] * 0.5 + (5 - params['boost_threshold']) * 1 final_score = composite - complexity_penalty * 0.5 self.search_history.append({'trial': trial + 1, 'params': params, 'reward': reward, 'anti_score': anti_score, 'composite': composite, 'final_score': final_score}) if final_score > best_score: best_score = final_score best_params = params.copy() if verbose: print(f" Trial {trial+1}/{n_trials}: reward={reward:.1f} anti={anti_score:.1f} final={final_score:.1f}") except Exception: continue self.best_params = best_params return {'best_params': best_params, 'best_score': round(float(best_score), 2), 'n_trials': len(self.search_history), 'search_history': sorted(self.search_history, key=lambda x: -x['final_score'])[:10]} # ============================================================ # 五、快速入口 # ============================================================ def run_full_validation(fusion_scores, future_returns, future_vols=None, dates=None): validator = FusionBacktestValidator() return validator.validate(fusion_scores, future_returns, future_vols, dates) if __name__ == '__main__': print("=" * 60) print(" 蒸鱼融合评分回测验证引擎 v1.0") print("=" * 60) print("\n 模块清单:") print(" FusionRewardScorer - 四维奖励评分器") print(" FusionAntiOverfit - 反过拟合检测器") print(" FusionBacktestValidator - 完整验证管道") print(" FusionParameterLearner - 参数学习优化器") print("\n [OK] 模块加载成功")