"""Bounded BTC 4H ATR-stop screen. No promotion or production writes.""" from __future__ import annotations import argparse import json from pathlib import Path import platform import subprocess import sys import numpy as np import pandas as pd REPO = Path(__file__).resolve().parent.parent sys.path.insert(0, str(REPO)) from config import SCORE_START, WFO_FOLDS, OOS_START from strategies.mlp_volatility_stop import event_study, replay_stop, volatility from strategies.strategy_mlp_scores import _score_to_signals from tools.run_mlp_learned_exit_experiment import load_input from tools.run_mlp_state_policy_experiment import digest, window_metrics PROTOCOL = 'docs/mlp_volatility_stop_protocol_2026_09_21.md' TRAIN_END = '2020-12-31' ATR_NAMES = ('atr_5', 'atr_7_5') def calibrate(frame: pd.DataFrame, opportunities: dict, vol: pd.DataFrame) -> tuple[dict, dict]: """Only training bars enter any estimated constant, including control widths.""" train = (frame.time >= pd.Timestamp(SCORE_START)) & (frame.time < pd.Timestamp('2021-01-01')) values = vol.atr_price[train].to_numpy() if not len(values) or not np.isfinite(values).all() or np.median(values) <= 0: raise ValueError('Invalid training volatility') baseline, _ = replay_stop(frame, opportunities, start=SCORE_START, end=TRAIN_END) active = train & baseline.position.shift(1, fill_value=0).astype(bool) if not active.any(): raise ValueError('No training holding bars for matched controls') median = float(np.median(values)) variants = {'baseline': None} details = {} for name, target in zip(ATR_NAMES, (.05, .075)): scale = target/median widths = np.clip(vol.atr_price.to_numpy()*scale, .025, .10) matched = float(pd.Series(widths, index=frame.index).shift(1)[active].mean()) variants[name] = widths variants['fixed_'+name] = np.full(len(frame), matched) details[name] = {'target_median': target, 'scale': scale, 'matched_fixed_width': matched} return variants, {'start': SCORE_START, 'end': TRAIN_END, 'bars': int(train.sum()), 'holding_bars': int(active.sum()), 'atr_price_median': median, 'min_width': .025, 'max_width': .10, 'max_threshold_loss': opportunities['trail'], 'candidates': details} def nominate(metrics: pd.DataFrame) -> dict: """Frozen validation-only rule; test/later rows are never read.""" val = metrics[metrics.stage == 'validation'].set_index(['window', 'variant']) eligible, audits = [], {} for name in ATR_NAMES: candidate = val.xs(name, level='variant') checks = {'enough_entries': bool(candidate.entries.sum() >= 20)} for control in ('baseline', 'fixed_'+name): reference = val.xs(control, level='variant') delta = candidate[['pnl_pct', 'calmar', 'dd_pct']]-reference[['pnl_pct', 'calmar', 'dd_pct']] checks[control] = {'median_pnl_delta': float(delta.pnl_pct.median()), 'median_calmar_delta': float(delta.calmar.median()), 'pnl_wins': int((delta.pnl_pct > 0).sum()), 'worst_dd_delta': float(delta.dd_pct.min())} passed = checks['enough_entries'] and all( x['median_pnl_delta'] > 0 and x['median_calmar_delta'] > 0 and x['pnl_wins'] >= 2 and x['worst_dd_delta'] >= -2 for key, x in checks.items() if key != 'enough_entries') checks['eligible'] = bool(passed) audits[name] = checks if passed: eligible.append(name) eligible.sort(key=lambda name: (-audits[name]['fixed_'+name]['median_calmar_delta'], ATR_NAMES.index(name))) return {'nominated': eligible[0] if eligible else None, 'validation_only': True, 'checks': audits} def save_json(path: Path, value: dict) -> None: path.write_text(json.dumps(value, indent=2, allow_nan=False)+'\n') def run(output: Path) -> pd.DataFrame: output.mkdir(parents=True, exist_ok=True) if (output/'manifest.json').exists(): raise ValueError('Use a new output directory to preserve previous run evidence') checks = {} for tool in ('sync_params.py', 'check_pine.py', 'validate_chart_data.py'): check = subprocess.run([sys.executable, str(REPO/'tools'/tool)], cwd=REPO, capture_output=True, text=True) checks[tool] = {'returncode': check.returncode, 'stdout': check.stdout, 'stderr': check.stderr} save_json(output/'prerequisites.json', checks) check.check_returncode() frame, opportunities, provenance = load_input('4H') if opportunities['trail'] != .05: raise ValueError('Protocol requires the frozen 5% baseline; do not silently change design') vol = volatility(frame) variants, calibration = calibrate(frame, opportunities, vol) save_json(output/'calibration.json', calibration) provenance['fresh'] = False provenance['untouched'] = False rows, traces, events = [], [], [] windows = [('training', 'training', SCORE_START, TRAIN_END)] for i, (_, start, end) in enumerate(WFO_FOLDS): windows.append(('validation' if i < 3 else 'historical_evaluation', start[:7], start, end)) windows.append(('previously_inspected_later', 'later_2026', OOS_START, str(frame.time.iloc[-1].date()))) selection_hash = None for stage, label, start, end in windows: # Freeze nomination before replaying ANY later candidate windows. if stage == 'historical_evaluation' and selection_hash is None: save_json(output/'selection.json', nominate(pd.DataFrame(rows))) selection_hash = digest(output/'selection.json') baseline_returns = None for name, widths in variants.items(): signals, decisions = replay_stop(frame, opportunities, start=start, end=end, widths=widths) mask = (frame.time >= pd.Timestamp(start)) & (frame.time < pd.Timestamp(end)+pd.Timedelta(days=1)) if name == 'baseline': params = {**provenance['params'], '_pine_time_start': start, '_pine_time_end': str(pd.Timestamp(end)+pd.Timedelta(days=1)-pd.Timedelta(nanoseconds=1))} canonical = _score_to_signals(frame.copy(), params, frame.stoch_peak_norm.to_numpy(), frame.close.to_numpy(), frame.high.to_numpy()) control, _ = replay_stop(frame, opportunities, start=start, end=end, widths=np.full(len(frame), opportunities['trail'])) for col in ('position', 'execute_entry', 'execute_exit'): np.testing.assert_array_equal(signals[col], canonical[col]) np.testing.assert_array_equal(signals[col], control[col]) np.testing.assert_array_equal(signals.strategy_return, control.strategy_return) assert window_metrics(canonical, start, end) == window_metrics(signals, start, end) baseline_returns = signals.strategy_return.to_numpy() # Every candidate trace must obey monotonicity and the entry floor. active_trace = signals.loc[mask & signals.stop_price.notna()] for _, trade in active_trace.groupby('entry_bar'): assert (trade.stop_price.diff().dropna() >= -1e-10).all() assert (trade.stop_price >= frame.close.iloc[int(trade.entry_bar.iloc[0])]*.95-1e-10).all() metrics = window_metrics(signals, start, end) rows.append({'stage': stage, 'window': label, 'start': start, 'end': end, 'variant': name, **metrics, 'stop_exits': sum(d['stop_hit'] for d in decisions), 'score_only_exits': sum(not d['stop_hit'] for d in decisions), 'both_exits': sum(d['stop_hit'] and d['score_exit'] for d in decisions), 'threshold_overshoots': sum(d['threshold_overshoot_pct'] > 1e-10 for d in decisions), 'entry_bound_breaches': sum(d['entry_bound_breach_pct'] > 1e-10 for d in decisions), 'open_at_end': bool(signals.loc[mask, 'position'].iloc[-1]), 'terminal_unfilled_entry': bool(signals.loc[mask, 'execute_entry'].iloc[-1])}) trace = signals.loc[mask, ['time', 'close', 'position', 'execute_entry', 'execute_exit', 'entry_bar', 'trade_high', 'stop_price', 'requested_width', 'strategy_return']].copy() trace['bar'] = np.flatnonzero(mask) trace['stage'], trace['window'], trace['variant'] = stage, label, name trace['paired_bar_return'] = signals.strategy_return.to_numpy()[mask]-baseline_returns[mask] traces.append(trace) event = event_study(frame, decisions, vol, end) if len(event): exit_details = pd.DataFrame(decisions)[['bar', 'threshold_overshoot_pct', 'entry_bound_breach_pct']] event = event.merge(exit_details, on='bar', validate='many_to_one') event['stage'], event['window'], event['variant'] = stage, label, name events.append(event) print(f'{stage}: {label} complete', flush=True) assert digest(output/'selection.json') == selection_hash metrics = pd.DataFrame(rows) paired = [] for name in variants: if name == 'baseline': continue for control in ['baseline'] + (['fixed_'+name] if name in ATR_NAMES else []): candidate = metrics[metrics.variant == name].set_index('window') reference = metrics[metrics.variant == control].set_index('window') delta = candidate[['pnl_pct', 'calmar', 'dd_pct', 'entries', 'exposure_pct']]-reference[['pnl_pct', 'calmar', 'dd_pct', 'entries', 'exposure_pct']] delta = delta.add_prefix('delta_').reset_index() delta['variant'], delta['control'] = name, control delta['stage'] = candidate.stage.to_numpy() paired.append(delta) metrics.to_csv(output/'metrics.csv', index=False) pd.concat(paired, ignore_index=True).to_csv(output/'paired_metrics.csv', index=False) pd.concat(traces, ignore_index=True).to_csv(output/'bar_traces.csv.gz', index=False, compression={'method': 'gzip', 'mtime': 0}) all_events = pd.concat(events, ignore_index=True) all_events.to_csv(output/'events.csv', index=False) summaries = [] for subset in ['all', 'nonoverlap']: valid = all_events[all_events.valid & (all_events.nonoverlap if subset == 'nonoverlap' else True)] grouped = valid.groupby(['stage', 'variant', 'reason', 'horizon_bars']).agg( count=('bar', 'size'), mean_terminal_pct=('terminal_return_pct', 'mean'), median_terminal_pct=('terminal_return_pct', 'median'), recovery_fraction=('terminal_return_pct', lambda x: float((x > 0).mean())), mean_loss_avoided_pct=('terminal_loss_avoided_pct', 'mean'), mean_missed_recovery_pct=('missed_recovery_pct', 'mean'), mean_decline_pct=('subsequent_decline_pct', 'mean'), median_holding_hours=('holding_hours', 'median'), median_entry_atr_pct=('entry_atr_pct', 'median'), median_exit_atr_pct=('exit_atr_pct', 'median'), median_entry_rv_pct=('entry_rv_pct', 'median'), median_exit_rv_pct=('exit_rv_pct', 'median')).reset_index() grouped['subset'] = subset summaries.append(grouped) pd.concat(summaries, ignore_index=True).to_csv(output/'event_summary.csv', index=False) # Hash the actual imported repository source, including previously dirty dependencies. sources = {PROTOCOL, 'tests/test_mlp_volatility_stop.py', str(Path(__file__).relative_to(REPO)), 'tools/sync_params.py', 'tools/check_pine.py', 'tools/validate_chart_data.py'} for module in tuple(sys.modules.values()): source = getattr(module, '__file__', None) if source: path = Path(source).resolve() if path.is_relative_to(REPO) and '.venv' not in path.parts and path.suffix == '.py': sources.add(str(path.relative_to(REPO))) outputs = ['prerequisites.json', 'calibration.json', 'selection.json', 'metrics.csv', 'paired_metrics.csv', 'bar_traces.csv.gz', 'events.csv', 'event_summary.csv'] save_json(output/'manifest.json', {'non_promoting': True, 'protocol': PROTOCOL, 'inputs': provenance, 'variants': list(variants), 'windows': windows, 'selection_sha256_before_evaluation': selection_hash, 'environment': {'python': platform.python_version(), 'numpy': np.__version__, 'pandas': pd.__version__}, 'source_sha256': {name: digest(REPO/name) for name in sorted(sources)}, 'outputs_sha256': {name: digest(output/name) for name in outputs}}) return metrics if __name__ == '__main__': parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--output', type=Path, default=REPO/'results/mlp_experiments/volatility_stop') run(parser.parse_args().output)