okx-codex-trader/scripts/validate_trend_exhaustion_candidate.py

from __future__ import annotations

import argparse
import sys
from dataclasses import replace
from pathlib import Path

import pandas as pd

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

from scripts.search_eth_bearish_failure_confirmation import (
    INITIAL_EQUITY,
    Spec,
    close_return,
    joined_frames,
    load_frame,
    markdown_table,
    period_metrics,
    resample,
    signals,
)
from scripts.search_long_short_fusion import component_returns, metrics


OUTPUT_DIR = Path("reports/eth-exploration")
PREFIX = "trend-exhaustion-narrow-validation"
BASELINE_EQUITY = Path("reports/eth-exploration/eth-focused-portfolio-conservative-equity.csv")
BASELINE_PORTFOLIO = "all_legs-risk-3-c0124-eth_btc_rsi_filter+btc_lead_eth_lag_15m+eth_robust_twap"
CANDIDATE = Spec("trend_exhaustion", "1H", 50, 240, 8, 0.012, 0.03, 0.045, 72, "none")
HORIZONS = (
    ("full", None),
    ("3y", pd.DateOffset(years=3)),
    ("1y", pd.DateOffset(years=1)),
    ("6m", pd.DateOffset(months=6)),
    ("3m", pd.DateOffset(months=3)),
)
OVERLAY_WEIGHTS = (0.025, 0.05, 0.075, 0.10)


def run_spec(spec: Spec, frame: pd.DataFrame, entry_filter: pd.Series | None = None) -> tuple[pd.Series, list[dict[str, object]]]:
    entry, exit_ = signals(spec, frame)
    if entry_filter is not None:
        entry = entry & entry_filter.reindex(entry.index).fillna(False)

    warmup = max(spec.slow, 260, spec.lookback * 3) + 2
    equity = INITIAL_EQUITY
    position: dict[str, object] | None = None
    pending_entry = False
    pending_exit = False
    trades: list[dict[str, object]] = []
    curve: list[tuple[pd.Timestamp, float]] = []

    rows = list(frame.itertuples())
    for index in range(warmup, len(rows)):
        candle = rows[index]
        ts = frame.index[index]
        if pending_exit and position is not None:
            net = close_return(float(position["entry_price"]), float(candle.open))
            equity *= 1 + net
            trades.append({"entry_time": position["entry_time"], "exit_time": ts, "return": net})
            position = None
            pending_exit = False
        if pending_entry and position is None and equity > 0:
            position = {
                "entry_time": ts,
                "entry_index": index,
                "entry_price": float(candle.open),
                "stop": float(candle.open) * (1 + spec.stop),
                "take": float(candle.open) * (1 - spec.take),
            }
            pending_entry = False

        mark = equity
        if position is not None:
            stop_hit = candle.high >= float(position["stop"])
            take_hit = candle.low <= float(position["take"])
            if stop_hit or take_hit:
                price = float(position["stop"] if stop_hit else position["take"])
                net = close_return(float(position["entry_price"]), price)
                equity *= 1 + net
                trades.append({"entry_time": position["entry_time"], "exit_time": ts, "return": net})
                position = None
                mark = equity
            else:
                gross = float(position["entry_price"]) / candle.close - 1
                mark = equity * (1 + gross - 0.0004)

        curve.append((ts, mark))
        if index == len(rows) - 1 or equity <= 0:
            continue
        if position is None and bool(entry.iloc[index]):
            pending_entry = True
        elif position is not None and (bool(exit_.iloc[index]) or index - int(position["entry_index"]) >= spec.hold):
            pending_exit = True

    series = pd.Series({ts: value for ts, value in curve}).sort_index()
    daily = series.resample("1D").last().ffill()
    daily = pd.concat([pd.Series([INITIAL_EQUITY], index=[daily.index[0].normalize()]), daily]).sort_index()
    return daily.groupby(level=0).last(), trades


