Update practicality_axioms.py

practicality_axioms.py

@@ -325,6 +325,124 @@ def generate_templates(problem: core.Problem, sketch: Dict[str, Any]) -> List[Hy
                 )
     return templates
 
+# ══════════════════════════════════════════════════════════════════════
+# MATH ENGINE METHODS (Explicitly implemented inside local scope)
+# ══════════════════════════════════════════════════════════════════════
+def _batched_deduce_and_evaluate(problem: core.Problem, hyps: List[Hypothesis], steps: int=80) -> List[Tuple[Dict, float, List[str], str]]:
+    if not hyps: return []
+    
+    skip_indices = [i for i, h in enumerate(hyps) if getattr(h, 'is_fully_determined', False) or len(h.free_vars) == 0]
+    solve_indices = [i for i, h in enumerate(hyps) if i not in skip_indices]
+    results = [None] * len(hyps)
+
+    for i in skip_indices:
+        hyp = hyps[i]
+        try:
+            ce = problem.scalar_energy(hyp.binding)
+            dom_vars = list(hyp.pinned_vars.keys())[:3]
+            results[i] = (hyp.binding, ce, dom_vars, "algebraic")
+        except: results[i] = (hyp.binding, float('inf'), [], "algebraic_error")
+
+    if not solve_indices: return [r for r in results if r is not None]
+
+    adam_hyps = [hyps[i] for i in solve_indices]
+    V = len(problem.variables)
+    
+    log_mask = []
+    log_lo, log_hi = [], []
+    for v in problem.variables:
+        lo, hi = core._c15(problem.bounds[v][0]), core._c15(problem.bounds[v][1])
+        if v in problem.log_space_vars and lo > 0:
+            log_mask.append(True)
+            log_lo.append(math.log10(max(lo, 1e-30)))
+            log_hi.append(math.log10(max(hi, 1e-30)))
+        else:
+            log_mask.append(False)
+            log_lo.append(lo)
+            log_hi.append(hi)
+
+    log_mask_t = torch.tensor(log_mask, device=core.DEVICE, dtype=torch.bool)
+    lo_param_t = torch.tensor(log_lo, device=core.DEVICE, dtype=torch.float32)
+    hi_param_t = torch.tensor(log_hi, device=core.DEVICE, dtype=torch.float32)
+
+    def _param_to_orig(P):
+        orig = P.clone()
+        if log_mask_t.any(): orig[:, log_mask_t] = torch.pow(10.0, P[:, log_mask_t])
+        return orig
+
+    def _orig_to_param(x_val, j):
+        if log_mask[j] and x_val > 0: return math.log10(max(x_val, 1e-30))
+        return x_val
+
+    x_data_p, mask_data, target_data_p = [], [], []
+    for hyp in adam_hyps:
+        xr, mr, tr = [], [], []
+        active_vars = problem.get_markov_blanket(set(hyp.pinned_vars.keys()), depth=2)
+        for j, v in enumerate(problem.variables):
+            lo, hi = core._c15(problem.bounds[v][0]), core._c15(problem.bounds[v][1])
+            if v in hyp.pinned_vars:
+                p_val = _orig_to_param(core._c15(hyp.pinned_vars[v]), j)
+                xr.append(p_val); mr.append(0.0); tr.append(p_val)
+            else:
+                p_val = _orig_to_param(core._c15(hyp.binding.get(v, (lo+hi)/2)), j)
+                is_active = (v in active_vars) or (len(hyp.pinned_vars) == 0)
+                xr.append(p_val); mr.append(1.0 if is_active else 0.0); tr.append(0.0)
+        x_data_p.append(xr); mask_data.append(mr); target_data_p.append(tr)
+
+    P = torch.tensor(x_data_p, device=core.DEVICE, dtype=torch.float32, requires_grad=True)
+    mask = torch.tensor(mask_data, device=core.DEVICE, dtype=torch.float32)
+    target = torch.tensor(target_data_p, device=core.DEVICE, dtype=torch.float32)
+
+    optimizer = torch.optim.Adam([P], lr=0.01)
+
+    for step in range(steps):
+        optimizer.zero_grad()
+        step_ratio = min(1.0, step / (steps * 0.8))
+        X_orig = _param_to_orig(P)
+        ce = problem.tensor_energy(X_orig, step_ratio, is_optimizing=True)
+        if isinstance(ce, torch.Tensor) and (ce < core.SOLVE_THRESHOLD).all() and step_ratio == 1.0: break
+        ce.sum().backward()
+        with torch.no_grad():
+            P.grad.clamp_(-10.0, 10.0)
+            P.grad *= mask
+            optimizer.step()
+            P.data = torch.where(mask == 0.0, target, P.data)
+            margin = 0.1 * (1.0 - step_ratio)
+            lo_m = lo_param_t - (hi_param_t - lo_param_t) * margin
+            hi_m = hi_param_t + (hi_param_t - lo_param_t) * margin
+            P.data = torch.clamp(P.data, lo_m.unsqueeze(0), hi_m.unsqueeze(0))
+
+    X_orig_final = _param_to_orig(P)
+    final_ce = problem.tensor_energy(X_orig_final, 1.0, is_optimizing=False).view(-1)
+    ce_vals = final_ce.detach().cpu().numpy()
+    X_vals = X_orig_final.detach().cpu().numpy()
+
+    for b_idx, orig_idx in enumerate(solve_indices):
+        final_b = {problem.variables[j]: float(X_vals[b_idx, j]) for j in range(V)}
+        results[orig_idx] = (final_b, float(ce_vals[b_idx]), [], "systemic")
+
+    return [r for r in results if r is not None]
+
+def _mprt_sample(problem: core.Problem, N: int):
+    var_list = problem.variables; V = len(var_list)
+    lo_t = torch.tensor([core._c15(problem.bounds.get(v, (-10.0, 10.0))[0]) for v in var_list], device=core.DEVICE, dtype=torch.float32)
+    hi_t = torch.tensor([core._c15(problem.bounds.get(v, (-10.0, 10.0))[1]) for v in var_list], device=core.DEVICE, dtype=torch.float32)
+    for i in range(V):
+        if lo_t[i] >= hi_t[i]: m = (lo_t[i]+hi_t[i])/2; lo_t[i] = m - 1e-6; hi_t[i] = m + 1e-6
+    
+    rand_base = torch.rand((N, V), device=core.DEVICE)
+    lsv_indices = [problem.var_idx[v] for v in problem.log_space_vars if v in problem.var_idx]
+    for idx in lsv_indices:
+        lo_v, hi_v = lo_t[idx].item(), hi_t[idx].item()
+        if lo_v > 0 and hi_v > lo_v:
+            log_lo, log_hi = math.log10(max(lo_v, 1e-30)), math.log10(max(hi_v, 1e-30))
+            rand_base[:, idx] = torch.pow(10.0, torch.rand(N, device=core.DEVICE)*(log_hi-log_lo)+log_lo) / hi_v
+            
+    X = lo_t.unsqueeze(0) + (hi_t - lo_t).unsqueeze(0) * rand_base
+    ce_batch = problem.tensor_energy(X, 1.0, is_optimizing=False).view(-1)
+    best_idx = torch.argmin(ce_batch).item()
+    return {v: float(X[best_idx, i].item()) for i, v in enumerate(var_list)}, ce_batch[best_idx].item()
+
 # ══════════════════════════════════════════════════════════════════════
 # THE DECOUPLED SEQUENCE RAY TRACER
 # ══════════════════════════════════════════════════════════════════════
@@ -466,8 +584,4 @@ class UCB1BanditSeeder:
             if child: 
                 child.baton = out_baton
                 children.append(child)
-        return children
-
-# Re-expose batch optimization so axioms engine remains self-contained
-_batched_deduce_and_evaluate = axioms._batched_deduce_and_evaluate
-_mprt_sample = axioms._mprt_sample
+        return children