"""Isolated, preregistered BTC4H chronological pipeline comparison; never promotes.""" from __future__ import annotations import argparse import hashlib import importlib.metadata import json import platform from pathlib import Path import signal import subprocess import sys import time import numpy as np import pandas as pd ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT)) from strategies.mlp_accounting import net_returns, window_metrics from strategies.strategy_activation_scores import _prepare_features from strategies.strategy_mlp_scores import FEATURE_COLS, _score_to_signals, mlp_forward, save_mlp_artifact, load_mlp_artifact from tools.train_mlp import build_target, pretrain, flatten_layers, unflatten_layers, run_cma, THRESHOLD_SCALE SEEDS = (101, 202) ARMS = ('expanding', 'three_year', 'five_year') YEARS = (2023, 2024, 2025, 2026) BUDGET = dict(fits=24, epochs=60, generations=8, population=8, thresholds=128, timeout_seconds=1800) PROTOCOL = ROOT / 'docs/mlp_pipeline_comparison_protocol_2026_09_21.md' def digest(path): return hashlib.sha256(Path(path).read_bytes()).hexdigest() def write_json(path, value): path = Path(path) path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(value, indent=2, default=str, allow_nan=False) + '\n') def split_plan(year, arm): origin = pd.Timestamp(year=year, month=1, day=1) start = pd.Timestamp('2017-12-01') if arm == 'expanding' else origin - pd.DateOffset(years=3 if arm == 'three_year' else 5) return dict(start=start, origin=origin, end=origin + pd.DateOffset(years=1), calibration=origin - pd.DateOffset(months=6), validation=origin - pd.DateOffset(months=9), folds=[(origin - pd.DateOffset(months=18), origin - pd.DateOffset(months=12)), (origin - pd.DateOffset(months=12), origin - pd.DateOffset(months=6))]) def load_inputs(path): df = pd.read_csv(path) df.columns = df.columns.str.lower().str.strip() if df.columns.duplicated().any(): raise ValueError('Duplicate CSV columns') df['time'] = pd.to_datetime(df.time, unit='s', utc=True).dt.tz_localize(None) if pd.api.types.is_numeric_dtype(df.time) else pd.to_datetime(df.time, utc=True).dt.tz_localize(None) dropped = str(df.time.iloc[-1]) df = df.iloc[:-1].copy().reset_index(drop=True) required = ['close', 'high', *FEATURE_COLS] missing = sorted(set(required) - set(df.columns)) if missing: raise ValueError(f'Missing feature/price columns: {missing}') if df.time.isna().any() or not df.time.is_monotonic_increasing or df.time.duplicated().any(): raise ValueError('Invalid timestamps') finite = np.isfinite(df[required].to_numpy(dtype=float)).all(axis=1) if not finite.any(): raise ValueError('No finite rows') df = df.iloc[np.flatnonzero(finite)[0]:].reset_index(drop=True) if not np.isfinite(df[required].to_numpy(dtype=float)).all() or (df[['close', 'high']] <= 0).any().any(): raise ValueError('Nonfinite interior feature or invalid price') if df.time.iloc[0] > pd.Timestamp('2017-12-01') or df.time.iloc[-1] < pd.Timestamp('2026-01-02'): raise ValueError('Insufficient historical coverage') return df, dropped def params_for(thresholds, start): return dict(i_long_entry_activation_threshold=float(thresholds[0]), i_long_exit_activation_threshold=float(thresholds[1]), i_long_exit_activation_confirmation_threshold=float(thresholds[2]), i_trailing_stop_threshold=float(thresholds[3]), i_use_long_exit_confirmation=True, i_use_long_entry_confirmation=False, i_exit_score_window=1, i_entry_score_window=1, i_mvrv_suppress_bear=False, i_regime_window=0, _pine_time_start=str(start)) def signals_for(frame, layers, thresholds, start, params=None): d = frame.copy() X = _prepare_features(d, FEATURE_COLS) d['activation_score'] = mlp_forward(X, layers) p = params_for(thresholds, start) if params is None else {**params, '_pine_time_start': str(start)} return _score_to_signals(d, p, X[:, 9], d.close.to_numpy(float), d.high.to_numpy(float)) def measured(signals, start, end): return window_metrics(signals, start=start, end=pd.Timestamp(end) - pd.Timedelta(nanoseconds=1)) def selection_rank(metrics): return (metrics['calmar'] if metrics['calmar'] is not None else 0., metrics['pnl_pct'], metrics['dd_pct']) def threshold_candidates(recommended): rng = np.random.default_rng(20260921) yield np.asarray(recommended, dtype=float) for _ in range(BUDGET['thresholds'] - 1): yield np.array([rng.uniform(-600, 600), rng.uniform(-750, 750), rng.uniform(-750, 750), rng.choice(np.arange(0, 50, 5))]) def fit_seed(frame, plan, seed, out): fit = frame[(frame.time >= plan['start']) & (frame.time < plan['calibration'])].reset_index(drop=True) X = _prepare_features(fit, FEATURE_COLS) y, valid = build_target(fit, 10) log = lambda message: print(message, flush=True) layers, meta = pretrain(X, y, valid, fit.time, [16, 8], BUDGET['epochs'], seed, plan['validation'], fit.time.iloc[-1], log, target_horizon=10, device_override='cpu') arch = [len(FEATURE_COLS), 16, 8, 1] theta_w = flatten_layers(layers) scores = mlp_forward(X, layers) thresholds = np.r_[np.percentile(scores, [60, 75, 30]), 0.] theta0 = np.r_[theta_w, thresholds / THRESHOLD_SCALE] calls = [0] def fitness(theta): calls[0] += 1 fitted = unflatten_layers(theta[:len(theta_w)], arch) thr = np.clip(theta[len(theta_w):] * THRESHOLD_SCALE, [-900, -900, -900, 0], [900, 900, 900, 30]) values = [] for start, end in plan['folds']: sig = signals_for(fit[fit.time < end], fitted, thr, start) metric = measured(sig, start, end) values.append(metric['calmar'] if metric['calmar'] is not None else 0.) objective = -float(np.mean(values)) + .001 * float(np.mean((theta[:len(theta_w)] - theta_w) ** 2)) if not np.isfinite(objective): raise ValueError('Nonfinite fitting objective') return objective, values, thr best = run_cma(fitness, theta0, .05, BUDGET['population'], BUDGET['generations'], seed, log, workers=1) layers = unflatten_layers(best['theta'][:len(theta_w)], arch) artifact = out / f'seed{seed}.json' stop_reason = 'generation_budget' if calls[0] == BUDGET['generations'] * BUDGET['population'] else 'cma_convergence' save_mlp_artifact(str(artifact), layers, asset='COINBASE_BTCUSD', timeframe='4H', feature_cols=FEATURE_COLS, training=dict(seed=seed, phase1=meta, splits=serial_plan(plan), target_k=10, cma_evaluations=calls[0], cma_stop_reason=stop_reason)) calibration_frame = frame[(frame.time >= plan['start']) & (frame.time < plan['origin'])].copy() rows = [] for index, thr in enumerate(threshold_candidates(best['thr'])): metrics = measured(signals_for(calibration_frame, layers, thr, plan['calibration']), plan['calibration'], plan['origin']) rows.append(dict(seed=seed, index=index, thresholds=thr.tolist(), metrics=metrics)) write_json(out / f'calibration_seed{seed}.json', rows) chosen = max(rows, key=lambda row: selection_rank(row['metrics'])) return {**chosen, 'artifact': str(artifact), 'sha256': digest(artifact), 'cma_evaluations': calls[0], 'cma_stop_reason': stop_reason, 'epochs': meta['epochs_run']} def serial_plan(plan): return {key: [[str(a), str(b)] for a, b in value] if key == 'folds' else str(value) for key, value in plan.items()} def freeze_selection(path, candidates, plan): chosen = max(candidates, key=lambda row: selection_rank(row['metrics'])) payload = dict(selected=chosen, candidates=candidates, splits=serial_plan(plan), protocol_sha256=digest(PROTOCOL)) write_json(path, payload) Path(str(path) + '.sha256').write_text(digest(path) + '\n') return json.loads(Path(path).read_text()) def baseline_signals(frame, start, kind): d = frame.copy() eligible = d.time >= start if kind == 'trend': daily = d.set_index('time').close.resample('1D').last() trend = (daily > daily.rolling(200, min_periods=200).mean()).shift(1).fillna(False) desired = d.time.dt.normalize().map(trend).fillna(False).to_numpy(bool) & eligible.to_numpy() else: desired = eligible.to_numpy() & (kind == 'buy_hold') prior = np.r_[False, desired[:-1]] d['position'] = desired.astype(int) d['execute_entry'] = desired & ~prior d['execute_exit'] = ~desired & prior return d def daily_returns(signals, start, end): values = net_returns(signals) mask = (signals.time >= start) & (signals.time < end) s = pd.Series(values[mask].to_numpy(), index=signals.loc[mask, 'time']) return (1 + s).resample('1D').prod() - 1 def run(args): out = Path(args.out) if out.exists() and any(out.iterdir()): raise ValueError('Refusing to overwrite nonempty experiment directory; choose a new --out') out.mkdir(parents=True, exist_ok=True) df, dropped = load_inputs(args.data) import torch torch.set_num_threads(1) torch.use_deterministic_algorithms(True) gates = {} for script in ('sync_params.py', 'check_pine.py', 'validate_chart_data.py'): result = subprocess.run([sys.executable, str(ROOT / 'tools' / script)], cwd=ROOT, capture_output=True, text=True) gates[script] = dict(returncode=result.returncode, stdout=result.stdout, stderr=result.stderr) write_json(out / 'gates.json', gates) if result.returncode: raise ValueError(f'Startup gate failed: {script}; see gates.json') code = ['config.py', 'tools/run_mlp_pipeline_comparison.py', 'tools/train_mlp.py', 'strategies/mlp_accounting.py', 'strategies/strategy_mlp_scores.py', 'strategies/strategy_activation_scores.py'] manifest = dict(data=str(args.data), data_sha256=digest(args.data), protocol_sha256=digest(PROTOCOL), code_hashes={name: digest(ROOT / name) for name in code}, environment=dict(python=sys.version, platform=platform.platform(), device='cpu', deterministic_algorithms=True, versions={name: importlib.metadata.version(name) for name in ('numpy', 'pandas', 'torch', 'cma')}), feature_cols=FEATURE_COLS, budget=BUDGET, seeds=SEEDS, arms=ARMS, years=YEARS, rows=len(df), first=str(df.time.iloc[0]), last=str(df.time.iloc[-1]), dropped_final_row=dropped, plans=[dict(arm=arm, **serial_plan(split_plan(year, arm))) for year in YEARS for arm in ARMS]) write_json(out / 'input_manifest.json', manifest) if args.validate_only: print('Input and chronological budget validation passed; no training performed.') return start_clock = time.monotonic() def timeout(*_): raise TimeoutError('Preregistered30-minute whole-comparison deadline exceeded') old_handler = signal.signal(signal.SIGALRM, timeout) signal.alarm(BUDGET['timeout_seconds']) rows, returns = [], {} try: # Freeze every chronological selection before evaluating any outer candidate. frozen = [] for year in YEARS: for arm in ARMS: plan = split_plan(year, arm) cell = out / f'{year}_{arm}' cell.mkdir(exist_ok=True) candidates = [fit_seed(df[df.time < plan['origin']], plan, seed, cell) for seed in SEEDS] frozen.append((year, arm, plan, freeze_selection(cell / 'selection.json', candidates, plan))) winner = pd.read_csv(ROOT / 'results/winners/optimization_winner_strategy_mlp_scores_COINBASE_BTCUSD_4H.csv').iloc[0].to_dict() incumbent_path = ROOT / winner['mlp_weights_file'] incumbent = load_mlp_artifact(str(incumbent_path)) if incumbent['feature_cols'] != FEATURE_COLS: raise ValueError('Incumbent schema differs from experiment') write_json(out / 'incumbent_reference.json', dict(params=winner, artifact=str(incumbent_path), sha256=digest(incumbent_path), label='hindsight-selected')) for year, arm, plan, freeze in frozen: selection_path = out / f'{year}_{arm}' / 'selection.json' if digest(selection_path) != Path(str(selection_path) + '.sha256').read_text().strip(): raise ValueError('Frozen selection changed') if any(digest(ROOT / name) != value for name, value in manifest['code_hashes'].items()): raise ValueError('Source code changed during experiment') chosen = freeze['selected'] if digest(chosen['artifact']) != chosen['sha256']: raise ValueError('Frozen artifact changed') frame = df[df.time < plan['end']].copy() layers = load_mlp_artifact(chosen['artifact'])['layers'] sig = signals_for(frame, layers, chosen['thresholds'], plan['origin']) rows.append(dict(year=year, arm=arm, **measured(sig, plan['origin'], plan['end']))) returns[f'{year}_{arm}'] = daily_returns(sig, plan['origin'], plan['end']) if arm == ARMS[0]: for name in ('cash', 'buy_hold', 'trend', 'incumbent_hindsight'): base = signals_for(frame, incumbent['layers'], chosen['thresholds'], plan['origin'], winner) if name == 'incumbent_hindsight' else baseline_signals(frame, plan['origin'], name) rows.append(dict(year=year, arm=name, **measured(base, plan['origin'], plan['end']))) returns[f'{year}_{name}'] = daily_returns(base, plan['origin'], plan['end']) write_json(out / 'results.json', dict(status='complete', runtime_seconds=time.monotonic()-start_clock, rows=rows)) pd.DataFrame(rows).to_csv(out / 'results.csv', index=False) pd.DataFrame(returns).to_csv(out / 'daily_returns.csv', index_label='date') except Exception as exc: write_json(out / 'status.json', dict(status='failed', reason=str(exc), runtime_seconds=time.monotonic()-start_clock)) raise finally: signal.alarm(0) signal.signal(signal.SIGALRM, old_handler) if __name__ == '__main__': parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--data', default=str(ROOT / 'data/mlp/COINBASE_BTCUSD, 240.csv')) parser.add_argument('--out', default=str(ROOT / 'results/mlp_measurement_audit_2026_09_21/pipeline')) parser.add_argument('--validate-only', action='store_true') run(parser.parse_args())