okx-codex-trader/scripts/search_eth_bearish_price_proxy.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:
    path = DATA_DIR / symbol / "15m.csv"
    frame = pd.read_csv(path)
    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:
    eth = frame["close"]
    btc = frame["btc_close"]
    if gate == "none":
        return pd.Series(True, index=frame.index)
    if gate == "btc_riskoff":
        btc_slow = btc.rolling(120).mean()
        btc_drop = btc / btc.shift(24) - 1
        return (btc < btc_slow) & (btc_drop < -0.015)
    if gate == "eth_riskoff":
        eth_slow = eth.rolling(160).mean()
        eth_drop = eth / eth.shift(24) - 1
        return (eth < eth_slow) & (eth_drop < -0.012)
    raise ValueError(gate)


def signals(spec: Spec, frame: pd.DataFrame) -> tuple[pd.Series, 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()
    ret = close / close.shift(spec.lookback) - 1
    range_pct = (high - low) / close
    range_rank = range_pct.rolling(160).rank(pct=True)
    vol_rank = frame["volume"].rolling(160).rank(pct=True)
    rsi14 = rsi(close, 14)
    gate = risk_gate(frame, spec.gate)

    if spec.family == "crash_follow":
        entry = gate & (close < slow) & (ret < -spec.threshold) & (range_rank > 0.75)
        exit_ = (close > fast) | (rsi14 > 52)
        side = pd.Series("short", index=frame.index)
    elif spec.family == "rebound_exhaustion":
        prior_drop = close / close.shift(spec.lookback * 2) - 1
        rebound = close / close.shift(spec.lookback) - 1
        weak_close = close < open_
        entry = gate & (close < slow) & (prior_drop < -spec.threshold * 1.4) & (rebound > spec.threshold * 0.45) & (high >= fast) & weak_close
        exit_ = close > slow
        side = pd.Series("short", index=frame.index)
    elif spec.family == "candle_funding_proxy":
        premium_proxy = close / slow - 1
        failed_high = (high / close - 1) > range_pct.rolling(80).median()
        entry = gate & (premium_proxy > spec.threshold) & (rsi14 > 62) & (vol_rank > 0.65) & failed_high & (close < open_)
        exit_ = (close < fast) | (rsi14 < 45)
        side = pd.Series("short", index=frame.index)
    elif spec.family == "riskoff_bidir":
        breakdown = gate & (close < slow) & (ret < -spec.threshold)
        capitulation_rebound = (close > fast) & (rsi14 < 35) & (range_rank > 0.75)
        entry = breakdown | capitulation_rebound
        exit_ = close > fast
        side = pd.Series("short", index=frame.index)
        side = side.mask(capitulation_rebound, "long")
    else:
        raise ValueError(spec.family)
    return entry.fillna(False), exit_.fillna(False), side


def close_return(side: str, entry: float, exit_: float) -> float:
    gross = exit_ / entry - 1 if side == "long" else entry / exit_ - 1
    return gross - ROUNDTRIP_FEE


def run_spec(spec: Spec, frame: pd.DataFrame) -> tuple[pd.Series, list[dict[str, object]]]:
    entry, exit_, side_signal = signals(spec, frame)
    fast = frame["close"].ewm(span=spec.fast, adjust=False).mean()
    warmup = max(spec.slow, 180, spec.lookback * 2) + 2
    equity = INITIAL_EQUITY
    position: dict[str, object] | None = None
    pending_entry: str | None = None
    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(str(position["side"]), float(position["entry_price"]), float(candle.open))
            equity *= 1 + net
            trades.append({"entry_time": position["entry_time"], "exit_time": ts, "side": position["side"], "return": net})
            position = None
            pending_exit = False
        if pending_entry and position is None and equity > 0:
            side = pending_entry
            position = {
                "side": side,
                "entry_time": ts,
                "entry_index": index,
                "entry_price": float(candle.open),
                "stop": float(candle.open) * (1 - spec.stop if side == "long" else 1 + spec.stop),
                "take": float(candle.open) * (1 + spec.take if side == "long" else 1 - spec.take),
            }
            pending_entry = None

        mark = equity
        if position is not None:
            side = str(position["side"])
            stop_hit = candle.low <= float(position["stop"]) if side == "long" else candle.high >= float(position["stop"])
            take_hit = candle.high >= float(position["take"]) if side == "long" else candle.low <= float(position["take"])
            if stop_hit or take_hit:
                price = float(position["stop"] if stop_hit else position["take"])
                net = close_return(side, float(position["entry_price"]), price)
                equity *= 1 + net
                trades.append({"entry_time": position["entry_time"], "exit_time": ts, "side": side, "return": net})
                position = None
                mark = equity
            else:
                gross = candle.close / float(position["entry_price"]) - 1 if side == "long" else 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 = str(side_signal.iloc[index])
        elif position is not None:
            held = index - int(position["entry_index"])
            exit_now = bool(exit_.iloc[index])
            if spec.family == "riskoff_bidir" and position["side"] == "long":
                exit_now = bool(frame["close"].iloc[index] < fast.iloc[index])
            if exit_now or held >= 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 (0.0 if not wins else 999.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 build_specs() -> list[Spec]:
    specs: list[Spec] = []
    trend_pairs = ((20, 120), (40, 240))
    risk_gates = ("btc_riskoff", "eth_riskoff")
    for bar in ("15m", "1H", "4H"):
        for fast, slow in trend_pairs:
            for lookback in (8, 24):
                for threshold in (0.02, 0.035):
                    specs.append(Spec("crash_follow", bar, fast, slow, lookback, threshold, 0.02, 0.035, 48, "none"))
                    for gate in risk_gates:
                        specs.append(Spec("crash_follow", bar, fast, slow, lookback, threshold, 0.02, 0.06, 96, gate))
                        specs.append(Spec("rebound_exhaustion", bar, fast, slow, lookback, threshold, 0.035, 0.035, 48, gate))
                        specs.append(Spec("riskoff_bidir", bar, fast, slow, lookback, threshold, 0.02, 0.035, 48, gate))
            for threshold in (0.02, 0.035):
                specs.append(Spec("candle_funding_proxy", bar, fast, slow, 16, threshold, 0.02, 0.035, 48, "none"))
                specs.append(Spec("candle_funding_proxy", bar, fast, slow, 16, threshold, 0.02, 0.06, 96, "eth_riskoff"))
    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,
        "short_ratio": sum(1 for trade in trades if trade["side"] == "short") / len(trades) if trades else 0.0,
    }
    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, path: Path) -> str:
    top = totals.head(10)
    best_full = totals.iloc[0]
    positive = totals[
        (totals["full_total_return"] > 0)
        & (totals["3y_total_return"] > 0)
        & (totals["1y_total_return"] > 0)
        & (totals["6m_total_return"] > 0)
        & (totals["3m_total_return"] > 0)
    ].sort_values(["full_total_return", "3m_total_return"], ascending=[False, False])
    best_positive = positive.iloc[0] if len(positive) else best_full
    keep = [
        "name",
        "family",
        "bar",
        "gate",
        "short_ratio",
        "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",
    ]
    table = markdown_table(top[keep])
    period_rows = []
    for label, _ in HORIZONS:
        period_rows.append(
            {
                "period": label,
                "total_return": best_positive[f"{label}_total_return"],
                "annualized_return": best_positive[f"{label}_annualized_return"],
                "max_drawdown": best_positive[f"{label}_max_drawdown"],
                "win_rate": best_positive[f"{label}_win_rate"],
                "profit_factor": best_positive[f"{label}_profit_factor"],
                "trades": best_positive[f"{label}_trades"],
            }
        )
    period_table = markdown_table(pd.DataFrame(period_rows))
    positive_table = markdown_table(positive[keep].head(7)) if len(positive) else "No candidate was positive across all requested windows."
    verdict = (
        "worth continuing as a narrow crash-follow research branch, but not ready for live work"
        if len(positive) and best_positive["full_profit_factor"] > 1.1
        else "not worth continuing yet"
    )
    return (
        "# ETH Bearish Price-Proxy Search\n\n"
        f"Output: `{path}`\n\n"
        "Scope: read-only local OKX candles under `data/okx-candles`; ETH trades, BTC absolute risk-off filter only; no staged entry, no ETH/BTC relative momentum, no live path.\n\n"
        "Families: crash-follow short, rebound-exhaustion short, candle-only funding proxy short, and risk-off bidirectional with bearish bias.\n\n"
        f"Best full-sample candidate: `{best_full['name']}`; it fails the 3m window.\n\n"
        f"Best all-window-positive candidate: `{best_positive['name']}`. Verdict: {verdict}.\n\n"
        "## Best all-window-positive metrics\n\n"
        f"{period_table}\n\n"
        "## All-window-positive candidates\n\n"
        f"{positive_table}\n\n"
        "## Top 10\n\n"
        f"{table}\n"
    )


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--output-dir", type=Path, default=OUTPUT_DIR)
    parser.add_argument("--max-candidates", type=int, default=0)
    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 ("15m", "1H", "4H")}
    specs = build_specs()
    if args.max_candidates:
        specs = specs[: args.max_candidates]

    rows = []
    for index, spec in enumerate(specs, start=1):
        equity, trades = run_spec(spec, frames[spec.bar])
        rows.append(row_for_spec(spec, equity, 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],
    )
    args.output_dir.mkdir(parents=True, exist_ok=True)
    totals_path = args.output_dir / "eth-bearish-price-proxy-totals.csv"
    report_path = args.output_dir / "eth-bearish-price-proxy-report.md"
    totals.to_csv(totals_path, index=False)
    report_path.write_text(markdown_report(totals, totals_path), encoding="utf-8")
    print(totals.head(10).to_string(index=False))
    print(f"wrote {totals_path} and {report_path}")
    return 0


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