"""Fixed, non-promoting experiments for MLP-score swing structure on one timeframe. This runner deliberately does not optimize thresholds, MLP weights, or pivot windows. It compares the promoted winner with a fixed confirmation ablation and with three causal score-structure overlays. A pivot is acted upon only on its right-hand confirmation bar, preventing look-ahead bias. """ from __future__ import annotations import argparse import csv 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 (OOS_START, SCORE_START, TRAIN_END, WFO_FOLDS, WFO_MIN_OOS_TRADES, WFO_MIN_VALID_FOLDS) from strategies.mlp_score_structure import ScoreStructure, confirmed_score_structure from strategies.strategy_mlp_scores import _score_to_signals, calculate_metrics, generate_signals def _winner_path(asset: str, tf: str) -> Path: return REPO / "results" / "winners" / f"optimization_winner_strategy_mlp_scores_{asset}_{tf}.csv" def _data_path(asset: str, tf: str) -> Path: periods = {"4H": "240", "6H": "360", "8H": "480", "12H": "720", "1D": "1D"} return REPO / "data" / "mlp" / f"{asset}, {periods[tf]}.csv" def _load_winner(asset: str, tf: str) -> dict[str, object]: with _winner_path(asset, tf).open() as handle: row = next(csv.DictReader(handle)) for key, value in list(row.items()): if key.startswith("i_") and key not in {"i_use_long_exit_confirmation", "i_use_long_entry_confirmation", "i_mvrv_suppress_bear"}: row[key] = float(value) row["i_use_long_exit_confirmation"] = row["i_use_long_exit_confirmation"] in {"1", "True", "true"} row["i_use_long_entry_confirmation"] = row["i_use_long_entry_confirmation"] in {"1", "True", "true"} row["i_mvrv_suppress_bear"] = row["i_mvrv_suppress_bear"] in {"1", "True", "true"} weights_path = Path(str(row["mlp_weights_file"])) if not weights_path.exists() and "strategies/" in str(weights_path): weights_path = REPO / str(weights_path).split("strategies/", 1)[1] weights_path = REPO / "strategies" / weights_path row["mlp_weights_file"] = str(weights_path) return row def _metrics(signals: pd.DataFrame, start: str, end: str | None = None) -> dict[str, float]: frame = signals if end is None else signals[signals["time"] <= pd.Timestamp(end)] return calculate_metrics(frame.copy(), score_start=start, min_trades=1) def _wfo(signals: pd.DataFrame) -> tuple[float, float, int, bool]: fold_calmars: list[float] = [] for _, start, end in WFO_FOLDS: fold = _metrics(signals, start, end) if fold["Total Trades"] >= WFO_MIN_OOS_TRADES: fold_calmars.append(float(fold["Calmar Ratio"])) eligible = len(fold_calmars) >= WFO_MIN_VALID_FOLDS if not eligible: return 0.0, -99.0, len(fold_calmars), False return float(np.mean(fold_calmars)), float(np.min(fold_calmars)), len(fold_calmars), True def _summary(name: str, signals: pd.DataFrame) -> dict[str, object]: is_metrics = _metrics(signals[signals["time"] <= pd.Timestamp(TRAIN_END)], SCORE_START) oos_metrics = _metrics(signals, OOS_START) # The available OOS slice has no completed trades for this winner. The # canonical metric function uses -10 as a low-trade sentinel; report zero # instead so readers do not mistake the absence of OOS trades for a loss. if oos_metrics["Total Trades"] == 0: oos_metrics = dict(oos_metrics) for key in ("Calmar Ratio", "Total P&L %"): oos_metrics[key] = 0.0 wfo_mean, wfo_min, wfo_folds, wfo_eligible = _wfo(signals) return { "experiment": name, "is_calmar": is_metrics["Calmar Ratio"], "is_pnl_pct": is_metrics["Total P&L %"], "is_max_dd_pct": is_metrics["Max Drawdown %"], "is_trades": is_metrics["Total Trades"], "wfo_mean_calmar": wfo_mean, "wfo_min_calmar": wfo_min, "wfo_valid_folds": wfo_folds, "wfo_eligible": wfo_eligible, "oos_calmar": oos_metrics["Calmar Ratio"], "oos_pnl_pct": oos_metrics["Total P&L %"], "oos_trades": oos_metrics["Total Trades"], } def _structure_overlay(base: pd.DataFrame, params: dict[str, object], structure: ScoreStructure, *, gate_entries: bool, exit_lower_high: bool) -> pd.DataFrame: return _score_to_signals( base.copy(), params, base["stoch_peak_norm"].to_numpy(), base["close"].to_numpy(np.float64), base["high"].to_numpy(np.float64), entry_gate=structure.bullish_state if gate_entries else None, additional_exit=structure.lower_high if exit_lower_high else None, ) def run(asset: str, tf: str, pivots: list[int]) -> tuple[pd.DataFrame, pd.DataFrame]: data = pd.read_csv(_data_path(asset, tf)) data.columns = data.columns.str.lower().str.strip() data["time"] = pd.to_datetime(data["time"], utc=True).dt.tz_localize(None) params = _load_winner(asset, tf) params["_pine_time_start"] = SCORE_START base = generate_signals(data, **params) results = [_summary("baseline (promoted winner)", base)] no_exit_confirmation = dict(params) no_exit_confirmation["i_use_long_exit_confirmation"] = False results.append(_summary("ablation: exit confirmation off", generate_signals(data, **no_exit_confirmation))) events: list[dict[str, object]] = [] for pivot in pivots: structure = confirmed_score_structure(base["activation_score"].to_numpy(), left=pivot, right=pivot) events.append({"pivot_left_right": pivot, "higher_lows": int(structure.higher_low.sum()), "lower_highs": int(structure.lower_high.sum())}) for label, gate_entries, exit_lower_high in ( ("entry gate: higher-low state", True, False), ("exit: confirmed lower high", False, True), ("combined higher-low gate + lower-high exit", True, True), ): results.append(_summary( f"pivot {pivot}/{pivot} — {label}", _structure_overlay(base, params, structure, gate_entries=gate_entries, exit_lower_high=exit_lower_high), )) return pd.DataFrame(results), pd.DataFrame(events) def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--asset", default="COINBASE_BTCUSD") parser.add_argument("--tf", default="6H") parser.add_argument("--pivots", type=int, nargs="+", default=[2, 3, 4]) parser.add_argument("--output", type=Path) args = parser.parse_args() results, events = run(args.asset, args.tf, args.pivots) print(results.to_string(index=False, float_format=lambda value: f"{value:.3f}")) print("\nConfirmed score-structure events:") print(events.to_string(index=False)) if args.output: args.output.parent.mkdir(parents=True, exist_ok=True) with args.output.open("w") as handle: handle.write("# MLP score-structure experiment\n\n") handle.write("This is a fixed historical screen, not an optimization or promotion. " "Pivots are acted on only after right-hand confirmation.\n\n") handle.write("## Results\n\n```text\n") handle.write(results.to_string(index=False, float_format=lambda value: f"{value:.3f}")) handle.write("\n```\n\n## Confirmed events\n\n```text\n") handle.write(events.to_string(index=False)) handle.write("\n```\n") return 0 if __name__ == "__main__": raise SystemExit(main())