okx-codex-trader/scripts/search_eth_bearish_failure_confirmation.py

from __future__ import annotations

import argparse
from dataclasses import dataclass
from pathlib import Path

import pandas as pd


DATA_DIR = Path("data/okx-candles")
OUTPUT_DIR = Path("reports/eth-exploration")
SYMBOL = "ETH-USDT-SWAP"
BTC_SYMBOL = "BTC-USDT-SWAP"
INITIAL_EQUITY = 10_000.0
FEE = 0.0004
ROUNDTRIP_FEE = FEE * 2
HORIZONS = (
    ("full", None),
    ("3y", pd.DateOffset(years=3)),
    ("1y", pd.DateOffset(years=1)),
    ("6m", pd.DateOffset(months=6)),
    ("3m", pd.DateOffset(months=3)),
)


@dataclass(frozen=True)
class Spec:
    family: str
    bar: str
    fast: int
    slow: int
    lookback: int
    threshold: float
    stop: float
    take: float
    hold: int
    gate: str

    @property
    def name(self) -> str:
        return (
            f"{self.family}-{self.bar}-f{self.fast}-s{self.slow}-lb{self.lookback}"
            f"-th{self.threshold:g}-sl{self.stop:g}-tp{self.take:g}-h{self.hold}-{self.gate}"
        )


def load_frame(symbol: str) -> pd.DataFrame:
    frame = pd.read_csv(DATA_DIR / symbol / "15m.csv")
    frame["ts"] = pd.to_datetime(frame["ts"], unit="ms", utc=True)
    return frame.sort_values("ts").drop_duplicates("ts", keep="last").set_index("ts")


def resample(frame: pd.DataFrame, bar: str) -> pd.DataFrame:
    rule = {"15m": "15min", "1H": "1h", "4H": "4h"}[bar]
    return (
        frame.resample(rule, label="left", closed="left")
        .agg(
            open=("open", "first"),
            high=("high", "max"),
            low=("low", "min"),
            close=("close", "last"),
            volume=("volume", "sum"),
        )
        .dropna()
    )


def rsi(close: pd.Series, length: int) -> pd.Series:
    diff = close.diff()
    gain = diff.clip(lower=0).ewm(alpha=1 / length, adjust=False).mean()
    loss = (-diff.clip(upper=0)).ewm(alpha=1 / length, adjust=False).mean()
    return 100 - 100 / (1 + gain / loss)


def joined_frames(eth: pd.DataFrame, btc: pd.DataFrame) -> pd.DataFrame:
    return eth.join(btc[["close"]].rename(columns={"close": "btc_close"}), how="inner")


def risk_gate(frame: pd.DataFrame, gate: str) -> pd.Series:
    if gate == "none":
        return pd.Series(True, index=frame.index)
    btc = frame["btc_close"]
    btc_slow = btc.rolling(160).mean()
    btc_return = btc / btc.shift(24) - 1
    btc_vol = btc.pct_change().rolling(48).std()
    if gate == "btc_riskoff":
        return (btc < btc_slow) & (btc_return < -0.01)
    if gate == "btc_riskoff_vol":
        return (btc < btc_slow) & (btc_return < 0) & (btc_vol > btc_vol.rolling(240).median())
    raise ValueError(gate)


def signals(spec: Spec, frame: pd.DataFrame) -> tuple[pd.Series, pd.Series]:
    close = frame["close"]
    high = frame["high"]
    low = frame["low"]
    open_ = frame["open"]
    fast = close.ewm(span=spec.fast, adjust=False).mean()
    slow = close.ewm(span=spec.slow, adjust=False).mean()
    rsi14 = rsi(close, 14)
    ret = close / close.shift(spec.lookback) - 1
    body = (close - open_) / open_
    range_pct = (high - low) / close
    range_rank = range_pct.rolling(200).rank(pct=True)
    volume_rank = frame["volume"].rolling(200).rank(pct=True)
    gate = risk_gate(frame, spec.gate)

    if spec.family == "mr_failure":
        prior_oversold = rsi14.shift(2).rolling(spec.lookback).min() < 34
        rebound_to_fast = high >= fast
        failed_reclaim = (close < fast) & (close < open_) & (ret > spec.threshold * 0.25)
        entry = gate & (close < slow) & prior_oversold & rebound_to_fast & failed_reclaim
        exit_ = (close > fast) | (rsi14 < 32)
    elif spec.family == "vol_second_confirm":
        prior_expansion = (
            (range_rank.shift(1) > 0.82)
            & (volume_rank.shift(1) > 0.60)
            & (body.shift(1) < -spec.threshold)
        )
        second_fail = (high < high.shift(1)) & (close < (open_.shift(1) + close.shift(1)) / 2) & (close < open_)
        entry = gate & (close < slow) & prior_expansion & second_fail
        exit_ = (close > fast) | (rsi14 < 30)
    elif spec.family == "trend_exhaustion":
        downtrend = (fast < slow) & (slow < slow.shift(spec.lookback))
        relief = close / close.rolling(spec.lookback * 2).min() - 1
        rejection = (high > fast) & (close < fast) & (close < open_) & (rsi14 > 45)
        entry = gate & downtrend & (relief > spec.threshold) & rejection
        exit_ = (close > slow) | (rsi14 < 35)
    else:
        raise ValueError(spec.family)
    return entry.fillna(False), exit_.fillna(False)


