小果量化因子分析系统助手

小果量化因子分析系统完整教程

目录

1. 系统概述
2. 核心功能模块
3. 环境依赖与安装
4. 数据准备与格式要求
5. 核心类与初始化
6. 数据清洗函数详解
7. IC分析模块
8. 分层收益分析模块
9. 多空组合分析模块
10. 换手率分析模块
11. 因子稳定性分析模块
12. 统计检验模块
13. 可视化分析模块
14. 综合评分模块（28项指标）
15. 一键运行完整分析
16. 完整使用示例
17. 输出解读与评级标准
18. 优化建议速查表
19. 常见问题解答
20. 完整源代码

---

一、系统概述

1.1 什么是小果量化因子分析系统？

小果量化因子分析系统是一套基于 Alphalens 框架扩展开发的量化因子评价工具，提供了从数据清洗到综合评分的完整因子分析流程。系统包含 28+ 项因子评估指标，涵盖 IC 分析、分层收益、多空组合、换手率、Fama-MacBeth 回归、因子稳定性检验等多个维度，能够全面评价因子的有效性和稳定性。

1.2 系统特点

• 全面性：覆盖因子评价的各个方面，从 IC 分析到综合评分
• 自动化：一键运行完整分析流程，自动生成图表和报告
• 可扩展性：基于 Alphalens 框架，支持自定义扩展
• 实用性：提供具体的优化建议，指导因子改进方向
• 可视化：丰富的图表展示，直观理解因子表现

1.3 适用场景

• 量化因子研究与开发
• 多因子模型构建
• 因子有效性验证
• 策略回测与优化
• 学术研究与教学

---

二、核心功能模块

---

三、环境依赖与安装

3.1 依赖库清单

# 核心依赖
import pandas as pd          # 数据处理
import numpy as np           # 数值计算
import alphalens as al       # 因子分析框架

# 可视化
import matplotlib.pyplot as plt
import seaborn as sns

# 统计分析
from scipy import stats
from scipy.stats import norm, ttest_1samp, jarque_bera, spearmanr, pearsonr
from statsmodels.tsa.stattools import adfuller
import statsmodels.api as sm

# 量化分析
import quantstats as qs

# 工具
import warnings

3.2 安装命令
bash
pip install pandas numpy matplotlib seaborn scipy statsmodels alphalens quantstats
3.3 全局设置
python
# 字体设置（纯英文字体）
plt.rcParams['font.sans-serif'] = ['DejaVu Sans']
plt.rcParams['axes.unicode_minus'] = False

# 忽略警告
warnings.filterwarnings("ignore", category=FutureWarning)
warnings.filterwarnings("ignore", category=UserWarning)

# 样式设置
plt.style.use('seaborn-v0_8-darkgrid')
sns.set_palette("husl")
plt.rcParams['figure.figsize'] = (12, 6)
四、数据准备与格式要求
4.1 因子数据格式要求
python
import pandas as pd
import numpy as np
import alphalens as al

# 因子数据必须是 Series，MultiIndex 为 (date, asset)
# date: 日期（Timestamp）
# asset: 资产代码（字符串）
factor_series = pd.Series(
    index=pd.MultiIndex.from_tuples([], names=['date', 'asset']),
    dtype=float
)

# 价格数据必须是 DataFrame，index=日期，columns=资产代码
price_df = pd.DataFrame(
    index=pd.DatetimeIndex([]),
    columns=[]
)
4.2 从通达信数据生成因子数据（完整流程）
python
import pandas as pd
import numpy as np
from tqdm import tqdm

# ==================== 1. 读取成分股 ====================
def get_read_tdx_data(path):
    '''
    读取通达信成分股代码
    参数:
        path: 通达信板块文件路径
    返回:
        df: 包含证券代码和名称的 DataFrame
    '''
    try:
        stock_list = []
        with open(r'{}'.format(path)) as p:
            com = p.readlines()
        for stock in com:
            if len(stock) >= 6:
                stock = stock.replace("\n", "")
                stock_list.append(stock)
        df = pd.DataFrame()
        df['证券代码'] = stock_list
        
        def select_stock(x):
            stock = str(x)
            if stock[0] == '0':
                stock = stock[1:] + '.SZ'
            elif stock[0] == '1':
                stock = stock[1:] + '.SH'
            else:
                stock = stock[1:] + '.SZ'
            return stock
        
        df['证券代码'] = df['证券代码'].apply(lambda x: select_stock(x))
        df['名称'] = df['证券代码']
    except Exception as e:
        try:
            print(e, '通达信路径有问题可能不存在', path)
            df = pd.read_excel(r'{}'.format(path))
        except Exception as e:
            try:
                df = pd.read_csv(r'{}'.format(path))
            except Exception as e:
                print(e)
                df = pd.DataFrame()
    return df

# ==================== 2. 读取交易日历 ====================
def get_trading_dates(start_date='20210101', end_date='20500101', index_stock='000001.SH'):
    '''
    获取交易日列表
    参数:
        start_date: 开始日期，格式 'YYYYMMDD'
        end_date: 结束日期，格式 'YYYYMMDD'
        index_stock: 基准指数代码
    返回:
        date_list: 交易日列表
    '''
    xtdata.subscribe_quote(
        stock_code=index_stock,
        start_time=start_date,
        end_time=end_date,
        count=-1
    )
    index_data = xtdata.get_market_data_ex(
        stock_list=[index_stock],
        start_time=start_date,
        end_time=end_date,
        count=-1
    )
    index_data = index_data[index_stock]
    index_data['date'] = pd.to_datetime(index_data.index)
    return index_data['date'].tolist()

