okx-codex-trader/scripts/search_t_overlay_other_strategies.py

from __future__ import annotations

import argparse
import sys
from dataclasses import dataclass
from pathlib import Path

import pandas as pd

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

from okx_codex_trader.models import Candle
from okx_codex_trader.sampled_report import SegmentResult, mark_to_market, trade_equity


ETH_SYMBOL = "ETH-USDT-SWAP"
BTC_SYMBOL = "BTC-USDT-SWAP"
BAR = "15m"
YEARS = 10.0
LEVERAGE = 3
INITIAL_EQUITY = 10_000.0
DATA_DIR = Path("data/okx-candles")
OUTPUT_DIR = Path("reports/eth-exploration")
PRIMARY_COST = "maker_taker"
COSTS = (
    ("maker_maker", 0.0012),
    ("maker_taker", 0.0021),
    ("taker_taker", 0.0030),
)


@dataclass(frozen=True)
class StrategySpec:
    family: str
    symbol: str
    signal_symbol: str
    name: str
    params: dict[str, float | int | str]
    reentry_bars: int
    reentry_pullback_pct: float


def _format_ts(ts: int) -> str:
    return pd.to_datetime(ts, unit="ms", utc=True).strftime("%Y-%m-%d %H:%M")


def _load_candles(symbol: str, bar: str) -> list[Candle]:
    frame = pd.read_csv(DATA_DIR / symbol / f"{bar}.csv")
    return [
        Candle(symbol=symbol, ts=int(row.ts), open=float(row.open), high=float(row.high), low=float(row.low), close=float(row.close), volume=float(row.volume))
        for row in frame.itertuples(index=False)
    ]


def _align_pair(left: list[Candle], right: list[Candle]) -> tuple[list[Candle], list[Candle]]:
    right_by_ts = {candle.ts: candle for candle in right}
    left_out: list[Candle] = []
    right_out: list[Candle] = []
    for candle in left:
        other = right_by_ts.get(candle.ts)
        if other is not None:
            left_out.append(candle)
            right_out.append(other)
    return left_out, right_out


def _close_trade(trades: list[dict[str, object]], exits: list[dict[str, object]], position: dict[str, object], candle: Candle, exit_price: float, reason: str) -> tuple[float, bool]:
    margin_used = float(position["margin_used"])
    exit_equity = trade_equity(
        side=str(position["side"]),
        margin_used=margin_used,
        entry_price=float(position["entry_price"]),
        exit_price=exit_price,
        leverage=LEVERAGE,
    )
    pnl = exit_equity - margin_used
    trades.append(
        {
            "side": "Long" if position["side"] == "long" else "Short",
            "entry_time": _format_ts(int(position["entry_time"])),
            "exit_time": _format_ts(candle.ts),
            "entry_price": round(float(position["entry_price"]), 4),
            "exit_price": round(exit_price, 4),
            "pnl": round(pnl, 4),
            "return_pct": round(pnl / margin_used * 100.0, 4),
            "cost_weight": 1.0,
            "entry_kind": position["entry_kind"],
            "exit_reason": reason,
        }
    )
    exits.append({"ts": candle.ts, "price": exit_price, "side": position["side"]})
    return exit_equity, pnl > 0.0


def rsi(series: pd.Series, length: int) -> pd.Series:
    delta = series.diff()
    gain = delta.clip(lower=0.0).ewm(alpha=1 / length, adjust=False).mean()
    loss = (-delta.clip(upper=0.0)).ewm(alpha=1 / length, adjust=False).mean()
    rs = gain / loss.replace(0.0, pd.NA)
    return 100.0 - (100.0 / (1.0 + rs))


def true_range(frame: pd.DataFrame) -> pd.Series:
    high = frame["high"]
    low = frame["low"]
    close = frame["close"]
    return pd.concat([high - low, (high - close.shift(1)).abs(), (low - close.shift(1)).abs()], axis=1).max(axis=1)


