组合优化与风险模型

因子值 → 因子合成 → 预期收益预测 → 组合优化 → 目标权重
  → 交易成本约束 → 订单生成 → 执行

def signal_to_weights(factor_composite, cov, prev_w,
                      risk_aversion=5.0, max_weight=0.03):
    mu = factor_composite * 0.01          # 因子 → 预期收益（年化 1% 尺度）
    n = len(mu)
    def neg_utility(w):   # 目标：max mu^T w - lambda/2 * w^T Σ w
        return -(mu @ w - risk_aversion / 2 * w @ cov @ w)
    constraints = [{'type': 'eq', 'fun': lambda w: w.sum() - 1}]  # 满仓
    bounds = [(-max_weight, max_weight)] * n                      # 单票上限
    x0 = prev_w if prev_w is not None else np.ones(n) / n
    res = minimize(neg_utility, x0, bounds=bounds, constraints=constraints)
    return res.x

max  μᵀw − (λ/2)·wᵀΣw
s.t. 1ᵀw = 1        （满仓）
     w ≥ 0          （不做空）
     |w| ≤ u        （单票上限）

def sensitivity_test(mu, cov, n_trials=100, noise=0.01):
    base = optimize(mu, cov)
    turns = []
    for _ in range(n_trials):
        mu_noisy = mu + np.random.normal(0, noise, len(mu))
        w = optimize(mu_noisy, cov)
        turns.append(np.abs(w - base).sum())
    return np.mean(turns), np.percentile(turns, 95)

from sklearn.covariance import LedoitWolf

def estimate_cov(returns):   # returns 为 T×N 矩阵，T 天数，N 股票数
    lw = LedoitWolf(assume_centered=False).fit(returns)
    shrunk = lw.covariance_          # 收缩后的协方差
    shrinkage = lw.shrinkage_        # 收缩强度，0~1
    return shrunk * 252, shrinkage   # 年化

def ewma_cov(returns, halflife=60):
    lam = np.exp(np.log(0.5) / halflife)   # RiskMetrics 风格衰减因子
    T, N = returns.shape
    weights = lam ** np.arange(T)[::-1]
    weights /= weights.sum()
    demeaned = returns - returns.mean(axis=0)
    return (demeaned * weights[:, None]).T @ demeaned * 252

r = X·f + u

X：N×K 的因子暴露矩阵（行业 + 风格）
f：K 维因子收益
u：特质收益（假设对角）

Σ = X·Σf·Xᵀ + Δ

Σf：K×K 的因子协方差（K 通常 30~60）
Δ：N×N 的对角矩阵（特质风险）

def factor_cov(X, factor_returns, spec_var):   # X 为 N×K 暴露，factor_returns 为 T×K
    Sigma_f = np.cov(factor_returns, rowvar=False) * 252   # K×K
    Delta = np.diag(spec_var)                              # 特质风险对角阵
    return X @ Sigma_f @ X.T + Delta

def build_l1_turnover(n, prev_w):
    c = np.ones(2 * n)   # 最小化 sum(c*(u+v))，约束 w - prev_w = u - v
    A_eq = np.hstack([np.eye(n), -np.eye(n)])   # w - prev_w = u - v
    return c, A_eq

def shrink_mu(mu, alpha=0.5):   # alpha 越大，收缩越强
    return alpha * mu.mean() + (1 - alpha) * mu

RC_i = w_i · (Σw)_i / (wᵀΣw)

目标：RC_i = 1/N，对所有 i

from scipy.optimize import minimize

def risk_parity(cov):
    n = cov.shape[0]
    def objective(w):
        port_var = w @ cov @ w
        rc = w * (cov @ w) / port_var      # 各资产风险贡献
        return ((rc - 1.0 / n) ** 2).sum() # 目标：都等于 1/N
    cons = [{'type': 'eq', 'fun': lambda w: w.sum() - 1}]
    res = minimize(objective, np.ones(n) / n, bounds=[(1e-6, 1)] * n, constraints=cons)
    return res.x

import cvxpy as cp

def optimize_cvxpy(mu, cov, prev_w, risk_aversion=5.0,
                   max_weight=0.03, turnover_limit=0.3):
    n = len(mu)
    w = cp.Variable(n)
    objective = cp.Maximize(mu @ w - risk_aversion / 2 * cp.quad_form(w, cov))
    constraints = [
        cp.sum(w) == 1,
        w <= max_weight,
        w >= -max_weight,
        cp.norm1(w - prev_w) <= turnover_limit,
    ]
    prob = cp.Problem(objective, constraints)
    prob.solve(solver=cp.OSQP)     # OSQP 适合大规模 QP
    return w.value

1. 不交易带（no-trade band）：偏离目标权重 < 0.5% 不动
2. 分批执行：把大调仓拆成几天完成
3. 换手上限：约束每日换手不超过 20%

def apply_no_trade_band(target_w, current_w, band=0.005):
    diff = target_w - current_w   # 偏离小于阈值的股票保持不动
    new_w = np.where(np.abs(diff) < band, current_w, target_w)
    return new_w / new_w.sum()    # 重新归一化