def close_return(entry: float, exit_: float) -> float:
    return entry / exit_ - 1 - ROUNDTRIP_FEE


def run_spec(spec: Spec, frame: pd.DataFrame) -> tuple[pd.Series, list[dict[str, object]]]:
    entry, exit_ = signals(spec, frame)
    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 - FEE)

        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 period_metrics(equity: pd.Series, trades: list[dict[str, object]], offset: pd.DateOffset | None) -> dict[str, object]:
    start = equity.index[0] if offset is None else equity.index[-1] - offset
    scoped = equity[equity.index >= start]
    scoped_trades = [trade for trade in trades if pd.Timestamp(trade["entry_time"]) >= scoped.index[0]]
    total = float(scoped.iloc[-1] / scoped.iloc[0] - 1)
    years = (scoped.index[-1] - scoped.index[0]).total_seconds() / 86_400 / 365
    annual = (1 + total) ** (1 / years) - 1 if total > -1 and years > 0 else 0.0
    drawdown = float(((scoped.cummax() - scoped) / scoped.cummax()).max())
    returns = [float(trade["return"]) for trade in scoped_trades]
    wins = [value for value in returns if value > 0]
    losses = [value for value in returns if value < 0]
    profit_factor = sum(wins) / abs(sum(losses)) if losses else (999.0 if wins else 0.0)
    return {
        "total_return": total,
        "annualized_return": annual,
        "max_drawdown": drawdown,
        "win_rate": len(wins) / len(returns) if returns else 0.0,
        "profit_factor": profit_factor,
        "trades": len(returns),
    }


def stability_rows(equity: pd.Series, trades: list[dict[str, object]]) -> pd.DataFrame:
    rows: list[dict[str, object]] = []
    for freq, label_name in (("YE", "year"), ("ME", "month")):
        sampled = equity.resample(freq).last().dropna()
        starts = equity.resample(freq).first().reindex(sampled.index)
        returns = sampled / starts - 1
        period_freq = "Y" if freq == "YE" else "M"
        periods = sampled.index.tz_localize(None).to_period(period_freq)
        for period, value in zip(periods.astype(str), returns):
            scoped = [
                trade
                for trade in trades
                if pd.Timestamp(trade["entry_time"]).tz_localize(None).to_period(period_freq) == pd.Period(period)
            ]
            rows.append({"bucket": label_name, "period": period, "return": float(value), "trades": len(scoped)})
    return pd.DataFrame(rows)


def build_specs() -> list[Spec]:
    specs: list[Spec] = []
    for bar in ("1H", "4H"):
        for fast, slow in ((20, 120), (34, 180), (50, 240)):
            for gate in ("none", "btc_riskoff", "btc_riskoff_vol"):
                for lookback in (8, 16, 24):
                    for threshold in (0.012, 0.02, 0.032):
                        specs.append(Spec("mr_failure", bar, fast, slow, lookback, threshold, 0.025, 0.04, 48, gate))
                        specs.append(Spec("trend_exhaustion", bar, fast, slow, lookback, threshold, 0.03, 0.045, 72, gate))
                for threshold in (0.008, 0.014, 0.02):
                    specs.append(Spec("vol_second_confirm", bar, fast, slow, 8, threshold, 0.025, 0.045, 48, gate))
    return specs


def row_for_spec(spec: Spec, equity: pd.Series, trades: list[dict[str, object]]) -> dict[str, object]:
    row: dict[str, object] = {"name": spec.name, "family": spec.family, "bar": spec.bar, "gate": spec.gate}
    for label, offset in HORIZONS:
        metrics = period_metrics(equity, trades, offset)
        for key, value in metrics.items():
            row[f"{label}_{key}"] = value
    return row


def markdown_table(frame: pd.DataFrame) -> str:
    def cell(value: object) -> str:
        if isinstance(value, float):
            return f"{value:.4f}"
        return str(value).replace("|", "\\|")

    rows = [list(frame.columns), ["---" for _ in frame.columns]]
    rows.extend(frame.astype(object).where(pd.notna(frame), "").values.tolist())
    return "\n".join("| " + " | ".join(cell(value) for value in row) + " |" for row in rows)


