"""Fixed, non-promoting action-state ablations on existing BTC MLP winners. No fitting, threshold search, or promotion. Historical regime slices are not out-of-fold model validation: the frozen winners were trained through TRAIN_END. """ from __future__ import annotations import argparse from dataclasses import asdict import hashlib import json from pathlib import Path 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, TRAIN_END, OOS_START, WFO_FOLDS, WFO_MIN_OOS_TRADES from strategies.mlp_state_policy import StatePolicy from strategies.strategy_activation_scores import COMMISSION_RATE from strategies.strategy_mlp_scores import generate_signals, _score_to_signals from tools.run_mlp_score_structure_experiment import _data_path, _load_winner, _winner_path from tools.validate_chart_data import validate_file # Predeclared small ablation set. Same durations in bars across timeframes; # 4 and 12 bars mean one and three days at the primary 6H timeframe. VARIANTS = { 'baseline': (StatePolicy(), False), 'macro_now': (StatePolicy(), True), 'macro_memory_4': (StatePolicy(macro_memory=4), True), 'macro_memory_12': (StatePolicy(macro_memory=12), True), 'cooldown_4': (StatePolicy(exit_cooldown=4), False), 'cooldown_12': (StatePolicy(exit_cooldown=12), False), 'hold_4': (StatePolicy(min_hold=4), False), 'hold_12': (StatePolicy(min_hold=12), False), 'combined_4': (StatePolicy(macro_memory=4, exit_cooldown=4, min_hold=4), True), } def window_metrics(signals, start, end): """Calendar annualization; preserve carry-in P&L even with zero new entries. Matches canonical close-to-close returns and fee timing. Windows include the whole last day. Drawdown includes initial equity, including early loss. """ returns = signals.position.shift(1, fill_value=0) * signals.close.pct_change().fillna(0) returns -= COMMISSION_RATE * (signals.execute_entry.shift(1, fill_value=False).astype(float) + signals.execute_exit.astype(float)) mask = (signals.time >= pd.Timestamp(start)) & (signals.time < pd.Timestamp(end) + pd.Timedelta(days=1)) frame, values = signals.loc[mask], returns.loc[mask].to_numpy() if not len(frame): raise ValueError(f'No bars in requested window {start} through {end}') equity = np.r_[1., np.cumprod(1 + values)] dd = float(np.min(equity / np.maximum.accumulate(equity) - 1)) interval = signals.time.diff().median() years = (frame.time.iloc[-1] - frame.time.iloc[0] + interval).total_seconds() / (365.25*86400) cagr = equity[-1] ** (1/years) - 1 if equity[-1] > 0 else -1. return {'pnl_pct': (equity[-1]-1)*100, 'dd_pct': dd*100, 'calmar': cagr/abs(dd) if dd < 0 else None, 'entries': int(frame.execute_entry.sum()), 'exits': int(frame.execute_exit.sum()), 'exposure_pct': float(frame.position.mean()*100), 'blocked_entries': int(frame.get('policy_blocked_entry', pd.Series(dtype=bool)).sum()), 'deferred_exit_bars': int(frame.get('policy_deferred_exit', pd.Series(dtype=bool)).sum())} def digest(path): return hashlib.sha256(Path(path).read_bytes()).hexdigest() def run(tfs, output): rows, provenance = [], [] asset = 'COINBASE_BTCUSD' for tf in tfs: path = _data_path(asset, tf) errors, warnings, _ = validate_file(str(path)) if errors: raise ValueError(f'{path}: {errors}') frame = pd.read_csv(path) frame.columns = frame.columns.str.lower().str.strip() unit = 's' if pd.api.types.is_numeric_dtype(frame.time) else None frame.time = pd.to_datetime(frame.time, unit=unit, utc=True).dt.tz_localize(None) dropped_incomplete_tail = 0 if 'mvrv_regime' in frame and pd.isna(frame.mvrv_regime.iloc[-1]): frame = frame.iloc[:-1].copy() dropped_incomplete_tail = 1 # Require the actual input instead of silently making an absent macro # signal zero. MVRV regime already exists in the Pine chart exports. if 'mvrv_regime' not in frame or not frame.mvrv_regime.isin([-1, 0, 1]).all(): raise ValueError('Expected finite mvrv_regime values in {-1, 0, 1}') params = _load_winner(asset, tf) params['_pine_time_start'] = SCORE_START base = generate_signals(frame, **params) provenance.append({'tf': tf, 'data': str(path.relative_to(REPO)), 'data_sha256': digest(path), 'winner_sha256': digest(_winner_path(asset, tf)), 'weights_sha256': digest(params['mlp_weights_file']), 'first': str(frame.time.iloc[0]), 'last': str(frame.time.iloc[-1]), 'warnings': warnings, 'dropped_incomplete_tail': dropped_incomplete_tail, 'params': params}) windows = [('IS', SCORE_START, TRAIN_END), *[(f'historical_{start[:7]}', start, end) for _, start, end in WFO_FOLDS], ('OOS', OOS_START, str(frame.time.iloc[-1].date()))] for name, (policy, macro) in VARIANTS.items(): signals = _score_to_signals(base.copy(), params, base.stoch_peak_norm.to_numpy(), base.close.to_numpy(), base.high.to_numpy(), state_policy=policy, policy_risk=(frame.mvrv_regime.to_numpy() < 0) if macro else None) if name == 'baseline': for column in ('position', 'execute_entry', 'execute_exit'): np.testing.assert_array_equal(signals[column], base[column]) for label, start, end in windows: metrics = window_metrics(signals, start, end) rows.append({'tf': tf, 'variant': name, 'window': label, **metrics, 'trade_coverage': metrics['entries'] >= WFO_MIN_OOS_TRADES}) results = pd.DataFrame(rows) output.mkdir(parents=True, exist_ok=True) results.to_csv(output / 'metrics.csv', index=False) (output / 'manifest.json').write_text(json.dumps({ 'non_promoting': True, 'variants': {name: {'policy': asdict(p), 'macro': m} for name, (p, m) in VARIANTS.items()}, 'inputs': provenance, 'code_sha256': {name: digest(REPO / name) for name in ['strategies/mlp_state_policy.py', 'strategies/strategy_mlp_scores.py', 'strategies/strategy_activation_scores.py', 'strategies/library_activation_scores.py', 'strategies/mlp_temporal_features.py', 'tools/validate_chart_data.py', 'tools/run_mlp_score_structure_experiment.py', 'tools/run_mlp_state_policy_experiment.py', 'config.py']}, }, indent=2)) print(results[results.window.isin(['IS', 'OOS'])].to_string(index=False, float_format=lambda x: f'{x:.3f}')) return results if __name__ == '__main__': parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--tfs', nargs='+', choices=['4H', '6H', '8H', '12H', '1D'], default=['4H', '6H', '8H', '12H', '1D']) parser.add_argument('--output', type=Path, default=REPO / 'results/mlp_experiments/state_policy') args = parser.parse_args() run(args.tfs, args.output)