多臂老虎机与策略动态配置：探索-利用权衡的量化实战

import numpy as np

def epsilon_greedy_weights(mean_returns, epsilon=0.1):
    """
    mean_returns: 各策略的历史平均收益估计 (array)
    返回: 资金权重 (array, 和为1)
    """
    n = len(mean_returns)
    weights = np.full(n, epsilon / n)          # 每个策略先分探索份额
    best = np.argmax(mean_returns)
    weights[best] += (1 - epsilon)             # 最优策略额外拿利用份额
    return weights

means = np.array([0.18, 0.06, 0.03])
print(epsilon_greedy_weights(means, epsilon=0.15))
# [0.90  0.05  0.05]

def ucb_scores(mean_returns, counts, total_rounds, c=2.0):
    """
    mean_returns: 各策略平均收益估计
    counts: 各策略被观测/分配的次数
    total_rounds: 累计总轮数
    """
    counts = np.maximum(counts, 1e-9)          # 防止除零
    bonus = c * np.sqrt(np.log(total_rounds + 1) / counts)
    return mean_returns + bonus

def ucb_weights(mean_returns, counts, total_rounds, c=2.0, temp=10.0):
    scores = ucb_scores(mean_returns, counts, total_rounds, c)
    # 用 softmax 把分数转成平滑权重，而非硬性 all-in
    exp_s = np.exp(temp * (scores - scores.max()))
    return exp_s / exp_s.sum()

means = np.array([0.18, 0.06, 0.03])
counts = np.array([12, 3, 3])                  # A 观测多，B/C 观测少
w = ucb_weights(means, counts, total_rounds=18, c=0.05)
print(np.round(w, 3))

class ThompsonAllocator:
    def __init__(self, n_strategies, prior_mu=0.0, prior_var=0.04):
        self.n = n_strategies
        self.mu = np.full(n_strategies, prior_mu)      # 后验均值
        self.var = np.full(n_strategies, prior_var)    # 后验方差
        self.obs_var = 0.04                            # 观测噪声（月收益方差）

    def update(self, strategy_idx, observed_return):
        """贝叶斯更新：正态-正态共轭"""
        prior_prec = 1.0 / self.var[strategy_idx]
        obs_prec = 1.0 / self.obs_var
        post_prec = prior_prec + obs_prec
        self.mu[strategy_idx] = (
            prior_prec * self.mu[strategy_idx] + obs_prec * observed_return
        ) / post_prec
        self.var[strategy_idx] = 1.0 / post_prec

    def allocate(self, n_samples=2000):
        """蒙特卡洛：多次采样取平均，得到平滑权重"""
        wins = np.zeros(self.n)
        for _ in range(n_samples):
            sampled = np.random.normal(self.mu, np.sqrt(self.var))
            wins[np.argmax(sampled)] += 1
        return wins / n_samples

import numpy as np

np.random.seed(42)
T = 120                    # 120 个月
n_strat = 3

# 构造 regime 切换的真实收益：每 40 个月轮换领跑者
true_mu = np.zeros((T, n_strat))
for t in range(T):
    regime = (t // 40) % 3
    base = np.array([0.002, 0.002, 0.002])
    base[regime] = 0.015   # 当前 regime 的领跑策略
    true_mu[t] = base

# 生成月度收益（真实均值 + 噪声）
returns = true_mu + np.random.normal(0, 0.04, (T, n_strat))

def run_thompson(returns):
    alloc = ThompsonAllocator(n_strat)
    port = []
    for t in range(len(returns)):
        w = alloc.allocate(n_samples=500)
        port.append(np.dot(w, returns[t]))
        for i in range(n_strat):      # 用实现收益更新后验（加权观测）
            alloc.update(i, returns[t, i])
    return np.array(port)

def run_greedy(returns, lookback=12):
    port = []
    for t in range(len(returns)):
        if t < lookback:
            w = np.ones(n_strat) / n_strat
        else:
            hist_mean = returns[t-lookback:t].mean(axis=0)
            w = np.zeros(n_strat)
            w[np.argmax(hist_mean)] = 1.0     # all-in 历史最优
        port.append(np.dot(w, returns[t]))
    return np.array(port)

def run_equal(returns):
    w = np.ones(n_strat) / n_strat
    return returns @ w

ts = run_thompson(returns)
gr = run_greedy(returns)
eq = run_equal(returns)

for name, p in [("Thompson", ts), ("Greedy", gr), ("Equal", eq)]:
    ann = p.mean() * 12
    sharpe = p.mean() / p.std() * np.sqrt(12)
    print(f"{name:10s} 年化 {ann:6.2%}  夏普 {sharpe:5.2f}")