# ==================== 3. 批量获取股票数据并计算因子 ====================
def generate_factor_data(stock_df, date_list, start_date='20210101', end_date='20500101', factor_func=None):
    '''
    批量获取股票数据并计算因子
    参数:
        stock_df: 股票列表 DataFrame
        date_list: 交易日列表
        start_date: 开始日期
        end_date: 结束日期
        factor_func: 因子计算函数，接收 CLOSE 返回因子值
    返回:
        factor_df: 因子宽表
        price_df: 价格宽表
    '''
    stock_list = stock_df['证券代码'].tolist()
    
    # 初始化空的 DataFrame（宽表）
    factor_df = pd.DataFrame(index=date_list, columns=stock_list, dtype=float)
    price_df = pd.DataFrame(index=date_list, columns=stock_list, dtype=float)
    
    # 默认因子函数：MACD
    if factor_func is None:
        def default_factor(CLOSE):
            SHORT, LONG, MID = 12, 26, 9
            # MACD 计算逻辑
            EMA_SHORT = CLOSE.ewm(span=SHORT, adjust=False).mean()
            EMA_LONG = CLOSE.ewm(span=LONG, adjust=False).mean()
            DIF = EMA_SHORT - EMA_LONG
            DEA = DIF.ewm(span=MID, adjust=False).mean()
            MACD = (DIF - DEA) * 2
            return DIF, DEA, MACD
        factor_func = default_factor
    
    # 循环获取数据
    for i in tqdm(range(len(stock_list))):
        stock = stock_list[i]
        xtdata.subscribe_quote(
            stock_code=stock,
            start_time=start_date,
            end_time=end_date,
            count=-1
        )
        data = xtdata.get_market_data_ex(
            stock_list=[stock],
            start_time=start_date,
            end_time=end_date,
            count=-1
        )
        data = data[stock]
        data.index = pd.to_datetime(data.index)
        
        CLOSE = data['close']
        
        # 计算因子
        DIF, DEA, factor_value = factor_func(CLOSE)
        factor_value = pd.Series(factor_value.tolist(), index=data.index, name=stock)
        
        # 填充因子和价格数据
        factor_df[stock] = factor_value.reindex(date_list)
        price_df[stock] = data['close'].reindex(date_list)
    
    return factor_df, price_df

# ==================== 4. 转换为factor_series格式 ====================
def convert_to_factor_series(factor_df):
    '''
    将宽表转换为factor_series格式
    参数:
        factor_df: 因子宽表 (index=日期, columns=资产代码)
    返回:
        factor_series: MultiIndex Series (date, asset)
    '''
    factor_series = factor_df.stack()
    factor_series.index.names = ['date', 'asset']
    return factor_series

# ==================== 5. 完整数据准备流程 ====================
def prepare_factor_data(block_file='QBKZZ.blk', start_date='20210101', end_date='20500101', factor_func=None):
    '''
    完整的数据准备流程
    参数:
        block_file: 通达信板块文件路径
        start_date: 开始日期
        end_date: 结束日期
        factor_func: 自定义因子计算函数
    返回:
        factor_df: 因子宽表
        price_df: 价格宽表
        factor_series: 因子序列
        stock_list: 股票列表
    '''
    # 1. 读取成分股
    stock_df = get_read_tdx_data(block_file)
    stock_list = stock_df['证券代码'].tolist()
    
    # 2. 获取交易日列表
    date_list = get_trading_dates(start_date, end_date)
    
    # 3. 生成因子数据
    factor_df, price_df = generate_factor_data(
        stock_df, date_list, start_date, end_date, factor_func
    )
    
    # 4. 转换为factor_series
    factor_series = convert_to_factor_series(factor_df)
    
    return factor_df, price_df, factor_series, stock_list
4.3 自定义因子计算函数示例
python
# 示例1: RSI 因子
def calculate_rsi(CLOSE, period=14):
    delta = CLOSE.diff()
    gain = (delta.where(delta > 0, 0)).rolling(window=period).mean()
    loss = (-delta.where(delta < 0, 0)).rolling(window=period).mean()
    rs = gain / loss
    rsi = 100 - (100 / (1 + rs))
    return rsi

# 示例2: 动量因子
def calculate_momentum(CLOSE, period=20):
    momentum = CLOSE / CLOSE.shift(period) - 1
    return momentum

# 示例3: 波动率因子
def calculate_volatility(CLOSE, period=20):
    returns = CLOSE.pct_change()
    volatility = returns.rolling(window=period).std()
    return volatility

# 示例4: 自定义因子包装函数
def custom_factor_wrapper(CLOSE):
    # 可以返回多个值，但只取最后一个作为因子
    rsi = calculate_rsi(CLOSE, 14)
    return None, None, rsi  # DIF, DEA, factor_value

# 使用自定义因子
factor_df, price_df, factor_series, stock_list = prepare_factor_data(
    block_file='QBKZZ.blk',
    start_date='20210101',
    end_date='20500101',
    factor_func=custom_factor_wrapper
)
4.4 生成模拟数据（测试用）
python
def generate_synthetic_data(start_date='2024-01-01', end_date='2024-12-31',
                            n_stocks=30, seed=42):
    """
    生成模拟的因子数据和价格数据
    参数:
        start_date: 开始日期
        end_date: 结束日期
        n_stocks: 股票数量
        seed: 随机种子
    返回:
        factor_df: 因子宽表
        price_df: 价格宽表
        factor_series: 因子序列
    """
    np.random.seed(seed)
    dates = pd.date_range(start_date, end_date, freq='B')
    stocks = [f'Stock_{i:03d}' for i in range(n_stocks)]

    factor_df = pd.DataFrame(index=dates, columns=stocks)
    price_df = pd.DataFrame(index=dates, columns=stocks)

    for stock in stocks:
        trend = np.linspace(0, 1, len(dates)) * 0.3
        style = np.random.randn() * 0.5
        factor_df[stock] = np.random.randn(len(dates)) + trend + style
        ret = np.random.randn(len(dates)) * 0.02
        ret += factor_df[stock].shift(1).fillna(0) * 0.01
        price_df[stock] = 100 * (1 + ret).cumprod()

    factor_series = factor_df.stack()
    factor_series.index.names = ['date', 'asset']
    return factor_df, price_df, factor_series
