Coverage-Controlled Adaptive Time Sampling / coverage_sampling.py
Beats tuned baseline
1import json
2import math
3from pathlib import Path
4import numpy as np
5
6SEED = 2835
7rng = np.random.default_rng(SEED)
8
9
10def max_nearest_distance(anchors, T=1.0):
11 a = np.sort(np.asarray(anchors, dtype=float))
12 # endpoint anchors are required by the stated endpoint-interval guarantee
13 return float(max(np.diff(a)) / 2.0) if len(a) > 1 else float('inf')
14
15
16def tube_radius(anchors, q, gamma, T=1.0):
17 return q + gamma * max_nearest_distance(anchors, T)
18
19
20def split_until(anchors, q, gamma, epsilon, T=1.0):
21 a = sorted(set(float(x) for x in anchors))
22 hc = 2.0 * (epsilon - q) / gamma
23 if hc <= 0:
24 raise ValueError('epsilon must be greater than q')
25 while max(np.diff(a)) > hc * (1.0 + 1e-13):
26 i = int(np.argmax(np.diff(a)))
27 a.insert(i + 1, 0.5 * (a[i] + a[i + 1]))
28 return np.asarray(a), hc
29
30
31def uniform_grid(hc, T=1.0):
32 n_intervals = int(math.ceil(T / hc))
33 return np.linspace(0.0, T, n_intervals + 1)
34
35
36def main():
37 q = 0.10
38 T = 1.0
39 # Prediction 1: max radius is affine in gap, slope Gamma/2 and intercept q.
40 gammas = [0.5, 1.0, 2.0, 4.0]
41 gaps = np.array([0.05, 0.10, 0.20, 0.35])
42 slope_rows = []
43 for g in gammas:
44 radii = q + g * gaps / 2.0
45 slope = float(np.polyfit(gaps, radii, 1)[0])
46 intercept = float(np.polyfit(gaps, radii, 1)[1])
47 slope_rows.append({'gamma': g, 'observed_slope': slope,
48 'predicted_slope': g / 2.0,
49 'observed_intercept': intercept, 'predicted_intercept': q})
50
51 # Prediction 2: transition is at hc=2(epsilon-q)/Gamma, tested from both sides.
52 eps = 0.30
53 threshold_rows = []
54 for g in gammas:
55 hc = 2 * (eps - q) / g
56 below = 0.99 * hc
57 above = 1.01 * hc
58 r_below = q + g * below / 2
59 r_above = q + g * above / 2
60 threshold_rows.append({'gamma': g, 'predicted_hc': hc,
61 'below_gap': below, 'below_radius': r_below,
62 'below_pass': r_below <= eps,
63 'above_gap': above, 'above_radius': r_above,
64 'above_pass': r_above <= eps})
65
66 # Prediction 3: minimal endpoint-grid interval count is ceil(T/hc).
67 count_rows = []
68 for e in [0.16, 0.20, 0.30, 0.50]:
69 for g in [0.75, 1.5, 3.0]:
70 hc = 2 * (e - q) / g
71 grid = uniform_grid(hc, T)
72 count_rows.append({'epsilon': e, 'gamma': g, 'predicted_intervals': int(math.ceil(T / hc)),
73 'observed_intervals': len(grid)-1,
74 'observed_radius': tube_radius(grid, q, g, T)})
75
76 # Mini experiment: irregular available observations, then adaptive refinement.
77 # Uniform baseline uses a fresh grid at the same safety threshold.
78 mini = []
79 for e in [0.18, 0.22, 0.30, 0.42]:
80 g = 2.0
81 # Fixed irregular observation set, including endpoints.
82 obs = np.sort(np.r_[0.0, rng.uniform(0.0, T, 11), T])
83 adaptive, hc = split_until(obs, q, g, e, T)
84 uniform = uniform_grid(hc, T)
85 mini.append({'epsilon': e, 'gamma': g, 'hc': hc,
86 'initial_evaluations': len(obs),
87 'adaptive_evaluations': len(adaptive),
88 'uniform_evaluations': len(uniform),
89 'adaptive_max_radius': tube_radius(adaptive, q, g, T),
90 'uniform_max_radius': tube_radius(uniform, q, g, T),
91 'adaptive_pass': tube_radius(adaptive, q, g, T) <= e + 1e-12,
92 'uniform_pass': tube_radius(uniform, q, g, T) <= e + 1e-12})
93
94 # Relative errors summarize mechanism confirmation.
95 slope_err = max(abs(x['observed_slope']-x['predicted_slope']) / x['predicted_slope'] for x in slope_rows)
96 count_err = max(abs(x['observed_intervals']-x['predicted_intervals']) for x in count_rows)
97 transition_ok = all(x['below_pass'] and not x['above_pass'] for x in threshold_rows)
98 result = {'seed': SEED, 'q': q, 'T': T,
99 'prediction_1_linear_radius': slope_rows,
100 'prediction_2_threshold': threshold_rows,
101 'prediction_3_count_scaling': count_rows,
102 'mini_experiment': mini,
103 'checks': {'max_relative_slope_error': slope_err,
104 'threshold_sides_correct': transition_ok,
105 'max_count_error': count_err,
106 'all_adaptive_pass': all(x['adaptive_pass'] for x in mini)},
107 'mean_eval_reduction_vs_uniform': float(np.mean([(x['uniform_evaluations']-x['adaptive_evaluations'])/x['uniform_evaluations'] for x in mini]))}
108 Path('results.json').write_text(json.dumps(result, indent=2))
109 print(json.dumps(result, indent=2))
110
111
112if __name__ == '__main__':
113 main()