okx-codex-trader/tests/test_cli.py

import json
from dataclasses import asdict
from pathlib import Path

import pytest

from okx_codex_trader.backtest import run_backtest
from okx_codex_trader.cli import main_factory
from okx_codex_trader.config import Config
from okx_codex_trader.models import Candle, TradeSignal


def sample_config() -> Config:
    return Config(api_key="key", api_secret="secret", api_passphrase="passphrase")


def sample_candles(limit: int = 60, symbol: str = "BTC-USDT-SWAP") -> list[Candle]:
    candles = []
    for index in range(limit):
        price = 100.0 + index
        candles.append(
            Candle(
                symbol=symbol,
                ts=index,
                open=price,
                high=price + 1.0,
                low=price - 1.0,
                close=price + 0.5,
                volume=1_000.0 + index,
            )
        )
    return candles


def valid_signal() -> dict[str, object]:
    return {
        "action": "long",
        "confidence": 0.8,
        "leverage": 2,
        "entry_price": 123.5,
        "take_profit_price": 130.0,
        "stop_loss_price": 119.0,
        "reason": "trend",
    }


def fake_analyze_with_codex(candles: list[Candle], symbol: str, bar: str) -> TradeSignal:
    assert candles
    assert symbol == "BTC-USDT-SWAP"
    assert bar == "1H"
    return TradeSignal(**valid_signal())


def real_write_text(path: str, text: str) -> None:
    Path(path).write_text(text)


class FakeClient:
    def __init__(self):
        self.get_candles_called_with: tuple[str, str, int] | None = None
        self.get_last_price_called_with: str | None = None

    def get_candles(self, symbol: str, bar: str, limit: int) -> list[Candle]:
        self.get_candles_called_with = (symbol, bar, limit)
        return sample_candles(limit=limit, symbol=symbol)

    def get_last_price(self, symbol: str) -> float:
        self.get_last_price_called_with = symbol
        return 250.0


def fake_client() -> FakeClient:
    return FakeClient()


def build_main_with_stubs(*, state_path: Path | None = None):
    client = fake_client()
    report_calls: list[dict[str, object]] = []
    bbmr_report_calls: list[dict[str, object]] = []
    bbsb_report_calls: list[dict[str, object]] = []
    donchian_report_calls: list[dict[str, object]] = []

    def fake_report(*, candles, leverage, output_file, symbol, bar):
        report_calls.append(
            {
                "candles": candles,
                "leverage": leverage,
                "output_file": output_file,
                "symbol": symbol,
                "bar": bar,
            }
        )
        return {
            "report_file": str(output_file),
            "plot_file": str(output_file).replace(".html", ".plot.html"),
            "trade_count": 3,
            "total_return": 0.12,
        }

    def fake_bbmr_report(*, candles, leverage, output_file, symbol, bar, segments, window_size):
        bbmr_report_calls.append(
            {
                "candles": candles,
                "leverage": leverage,
                "output_file": output_file,
                "symbol": symbol,
                "bar": bar,
                "segments": segments,
                "window_size": window_size,
            }
        )
        return {
            "report_file": str(output_file),
            "segment_count": segments,
            "window_size": window_size,
            "aggregate_trade_count": 11,
            "average_return": 0.031,
        }

    def fake_bbsb_report(*, candles, leverage, output_file, symbol, bar, segments, window_size):
        bbsb_report_calls.append(
            {
                "candles": candles,
                "leverage": leverage,
                "output_file": output_file,
                "symbol": symbol,
                "bar": bar,
                "segments": segments,
                "window_size": window_size,
            }
        )
        return {
            "report_file": str(output_file),
            "segment_count": segments,
            "window_size": window_size,
            "aggregate_trade_count": 11,
            "average_return": 0.031,
        }

    def fake_donchian_report(
        *,
        candles,
        leverage,
        output_file,
        symbol,
        bar,
        segments,
        window_size,
        entry_window,
        exit_window,
        stop_loss_pct,
    ):
        donchian_report_calls.append(
            {
                "candles": candles,
                "leverage": leverage,
                "output_file": output_file,
                "symbol": symbol,
                "bar": bar,
                "segments": segments,
                "window_size": window_size,
                "entry_window": entry_window,
                "exit_window": exit_window,
                "stop_loss_pct": stop_loss_pct,
            }
        )
        return {
            "report_file": str(output_file),
            "segment_count": segments,
            "window_size": window_size,
            "aggregate_trade_count": 7,
            "average_return": 0.024,
        }

    main = main_factory(
        load_config=lambda: sample_config(),
        client_factory=lambda: client,
        analyze_fn=fake_analyze_with_codex,
        write_text=real_write_text,
        state_path=Path("paper_state.json") if state_path is None else state_path,
        now_fn=lambda: "1970-01-01T00:00:00Z",
        report_fn=fake_report,
        bbmr_report_fn=fake_bbmr_report,
        bbsb_report_fn=fake_bbsb_report,
        donchian_report_fn=fake_donchian_report,
        ema_pullback_report_fn=lambda **kwargs: {},
    )
    return main, client, report_calls, bbmr_report_calls, bbsb_report_calls, donchian_report_calls


