okx-codex-trader/okx_codex_trader/bb_squeeze_strategy.py

from __future__ import annotations

from dataclasses import dataclass

import pandas as pd

from okx_codex_trader.time_rules import entry_allowed, is_us_open_window


BAND_LENGTH = 96
BANDWIDTH_LOOKBACK = 960
BANDWIDTH_QUANTILE = 0.25
STOP_LOSS_PCT = 0.01
TAKE_PROFIT_PCT = 0.03
MIDDLE_EXIT_BUFFER_PCT = 0.001
MIDDLE_EXIT_CONFIRM_BARS = 1
ETH_VOL_CAP = 0.006
COOLDOWN_BARS = 24
BTC_TREND_SMA = 480
BTC_MOMENTUM_LOOKBACK = 96
ENTRY_TIME_FILTER = "weekday"
US_OPEN_EXIT_MODE = "skip"
BREAKEVEN_TRIGGER_PCT = 0.008
BREAKEVEN_LOCK_PCT = 0.001


@dataclass(frozen=True)
class StrategyState:
    last_candle_ts: int | None
    active_side: str | None
    entry_price: float | None
    entry_candle_ts: int | None
    cooldown_until_ts: int | None
    middle_exit_streak: int
    max_favorable_move_pct: float | None


EMPTY_STATE = StrategyState(None, None, None, None, None, 0, None)


def strategy_name() -> str:
    return (
        f"bb-rr-time-l{BAND_LENGTH}-bw{BANDWIDTH_LOOKBACK}"
        f"-q{BANDWIDTH_QUANTILE:g}-sl{STOP_LOSS_PCT:g}-rr{TAKE_PROFIT_PCT / STOP_LOSS_PCT:g}"
        f"-hybrid_signal_rr-both-btc-up-vc{ETH_VOL_CAP:g}-ddnone"
        f"-cd{COOLDOWN_BARS}-mxbuf{MIDDLE_EXIT_BUFFER_PCT:g}-mxc{MIDDLE_EXIT_CONFIRM_BARS}"
        f"-entry{ENTRY_TIME_FILTER}-openexit{US_OPEN_EXIT_MODE}"
        f"-be{BREAKEVEN_TRIGGER_PCT:g}-{BREAKEVEN_LOCK_PCT:g}"
    )


