"""Frozen retrospective diagnostics; never select or promote perturbations.""" from __future__ import annotations import argparse from dataclasses import replace import json from pathlib import Path import subprocess import sys import numpy as np import pandas as pd REPO = Path(__file__).resolve().parents[1] sys.path.insert(0, str(REPO)) from strategies.mlp_accounting import net_returns from strategies.mlp_comparison import digest, fingerprint, require_replay from strategies.mlp_score_structure import confirmed_score_structure from strategies.strategy_mlp_scores import _score_to_signals from tools.run_mlp_entry_loss_experiment import ( FROZEN, MODEL_HASH, evaluator, gate_decisions, hashes, protected_hashes, ) REFERENCE = REPO / 'results/mlp_threshold_comparison_2026_09_21/final' PROTOCOL = 'docs/mlp_revisit_protocol_2026_09_21.md' THRESHOLDS = ('i_long_entry_activation_threshold', 'i_long_exit_activation_threshold', 'i_long_exit_activation_confirmation_threshold') DELTAS = (-1., -.05, .05, 1.) SOURCES = [PROTOCOL, 'tools/run_mlp_revisit.py', 'tests/test_mlp_revisit.py', 'tools/run_mlp_entry_loss_experiment.py', 'strategies/mlp_score_structure.py'] TRACE_KEYS = ('position', 'execute_entry', 'execute_exit', 'net_return') def write_json(path, payload): path.write_text(json.dumps(payload, indent=2, default=str, allow_nan=False) + '\n') def select_centers(rows, cases, param_cols): """Stratify on prior proximity flags, never on new sensitivity outcomes.""" rng = np.random.default_rng(20260921) selected = {} def add(row, sample=False, case=None): params = {key: row[key] for key in param_cols} key = fingerprint(params) item = selected.setdefault(key, dict(id=key, params=params, sample=False, cases=[])) item['sample'] |= sample if case is not None: item['cases'].append(case) for model in sorted({r['mlp_weights_file'] for r in rows}): for flagged in (False, True): # Deduplicate configurations before sampling, not after it. pool = {fingerprint({k: r[k] for k in param_cols}): r for r in rows if r['mlp_weights_file'] == model and (r['Fragile_0.05'] > 0) == flagged} values = list(pool.values()) for index in rng.choice(len(values), min(10, len(values)), replace=False): add(values[index], sample=True) for index, row in enumerate(cases): add(row, case=index) return list(selected.values()) def gate_signal(width): def signal(frame, params, stoch, close, high): gate = (confirmed_score_structure(frame.activation_score.to_numpy(), left=width, right=width).bullish_state if width else np.ones(len(frame), dtype=bool)) return _score_to_signals(frame, params, stoch, close, high, entry_gate=gate) return signal def replay(module, params, cache, policy='canonical'): """Use the actual promotion evaluator with a private policy implementation.""" row = require_replay(module.evaluate('6H', params, cache)) cache_key = (module._COMPARISON.fingerprint, params['mlp_weights_file'], row['Artifact_SHA256']) scored, stoch, close, high = cache[cache_key] trace = module._score_to_signals(scored.copy(), dict(params, _pine_time_start=module.SCORE_START), stoch, close, high) trace['net_return'] = net_returns(trace) row['Policy'] = policy row['Evaluation_Key'] = fingerprint(dict(context=row['Comparison_Fingerprint'], artifact=row['Artifact_SHA256'], params={k: row[k] for k in module.PARAM_COLS}, policy=policy)) return row, trace def trace_arrays(trace): return {key: trace[key].to_numpy() for key in TRACE_KEYS} def sensitivity(base, changed, base_row, row, module): result = {} for name, mask in ( ('IS', (base.time >= pd.Timestamp(module.SCORE_START)) & (base.time < pd.Timestamp(module.TRAIN_END) + pd.Timedelta(days=1))), ('OOS', base.time >= pd.Timestamp(module.OOS_START)), ): actions = ((base.execute_entry != changed.execute_entry) | (base.execute_exit != changed.execute_exit)) & mask a = next(p for p in json.loads(base_row['Period_Measurements']) if p.get('name') == name) b = next(p for p in json.loads(row['Period_Measurements']) if p.get('name') == name) diff = {key: b[key] - a[key] if a[key] is not None and b[key] is not None else None for key in ('pnl_pct', 'dd_pct', 'calmar')} result[name] = dict(changed_action_bars=int(actions.sum()), changed_position_bars=int(((base.position != changed.position) & mask).sum()), changed_return_bars=int(((base.net_return != changed.net_return) & mask).sum()), first_changed_action=str(base.time[actions].iloc[0]) if actions.any() else None, deltas=diff, material=any(abs(diff[k]) >= limit for k, limit in (('pnl_pct', 1.), ('dd_pct', 1.), ('calmar', .05)) if diff[k] is not None)) return result def proximity_events(trace, params, module): rows = [] for threshold in THRESHOLDS: near = np.flatnonzero((np.abs(trace.activation_score - params[threshold]) < .05) & (trace.time < pd.Timestamp(module.TRAIN_END) + pd.Timedelta(days=1))) for index in near: nearby = trace.iloc[index:index+3] rows.append(dict(bar=int(index), time=str(trace.time.iloc[index]), threshold=threshold, score=float(trace.activation_score.iloc[index]), threshold_value=params[threshold], in_trading_window=bool(trace.time.iloc[index] >= pd.Timestamp(module.SCORE_START)), exit_confirmation_enabled=params['i_use_long_exit_confirmation'], entry_window=params['i_entry_score_window'], exit_window=params['i_exit_score_window'], executed_entry_offsets=np.flatnonzero(nearby.execute_entry).tolist(), executed_exit_offsets=np.flatnonzero(nearby.execute_exit).tolist())) return rows def setup(output, stage): if output.exists() and any(output.iterdir()): raise ValueError('Output must be new or empty') output.mkdir(parents=True, exist_ok=True) protected = protected_hashes() for script in ('tools/sync_params.py', 'tools/check_pine.py'): run = subprocess.run([sys.executable, script], cwd=REPO, text=True, capture_output=True) (output / (Path(script).name + '.log')).write_text(run.stdout + run.stderr) if run.returncode: raise RuntimeError(f'Preflight failed: {script}') module = evaluator() incumbent = module.normalise_param(module.load_winner('6H')) if any(incumbent[key] != value for key, value in FROZEN.items()): raise ValueError('Incumbent parameter drift') if digest(Path(incumbent['mlp_weights_file']).read_bytes()) != MODEL_HASH: raise ValueError('Incumbent artifact drift') old_rows = json.loads((REFERENCE / 'all_results.json').read_text()) snapshot = module.capture_comparison('6H', [r['mlp_weights_file'] for r in old_rows]) manifest = json.loads(snapshot.manifest_json) reference = json.loads((REFERENCE / 'comparison_manifest.json').read_text()) for key in ('data_sha256', 'contract', 'source_hashes'): if manifest[key] != reference[key]: raise ValueError(f'Historical comparison drift: {key}') if any(snapshot.artifact_hashes[r['mlp_weights_file']] != r['Artifact_SHA256'] for r in old_rows): raise ValueError('Archived model drift') reference_hashes = hashes([REFERENCE / name for name in ('all_results.json', 'summary.json', 'incumbent.json', 'comparison_manifest.json')]) manifest['source_hashes'].update(hashes([REPO / p for p in SOURCES])) manifest['contract']['revisit'] = dict(stage=stage, protocol=PROTOCOL, seed=20260921, magnitudes=[.05, 1.], pivots=[2, 3, 4], no_selection=True) snapshot = replace(snapshot, manifest_json=json.dumps(manifest, sort_keys=True, separators=(',', ':'))) module._COMPARISON = snapshot write_json(output / 'comparison_manifest.json', manifest) write_json(output / 'protected_hashes.json', protected) write_json(output / 'reference_hashes.json', reference_hashes) return module, incumbent, snapshot, protected, reference_hashes, old_rows def finish(output, module, snapshot, protected, references, count): snapshot.validate_sources(REPO) assert protected == protected_hashes(), 'Protected artifact drift' assert references == hashes([REPO / p for p in references]), 'Reference drift' assert digest(module.data_path('6H').read_bytes()) == json.loads(snapshot.manifest_json)['data_sha256'] write_json(output / 'verification.json', dict(complete=True, evaluations=count, protected_files=len(protected), protected_unchanged=True, source_input_hashes_unchanged=True, promotions=0, retrospective=True)) def run_fragility(output): module, incumbent, snapshot, protected, references, old_rows = setup(output, 'fragility') summary = json.loads((REFERENCE / 'summary.json').read_text()) cases = [json.loads((REFERENCE / 'incumbent.json').read_text())] + [m['best_ranked'] for m in summary['models']] case_names = ['frozen_incumbent'] + [m['model'] for m in summary['models']] centers = select_centers(old_rows, cases, module.PARAM_COLS) write_json(output / 'frozen_centers.json', dict(case_names=case_names, centers=centers)) cache = {} incumbent['source'] = 'current' baseline, _ = replay(module, incumbent, cache) write_json(output / 'incumbent.json', baseline) results, comparisons, center_summaries, events = [], [], [], [] for number, center in enumerate(centers): params = dict(center['params'], source='diagnostic_center') base_row, base = replay(module, params, cache) # The experiment only changes the fingerprint and identity, never center metrics. original = next(r for r in old_rows if all(r[k] == params[k] for k in module.PARAM_COLS)) for key in ('Total P&L %', 'Max Drawdown %', 'Calmar Ratio', 'Fragile_0.05', 'Total Trades'): np.testing.assert_allclose(base_row[key], original[key], rtol=1e-12, atol=1e-10) center_id = center['id'] base_row.update(center_id=center_id, threshold=None, delta=0.) base_row['gates'] = gate_decisions(module, base_row, baseline) results.append(base_row) arrays = {f'center_{key}': value for key, value in trace_arrays(base).items()} arrays['time_ns'] = base.time.to_numpy(dtype='datetime64[ns]').astype('int64') for event in proximity_events(base, params, module): events.append(dict(center_id=center_id, **event)) changes = [] for threshold in THRESHOLDS: for delta in DELTAS: changed_params = dict(params, **{threshold: params[threshold] + delta}) row, trace = replay(module, changed_params, cache) row.update(center_id=center_id, threshold=threshold, delta=delta) row['gates'] = gate_decisions(module, row, baseline) results.append(row) diff = dict(center_id=center_id, threshold=threshold, delta=delta, **sensitivity(base, trace, base_row, row, module)) comparisons.append(diff) changes.append(diff) tag = f'{THRESHOLDS.index(threshold)}_{DELTAS.index(delta)}' arrays.update({f'{tag}_{key}': value for key, value in trace_arrays(trace).items()}) np.savez_compressed(output / f'{center_id}.npz', **arrays) center_summaries.append(dict(center_id=center_id, model=Path(params['mlp_weights_file']).stem, sample=center['sample'], cases=[case_names[i] for i in center['cases']], flagged=base_row['Fragile_0.05'] > 0, flagged_count=base_row['Fragile_0.05'], sensitivity={str(magnitude): dict( any_actions=any(x['IS']['changed_action_bars'] > 0 for x in changes if abs(x['delta']) == magnitude), any_material=any(x['IS']['material'] for x in changes if abs(x['delta']) == magnitude), max_action_bars=max(x['IS']['changed_action_bars'] for x in changes if abs(x['delta']) == magnitude)) for magnitude in (.05, 1.)})) if number % 10 == 0: print(f'Completed centers {number+1}/{len(centers)}', flush=True) write_json(output / 'results.json', results) write_json(output / 'sensitivity.json', comparisons) write_json(output / 'center_summary.json', center_summaries) write_json(output / 'proximity_events.json', events) finish(output, module, snapshot, protected, references, len(results) + 1) def run_entry_gates(output, prior): if prior is None or not json.loads((prior / 'audit.json').read_text()).get('passed'): raise ValueError('Verified fragility study must finish first') module, incumbent, snapshot, protected, references, _ = setup(output, 'entry_gates') incumbent['source'] = 'current' canonical, base = replay(module, incumbent, {}) module._score_to_signals = gate_signal(0) disabled, control = replay(module, incumbent, {}) assert canonical == disabled, 'Disabled gate metrics differ' for key in TRACE_KEYS: np.testing.assert_array_equal(base[key], control[key]) results = [dict(canonical, Policy='baseline')] base.to_csv(output / 'baseline_bars.csv', index=False) for width in (2, 3, 4): module._score_to_signals = gate_signal(width) row, trace = replay(module, dict(incumbent, source=f'higher_low_{width}'), {}, f'higher_low_{width}') row['gates'] = gate_decisions(module, row, canonical) row['sensitivity'] = sensitivity(base, trace, canonical, row, module) results.append(row) trace.to_csv(output / f'higher_low_{width}_bars.csv', index=False) write_json(output / 'results.json', results) write_json(output / 'control_verification.json', dict(actions_returns_metrics_equal=True, prior_audit_sha256=digest((prior / 'audit.json').read_bytes()))) finish(output, module, snapshot, protected, references, 5) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('stage', choices=['fragility', 'entry-gates']) parser.add_argument('--output', required=True, type=Path) parser.add_argument('--prior-fragility', type=Path) args = parser.parse_args() if args.stage == 'fragility': run_fragility(args.output.resolve()) else: run_entry_gates(args.output.resolve(), args.prior_fragility) if __name__ == '__main__': main()