Create practicality_axioms.py

practicality_axioms.py

@@ -0,0 +1,191 @@
+import math
+import random
+import torch
+from typing import Dict, List, Tuple, Set, Optional, Any
+from dataclasses import dataclass, field
+from practicality_core import Problem, DEVICE, SOLVE_THRESHOLD, _c15, IV
+
+@dataclass
+class Hypothesis:
+    hid: str
+    binding: Dict[str, float]
+    h_type: str
+    claim: str
+    derivation: List[str]
+    pinned_vars: Dict[str, float]
+    free_vars: List[str]
+    confidence: float
+    ce: float = float('inf')
+    is_fully_determined: bool = False
+
+@dataclass
+class L9Certificate:
+    residual_ce: float; dominant_vars: List[str]; dominant_exprs: List[str]; tension_class: str="unknown"
+
+@dataclass
+class Baton:
+    binding: Dict[str, float]; ce: float; ray_id: str=""; depth: int=0; l9: Optional[L9Certificate]=None
+
+@dataclass
+class AxiomRay:
+    sequence: List[str]; ray_id: str; ray_type: str="seed"
+    depth: int=0; parent_id: Optional[str]=None
+    baton: Optional[Baton]=None; ce_prior: float=float('inf')
+    @property
+    def name(self) -> str: return "→".join(a[:3] for a in self.sequence)
+    def extend(self, axiom, rtype, new_prior=float('inf')):
+        return AxiomRay(sequence=self.sequence+[axiom], ray_id=f"{self.ray_id}+{axiom[:3]}",
+                        ray_type=rtype, depth=self.depth+1, parent_id=self.ray_id, ce_prior=new_prior)
+
+class Axiom:
+    CONTINUOUS="CONTINUOUS"; DISCRETE="DISCRETE"; QUADRATIC="QUADRATIC"
+    BILINEAR="BILINEAR"; METRIC="METRIC"; SYMMETRIC="SYMMETRIC"
+    MUTABLE="MUTABLE"; EXTREMAL="EXTREMAL"; ENTROPY="ENTROPY"
+    ATOMIC="ATOMIC"; PARSIMONY="PARSIMONY"; DUALITY="DUALITY"
+
+ALL_AXIOMS = [getattr(Axiom, a) for a in dir(Axiom) if not a.startswith("__")]
+
+class UCB1BanditSeeder:
+    def __init__(self):
+        self.stats = {a: {"tries": 0, "reward": 0.0} for a in ALL_AXIOMS}
+        self.total_tries = 0
+
+    def record_reward(self, axiom, ce_before, ce_after):
+        reward = max(0.0, ce_before - ce_after)
+        self.stats[axiom]["tries"] += 1
+        self.stats[axiom]["reward"] += reward
+        self.total_tries += 1
+
+    def intelligent_branch(self, ray, out_baton, remaining_axioms, branch_width):
+        scored = []
+        for ax in remaining_axioms:
+            tries = self.stats[ax]["tries"]
+            if tries == 0: ucb = 999.0
+            else:
+                avg_reward = self.stats[ax]["reward"] / tries
+                exploration = math.sqrt(math.log(self.total_tries + 1) / tries)
+                ucb = avg_reward + 0.5 * exploration
+            scored.append((ucb, ax))
+        scored.sort(key=lambda x: x[0] * random.random(), reverse=True)
+        children = []
+        for _, axiom in scored[:branch_width]:
+            child = ray.extend(axiom, "branch", new_prior=out_baton.ce)
+            if child: 
+                child.baton = out_baton
+                children.append(child)
+        return children
+
+def _batched_deduce_and_evaluate(problem: 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 = _c15(problem.bounds[v][0]), _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=DEVICE, dtype=torch.bool)
+    lo_param_t = torch.tensor(log_lo, device=DEVICE, dtype=torch.float32)
+    hi_param_t = torch.tensor(log_hi, device=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 = _c15(problem.bounds[v][0]), _c15(problem.bounds[v][1])
+            if v in hyp.pinned_vars:
+                p_val = _orig_to_param(_c15(hyp.pinned_vars[v]), j)
+                xr.append(p_val); mr.append(0.0); tr.append(p_val)
+            else:
+                p_val = _orig_to_param(_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=DEVICE, dtype=torch.float32, requires_grad=True)
+    mask = torch.tensor(mask_data, device=DEVICE, dtype=torch.float32)
+    target = torch.tensor(target_data_p, device=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 < 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: Problem, N: int):
+    var_list = problem.variables; V = len(var_list)
+    lo_t = torch.tensor([_c15(problem.bounds.get(v, (-10.0, 10.0))[0]) for v in var_list], device=DEVICE, dtype=torch.float32)
+    hi_t = torch.tensor([_c15(problem.bounds.get(v, (-10.0, 10.0))[1]) for v in var_list], device=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=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=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()