五、核心类与初始化
5.1 xg_alphalens 类完整定义
python
class xg_alphalens:
    """
    小果因子分析框架
    
    作者: 小果
    微信: xg_quant
    
    参数:
    ----------
    factor_data : DataFrame, 可选
        预清洗的因子数据，Alphalens 标准格式
    quantiles : int, 默认 5
        分组数量，用于分层收益分析
    periods : tuple, 默认 (1,5,10)
        未来收益持有期（交易日）
    max_loss : float, 默认 0.95
        数据清洗时允许丢失的最大比例
    """
    
    def __init__(self,
            factor_data='',
            quantiles=5,
            periods=(1,5,10),
            max_loss=0.95):
        
        self.factor_data = factor_data
        self.quantiles = quantiles
        self.periods = periods
        self.max_loss = max_loss
        self.al = al
        self.qs = qs
        
    # ========== 以下为所有方法 ==========
    # 1. 数据读取
    # 2. 数据生成
    # 3. 数据清洗
    # 4. IC分析
    # 5. 分层收益分析
    # 6. 多空组合分析
    # 7. 换手率分析
    # 8. 因子稳定性分析
    # 9. 统计检验
    # 10. 可视化分析
    # 11. 综合评分
    # 12. 完整分析流程
5.2 参数说明
参数	类型	默认值	说明
factor_data	DataFrame	''	Alphalens 格式的因子数据（含 factor, factor_quantile, 1D,5D,10D）
quantiles	int	5	分组数量，用于分层收益分析
periods	tuple	(1,5,10)	未来收益计算周期（交易日）
max_loss	float	0.95	数据清洗时允许丢失的最大比例
六、数据清洗函数详解
6.1 clean_factor_data() 完整实现
python
def clean_factor_data(self, factor_series, price_df, quantiles=5, periods=(1,5,10), max_loss=0.95):
    """
    数据清洗与对齐，返回 Alphalens 格式的 factor_data
    
    参数:
    ----------
    factor_series : pd.Series
        因子数据，MultiIndex 为 (date, asset)
    price_df : pd.DataFrame
        价格数据，index=日期，columns=资产代码
    quantiles : int, 默认 5
        分组数量
    periods : tuple, 默认 (1,5,10)
        未来收益持有期
    max_loss : float, 默认 0.95
        允许丢失的最大数据比例
    
    返回:
    ----------
    factor_data : pd.DataFrame
        Alphalens 标准因子数据，包含:
        - 1D, 5D, 10D: 未来收益
        - factor: 因子值
        - factor_quantile: 因子分组标签
    """
    print("正在清洗数据...")
    factor_data = al.utils.get_clean_factor_and_forward_returns(
        factor=factor_series,
        prices=price_df,
        quantiles=quantiles,
        periods=periods,
        max_loss=max_loss,
    )
    print(f"数据清洗完成，形状: {factor_data.shape}\n")
    return factor_data
6.2 数据清洗示例
python
# 清洗数据
factor_data = models.clean_factor_data(
    factor_series=factor_series,
    price_df=price_df,
    quantiles=5,
    periods=(1,5,10),
    max_loss=0.95
)

# 查看清洗后的数据
print(factor_data.head())
#                                    1D        5D       10D    factor  factor_quantile
# date       asset
# 2021-01-04 113597.SH        0.000000 -0.004883 -0.006181  0.000000                1
#            113588.SH        0.000000  0.005067  0.007357  0.000000                1
#            113589.SH        0.000000 -0.000731 -0.008962  0.000000                1

# 查看数据统计
print(factor_data.describe())
七、IC分析模块
7.1 ic_detailed_statistics() - IC详细统计
python
def ic_detailed_statistics(self):
    """
    信息系数 (IC) 详细统计
    
    返回:
    ----------
    ic_summary : pd.DataFrame
        包含以下列:
        - IC_mean: IC均值
        - IC_std: IC标准差
        - ICIR: 信息比率 (IC均值/IC标准差)
        - IC_positive_ratio: IC为正的比例
        - IC_abs_mean: IC绝对值均值
        - Skewness: 偏度
        - Kurtosis: 峰度
        - Jarque-Bera p-value: 正态性检验p值
        - t_stat: t统计量
        - p_value: p值
    """
    print("\n>>> 信息系数 (IC) 详细统计 <<<")
    ic = al.performance.factor_information_coefficient(self.factor_data)
    ic_stats_list = []
    for col in ic.columns:
        series = ic[col].dropna()
        jb_stat, jb_p = jarque_bera(series)
        ic_stats_list.append({
            'period': col,
            'IC_mean': series.mean(),
            'IC_std': series.std(),
            'ICIR': series.mean() / series.std() if series.std() != 0 else np.nan,
            'IC_positive_ratio': (series > 0).mean(),
            'IC_abs_mean': series.abs().mean(),
            'Skewness': series.skew(),
            'Kurtosis': series.kurtosis(),
            'Jarque-Bera p-value': jb_p,
            't_stat': series.mean() / (series.std() / np.sqrt(len(series))) if series.std() != 0 else np.nan,
            'p_value': 2 * (1 - stats.t.cdf(np.abs(series.mean() / (series.std() / np.sqrt(len(series)))), df=len(series)-1)) if series.std() != 0 else np.nan
        })
    ic_summary = pd.DataFrame(ic_stats_list).set_index('period')
    print(ic_summary.round(4))
    return ic_summary
使用示例：

python
ic_summary = models.ic_detailed_statistics()
# 输出示例:
#         IC_mean  IC_std   ICIR  IC_positive_ratio  IC_abs_mean  Skewness  Kurtosis  Jarque-Bera p-value  t_stat  p_value
# period
# 1D       -0.036  0.189 -0.192              0.456        0.118    -0.050     0.438                 0.000  -6.839    0.000
# 5D       -0.042  0.182 -0.232              0.443        0.123    -0.166     0.617                 0.000  -8.262    0.000
# 10D      -0.031  0.172 -0.181              0.452        0.115     0.054     0.213                 0.000  -6.452    0.000
7.2 plot_ic_distribution() - IC分布图
python
def plot_ic_distribution(self, periods=[1,5,10]):
    """
    绘制IC分布直方图和Q-Q图
    
    参数:
    ----------
    periods : list, 默认 [1,5,10]
        需要绘制的持有期
    """
    ic = al.performance.factor_information_coefficient(self.factor_data)
    fig, axes = plt.subplots(2, 3, figsize=(15, 8))
    for i, period in enumerate(periods):
        col = f'{period}D'
        axes[0, i].hist(ic[col].dropna(), bins=30, edgecolor='black', alpha=0.7)
        axes[0, i].axvline(0, color='r', linestyle='--')
        axes[0, i].set_title(f'IC Distribution ({period}D)')
        stats.probplot(ic[col].dropna(), dist="norm", plot=axes[1, i])
        axes[1, i].set_title(f'IC Q-Q Plot ({period}D)')
    plt.tight_layout()
    plt.show()