def row_for_spec(name: str, equity: pd.Series, trades: list[dict[str, object]]) -> dict[str, object]:
    row: dict[str, object] = {"name": name}
    for label, offset in HORIZONS:
        for key, value in period_metrics(equity, trades, offset).items():
            row[f"{label}_{key}"] = value
    years = yearly_returns(equity)
    row["return_2022"] = float(years.get("2022", 0.0))
    row["return_2023"] = float(years.get("2023", 0.0))
    row["worst_year"] = float(years.min()) if len(years) else 0.0
    row["left_tail_2022_2023"] = min(row["return_2022"], row["return_2023"])
    return row


def yearly_returns(equity: pd.Series) -> pd.Series:
    sampled = equity.resample("YE").last().dropna()
    starts = equity.resample("YE").first().reindex(sampled.index)
    returns = sampled / starts - 1.0
    returns.index = sampled.index.tz_localize(None).to_period("Y").astype(str)
    return returns


def neighborhood_specs() -> list[Spec]:
    specs = {CANDIDATE.name: CANDIDATE}
    for fast, slow in ((40, 220), (40, 240), (50, 220), (50, 260), (60, 240), (60, 260)):
        spec = replace(CANDIDATE, fast=fast, slow=slow)
        specs[spec.name] = spec
    for field, values in (
        ("lookback", (6, 10)),
        ("threshold", (0.010, 0.014)),
        ("stop", (0.025, 0.035)),
        ("take", (0.040, 0.050)),
        ("hold", (60, 84)),
    ):
        for value in values:
            spec = replace(CANDIDATE, **{field: value})
            specs[spec.name] = spec
    return list(specs.values())


def feature_filters(frame: pd.DataFrame) -> dict[str, pd.Series]:
    close = frame["close"]
    btc = frame["btc_close"]
    slow = close.ewm(span=CANDIDATE.slow, adjust=False).mean()
    eth_slope = slow / slow.shift(24) - 1.0
    btc_sma = btc.rolling(240).mean()
    btc_slope = btc_sma / btc_sma.shift(24) - 1.0
    eth_vol = close.pct_change().rolling(72).std()
    vol_rank = eth_vol.rolling(720).rank(pct=True)
    return {
        "btc_above_sma240": btc > btc_sma,
        "btc_sma240_slope_ge_0": btc_slope >= 0.0,
        "eth_slope_ge_-0.015": eth_slope >= -0.015,
        "eth_slope_-0.015_to_0": (eth_slope >= -0.015) & (eth_slope <= 0.0),
        "vol_rank_0.35_to_0.85": (vol_rank >= 0.35) & (vol_rank <= 0.85),
        "vol_rank_le_0.85": vol_rank <= 0.85,
        "btc_up_and_mid_vol": (btc > btc_sma) & (vol_rank >= 0.35) & (vol_rank <= 0.85),
        "btc_up_eth_slope_ge_-0.015": (btc > btc_sma) & (eth_slope >= -0.015),
        "btc_up_mid_vol_eth_slope": (btc > btc_sma) & (vol_rank >= 0.35) & (vol_rank <= 0.85) & (eth_slope >= -0.015),
        "btc_slope_up_mid_vol_eth_slope": (btc_slope >= 0.0) & (vol_rank >= 0.35) & (vol_rank <= 0.85) & (eth_slope >= -0.015),
    }


def baseline_equity(path: Path, portfolio: str) -> pd.Series:
    frame = pd.read_csv(path)
    selected = frame[
        (frame["portfolio"] == portfolio)
        & (frame["cost_model"] == "maker_taker")
        & (frame["scope"] == "all_legs")
    ].copy()
    selected["date"] = pd.to_datetime(selected["date"], utc=True)
    series = selected.sort_values("date").set_index("date")["equity"].astype(float)
    series.name = portfolio
    return series


