"""Reproduce the fixed-budget, non-promoting BTC6H comparison migration evidence. Only writes a new output directory. Never calls training, sweep generation, promotion, preset generation, or deployment. The archived evaluator snapshot isolates old decisions; its imported accounting dependencies are current code. """ from __future__ import annotations import argparse import hashlib import importlib.util import json import math import shutil from pathlib import Path import subprocess import sys from typing import Any REPO = Path(__file__).resolve().parents[1] sys.path.insert(0, str(REPO)) EVIDENCE = REPO / "results/mlp_comparison_migration_2026_09_21" ASSET, TF = "COINBASE_BTCUSD", "6H" def digest(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest() def clean(value: Any) -> Any: if isinstance(value, dict): return {str(k): clean(v) for k, v in value.items()} if isinstance(value, (list, tuple)): return [clean(v) for v in value] if hasattr(value, "item"): return clean(value.item()) if isinstance(value, float) and not math.isfinite(value): return None if isinstance(value, Path): return str(value) return value def write_json(path: Path, value: Any) -> None: path.write_text(json.dumps(clean(value), indent=2, sort_keys=True, allow_nan=False) + "\n") def verify_protected() -> int: expected = json.loads((EVIDENCE / "protected_hashes.json").read_text()) changed = [name for name, sha in expected.items() if not (REPO / name).is_file() or digest(REPO / name) != sha] if changed: raise RuntimeError(f"Protected inputs changed: {changed}") return len(expected) def old_module(): spec = importlib.util.spec_from_file_location("migration_initial_sweep", EVIDENCE / "initial_sweep.py") module = importlib.util.module_from_spec(spec) sys.modules[spec.name] = module spec.loader.exec_module(module) module.REPO, module.ASSET = REPO, ASSET module.WINNER_DIR = REPO / "results/winners" return module def gate_diagnostics(row: dict, reference: dict, stored: dict | None = None) -> dict: """Expose preserved substantive conditions independently of evaluator API.""" historical = stored or {} floor = lambda key, default: max(float(reference.get(key, default)), float(historical.get(key, default))) oos_applies = (float(row.get("OOS_Total Trades", 0)) >= 1 and floor("OOS_Total Trades", 0) >= 1 and float(row.get("OOS_Bear_Pct", 0)) < 85) return { "candidate_not_incumbent": row.get("source") != "current", "positive_pnl_dd": float(row["P&L/DD Ratio"]) > 0, "nonfragile": float(row["Fragile_0.05"]) == 0, "both_subperiods_positive": float(row["Sub1_Calmar"]) > 0 and float(row["Sub2_Calmar"]) > 0, "no_negative_qualifying_wfo": int(row["WFO_Neg_Folds"]) == 0, "pnl_dd_floor": floor("P&L/DD Ratio", 0), "calmar_floor": floor("Calmar Ratio", -10), "pnl_dd_meets_floor": float(row["P&L/DD Ratio"]) >= floor("P&L/DD Ratio", 0), "calmar_meets_floor": float(row["Calmar Ratio"]) >= floor("Calmar Ratio", -10), "oos_gate_applies": oos_applies, "oos_sortino_protected": not oos_applies or float(row["OOS_Sortino Ratio"]) >= floor("OOS_Sortino Ratio", -10) - 1, } def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--output", type=Path, required=True) args = parser.parse_args() out = args.output.resolve() if out.exists() and any(out.iterdir()): raise SystemExit("Output directory must be empty") out.mkdir(parents=True, exist_ok=True) before = verify_protected() for script in ("sync_params.py", "check_pine.py"): run = subprocess.run([sys.executable, str(REPO / "tools" / script)], cwd=REPO, capture_output=True, text=True) (out / f"{script}.log").write_text(run.stdout + run.stderr) if run.returncode: raise RuntimeError(f"Prerequisite failed: {script}") from tools import run_mlp_deep_sweep as sweep from tools import run_mlp_adaptive as adaptive sweep.ASSET = ASSET old = old_module() stored = old.load_winner(TF) incumbent_params = old.normalise_param(stored) incumbent_params["source"] = "current" configs = [("incumbent", incumbent_params)] for family in ("grouped_v1", "random_sparse_v1", "hybrid_grouped_v1"): for seed in (9090, 9191, 9292): path = REPO / f"strategies/params/mlp/mlp_weights_{ASSET}_{TF}_{family}_seed{seed}.json" params = old.normalise_param(old.recommended_from_artifact(str(path))) params["source"] = f"{family}_seed{seed}" configs.append((params["source"], params)) write_json(out / "frozen_configurations.json", configs) # Shared snapshot/cache API is intentionally the same one used by sweep. corrected_rows, manifest = evaluate_corrected(sweep, configs) write_json(out / "comparison_manifest.json", manifest) old_cache: dict = {} old_rows = {name: old.evaluate(TF, params, old_cache) for name, params in configs} current = corrected_rows["incumbent"] adaptive_current = adaptive.read_winner_evidence(ASSET, TF) assert adaptive_current["Comparison_Fingerprint"] == current["Comparison_Fingerprint"] assert math.isclose(adaptive_current["Calmar Ratio"], current["Calmar Ratio"], rel_tol=1e-12) state = adaptive.TFState(TF, comparison_fingerprint=current["Comparison_Fingerprint"], last_calmar=current["Calmar Ratio"]) delta = adaptive.comparison_delta(state, current, adaptive_current) assert delta == 0 old_order = sorted(old_rows, key=lambda name: old.rank_tuple(old_rows[name]), reverse=True) corrected_order = sorted(corrected_rows, key=lambda name: sweep.rank_tuple(corrected_rows[name]), reverse=True) decisions = [] for name, _ in configs: old_row, row = old_rows[name], corrected_rows[name] old_decision = old.promotable(old_row, old_rows["incumbent"], stored) old_policy_same_evidence = old.promotable(row, current, stored) new_decision = sweep.promotable(row, current, stored) old_gates = gate_diagnostics(row, current, stored) new_gates = gate_diagnostics(row, current) differences = {key: {"old": old_gates[key], "corrected": val} for key, val in new_gates.items() if val != old_gates[key]} decisions.append({"model": name, "old_promotable": old_decision, "old_policy_on_corrected_evidence": old_policy_same_evidence, "corrected_promotable": new_decision, "old_rank": old_order.index(name) + 1, "corrected_rank": corrected_order.index(name) + 1, "old_gates_on_corrected_evidence": old_gates, "corrected_gates": new_gates, "changed_reference_conditions": differences, "failed_preserved_conditions": [k for k, v in new_gates.items() if v is False and k != "oos_gate_applies"], "changed_replay_metrics": {k: {"old": clean(old_row.get(k)), "corrected": clean(v)} for k, v in row.items() if k in old_row and clean(old_row[k]) != clean(v)}, "explanation": decision_explanation(old_decision, old_policy_same_evidence, new_decision, differences)}) after = verify_protected() sources = ["tools/run_mlp_deep_sweep.py", "tools/run_mlp_adaptive.py", "tools/verify_mlp_comparison_migration.py", "strategies/mlp_accounting.py", "strategies/mlp_comparison.py", "strategies/strategy_mlp_scores.py", "config.py"] sources += [str(path.relative_to(REPO)) for path in (REPO / "tools").glob("mlp_comparison*.py")] sources = sorted(set(sources) | set(manifest["source_hashes"])) for name in sources: target = out / "source" / name target.parent.mkdir(parents=True, exist_ok=True) shutil.copyfile(REPO / name, target) shutil.copyfile(EVIDENCE / "initial_sweep.py", out / "source/initial_sweep.py") shutil.copyfile(EVIDENCE / "protected_hashes.json", out / "protected_hashes.json") write_json(out / "comparison.json", { "status": "passed", "asset": ASSET, "tf": TF, "models": len(configs), "additional_threshold_trials_per_model": 0, "promotions": 0, "historical_search_budget": "Unequal; 66 dense bb50 + 36 bull surviving BTC6H artifacts versus three seeds per alternative family. These are artifact counts, not complete search ledgers.", "source_hashes": {name: digest(REPO / name) for name in sources}, "old_sweep_source_sha256": digest(EVIDENCE / "initial_sweep.py"), "protected_artifacts_verified_before": before, "protected_artifacts_verified_after": after, "historical_winner_unchanged": stored, "old_replays": old_rows, "corrected_replays": corrected_rows, "decisions": decisions, "adaptive_identity": {"fingerprint": current["Comparison_Fingerprint"], "calmar": current["Calmar Ratio"], "unchanged_incumbent_delta": delta}, "limitations": ["Old evaluator snapshot imports current shared accounting; it is the pre-migration comparison path, not pre-audit accounting.", "Historical BTC6H saved/replayed return discrepancy remains unresolved.", "Common older data is valid for a historical comparison; recency alone is not incompatibility.", "No untouched holdout or live TradingView parity claim."]}) print(json.dumps({"status": "passed", "models": len(configs), "output": str(out)})) def decision_explanation(old: bool, isolated: bool, new: bool, differences: dict) -> str: parts = [] if differences: parts.append("Stored incompatible metric maxima are removed; changed reference conditions are enumerated explicitly.") if old != isolated: parts.append("The legacy evaluator and corrected snapshot/coverage evaluator change the decision before changing the reference policy; inspect saved complete replay rows.") if isolated != new: parts.append("On identical corrected replay rows the reference/evidence policy changes eligibility.") if old == new: parts.append("Final promotion decision is unchanged.") return " ".join(parts) def evaluate_corrected(sweep, configs): previous = sweep._COMPARISON try: snapshot = sweep.capture_comparison(TF, [params["mlp_weights_file"] for _, params in configs]) sweep._COMPARISON = snapshot cache = {} rows = {name: sweep.evaluate(TF, params, cache) for name, params in configs} sweep.require_replay(rows["incumbent"]) snapshot.validate_sources(REPO) assert len({row["Comparison_Fingerprint"] for row in rows.values()}) == 1 return rows, json.loads(snapshot.manifest_json) finally: sweep._COMPARISON = previous if __name__ == "__main__": main()