def test_fetch_history_prints_candle_json(capsys):
    main, client, _, _, _, _ = build_main_with_stubs()
    expected = [asdict(candle) for candle in sample_candles(limit=20)]

    exit_code = main(["fetch-history", "--symbol", "BTC-USDT-SWAP", "--bar", "1H", "--limit", "20"])

    assert exit_code == 0
    assert client.get_candles_called_with == ("BTC-USDT-SWAP", "1H", 20)
    assert json.loads(capsys.readouterr().out) == expected


def test_backtest_prints_summary_json(capsys):
    main, client, _, _, _, _ = build_main_with_stubs()
    expected = run_backtest(candles=sample_candles(limit=50), leverage=2).to_dict()

    exit_code = main(["backtest", "--symbol", "BTC-USDT-SWAP", "--bar", "1H", "--limit", "50", "--leverage", "2"])

    assert exit_code == 0
    assert client.get_candles_called_with == ("BTC-USDT-SWAP", "1H", 50)
    assert json.loads(capsys.readouterr().out) == expected


def test_analyze_writes_output_file_and_stdout(tmp_path, capsys):
    main, client, _, _, _, _ = build_main_with_stubs()
    output_file = tmp_path / "signal.json"

    exit_code = main(
        [
            "analyze",
            "--symbol",
            "BTC-USDT-SWAP",
            "--bar",
            "1H",
            "--limit",
            "20",
            "--output-file",
            str(output_file),
        ]
    )

    assert exit_code == 0
    assert output_file.exists()
    assert client.get_candles_called_with == ("BTC-USDT-SWAP", "1H", 20)
    stdout = capsys.readouterr().out.strip()
    file_text = output_file.read_text()
    assert stdout == file_text
    assert json.loads(stdout) == valid_signal()


def test_paper_order_initializes_local_state_and_outputs_local_order_json(tmp_path, capsys):
    state_path = tmp_path / "paper_state.json"
    main, client, _, _, _, _ = build_main_with_stubs(state_path=state_path)
    signal_file = tmp_path / "signal.json"
    signal_file.write_text(json.dumps(valid_signal()))

    exit_code = main(
        [
            "paper-order",
            "--symbol",
            "BTC-USDT-SWAP",
            "--signal-file",
            str(signal_file),
            "--margin-usdt",
            "100",
        ]
    )

    assert exit_code == 0
    assert client.get_last_price_called_with is None
    payload = json.loads(capsys.readouterr().out)
    assert payload == {
        "status": "filled",
        "symbol": "BTC-USDT-SWAP",
        "side": "long",
        "price": 123.5,
        "quantity": pytest.approx((100.0 * 2) / 123.5),
        "margin_used": 100.0,
        "cash_usdt": 9900.0,
    }
    state = json.loads(state_path.read_text())
    assert state["cash_usdt"] == 9900.0
    assert state["realized_pnl"] == 0.0
    assert state["updated_at"] == "1970-01-01T00:00:00Z"
    assert len(state["positions"]) == 1
    assert state["positions"][0]["symbol"] == "BTC-USDT-SWAP"
    assert state["positions"][0]["side"] == "long"
    assert state["positions"][0]["quantity"] == pytest.approx((100.0 * 2) / 123.5)
    assert state["positions"][0]["avg_entry_price"] == 123.5
    assert state["positions"][0]["margin_used"] == 100.0


