"""Exactly two frozen policies; retrospective evidence only, never promotion.""" from __future__ import annotations import argparse from dataclasses import replace import importlib.util 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 COMMISSION_RATE, net_returns from strategies.mlp_comparison import digest, fingerprint, require_replay from strategies.mlp_entry_loss import entry_loss_signals from strategies.strategy_mlp_scores import _score_to_signals MODEL_HASH = '91fa9223c87c489c24ec1d4711099c92fe331d189f660e7f080ee596c052763b' FROZEN = dict(i_long_entry_activation_threshold=90., i_long_exit_activation_threshold=210., i_long_exit_activation_confirmation_threshold=-260., i_trailing_stop_threshold=20., i_use_long_entry_confirmation=False, i_use_long_exit_confirmation=True, i_entry_score_window=1, i_exit_score_window=1, i_mvrv_suppress_bear=False, i_regime_window=0, i_regime_entry_min_score=-1000.) EXPERIMENT_SOURCES = ['strategies/mlp_entry_loss.py', 'tools/run_mlp_entry_loss_experiment.py', 'tests/test_mlp_entry_loss.py', 'docs/mlp_entry_loss_protocol_2026_09_21.md'] def evaluator(snapshot=None, signal=None): """Load a private evaluator instance; never mutate shared module globals.""" spec = importlib.util.spec_from_file_location('_entry_loss_private_sweep', REPO / 'tools/run_mlp_deep_sweep.py') module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) module._COMPARISON = snapshot if signal is not None: module._score_to_signals = signal return module def evaluate_policy(snapshot, params, loss_fraction, *, canonical=False): signal = _score_to_signals if canonical else lambda *args: entry_loss_signals(*args, loss_fraction=loss_fraction) module = evaluator(snapshot, signal) cache = {} row = module.evaluate('6H', params, cache) scored, stoch, close, high = next(iter(cache.values())) trace = signal(scored.copy(), dict(params, _pine_time_start=module.SCORE_START), stoch, close, high) trace['net_return'] = net_returns(trace) trace['fee_return'] = COMMISSION_RATE * (trace.execute_entry.shift(1, fill_value=False).astype(float) + trace.execute_exit.astype(float)) trace['held_interval'] = trace.position.shift(1, fill_value=0) return row, trace def assert_control(canonical, disabled): left, a = canonical right, b = disabled if json.dumps(left, sort_keys=True) != json.dumps(right, sort_keys=True): raise AssertionError('Disabled control metrics differ from canonical evaluator') for key in ('position', 'execute_entry', 'execute_exit', 'net_return', 'fee_return', 'held_interval'): np.testing.assert_array_equal(a[key], b[key], err_msg=key) def identify(row, snapshot, policy): row = dict(row, Policy=policy) row['Evaluation_Key'] = fingerprint(dict(context=snapshot.fingerprint, artifact=row['Artifact_SHA256'], params={k: row[k] for k in evaluator().PARAM_COLS}, policy=policy)) return row def diagnostics(trace, baseline): """Keep each event, censor missing horizons, and retain both actual paths.""" entries = np.flatnonzero(trace.execute_entry) exits = np.flatnonzero(trace.execute_exit) trades = [] for entry in entries: following = exits[exits > entry] end = int(following[0]) if len(following) else len(trace) - 1 completed = bool(len(following)) trades.append(dict(entry_bar=int(entry), end_bar=end, completed=completed, entry_time=str(trace.time.iloc[entry]), end_time=str(trace.time.iloc[end]), net_pct=float((np.prod(1 + trace.net_return.iloc[entry + 1:end + 1]) - 1) * 100), fee_return_sum=float(trace.fee_return.iloc[entry + 1:end + 1].sum()))) events = [] for i in exits: later = entries[entries > i] flags = {key: bool(trace[key].iloc[i]) for key in ('score_exit_flag', 'trailing_exit_flag', 'entry_loss_flag')} item = dict(bar=int(i), time=str(trace.time.iloc[i]), **flags, simultaneous_flags=sum(flags.values()), reentry_bar=int(later[0]) if len(later) else None, reentry_delay_hours=float((trace.time.iloc[later[0]] - trace.time.iloc[i]).total_seconds()/3600) if len(later) else None, reentry_censored=not bool(len(later)), overshoot_pct=float(trace.entry_loss_overshoot_pct.iloc[i]) if flags['entry_loss_flag'] else None) # Loss-only exits are causal additional exits for this position. Also # retain whether the baseline actually exits: its path can already differ. item['candidate_specific_loss_exit'] = flags['entry_loss_flag'] and not flags['score_exit_flag'] and not flags['trailing_exit_flag'] item['baseline_exit_same_bar'] = bool(baseline.execute_exit.iloc[i]) item['recoveries'] = [] if item['candidate_specific_loss_exit']: for days, bars in ((1, 4), (4, 16), (7, 28)): end = int(i + bars) horizon = trace.iloc[i:min(end + 1, len(trace))] gap = bool((horizon.time.diff().dropna() != pd.Timedelta(hours=6)).any()) censor = 'end_of_data' if end >= len(trace) else 'gap' if gap else None rec = dict(days=days, bars=bars, censored=censor, gap=gap) if censor is None: rec.update(close_return_pct=float((trace.close.iloc[end]/trace.close.iloc[i]-1)*100), max_close_return_pct=float((horizon.close.max()/trace.close.iloc[i]-1)*100), recovered_entry=bool((horizon.close >= trace.entry_signal_close.iloc[i]).any()), candidate_path_net_pct=float((np.prod(1+trace.net_return.iloc[i+1:end+1])-1)*100), baseline_path_net_pct=float((np.prod(1+baseline.net_return.iloc[i+1:end+1])-1)*100)) item['recoveries'].append(rec) events.append(item) return dict(trades=trades, exit_events=events, entries=int(trace.execute_entry.sum()), exits=int(trace.execute_exit.sum()), fee_return_sum=float(trace.fee_return.sum()), flag_counts={key: int((trace.execute_exit & trace[key]).sum()) for key in ('score_exit_flag','trailing_exit_flag','entry_loss_flag')}) def gate_decisions(module, row, baseline): conditional = (row['OOS_Total Trades'] >= module.OOS_MIN_TRADES and baseline['OOS_Total Trades'] >= module.OOS_MIN_TRADES and row['OOS_Bear_Pct'] < module.OOS_BEAR_PCT_SKIP) return dict(positive_eligible_pnl_dd=row['P&L/DD Ratio'] > 0, fragility=row['Fragile_0.05'] == 0, positive_sub1=row['Sub1_Calmar'] > 0, positive_sub2=row['Sub2_Calmar'] > 0, wfo=row['WFO_Neg_Folds'] == 0, common_contract=row['Comparison_Fingerprint'] == baseline['Comparison_Fingerprint'], complete=row['Evaluation_Status'] == baseline['Evaluation_Status'] == 'complete', incumbent_calmar=row['Calmar Ratio'] >= baseline['Calmar Ratio'], incumbent_pnl_dd=row['P&L/DD Ratio'] >= baseline['P&L/DD Ratio'], oos_sortino_gate_applies=bool(conditional), oos_sortino=(not conditional or row['OOS_Sortino Ratio'] >= baseline['OOS_Sortino Ratio']-module.OOS_SORTINO_DELTA), qualifies=module.promotable(row, baseline)) def hashes(paths): return {str(path.relative_to(REPO)): digest(path.read_bytes()) for path in sorted(set(paths)) if path.is_file()} def protected_hashes(): return hashes(list((REPO/'strategies/params').rglob('*')) + list((REPO/'results/winners').rglob('*')) + list((REPO/'strategies').rglob('*.pine')) + list((REPO/'libraries').rglob('*.pine')) + list((REPO/'indicators').rglob('*.pine')) + [REPO/'config.py']) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--output', type=Path, required=True) args = parser.parse_args() output = args.output.resolve() if output.exists() and any(output.iterdir()): raise ValueError('Output must be new or empty') module = evaluator() params = module.normalise_param(module.load_winner('6H')) if any(params[key] != value for key, value in FROZEN.items()): raise ValueError('Frozen incumbent parameters drifted') if 'bb50_seed4942' not in params['mlp_weights_file'] or digest(Path(params['mlp_weights_file']).read_bytes()) != MODEL_HASH: raise ValueError('Frozen model drifted') protected = protected_hashes() preflights = {} for script in ('tools/sync_params.py', 'tools/check_pine.py'): run = subprocess.run([sys.executable, script], cwd=REPO, text=True, capture_output=True) preflights[script] = dict(returncode=run.returncode, stdout=run.stdout, stderr=run.stderr) if run.returncode: raise RuntimeError(f'Preflight failed: {script}\n{run.stdout}\n{run.stderr}') snapshot = module.capture_comparison('6H', [params['mlp_weights_file']]) manifest = json.loads(snapshot.manifest_json) reference_path = REPO/'results/mlp_threshold_comparison_2026_09_21/final/comparison_manifest.json' reference = json.loads(reference_path.read_bytes()) if manifest['data_sha256'] != reference['data_sha256']: raise ValueError('Data differs from completed threshold comparison') if manifest['contract'] != reference['contract']: raise ValueError('Measurement contract differs from completed threshold comparison') if manifest['source_hashes'] != reference['source_hashes']: raise ValueError('Evaluator sources differ from completed threshold comparison') manifest['reference_manifest_sha256'] = digest(reference_path.read_bytes()) manifest['source_hashes'].update(hashes([REPO/p for p in EXPERIMENT_SOURCES])) manifest['contract']['entry_loss_experiment'] = dict(policies=['baseline', 'entry_loss_10pct'], loss_fraction=0.10, original_trail_pct=20, protocol='docs/mlp_entry_loss_protocol_2026_09_21.md') snapshot = replace(snapshot, manifest_json=json.dumps(manifest, sort_keys=True, separators=(',', ':'))) params['source'] = 'current' canonical = evaluate_policy(snapshot, params, None, canonical=True) disabled = evaluate_policy(snapshot, params, None) assert_control(canonical, disabled) baseline, base_trace = disabled baseline = identify(require_replay(baseline), snapshot, 'baseline') candidate, trace = evaluate_policy(snapshot, dict(params, source='entry_loss_10pct'), 0.10) candidate = identify(require_replay(candidate), snapshot, 'entry_loss_10pct') gates = gate_decisions(module, candidate, baseline) snapshot.validate_sources(REPO) if (protected != protected_hashes() or digest(module.data_path('6H').read_bytes()) != manifest['data_sha256'] or digest(reference_path.read_bytes()) != manifest['reference_manifest_sha256']): raise ValueError('Protected or input files drifted') output.mkdir(parents=True, exist_ok=True) payloads = dict(comparison_manifest=manifest, protected_hashes=protected, preflights=preflights, results=[baseline, candidate], gates=gates, diagnostics=dict(baseline=diagnostics(base_trace, base_trace), candidate=diagnostics(trace, base_trace)), verification=dict(control_actions_returns_metrics_equal=True, protected_unchanged=True, source_input_hashes_unchanged=True, distinct_evaluation_keys=baseline['Evaluation_Key'] != candidate['Evaluation_Key'], policies_evaluated=2, promotions=0, retrospective=True)) for name, payload in payloads.items(): (output/f'{name}.json').write_text(json.dumps(payload, indent=2, default=str)+'\n') base_trace.to_csv(output/'baseline_bars.csv', index=False) trace.to_csv(output/'entry_loss_10pct_bars.csv', index=False) print(json.dumps(gates, indent=2)) if __name__ == '__main__': main()