"""Causal, non-promoting MLP score-peak deterioration event study. The only model inputs are the local TradingView export and the current winner CSV for the requested asset/timeframe. This script never writes either. """ from __future__ import annotations import argparse import csv 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, TRAIN_END, WFO_FOLDS from strategies.strategy_mlp_scores import generate_signals HORIZONS = (1, 3, 7, 14) PIVOT_WIDTHS = (2, 3, 4) 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: minutes = {"4H": "240", "6H": "360", "8H": "480", "12H": "720", "1D": "1D"} return REPO / "data" / "mlp" / f"{asset}, {minutes[tf.upper()]}.csv" def _load_winner(asset: str, tf: str) -> dict[str, object]: with _winner_path(asset, tf).open() as handle: row = next(csv.DictReader(handle)) booleans = {"i_use_long_exit_confirmation", "i_use_long_entry_confirmation", "i_mvrv_suppress_bear"} for key, value in list(row.items()): if key.startswith("i_"): row[key] = value in {"1", "True", "true"} if key in booleans else float(value) weight_path = Path(str(row["mlp_weights_file"])) if not weight_path.exists(): weight_path = REPO / "strategies" / str(weight_path).split("strategies/", 1)[-1] row["mlp_weights_file"] = str(weight_path) return row def confirmed_pivot_highs(scores: np.ndarray, left: int, right: int) -> list[tuple[int, int, float]]: """Return (pivot index, confirmation index, score), with no future access. At ``confirmation`` the complete [pivot-left, pivot+right] window is closed. Strict comparison on the left and non-strict on the right makes a flat top deterministic: its first bar is not selected before a later bar could invalidate it. """ result = [] for pivot in range(left, len(scores) - right): value = scores[pivot] if not np.isfinite(value): continue before, after = scores[pivot - left:pivot], scores[pivot + 1:pivot + right + 1] if np.all(value > before) and np.all(value >= after): result.append((pivot, pivot + right, float(value))) return result def detect_deterioration_events(pivots: list[tuple[int, int, float]], *, horizon: int = 14) -> list[dict[str, float | int]]: """Find qualifying second peaks, deduping chronologically by event horizon.""" candidates: list[dict[str, float | int]] = [] for second_pivot, second_confirm, second_score in pivots: for first_pivot, first_confirm, first_score in pivots: spacing = second_confirm - first_confirm if not (8 <= spacing <= 40): continue if first_score < 300 or second_score > first_score - 200 or second_score > .6 * first_score: continue candidates.append({"first_pivot_index": first_pivot, "first_confirmation_index": first_confirm, "first_score": first_score, "second_pivot_index": second_pivot, "event_index": second_confirm, "event_score": second_score, "second_score": second_score}) # More than one qualifying earlier peak can point to a second peak. Keep # the most recent first peak, then earliest event, before the horizon lock. candidates.sort(key=lambda event: (int(event["event_index"]), -int(event["first_confirmation_index"]))) accepted: list[dict[str, float | int]] = [] next_allowed = -1 seen_events: set[int] = set() for event in candidates: index = int(event["event_index"]) if index in seen_events or index < next_allowed: continue accepted.append(event) seen_events.add(index) next_allowed = index + horizon + 1 return accepted def forward_metrics(closes: np.ndarray, event_index: int, horizons: tuple[int, ...] = HORIZONS) -> dict[int, dict[str, float]]: entry = float(closes[event_index]) output: dict[int, dict[str, float]] = {} for horizon in horizons: path = closes[event_index + 1:event_index + horizon + 1] if len(path) != horizon or not np.isfinite(entry) or entry == 0 or not np.all(np.isfinite(path)): continue output[horizon] = {"return_pct": 100 * (float(path[-1]) / entry - 1), # MAE is adverse only: a path entirely above entry # has zero excursion rather than a positive "MAE". "mae_pct": min(0.0, 100 * (float(np.min(path)) / entry - 1))} return output def match_controls(frame: pd.DataFrame, events: list[dict[str, float | int]], *, horizon: int = 14, score_band: float = 150.0, region_days: int = 180, allowed_indices: set[int] | None = None, forbidden_events: list[dict[str, float | int]] | None = None) -> list[int]: """One deterministic, without-replacement control per event. Candidates share the event's +/- ``region_days`` calendar region and have score within ``score_band`` points. All event action bars and forward horizon bars are excluded. Ties choose smallest score distance, then nearest timestamp, then earliest index. """ times = pd.to_datetime(frame["time"]) scores = frame["activation_score"].to_numpy(float) forbidden_events = events if forbidden_events is None else forbidden_events forbidden = {idx for event in forbidden_events for idx in range(int(event["event_index"]), int(event["event_index"]) + horizon + 1)} used: set[int] = set() selected: list[int] = [] for event in sorted(events, key=lambda value: int(value["event_index"])): index, score = int(event["event_index"]), float(event["event_score"]) candidates = [] for candidate in range(len(frame) - horizon): if allowed_indices is not None and candidate not in allowed_indices: continue if candidate in forbidden or candidate in used or abs(scores[candidate] - score) > score_band: continue if abs(times.iloc[candidate] - times.iloc[index]) > pd.Timedelta(days=region_days): continue candidates.append(candidate) if candidates: best = min(candidates, key=lambda candidate: (abs(scores[candidate] - score), abs(candidate - index), candidate)) selected.append(best) used.add(best) return selected def _summarize(indices: list[int], closes: np.ndarray) -> dict[str, object]: output: dict[str, object] = {"count": len(indices), "horizons": {}} metrics_by_index = [forward_metrics(closes, index) for index in indices] for horizon in HORIZONS: values = [metrics.get(horizon) for metrics in metrics_by_index] values = [value for value in values if value is not None] returns = np.array([value["return_pct"] for value in values], dtype=float) maes = np.array([value["mae_pct"] for value in values], dtype=float) output["horizons"][str(horizon)] = {} if not len(returns) else { "n": len(returns), "median_return_pct": float(np.median(returns)), "mean_return_pct": float(np.mean(returns)), "negative_return_pct": float(100 * np.mean(returns < 0)), "median_mae_pct": float(np.median(maes)), "max_mae_pct": float(np.min(maes)), } return output def _study_window(frame: pd.DataFrame, events: list[dict[str, float | int]], start: str, end: str | None) -> dict[str, object]: dates = pd.to_datetime(frame["time"]) valid_positions = set(np.flatnonzero((dates >= pd.Timestamp(start)) & ((dates <= pd.Timestamp(end)) if end else True))) # Both event and control need a complete 14-bar outcome inside this named # evaluation window; this prevents IS returns leaking into OOS. usable_positions = {index for index in valid_positions if all(index + horizon in valid_positions for horizon in HORIZONS)} local_events = [event for event in events if int(event["event_index"]) in usable_positions] # Exclude every detected event horizon, even when an event's 14-bar # outcome crosses this reporting-window boundary and is not reportable. controls = match_controls(frame, local_events, allowed_indices=usable_positions, forbidden_events=events) closes = frame["close"].to_numpy(float) return {"bars": len(valid_positions), "events": _summarize([int(event["event_index"]) for event in local_events], closes), "controls": _summarize(controls, closes), "control_count": len(controls)} def run(asset: str = "COINBASE_BTCUSD", tf: str = "6H") -> dict[str, object]: 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 scored = generate_signals(data, **params).reset_index(drop=True) scores = scored["activation_score"].to_numpy(float) all_events: dict[str, list[dict[str, float | int]]] = {} for width in PIVOT_WIDTHS: all_events[f"{width}/{width}"] = detect_deterioration_events(confirmed_pivot_highs(scores, width, width)) # Combined is the predeclared union: events detected by any of 2/2, 3/3, # or 4/4, sorted chronologically then subject to the same 14-bar lock. union = [event for events in all_events.values() for event in events] union.sort(key=lambda event: int(event["event_index"])) combined: list[dict[str, float | int]] = [] locked_until = -1 for event in union: if int(event["event_index"]) > locked_until: combined.append(event) locked_until = int(event["event_index"]) + 14 all_events["combined (union of 2/2, 3/3, 4/4)"] = combined return {"asset": asset, "timeframe": tf, "definitions": {"first_peak_min": 300, "decline_min": 200, "second_to_first_max": .6, "confirmation_spacing_bars": [8, 40], "dedup_horizon_bars": 14, "control_score_band": 150, "control_region_days": 180}, "studies": {name: {"IS": _study_window(scored, events, SCORE_START, TRAIN_END), "OOS": _study_window(scored, events, OOS_START, None), "WFO": {start: _study_window(scored, events, start, end) for _, start, end in WFO_FOLDS}} for name, events in all_events.items()}} def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--asset", default="COINBASE_BTCUSD") parser.add_argument("--tf", default="6H") parser.add_argument("--output", type=Path, help="Write machine-readable JSON (safe outside tracked inputs).") args = parser.parse_args() result = run(args.asset, args.tf.upper()) 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())