def test_paper_order_uses_latest_price_when_entry_price_is_null(tmp_path, capsys):
    state_path = tmp_path / "paper_state.json"
    main, client, _, _, _, _ = build_main_with_stubs(state_path=state_path)
    signal_file = tmp_path / "signal.json"
    payload = valid_signal()
    payload["entry_price"] = None
    signal_file.write_text(json.dumps(payload))

    exit_code = main(
        [
            "paper-order",
            "--symbol",
            "BTC-USDT-SWAP",
            "--signal-file",
            str(signal_file),
            "--margin-usdt",
            "100",
        ]
    )

    assert exit_code == 0
    assert client.get_last_price_called_with == "BTC-USDT-SWAP"
    payload = json.loads(capsys.readouterr().out)
    assert payload["price"] == 250.0
    assert payload["quantity"] == 0.8


def test_paper_order_rejects_when_local_cash_is_insufficient(tmp_path):
    state_path = tmp_path / "paper_state.json"
    state_path.write_text(
        json.dumps(
            {
                "cash_usdt": 50.0,
                "realized_pnl": 0.0,
                "positions": [],
                "updated_at": "1970-01-01T00:00:00Z",
            }
        )
    )
    main, _, _, _, _, _ = build_main_with_stubs(state_path=state_path)
    signal_file = tmp_path / "signal.json"
    signal_file.write_text(json.dumps(valid_signal()))

    with pytest.raises(ValueError, match="insufficient local cash"):
        main(
            [
                "paper-order",
                "--symbol",
                "BTC-USDT-SWAP",
                "--signal-file",
                str(signal_file),
                "--margin-usdt",
                "100",
            ]
        )


def test_positions_prints_local_state_positions(tmp_path, capsys):
    state_path = tmp_path / "paper_state.json"
    state_path.write_text(
        json.dumps(
            {
                "cash_usdt": 9800.0,
                "realized_pnl": 0.0,
                "positions": [
                    {
                        "symbol": "BTC-USDT-SWAP",
                        "side": "long",
                        "quantity": 2.0,
                        "avg_entry_price": 123.5,
                        "margin_used": 200.0,
                    },
                    {
                        "symbol": "ETH-USDT-SWAP",
                        "side": "short",
                        "quantity": 1.0,
                        "avg_entry_price": 3000.0,
                        "margin_used": 150.0,
                    },
                ],
                "updated_at": "1970-01-01T00:00:00Z",
            }
        )
    )
    main, client, _, _, _, _ = build_main_with_stubs(state_path=state_path)
    expected = [
        {
            "symbol": "BTC-USDT-SWAP",
            "side": "long",
            "quantity": 2.0,
            "avg_entry_price": 123.5,
            "margin_used": 200.0,
        }
    ]

    exit_code = main(["positions", "--symbol", "BTC-USDT-SWAP"])

    assert exit_code == 0
    assert client.get_last_price_called_with is None
    assert json.loads(capsys.readouterr().out) == expected


def test_fetch_history_does_not_require_credentials(capsys):
    client = fake_client()
    main = main_factory(
        load_config=lambda: (_ for _ in ()).throw(AssertionError("should not load config")),
        client_factory=lambda: client,
        analyze_fn=fake_analyze_with_codex,
        write_text=real_write_text,
        state_path=Path("paper_state.json"),
        now_fn=lambda: "1970-01-01T00:00:00Z",
        report_fn=lambda **kwargs: {},
        bbmr_report_fn=lambda **kwargs: {},
    )

    exit_code = main(["fetch-history", "--symbol", "BTC-USDT-SWAP", "--bar", "1H", "--limit", "2"])

    assert exit_code == 0
    assert client.get_candles_called_with == ("BTC-USDT-SWAP", "1H", 2)
    assert json.loads(capsys.readouterr().out) == [asdict(candle) for candle in sample_candles(limit=2)]