7.3 plot_ic_decay() - IC衰减分析
python
def plot_ic_decay(self, periods=[1,5,10]):
    """
    IC衰减（自相关）图
    
    参数:
    ----------
    periods : list, 默认 [1,5,10]
        需要分析的持有期
    """
    print("\n>>> IC 衰减分析 <<<")
    ic = al.performance.factor_information_coefficient(self.factor_data)
    ic_lag_acf = {}
    for period in periods:
        col = f'{period}D'
        ic_series = ic[col].dropna()
        acf_values = [ic_series.autocorr(lag) for lag in range(1, 11)]
        ic_lag_acf[period] = acf_values
    pd.DataFrame(ic_lag_acf, index=range(1,11)).plot(marker='o', title='IC Series Autocorrelation')
    plt.xlabel('Lag (days)')
    plt.ylabel('Autocorrelation')
    plt.grid(True)
    plt.show()
八、分层收益分析模块
8.1 quantile_performance_metrics() - 分层收益绩效
python
def quantile_performance_metrics(self, period='1D'):
    """
    计算分层收益绩效指标
    
    参数:
    ----------
    period : str, 默认 '1D'
        持有期（'1D', '5D', '10D'）
    
    返回:
    ----------
    daily_ret_by_q : pd.DataFrame
        每日各分位组的平均收益
    cumulative_ret : pd.DataFrame
        各分位组的累计收益
    metrics_df : pd.DataFrame
        绩效指标DataFrame，包含:
        - 日均收益 (bps)
        - 年化收益 (%)
        - 波动率 (年化)
        - 夏普比率
        - 最大回撤 (%)
        - 卡玛比率
        - 胜率 (%)
        - 最大单日收益 (bps)
        - 最小单日收益 (bps)
    """
    print(f"\n>>> 分层收益绩效 (未来{period}日) <<<")
    daily_ret_by_q = self.factor_data.groupby(['date', 'factor_quantile'])[period].mean().unstack()
    cumulative_ret = (1 + daily_ret_by_q).cumprod()

    metrics = {}
    for q in daily_ret_by_q.columns:
        series = daily_ret_by_q[q].dropna()
        cum = cumulative_ret[q]
        rolling_max = cum.expanding().max()
        drawdown = (cum / rolling_max - 1)
        max_dd = drawdown.min()
        ann_ret = (1 + series.mean())**252 - 1
        calmar = ann_ret / abs(max_dd) if max_dd != 0 else np.nan
        metrics[q] = {
            '日均收益 (bps)': series.mean() * 10000,
            '年化收益 (%)': ann_ret * 100,
            '波动率 (年化)': series.std() * np.sqrt(252) * 100,
            '夏普比率': (series.mean() / series.std()) * np.sqrt(252) if series.std() > 0 else np.nan,
            '最大回撤 (%)': max_dd * 100,
            '卡玛比率': calmar,
            '胜率 (%)': (series > 0).mean() * 100,
            '最大单日收益 (bps)': series.max() * 10000,
            '最小单日收益 (bps)': series.min() * 10000,
        }
    metrics_df = pd.DataFrame(metrics).T
    print(metrics_df.round(4))
    return daily_ret_by_q, cumulative_ret, metrics_df
使用示例：

python
daily_ret_by_q, cumulative_ret, metrics_df = models.quantile_performance_metrics(period='1D')
# 输出示例:
#       日均收益 (bps)  年化收益 (%)  波动率 (年化)  夏普比率  最大回撤 (%)  卡玛比率  胜率 (%)  最大单日收益 (bps)  最小单日收益 (bps)
# 1.0        3.8299     10.1303     21.6363  0.4461  -25.3737  0.3992  51.8898          519.0975        -1059.7116
# 2.0        5.3086     14.3096     14.0490  0.9522  -14.8582  0.9631  53.6220          425.0273         -631.6201
# 3.0        4.0799     10.8260     12.1789  0.8442  -19.3124  0.5606  53.5433          340.8297         -502.6342
8.2 monthly_return_heatmap() - 月度分层收益热力图
python
def monthly_return_heatmap(self, period='1D'):
    """
    月度分层收益热力图
    
    参数:
    ----------
    period : str, 默认 '1D'
        持有期
    """
    print("\n>>> 月度分层收益热力图 <<<")
    daily_ret_by_q, cumulative_ret, metrics_df = self.quantile_performance_metrics(period=period)
    
    monthly_ret = daily_ret_by_q.copy()
    monthly_ret.index = pd.to_datetime(monthly_ret.index)
    monthly_ret['year'] = monthly_ret.index.year
    monthly_ret['month'] = monthly_ret.index.month
    monthly_pivot = monthly_ret.groupby(['year', 'month']).mean()
    
    fig, ax = plt.subplots(figsize=(14, 8))
    sns.heatmap(monthly_pivot, annot=True, fmt='.5f', cmap='RdYlGn', center=0, ax=ax)
    ax.set_title(f'Monthly Average Return by Quantile ({period})')
    plt.show()
