okx-codex-trader/tests/test_bbsb_report.py

import pytest

from okx_codex_trader import bbsb_report
from okx_codex_trader.bbsb_report import generate_bbsb_sampled_report, run_bbsb_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 make_candle(index: int, open_price: float, high: float, low: float, close: float) -> Candle:
    return Candle(
        symbol="BTC-USDT-SWAP",
        ts=index * 60_000,
        open=open_price,
        high=high,
        low=low,
        close=close,
        volume=1_000.0 + index,
    )


def build_warmup() -> list[Candle]:
    candles: list[Candle] = []
    for index in range(bbsb_report.WARMUP_BARS):
        close = 100.5 if index % 2 else 99.5
        candles.append(make_candle(index, close, close + 0.2, close - 0.2, close))
    return candles


def build_long_breakout_fixture() -> list[Candle]:
    candles = build_warmup()
    base = len(candles)
    for offset in range(19):
        candles.append(make_candle(base + offset, 100.0, 100.02, 99.98, 100.0))
    candles.append(make_candle(base + 19, 100.0, 100.25, 99.98, 100.2))
    candles.append(make_candle(base + 20, 100.2, 100.3, 100.15, 100.25))
    candles.append(make_candle(base + 21, 100.25, 101.3, 100.2, 101.0))
    return candles


def build_short_breakout_fixture() -> list[Candle]:
    candles = build_warmup()
    base = len(candles)
    for offset in range(19):
        candles.append(make_candle(base + offset, 100.0, 100.02, 99.98, 100.0))
    candles.append(make_candle(base + 19, 100.0, 100.02, 99.75, 99.8))
    candles.append(make_candle(base + 20, 99.8, 99.85, 99.7, 99.75))
    candles.append(make_candle(base + 21, 99.75, 99.8, 98.7, 98.9))
    return candles


def build_ambiguous_exit_fixture() -> list[Candle]:
    candles = build_long_breakout_fixture()[:-1]
    base = len(candles)
    candles.append(make_candle(base, 100.25, 101.4, 99.6, 100.8))
    return candles


def build_final_bar_breakout_fixture() -> list[Candle]:
    candles = build_warmup()
    base = len(candles)
    candles.append(make_candle(base, 100.0, 100.02, 99.98, 100.0))
    candles.append(make_candle(base + 1, 100.0, 100.25, 99.98, 100.2))
    return candles


def build_open_tail_fixture() -> list[Candle]:
    candles = build_warmup()
    base = len(candles)
    for offset in range(19):
        candles.append(make_candle(base + offset, 100.0, 100.02, 99.98, 100.0))
    candles.append(make_candle(base + 19, 100.0, 100.25, 99.98, 100.2))
    candles.append(make_candle(base + 20, 100.2, 100.3, 100.15, 100.25))
    candles.append(make_candle(base + 21, 100.25, 100.9, 100.2, 100.6))
    return candles


def build_entry_bar_tp_sl_fixture() -> list[Candle]:
    candles = build_warmup()
    base = len(candles)
    for offset in range(19):
        candles.append(make_candle(base + offset, 100.0, 100.02, 99.98, 100.0))
    candles.append(make_candle(base + 19, 100.0, 100.25, 99.98, 100.2))
    candles.append(make_candle(base + 20, 100.2, 101.4, 99.6, 100.5))
    return candles


def build_same_bar_reentry_fixture() -> list[Candle]:
    candles = build_long_breakout_fixture()[:-1]
    base = len(candles)
    candles.append(make_candle(base, 100.25, 101.3, 100.2, 100.8))
    candles.append(make_candle(base + 1, 100.8, 101.0, 99.7, 99.8))
    return candles


def test_run_bbsb_segment_produces_long_breakout_trade():
    result = run_bbsb_segment(candles=build_long_breakout_fixture(), leverage=2, warmup_bars=bbsb_report.WARMUP_BARS)

    assert isinstance(result, SegmentResult)
    assert result.trade_count == 1
    assert result.trades[0]["side"] == "Long"


def test_run_bbsb_segment_produces_short_breakout_trade():
    result = run_bbsb_segment(candles=build_short_breakout_fixture(), leverage=2, warmup_bars=bbsb_report.WARMUP_BARS)

    assert isinstance(result, SegmentResult)
    assert result.trade_count == 1
    assert result.trades[0]["side"] == "Short"


def test_run_bbsb_segment_stop_loss_takes_precedence_over_take_profit():
    result = run_bbsb_segment(candles=build_ambiguous_exit_fixture(), leverage=2, warmup_bars=bbsb_report.WARMUP_BARS)

    assert result.trade_count == 1
    assert result.trades[0]["exit_price"] == pytest.approx(99.699)


def test_run_bbsb_segment_does_not_generate_entry_from_final_reported_candle():
    result = run_bbsb_segment(candles=build_final_bar_breakout_fixture(), leverage=2, warmup_bars=bbsb_report.WARMUP_BARS)

    assert result.trade_count == 0
    assert result.trades == []


def test_run_bbsb_segment_marks_open_position_to_market_but_keeps_journal_realized_only():
    result = run_bbsb_segment(candles=build_open_tail_fixture(), leverage=2, warmup_bars=bbsb_report.WARMUP_BARS)

    assert result.trade_count == 0
    assert result.trades == []
    assert result.total_return != 0
    assert result.open_position is not None


def test_run_bbsb_segment_allows_tp_or_sl_on_entry_candle():
    result = run_bbsb_segment(candles=build_entry_bar_tp_sl_fixture(), leverage=2, warmup_bars=bbsb_report.WARMUP_BARS)

    assert result.trade_count == 1
    assert result.open_position is None


def test_run_bbsb_segment_exit_exhausts_bar_without_same_bar_reentry():
    result = run_bbsb_segment(candles=build_same_bar_reentry_fixture(), leverage=2, warmup_bars=bbsb_report.WARMUP_BARS)

    assert result.trade_count == 1
    assert len(result.entries) == 1


def test_generate_bbsb_sampled_report_forwards_shared_report_arguments(monkeypatch, tmp_path):
    expected = {"report_file": str(tmp_path / "bbsb.html"), "segment_count": 2}
    captured: dict[str, object] = {}

    def fake_generate_sampled_report(**kwargs):
        captured.update(kwargs)
        return expected

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

    result = generate_bbsb_sampled_report(
        candles=build_candles_from_closes([100.0 + index for index in range(5_000)]),
        leverage=2,
        output_file=tmp_path / "bbsb.html",
        symbol="BTC-USDT-SWAP",
        bar="3m",
        segments=2,
        window_size=300,
    )

    assert result == expected
    assert captured["report_title"] == "BBSB Sampled Report"
    assert captured["strategy_label"] == "BBSB"
    assert captured["run_segment"] is run_bbsb_segment
    assert captured["strategy_params"] == {
        "band_length": 20,
        "std_multiplier": 2.0,
        "bandwidth_lookback": 50,
        "take_profit_pct": 0.01,
        "stop_loss_pct": 0.005,
    }