import json import numpy as np SEED = 1099 rng = np.random.default_rng(SEED) def kernel(x, y, ell=0.18, amp=1.0): x = np.asarray(x)[:, None] y = np.asarray(y)[None, :] return amp * np.exp(-0.5 * ((x-y)/ell)**2) def posterior_var(train_x, eval_x, noise=0.04, ell=0.18): train_x = np.asarray(train_x, dtype=float) eval_x = np.asarray(eval_x, dtype=float) K = kernel(train_x, train_x, ell) + noise * np.eye(len(train_x)) # solve rather than explicitly invert; this is the formula in the proposal cross = kernel(eval_x, train_x, ell) sol = np.linalg.solve(K, cross.T) v = np.diag(kernel(eval_x, eval_x, ell)) - np.sum(cross * sol.T, axis=1) return np.maximum(v, 0.0) def reduction(train_x, action, eval_x, weights): before = posterior_var(train_x, eval_x) after = posterior_var(np.r_[train_x, action], eval_x) return float(np.dot(weights, before-after)), before, after def make_candidates(n=9, batch=5): # Three overlapping clusters plus two isolated-ish alternatives. Every action # is a possible simulator trajectory/batch and has a spatial transition cost. centers = np.array([-0.72, -0.42, -0.10, 0.22, 0.55, 0.82]) actions = [] for c in centers: pts = np.linspace(c-0.10, c+0.10, batch) actions.append(np.clip(pts, -1, 1)) return centers, actions def tree_objective(train_x, actions, eval_x, weights, gamma=0.9, lam=0.0): # Exact shallow MCTS value: exhaustive tree is the small-MVP equivalent of # rollouts/UCB, and is deterministic for reproducibility. immediate = np.array([reduction(train_x, a, eval_x, weights)[0] for a in actions]) best_by_root = [] paths = [] for i, a in enumerate(actions): best = -np.inf; bestj = None for j, b in enumerate(actions): r2 = reduction(np.r_[train_x, a], b, eval_x, weights)[0] cost = abs(i-j) / max(1, len(actions)-1) val = immediate[i] + gamma * (r2 - lam*cost) if val > best: best, bestj = val, j best_by_root.append(best) paths.append((i, bestj)) root = int(np.argmax(best_by_root)) return root, paths[root], float(best_by_root[root]), immediate def overlap_sweep(): # Fixed first batch; construct second batches with controlled point overlap. train = np.array([-1.0, 1.0]) z = np.linspace(-1, 1, 401); w = np.ones(len(z))/len(z) a = np.linspace(-0.10, 0.10, 8) # Same candidate location but replace points with distant points to control overlap. far = np.linspace(0.62, 0.82, 8) rows=[] for frac in np.linspace(0, 1, 5): m = int(round(frac*len(a))) b = np.r_[a[:m], far[:len(a)-m]] # marginal reward after observing a r, _, _ = reduction(np.r_[train, a], b, z, w) rows.append((float(frac), r)) return rows def lambda_sweep(): train = np.array([-1.0, 1.0]) z = np.linspace(-1, 1, 401); w = np.ones(len(z))/len(z) centers, actions = make_candidates(); gamma = .9 # Every fixed depth-2 path has affine value A - B*lambda. Precompute these # lines once, then find the exact first root-policy transition. lines = {} for i in range(len(actions)): r1 = reduction(train, actions[i], z, w)[0] for j in range(len(actions)): r2 = reduction(np.r_[train, actions[i]], actions[j], z, w)[0] lines[(i,j)] = (r1 + gamma*r2, gamma*abs(i-j)/5) def policy(lam): vals=[] paths=[] for i in range(len(actions)): candidates=[(lines[(i,j)][0]-lam*lines[(i,j)][1],j) for j in range(len(actions))] val,j=max(candidates) vals.append(val); paths.append((i,j)) root=int(np.argmax(vals)) return root,paths[root] grid=np.linspace(0,2,401) chosen=[policy(x)[0] for x in grid] changes=np.where(np.diff(chosen)!=0)[0] if len(changes): idx=int(changes[0]); observed=(idx+0.5)*2/400 old_path=policy(max(0,observed-1e-7))[1] new_path=policy(min(2,observed+1e-7))[1] a0,a1=lines[old_path]; b0,b1=lines[new_path] predicted=(a0-b0)/(a1-b1) if abs(a1-b1)>1e-12 else None return {'observed_threshold':float(observed),'predicted_threshold':None if predicted is None else float(predicted), 'old_root':int(chosen[idx]),'new_root':int(chosen[idx+1]),'num_changes':int(len(changes))} return {'observed_threshold':None,'predicted_threshold':None,'num_changes':0} def main(): # Core numerical formula sanity check: conditioning cannot increase variance. train=np.array([-0.85,-0.25,0.35,0.9]); z=np.linspace(-1,1,301) v0=posterior_var(train,z); v1=posterior_var(np.r_[train,0.0],z) max_increase=float(np.max(v1-v0)) overlap=overlap_sweep() # Check monotonic diminishing returns and report endpoint ratio. overlap_monotone=all(overlap[i+1][1] <= overlap[i][1]+1e-10 for i in range(len(overlap)-1)) lam=lambda_sweep() # Mini comparison: random, greedy, and planned tree, two executed batches. train0=np.array([-1.0,1.0]); z=np.linspace(-1,1,401); w=np.ones(len(z))/len(z) centers, actions=make_candidates(); gamma=.9 def run(method): tr=train0.copy(); total=0.; roots=[] for t in range(2): if method=='random': i=int(rng.integers(len(actions))) elif method=='greedy': i=int(np.argmax([reduction(tr,a,z,w)[0] for a in actions])) else: i=tree_objective(tr,actions,z,w,gamma,lam=0.8)[0] total += reduction(tr,actions[i],z,w)[0]; tr=np.r_[tr,actions[i]]; roots.append(i) return total,roots, float(np.mean(posterior_var(tr,z))) cmp={m:run(m) for m in ['random','greedy','tree']} out={'seed':SEED,'math_check':{'max_variance_increase':max_increase, 'nonincrease_verified':max_increase < 1e-9}, 'prediction_1':{'claim':'posterior variance never increases after conditioning', 'predicted':'max increase <= numerical tolerance 1e-9', 'observed':max_increase}, 'prediction_2':{'claim':'overlap makes later marginal reduction smaller', 'overlap_fraction_and_reduction':overlap,'monotone_verified':overlap_monotone, 'endpoint_reduction_ratio':overlap[-1][1]/max(overlap[0][1],1e-12)}, 'prediction_3':{'claim':'tree root changes at the analytic reward/cost crossover',**lam}, 'comparison':{k:{'cumulative_reduction':v[0],'actions':v[1],'final_mean_variance':v[2]} for k,v in cmp.items()}} with open('results.json','w') as f: json.dump(out,f,indent=2) print(json.dumps(out,indent=2)) if __name__=='__main__': main()