# -*- coding: utf-8 -*- """ 蒸鱼股票分析系统 - 多维因子评分体系 基于QuantGPT iteration.py compute_factor_score适配 6维度评分: IC Mean 15% + IC IR 15% + 稳定性 15% + 反过拟合 15% + 分组回测 15% + Cloud对齐 25% """ import numpy as np import pandas as pd from datetime import datetime import warnings warnings.filterwarnings('ignore') # 评分权重配置 SCORE_WEIGHTS = { 'ic_mean': 0.15, 'ic_ir': 0.15, 'stability': 0.15, 'anti_overfit': 0.15, 'backtest': 0.15, 'cloud_alignment': 0.25, } def compute_factor_score(backtest_result, anti_overfit_result=None, cloud_result=None): """ 计算因子综合评分（0-100） Parameters ---------- backtest_result : dict run_factor_backtest的返回结果 anti_overfit_result : dict, optional 反过拟合检测结果 cloud_result : dict, optional 云端因子库对齐评分 {'ic_mean': float, 'ic_ir': float, 'turnover': float, 'sufficiency': float} Returns ------- dict : 6维度评分 + 综合评分 + 评级 """ if not backtest_result or not backtest_result.get('success'): return {'overall': 0, 'grade': 'N/A', 'dimensions': {}, 'error': '回测结果无效'} summary = backtest_result.get('summary', {}) performance = backtest_result.get('performance', {}) scores = {} # ─── 维度1: IC Mean（15%）─── ic_mean = abs(summary.get('ic_mean', 0)) scores['ic_mean'] = min(ic_mean / 0.05, 1.0) * 100 # ─── 维度2: IC IR（15%）─── ic_ir = abs(summary.get('ic_ir', 0)) scores['ic_ir'] = min(ic_ir / 1.5, 1.0) * 100 # ─── 维度3: 稳定性（15%）─── ic_win_rate = summary.get('ic_win_rate', 0) # Sharpe稳定性：比较第1组和最后1组的Sharpe符号一致性 g1_sharpe = performance.get('group_1', {}).get('sharpe', 0) gn_sharpe = performance.get('group_5', {}).get('sharpe', 0) sharpe_consistent = 1 if g1_sharpe * gn_sharpe <= 0 else 0 # 一正一负为好 stability_score = ic_win_rate * 100 * 0.6 + sharpe_consistent * 40 scores['stability'] = min(stability_score, 100) # ─── 维度4: 反过拟合（15%）─── if anti_overfit_result: ao_score = anti_overfit_result.get('overall_score', 0) scores['anti_overfit'] = min(float(ao_score), 100) else: # 无反过拟合数据时用IC胜率近似 scores['anti_overfit'] = ic_win_rate * 100 * 0.7 # ─── 维度5: 分组回测（15%）─── ls_sharpe = abs(performance.get('long_short', {}).get('sharpe', 0)) mono = abs(performance.get('monotonicity', 0)) spread = abs(performance.get('spread', 0)) bt_score = (min(ls_sharpe / 2.0, 1) * 40 + min(mono, 1) * 40 + min(spread / 0.1, 1) * 20) scores['backtest'] = min(bt_score, 100) # ─── 维度6: Cloud对齐（25%）─── if cloud_result: c_ic = cloud_result.get('ic_mean', 0) c_ir = cloud_result.get('ic_ir', 0) c_turn = cloud_result.get('turnover', 0.5) c_suff = cloud_result.get('sufficiency', 0.5) scores['cloud_alignment'] = ( min(abs(c_ic) / 0.05, 1) * 30 + min(abs(c_ir) / 1.0, 1) * 30 + min(c_turn, 1) * 20 + min(c_suff, 1) * 20 ) else: # 无Cloud数据时，用本地指标的置信度加权 local_confidence = (scores['ic_mean'] + scores['ic_ir']) / 200 scores['cloud_alignment'] = local_confidence * 100 # ─── 综合评分 ─── overall = 0 for key, weight in SCORE_WEIGHTS.items(): overall += scores.get(key, 0) * weight overall = round(overall, 2) # ─── 评级 ─── if overall >= 80: grade = 'S' elif overall >= 70: grade = 'A' elif overall >= 55: grade = 'B' elif overall >= 40: grade = 'C' elif overall >= 25: grade = 'D' else: grade = 'F' return { 'overall': overall, 'grade': grade, 'grade_label': _grade_label(grade), 'weights': SCORE_WEIGHTS, 'dimensions': {k: round(v, 2) for k, v in scores.items()}, 'timestamp': datetime.now().isoformat(), } def _grade_label(grade): labels = { 'S': '卓越因子，强烈推荐', 'A': '优秀因子，推荐使用', 'B': '良好因子，可考虑', 'C': '一般因子，需谨慎', 'D': '较弱因子，不建议', 'F': '无效因子，不推荐', } return labels.get(grade, '未知') def compare_factors(factor_results): """ 比较多个因子的评分 Parameters ---------- factor_results : list of dict [{'name': str, 'score': dict from compute_factor_score}, ...] Returns ------- dict : 排名结果 """ ranked = sorted(factor_results, key=lambda x: x['score']['overall'], reverse=True) comparison = { 'factors': ranked, 'n_factors': len(ranked), 'best': ranked[0]['name'] if ranked else None, 'ranking': [], } for i, f in enumerate(ranked): comparison['ranking'].append({ 'rank': i + 1, 'name': f['name'], 'overall': f['score']['overall'], 'grade': f['score']['grade'], }) return comparison # 常用因子评估预设 FACTOR_EVAL_PRESETS = { 'fast_momentum': { 'name': '快速动量', 'expression': 'ts_rank(ts_delta(close, 5), 20)', 'category': '动量', 'expected_ic': 0.03, }, 'medium_momentum': { 'name': '中期动量', 'expression': 'ts_rank(ts_delta(close, 20), 60)', 'category': '动量', 'expected_ic': 0.04, }, 'mean_reversion': { 'name': '均值回归', 'expression': '-ts_zscore(close, 20)', 'category': '反转', 'expected_ic': 0.02, }, 'volatility': { 'name': '波动率因子', 'expression': 'ts_std(close / delay(close, 1) - 1, 20)', 'category': '波动率', 'expected_ic': 0.01, }, 'volume_momentum': { 'name': '量价动量', 'expression': 'ts_corr(close, volume, 20)', 'category': '量价', 'expected_ic': 0.02, }, 'boll_deviation': { 'name': '布林偏离', 'expression': '(close - boll_mid(close, 20)) / ts_std(close, 20)', 'category': '技术', 'expected_ic': 0.03, }, 'rsi_extreme': { 'name': 'RSI极值', 'expression': 'rsi(close, 14) - 50', 'category': '技术', 'expected_ic': 0.02, }, }