因子拥挤与拥挤崩塌：当量化策略同质化引发市场风暴

# 因子拥挤度的简单衡量指标
def calculate_crowding_score(factor_returns, aum_data):
    """
    计算因子拥挤度评分
    factor_returns: 因子历史收益序列
    aum_data: 追踪该因子的资产管理规模
    """
    # 1. 资金集中度：AUM增长率 vs 因子收益
    aum_growth = aum_data.pct_change(12).mean()  # 年化AUM增长
    factor_return = factor_returns.mean() * 12  # 年化因子收益
    
    # 2.  turnover比率：高turnover暗示拥挤
    turnover = calculate_factor_turnover(factor_returns)
    
    # 3. 相关性集中度：因子与基准的相关性
    correlation_concentration = factor_returns.rolling(60).corr(benchmark).mean()
    
   拥挤_score = (aum_growth / factor_return) * turnover * correlation_concentration
    return crowding_score

# 计算因子资金流压力
def funding_pressure_factor(factor_exposure, fund_flows):
    """
    衡量因子承受的资金流压力
    """
    # 标准化因子暴露
    norm_exposure = (factor_exposure - factor_exposure.mean()) / factor_exposure.std()
    
    # 计算资金流与因子暴露的相关性
    flow_factor_corr = fund_flows.rolling(60).apply(
        lambda x: np.corrcoef(x, norm_exposure[:len(x)])[0,1]
    )
    
    # 高相关性暗示拥挤
    crowding_signal = flow_factor_corr > 0.7
    return crowding_signal

# 估值离散度衡量拥挤程度
def valuation_dispersion(factor_portfolio):
    """
    计算因子组合内股票的估值离散度
    离散度越低，暗示拥挤越严重
    """
    # 提取组合内股票的估值指标（如P/E, P/B）
    valuations = factor_portfolio['valuation'].values
    
    # 计算离散度（标准差/均值）
    dispersion = np.std(valuations) / np.mean(valuations)
    
    # 离散度Z-score（相对于历史）
    historical_dispersion = load_historical_dispersion(factor_portfolio.name)
    z_score = (dispersion - historical_dispersion.mean()) / historical_dispersion.std()
    
    return z_score  # Z-score < -2 表示极度拥挤

# 因子相关性矩阵分析
def correlation_concentration(factor_returns_df):
    """
    分析因子间相关性是否异常升高
    """
    # 计算因子收益相关性矩阵
    corr_matrix = factor_returns_df.rolling(60).corr()
    
    # 提取上三角矩阵（排除自相关）
    upper_tri = corr_matrix.where(np.triu(np.ones(corr_matrix.shape), k=1).astype(bool))
    
    # 平均相关性
    avg_corr = upper_tri.mean(axis=1)
    
    # 相关性集中度（赫芬达尔指数）
    hhi = (upper_tri**2).sum(axis=1)
    
    # 高HHI暗示因子策略同质化严重
    crowding_alert = hhi > 0.3  # 经验阈值
    return crowding_alert, avg_corr

# 动态仓位调整模型
def adaptive_position_sizing(factor_score, crowding_indicator, max_position=0.05):
    """
    根据拥挤度动态调整因子敞口
    """
    # 基础仓位
    base_position = factor_score.rank(pct=True) 
    
    # 拥挤度调整
    if crowding_indicator > 0.8:  # 高拥挤
        adjustment = 0.5  # 减半仓位
    elif crowding_indicator > 0.6:  # 中等拥挤
        adjustment = 0.75
    else:  # 低拥挤
        adjustment = 1.0
    
    # 最终仓位
    final_position = base_position * adjustment
    final_position = final_position.clip(upper=max_position)
    
    return final_position