def overlay_rows(base: pd.Series, overlay: pd.Series) -> pd.DataFrame:
    base_metrics = {label: horizon_metrics(base, offset) for label, offset in HORIZONS}
    rows = []
    overlay_returns = component_returns(overlay)
    for weight in OVERLAY_WEIGHTS:
        aligned = pd.DataFrame({"base": component_returns(base), "overlay": overlay_returns}).dropna()
        combined = aligned["base"] + aligned["overlay"] * weight
        equity = INITIAL_EQUITY * (1.0 + combined).cumprod()
        for label, offset in HORIZONS:
            row = horizon_metrics(equity, offset)
            baseline = base_metrics[label]
            rows.append(
                {
                    "overlay_weight": weight,
                    "horizon": label,
                    **row,
                    "baseline_total_return": baseline["total_return"],
                    "baseline_max_drawdown": baseline["max_drawdown"],
                    "baseline_calmar": baseline["calmar"],
                    "delta_total_return": row["total_return"] - baseline["total_return"],
                    "delta_max_drawdown": row["max_drawdown"] - baseline["max_drawdown"],
                    "delta_calmar": row["calmar"] - baseline["calmar"],
                }
            )
    return pd.DataFrame(rows)


def horizon_metrics(series: pd.Series, offset: pd.DateOffset | None) -> dict[str, object]:
    scoped = series if offset is None else series[series.index >= series.index[-1] - offset]
    if len(scoped) < 2:
        scoped = series
    return {"start": scoped.index[0].strftime("%Y-%m-%d"), "end": scoped.index[-1].strftime("%Y-%m-%d"), **metrics(scoped)}


