okx-codex-trader/tests/test_backtest.py

from okx_codex_trader.backtest import run_backtest
from okx_codex_trader.models import Candle
from okx_codex_trader.strategy import simple_moving_average


def build_crossing_series() -> list[Candle]:
    closes = [
        60.0,
        60.0,
        60.0,
        60.0,
        60.0,
        60.0,
        60.0,
        60.0,
        60.0,
        60.0,
        60.0,
        60.0,
        60.0,
        60.0,
        60.0,
        60.0,
        60.0,
        60.0,
        60.0,
        60.0,
        140.0,
        50.0,
        50.0,
        50.0,
        50.0,
        50.0,
        50.0,
        50.0,
        50.0,
        50.0,
        60.0,
        60.0,
        60.0,
        60.0,
        60.0,
        60.0,
        60.0,
        60.0,
        60.0,
        60.0,
        60.0,
        60.0,
        60.0,
        60.0,
        60.0,
    ]
    opens = list(closes)
    opens[21] = 100.0
    opens[30] = 90.0
    opens[40] = 80.0

    candles = []
    for index, (open_price, close_price) in enumerate(zip(opens, closes)):
        high = max(open_price, close_price)
        low = min(open_price, close_price)
        candles.append(
            Candle(
                symbol="BTC-USDT-SWAP",
                ts=index,
                open=open_price,
                high=high,
                low=low,
                close=close_price,
                volume=1_000.0,
            )
        )
    return candles


def build_open_position_series() -> list[Candle]:
    candles = build_crossing_series()[:29]
    return candles


def build_tail_drawdown_series() -> list[Candle]:
    candles = build_crossing_series()[:22]
    return candles


def build_drawdown_series() -> list[Candle]:
    closes = [60.0] * 20 + [120.0, 75.0, 100.0] + [75.0] * 14
    opens = list(closes)
    opens[21] = 100.0
    opens[36] = 100.0

    candles = []
    for index, (open_price, close_price) in enumerate(zip(opens, closes)):
        high = max(open_price, close_price)
        low = min(open_price, close_price)
        candles.append(
            Candle(
                symbol="BTC-USDT-SWAP",
                ts=index,
                open=open_price,
                high=high,
                low=low,
                close=close_price,
                volume=1_000.0,
            )
        )
    return candles


def test_simple_moving_average_requires_full_window():
    candles = [
        Candle(symbol="BTC-USDT-SWAP", ts=index, open=close, high=close, low=close, close=close, volume=1_000.0)
        for index, close in enumerate([10.0, 20.0, 30.0, 40.0])
    ]

    assert simple_moving_average(candles, 3) == [None, None, 20.0, 30.0]


def test_backtest_rejects_invalid_leverage():
    candles = build_crossing_series()

    for leverage in (0, 4):
        try:
            run_backtest(candles=candles, leverage=leverage)
        except ValueError as exc:
            assert str(exc) == "leverage is invalid"
        else:
            raise AssertionError("expected ValueError")


def test_backtest_runs_fixed_sma_crossover_series():
    candles = build_crossing_series()

    result = run_backtest(candles=candles, leverage=2)

    assert result.initial_equity == 10_000
    assert result.trade_count == 2
    assert result.trades[0].entry_price == candles[21].open
    assert result.trades[0].exit_price == candles[30].open
    assert result.trades[0].margin_used == 10_000
    assert result.trades[1].margin_used == result.trades[0].ending_equity
    assert result.ending_equity == 4_888.888888888889
    assert result.total_return == -0.5111111111111112
    assert "total_return" in result.to_dict()
    assert "max_drawdown" in result.to_dict()
    assert result.win_rate == 0.5


def test_backtest_does_not_force_close_open_position_at_series_end():
    candles = build_open_position_series()

    result = run_backtest(candles=candles, leverage=2)

    assert result.trade_count == 0
    assert result.trades == []
    assert result.ending_equity == 0.0
    assert result.total_return == -1.0


def test_backtest_tracks_open_trade_drawdown_from_candle_close():
    candles = build_drawdown_series()

    result = run_backtest(candles=candles, leverage=2)

    assert result.trade_count == 1
    assert result.trades[0].entry_price == candles[21].open
    assert result.trades[0].exit_price == candles[36].open
    assert result.max_drawdown == 0.5


def test_backtest_tracks_tail_drawdown_for_final_open_position():
    candles = build_tail_drawdown_series()

    result = run_backtest(candles=candles, leverage=2)

    assert result.trade_count == 0
    assert result.trades == []
    assert result.ending_equity == 0.0
    assert result.total_return == -1.0
    assert result.max_drawdown == 1.0