"""Non-promoting screen for a frozen MLP trunk plus tiny slow-regime heads. This is deliberately not part of ``run_mlp_train.py``: artifacts are written under ``results/mlp_experiments`` and the runner has no sweep/promote path. It uses the winner's thresholds unchanged, so any observed change is due to the 27 residual head parameters, not threshold search. """ from __future__ import annotations import argparse import json from pathlib import Path import numpy as np import pandas as pd REPO = Path(__file__).resolve().parent.parent import sys sys.path.insert(0, str(REPO)) from config import SCORE_START, TRAIN_END, WFO_FOLDS from strategies.mlp_slow_regime_router import (SlowRouterConfig, regime_head_scores, shared_trunk_forward, slow_regime_states) from strategies.strategy_activation_scores import _prepare_features from strategies.strategy_mlp_scores import _score_to_signals, load_mlp_artifact from tools.run_mlp_score_structure_experiment import _data_path, _load_winner, _summary def _parse_chart_time(values: pd.Series) -> pd.Series: """Accept either TradingView's ISO timestamps or its Unix-second export.""" if pd.api.types.is_numeric_dtype(values): return pd.to_datetime(values, unit="s", utc=True).dt.tz_localize(None) return pd.to_datetime(values, utc=True).dt.tz_localize(None) def _target(frame: pd.DataFrame, k: int = 10, vol_window: int = 60) -> tuple[np.ndarray, np.ndarray]: """Same causal target as train_mlp, kept local to isolate this experiment.""" log_close = np.log(frame["close"].to_numpy(dtype=np.float64)) ret1 = pd.Series(np.diff(log_close, prepend=log_close[0])) vol = ret1.rolling(vol_window).std().shift(1).to_numpy() forward = np.full(len(frame), np.nan) forward[:-k] = (log_close[k:] - log_close[:-k]) / k y = np.tanh(2.0 * forward / (vol + 1e-6)) valid = ~np.isnan(y) & (frame["time"] >= pd.Timestamp(SCORE_START)).to_numpy() return y, valid def fit_residual_heads(trunk: np.ndarray, states: np.ndarray, base_head: tuple[np.ndarray, np.ndarray], y: np.ndarray, valid: np.ndarray, times: pd.Series, *, seed: int, val_start: pd.Timestamp, epochs: int = 300, l2: float = 1e-3) -> tuple[tuple[np.ndarray, np.ndarray], dict[str, float | int]]: """Fit only 3×(trunk width + bias) residuals; the dense trunk stays frozen.""" import torch torch.manual_seed(seed) np.random.seed(seed) state_index = states.astype(np.int64) + 1 train_mask = valid & (times < val_start).to_numpy() val_mask = valid & (times >= val_start).to_numpy() & (times <= pd.Timestamp(TRAIN_END)).to_numpy() if not train_mask.any() or not val_mask.any(): raise ValueError("Experiment requires non-empty chronological train and validation partitions") base_weights, base_bias = base_head with np.errstate(over="ignore", divide="ignore", invalid="ignore"): base_logits = (trunk @ base_weights.T + base_bias).reshape(-1) features = torch.tensor(trunk, dtype=torch.float32) indices = torch.tensor(state_index, dtype=torch.long) targets = torch.tensor(y, dtype=torch.float32) train_idx = torch.tensor(np.flatnonzero(train_mask), dtype=torch.long) val_idx = torch.tensor(np.flatnonzero(val_mask), dtype=torch.long) # Tiny seeded noise makes each seed an independent optimization trajectory; # exact-zero initialization would make Adam deterministic across all runs. heads = torch.nn.Parameter(torch.empty((3, trunk.shape[1]), dtype=torch.float32).normal_(mean=0.0, std=1e-3)) bias = torch.nn.Parameter(torch.empty(3, dtype=torch.float32).normal_(mean=0.0, std=1e-3)) fixed = torch.tensor(base_logits, dtype=torch.float32) optimizer = torch.optim.Adam((heads, bias), lr=1e-2) best_loss, best_state, since_best = float("inf"), None, 0 for epoch in range(epochs): optimizer.zero_grad() logits = fixed[train_idx] + (features[train_idx] * heads[indices[train_idx]]).sum(1) + bias[indices[train_idx]] loss = torch.mean((torch.tanh(logits) - targets[train_idx]) ** 2) + l2 * (heads.square().mean() + bias.square().mean()) loss.backward() optimizer.step() with torch.no_grad(): val_logits = fixed[val_idx] + (features[val_idx] * heads[indices[val_idx]]).sum(1) + bias[indices[val_idx]] val_loss = float(torch.mean((torch.tanh(val_logits) - targets[val_idx]) ** 2)) if val_loss < best_loss - 1e-7: best_loss, since_best = val_loss, 0 best_state = (heads.detach().cpu().numpy().copy(), bias.detach().cpu().numpy().copy()) else: since_best += 1 if since_best >= 25: break assert best_state is not None return best_state, {"epochs_run": epoch + 1, "best_val_mse": best_loss, "train_bars": int(train_mask.sum()), "val_bars": int(val_mask.sum())} def _signals(frame: pd.DataFrame, params: dict[str, object], score: np.ndarray) -> pd.DataFrame: scored = frame.copy() scored["activation_score"] = score return _score_to_signals(scored, params, scored["stoch_peak_norm"].to_numpy(), scored["close"].to_numpy(np.float64), scored["high"].to_numpy(np.float64)) def run(asset: str, tf: str, seeds: list[int], *, epochs: int = 300, router_config: SlowRouterConfig = SlowRouterConfig()) -> tuple[pd.DataFrame, dict[str, object]]: frame = pd.read_csv(_data_path(asset, tf)) frame.columns = frame.columns.str.lower().str.strip() frame["time"] = _parse_chart_time(frame["time"]) params = _load_winner(asset, tf) params["_pine_time_start"] = SCORE_START artifact = load_mlp_artifact(str(params["mlp_weights_file"])) if artifact["arch"][-3:] != [16, 8, 1]: raise ValueError(f"Expected dense 55→16→8→1 winner, got {artifact['arch']}") X = _prepare_features(frame, artifact["feature_cols"]) trunk = shared_trunk_forward(X, artifact["layers"]) states = slow_regime_states(frame, router_config) y, valid = _target(frame) val_start = pd.Timestamp(WFO_FOLDS[-1][1]) baseline = _signals(frame, params, regime_head_scores(trunk, states, artifact["layers"][-1], (np.zeros((3, trunk.shape[1])), np.zeros(3)))) rows = [_summary("null heads (exact shared-trunk control)", baseline)] runs = [] for seed in seeds: heads, training = fit_residual_heads(trunk, states, artifact["layers"][-1], y, valid, frame["time"], seed=seed, val_start=val_start, epochs=epochs) signals = _signals(frame, params, regime_head_scores(trunk, states, artifact["layers"][-1], heads)) row = _summary(f"slow-regime residual heads seed {seed}", signals) rows.append(row) runs.append({"seed": seed, "training": training, "residual_heads": [heads[0].tolist(), heads[1].tolist()]}) metadata = {"non_promoting": True, "asset": asset, "tf": tf, "router": vars(router_config), "state_counts": {str(state): int((states == state).sum()) for state in (-1, 0, 1)}, "state_switches": int(np.count_nonzero(np.diff(states))), "validation_start": str(val_start.date()), "fixed_policy": "winner thresholds unchanged", "runs": runs} return pd.DataFrame(rows), metadata def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--asset", default="COINBASE_BTCUSD") parser.add_argument("--tf", default="6H") parser.add_argument("--seeds", nargs="+", type=int, default=[8080, 8181, 8282]) parser.add_argument("--epochs", type=int, default=300) parser.add_argument("--output", type=Path, default=REPO / "docs" / "mlp_slow_regime_head_experiment_btc_6h_2026_08_21.md") args = parser.parse_args() results, metadata = run(args.asset, args.tf, args.seeds, epochs=args.epochs) print(results.to_string(index=False, float_format=lambda value: f"{value:.3f}")) control = results.iloc[0] candidates = results.iloc[1:] eligible = int(candidates["wfo_eligible"].sum()) median_wfo = float(candidates["wfo_mean_calmar"].median()) decision = ( "Rejected at the exploratory screen: no candidate met the canonical WFO coverage " f"requirement ({eligible}/{len(candidates)} eligible), and median WFO Calmar " f"({median_wfo:.3f}) was below the paired null-head control ({float(control['wfo_mean_calmar']):.3f}). " "This is independent of the stricter live-promotion benchmark." ) args.output.parent.mkdir(parents=True, exist_ok=True) args.output.write_text( "# Slow Regime-Head Experiment — BTC 6H\n\n" "Non-promoting, fixed-policy screen. The current winner's thresholds were held fixed; " "the only trainable parameters are three residual 8→1 output heads.\n\n" "## Decision\n\n" + decision + "\n\n" "## Results\n\n```text\n" + results.to_string(index=False, float_format=lambda value: f"{value:.3f}") + "\n```\n\n## Provenance\n\n```json\n" + json.dumps(metadata, indent=2) + "\n```\n" ) return 0 if __name__ == "__main__": raise SystemExit(main())