okx-codex-trader/tests/test_cli.py

import json
from dataclasses import asdict
from pathlib import Path

import pytest

from okx_codex_trader.cli import main_factory
from okx_codex_trader.config import Config
from okx_codex_trader.models import Candle, OrderResult, Position, 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_positions_called_with: str | None = None
        self.place_demo_order_called = False
        self.place_demo_order_called_with: tuple[str, TradeSignal, float] | 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 place_demo_order(self, symbol: str, signal: TradeSignal, margin_usdt: float) -> OrderResult:
        self.place_demo_order_called = True
        self.place_demo_order_called_with = (symbol, signal, margin_usdt)
        return OrderResult(
            status="placed",
            order_id="demo-order-1",
            symbol=symbol,
            side="buy",
            pos_side="long",
            order_type="limit",
            size=1.0,
        )

    def get_positions(self, symbol: str) -> list[Position]:
        self.get_positions_called_with = symbol
        return [Position(symbol=symbol, pos_side="long", size=2.0, avg_price=123.5)]


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


def build_main_with_stubs():
    client = fake_client()
    main = main_factory(
        load_config=lambda: sample_config(),
        client_factory=lambda config: client,
        analyze_fn=fake_analyze_with_codex,
        write_text=real_write_text,
    )
    return main, client


def test_fetch_history_prints_candle_json(capsys):
    main, client = build_main_with_stubs()

    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 '"symbol": "BTC-USDT-SWAP"' in capsys.readouterr().out


def test_backtest_prints_summary_json(capsys):
    main, client = build_main_with_stubs()

    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 '"trade_count"' in capsys.readouterr().out


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 json.loads(output_file.read_text()) == valid_signal()
    assert client.get_candles_called_with == ("BTC-USDT-SWAP", "1H", 20)
    assert '"action"' in capsys.readouterr().out


def test_paper_order_reads_signal_file_and_outputs_order_json(tmp_path, capsys):
    main, client = build_main_with_stubs()
    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.place_demo_order_called
    assert client.place_demo_order_called_with is not None
    assert client.place_demo_order_called_with[0] == "BTC-USDT-SWAP"
    assert asdict(client.place_demo_order_called_with[1]) == valid_signal()
    assert client.place_demo_order_called_with[2] == 100.0
    assert '"status"' in capsys.readouterr().out


def test_positions_prints_position_json(capsys):
    main, client = build_main_with_stubs()

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

    assert exit_code == 0
    assert client.get_positions_called_with == "BTC-USDT-SWAP"
    assert '"symbol": "BTC-USDT-SWAP"' in capsys.readouterr().out


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