def build_signals(spec: StrategySpec, frame: pd.DataFrame) -> tuple[pd.Series, pd.Series]:
    close = frame["close"]
    open_ = frame["open"]
    high = frame["high"]
    low = frame["low"]
    p = spec.params
    if spec.family == "range_momentum":
        lookback = int(p["lookback"])
        entry_high = high.shift(1).rolling(lookback).max()
        entry_low = low.shift(1).rolling(lookback).min()
        return (close > entry_high).fillna(False), (close < entry_low).fillna(False)
    if spec.family == "btc_lead_eth_lag":
        sig_close = frame["sig_close"]
        lead = int(p["lead_lookback"])
        btc_ret = sig_close / sig_close.shift(lead) - 1.0
        eth_ret = close / close.shift(lead) - 1.0
        entry = (btc_ret >= float(p["btc_return_threshold"])) & (eth_ret < btc_ret - float(p["lag_gap"]))
        exit_ = pd.Series(False, index=frame.index)
        return entry.fillna(False), exit_
    if spec.family == "short_bb_upper_rejection":
        trend = close.rolling(int(p["trend"])).mean()
        mid = close.rolling(int(p["bb"])).mean()
        std = close.rolling(int(p["bb"])).std(ddof=0)
        upper = mid + std * float(p["std"])
        atr = true_range(frame).rolling(int(p["atr"])).mean()
        entry = (close < trend) & (high >= upper) & (close < upper) & (close < open_) & atr.notna() & (atr > 0)
        exit_ = (close <= mid) | (close > trend)
        return entry.fillna(False), exit_.fillna(False)
    raise ValueError(f"unknown family {spec.family}")


def run_strategy(candles: list[Candle], spec: StrategySpec, with_t: bool) -> tuple[SegmentResult, dict[str, int]]:
    frame = pd.DataFrame([c.__dict__ for c in candles])
    if spec.signal_symbol != spec.symbol:
        signal = _load_candles(spec.signal_symbol, BAR)
        left, right = _align_pair(candles, signal)
        candles = left
        frame = pd.DataFrame([c.__dict__ for c in left])
        frame["sig_close"] = [c.close for c in right]
    entry_signal, exit_signal = build_signals(spec, frame)
    warmup = max(int(value) for key, value in spec.params.items() if isinstance(value, int) and key not in {"max_hold"})

    equity = INITIAL_EQUITY
    ending_equity = equity
    peak_equity = equity
    max_drawdown = 0.0
    wins = 0
    trades: list[dict[str, object]] = []
    entries: list[dict[str, object]] = []
    exits: list[dict[str, object]] = []
    equity_curve: list[dict[str, float | int]] = []
    position: dict[str, object] | None = None
    pending_entry: dict[str, str] | None = None
    pending_exit = False
    reentry: dict[str, object] | None = None
    stats = {"take_exits": 0, "reentries": 0}

    for index in range(warmup, len(candles)):
        candle = candles[index]
        if pending_exit and position is not None:
            equity, won = _close_trade(trades, exits, position, candle, candle.open, "signal_exit")
            wins += int(won)
            position = None
            pending_exit = False
        if pending_entry is not None and position is None and equity > 0.0:
            side = pending_entry["side"]
            stop = float(spec.params["stop"])
            take = float(spec.params["take"])
            position = {
                "side": side,
                "entry_kind": pending_entry["kind"],
                "entry_time": candle.ts,
                "entry_price": candle.open,
                "entry_index": index,
                "margin_used": equity,
                "stop_price": candle.open * (1.0 - stop if side == "long" else 1.0 + stop),
                "take_price": candle.open * (1.0 + take if side == "long" else 1.0 - take),
            }
            entries.append({"ts": candle.ts, "price": candle.open, "side": side})
            stats["reentries"] += int(pending_entry["kind"] == "reentry")
            pending_entry = None

        current_equity = equity
        if position is not None:
            side = str(position["side"])
            stop_hit = (side == "long" and candle.low <= float(position["stop_price"])) or (side == "short" and candle.high >= float(position["stop_price"]))
            take_hit = (side == "long" and candle.high >= float(position["take_price"])) or (side == "short" and candle.low <= float(position["take_price"]))
            if stop_hit or take_hit:
                reason = "stop" if stop_hit else "take"
                exit_price = float(position["stop_price"] if stop_hit else position["take_price"])
                equity, won = _close_trade(trades, exits, position, candle, exit_price, reason)
                wins += int(won)
                current_equity = equity
                if take_hit and not stop_hit and with_t:
                    stats["take_exits"] += 1
                    reentry = {"side": side, "until": index + spec.reentry_bars, "anchor": exit_price}
                position = None
        if position is not None:
            current_equity = mark_to_market(side=str(position["side"]), margin_used=float(position["margin_used"]), entry_price=float(position["entry_price"]), mark_price=candle.close, leverage=LEVERAGE)
        peak_equity = max(peak_equity, current_equity)
        max_drawdown = max(max_drawdown, (peak_equity - current_equity) / peak_equity)
        equity_curve.append({"ts": candle.ts, "equity": current_equity, "close": candle.close})
        ending_equity = current_equity
        if index == len(candles) - 1 or equity <= 0.0:
            continue

        if position is None and reentry is not None:
            side = str(reentry["side"])
            anchor = float(reentry["anchor"])
            if index > int(reentry["until"]):
                reentry = None
            elif (side == "long" and candle.close <= anchor * (1.0 - spec.reentry_pullback_pct)) or (
                side == "short" and candle.close >= anchor * (1.0 + spec.reentry_pullback_pct)
            ):
                pending_entry = {"side": side, "kind": "reentry"}
                reentry = None
                continue
        if position is not None:
            held = index - int(position["entry_index"])
            if bool(exit_signal.iloc[index]) or held >= int(spec.params["max_hold"]):
                pending_exit = True
            continue
        if reentry is not None:
            continue
        if bool(entry_signal.iloc[index]):
            side = "short" if spec.family == "short_bb_upper_rejection" else "long"
            if spec.family == "range_momentum":
                side = "long" if frame.iloc[index]["close"] > frame.iloc[max(index - 1, 0)]["close"] else "short"
            pending_entry = {"side": side, "kind": "initial"}

    result = SegmentResult(
        trade_count=len(trades),
        total_return=(ending_equity - INITIAL_EQUITY) / INITIAL_EQUITY,
        win_rate=wins / len(trades) if trades else 0.0,
        max_drawdown=max_drawdown,
        trades=trades,
        open_position=position,
        candles=candles[warmup:],
        equity_curve=equity_curve,
        entries=entries,
        exits=exits,
    )
    return result, stats


