okx-codex-trader/tests/test_run_crash_follow_short_observer.py

import math

from okx_codex_trader.models import Candle
from scripts.run_crash_follow_short_observer import frame_from_candles, latest_signal, load_pair_frame


def test_latest_signal_detects_crash_follow_short() -> None:
    eth = []
    btc = []
    for index in range(800):
        eth_price = 100.0 + math.sin(index / 9.0)
        btc_price = 100.0
        if index >= 760:
            btc_price = 96.0
        if index == 799:
            eth_price = 95.0
            btc_price = 94.0
        ts = 1_700_000_000_000 + (index * 3_600_000)
        high = eth_price + (10.0 if index == 799 else 2.0)
        low = eth_price - (10.0 if index == 799 else 2.0)
        eth.append(Candle("ETH-USDT-SWAP", ts, eth_price, high, low, eth_price, 1_000.0))
        btc.append(Candle("BTC-USDT-SWAP", ts, btc_price, btc_price, btc_price, btc_price, 1_000.0))
    eth_frame = frame_from_candles(eth)
    btc_frame = frame_from_candles(btc)[["ts", "close"]].rename(columns={"close": "btc_close"})
    frame = eth_frame.merge(btc_frame, on="ts", how="inner").sort_values("ts").reset_index(drop=True)

    signal = latest_signal(frame)

    assert signal["signal"] == "entry_short"
    assert signal["target_side"] == "short"
    assert signal["indicators"]["btc_riskoff_gate"] is True