def test_cli_rejects_unsupported_symbol():
    main, _, _, _, _, _ = build_main_with_stubs()

    with pytest.raises(SystemExit):
        main(["fetch-history", "--symbol", "SOL-USDT-SWAP", "--bar", "1H", "--limit", "20"])


def test_cli_rejects_leverage_out_of_range():
    main, _, _, _, _, _ = build_main_with_stubs()

    with pytest.raises(SystemExit):
        main(["backtest", "--symbol", "BTC-USDT-SWAP", "--bar", "1H", "--limit", "50", "--leverage", "4"])


def test_backtest_report_generates_html_report(capsys, tmp_path):
    main, client, report_calls, _, _, _ = build_main_with_stubs()
    output_file = tmp_path / "report.html"

    exit_code = main(
        [
            "backtest-report",
            "--symbol",
            "BTC-USDT-SWAP",
            "--bar",
            "1H",
            "--limit",
            "50",
            "--leverage",
            "2",
            "--output-file",
            str(output_file),
        ]
    )

    assert exit_code == 0
    assert client.get_candles_called_with == ("BTC-USDT-SWAP", "1H", 50)
    assert len(report_calls) == 1
    assert report_calls[0]["leverage"] == 2
    assert report_calls[0]["output_file"] == output_file
    assert report_calls[0]["symbol"] == "BTC-USDT-SWAP"
    assert report_calls[0]["bar"] == "1H"
    assert json.loads(capsys.readouterr().out) == {
        "report_file": str(output_file),
        "plot_file": str(output_file).replace(".html", ".plot.html"),
        "trade_count": 3,
        "total_return": 0.12,
    }


def test_backtest_bbmr_report_generates_single_page_report(capsys, tmp_path):
    main, client, _, bbmr_report_calls, _, _ = build_main_with_stubs()
    output_file = tmp_path / "bbmr.html"

    exit_code = main(
        [
            "backtest-bbmr-report",
            "--symbol",
            "BTC-USDT-SWAP",
            "--bar",
            "3m",
            "--history-limit",
            "5000",
            "--leverage",
            "2",
            "--segments",
            "8",
            "--window-size",
            "300",
            "--output-file",
            str(output_file),
        ]
    )

    assert exit_code == 0
    assert client.get_candles_called_with == ("BTC-USDT-SWAP", "3m", 5000)
    assert len(bbmr_report_calls) == 1
    assert bbmr_report_calls[0]["symbol"] == "BTC-USDT-SWAP"
    assert bbmr_report_calls[0]["bar"] == "3m"
    assert bbmr_report_calls[0]["segments"] == 8
    assert bbmr_report_calls[0]["window_size"] == 300
    assert json.loads(capsys.readouterr().out) == {
        "report_file": str(output_file),
        "segment_count": 8,
        "window_size": 300,
        "aggregate_trade_count": 11,
        "average_return": 0.031,
    }


def test_backtest_bbsb_report_generates_single_page_report(capsys, tmp_path):
    main, client, _, _, bbsb_report_calls, _ = build_main_with_stubs()
    output_file = tmp_path / "bbsb.html"

    exit_code = main(
        [
            "backtest-bbsb-report",
            "--symbol",
            "BTC-USDT-SWAP",
            "--bar",
            "3m",
            "--history-limit",
            "5000",
            "--leverage",
            "2",
            "--segments",
            "8",
            "--window-size",
            "300",
            "--output-file",
            str(output_file),
        ]
    )

    assert exit_code == 0
    assert client.get_candles_called_with == ("BTC-USDT-SWAP", "3m", 5000)
    assert len(bbsb_report_calls) == 1
    assert bbsb_report_calls[0]["symbol"] == "BTC-USDT-SWAP"
    assert bbsb_report_calls[0]["bar"] == "3m"
    assert bbsb_report_calls[0]["segments"] == 8
    assert bbsb_report_calls[0]["window_size"] == 300
    assert json.loads(capsys.readouterr().out) == {
        "report_file": str(output_file),
        "segment_count": 8,
        "window_size": 300,
        "aggregate_trade_count": 11,
        "average_return": 0.031,
    }