def report_text(paths: list[Path], neighborhood: pd.DataFrame, filters: pd.DataFrame, overlay: pd.DataFrame, selected_name: str) -> str:
    base = filters[filters["name"] == "unfiltered"].iloc[0]
    best_filter = filters.iloc[0]
    full_overlay = overlay[overlay["horizon"] == "full"].sort_values(["delta_calmar", "delta_max_drawdown"], ascending=[False, True])
    best_overlay = full_overlay.iloc[0]
    neighborhood_pass = int(
        (
            (neighborhood["full_total_return"] > 0.0)
            & (neighborhood["3y_total_return"] > 0.0)
            & (neighborhood["1y_total_return"] > 0.0)
            & (neighborhood["return_2022"] > -0.10)
            & (neighborhood["return_2023"] > -0.05)
        ).sum()
    )
    include = bool(
        neighborhood_pass >= 5
        and best_filter["return_2022"] > -0.10
        and best_filter["return_2023"] > -0.05
        and best_filter["full_total_return"] > 0.0
        and best_overlay["delta_calmar"] > 0.0
        and best_overlay["delta_max_drawdown"] <= 0.0
    )
    verdict = (
        f"Include `{CANDIDATE.name}` with `{selected_name}` as a capped 0.025-0.10 overlay dimension in the conservative ETH portfolio search."
        if include
        else f"Reject `{CANDIDATE.name}` for portfolio search. The narrow validation does not satisfy stability, left-tail, and overlay drawdown/Calmar requirements together."
    )
    keep = [
        "name",
        "full_total_return",
        "full_annualized_return",
        "full_max_drawdown",
        "full_profit_factor",
        "full_trades",
        "3y_total_return",
        "1y_total_return",
        "6m_total_return",
        "3m_total_return",
        "return_2022",
        "return_2023",
        "worst_year",
    ]
    overlay_keep = [
        "overlay_weight",
        "horizon",
        "total_return",
        "max_drawdown",
        "calmar",
        "delta_total_return",
        "delta_max_drawdown",
        "delta_calmar",
    ]
    return "\n".join(
        [
            "# Trend Exhaustion Narrow Validation",
            "",
            "Run command: `rtk .venv/bin/python scripts/validate_trend_exhaustion_candidate.py`",
            "",
            "Output files:",
            *[f"- `{path}`" for path in paths],
            "",
            f"Candidate: `{CANDIDATE.name}`.",
            "Scope: local OKX `ETH-USDT-SWAP` and `BTC-USDT-SWAP` candles only. No live path touched.",
            "All filters use current or historical completed 1H candles; entries execute on the next open.",
            "",
            "## Small Parameter Neighborhood",
            "",
            f"Neighborhood pass count under fixed left-tail thresholds: {neighborhood_pass}/{len(neighborhood)}.",
            "",
            markdown_table(neighborhood.sort_values(["full_total_return", "left_tail_2022_2023"], ascending=[False, False]).head(12)[keep]),
            "",
            "## Structural Filters",
            "",
            "Filter objective: improve 2022/2023 left tail without using future candles.",
            "",
            markdown_table(filters.head(12)[keep]),
            "",
            "Unfiltered 2022/2023:",
            f"- 2022 `{base['return_2022']:.4f}`, 2023 `{base['return_2023']:.4f}`",
            "",
            "## Conservative Portfolio Overlay",
            "",
            f"Overlay source: `{CANDIDATE.name}` filtered by `{selected_name}`.",
            "",
            markdown_table(overlay[overlay["horizon"].isin(["full", "3y", "1y"])][overlay_keep]),
            "",
            "## Verdict",
            "",
            verdict,
            "",
        ]
    )


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--output-dir", type=Path, default=OUTPUT_DIR)
    parser.add_argument("--baseline-equity", type=Path, default=BASELINE_EQUITY)
    parser.add_argument("--baseline-portfolio", default=BASELINE_PORTFOLIO)
    args = parser.parse_args()

    eth = load_frame("ETH-USDT-SWAP")
    btc = load_frame("BTC-USDT-SWAP")
    frame = joined_frames(resample(eth, "1H"), resample(btc, "1H"))

    neighborhood_rows = []
    best_equity: pd.Series | None = None
    selected_name = ""
    for spec in neighborhood_specs():
        equity, trades = run_spec(spec, frame)
        row = row_for_spec(spec.name, equity, trades)
        neighborhood_rows.append(row)
        if spec == CANDIDATE:
            best_equity = equity
            selected_name = spec.name
    neighborhood = pd.DataFrame(neighborhood_rows)

    filter_rows = []
    filter_equities: dict[str, pd.Series] = {}
    equity, trades = run_spec(CANDIDATE, frame)
    filter_rows.append(row_for_spec("unfiltered", equity, trades))
    filter_equities["unfiltered"] = equity
    for name, mask in feature_filters(frame).items():
        equity, trades = run_spec(CANDIDATE, frame, mask)
        filter_rows.append(row_for_spec(name, equity, trades))
        filter_equities[name] = equity
    filters = pd.DataFrame(filter_rows).sort_values(
        ["left_tail_2022_2023", "full_max_drawdown", "full_total_return"],
        ascending=[False, True, False],
    )
    selected_filter_name = str(filters.iloc[0]["name"])
    selected_equity = filter_equities[selected_filter_name]
    if best_equity is None:
        best_equity = filter_equities["unfiltered"]
        selected_name = CANDIDATE.name

    overlay = overlay_rows(baseline_equity(args.baseline_equity, args.baseline_portfolio), selected_equity)

    args.output_dir.mkdir(parents=True, exist_ok=True)
    neighborhood_path = args.output_dir / f"{PREFIX}-neighborhood.csv"
    filters_path = args.output_dir / f"{PREFIX}-filters.csv"
    overlay_path = args.output_dir / f"{PREFIX}-overlay.csv"
    report_path = args.output_dir / f"{PREFIX}.md"
    neighborhood.to_csv(neighborhood_path, index=False)
    filters.to_csv(filters_path, index=False)
    overlay.to_csv(overlay_path, index=False)
    report_path.write_text(
        report_text(
            [neighborhood_path, filters_path, overlay_path, report_path],
            neighborhood,
            filters,
            overlay,
            selected_filter_name,
        ),
        encoding="utf-8",
    )
    print(report_path)
    print(filters.head(8).to_string(index=False))
    print(overlay[overlay["horizon"] == "full"].to_string(index=False))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())