okx-codex-trader/tests/test_donchian_report.py

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"])