Feasibility-Preserving Error Compensator / run_experiment.py
Failed on benchmark
1import json
2from pathlib import Path
3import numpy as np
4
5SEED = 2907
6
7
8def aug_matrix(A, B, KC, KI, rho):
9 return np.array([[A, -B * KC], [KI, rho]], dtype=float)
10
11
12def spectral_radius(M):
13 return float(np.max(np.abs(np.linalg.eigvals(M))))
14
15
16def unsaturated_decay(A, B, KC, KI, rho, n=100):
17 M = aug_matrix(A, B, KC, KI, rho)
18 v = np.array([1.0, 0.0])
19 norms = []
20 for _ in range(n):
21 norms.append(float(np.linalg.norm(v)))
22 v = M @ v
23 # Dominant asymptotic factor estimated from log(norm) after transients.
24 fit = np.polyfit(np.arange(25, n), np.log(np.maximum(np.asarray(norms[25:]), 1e-15)), 1)
25 return spectral_radius(M), float(np.exp(fit[0]))
26
27
28def stability_sweep():
29 A, B, KC, rho = 0.82, 1.0, 0.32, 0.68
30 kis = np.linspace(0.0, 1.8, 181)
31 rows = []
32 for ki in kis:
33 sr = spectral_radius(aug_matrix(A, B, KC, ki, rho))
34 M = aug_matrix(A, B, KC, ki, rho)
35 v = np.array([1.0, 0.0])
36 for _ in range(500):
37 v = M @ v
38 # A deliberately strict finite-horizon divergence test, not the theory.
39 empirical_stable = bool(np.linalg.norm(v) < 1e-3)
40 rows.append((float(ki), sr, empirical_stable))
41 stable = [x for x in rows if x[2]]
42 empirical_boundary = max(x[0] for x in stable) if stable else float('nan')
43 crossing = [rows[i] for i in range(1, len(rows))
44 if rows[i-1][1] < 1.0 <= rows[i][1]]
45 predicted_boundary = crossing[0][0] if crossing else float('nan')
46 return {
47 'A': A, 'B': B, 'KC': KC, 'rho': rho,
48 'predicted_KI_boundary_rho_eq_1': predicted_boundary,
49 'observed_KI_boundary_500_step_state_decay_below_1e-3': empirical_boundary,
50 'boundary_abs_difference': abs(predicted_boundary - empirical_boundary),
51 'spectral_stability_classification_agreement': float(np.mean([x[2] == (x[1] < 1.0) for x in rows])),
52 }
53
54
55def saturation_episode(ki, ka, sat=0.25, n=180, hold_end=95, rho=0.92, kp=0.28, kc=0.55):
56 y, c = 0.0, 0.0
57 ys, rs, cs, es = [], [], [], []
58 for t in range(n):
59 yd = 1.0 if 10 <= t < hold_end else 0.0
60 e = yd - y
61 tilde = kp * e + kc * c # nominal neural reference z_t is zero
62 r = float(np.clip(tilde, -sat, sat))
63 d = r - tilde
64 y = 0.92 * y + 0.08 * r
65 c = rho * c + ki * e + ka * d
66 ys.append(y); rs.append(r); cs.append(c); es.append(e)
67 es, cs, rs = np.asarray(es), np.asarray(cs), np.asarray(rs)
68 recovery = None
69 for t in range(hold_end, n):
70 if np.max(np.abs(es[t:min(t + 8, n)])) < 0.1:
71 recovery = t - hold_end
72 break
73 return {
74 'rms_error': float(np.sqrt(np.mean(es ** 2))),
75 'post_release_rms_error': float(np.sqrt(np.mean(es[hold_end:] ** 2))),
76 'max_abs_c': float(np.max(np.abs(cs))),
77 'release_abs_c': float(abs(cs[hold_end])),
78 'recovery_steps': recovery if recovery is not None else n - hold_end,
79 'max_constraint_violation': float(np.max(np.maximum(np.abs(rs) - sat, 0.0))),
80 'terminal_abs_error': float(abs(es[-1])),
81 }
82
83
84def saturation_sweep():
85 durations = np.array([10, 25, 40, 55, 70])
86 ordinary = [saturation_episode(.075, 0.0, hold_end=10 + int(d)) for d in durations]
87 antiwindup = [saturation_episode(.075, .9, hold_end=10 + int(d)) for d in durations]
88 ordinary_c = np.array([x['release_abs_c'] for x in ordinary])
89 anti_c = np.array([x['release_abs_c'] for x in antiwindup])
90 # Prediction: residual feedback lowers the slope of windup versus saturation duration.
91 ordinary_slope = float(np.polyfit(durations, ordinary_c, 1)[0])
92 anti_slope = float(np.polyfit(durations, anti_c, 1)[0])
93 return {
94 'durations': durations.tolist(), 'ordinary_integral': ordinary,
95 'residual_antiwindup': antiwindup,
96 'predicted_ordering': 'antiwindup slope < ordinary slope',
97 'observed_release_c_slope_ordinary': ordinary_slope,
98 'observed_release_c_slope_antiwindup': anti_slope,
99 'observed_slope_ratio_anti_over_ordinary': anti_slope / ordinary_slope,
100 }
101
102
103def unsaturated_sweep():
104 A, B, KC, rho = .80, 1.0, .30, .70
105 kis = [.02, .08, .16, .24]
106 rows = []
107 for ki in kis:
108 predicted, observed = unsaturated_decay(A, B, KC, ki, rho)
109 rows.append({'KI': ki, 'predicted_spectral_radius': predicted,
110 'observed_asymptotic_decay_factor': observed,
111 'relative_error': abs(observed-predicted)/predicted})
112 return rows
113
114
115def main():
116 sat = saturation_sweep()
117 result = {
118 'seed': SEED,
119 'math_prediction_1_unsaturated_decay': unsaturated_sweep(),
120 'math_prediction_2_stability_boundary': stability_sweep(),
121 'math_prediction_3_saturation_windup_scaling': sat,
122 'mini_experiment_baseline_no_compensation': saturation_episode(0.0, 0.0),
123 'mini_experiment_ordinary_integral': saturation_episode(.075, 0.0),
124 'mini_experiment_idea_residual_antiwindup': saturation_episode(.075, .9),
125 }
126 Path('results.json').write_text(json.dumps(result, indent=2))
127 print(json.dumps(result, indent=2))
128
129if __name__ == '__main__':
130 main()