def cost_metrics(result: SegmentResult, cost: float, first_ts: int, last_ts: int) -> dict[str, float]:
    equity = INITIAL_EQUITY
    peak = equity
    dd = 0.0
    for trade in result.trades:
        equity *= 1.0 + float(trade["return_pct"]) / 100.0 - cost
        peak = max(peak, equity)
        dd = max(dd, (peak - equity) / peak)
    years = (last_ts - first_ts) / 86_400_000 / 365
    total = equity / INITIAL_EQUITY - 1.0
    ann = (1.0 + total) ** (1.0 / years) - 1.0 if total > -1 and years > 0 else 0.0
    return {"net_total_return": total, "net_annualized_return": ann, "net_max_drawdown": dd, "net_calmar": ann / dd if dd else 0.0}


def specs() -> list[StrategySpec]:
    return [
        StrategySpec("range_momentum", ETH_SYMBOL, ETH_SYMBOL, "range-momo-eth-l10-tp0.006-sl0.004", {"lookback": 10, "take": 0.006, "stop": 0.004, "max_hold": 9999}, 48, 0.002),
        StrategySpec("btc_lead_eth_lag", ETH_SYMBOL, BTC_SYMBOL, "btc-lead-eth-lag-lb6-br0.006-gap0.002-tp0.012-sl0.006", {"lead_lookback": 6, "btc_return_threshold": 0.006, "lag_gap": 0.002, "take": 0.012, "stop": 0.006, "max_hold": 12}, 48, 0.003),
        StrategySpec("short_bb_upper_rejection", ETH_SYMBOL, ETH_SYMBOL, "short-bb-upper-rejection-eth", {"trend": 96, "bb": 48, "atr": 48, "std": 1.5, "take": 0.010, "stop": 0.006, "max_hold": 12}, 48, 0.002),
    ]


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--bar", default=BAR)
    parser.add_argument("--years", type=float, default=YEARS)
    parser.add_argument("--output-dir", type=Path, default=OUTPUT_DIR)
    args = parser.parse_args()
    candles = _load_candles(ETH_SYMBOL, args.bar)
    candles = candles[-int(args.years * 365 * 24 * 60 / 15) :]

    rows: list[dict[str, object]] = []
    for spec in specs():
        for with_t in (False, True):
            result, stats = run_strategy(candles, spec, with_t)
            for cost_name, cost in COSTS:
                rows.append(
                    {
                        "family": spec.family,
                        "cost": cost_name,
                        "name": spec.name + ("-t" if with_t else "-base"),
                        "with_t": with_t,
                        "trades": result.trade_count,
                        "win_rate": result.win_rate,
                        "gross_total_return": result.total_return,
                        "gross_max_drawdown": result.max_drawdown,
                        **stats,
                        **cost_metrics(result, cost, candles[0].ts, candles[-1].ts),
                    }
                )
    output = pd.DataFrame(rows).sort_values(["cost", "net_calmar", "net_annualized_return"], ascending=[True, False, False])
    args.output_dir.mkdir(parents=True, exist_ok=True)
    path = args.output_dir / "other-strategies-t-overlay-summary.csv"
    output.to_csv(path, index=False)
    print(output[output.cost.eq(PRIMARY_COST)].to_string(index=False))
    return 0


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