Complementary-Channel Switched Latent Observer / experiment.py

Failed on benchmark

Raw ⬇ ZIP
  1import json
  2import numpy as np
  3from scipy.linalg import expm
  4
  5SEED = 2080
  6A = np.diag([0.4, 0.4])
  7C1 = np.array([[1.0, 0.0]])
  8C2 = np.array([[0.0, 1.0]])
  9I = np.eye(2)
 10
 11
 12def obs(A, C):
 13    n = A.shape[0]
 14    return np.vstack([C @ np.linalg.matrix_power(A, k) for k in range(n)])
 15
 16
 17def matrices(gain, tau1=0.5, tau2=0.5):
 18    L1 = np.array([[gain], [0.0]])
 19    L2 = np.array([[0.0], [gain]])
 20    F1 = A - L1 @ C1
 21    F2 = A - L2 @ C2
 22    P1, P2 = expm(F1 * tau1), expm(F2 * tau2)
 23    return F1, F2, P1, P2, P2 @ P1
 24
 25
 26def spectral_radius(M):
 27    return float(np.max(np.abs(np.linalg.eigvals(M))))
 28
 29
 30def exact_boundary():
 31    # For this diagonal complementary system, rho(Phi)=exp((.4-g) tau)
 32    # and the predicted boundary is gain=.8 for equal dwell time.
 33    gains = np.linspace(0.0, 1.2, 241)
 34    values = np.array([spectral_radius(matrices(g)[-1]) for g in gains])
 35    crossing = gains[np.argmin(np.abs(values - 1.0))]
 36    return crossing, gains, values
 37
 38
 39def empirical_decay(gain, tau=0.5, cycles=25):
 40    Phi = matrices(gain, tau, tau)[-1]
 41    e = np.array([1.0, -0.7])
 42    norms = []
 43    for _ in range(cycles):
 44        norms.append(np.linalg.norm(e))
 45        e = Phi @ e
 46    norms = np.maximum(np.asarray(norms), 1e-300)
 47    slope = np.polyfit(np.arange(cycles), np.log(norms), 1)[0]
 48    return slope, norms[-1]
 49
 50
 51def noisy_switched(gain, tau=0.5, noise=0.15, cycles=80, trials=200):
 52    # Discrete exact observer with measurement noise. For each dwell, use
 53    # midpoint-equivalent correction z-C h at the start of the dwell.
 54    rng = np.random.default_rng(SEED + 9)
 55    F1, F2, P1, P2, _ = matrices(gain, tau, tau)
 56    L1 = np.array([[gain], [0.0]])
 57    L2 = np.array([[0.0], [gain]])
 58    # Integral exp(F(t-s)) L ds, computed robustly by augmented exponential.
 59    def noise_map(F, L):
 60        aug = np.zeros((3, 3)); aug[:2, :2] = F; aug[:2, 2:] = L
 61        return expm(aug * tau)[:2, 2:]
 62    Q1, Q2 = noise_map(F1, L1), noise_map(F2, L2)
 63    vals = []
 64    for _ in range(trials):
 65        e = np.array([1.0, -0.7])
 66        for _ in range(cycles):
 67            e = P1 @ e + Q1 @ rng.normal(0, noise, size=(1,))
 68            e = P2 @ e + Q2 @ rng.normal(0, noise, size=(1,))
 69        vals.append(np.dot(e, e))
 70    return float(np.mean(vals))
 71
 72
 73def residual_fusion(gain, tau=0.5, noise=0.15, cycles=80, trials=200):
 74    # Baseline: each observed coordinate is independently corrected, but its
 75    # residual is held at zero during the missing-channel dwell (ordinary
 76    # modality-specific residual fusion, no cycle-aware gain design).
 77    rng = np.random.default_rng(SEED + 9)
 78    F1, F2, P1, P2, _ = matrices(gain, tau, tau)
 79    L1 = np.array([[gain], [0.0]]); L2 = np.array([[0.0], [gain]])
 80    def noise_map(F, L):
 81        aug = np.zeros((3, 3)); aug[:2, :2] = F; aug[:2, 2:] = L
 82        return expm(aug * tau)[:2, 2:]
 83    Q1, Q2 = noise_map(F1, L1), noise_map(F2, L2)
 84    vals = []
 85    for _ in range(trials):
 86        e = np.array([1.0, -0.7])
 87        for _ in range(cycles):
 88            # baseline uses only currently available channel but resets the
 89            # unobserved residual branch; this is equivalent here to no
 90            # complementary correction during that half-cycle.
 91            e = P1 @ e + Q1 @ rng.normal(0, noise, size=(1,))
 92            e = expm(A * tau) @ e + Q2 @ rng.normal(0, noise, size=(1,))
 93        vals.append(np.dot(e, e))
 94    return float(np.mean(vals))
 95
 96
 97def main():
 98    O1, O2 = obs(A, C1), obs(A, C2)
 99    Ostack = np.vstack([O1, O2])
100    crossing, gains, rhos = exact_boundary()
101    # Prediction 1: equal-dwell stability boundary at gain=1.2.
102    boundary = {"predicted": 0.8, "observed_grid": float(crossing),
103                "rho_at_0_7": float(spectral_radius(matrices(.7)[-1])),
104                "rho_at_0_9": float(spectral_radius(matrices(.9)[-1]))}
105    # Prediction 2: log decay slope equals log spectral radius per cycle.
106    decay = []
107    for g in (0.9, 1.0, 1.2):
108        slope, final = empirical_decay(g)
109        predicted = np.log(spectral_radius(matrices(g)[-1]))
110        decay.append({"gain": g, "predicted_log_slope": float(predicted),
111                      "observed_log_slope": float(slope), "final_norm": float(final)})
112    # Prediction 3: stable switched cycle can tolerate an individually expanding
113    # channel; scan dwell ratios while keeping gain=1.2.
114    dwell = []
115    for t1, t2 in ((0.2,0.8),(0.3,0.7),(0.4,0.6),(0.5,0.5),(0.6,0.4),(0.7,0.3),(0.8,0.2)):
116        F1,F2,P1,P2,Phi = matrices(1.2,t1,t2)
117        dwell.append({"tau1":t1,"tau2":t2,"rho_cycle":spectral_radius(Phi),
118                      "rho_channel1":spectral_radius(P1),"rho_channel2":spectral_radius(P2)})
119    gain = 1.2
120    comparison = {"switched_mse": noisy_switched(gain),
121                  "residual_fusion_mse": residual_fusion(gain),
122                  "gain": gain}
123    out = {"observability_rank_channel1": int(np.linalg.matrix_rank(O1)),
124           "observability_rank_channel2": int(np.linalg.matrix_rank(O2)),
125           "observability_rank_stacked": int(np.linalg.matrix_rank(Ostack)),
126           "predictions": {"boundary": boundary, "decay": decay, "dwell_sweep": dwell},
127           "comparison": comparison}
128    with open("results.json", "w") as f: json.dump(out, f, indent=2)
129    print(json.dumps(out, indent=2))
130
131if __name__ == "__main__": main()