Joint latent-actuator identification / experiment.py

✓✓ Beats tuned baseline

Raw ⬇ ZIP
  1import json
  2from pathlib import Path
  3import numpy as np
  4from scipy.optimize import least_squares
  5
  6SEED = 1411
  7TRUE_A = 0.70
  8TRUE_B = 1.20
  9TRUE_TAU = 0.80
 10NOISE = 0.01
 11
 12
 13def disturbance(u, alpha, tau):
 14    return alpha * np.tanh(u / tau)
 15
 16
 17def make_data(alpha, n, u_limit, seed, tau=TRUE_TAU):
 18    rng = np.random.default_rng(seed)
 19    x = rng.uniform(-1.0, 1.0, n)
 20    u = rng.uniform(-u_limit, u_limit, n)
 21    y = TRUE_A * x + TRUE_B * (u + disturbance(u, alpha, tau))
 22    y += rng.normal(0.0, NOISE, n)
 23    return x, u, y
 24
 25
 26def fit_baseline(x, u, y):
 27    X = np.column_stack([x, u, np.ones_like(x)])
 28    p, *_ = np.linalg.lstsq(X, y, rcond=None)
 29    return p
 30
 31
 32def fit_joint(x, u, y, tau_hint=TRUE_TAU):
 33    def residual(p):
 34        a, b, alpha, tau, c = p
 35        pred = a * x + b * (u + disturbance(u, alpha, tau)) + c
 36        return (pred - y) / NOISE
 37    p0 = np.array([0.6, 1.1, 0.15, tau_hint, 0.0])
 38    lo = np.array([-3., -3., -2., 0.15, -1.])
 39    hi = np.array([3., 3., 2., 3.0, 1.])
 40    result = least_squares(residual, p0, bounds=(lo, hi), max_nfev=1000)
 41    return result.x
 42
 43
 44def mse(y, pred):
 45    return float(np.mean((y - pred) ** 2))
 46
 47
 48def main():
 49    # Prediction 1: max_u |d'(u)| = |alpha|/tau, attained at u=0.
 50    grid = np.linspace(-8, 8, 20001)
 51    alpha, tau = 0.37, 0.63
 52    analytic = alpha / tau / np.cosh(grid / tau) ** 2
 53    numeric = np.gradient(disturbance(grid, alpha, tau), grid)
 54    derivative_check = {
 55        "predicted_max": float(abs(alpha) / tau),
 56        "observed_max": float(np.max(np.abs(numeric))),
 57        "relative_error": float(abs(np.max(np.abs(numeric)) - abs(alpha) / tau) / (abs(alpha) / tau)),
 58        "bound_location": float(grid[np.argmax(np.abs(numeric))]),
 59        "analytic_numeric_max_gap": float(np.max(np.abs(analytic - numeric))),
 60    }
 61
 62    # Prediction 2: for small alpha, omitted-variable bias in the linear action
 63    # coefficient is approximately B*alpha*E[u*tanh(u/tau)]/E[u^2], hence linear in alpha.
 64    alphas = np.array([0.0, 0.05, 0.10, 0.20, 0.40, 0.80])
 65    rows = []
 66    for i, a in enumerate(alphas):
 67        x, u, y = make_data(a, 5000, 1.5, SEED + i)
 68        pb = fit_baseline(x, u, y)
 69        pjoint = fit_joint(x, u, y)
 70        # Test on a fresh broad command range to expose saturation extrapolation.
 71        xt, ut, yt = make_data(a, 3000, 2.2, SEED + 100 + i)
 72        pred_b = pb[0] * xt + pb[1] * ut + pb[2]
 73        pred_j = pjoint[0] * xt + pjoint[1] * (ut + disturbance(ut, pjoint[2], pjoint[3])) + pjoint[4]
 74        ratio = np.mean(u * np.tanh(u / TRUE_TAU)) / np.mean(u * u)
 75        predicted_bias = TRUE_B * a * ratio
 76        rows.append({
 77            "alpha": float(a),
 78            "predicted_linear_bias": float(predicted_bias),
 79            "observed_baseline_b_bias": float(pb[1] - TRUE_B),
 80            "joint_train_mse": mse(y, pjoint[0] * x + pjoint[1] * (u + disturbance(u, pjoint[2], pjoint[3])) + pjoint[4]),
 81            "baseline_cross_regime_mse": mse(yt, pred_b),
 82            "joint_cross_regime_mse": mse(yt, pred_j),
 83            "fitted_alpha": float(pjoint[2]),
 84            "fitted_tau": float(pjoint[3]),
 85        })
 86
 87    # Prediction 3: no actuator disturbance (alpha=0) removes the mechanism;
 88    # the two models should have comparable cross-regime error.
 89    zero_rows = []
 90    for i in range(8):
 91        x, u, y = make_data(0.0, 2000, 1.5, SEED + 300 + i)
 92        pb = fit_baseline(x, u, y)
 93        pj = fit_joint(x, u, y)
 94        xt, ut, yt = make_data(0.0, 2000, 2.2, SEED + 400 + i)
 95        zero_rows.append({
 96            "baseline_mse": mse(yt, pb[0] * xt + pb[1] * ut + pb[2]),
 97            "joint_mse": mse(yt, pj[0] * xt + pj[1] * (ut + disturbance(ut, pj[2], pj[3])) + pj[4]),
 98            "fitted_alpha": float(pj[2]),
 99        })
100
101    # Fit slopes through the origin for the small-alpha predicted/observed bias.
102    small = rows[:4]
103    alpha_small = np.array([r["alpha"] for r in small])
104    observed_bias_small = np.array([r["observed_baseline_b_bias"] for r in small])
105    predicted_bias_small = np.array([r["predicted_linear_bias"] for r in small])
106    observed_slope = float(np.dot(alpha_small, observed_bias_small) / np.dot(alpha_small, alpha_small))
107    predicted_slope = float(np.dot(alpha_small, predicted_bias_small) / np.dot(alpha_small, alpha_small))
108    zero_base = float(np.mean([r["baseline_mse"] for r in zero_rows]))
109    zero_joint = float(np.mean([r["joint_mse"] for r in zero_rows]))
110    output = {
111        "seed": SEED,
112        "derivative_bound_check": derivative_check,
113        "bias_scaling_prediction": {
114            "predicted_slope": predicted_slope,
115            "observed_slope": observed_slope,
116            "slope_ratio_observed_over_predicted": observed_slope / predicted_slope,
117            "rows": rows,
118        },
119        "zero_disturbance_prediction": {
120            "mean_baseline_mse": zero_base,
121            "mean_joint_mse": zero_joint,
122            "joint_over_baseline": zero_joint / zero_base,
123            "rows": zero_rows,
124        },
125        "interpretation": "Joint model should recover saturation and improve held-out command-regime error when alpha is nonzero; at alpha=0 no systematic benefit is expected.",
126    }
127    Path("results.json").write_text(json.dumps(output, indent=2))
128    print(json.dumps(output, indent=2))
129
130
131if __name__ == "__main__":
132    main()