Lag-Compensated Spectral Scheduler / lag_scheduler_experiment.py

Failed on benchmark

Raw ⬇ ZIP
  1import json
  2import random
  3from pathlib import Path
  4import numpy as np
  5
  6
  7def simulate(theta0, theta1, t_end, dt, kappa, theta_c, slope=1.0, mode="baseline"):
  8    n = int(round(t_end / dt)) + 1
  9    ts = np.arange(n) * dt
 10    desired = theta0 + (theta1 - theta0) * np.minimum(ts / t_end, 1.0)
 11    r = (theta1 - theta0) / t_end
 12    if mode == "baseline":
 13        command = desired.copy()
 14    elif mode == "minus":
 15        command = desired - r / kappa
 16    elif mode == "plus":
 17        command = desired + r / kappa
 18    else:
 19        raise ValueError(mode)
 20    eff = np.empty(n)
 21    eff[0] = theta0
 22    for i in range(1, n):
 23        eff[i] = eff[i - 1] + dt * kappa * (command[i - 1] - eff[i - 1])
 24    lam_eff = slope * (eff - theta_c)
 25    lam_cmd = slope * (command - theta_c)
 26
 27    def first_cross(x):
 28        ix = np.flatnonzero(x >= 0)
 29        if len(ix) == 0:
 30            return float("nan")
 31        i = int(ix[0])
 32        if i == 0:
 33            return float(ts[0])
 34        # Linear interpolation makes timing less dependent on dt.
 35        return float(ts[i - 1] + dt * (-x[i - 1]) / (x[i] - x[i - 1]))
 36
 37    target_cross = (theta_c - theta0) / r
 38    return {
 39        "target_cross": target_cross,
 40        "command_cross": first_cross(lam_cmd),
 41        "effective_cross": first_cross(lam_eff),
 42        "delay": first_cross(lam_eff) - target_cross,
 43        "ts": ts, "desired": desired, "eff": eff,
 44    }
 45
 46
 47def linear_fit(x, y):
 48    x, y = np.asarray(x), np.asarray(y)
 49    coef = np.polyfit(x, y, 1)
 50    pred = np.polyval(coef, x)
 51    ssr = np.sum((y - pred) ** 2)
 52    sst = np.sum((y - y.mean()) ** 2)
 53    return float(coef[0]), float(coef[1]), float(1 - ssr / sst)
 54
 55
 56def math_verification():
 57    # Prediction 1: after ramp transients, theta_eff = theta_desired-r/kappa.
 58    kappa = 4.0
 59    t_end, dt = 20.0, 0.002
 60    out = simulate(0, 2, t_end, dt, kappa, 99, mode="baseline")
 61    r = 2 / t_end
 62    tail_error = np.mean((out["desired"][-1000:] - out["eff"][-1000:]))
 63    predicted_offset = r / kappa
 64
 65    # Prediction 2: threshold delay is (approximately) r/(kappa*slope).
 66    kappas = np.array([1., 2., 4., 8., 16., 32.])
 67    delays = np.array([simulate(0, 2, 20, .002, k, 1, mode="baseline")["delay"] for k in kappas])
 68    slope, intercept, r2 = linear_fit(1 / kappas, delays)
 69    predicted_slope = 1.0  # crossing delay=(r/kappa)/(|d theta_c/dt|)=1/kappa
 70
 71    # Prediction 3: inverse feed-forward reduces delay; printed proposal sign
 72    # is tested separately because for an increasing ramp it has the wrong sign.
 73    base = np.array([simulate(0, 2, 20, .002, k, 1, mode="baseline")["delay"] for k in kappas])
 74    minus = np.array([simulate(0, 2, 20, .002, k, 1, mode="minus")["delay"] for k in kappas])
 75    plus = np.array([simulate(0, 2, 20, .002, k, 1, mode="plus")["delay"] for k in kappas])
 76    return {
 77        "offset_observed": float(tail_error), "offset_predicted": predicted_offset,
 78        "delay_fit_slope": slope, "delay_fit_intercept": intercept,
 79        "delay_predicted_slope": predicted_slope, "delay_r2": r2,
 80        "kappas": kappas.tolist(), "baseline_delays": base.tolist(),
 81        "minus_delays": minus.tolist(), "plus_delays": plus.tolist(),
 82        "minus_reduction_fraction": float(1 - np.nanmean(np.abs(minus)) / np.nanmean(np.abs(base))),
 83        "plus_reduction_fraction": float(1 - np.nanmean(np.abs(plus)) / np.nanmean(np.abs(base))),
 84    }
 85
 86
 87def train_toy(seed=7, kappa=4.0, mode="baseline", steps=500):
 88    # 2-D binary classification; learning rate itself is EMA-filtered.
 89    rng = np.random.default_rng(seed)
 90    x0 = rng.normal([-0.7, -0.7], .8, (96, 2))
 91    x1 = rng.normal([0.7, 0.7], .8, (96, 2))
 92    X = np.vstack([x0, x1]); y = np.r_[np.zeros(96), np.ones(96)]
 93    w = np.zeros(2); b = 0.0; lr_eff = .01; lr0, lr1 = .01, .08
 94    losses = []
 95    dt = 1.0
 96    for step in range(steps):
 97        frac = min(step / 150, 1.0)
 98        desired = lr0 + (lr1 - lr0) * frac
 99        rate = (lr1 - lr0) / 150 if step < 150 else 0.0
100        if mode == "baseline": command = desired
101        elif mode == "minus": command = max(.0001, desired - rate / kappa)
102        elif mode == "plus": command = desired + rate / kappa
103        alpha = min(0.9, 0.1 * kappa)
104        lr_eff += alpha * (command - lr_eff)
105        z = X @ w + b
106        p = 1 / (1 + np.exp(-np.clip(z, -30, 30)))
107        loss = -np.mean(y*np.log(p+1e-8) + (1-y)*np.log(1-p+1e-8))
108        losses.append(float(loss))
109        gw = X.T @ (p-y) / len(y); gb = float(np.mean(p-y))
110        w -= lr_eff * gw; b -= lr_eff * gb
111        if not np.isfinite(loss) or np.linalg.norm(w) > 1e6:
112            return {"final_loss": float("inf"), "best_loss": float(np.min(losses)), "exploded": True}
113    return {"final_loss": losses[-1], "best_loss": float(np.min(losses)), "exploded": False}
114
115
116def main():
117    random.seed(7); np.random.seed(7)
118    math = math_verification()
119    training = {m: train_toy(mode=m, steps=300) for m in ["baseline", "minus", "plus"]}
120    result = {"math_verification": math, "training": training}
121    Path("results.json").write_text(json.dumps(result, indent=2))
122    print(json.dumps(result, indent=2))
123
124
125if __name__ == "__main__":
126    main()