import json, math, random from pathlib import Path import numpy as np # Reproducible MVP for PAC transition-cover monitoring. SEED = 2077 class PACCellMonitor: def __init__(self, n_cells, epsilon=0.1, delta=0.1): self.n_cells = n_cells self.epsilon = epsilon self.delta = delta self.required = int(math.ceil(math.log(1.0 / delta) / epsilon)) self.counts = np.zeros(n_cells, dtype=int) self.successes = np.zeros(n_cells, dtype=int) def observe(self, cell, hit): self.counts[cell] += 1 self.successes[cell] += int(hit) def deficient_cells(self): # A cell is not PAC-ready until it has the theorem's sample count. return np.flatnonzero(self.counts < self.required).tolist() def report(self): return {"required_n": self.required, "counts": self.counts.tolist(), "deficient": self.deficient_cells()} def math_sanity(rng): # Prediction 1: missed-region probability follows (1-eps)^n and its # exponential upper bound. Prediction 2: n >= log(1/delta)/eps crosses # below delta. Each row uses independent Bernoulli region hits. rows = [] reps = 40000 for eps in (0.02, 0.05, 0.10, 0.20): for n in (10, 25, 50, 100, 200): hits = rng.random((reps, n)) < eps observed = float(np.mean(~np.any(hits, axis=1))) exact = (1.0 - eps) ** n bound = math.exp(-n * eps) rows.append({"epsilon": eps, "n": n, "observed_miss": observed, "exact": exact, "exp_bound": bound, "abs_error_exact": abs(observed-exact)}) threshold_rows = [] for eps in (0.02, 0.05, 0.10, 0.20): delta = 0.1 n_req = math.ceil(math.log(1/delta) / eps) hits = rng.random((reps, n_req)) < eps observed = float(np.mean(~np.any(hits, axis=1))) threshold_rows.append({"epsilon": eps, "delta": delta, "n_required": n_req, "observed_miss": observed, "target_delta": delta, "exact_miss": (1-eps)**n_req}) max_err = max(x["abs_error_exact"] for x in rows) threshold_pass = all(x["observed_miss"] <= x["target_delta"] + 0.01 for x in threshold_rows) return {"scaling_rows": rows, "threshold_rows": threshold_rows, "max_abs_error_vs_exact": max_err, "threshold_pass": threshold_pass} def dynamics(x, rng): # A nonlinear noisy transition with a rare, dynamically important branch. # The branch is concentrated in cell 3 and is observable from x. cell = np.clip(((x + 1.0) / 2.0 * 8).astype(int), 0, 7) rare = (cell == 3) & (rng.random(len(x)) < 0.12) y = 0.72*x + 0.22*np.sin(3*x) + rng.normal(0, 0.025, len(x)) y = y + rare * (0.85 + 0.08*np.sin(5*x)) return np.clip(y, -1.5, 1.5), cell, rare def collect_uniform(rng, budget): x = rng.uniform(-1, 1, budget) y, cells, rare = dynamics(x, rng) return x, y, cells, rare def collect_pac(rng, budget, n_cells=8, epsilon=0.1, delta=0.1): mon = PACCellMonitor(n_cells, epsilon, delta) xs, ys, cs, rs = [], [], [], [] # Deficit-first allocation: this is the proposed extra-rollout policy. for t in range(budget): deficient = mon.deficient_cells() if deficient: c = deficient[t % len(deficient)] else: c = int(rng.integers(n_cells)) lo, hi = -1 + 2*c/n_cells, -1 + 2*(c+1)/n_cells x = np.array([rng.uniform(lo, hi)]) y, actual_cell, rare = dynamics(x, rng) # For this toy, a hit means observing the rare successor component; # ordinary transitions are also retained in the training buffer. mon.observe(c, bool(rare[0])) xs.append(x[0]); ys.append(y[0]); cs.append(c); rs.append(bool(rare[0])) return np.array(xs), np.array(ys), np.array(cs), np.array(rs), mon.report() def fit_and_score(x, y, rng, steps=350): # Small torch MLP; CPU fallback is automatic and CUDA is optional. try: import torch import torch.nn as nn device = "cuda" if torch.cuda.is_available() else "cpu" try: torch.manual_seed(SEED) model = nn.Sequential(nn.Linear(1, 24), nn.Tanh(), nn.Linear(24, 24), nn.Tanh(), nn.Linear(24, 1)).to(device) opt = torch.optim.Adam(model.parameters(), lr=0.008) X = torch.tensor(x[:,None], dtype=torch.float32, device=device) Y = torch.tensor(y[:,None], dtype=torch.float32, device=device) for _ in range(steps): idx = torch.randint(0, len(x), (min(128, len(x)),), device=device) loss = ((model(X[idx])-Y[idx])**2).mean() opt.zero_grad(); loss.backward(); opt.step() # Balanced test plus explicit rare-branch probe. tx = np.linspace(-1, 1, 1600) ty, tc, tr = dynamics(tx, np.random.default_rng(SEED+99)) with torch.no_grad(): pred = model(torch.tensor(tx[:,None], dtype=torch.float32, device=device)).cpu().numpy()[:,0] mse = float(np.mean((pred-ty)**2)) rare_mask = (tc == 3) & tr common_mask = (tc == 3) & ~tr rare_mse = float(np.mean((pred[rare_mask]-ty[rare_mask])**2)) if rare_mask.any() else float('nan') cell_mse = float(np.mean((pred[tc==3]-ty[tc==3])**2)) return {"test_mse": mse, "cell3_mse": cell_mse, "rare_branch_mse": rare_mse, "device": device} except Exception: pass except Exception: pass # Deterministic least-squares fallback if torch/CUDA fails. z = np.stack([x**k for k in range(6)], axis=1) coef = np.linalg.lstsq(z, y, rcond=None)[0] tx = np.linspace(-1,1,1600); ty,tc,tr = dynamics(tx, np.random.default_rng(SEED+99)) pred = np.stack([tx**k for k in range(6)], axis=1) @ coef return {"test_mse": float(np.mean((pred-ty)**2)), "cell3_mse": float(np.mean((pred[tc==3]-ty[tc==3])**2)), "rare_branch_mse": float(np.mean((pred[(tc==3)&tr]-ty[(tc==3)&tr])**2)), "device":"cpu-fallback"} def mini_experiment(rng): budget = 640 ux, uy, uc, ur = collect_uniform(rng, budget) px, py, pc, pr, report = collect_pac(rng, budget) baseline = fit_and_score(ux, uy, rng) idea = fit_and_score(px, py, rng) return {"budget": budget, "baseline": baseline, "idea": idea, "training_cell_counts": {"uniform": np.bincount(uc, minlength=8).tolist(), "pac": np.bincount(pc, minlength=8).tolist()}, "pac_monitor": report, "rare_samples": {"uniform": int(ur.sum()), "pac": int(pr.sum())}} def main(): random.seed(SEED); np.random.seed(SEED) rng = np.random.default_rng(SEED) out = {"seed": SEED, "math": math_sanity(rng), "mini_experiment": mini_experiment(rng)} Path("results.json").write_text(json.dumps(out, indent=2)) print(json.dumps(out, indent=2)) if __name__ == "__main__": main()