"""Research-only 4H SOPR entry-inhibition ablation. This never changes MLP weights, promoted parameters, Pine code, or exits. It compares the promoted Python replay with the same replay in which an entry is permitted only when SOPR is above a discovery-calibrated low-tail cutoff. """ from __future__ import annotations import argparse import json import sys from pathlib import Path import numpy as np import pandas as pd REPO = Path(__file__).resolve().parent.parent sys.path.insert(0, str(REPO)) from config import OOS_START, SCORE_START, WFO_FOLDS from strategies.strategy_mlp_scores import calculate_metrics, generate_signals from tools.check_mlp_parity import load_params from tools.run_mlp_extreme_bar_event_study import DISCOVERY_END, VALIDATION_START, parse_chart_time def _metrics(signals: pd.DataFrame, start: str, end: str | None = None) -> dict[str, float | int]: window = signals if end is None else signals[signals["time"] <= pd.Timestamp(end)].copy() result = calculate_metrics(window, score_start=start, min_trades=0) return { key: float(value) if isinstance(value, (float, np.floating)) else int(value) for key, value in result.items() } def _entry_diagnostics(signals: pd.DataFrame, blocked: pd.Series, start: str, end: str | None) -> dict[str, float | int | None]: mask = signals["time"] >= pd.Timestamp(start) if end is not None: mask &= signals["time"] <= pd.Timestamp(end) candidates = signals["execute_entry"] & mask selected = candidates & blocked next_return = signals["close"].shift(-1).div(signals["close"]).sub(1.0) def describe(rows: pd.Series) -> dict[str, float | int | None]: values = next_return[rows].dropna() return { "entries": int(rows.sum()), "next_bar_observations": int(len(values)), "next_bar_mean_return_pct": float(values.mean() * 100) if len(values) else None, "next_bar_positive_rate": float((values > 0).mean()) if len(values) else None, } return {"baseline_entries": describe(candidates), "blocked_baseline_entries": describe(selected)} def calibrate_sopr_cutoff(data: pd.DataFrame, history_end: str | pd.Timestamp, feature_tail: float) -> float: """Return a low-SOPR cutoff using only rows available through history_end.""" history = data[(data["time"] >= pd.Timestamp(SCORE_START)) & (data["time"] <= pd.Timestamp(history_end))] if history.empty: raise ValueError("no SOPR history available for cutoff calibration") return float(history["sopr_norm"].quantile(feature_tail)) def run(data_path: Path, winner_path: Path, *, feature_tail: float = 0.10) -> dict[str, object]: if not 0 < feature_tail < 0.5: raise ValueError("feature_tail must be between 0 and 0.5") data = pd.read_csv(data_path) data.columns = data.columns.str.lower().str.strip() if "sopr_norm" not in data.columns: raise ValueError("data has no sopr_norm column") data["time"] = parse_chart_time(data["time"]) data = data.sort_values("time").reset_index(drop=True) params = load_params(str(winner_path)) if not params.get("mlp_weights_file"): raise ValueError(f"winner has no mlp_weights_file: {winner_path}") cutoff = calibrate_sopr_cutoff(data, DISCOVERY_END, feature_tail) # This is evaluated at the decision bar. It applies only to entry raw # signals; exits and every other promoted parameter remain unchanged. entry_allowed = (data["sopr_norm"] > cutoff).fillna(False).to_numpy(dtype=bool) baseline = generate_signals(data.copy(), **params) gated = generate_signals(data.copy(), entry_allowed=entry_allowed, **params) all_allowed = generate_signals(data.copy(), entry_allowed=np.ones(len(data), dtype=bool), **params) for column in ("position", "execute_entry", "execute_exit"): if not baseline[column].equals(all_allowed[column]): raise AssertionError(f"all-true entry mask changed baseline {column}") blocked = ~pd.Series(entry_allowed, index=data.index) windows: dict[str, tuple[str, str | None]] = { "discovery": (SCORE_START, str(DISCOVERY_END)), "validation": (str(VALIDATION_START), str(pd.Timestamp(OOS_START) - pd.Timedelta(nanoseconds=1))), "oos": (OOS_START, None), } summaries = { name: { "baseline": _metrics(baseline, start, end), "sopr_entry_inhibited": _metrics(gated, start, end), "entry_diagnostics": _entry_diagnostics(baseline, blocked, start, end), } for name, (start, end) in windows.items() } folds: dict[str, dict[str, object]] = {} for is_end, start, end in WFO_FOLDS: # Unlike the fixed validation/OOS policy, each retrospective WFO fold # may calibrate only on history available before its OOS window. fold_cutoff = calibrate_sopr_cutoff(data, is_end, feature_tail) fold_blocked = data["sopr_norm"].le(fold_cutoff).fillna(True) fold_gated = generate_signals( data.copy(), entry_allowed=(~fold_blocked).to_numpy(dtype=bool), **params, ) folds[end] = { "cutoff": fold_cutoff, "baseline": _metrics(baseline, start, end), "sopr_entry_inhibited": _metrics(fold_gated, start, end), "entry_diagnostics": _entry_diagnostics(baseline, fold_blocked, start, end), } return { "data_path": str(data_path), "winner_path": str(winner_path), "definition": { "policy": "when flat and the unchanged MLP entry condition fires, inhibit entry if sopr_norm <= discovery cutoff", "feature_tail": feature_tail, "sopr_cutoff": cutoff, "unchanged": ["MLP weights", "all promoted parameters", "exit logic", "Pine code"], "costs": "strategy metrics include the repository's 0.5% entry and 0.5% exit commission assumptions", "wfo_calibration": "each fold recalibrates its fixed 10th-percentile cutoff using only history before that fold's OOS start", "warning": "this is a one-policy ablation, not a threshold sweep or a promotable candidate", }, "windows": summaries, "wfo_folds": folds, } def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--data", type=Path, required=True) parser.add_argument("--winner", type=Path, required=True) parser.add_argument("--output", type=Path) args = parser.parse_args() result = run(args.data, args.winner) text = json.dumps(result, indent=2) print(text) if args.output: args.output.parent.mkdir(parents=True, exist_ok=True) args.output.write_text(text + "\n") return 0 if __name__ == "__main__": raise SystemExit(main())