import pytest from okx_codex_trader import donchian_report from okx_codex_trader.donchian_report import ( DonchianConfig, generate_donchian_sampled_report, run_donchian_segment, ) from okx_codex_trader.models import Candle from okx_codex_trader.sampled_report import SegmentResult 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_linear_candles(count: int) -> list[Candle]: candles: list[Candle] = [] for index in range(count): price = 100.0 + index candles.append(make_candle(index, price, price + 1.0, price - 1.0, price)) return candles def build_long_trade_fixture() -> list[Candle]: return [ make_candle(0, 100.0, 101.0, 99.0, 100.0), make_candle(1, 100.0, 101.0, 99.0, 100.0), make_candle(2, 100.0, 101.0, 99.0, 100.0), make_candle(3, 101.0, 102.5, 100.5, 102.0), make_candle(4, 103.0, 103.5, 102.0, 103.0), make_candle(5, 102.5, 102.8, 99.8, 100.0), make_candle(6, 99.9, 100.2, 99.2, 99.7), ] def build_short_trade_fixture() -> list[Candle]: return [ make_candle(0, 100.0, 101.0, 99.0, 100.0), make_candle(1, 100.0, 101.0, 99.0, 100.0), make_candle(2, 100.0, 101.0, 99.0, 100.0), make_candle(3, 99.0, 99.5, 97.5, 98.0), make_candle(4, 97.0, 97.5, 96.5, 97.0), make_candle(5, 97.2, 100.5, 96.8, 100.0), make_candle(6, 100.1, 100.6, 99.5, 100.3), ] def build_stop_loss_fixture() -> list[Candle]: return [ make_candle(0, 100.0, 101.0, 99.0, 100.0), make_candle(1, 100.0, 101.0, 99.0, 100.0), make_candle(2, 100.0, 101.0, 99.0, 100.0), make_candle(3, 101.0, 102.5, 100.5, 102.0), make_candle(4, 103.0, 103.5, 102.5, 103.0), make_candle(5, 103.0, 103.2, 99.8, 100.0), ] def build_entry_bar_stop_loss_fixture() -> list[Candle]: return [ make_candle(0, 100.0, 101.0, 99.0, 100.0), make_candle(1, 100.0, 101.0, 99.0, 100.0), make_candle(2, 100.0, 101.0, 99.0, 100.0), make_candle(3, 101.0, 102.5, 100.5, 102.0), make_candle(4, 103.0, 103.5, 101.0, 102.5), ] def build_gap_through_stop_fixture() -> list[Candle]: return [ make_candle(0, 100.0, 101.0, 99.0, 100.0), make_candle(1, 100.0, 101.0, 99.0, 100.0), make_candle(2, 100.0, 101.0, 99.0, 100.0), make_candle(3, 101.0, 102.5, 100.5, 102.0), make_candle(4, 103.0, 103.5, 102.5, 103.0), make_candle(5, 101.0, 101.2, 99.8, 100.0), ] def build_final_bar_breakout_fixture() -> list[Candle]: return [ make_candle(0, 100.0, 101.0, 99.0, 100.0), make_candle(1, 100.0, 101.0, 99.0, 100.0), make_candle(2, 100.0, 101.0, 99.0, 100.0), make_candle(3, 100.0, 101.0, 99.0, 100.0), make_candle(4, 101.0, 102.5, 100.5, 102.0), ] def build_open_tail_fixture() -> list[Candle]: return [ make_candle(0, 100.0, 101.0, 99.0, 100.0), make_candle(1, 100.0, 101.0, 99.0, 100.0), make_candle(2, 100.0, 101.0, 99.0, 100.0), make_candle(3, 101.0, 102.5, 100.5, 102.0), make_candle(4, 103.0, 104.0, 102.5, 104.0), make_candle(5, 104.0, 105.5, 103.5, 105.0), ] def test_run_donchian_segment_produces_long_trade(): result = run_donchian_segment( candles=build_long_trade_fixture(), leverage=2, warmup_bars=3, config=DonchianConfig(entry_window=3, exit_window=2, stop_loss_pct=0.2), ) assert isinstance(result, SegmentResult) assert result.trade_count == 1 assert result.trades[0]["side"] == "Long" assert result.open_position is None def test_run_donchian_segment_produces_short_trade(): result = run_donchian_segment( candles=build_short_trade_fixture(), leverage=2, warmup_bars=3, config=DonchianConfig(entry_window=3, exit_window=2, stop_loss_pct=0.2), ) assert isinstance(result, SegmentResult) assert result.trade_count == 1 assert result.trades[0]["side"] == "Short" assert result.open_position is None def test_run_donchian_segment_stop_loss_takes_precedence(): result = run_donchian_segment( candles=build_stop_loss_fixture(), leverage=2, warmup_bars=3, config=DonchianConfig(entry_window=3, exit_window=2, stop_loss_pct=0.01), ) assert result.trade_count == 1 assert result.trades[0]["side"] == "Long" assert result.trades[0]["exit_price"] == pytest.approx(101.97) def test_run_donchian_segment_allows_entry_bar_stop_loss(): result = run_donchian_segment( candles=build_entry_bar_stop_loss_fixture(), leverage=2, warmup_bars=3, config=DonchianConfig(entry_window=3, exit_window=2, stop_loss_pct=0.01), ) assert result.trade_count == 1 assert result.open_position is None assert result.trades[0]["exit_price"] == pytest.approx(101.97) def test_run_donchian_segment_exits_gap_through_stop_at_open(): result = run_donchian_segment( candles=build_gap_through_stop_fixture(), leverage=2, warmup_bars=3, config=DonchianConfig(entry_window=3, exit_window=2, stop_loss_pct=0.01), ) assert result.trade_count == 1 assert result.trades[0]["exit_price"] == pytest.approx(101.0) def test_run_donchian_segment_does_not_generate_entry_from_final_bar(): result = run_donchian_segment( candles=build_final_bar_breakout_fixture(), leverage=2, warmup_bars=3, config=DonchianConfig(entry_window=3, exit_window=2, stop_loss_pct=0.01), ) assert result.trade_count == 0 assert result.entries == [] assert result.open_position is None def test_run_donchian_segment_marks_open_position_to_market(): result = run_donchian_segment( candles=build_open_tail_fixture(), leverage=2, warmup_bars=3, config=DonchianConfig(entry_window=3, exit_window=2, stop_loss_pct=0.2), ) assert result.trade_count == 0 assert result.trades == [] assert result.total_return == pytest.approx((10_388.349514563106 - 10_000.0) / 10_000.0) assert result.open_position is not None def test_generate_donchian_sampled_report_uses_shared_shell_defaults(monkeypatch, tmp_path): candles = build_linear_candles(5_000) output_file = tmp_path / "donchian.html" recorded: dict[str, object] = {} sentinel = { "report_file": str(output_file), "segment_count": 2, "window_size": 300, "aggregate_trade_count": 4, "average_return": 0.12, } def fake_generate_sampled_report(**kwargs): recorded.update(kwargs) return sentinel monkeypatch.setattr(donchian_report, "generate_sampled_report", fake_generate_sampled_report) result = generate_donchian_sampled_report( candles=candles, leverage=2, output_file=output_file, symbol="BTC-USDT-SWAP", bar="3m", segments=2, window_size=300, ) assert result == sentinel assert recorded["report_title"] == "Donchian Sampled Report" assert recorded["strategy_label"] == "Donchian" assert recorded["strategy_params"] == { "entry_window": 20, "exit_window": 10, "stop_loss_pct": 0.01, } assert recorded["warmup_bars"] == 20 assert callable(recorded["run_segment"])