okx-codex-trader/tests/test_bbmr_report.py

import pytest

from okx_codex_trader import bbmr_report
from okx_codex_trader.bbmr_report import (
    BBMRConfig,
    generate_bbmr_sampled_report,
    run_bbmr_segment,
)
from okx_codex_trader.models import Candle
from okx_codex_trader.sampled_report import SegmentResult


def build_candles_from_closes(closes: list[float]) -> list[Candle]:
    candles: list[Candle] = []
    for index, close in enumerate(closes):
        candles.append(
            Candle(
                symbol="BTC-USDT-SWAP",
                ts=index * 60_000,
                open=close,
                high=close + 1.0,
                low=close - 1.0,
                close=close,
                volume=1_000.0 + index,
            )
        )
    return candles


def build_linear_candles(count: int) -> list[Candle]:
    return build_candles_from_closes([100.0 + index for index in range(count)])


def test_run_bbmr_segment_can_enter_on_final_bar_from_prior_signal():
    config = BBMRConfig(band_length=2, std_multiplier=0.1, bandwidth_lookback=1, stop_loss_pct=0.005)
    candles = build_candles_from_closes([100.0, 110.0, 106.0, 96.0])

    result = run_bbmr_segment(candles=candles, leverage=2, warmup_bars=2, config=config)

    assert isinstance(result, SegmentResult)
    assert result.trade_count == 1
    assert result.trades[0]["exit_price"] == pytest.approx(95.52)
    assert result.entries == [{"ts": 180_000, "price": 96.0, "side": "long"}]
    assert result.open_position is None


def test_run_bbmr_segment_stop_loss_takes_precedence_and_no_reverse_entry():
    config = BBMRConfig(band_length=2, std_multiplier=0.1, bandwidth_lookback=1, stop_loss_pct=0.01)
    candles = [
        Candle(symbol="BTC-USDT-SWAP", ts=0, open=100.0, high=101.0, low=99.0, close=100.0, volume=1000.0),
        Candle(symbol="BTC-USDT-SWAP", ts=60_000, open=110.0, high=111.0, low=109.0, close=110.0, volume=1001.0),
        Candle(symbol="BTC-USDT-SWAP", ts=120_000, open=104.0, high=105.0, low=103.0, close=104.0, volume=1002.0),
        Candle(symbol="BTC-USDT-SWAP", ts=180_000, open=105.0, high=105.2, low=103.7, close=103.8, volume=1003.0),
        Candle(symbol="BTC-USDT-SWAP", ts=240_000, open=106.0, high=109.0, low=103.5, close=108.0, volume=1004.0),
        Candle(symbol="BTC-USDT-SWAP", ts=300_000, open=107.0, high=108.0, low=106.0, close=107.0, volume=1005.0),
    ]

    result = run_bbmr_segment(candles=candles, leverage=2, warmup_bars=2, config=config)

    assert result.trade_count == 1
    assert len(result.trades) == 1
    assert result.trades[0]["side"] == "Long"
    assert result.trades[0]["exit_price"] == pytest.approx(103.95)
    assert result.open_position is None


def test_run_bbmr_segment_marks_open_position_to_market_but_keeps_journal_realized_only():
    config = BBMRConfig(band_length=2, std_multiplier=0.1, bandwidth_lookback=1, stop_loss_pct=0.2)
    candles = build_candles_from_closes([100.0, 110.0, 104.0, 103.0, 102.0])

    result = run_bbmr_segment(candles=candles, leverage=2, warmup_bars=2, config=config)

    assert result.trade_count == 0
    assert result.trades == []
    assert isinstance(result, SegmentResult)
    assert result.total_return == pytest.approx((9805.825242718447 - 10_000.0) / 10_000.0)
    assert result.open_position is not None


def test_generate_bbmr_sampled_report_rejects_insufficient_history_pool(tmp_path):
    candles = build_linear_candles(1_000)

    with pytest.raises(ValueError, match="history pool is too small"):
        generate_bbmr_sampled_report(
            candles=candles,
            leverage=2,
            output_file=tmp_path / "bbmr.html",
            symbol="BTC-USDT-SWAP",
            bar="3m",
            segments=8,
            window_size=300,
        )


def test_generate_bbmr_sampled_report_uses_shared_shell(monkeypatch, tmp_path):
    candles = build_linear_candles(500)
    output_file = tmp_path / "bbmr.html"
    recorded: dict[str, object] = {}
    sentinel = {"report_file": str(output_file), "segment_count": 2, "window_size": 50, "aggregate_trade_count": 3, "average_return": 0.25}

    def fake_generate_sampled_report(**kwargs):
        recorded.update(kwargs)
        return sentinel

    monkeypatch.setattr(bbmr_report, "generate_sampled_report", fake_generate_sampled_report)

    result = generate_bbmr_sampled_report(
        candles=candles,
        leverage=3,
        output_file=output_file,
        symbol="BTC-USDT-SWAP",
        bar="3m",
        segments=2,
        window_size=50,
    )

    assert result == sentinel
    assert recorded["candles"] == candles
    assert recorded["leverage"] == 3
    assert recorded["output_file"] == output_file
    assert recorded["symbol"] == "BTC-USDT-SWAP"
    assert recorded["bar"] == "3m"
    assert recorded["segments"] == 2
    assert recorded["window_size"] == 50
    assert recorded["report_title"] == "BBMR Sampled Report"
    assert recorded["strategy_label"] == "BBMR"
    assert recorded["strategy_description"] == (
        "Bollinger Band mean reversion, bandwidth filter against previous 50 completed values, "
        "close-based return-to-middle exits at next open, intrabar 0.5% stop-loss."
    )
    assert recorded["strategy_params"] == {
        "band_length": 20,
        "std_multiplier": 2.0,
        "bandwidth_lookback": 50,
        "stop_loss_pct": 0.005,
    }
    assert recorded["run_segment"] is run_bbmr_segment