#!/usr/bin/env python3 """ rrc_slo_sweep.py — Sweep merge aggressiveness to find the optimal refactoring configuration that balances speed vs spectral richness. Composite score: speedup × community_retention × node_retention × (1 + mod_gain) × (1 - cent_spread_loss) """ from __future__ import annotations import json import subprocess import sys import time from pathlib import Path SHIM = Path(__file__).resolve().parent ORIGINAL = SHIM.parent.parent / "shared-data" / "data" / "domain_manifold_graph_v1.json" SACRIFICIAL = Path("/tmp") / "rrc_sweep_sacrificial.json" ORACLE = SHIM / "rrc_refactor_oracle.py" ANALYZER = SHIM / "rrc_slo_analyzer.py" SLO_RECEIPT_PATH = SHIM / "rrc_slo_receipt.json" SWEEP_OUT = SHIM / "rrc_slo_sweep_receipt.json" def run_oracle(max_merges: int, max_iter: int, sac: Path) -> dict: start = time.time() r = subprocess.run( [sys.executable, str(ORACLE), "--graph", str(sac), "--threshold-prune", "0.005", "--threshold-merge", "0.01", "--max-iterations", str(max_iter), "--max-merges", str(max_merges), "--apply"], capture_output=True, text=True, timeout=300, ) return dict(returncode=r.returncode, stdout=r.stdout, stderr=r.stderr, elapsed_s=round(time.time() - start, 1)) def run_analyzer(original: Path, refactored: Path) -> dict: r = subprocess.run( [sys.executable, str(ANALYZER), "--baseline", str(original), "--target", str(refactored), "--output", str(SLO_RECEIPT_PATH)], capture_output=True, text=True, timeout=120, ) if SLO_RECEIPT_PATH.exists(): return json.loads(SLO_RECEIPT_PATH.read_text()) return {} def compute_score(a: dict) -> float: bs = a.get("baseline_structural") or {} ts = a.get("target_structural") or {} bp = a.get("baseline_performance") or {} tp = a.get("target_performance") or {} speedup = bp.get("total_ms", 1) / max(tp.get("total_ms", 1), 0.001) comm_ret = ts.get("community_count", 1) / max(bs.get("community_count", 1), 1) node_ret = ts.get("num_nodes", 1) / max(bs.get("num_nodes", 1), 1) mod_gain = max(0, ts.get("modularity", 0) - bs.get("modularity", 0)) cent_loss = max(0, (bs.get("centrality_spread", 0) - ts.get("centrality_spread", 0)) / max(bs.get("centrality_spread", 1e-6), 1e-6)) return round(speedup * comm_ret * node_ret * (1 + mod_gain) * (1 - cent_loss), 4) def main() -> int: import shutil sweeps = [(20, 5, "aggresive"), (10, 5, "moderate"), (5, 5, "conservative"), (2, 5, "gentle"), (1, 5, "minimal")] results = [] print(f"{'Config':>15s} {'Merges':>6s} {'Nodes':>8s} " f"{'Speedup':>8s} {'Comm':>6s} {'Mod':>8s} " f"{'CentSpd':>8s} {'Score':>8s} {'Time':>6s}") print("-" * 95) for max_m, max_i, label in sweeps: shutil.copy2(str(ORIGINAL), str(SACRIFICIAL)) t0 = time.time() o = run_oracle(max_m, max_i, SACRIFICIAL) if o["returncode"] != 0: print(f" {label:>15s} FAILED (rc={o['returncode']})") print(f" {o['stderr'][:200]}") continue a = run_analyzer(ORIGINAL, SACRIFICIAL) elapsed = round(time.time() - t0, 1) bs = a.get("baseline_structural", {}) ts = a.get("target_structural", {}) bp = a.get("baseline_performance", {}) tp = a.get("target_performance", {}) speedup = bp.get("total_ms", 1) / max(tp.get("total_ms", 1), 0.001) score = compute_score(a) entry = dict(label=label, max_merges=max_m, max_iter=max_i, initial_nodes=bs.get("num_nodes", 0), final_nodes=ts.get("num_nodes", 0), initial_edges=bs.get("num_edges", 0), final_edges=ts.get("num_edges", 0), speedup=round(speedup, 2), communities_initial=bs.get("community_count", 0), communities_final=ts.get("community_count", 0), modularity_initial=bs.get("modularity", 0), modularity_final=ts.get("modularity", 0), cent_spread_initial=bs.get("centrality_spread", 0), cent_spread_final=ts.get("centrality_spread", 0), isolation_ratio_initial=bs.get("isolation_ratio", 0), isolation_ratio_final=ts.get("isolation_ratio", 0), composite_score=score, elapsed_s=elapsed) results.append(entry) print(f" {label:>15s} {max_m:>6d} " f"{entry['final_nodes']:>4d}/{entry['initial_nodes']:<2d} " f"{speedup:>7.2f}x " f"{entry['communities_final']:>4d}/{entry['communities_initial']:<1d} " f"{entry['modularity_final']:>7.4f} " f"{entry['cent_spread_final']:>7.4f} " f"{score:>7.4f} {elapsed:>5.1f}s") if not results: print("\nAll sweeps failed.") return 1 winner = max(results, key=lambda r: r["composite_score"]) print("\n" + "=" * 95) print(f"Winner: {winner['label']} (max_merges={winner['max_merges']}, " f"score={winner['composite_score']})") print(f" {winner['final_nodes']}/{winner['initial_nodes']} nodes " f"({100*winner['final_nodes']//max(winner['initial_nodes'],1)}%)") print(f" {winner['speedup']}x speedup") print(f" {winner['communities_final']}/{winner['communities_initial']} communities") print(f" modularity {winner['modularity_initial']} → {winner['modularity_final']}") print(f" centrality spread {winner['cent_spread_initial']} → {winner['cent_spread_final']}") print(f" no isolated nodes: {winner['isolation_ratio_final'] == 0}") import hashlib from datetime import datetime, timezone receipt = dict( schema="rrc_slo_sweep_v1", claim_boundary=( "parameter-sweep-over-merge-aggressiveness;" "composite-score-maximizes-speedup-community-retention-modularity;" "no-decision-logic;measurement-only" ), sweeps=results, winner=winner, composite_formula=( "score = speedup * community_retention * node_retention " "* (1 + mod_gain) * (1 - cent_spread_loss)" ), ) canonical = json.dumps(receipt, sort_keys=True, separators=(",", ":")) receipt["receipt_sha256"] = hashlib.sha256(canonical.encode()).hexdigest() receipt["computed_at"] = datetime.now(timezone.utc).isoformat() SWEEP_OUT.write_text(json.dumps(receipt, indent=2, default=str)) print(f"\nFull receipt: {SWEEP_OUT}") print(f"SHA256: {receipt['receipt_sha256']}") return 0 if __name__ == "__main__": sys.exit(main())