九、多空组合分析模块
9.1 long_short_portfolio_analysis() - 多空组合分析
python
def long_short_portfolio_analysis(self, period='1D', window=60):
    """
    多空组合分析（Top - Bottom）
    
    参数:
    ----------
    period : str, 默认 '1D'
        持有期
    window : int, 默认 60
        滚动窗口大小
    
    返回:
    ----------
    long_short_ret : pd.Series
        多空组合日收益
    long_short_cum : pd.Series
        多空组合累计收益
    sharpe_ls : float
        多空组合夏普比率
    """
    daily_ret_by_q, cumulative_ret, metrics_df = self.quantile_performance_metrics(period=period)
    
    top_q = daily_ret_by_q.columns.max()
    bottom_q = daily_ret_by_q.columns.min()
    long_short_ret = daily_ret_by_q[top_q] - daily_ret_by_q[bottom_q]
    long_short_cum = (1 + long_short_ret).cumprod()
    rolling_ls_sharpe = long_short_ret.rolling(window).apply(
        lambda x: (x.mean()/x.std())*np.sqrt(252) if x.std()>0 else 0)
    
    ann_ret_ls = (1 + long_short_ret.mean())**252 - 1
    vol_ls = long_short_ret.std() * np.sqrt(252)
    sharpe_ls = ann_ret_ls / vol_ls if vol_ls != 0 else np.nan
    max_dd_ls = (long_short_cum / long_short_cum.cummax() - 1).min()
    
    print(f"\n>>> 多空组合 (Q{top_q} - Q{bottom_q}) 绩效 <<<")
    print(f"年化收益: {ann_ret_ls:.4%}")
    print(f"年化波动: {vol_ls:.4%}")
    print(f"夏普比率: {sharpe_ls:.4f}")
    print(f"最大回撤: {max_dd_ls:.4%}")
    print(f"胜率: {(long_short_ret>0).mean():.2%}")
    
    fig, axes = plt.subplots(2, 1, figsize=(12, 8))
    long_short_cum.plot(ax=axes[0], title='Long-Short Portfolio Cumulative Return')
    axes[0].set_ylabel('Cumulative Return')
    rolling_ls_sharpe.plot(ax=axes[1], title=f'{window}-Day Rolling Sharpe Ratio (Long-Short)', color='green')
    axes[1].axhline(0, linestyle='--', color='black')
    axes[1].set_ylabel('Sharpe Ratio')
    plt.tight_layout()
    plt.show()
    
    return long_short_ret, long_short_cum, sharpe_ls
十、换手率分析模块
10.1 turnover_analysis() - 换手率分析
python
def turnover_analysis(self):
    """
    换手率分析（每日分组变动率）
    
    返回:
    ----------
    turnover_mat : pd.DataFrame
        每日各分位组的换手率
    """
    print("\n>>> 分组换手率分析 <<<")
    group_members = self.factor_data.reset_index().groupby(['date', 'factor_quantile'])['asset'].apply(frozenset).unstack()
    turnover_mat = pd.DataFrame(index=group_members.index)
    
    for q in group_members.columns:
        current = group_members[q]
        prev = current.shift(1)
        
        def turnover_rate(curr, prev):
            if pd.isna(curr) or pd.isna(prev):
                return 0.0
            added = len(curr - prev)
            removed = len(prev - curr)
            total = len(curr)
            return (added + removed) / total if total > 0 else 0.0
        
        turnover = pd.Series(index=current.index, dtype=float)
        for dt in current.index:
            turnover[dt] = turnover_rate(current[dt], prev[dt])
        turnover_mat[f'Q{q}_turnover'] = turnover
    
    print("平均换手率 (每日组内成员变动比例):")
    print(turnover_mat.mean().round(4))
    
    turnover_mat.plot(figsize=(12,5), title='Daily Turnover by Quantile')
    plt.ylabel('Turnover Rate')
    plt.show()
    
    return turnover_mat
10.2 turnover_return_tradeoff() - 换手率与收益权衡
python
def turnover_return_tradeoff(self, period='1D'):
    """
    换手率与收益权衡散点图
    
    参数:
    ----------
    period : str, 默认 '1D'
        持有期
    """
    print("\n>>> 换手率 vs 收益 权衡 <<<")
    turnover_mat = self.turnover_analysis()
    daily_ret_by_q, cumulative_ret, metrics_df = self.quantile_performance_metrics(period=period)
    
    avg_turnover = turnover_mat.mean()
    avg_ret = daily_ret_by_q.mean()
    tradeoff = pd.DataFrame({'Turnover': avg_turnover, 'Return': avg_ret})
    
    plt.figure(figsize=(8,6))
    sns.scatterplot(data=tradeoff, x='Turnover', y='Return', s=200, hue=tradeoff.index)
    for idx in tradeoff.index:
        plt.annotate(idx, (tradeoff.loc[idx, 'Turnover'], tradeoff.loc[idx, 'Return']))
    plt.title('Turnover vs Average Daily Return')
    plt.xlabel('Average Turnover')
    plt.ylabel('Average Daily Return (bps)')
    plt.grid(True)
    plt.show()
十一、因子稳定性分析模块
11.1 factor_autocorrelation_stationarity() - 因子自相关与平稳性
python
def factor_autocorrelation_stationarity(self):
    """
    因子自相关与平稳性检验
    
    返回:
    ----------
    mean_autocorr : float
        因子一阶自相关系数均值
    """
    print("\n>>> 因子稳定性与平稳性检验 <<<")
    factor_wide = self.factor_data['factor'].unstack()
    autocorr_list = [factor_wide[col].autocorr(lag=1) for col in factor_wide.columns]
    mean_autocorr = np.nanmean(autocorr_list)
    print(f"因子一阶自相关系数 (均值): {mean_autocorr:.4f}")
    
    mean_factor_series = factor_wide.mean(axis=1).dropna()
    adf_stat, p_value, usedlag, nobs, crit_values, icbest = adfuller(mean_factor_series, autolag='AIC')
    print(f"因子均值序列 ADF 检验: 统计量={adf_stat:.4f}, p-value={p_value:.4f}")
    if p_value < 0.05:
        print("  -> 因子序列平稳")
    else:
        print("  -> 因子序列非平稳")
    
    plt.figure(figsize=(10,4))
    plt.hist(autocorr_list, bins=20, edgecolor='black')
    plt.axvline(mean_autocorr, color='r', linestyle='--', label=f'Mean={mean_autocorr:.3f}')
    plt.title('Distribution of 1st-order Autocorrelation by Asset')
    plt.xlabel('Autocorrelation')
    plt.legend()
    plt.show()
    
    return mean_autocorr
十二、统计检验模块
12.1 quantile_ttest() - 分层收益t检验
python
def quantile_ttest(self, periods=[1,5,10]):
    """
    分层收益t检验（vs 0）
    
    参数:
    ----------
    periods : list, 默认 [1,5,10]
        需要检验的持有期
    """
    print("\n>>> 分层收益统计显著性检验 (t检验 vs 0) <<<")
    
    for period in periods:
        col = f'{period}D'
        print(f"\n预测周期 {period}天:")
        for q in range(1, min(self.quantiles, 6)):
            rets = self.factor_data[self.factor_data['factor_quantile'] == q][col].dropna()
            t_stat, p_val = ttest_1samp(rets, 0)
            print(f"  Q{q}: 均值={rets.mean():.6f}, t={t_stat:.4f}, p={p_val:.4f}")
