"""Fit purged position-conditioned value heads; never promote or change winners.""" from __future__ import annotations import argparse import json from pathlib import Path 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, OOS_START, WFO_FOLDS from strategies.mlp_learned_exit import (HORIZON, MAX_EXTENSION, MIN_TRAIN_TRADES, fit_head, training_records, replay, continuation_target) 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.run_mlp_state_policy_experiment import window_metrics, digest from tools.validate_chart_data import validate_file VARIANTS = {'baseline': (None, 'baseline'), 'delay_12': (None, 'delay'), 'market_veto': ('market', 'veto'), 'state_veto': ('state', 'veto'), 'combined_veto': ('combined', 'veto'), 'combined_manage': ('combined', 'manage'), 'tree_market_veto': ('tree_market', 'veto'), 'tree_combined_veto': ('tree_combined', 'veto'), 'tree_combined_manage': ('tree_combined', 'manage')} def load_input(tf): path = _data_path('COINBASE_BTCUSD', tf) errors, warnings, _ = validate_file(str(path)) if errors: raise ValueError(f'{path}: {errors}') data = pd.read_csv(path) data.columns = data.columns.str.lower().str.strip() unit = 's' if pd.api.types.is_numeric_dtype(data.time) else None data.time = pd.to_datetime(data.time, unit=unit, utc=True).dt.tz_localize(None) dropped_tail = int(pd.isna(data.mvrv_regime.iloc[-1])) if dropped_tail: data = data.iloc[:-1].copy() required = ['rvol_norm', 'fed_net_liq_sign', 'mvrv_regime'] complete = np.isfinite(data[required].to_numpy(dtype=float)).all(axis=1) if not complete.any(): raise ValueError('No complete market-input rows') leading = int(np.flatnonzero(complete)[0]) data = data.iloc[leading:].reset_index(drop=True) if not np.isfinite(data[required].to_numpy(dtype=float)).all(): raise ValueError('Interior missing market inputs') if data.time.iloc[0] > pd.Timestamp(SCORE_START): raise ValueError('Warmup exclusion removes required score history') params = _load_winner('COINBASE_BTCUSD', tf) params['_pine_time_start'] = SCORE_START base = generate_signals(data, **params) opportunities = _score_to_signals(base.copy(), params, base.stoch_peak_norm.to_numpy(), base.close.to_numpy(), base.high.to_numpy(), return_opportunities=True) metadata = {'tf': tf, 'data': str(path.relative_to(REPO)), 'data_sha256': digest(path), 'winner_sha256': digest(_winner_path('COINBASE_BTCUSD', tf)), 'weights_sha256': digest(params['mlp_weights_file']), 'params': params, 'first': str(data.time.iloc[0]), 'last': str(data.time.iloc[-1]), 'excluded_leading_warmup_rows': leading, 'excluded_tail_rows': dropped_tail, 'warnings': warnings} return base, opportunities, metadata def run(tfs, output): # Mandatory project prerequisites are enforced on every invocation. for tool in ('sync_params.py', 'check_pine.py'): subprocess.run([sys.executable, str(REPO/'tools'/tool)], cwd=REPO, check=True) output.mkdir(parents=True, exist_ok=True) rows, audits, inputs, predictions, events = [], [], [], [], [] for tf in tfs: base, opportunities, provenance = load_input(tf) inputs.append(provenance) records = training_records(base, opportunities) folds = [(start, end) for _, start, end in WFO_FOLDS] folds.append((OOS_START, str(base.time.iloc[-1].date()))) for start, end in folds: label = 'later_2026' if start == OOS_START else f'historical_{start[:7]}' heads = {} for kind in ('market', 'state', 'combined', 'tree_market', 'tree_combined'): model_type = 'tree' if kind.startswith('tree_') else 'ridge' fitted, audit = fit_head(records, base.time, start, kind.removeprefix('tree_'), model_type) heads[kind] = fitted audits.append({'tf': tf, 'fold': label, 'kind': kind, **audit, 'model': None if fitted is None else fitted.artifact()}) for variant, (kind, mode) in VARIANTS.items(): signals, decisions = replay(base, opportunities, start=start, end=end, head=heads.get(kind), mode=mode) if variant == '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(base.copy(), params, base.stoch_peak_norm.to_numpy(), base.close.to_numpy(), base.high.to_numpy()) for col in ('position', 'execute_entry', 'execute_exit'): np.testing.assert_array_equal(signals[col], canonical[col]) metrics = window_metrics(signals, start, end) rows.append({'tf': tf, 'fold': label, 'start': start, 'end': end, 'variant': variant, 'fitted': kind is not None and heads[kind] is not None, **metrics, 'learned_exits': sum(d['reason']=='learned_exit' for d in decisions), 'cap_exits': sum(d['reason']=='extension_cap' for d in decisions), 'trailing_exits': sum(d['reason']=='trailing_stop' for d in decisions), 'hold_decisions': sum(d['predicted_advantage'] is not None and d['action']=='hold' for d in decisions)}) for decision in decisions: # Store every active head prediction and executed exit, plus # score exits; idle holding bars have no policy action to audit. if decision['predicted_advantage'] is not None or decision['action']=='exit' or decision['score_exit']: predictions.append({'tf': tf, 'fold': label, 'variant': variant, **decision}) if variant == 'baseline' and decision['reason']=='score_exit': target = continuation_target(base.close.to_numpy(), base.high.to_numpy(), decision['bar'], decision['entry_bar'], decision['trade_high'], opportunities['trail']) if target is not None and base.time.iloc[target[1]] < pd.Timestamp(end)+pd.Timedelta(days=1): events.append({'tf': tf, 'fold': label, 'bar': decision['bar'], 'time': decision['time'], 'hold_4_advantage': target[0], 'label_end': str(base.time.iloc[target[1]])}) print(f'{tf} {label}: fit from {audits[-1]["trades"]} trades / {audits[-1]["samples"]} rows', flush=True) pd.DataFrame(rows).to_csv(output/'metrics.csv', index=False) pd.DataFrame(predictions).to_csv(output/'decisions.csv', index=False) pd.DataFrame(events).to_csv(output/'exit_opportunities.csv', index=False) (output/'models.json').write_text(json.dumps(audits, indent=2, allow_nan=False)) sources = ['strategies/mlp_learned_exit.py', 'strategies/strategy_mlp_scores.py', 'strategies/strategy_activation_scores.py', 'strategies/library_activation_scores.py', 'strategies/mlp_temporal_features.py', 'tools/run_mlp_learned_exit_experiment.py', 'tools/run_mlp_state_policy_experiment.py', 'tools/run_mlp_score_structure_experiment.py', 'tools/validate_chart_data.py', 'tools/check_pine.py', 'config.py', 'docs/mlp_learned_exit_protocol_2026_09_21.md'] manifest = {'non_promoting': True, 'horizon': HORIZON, 'max_extension': MAX_EXTENSION, 'min_train_trades': MIN_TRAIN_TRADES, 'variants': VARIANTS, 'inputs': inputs, 'code_sha256': {name: digest(REPO/name) for name in sources}, 'outputs_sha256': {name: digest(output/name) for name in ['metrics.csv', 'decisions.csv', 'exit_opportunities.csv', 'models.json']}} (output/'manifest.json').write_text(json.dumps(manifest, indent=2, allow_nan=False)) return pd.DataFrame(rows) 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/learned_exit') args = parser.parse_args() run(args.tfs, args.output)