def signal_from_frame(frame: pd.DataFrame, state: StrategyState) -> tuple[StrategyState, dict[str, object]]:
    if len(frame) < BANDWIDTH_LOOKBACK + 2 or "btc_close" not in frame.columns:
        raise ValueError("not enough candles")
    close = frame["close"].astype(float)
    btc_close = frame["btc_close"].astype(float)
    middle = close.rolling(BAND_LENGTH).mean()
    stdev = close.rolling(BAND_LENGTH).std(ddof=0)
    upper = middle + (2.0 * stdev)
    lower = middle - (2.0 * stdev)
    bandwidth = (upper - lower) / middle
    threshold = bandwidth.rolling(BANDWIDTH_LOOKBACK).quantile(BANDWIDTH_QUANTILE)
    eth_vol = close.pct_change().rolling(96).std(ddof=0)
    btc_sma = btc_close.rolling(BTC_TREND_SMA).mean()
    btc_momentum = btc_close / btc_close.shift(BTC_MOMENTUM_LOOKBACK) - 1.0
    decision_index = len(frame) - 1
    row = frame.iloc[decision_index]
    candle_ts = int(row["ts"])
    candle_time = pd.Timestamp(row["time"]).isoformat().replace("+00:00", "Z")

    indicators = {
        "eth_close": float(row["close"]),
        "middle": float(middle.iloc[decision_index]),
        "upper": float(upper.iloc[decision_index]),
        "lower": float(lower.iloc[decision_index]),
        "bandwidth": float(bandwidth.iloc[decision_index]),
        "bandwidth_threshold": float(threshold.iloc[decision_index]),
        "eth_vol_96": float(eth_vol.iloc[decision_index]),
        "btc_close": float(btc_close.iloc[decision_index]),
        "btc_sma_480": float(btc_sma.iloc[decision_index]),
        "btc_momentum_96": float(btc_momentum.iloc[decision_index]),
    }

    if state.last_candle_ts is not None and candle_ts <= state.last_candle_ts:
        return state, {
            "decision_candle_ts": candle_ts,
            "decision_candle_time": candle_time,
            "signal": "state_replay",
            "target_side": state.active_side or "flat",
            "indicators": indicators,
        }

    next_state = StrategyState(
        candle_ts,
        state.active_side,
        state.entry_price,
        state.entry_candle_ts,
        state.cooldown_until_ts,
        state.middle_exit_streak,
        state.max_favorable_move_pct,
    )
    signal = "hold"
    target_side = state.active_side or "flat"

    if state.active_side is not None:
        entry_price = float(state.entry_price)
        stop = entry_price * (1.0 - STOP_LOSS_PCT if state.active_side == "long" else 1.0 + STOP_LOSS_PCT)
        if state.max_favorable_move_pct is not None and state.max_favorable_move_pct >= BREAKEVEN_TRIGGER_PCT:
            breakeven_stop = entry_price * (1.0 + BREAKEVEN_LOCK_PCT if state.active_side == "long" else 1.0 - BREAKEVEN_LOCK_PCT)
            stop = max(stop, breakeven_stop) if state.active_side == "long" else min(stop, breakeven_stop)
        take = entry_price * (1.0 + TAKE_PROFIT_PCT if state.active_side == "long" else 1.0 - TAKE_PROFIT_PCT)
        stop_hit = (state.active_side == "long" and float(row["low"]) <= stop) or (state.active_side == "short" and float(row["high"]) >= stop)
        take_hit = (state.active_side == "long" and float(row["high"]) >= take) or (state.active_side == "short" and float(row["low"]) <= take)
        middle_exit = (state.active_side == "long" and float(row["close"]) < indicators["middle"] * (1.0 - MIDDLE_EXIT_BUFFER_PCT)) or (
            state.active_side == "short" and float(row["close"]) > indicators["middle"] * (1.0 + MIDDLE_EXIT_BUFFER_PCT)
        )
        if middle_exit and is_us_open_window(candle_ts) and US_OPEN_EXIT_MODE == "skip":
            middle_exit = False
        middle_exit_streak = state.middle_exit_streak + 1 if middle_exit else 0
        favorable_move = (float(row["high"]) / entry_price - 1.0) if state.active_side == "long" else (entry_price / float(row["low"]) - 1.0)
        max_favorable_move_pct = max(float(state.max_favorable_move_pct or 0.0), favorable_move)
        next_state = StrategyState(
            candle_ts,
            state.active_side,
            state.entry_price,
            state.entry_candle_ts,
            state.cooldown_until_ts,
            middle_exit_streak,
            max_favorable_move_pct,
        )
        if stop_hit or take_hit or middle_exit_streak >= MIDDLE_EXIT_CONFIRM_BARS:
            signal = (
                "exit_breakeven"
                if stop_hit and state.max_favorable_move_pct is not None and state.max_favorable_move_pct >= BREAKEVEN_TRIGGER_PCT
                else "exit_stop"
                if stop_hit
                else "exit_take_profit"
                if take_hit
                else "exit_middle"
            )
            target_side = "flat"
            next_state = StrategyState(
                candle_ts,
                None,
                None,
                None,
                candle_ts + (COOLDOWN_BARS * 900_000),
                0,
                None,
            )
    else:
        cooldown_ok = state.cooldown_until_ts is None or candle_ts >= state.cooldown_until_ts
        compressed = indicators["bandwidth"] <= indicators["bandwidth_threshold"]
        vol_ok = indicators["eth_vol_96"] <= ETH_VOL_CAP
        btc_up = indicators["btc_close"] > indicators["btc_sma_480"]
        time_ok = entry_allowed(candle_ts, ENTRY_TIME_FILTER)
        if cooldown_ok and compressed and vol_ok and btc_up and time_ok and float(row["close"]) > indicators["upper"]:
            signal = "entry_long"
            target_side = "long"
            next_state = StrategyState(candle_ts, "long", float(row["close"]), candle_ts, state.cooldown_until_ts, 0, 0.0)
        elif cooldown_ok and compressed and vol_ok and btc_up and time_ok and float(row["close"]) < indicators["lower"]:
            signal = "entry_short"
            target_side = "short"
            next_state = StrategyState(candle_ts, "short", float(row["close"]), candle_ts, state.cooldown_until_ts, 0, 0.0)

    return next_state, {
        "decision_candle_ts": candle_ts,
        "decision_candle_time": candle_time,
        "signal": signal,
        "target_side": target_side,
        "indicators": indicators,
        "params": {
            "band_length": BAND_LENGTH,
            "bandwidth_lookback": BANDWIDTH_LOOKBACK,
            "bandwidth_quantile": BANDWIDTH_QUANTILE,
            "stop_loss_pct": STOP_LOSS_PCT,
            "take_profit_pct": TAKE_PROFIT_PCT,
            "middle_exit_buffer_pct": MIDDLE_EXIT_BUFFER_PCT,
            "middle_exit_confirm_bars": MIDDLE_EXIT_CONFIRM_BARS,
            "eth_vol_cap": ETH_VOL_CAP,
            "cooldown_bars": COOLDOWN_BARS,
            "side_mode": "both",
            "btc_filter": "btc-up",
            "entry_time_filter": ENTRY_TIME_FILTER,
            "us_open_exit_mode": US_OPEN_EXIT_MODE,
            "breakeven_trigger_pct": BREAKEVEN_TRIGGER_PCT,
            "breakeven_lock_pct": BREAKEVEN_LOCK_PCT,
        },
    }