12.2 fama_macbeth_regression() - Fama-MacBeth回归
python
def fama_macbeth_regression(self, period='1D'):
    """
    Fama-MacBeth 回归估计因子溢价
    
    参数:
    ----------
    period : str, 默认 '1D'
        持有期
    
    返回:
    ----------
    fm_premium : float
        因子溢价系数
    """
    print("\n>>> Fama-MacBeth 回归 (因子溢价) <<<")
    dates = self.factor_data.index.get_level_values('date').unique()
    betas = []
    
    for dt in dates:
        data_day = self.factor_data.xs(dt, level='date')
        y = data_day[period]
        X = sm.add_constant(data_day['factor'])
        model = sm.OLS(y, X, missing='drop').fit()
        betas.append(model.params)
    
    fm_result = pd.DataFrame(betas).mean()
    print(f"Fama-MacBeth 因子溢价 ({period}): 常数项={fm_result['const']:.6f}, 因子系数={fm_result['factor']:.6f}")
    print(f"解释: 因子值每增加1单位，未来{period}日收益变化 {fm_result['factor']*10000:.2f} bps")
    return fm_result['factor']
十三、可视化分析模块
13.1 factor_vs_return_scatter() - 因子值与收益散点图
python
def factor_vs_return_scatter(self, periods=[1,5,10], sample_frac=0.1):
    """
    因子值与未来收益散点图
    
    参数:
    ----------
    periods : list, 默认 [1,5,10]
        需要绘制的持有期
    sample_frac : float, 默认 0.1
        采样比例
    """
    sample = self.factor_data.sample(frac=sample_frac, random_state=42)
    fig, axes = plt.subplots(1, 3, figsize=(18,5))
    
    for i, period in enumerate(periods):
        col = f'{period}D'
        axes[i].scatter(sample['factor'], sample[col], c=sample['factor_quantile'], cmap='viridis', alpha=0.5)
        axes[i].set_xlabel('Factor Value')
        axes[i].set_ylabel(f'Future {period}D Return')
        axes[i].set_title(f'Period {period}D | IC={sample["factor"].corr(sample[col]):.3f}')
    plt.tight_layout()
    plt.show()
13.2 rolling_sharpe_and_drawdown() - 滚动夏普与回撤
python
def rolling_sharpe_and_drawdown(self, period='1D', window=60):
    """
    滚动夏普比率和滚动最大回撤
    
    参数:
    ----------
    period : str, 默认 '1D'
        持有期
    window : int, 默认 60
        滚动窗口大小
    """
    print("\n>>> 分组收益滚动夏普 (60天) <<<")
    daily_ret_by_q, cumulative_ret, metrics_df = self.quantile_performance_metrics(period=period)
    
    rolling_sharpe = daily_ret_by_q.rolling(window).apply(
        lambda x: (x.mean()/x.std())*np.sqrt(252) if x.std()>0 else 0)
    rolling_sharpe.plot(figsize=(12,5), title=f'Rolling Sharpe Ratio ({window}-day window)')
    plt.ylabel('Sharpe Ratio')
    plt.axhline(0, color='k', linestyle='--')
    plt.show()
    
    rolling_max_dd = daily_ret_by_q.rolling(window).apply(
        lambda x: ((1+x).cumprod() / (1+x).cumprod().expanding().max() - 1).min())
    rolling_max_dd.plot(figsize=(12,5), title=f'Rolling {window}-day Max Drawdown')
    plt.ylabel('Max Drawdown')
    plt.show()
13.3 cumulative_excess_return() - 累积超额收益
python
def cumulative_excess_return(self, period='1D'):
    """
    累积超额收益（相对于等权市场）
    
    参数:
    ----------
    period : str, 默认 '1D'
        持有期
    """
    print("\n>>> 分组累积超额收益 (相对于等权市场) <<<")
    daily_ret_by_q, cumulative_ret, metrics_df = self.quantile_performance_metrics(period=period)
    
    market_ret = self.factor_data.groupby('date')['1D'].mean()
    market_cum = (1 + market_ret).cumprod()
    cum_excess = cumulative_ret.sub(market_cum, axis=0)
    
    cum_excess.plot(figsize=(12,5), title='Cumulative Excess Return by Quantile (vs Equal-Weight Market)')
    plt.ylabel('Excess Return')
    plt.axhline(0, color='k', linestyle='--')
    plt.show()
13.4 market_regime_ic() - 市场状态分割IC
python
def market_regime_ic(self):
    """
    市场状态分割：上涨日 vs 下跌日
    """
    print("\n>>> 市场状态分割：上涨日 vs 下跌日 <<<")
    market_ret = self.factor_data.groupby('date')['1D'].mean()
    up_days = market_ret > 0
    down_days = market_ret <= 0
    
    ic_daily = self.factor_data.groupby('date').apply(lambda g: g['factor'].corr(g['1D']))
    ic_up_daily = ic_daily[ic_daily.index.isin(up_days[up_days].index)]
    ic_down_daily = ic_daily[ic_daily.index.isin(down_days[down_days].index)]
    
    print(f"上涨日平均 IC: {ic_up_daily.mean():.4f}, 下跌日平均 IC: {ic_down_daily.mean():.4f}")
    print(f"上涨日 IC 胜率: {(ic_up_daily>0).mean():.2%}, 下跌日 IC 胜率: {(ic_down_daily>0).mean():.2%}")
