因子衰减与因子轮动：量化投资中的时间维度

import numpy as np
import matplotlib.pyplot as plt

def factor_decay_model(t, IR0=1.2, lam=0.3, IR_inf=0.1):
    """因子衰减模型"""
    return IR0 * np.exp(-lam * t) + IR_inf

# 绘制衰减曲线
t = np.linspace(0, 10, 100)
IR = factor_decay_model(t, IR0=1.5, lam=0.4, IR_inf=0.05)

plt.figure(figsize=(10, 6))
plt.plot(t, IR, 'b-', linewidth=2)
plt.axhline(y=0.3, color='r', linestyle='--', label='临界值 IR=0.3')
plt.xlabel('时间 (年)')
plt.ylabel('信息比率 (IR)')
plt.title('因子衰减曲线')
plt.grid(True)
plt.legend()
plt.show()

import pandas as pd
import numpy as np

def factor_rotation_by_economy(df_factors, df_macro, window=12):
    """
    df_factors: 因子收益率 DataFrame, shape=(T, N_factors)
    df_macro: 宏观经济指标 DataFrame, shape=(T, N_macro)
    """
    # 1. 计算宏观指标的趋势
    macro_signal = np.sign(df_macro.diff(window))
    
    # 2. 根据宏观信号选择因子
    factor_weights = pd.DataFrame(index=df_factors.index, columns=df_factors.columns)
    
    for i in range(len(df_factors)):
        if macro_signal.iloc[i]['GDP_growth'] > 0:  # 经济扩张
            factor_weights.iloc[i] = [0.3, 0.4, 0.1, 0.2]  # 侧重动量和质量
        else:  # 经济衰退
            factor_weights.iloc[i] = [0.2, 0.1, 0.4, 0.3]  # 侧重低波动和价值
    
    # 3. 计算因子组合收益
    portfolio_return = (factor_weights.shift(1) * df_factors).sum(axis=1)
    return portfolio_return

def factor_momentum_rotation(df_factors, lookback=12, hold=3):
    """
    因子动量轮动：选择过去表现最好的因子
    """
    factor_returns = df_factors.pct_change()
    
    # 计算滚动收益
    rolling_return = factor_returns.rolling(lookback).sum()
    
    # 选择前3名因子
    top_factors = rolling_return.apply(lambda x: x.nlargest(3).index.tolist(), axis=1)
    
    # 等权配置
    portfolio_return = pd.Series(index=df_factors.index)
    for i in range(hold, len(df_factors)):
        selected = top_factors.iloc[i-hold]
        portfolio_return.iloc[i] = factor_returns.iloc[i][selected].mean()
    
    return portfolio_return

┌─────────────────────────────────────┐
│         因子衰减监控系统             │
│  - 跟踪每个因子的实时IR             │
│  - 检测衰减加速度                   │
└──────────────┬──────────────────────┘
               │
               ▼
┌─────────────────────────────────────┐
│         因子轮动决策模块             │
│  - 宏观环境判断                     │
│  - 因子相对强度比较                 │
└──────────────┬──────────────────────┘
               │
               ▼
┌─────────────────────────────────────┐
│         因子组合优化模块             │
│  - 风险预算分配                     │
│  - 交易成本约束                     │
└─────────────────────────────────────┘

class FactorDecayMonitor:
    def __init__(self, factor_data, lookback=36):
        self.factor_data = factor_data
        self.lookback = lookback
        
    def calculate_decay_speed(self, factor_name):
        """计算因子衰减速度"""
        returns = self.factor_data[factor_name].pct_change()
        
        # 分段计算IR
        ir_list = []
        for start in range(0, len(returns)-self.lookback, 12):
            end = start + self.lookback
            segment = returns[start:end]
            ir = segment.mean() / segment.std() * np.sqrt(12)
            ir_list.append(ir)
        
        # 拟合衰减曲线
        t = np.arange(len(ir_list))
        popt, _ = curve_fit(factor_decay_model, t, ir_list)
        decay_rate = popt[1]  # lambda
        
        return decay_rate
    
    def alert_if_fast_decay(self, threshold=0.2):
        """衰减过快时报警"""
        for factor in self.factor_data.columns:
            decay_rate = self.calculate_decay_speed(factor)
            if decay_rate > threshold:
                print(f"警告：{factor} 衰减过快！衰减速率={decay_rate:.3f}")

def value_momentum_rotation_backtest(df_value, df_momentum, df_macro, initial_capital=1e6):
    """
    价值-动量轮动回测
    """
    # 合并因子收益率
    df_factors = pd.DataFrame({
        'value': df_value.pct_change(),
        'momentum': df_momentum.pct_change()
    }).dropna()
    
    # 信号生成：根据宏观经济状态选择因子
    signals = pd.DataFrame(index=df_factors.index)
    signals['regime'] = np.where(df_macro['GDP_growth'].diff(12) > 0, 'expansion', 'recession')
    
    # 因子权重
    signals['value_weight'] = np.where(signals['regime'] == 'expansion', 0.3, 0.7)
    signals['momentum_weight'] = 1 - signals['value_weight']
    
    # 计算策略收益
    strategy_return = (signals[['value_weight', 'momentum_weight']].shift(1).values * 
                       df_factors.values).sum(axis=1)
    
    # 计算净值
    cumulative_return = (1 + strategy_return).cumprod()
    
    # 性能指标
    annual_return = cumulative_return.iloc[-1] ** (12/len(cumulative_return)) - 1
    sharpe = strategy_return.mean() / strategy_return.std() * np.sqrt(12)
    max_dd = (cumulative_return / cumulative_return.cummax() - 1).min()
    
    performance = {
        '年化收益': annual_return,
        '夏普比率': sharpe,
        '最大回撤': max_dd
    }
    
    return cumulative_return, performance

import lightgbm as lgb
from sklearn.model_selection import TimeSeriesSplit

def predict_factor_performance(factor_returns, features, lookahead=3):
    """
    使用LightGBM预测因子未来表现
    """
    # 构建标签：未来3个月因子收益是否跑赢中位数
    y = (factor_returns.rolling(lookahead).sum().shift(-lookahead) > 
         factor_returns.rolling(lookahead).sum().shift(-lookahead).median(axis=1))
    
    # 时间序列交叉验证
    tscv = TimeSeriesSplit(n_splits=5)
    
    models = []
    for train_idx, test_idx in tscv.split(features):
        X_train, X_test = features.iloc[train_idx], features.iloc[test_idx]
        y_train, y_test = y.iloc[train_idx], y.iloc[test_idx]
        
        # 训练LightGBM
        model = lgb.LGBMClassifier(
            n_estimators=100,
            learning_rate=0.05,
            max_depth=5
        )
        model.fit(X_train, y_train)
        models.append(model)
    
    # 集成预测
    predictions = np.mean([m.predict_proba(features) for m in models], axis=0)
    return predictions