feat: process equation database samples through PIST pipeline (209 equations)

- Processed 209 unique equations from Research Stack 3-Mathematical-Models
  66% signal detected, mean λ=4.68
  36.8% CognitiveLoadField, 29.2% SignalShapedRouteCompiler, 34.0% LogogramProjection
- Full 1.2M equation database unreachable (neon down, custom compression)
- Added process_large_equations.py for batch processing
This commit is contained in:
allaun 2026-06-30 06:58:40 -05:00
parent 098e3ded11
commit f4b16e4f93
2 changed files with 161 additions and 0 deletions

View file

@ -0,0 +1,148 @@
#!/usr/bin/env python3
"""Process the large equation database through the PIST pipeline."""
import json, re, sys, hashlib
from pathlib import Path
from collections import Counter
ROOT = Path("/home/allaun/Research Stack/3-Mathematical-Models")
def tokenize(text: str) -> list[str]:
return re.findall(r"[a-z0-9]+|[+\-*/=^(){}\[\]]", text.lower())
def build_matrix(tokens: list[str]) -> list[list[int]]:
vocab = sorted(set(tokens))
if not vocab:
return [[0]*8]*8
matrix = [[0]*8 for _ in range(8)]
for t_i, t_j in zip(tokens, tokens[1:]):
matrix[vocab.index(t_i) % 8][vocab.index(t_j) % 8] += 1
return matrix
def spectral_radius(mat: list[list[int]], n_iter: int = 50) -> float:
n = len(mat)
v = [1.0] * n
for _ in range(n_iter):
w = [sum(mat[i][j] * v[j] for j in range(n)) for i in range(n)]
norm = max(abs(x) for x in w) if w else 1.0
if norm == 0: return 0.0
v = [x / norm for x in w]
return norm
def classify(lam: float) -> str:
if lam >= 4.0: return "CognitiveLoadField"
elif lam >= 1.5: return "SignalShapedRouteCompiler"
else: return "LogogramProjection"
def extract_equations() -> list[str]:
"""Extract all equation strings from the JSON files."""
eqs = []
# From unified_9pattern_samples
try:
d = json.loads((ROOT / "unified_9pattern_samples.json").read_text())
for cat, samples in d.items():
if isinstance(samples, list):
for s in samples:
if isinstance(s, str):
eqs.append(s)
elif isinstance(s, dict):
for v in s.values():
if isinstance(v, str) and len(v) > 5:
eqs.append(v)
except: pass
# From unknown_discovery_report
try:
d = json.loads((ROOT / "unknown_discovery_report.json").read_text())
samples = d.get("samples", {})
for cat, items in samples.items():
if isinstance(items, list):
eqs.extend(items)
except: pass
# From structural_discovery (top structures)
try:
d = json.loads((ROOT / "structural_discovery.json").read_text())
for item in d.get("top_structures", []):
eq = item.get("equation", {})
if isinstance(eq, dict):
s = eq.get("structure", "")
if s:
eqs.append(s)
elif isinstance(eq, str):
eqs.append(eq)
except: pass
return eqs
def main():
print("Extracting equations...")
eqs = extract_equations()
print(f" Found {len(eqs)} equation strings")
# Deduplicate
eqs = list(set(eqs))
print(f" Unique: {len(eqs)}")
# Process through pipeline
results = []
shape_counts = Counter()
signal_count = 0
for i, eq in enumerate(eqs):
if i % 1000 == 0 and i > 0:
print(f" processed {i}/{len(eqs)}...")
tokens = tokenize(str(eq))
if not tokens:
continue
matrix = build_matrix(tokens)
lam = spectral_radius(matrix)
shape = classify(lam)
mhash = hashlib.sha256(
json.dumps(matrix, separators=(",", ":")).encode()
).hexdigest()[:12]
shape_counts[shape] += 1
if lam >= 1.5:
signal_count += 1
results.append({
"lambda": round(lam, 4),
"shape": shape,
"hash": mhash,
"len": len(eq),
})
print(f"\n {'='*50}")
print(f" Results: {len(results)} equations processed")
print(f" Signal detected: {signal_count}/{len(results)} ({100*signal_count/len(results):.1f}%)")
print(f" Shape distribution:")
for shape, count in shape_counts.most_common():
print(f" {shape}: {count} ({100*count/len(results):.1f}%)")
print(f" Mean λ: {sum(r['lambda'] for r in results)/len(results):.3f}")
# Summary receipt
receipt = {
"schema": "large_equation_pipeline_v1",
"source": "equations_compressed (decompressed samples)",
"total_unique": len(eqs),
"processed": len(results),
"signals": signal_count,
"shapes": dict(shape_counts),
"mean_lambda": round(sum(r['lambda'] for r in results)/len(results), 4) if results else 0,
}
out = Path("signatures/large_equation_pipeline.json")
out.write_text(json.dumps(receipt, indent=2) + "\n")
print(f"\n Receipt: {out}")
if __name__ == "__main__":
main()

View file

@ -0,0 +1,13 @@
{
"schema": "large_equation_pipeline_v1",
"source": "equations_compressed (decompressed samples)",
"total_unique": 209,
"processed": 209,
"signals": 138,
"shapes": {
"SignalShapedRouteCompiler": 61,
"LogogramProjection": 71,
"CognitiveLoadField": 77
},
"mean_lambda": 4.6759
}