十四、综合评分模块（28项指标）
14.1 composite_score() - 综合评分
python
def composite_score(self):
    """
    综合评分与结论（超详细版 V2）
    包含 28+ 项因子评估指标，每项均有原理说明、应用场景和评价
    """
    print("\n" + "=" * 80)
    print("第三部分：因子综合评分与结论（超详细版 V2）")
    print("=" * 80)

    # ---------- 1. 获取基础数据 ----------
    ic = al.performance.factor_information_coefficient(self.factor_data)
    ic_1d = ic['1D']
    daily_ret_by_q, cumulative_ret, metrics_df = self.quantile_performance_metrics(period='1D')
    long_short_ret, long_short_cum, sharpe_ls = self.long_short_portfolio_analysis(window=60)
    turnover_mat = self.turnover_analysis()
    ic_summary = self.ic_detailed_statistics()
    mean_autocorr = self.factor_autocorrelation_stationarity()
    fm_premium = self.fama_macbeth_regression(period='1D')

    # 提取常用数值
    ic_mean_1d = ic_summary.loc['1D', 'IC_mean']
    ic_ir_1d = ic_summary.loc['1D', 'ICIR']
    turnover_mean = turnover_mat.mean().mean()
    fm_premium_bps = fm_premium * 10000
    factor_autocorr_mean = mean_autocorr

    # ---------- 2. 分组收益单调性（Spearman秩相关） ----------
    ret_col = None
    for col in metrics_df.columns:
        if '日均收益' in col and 'bps' in col:
            ret_col = col
            break
    if ret_col is None:
        ret_col = metrics_df.columns[0]
    mean_returns_bps = [metrics_df.loc[q, ret_col] for q in sorted(metrics_df.index)]
    quantile_ranks = np.arange(1, len(mean_returns_bps) + 1)
    spearman_corr, spearman_p = spearmanr(quantile_ranks, mean_returns_bps)

    # ---------- 3. 极端值分析 ----------
    factor_values = self.factor_data['factor']
    high_10 = factor_values.quantile(0.9)
    low_10 = factor_values.quantile(0.1)
    high_group_ret = self.factor_data[self.factor_data['factor'] >= high_10]['1D'].mean()
    low_group_ret = self.factor_data[self.factor_data['factor'] <= low_10]['1D'].mean()
    extreme_spread = (high_group_ret - low_group_ret) * 10000

    # ---------- 4. 月度IC稳定性 ----------
    ic_monthly_mean = ic_1d.resample('M').mean()
    ic_monthly_std = ic_monthly_mean.std()

    # ---------- 5. 多空组合最大回撤持续期 ----------
    drawdown_series = (long_short_cum / long_short_cum.cummax() - 1)
    in_drawdown = (drawdown_series < 0).astype(int)
    drawdown_duration = 0
    max_duration = 0
    for val in in_drawdown:
        if val == 1:
            drawdown_duration += 1
        else:
            max_duration = max(max_duration, drawdown_duration)
            drawdown_duration = 0
    max_duration = max(max_duration, drawdown_duration)

    # ---------- 6. 分层收益夏普对比 ----------
    top_q = daily_ret_by_q.columns.max()
    bottom_q = daily_ret_by_q.columns.min()
    top_q_ret = daily_ret_by_q[top_q].dropna()
    bottom_q_ret = daily_ret_by_q[bottom_q].dropna()
    top_sharpe = (top_q_ret.mean() / top_q_ret.std()) * np.sqrt(252) if top_q_ret.std() != 0 else np.nan
    bottom_sharpe = (bottom_q_ret.mean() / bottom_q_ret.std()) * np.sqrt(252) if bottom_q_ret.std() != 0 else np.nan

    # ---------- 7. 因子值分布特征 ----------
    factor_all = factor_values.dropna()
    factor_skew = factor_all.skew()
    factor_kurt = factor_all.kurtosis()

    # ---------- 8. 滚动IC趋势 ----------
    ic_rolling_mean = ic_1d.rolling(20).mean()
    recent_ic_trend = ic_rolling_mean.iloc[-1] - ic_rolling_mean.iloc[-20] if len(ic_rolling_mean) >= 20 else 0

    # ---------- 9. 多空组合卡玛比率 ----------
    ls_ann_ret = (1 + long_short_ret.mean()) ** 252 - 1
    ls_max_dd = (long_short_cum / long_short_cum.cummax() - 1).min()
    calmar_ls = ls_ann_ret / abs(ls_max_dd) if ls_max_dd != 0 else np.nan

    # ---------- 10. 多空组合回撤修复天数 ----------
    cummax_series = long_short_cum.cummax()
    trough_date = drawdown_series.idxmin()
    recovery_date = None
    for dt in long_short_cum.index[long_short_cum.index > trough_date]:
        if long_short_cum.loc[dt] >= cummax_series.loc[trough_date]:
            recovery_date = dt
            break
    if recovery_date is not None:
        recovery_days = (recovery_date - trough_date).days
    else:
        recovery_days = np.nan

    # ---------- 11. 多空组合月度胜率 ----------
    monthly_ls_ret = long_short_ret.resample('M').mean()
    monthly_win_rate = (monthly_ls_ret > 0).mean()

    # ---------- 12. 因子值与市场收益的相关性 ----------
    market_ret = self.factor_data.groupby('date')['1D'].mean()
    common_idx = long_short_ret.index.intersection(market_ret.index)
    if len(common_idx) > 5:
        corr_market, p_market = pearsonr(long_short_ret.loc[common_idx], market_ret.loc[common_idx])
    else:
        corr_market, p_market = np.nan, np.nan

    # ---------- 13. IC序列的偏度与峰度 ----------
    ic_skew = ic_1d.dropna().skew()
    ic_kurt = ic_1d.dropna().kurtosis()

    # ---------- 14. IC序列的自相关（滞后1天） ----------
    ic_autocorr = ic_1d.dropna().autocorr(lag=1)

    # ---------- 15. IC的滚动波动率（20天） ----------
    ic_rolling_vol = ic_1d.rolling(20).std()
    recent_ic_vol = ic_rolling_vol.iloc[-1] if len(ic_rolling_vol) > 0 else np.nan

    # ---------- 16. 因子值的极端值比例（超出±3σ） ----------
    factor_mean = factor_all.mean()
    factor_std = factor_all.std()
    outliers = factor_all[(factor_all < factor_mean - 3*factor_std) | (factor_all > factor_mean + 3*factor_std)]
    outlier_ratio = len(outliers) / len(factor_all)

    # ---------- 17. 因子值滚动标准差（20天）的稳定性 ----------
    factor_wide = self.factor_data['factor'].unstack()
    rolling_std = factor_wide.rolling(20).std().mean(axis=1)
    rolling_std_vol = rolling_std.std()

    # ---------- 18. 分组换手率的稳定性 ----------
    turnover_std = turnover_mat.mean(axis=0).std()

    # ---------- 19. 分组收益中位数差异 ----------
    top_median = top_q_ret.median()
    bottom_median = bottom_q_ret.median()
    median_spread = (top_median - bottom_median) * 10000

    # ---------- 20. 分组收益t检验p值 ----------
    t_stat, p_val_bottom = ttest_1samp(bottom_q_ret.dropna(), 0)
    t_stat, p_val_top = ttest_1samp(top_q_ret.dropna(), 0)

    # ---------- 21. 各组最大回撤深度 ----------
    group_max_dd = {}
    for q in daily_ret_by_q.columns:
        cum = (1 + daily_ret_by_q[q]).cumprod()
        dd = (cum / cum.cummax() - 1).min()
        group_max_dd[q] = dd

    # ---------- 22. 最高组连续盈利/亏损的最大天数 ----------
    ret_sign = (top_q_ret > 0).astype(int)
    streak = 0
    max_win_streak = 0
    max_loss_streak = 0
    for s in ret_sign:
        if s == 1:
            streak = streak + 1 if streak > 0 else 1
            max_win_streak = max(max_win_streak, streak)
        else:
            streak = streak - 1 if streak < 0 else -1
            max_loss_streak = max(max_loss_streak, -streak)

    # ================== 输出所有指标（含解释） ==================
    # [输出28项指标的详细说明和评价...]
    # （为节省篇幅，此处省略详细输出代码，完整实现见源代码）

    # ---------- 综合评分 ----------
    score = 0
    score += min(ic_mean_1d * 100, 10) if ic_mean_1d > 0 else max(ic_mean_1d * 100, -10)
    score += min(ic_ir_1d, 5) if ic_ir_1d > 0 else max(ic_ir_1d, -5)
    score += min(sharpe_ls, 5) if not np.isnan(sharpe_ls) else 0
    score += max(0, 5 - turnover_mean * 10)
    score += factor_autocorr_mean * 5
    score += min(fm_premium_bps, 5) if fm_premium > 0 else max(fm_premium_bps, -5)

    # 评级与建议
    if score >= 30:
        grade = "A+ (极优秀)"
        advice = "强烈建议纳入策略，可直接作为主要选股因子。"
    elif score >= 25:
        grade = "A (优秀)"
        advice = "建议使用，可与其他因子简单结合。"
    elif score >= 20:
        grade = "B (良好)"
        advice = "有一定价值，需优化或组合使用。"
    elif score >= 15:
        grade = "C (及格)"
        advice = "效果较弱，仅作为辅助参考。"
    else:
        grade = "D (较差)"
        advice = "不建议单独使用，请重新审视因子定义或数据。"

    print(f"\n📊 因子综合评分: {score:.2f} / 40")
    print(f"📝 因子诊断总结")
    print(f"  综合评级: {grade}")
    print(f"  操作建议: {advice}")

    return score