def test_backtest_donchian_report_dispatches_generator(capsys, tmp_path):
    main, client, _, _, _, donchian_report_calls = build_main_with_stubs()
    output_file = tmp_path / "donchian.html"

    exit_code = main(
        [
            "backtest-donchian-report",
            "--symbol",
            "BTC-USDT-SWAP",
            "--bar",
            "3m",
            "--history-limit",
            "5000",
            "--leverage",
            "2",
            "--segments",
            "8",
            "--window-size",
            "300",
            "--entry-window",
            "30",
            "--exit-window",
            "12",
            "--stop-loss-pct",
            "0.02",
            "--output-file",
            str(output_file),
        ]
    )

    assert exit_code == 0
    assert client.get_candles_called_with == ("BTC-USDT-SWAP", "3m", 5000)
    assert len(donchian_report_calls) == 1
    assert donchian_report_calls[0]["symbol"] == "BTC-USDT-SWAP"
    assert donchian_report_calls[0]["bar"] == "3m"
    assert donchian_report_calls[0]["segments"] == 8
    assert donchian_report_calls[0]["window_size"] == 300
    assert donchian_report_calls[0]["entry_window"] == 30
    assert donchian_report_calls[0]["exit_window"] == 12
    assert donchian_report_calls[0]["stop_loss_pct"] == pytest.approx(0.02)
    assert json.loads(capsys.readouterr().out) == {
        "report_file": str(output_file),
        "segment_count": 8,
        "window_size": 300,
        "aggregate_trade_count": 7,
        "average_return": 0.024,
    }


def test_backtest_rsi2_report_dispatches_generator(capsys, tmp_path):
    client = fake_client()
    rsi2_report_calls: list[dict[str, object]] = []

    def fake_rsi2_report(
        *,
        candles,
        leverage,
        output_file,
        symbol,
        bar,
        segments,
        window_size,
        trend_sma,
        rsi_length,
        rsi_long_threshold,
        rsi_short_threshold,
        exit_rsi,
    ):
        rsi2_report_calls.append(
            {
                "candles": candles,
                "leverage": leverage,
                "output_file": output_file,
                "symbol": symbol,
                "bar": bar,
                "segments": segments,
                "window_size": window_size,
                "trend_sma": trend_sma,
                "rsi_length": rsi_length,
                "rsi_long_threshold": rsi_long_threshold,
                "rsi_short_threshold": rsi_short_threshold,
                "exit_rsi": exit_rsi,
            }
        )
        return {
            "report_file": str(output_file),
            "segment_count": segments,
            "window_size": window_size,
            "aggregate_trade_count": 5,
            "average_return": 0.019,
        }

    main = main_factory(
        load_config=lambda: sample_config(),
        client_factory=lambda: client,
        analyze_fn=fake_analyze_with_codex,
        write_text=real_write_text,
        state_path=Path("paper_state.json"),
        now_fn=lambda: "1970-01-01T00:00:00Z",
        report_fn=lambda **kwargs: {},
        bbmr_report_fn=lambda **kwargs: {},
        bbsb_report_fn=lambda **kwargs: {},
        donchian_report_fn=lambda **kwargs: {},
        rsi2_report_fn=fake_rsi2_report,
    )
    output_file = tmp_path / "rsi2.html"

    exit_code = main(
        [
            "backtest-rsi2-report",
            "--symbol",
            "BTC-USDT-SWAP",
            "--bar",
            "3m",
            "--history-limit",
            "5000",
            "--leverage",
            "2",
            "--segments",
            "8",
            "--window-size",
            "300",
            "--trend-sma",
            "30",
            "--rsi-length",
            "3",
            "--rsi-long-threshold",
            "15",
            "--rsi-short-threshold",
            "85",
            "--exit-rsi",
            "55",
            "--output-file",
            str(output_file),
        ]
    )

    assert exit_code == 0
    assert client.get_candles_called_with == ("BTC-USDT-SWAP", "3m", 5000)
    assert len(rsi2_report_calls) == 1
    assert rsi2_report_calls[0]["symbol"] == "BTC-USDT-SWAP"
    assert rsi2_report_calls[0]["bar"] == "3m"
    assert rsi2_report_calls[0]["segments"] == 8
    assert rsi2_report_calls[0]["window_size"] == 300
    assert rsi2_report_calls[0]["trend_sma"] == 30
    assert rsi2_report_calls[0]["rsi_length"] == 3
    assert rsi2_report_calls[0]["rsi_long_threshold"] == pytest.approx(15.0)
    assert rsi2_report_calls[0]["rsi_short_threshold"] == pytest.approx(85.0)
    assert rsi2_report_calls[0]["exit_rsi"] == pytest.approx(55.0)
    assert json.loads(capsys.readouterr().out) == {
        "report_file": str(output_file),
        "segment_count": 8,
        "window_size": 300,
        "aggregate_trade_count": 5,
        "average_return": 0.019,
    }


