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, }, }