python
import math
import random
import statistics
from typing import Callable, Dict, List, Any, Tuple
import numpy as np
# Types:
# traffic_fn(t) -> int (number of users arriving at time t)
# reward_fn(t, action, n_samples=1) -> array-like rewards for n_samples users
# strategy(history, t, actions) -> action
# actions: list of possible actions/arms
def simulate_regret(
strategies: Dict[str, Callable[[Dict, int, List[Any]], Any]],
actions: List[Any],
traffic_fn: Callable[[int], int],
reward_fn: Callable[[int, Any, int], np.ndarray],
time_horizon: int,
trials: int = 1000,
oracle_samples_per_step: int = 1000,
seed: int = None,
) -> Dict[str, Dict]:
"""
Returns per-strategy stats on cumulative regret:
{ name: { 'mean':..., 'variance':..., 'ci95': (low, high), 'all': [...] } }
- strategies: mapping name -> strategy(history, t, actions) where history is dict
that the strategy can read/modify to store past observations. Implementations should
be deterministic given same seed and inputs.
- traffic_fn(t): number users at time t
- reward_fn(t, action, n): returns n IID rewards (np.ndarray)
- oracle_samples_per_step: samples used to estimate best action each timestep per trial
Performance: O(trials * time_horizon * oracle_samples_per_step) reward draws; vectorize reward_fn where possible.
"""
if seed is not None:
random.seed(seed)
np.random.seed(seed)
regrets_per_strategy = {name: [] for name in strategies}
for tr in range(trials):
# per-trial cumulative rewards per strategy
cum_rewards = {name: 0.0 for name in strategies}
# initialize separate histories (deep copy if needed)
histories = {name: {} for name in strategies}
for t in range(time_horizon):
n_users = int(traffic_fn(t))
if n_users <= 0:
continue
# Estimate oracle expected reward-per-user for each action this timestep
# (could be cached if stationary)
oracle_means = {}
for a in actions:
samples = reward_fn(t, a, n=oracle_samples_per_step)
oracle_means[a] = float(np.mean(samples))
best_mean = max(oracle_means.values())
# For each strategy, get chosen action and accumulate reward
for name, strat in strategies.items():
act = strat(histories[name], t, actions)
# sample rewards for n_users users
rewards = reward_fn(t, act, n=n_users)
total_reward = float(np.sum(rewards))
cum_rewards[name] += total_reward
# allow strategy to observe outcomes
histories[name].setdefault('records', []).append((t, act, rewards))
# regret is (best_mean * n_users) - achieved_reward_per_strategy
for name in strategies:
achieved = float(np.sum(histories[name]['records'][-1][2]))
regret = best_mean * n_users - achieved
# accumulate per-trial total regret (store later)
histories[name].setdefault('trial_regret', 0.0)
histories[name]['trial_regret'] += regret
# after trial, record cumulative regret per strategy
for name in strategies:
regrets_per_strategy[name].append(histories[name].get('trial_regret', 0.0))
results = {}
for name, vals in regrets_per_strategy.items():
mean = statistics.mean(vals)
var = statistics.pvariance(vals)
# 95% CI using t-approx ~ normal for large trials
se = math.sqrt(var / max(1, len(vals)))
ci95 = (mean - 1.96 * se, mean + 1.96 * se)
results[name] = {
'mean': mean,
'variance': var,
'ci95': ci95,
'all': vals,
}
return results
# Example strategy signatures:
# def epsilon_greedy(history, t, actions, epsilon=0.1):
# # history can keep counts/estimates; return an action from actions
# ...
