Blame for webui/factor_score.py

# -*- 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,

    },

}