十五、一键运行完整分析
15.1 run_full_analysis() - 完整分析流程
python
def run_full_analysis(self, periods=[1,5,10]):
    """
    运行完整的因子分析流程（包含所有模块）
    
    参数:
    ----------
    periods : list, 默认 [1,5,10]
        需要分析的持有期
    """
    factor_data = self.factor_data
    
    # Alphalens 标准报告
    print("\n" + "="*80)
    print("第一部分：Alphalens 标准完整报告")
    print("="*80)
    al.tears.create_full_tear_sheet(factor_data)

    print("\n" + "="*80)
    print("第二部分：自定义深度分析")
    print("="*80)

    # IC 分析
    ic = al.performance.factor_information_coefficient(factor_data)
    ic_summary = self.ic_detailed_statistics()
    self.plot_ic_distribution(periods)
    self.plot_ic_decay(periods)

    # 分层收益与多空组合
    daily_ret_by_q, cumulative_ret, metrics_df = self.quantile_performance_metrics(period='1D')
    long_short_ret, long_short_cum, sharpe_ls = self.long_short_portfolio_analysis(window=60)

    # 月度热力图
    self.monthly_return_heatmap()

    # 换手率
    turnover_mat = self.turnover_analysis()

    # 因子自相关与平稳性
    mean_autocorr = self.factor_autocorrelation_stationarity()

    # t检验
    self.quantile_ttest(periods)

    # 滚动指标
    self.rolling_sharpe_and_drawdown(window=60)

    # 散点图
    self.factor_vs_return_scatter(periods, sample_frac=0.1)

    # 市场状态分割
    self.market_regime_ic()

    # 换手率与收益权衡
    self.turnover_return_tradeoff()

    # Fama-MacBeth
    fm_premium = self.fama_macbeth_regression(period='1D')

    # 累积超额收益
    self.cumulative_excess_return()

    # 综合评分
    self.composite_score()
十六、完整使用示例
16.1 使用模拟数据（快速测试）
python
from xg_alphalens.xg_alphalens import xg_alphalens

# 1. 生成模拟数据
models = xg_alphalens()
factor_df, price_df, factor_series = models.generate_synthetic_data(
    start_date='2024-01-01',
    end_date='2024-12-31',
    n_stocks=30,
    seed=42
)

# 2. 数据清洗
factor_data = models.clean_factor_data(
    factor_series=factor_series,
    price_df=price_df,
    quantiles=5,
    periods=(1,5,10),
    max_loss=0.95
)

# 3. 赋值给模型
models.factor_data = factor_data

# 4. 运行完整分析
models.run_full_analysis()

# 或单独运行综合评分
score = models.composite_score()
16.2 使用真实数据（通达信）
python
from xg_alphalens.xg_alphalens import xg_alphalens
import pandas as pd

# 1. 初始化
models = xg_alphalens(quantiles=10, periods=(1,5,10), max_loss=0.95)

# 2. 使用数据准备函数
factor_df, price_df, factor_series, stock_list = prepare_factor_data(
    block_file='QBKZZ.blk',
    start_date='20210101',
    end_date='20500101'
)

# 3. 数据清洗
factor_data = models.clean_factor_data(
    factor_series=factor_series,
    price_df=price_df,
    quantiles=10,
    periods=(1,5,10),
    max_loss=0.95
)

# 4. 赋值并分析
models.factor_data = factor_data

# 5. 运行完整分析
models.r