def test_backtest_ema_pullback_report_dispatches_generator(capsys, tmp_path):
    client = fake_client()
    ema_pullback_report_calls: list[dict[str, object]] = []

    def fake_ema_pullback_report(
        *,
        candles,
        leverage,
        output_file,
        symbol,
        bar,
        segments,
        window_size,
        fast_ema,
        slow_ema,
        stop_buffer_pct,
    ):
        ema_pullback_report_calls.append(
            {
                "candles": candles,
                "leverage": leverage,
                "output_file": output_file,
                "symbol": symbol,
                "bar": bar,
                "segments": segments,
                "window_size": window_size,
                "fast_ema": fast_ema,
                "slow_ema": slow_ema,
                "stop_buffer_pct": stop_buffer_pct,
            }
        )
        return {
            "report_file": str(output_file),
            "segment_count": segments,
            "window_size": window_size,
            "aggregate_trade_count": 6,
            "average_return": 0.021,
        }

    main = main_factory(
        load_config=lambda: sample_config(),
        client_factory=lambda: client,
        analyze_fn=fake_analyze_with_codex,
        write_text=real_write_text,
        state_path=Path("paper_state.json"),
        now_fn=lambda: "1970-01-01T00:00:00Z",
        report_fn=lambda **kwargs: {},
        bbmr_report_fn=lambda **kwargs: {},
        bbsb_report_fn=lambda **kwargs: {},
        donchian_report_fn=lambda **kwargs: {},
        rsi2_report_fn=lambda **kwargs: {},
        ema_pullback_report_fn=fake_ema_pullback_report,
    )
    output_file = tmp_path / "ema-pullback.html"

    exit_code = main(
        [
            "backtest-ema-pullback-report",
            "--symbol",
            "BTC-USDT-SWAP",
            "--bar",
            "3m",
            "--history-limit",
            "5000",
            "--leverage",
            "2",
            "--segments",
            "8",
            "--window-size",
            "300",
            "--fast-ema",
            "30",
            "--slow-ema",
            "80",
            "--stop-buffer-pct",
            "0.01",
            "--output-file",
            str(output_file),
        ]
    )

    assert exit_code == 0
    assert client.get_candles_called_with == ("BTC-USDT-SWAP", "3m", 5000)
    assert len(ema_pullback_report_calls) == 1
    assert ema_pullback_report_calls[0]["symbol"] == "BTC-USDT-SWAP"
    assert ema_pullback_report_calls[0]["bar"] == "3m"
    assert ema_pullback_report_calls[0]["segments"] == 8
    assert ema_pullback_report_calls[0]["window_size"] == 300
    assert ema_pullback_report_calls[0]["fast_ema"] == 30
    assert ema_pullback_report_calls[0]["slow_ema"] == 80
    assert ema_pullback_report_calls[0]["stop_buffer_pct"] == pytest.approx(0.01)
    assert json.loads(capsys.readouterr().out) == {
        "report_file": str(output_file),
        "segment_count": 8,
        "window_size": 300,
        "aggregate_trade_count": 6,
        "average_return": 0.021,
    }