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