def markdown_report(totals: pd.DataFrame, selected: pd.Series, stability: pd.DataFrame) -> str:
    metric_rows = []
    for label, _ in HORIZONS:
        metric_rows.append(
            {
                "period": label,
                "total_return": selected[f"{label}_total_return"],
                "annualized_return": selected[f"{label}_annualized_return"],
                "max_drawdown": selected[f"{label}_max_drawdown"],
                "win_rate": selected[f"{label}_win_rate"],
                "profit_factor": selected[f"{label}_profit_factor"],
                "trades": selected[f"{label}_trades"],
            }
        )
    keep = [
        "name",
        "family",
        "bar",
        "gate",
        "full_total_return",
        "full_annualized_return",
        "full_max_drawdown",
        "full_win_rate",
        "full_profit_factor",
        "full_trades",
        "3y_total_return",
        "1y_total_return",
        "6m_total_return",
        "3m_total_return",
    ]
    years = stability[stability["bucket"] == "year"]
    months = stability[stability["bucket"] == "month"]
    losing_years = int((years["return"] < 0).sum())
    losing_months = int((months["return"] < 0).sum())
    active_months = months[months["trades"] > 0]
    verdict = "not worth continuing"
    if (
        selected["full_profit_factor"] > 1.12
        and selected["3y_total_return"] > 0
        and selected["1y_total_return"] > 0
        and selected["6m_total_return"] > 0
        and selected["3m_total_return"] > 0
        and losing_years <= 2
    ):
        verdict = "worth a narrow follow-up, but only after reducing drawdown"

    return (
        "# ETH Bearish Failure/Confirmation Search\n\n"
        "Scope: local OKX ETH/BTC candle CSV only; ETH short-only entries; BTC absolute risk-off filters only. "
        "Excluded: staged entry, ETH/BTC relative momentum, crash-follow, calendar/time buckets.\n\n"
        "Families: counter-trend mean-reversion failure, volatility expansion second confirmation, and trend exhaustion.\n\n"
        f"Selected candidate: `{selected['name']}`.\n\n"
        f"Verdict: {verdict}.\n\n"
        "## Selected metrics\n\n"
        f"{markdown_table(pd.DataFrame(metric_rows))}\n\n"
        "## Stability\n\n"
        f"Years: {len(years)}, losing years: {losing_years}. "
        f"Months: {len(months)}, active months: {len(active_months)}, losing months: {losing_months}.\n\n"
        f"{markdown_table(years[['period', 'return', 'trades']])}\n\n"
        "Worst active months:\n\n"
        f"{markdown_table(active_months.sort_values('return').head(12)[['period', 'return', 'trades']])}\n\n"
        "## Top 10\n\n"
        f"{markdown_table(totals.head(10)[keep])}\n"
    )


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

    eth_15m = load_frame(SYMBOL)
    btc_15m = load_frame(BTC_SYMBOL)
    frames = {bar: joined_frames(resample(eth_15m, bar), resample(btc_15m, bar)) for bar in ("1H", "4H")}

    rows = []
    curves: dict[str, pd.Series] = {}
    trade_sets: dict[str, list[dict[str, object]]] = {}
    specs = build_specs()
    for index, spec in enumerate(specs, start=1):
        equity, trades = run_spec(spec, frames[spec.bar])
        rows.append(row_for_spec(spec, equity, trades))
        curves[spec.name] = equity
        trade_sets[spec.name] = trades
        if index % 100 == 0:
            print(f"done {index}/{len(specs)}", flush=True)

    totals = pd.DataFrame(rows).sort_values(
        ["full_total_return", "3y_total_return", "1y_total_return", "full_profit_factor"],
        ascending=[False, False, False, False],
    )
    viable = totals[
        (totals["full_trades"] >= 25)
        & (totals["full_profit_factor"] > 1)
        & (totals["3y_total_return"] > 0)
        & (totals["1y_total_return"] > 0)
        & (totals["6m_total_return"] > 0)
        & (totals["3m_total_return"] > 0)
    ]
    selected = (viable if len(viable) else totals).iloc[0]
    stability = stability_rows(curves[str(selected["name"])], trade_sets[str(selected["name"])])

    args.output_dir.mkdir(parents=True, exist_ok=True)
    totals_path = args.output_dir / "eth-bearish-failure-confirmation-totals.csv"
    stability_path = args.output_dir / "eth-bearish-failure-confirmation-stability.csv"
    report_path = args.output_dir / "eth-bearish-failure-confirmation-report.md"
    totals.to_csv(totals_path, index=False)
    stability.to_csv(stability_path, index=False)
    report_path.write_text(markdown_report(totals, selected, stability), encoding="utf-8")
    print(markdown_table(pd.DataFrame([selected]).drop(columns=["name"]).iloc[:, :12]))
    print(f"selected {selected['name']}")
    print(f"wrote {totals_path}, {stability_path}, {report_path}")
    return 0


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