Research-Stack/4-Infrastructure/shim/braid_search.py
Brandon Schneider fdb33aa08e fix: braid_search.py QUBO/soliton from float to Q16_16 integer arithmetic
AGENTS.md §1.4 compliance: all internal computation now uses Q16_16 integers.
Float only at HiGHS API boundary and display statements.

- Q16_SCALE = 65536, _q16(), _q16_to_float(), _q16_signed()
- bracket_cost, crossing_penalty, build_qubo_matrix: all int
- soliton_search, qubo_optimize: Q16_16 temperature/energy
- 68/68 tests pass
2026-05-28 17:47:40 -05:00

707 lines
25 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

#!/usr/bin/env python3
"""
Braid Search — Soliton Search, QUBO Optimization, and Sidon Slot Assignment
Provides combinatorial optimization routines for braid crossing configurations:
1. Soliton-inspired search: uses a soliton pulse shape (discrete approximation)
to guide a stochastic search over crossing configurations, concentrating
exploration near high-energy regions.
2. QUBO (Quadratic Unconstrained Binary Optimization) formulation for bracket
selection: encodes pairwise crossing costs into a QUBO matrix and solves
via simulated annealing.
3. Sidon set construction for unique slot assignment: assigns braid strands
to collision-free time/frequency slots using a Sidon set (all pairwise
sums are distinct).
All arithmetic is integer/Q16_16 where applicable. No float in compute paths.
"""
from __future__ import annotations
import math
import random
import hashlib
from typing import Dict, List, Optional, Tuple
# HiGHS QUBO solver — lazy import with fallback
try:
from qubo_highs import solve_qubo_highs
_HIGHS_AVAILABLE = True
except ImportError:
_HIGHS_AVAILABLE = False
# Q16_16 constants
Q16_ONE = 0x00010000
Q16_MASK = 0xFFFFFFFF
Q16_SCALE = 65536 # alias for readability
def _q16(f: float) -> int:
"""Convert float to Q16_16 integer."""
return max(-2147483648, min(2147483647, int(round(f * Q16_SCALE))))
def _q16_to_float(q: int) -> float:
"""Convert Q16_16 integer to float (for display/comparison only)."""
if q >= 0x80000000:
return (q - 0x100000000) / Q16_SCALE
return q / Q16_SCALE
def _q16_signed(v: int) -> int:
"""Convert unsigned Q16_16 to signed Python int (still in Q16_16 units)."""
if v >= 0x80000000:
return v - 0x100000000
return v
# ── Sidon Set Slot Assignment ────────────────────────────────────────────────
def assign_sidon_slots(num_slots: int, method: str = 'greedy_optimal') -> List[int]:
"""Generate a Sidon set of size *num_slots* and return sorted slot indices.
A Sidon set (B₂ sequence) is a set of integers where all pairwise sums
are distinct. This guarantees collision-free slot assignment for braid
strands: no two pairs of strands occupy the same combined slot.
Args:
num_slots: number of slots needed.
method: construction method.
'powers_of_2'{1, 2, 4, 8, 16, ...} (conservative, max=2^(n-1))
'greedy_optimal' — Mian-Chowla sequence (default, densest practical)
'algebraic' — algebraic integer construction from totally real
number fields (scales well for large n)
Returns:
Sorted list of *num_slots* integers forming a Sidon set.
"""
if num_slots <= 0:
return []
if method == 'powers_of_2':
return [1 << i for i in range(num_slots)]
elif method == 'greedy_optimal':
return _mian_chowla(num_slots)
elif method == 'algebraic':
return _algebraic_sidon(num_slots)
else:
raise ValueError(f"Unknown Sidon method: {method!r}")
def _mian_chowla(n: int) -> List[int]:
"""Mian-Chowla sequence: densest known Sidon set construction.
a(0)=1, a(n) = smallest integer > a(n-1) such that all pairwise
sums a(i)+a(j) (i<=j) are distinct.
Returns a list of *n* positive integers forming a Sidon set.
For n=8 the result is {1, 3, 7, 12, 20, 30, 44, 65} — max is 65
vs 128 for powers-of-2 (49% smaller).
"""
if n <= 0:
return []
seq = [1]
sums = {2} # a(0)+a(0) = 2
while len(seq) < n:
candidate = seq[-1] + 1
while True:
# Check if candidate + each existing element produces a new sum
new_sums = {candidate + s for s in seq} | {candidate * 2}
if new_sums.isdisjoint(sums):
seq.append(candidate)
sums |= new_sums
break
candidate += 1
return seq
def _algebraic_sidon(n: int) -> List[int]:
"""Algebraic integer Sidon set from totally real number field.
Uses the Bloom-Sawin construction: elements of O_K in a box.
For practical n, falls back to Mian-Chowla with scaling.
For small n (≤20), delegates to Mian-Chowla directly.
For larger n, uses the number field construction: take elements
of Z[√d] in a box, project to integers via the norm map.
d = smallest prime > n (totally real quadratic field).
"""
if n <= 20:
return _mian_chowla(n)
# Find the smallest prime > n for the quadratic field Q(√d)
d = n + 1
while not _is_prime(d):
d += 1
# Elements: a + b*√d with 0 ≤ a, b ≤ √(n/d)
box = max(1, int(math.sqrt(n / d)))
slots: List[int] = []
seen: set = set()
for b in range(box + 1):
for a in range(box + 1):
val = a + b * b * d # norm map projection
if val > 0 and val not in seen:
slots.append(val)
seen.add(val)
if len(slots) >= n:
break
if len(slots) >= n:
break
result = sorted(slots[:n])
# Verify Sidon property; fall back to Mian-Chowla if construction fails
if not _verify_sidon(result):
return _mian_chowla(n)
return result
def _verify_sidon(seq: List[int]) -> bool:
"""Verify all pairwise sums are distinct (internal helper)."""
sums: set = set()
for i in range(len(seq)):
for j in range(i, len(seq)):
s = seq[i] + seq[j]
if s in sums:
return False
sums.add(s)
return True
def _is_prime(n: int) -> bool:
"""Deterministic primality test (trial division)."""
if n < 2:
return False
if n < 4:
return True
if n % 2 == 0 or n % 3 == 0:
return False
i = 5
while i * i <= n:
if n % i == 0 or n % (i + 2) == 0:
return False
i += 6
return True
def verify_sidon(slots: List[int]) -> bool:
"""Verify that *slots* is a valid Sidon set (all pairwise sums distinct)."""
sums_seen: set = set()
for i in range(len(slots)):
for j in range(i, len(slots)):
s = slots[i] + slots[j]
if s in sums_seen:
return False
sums_seen.add(s)
return True
# ── Soliton-Inspired Search ─────────────────────────────────────────────────
def _discrete_soliton(k: int, n: int) -> int:
"""Evaluate the (discrete) ideal soliton distribution at index *k* for size *n*.
μ(1) = 1/n, μ(k) = 1/(k*(k-1)) for k = 2..n
Returns Q16_16 integer weight.
"""
if k == 1:
return Q16_ONE // n
elif 2 <= k <= n:
return Q16_ONE // (k * (k - 1))
return 0
def _candidate_energy(crossing: dict) -> int:
"""Compute an 'energy' score for a crossing configuration (Q16_16 int).
Higher energy ↔ more promising configuration.
Uses bracket admissibility, gap size, and parity diversity.
"""
energy = 0
brackets = crossing.get("brackets", [])
for b in brackets:
if b.get("admissible", False):
energy += 2 * Q16_ONE
gap = _q16_signed(b.get("gap", 0))
# Larger gaps are generally better (more room for braiding)
# abs(gap) * 0.5 in Q16_16 = abs(gap) // 2
energy += abs(gap) // 2
# Parity diversity bonus
parities = {b.get("admissible", False) for b in brackets}
if len(parities) > 1:
energy += Q16_ONE
return energy
def soliton_search(target_energy: float, candidates: List[dict],
max_iterations: int = 1000, seed: int = 0) -> dict:
"""Soliton-inspired stochastic search over crossing configurations.
The search distributes exploration effort according to a discrete soliton
pulse: heavy sampling of promising candidates (high energy), lighter
sampling of others, with stochastic jumps to avoid local optima.
All internal computation uses Q16_16 integers. Energy values in the
returned dict are converted to float at the API boundary.
Args:
target_energy: desired energy threshold for a satisfactory solution.
candidates: list of crossing configuration dicts.
max_iterations: maximum search iterations.
seed: RNG seed for reproducibility.
Returns:
{
"best": dict, # best crossing found
"best_energy": float,
"iterations": int,
"converged": bool, # reached target_energy
"history": list, # (iteration, energy) pairs (sampled)
}
"""
rng = random.Random(seed)
n = len(candidates)
if n == 0:
return {"best": {}, "best_energy": 0.0, "iterations": 0,
"converged": False, "history": []}
# Convert target to Q16_16 for internal comparison
if target_energy == float("inf"):
target_q16 = 2 ** 62 # effectively infinite
else:
target_q16 = _q16(target_energy)
# Pre-compute energies (Q16_16 ints)
energies = [_candidate_energy(c) for c in candidates]
# Build soliton weights (Q16_16 ints)
weights = [_discrete_soliton(i + 1, n) for i in range(n)]
total_w = sum(weights)
# Convert to Q16_16 probabilities (sum ≈ Q16_SCALE)
if total_w > 0:
probs_q16 = [w * Q16_SCALE // total_w for w in weights]
else:
probs_q16 = [Q16_SCALE // n] * n
# Sort candidates by energy descending; remap probs accordingly
ranked = sorted(range(n), key=lambda i: energies[i], reverse=True)
# Assign higher soliton weight to higher-energy candidates
weighted_probs_q16 = [0] * n
for rank_idx, orig_idx in enumerate(ranked):
weighted_probs_q16[orig_idx] = probs_q16[rank_idx]
best_idx = max(range(n), key=lambda i: energies[i])
best_energy = energies[best_idx]
best = candidates[best_idx]
history: List[Tuple[int, float]] = []
for iteration in range(1, max_iterations + 1):
# Sample a candidate index from the soliton-weighted distribution
r_q16 = int(rng.random() * Q16_SCALE)
cumulative = 0
chosen = rng.randint(0, n - 1)
for idx in range(n):
cumulative += weighted_probs_q16[idx]
if cumulative >= r_q16:
chosen = idx
break
# Evaluate (already pre-computed)
e = energies[chosen]
if e > best_energy:
best_energy = e
best = candidates[chosen]
best_idx = chosen
history.append((iteration, _q16_to_float(best_energy)))
# Check convergence
if best_energy >= target_q16:
return {
"best": best,
"best_energy": _q16_to_float(best_energy),
"iterations": iteration,
"converged": True,
"history": history,
}
# Stochastic jump: small probability of random exploration
if rng.random() < 0.05:
jump_idx = rng.randint(0, n - 1)
if energies[jump_idx] > best_energy:
best_energy = energies[jump_idx]
best = candidates[jump_idx]
best_idx = jump_idx
history.append((iteration, _q16_to_float(best_energy)))
return {
"best": best,
"best_energy": _q16_to_float(best_energy),
"iterations": max_iterations,
"converged": False,
"history": history,
}
# ── QUBO Matrix (dict format for HiGHS) ─────────────────────────────────────
def bracket_cost(bracket: dict) -> int:
"""Compute the individual (diagonal) cost for a single bracket (Q16_16 int).
Negative cost → prefer selecting this bracket (admissible is good).
"""
base = Q16_ONE if bracket.get("admissible", False) else 2 * Q16_ONE
# Gap magnitude bonus: larger gaps reduce cost
# abs(gap) * 0.1 in Q16_16 = abs(gap) * _q16(0.1) // Q16_SCALE
gap = _q16_signed(bracket.get("gap", 0))
return -base + abs(gap) * _q16(0.1) // Q16_SCALE
def crossing_penalty(b1: dict, b2: dict) -> int:
"""Compute the off-diagonal interaction cost between two brackets (Q16_16 int).
Positive penalty → discourage selecting both when they conflict.
"""
penalty = 0
if _brackets_overlap(b1, b2):
penalty += 2 * Q16_ONE
# Gap similarity: reward diversity
g1 = _q16_signed(b1.get("gap", 0))
g2 = _q16_signed(b2.get("gap", 0))
penalty -= abs(g1 - g2) * _q16(0.1) // Q16_SCALE
return penalty
def build_qubo_matrix(brackets: List[dict]) -> Dict[Tuple[int, int], int]:
"""Convert brackets to a dict-format QUBO matrix Q[i,j] (Q16_16 values).
Q[i,i] = bracket_cost(bracket) — diagonal (linear bias).
Q[i,j] = crossing_penalty(b_i, b_j) — off-diagonal (interaction).
This format is directly consumable by ``solve_qubo_highs`` (after
converting Q16_16 values to float at the API boundary).
Args:
brackets: list of BraidBracket dicts.
Returns:
Dict mapping (i, j) → Q16_16 cost coefficient.
"""
n = len(brackets)
Q: Dict[Tuple[int, int], int] = {}
for i in range(n):
Q[(i, i)] = bracket_cost(brackets[i])
for j in range(i + 1, n):
Q[(i, j)] = crossing_penalty(brackets[i], brackets[j])
Q[(j, i)] = Q[(i, j)]
return Q
# ── QUBO Optimization (list-format, SA) ────────────────────────────────────
def _build_qubo_matrix(bracket_pairs: List[Tuple[dict, dict]]) -> List[List[int]]:
"""Build a QUBO matrix from bracket pair costs (Q16_16 values).
Each bracket pair (A, B) has a crossing cost derived from gap difference,
admissibility conflict, and parity mismatch.
The QUBO matrix Q is symmetric with shape (n × n) where n = len(bracket_pairs).
Diagonal Q[i][i] = individual pair cost (linear bias).
Off-diagonal Q[i][j] = interaction cost between pairs i and j.
Goal: minimise x^T Q x over binary x ∈ {0,1}^n.
"""
n = len(bracket_pairs)
Q = [[0] * n for _ in range(n)]
for i in range(n):
a_i, b_i = bracket_pairs[i]
# Linear bias: admissible pairs get negative cost (prefer them)
if a_i.get("admissible", False) and b_i.get("admissible", False):
Q[i][i] = -Q16_ONE
else:
Q[i][i] = Q16_ONE
# Quadratic interaction
for j in range(i + 1, n):
a_j, b_j = bracket_pairs[j]
cost = 0
# Overlap penalty: if pairs share a bracket, penalise
if _brackets_overlap(a_i, a_j) or _brackets_overlap(b_i, b_j):
cost += 2 * Q16_ONE
# Gap similarity reward: diverse gaps are better
gap_i = _q16_signed(a_i.get("gap", 0))
gap_j = _q16_signed(a_j.get("gap", 0))
cost -= abs(gap_i - gap_j) * _q16(0.1) // Q16_SCALE
Q[i][j] = cost
Q[j][i] = cost
return Q
def _brackets_overlap(b1: dict, b2: dict) -> bool:
"""Check if two brackets overlap in their [lower, upper] ranges."""
l1 = _q16_signed(b1.get("lower", 0))
u1 = _q16_signed(b1.get("upper", 0))
l2 = _q16_signed(b2.get("lower", 0))
u2 = _q16_signed(b2.get("upper", 0))
return l1 < u2 and l2 < u1
def _qubo_energy(Q: List[List[int]], x: List[int]) -> int:
"""Compute x^T Q x (Q16_16 result)."""
n = len(x)
energy = 0
for i in range(n):
for j in range(n):
energy += Q[i][j] * x[i] * x[j]
return energy
def qubo_optimize(bracket_pairs: List[Tuple[dict, dict]],
max_iterations: int = 5000,
seed: int = 42,
initial_temp: float = 10.0,
cooling_rate: float = 0.9995) -> dict:
"""Solve the QUBO via simulated annealing.
All QUBO matrix values and energy computations use Q16_16 integers.
Temperature and acceptance use Q16_16 internally; the returned energy
is converted to float at the API boundary.
Args:
bracket_pairs: list of (bracket_a, bracket_b) tuples.
max_iterations: SA iteration count.
seed: RNG seed.
initial_temp: starting temperature.
cooling_rate: geometric cooling factor per step.
Returns:
{
"selection": List[int], # binary vector (1 = selected pair)
"selected_pairs": List[Tuple[dict, dict]],
"energy": float,
"iterations": int,
}
"""
rng = random.Random(seed)
n = len(bracket_pairs)
if n == 0:
return {"selection": [], "selected_pairs": [], "energy": 0.0, "iterations": 0}
Q = _build_qubo_matrix(bracket_pairs)
# Initial solution: all ones (select all pairs)
x = [1] * n
current_energy = _qubo_energy(Q, x)
best_x = list(x)
best_energy = current_energy
# Temperature in Q16_16
temp_q16 = _q16(initial_temp)
cooling_q16 = _q16(cooling_rate)
for iteration in range(1, max_iterations + 1):
# Flip a random bit
idx = rng.randint(0, n - 1)
x[idx] = 1 - x[idx]
new_energy = _qubo_energy(Q, x)
delta = new_energy - current_energy
# Accept or reject
# delta/temp_q16 gives dimensionless ratio via Python float division
if delta < 0 or rng.random() < math.exp(-delta / max(temp_q16, 1)):
current_energy = new_energy
if current_energy < best_energy:
best_energy = current_energy
best_x = list(x)
else:
x[idx] = 1 - x[idx] # revert
temp_q16 = temp_q16 * cooling_q16 // Q16_SCALE
selected = [bracket_pairs[i] for i in range(n) if best_x[i]]
return {
"selection": best_x,
"selected_pairs": selected,
"energy": _q16_to_float(best_energy), # convert at API boundary
"iterations": max_iterations,
}
# ── Convenience: combined search ────────────────────────────────────────────
def find_optimal_crossing(brackets: List[dict],
max_iterations: int = 1000) -> dict:
"""Find the optimal crossing configuration from a set of brackets.
1. Build QUBO matrix from brackets (dict format, Q16_16 values).
2. Try HiGHS MIP solver first (exact, fast for small instances).
Q16_16 values are converted to float at the HiGHS API boundary.
3. Fall back to simulated annealing if HiGHS unavailable or fails.
4. Run soliton search on the candidates.
Args:
brackets: list of BraidBracket dicts.
max_iterations: iteration budget for soliton search and SA fallback.
Returns:
{
"qubo_result": dict,
"soliton_result": dict,
"optimal_pairs": List[Tuple[dict, dict]],
"method": str, # 'highs_mip' or 'simulated_annealing'
}
"""
# Build dict-format QUBO matrix from raw brackets (Q16_16 values)
Q = build_qubo_matrix(brackets)
qubo_result = None
method = "simulated_annealing"
# ── Try HiGHS first ────────────────────────────────────────────────
if _HIGHS_AVAILABLE:
try:
# Convert Q16_16 ints to float at the HiGHS API boundary
Q_float = {(i, j): v / Q16_SCALE for (i, j), v in Q.items()}
result = solve_qubo_highs(Q_float, time_limit=30.0)
selection = result.get("solution", [])
# Convert selection vector to bracket pairs
selected_brackets = [brackets[i] for i in range(len(selection))
if i < len(brackets) and selection[i]]
# Reconstruct pairs from selected brackets
selected_pairs = []
for i in range(len(selected_brackets)):
for j in range(i + 1, len(selected_brackets)):
selected_pairs.append((selected_brackets[i],
selected_brackets[j]))
qubo_result = {
"selection": selection,
"selected_pairs": selected_pairs,
"energy": result.get("objective", 0.0),
"iterations": 1,
"highs_status": result.get("status", "unknown"),
}
method = "highs_mip"
except Exception as exc:
# Log and fall through to SA
print(f"[braid_search] HiGHS failed ({exc}), falling back to SA")
# ── Simulated annealing fallback ───────────────────────────────────
if qubo_result is None:
# Generate bracket pairs (only admissible pairs are candidates)
pairs = []
for i in range(len(brackets)):
for j in range(i + 1, len(brackets)):
if (brackets[i].get("admissible", False)
and brackets[j].get("admissible", False)):
pairs.append((brackets[i], brackets[j]))
if not pairs:
# Fall back to all pairs even if not admissible
for i in range(len(brackets)):
for j in range(i + 1, len(brackets)):
pairs.append((brackets[i], brackets[j]))
qubo_result = qubo_optimize(pairs, max_iterations=max_iterations)
method = "simulated_annealing"
# Build crossing candidates from QUBO-selected pairs
selected_pairs = qubo_result.get("selected_pairs", [])
crossing_candidates = []
for a, b in selected_pairs:
crossing_candidates.append({
"brackets": [a, b],
"admissible": a.get("admissible", False) and b.get("admissible", False),
})
# Fallback: if HiGHS selected nothing, build from all pairs
if not crossing_candidates:
for i in range(len(brackets)):
for j in range(i + 1, len(brackets)):
crossing_candidates.append({
"brackets": [brackets[i], brackets[j]],
"admissible": (brackets[i].get("admissible", False)
and brackets[j].get("admissible", False)),
})
soliton_result = soliton_search(
target_energy=float("inf"),
candidates=crossing_candidates,
max_iterations=max_iterations,
)
return {
"qubo_result": qubo_result,
"soliton_result": soliton_result,
"optimal_pairs": selected_pairs if selected_pairs else qubo_result.get("selected_pairs", []),
"method": method,
}
# ── CLI ──────────────────────────────────────────────────────────────────────
def main():
import sys
print("braid_search.py — Soliton Search / QUBO / Sidon Slot Assignment")
print()
print("Functions:")
print(" assign_sidon_slots(num_slots) -> List[int]")
print(" soliton_search(target, candidates) -> dict")
print(" qubo_optimize(bracket_pairs) -> dict")
print(" build_qubo_matrix(brackets) -> Dict[(i,j), int]")
print(" find_optimal_crossing(brackets) -> dict")
print()
print(f" HiGHS available: {_HIGHS_AVAILABLE}")
print()
# Demo: Sidon set methods
for n in (8, 16):
slots_old = assign_sidon_slots(n, 'powers_of_2')
slots_new = assign_sidon_slots(n, 'greedy_optimal')
print(f" Sidon set (n={n}):")
print(f" powers_of_2 : {slots_old} max={max(slots_old)}")
print(f" greedy_optimal: {slots_new} max={max(slots_new)}")
print(f" max reduction: {(1 - max(slots_new)/max(slots_old))*100:.0f}%")
assert verify_sidon(slots_old), "powers_of_2 failed Sidon check"
assert verify_sidon(slots_new), "greedy_optimal failed Sidon check"
if __name__ == "__main__":
import time
# Generate test brackets with varying properties
test_brackets = [
{"admissible": True, "gap": 0x00020000, "lower": 0x00000000, "upper": 0x00030000},
{"admissible": True, "gap": 0x00040000, "lower": 0x00010000, "upper": 0x00050000},
{"admissible": False, "gap": 0x00010000, "lower": 0x00020000, "upper": 0x00030000},
{"admissible": True, "gap": 0x00030000, "lower": 0x00000000, "upper": 0x00030000},
{"admissible": True, "gap": 0x00050000, "lower": 0x00040000, "upper": 0x00090000},
{"admissible": False, "gap": 0x00028000, "lower": 0x00010000, "upper": 0x00038000},
]
main()
print()
# ── Timing: HiGHS vs SA ────────────────────────────────────────────
# Time HiGHS (or full pipeline with HiGHS)
t0 = time.time()
r1 = find_optimal_crossing(test_brackets)
t_highs = time.time() - t0
# Time SA fallback directly
pairs = []
for i in range(len(test_brackets)):
for j in range(i + 1, len(test_brackets)):
pairs.append((test_brackets[i], test_brackets[j]))
t0 = time.time()
r2 = qubo_optimize(pairs)
t_sa = time.time() - t0
print(f" find_optimal_crossing method: {r1.get('method', 'unknown')}")
print(f" find_optimal_crossing: {t_highs*1000:.1f}ms | SA direct: {t_sa*1000:.1f}ms")
print(f" HiGHS available: {_HIGHS_AVAILABLE}")