okx-codex-trader/tests/test_live_execution.py

import pytest

from okx_codex_trader.live_execution import (
    RuntimeState,
    TargetPosition,
    current_position_from_okx,
    plan_position_delta,
    render_market_order_bodies,
    target_from_signal,
)
from okx_codex_trader.models import InstrumentMeta, Position


def nextgen_payload(*, candle_ts=1000, first_signal=False, first_exit=False, second_signal=False, second_exit=False):
    return {
        "decision": {"active_engine": "nextgen"},
        "nextgen": {
            "decision": {"decision_candle_ts": candle_ts},
            "legs": [
                {"leg_id": "a", "suggested_weight": 0.5, "signal": first_signal, "exit_signal": first_exit},
                {"leg_id": "b", "suggested_weight": 0.5, "signal": second_signal, "exit_signal": second_exit},
            ],
        },
    }


def test_nextgen_target_opens_one_virtual_leg():
    state, target = target_from_signal(nextgen_payload(first_signal=True), RuntimeState(None, (), None))

    assert state.nextgen_active_legs == ("a",)
    assert target == TargetPosition(side="long", unit=0.5, known=True, reason="active nextgen virtual legs net to one long ETH target")


def test_nextgen_target_nets_two_virtual_legs_to_one_unit():
    state, target = target_from_signal(nextgen_payload(first_signal=True, second_signal=True), RuntimeState(None, (), None))

    assert state.nextgen_active_legs == ("a", "b")
    assert target.side == "long"
    assert target.unit == 1.0


def test_nextgen_exit_removes_only_active_leg():
    previous = RuntimeState(last_candle_ts=1000, nextgen_active_legs=("a", "b"), micro_side=None)
    state, target = target_from_signal(nextgen_payload(candle_ts=2000, first_exit=True), previous)

    assert state.nextgen_active_legs == ("b",)
    assert target.side == "long"
    assert target.unit == 0.5


def test_repeated_candle_is_idempotent():
    previous = RuntimeState(last_candle_ts=1000, nextgen_active_legs=("a",), micro_side=None)
    state, target = target_from_signal(nextgen_payload(candle_ts=1000, first_signal=True, second_signal=True), previous)

    assert state == previous
    assert target.side == "long"
    assert target.unit == 0.5


def test_micro_target_is_blocked_until_exit_state_exists():
    payload = {"decision": {"active_engine": "micro"}, "micro": {"decision_candle_ts": 1000}}

    _, target = target_from_signal(payload, RuntimeState(None, (), None))

    assert target.known is False
    assert target.unit == 0.0


def test_position_delta_plans_reduce_only_close_before_reverse():
    current = TargetPosition(side="long", unit=1.0, known=True, reason="current")
    target = TargetPosition(side="short", unit=0.5, known=True, reason="target")

    plan = plan_position_delta(current, target)

    assert [(action.action, action.side, action.unit, action.reduce_only) for action in plan.actions] == [
        ("close", "long", 1.0, True),
        ("reverse", "short", 0.5, False),
    ]


def test_position_delta_does_not_plan_when_current_position_is_unknown():
    current = TargetPosition(side="flat", unit=0.0, known=False, reason="unknown")
    target = TargetPosition(side="long", unit=0.5, known=True, reason="target")

    plan = plan_position_delta(current, target)

    assert plan.actions == ()


def test_current_position_from_okx_normalizes_contracts_to_strategy_units():
    current = current_position_from_okx(
        positions=[Position(symbol="ETH-USDT-SWAP", pos_side="long", size=10.0, avg_price=3000.0)],
        mark_price=3000.0,
        metadata=InstrumentMeta(ct_val=0.1, lot_sz=1.0, min_sz=1.0),
        leverage=3,
        margin_per_unit_usdt=1000.0,
    )

    assert current.side == "long"
    assert current.unit == pytest.approx(1.0)
    assert current.known is True


def test_current_position_from_okx_blocks_when_both_hedge_sides_are_open():
    current = current_position_from_okx(
        positions=[
            Position(symbol="ETH-USDT-SWAP", pos_side="long", size=1.0, avg_price=3000.0),
            Position(symbol="ETH-USDT-SWAP", pos_side="short", size=1.0, avg_price=3000.0),
        ],
        mark_price=3000.0,
        metadata=InstrumentMeta(ct_val=0.1, lot_sz=1.0, min_sz=1.0),
        leverage=3,
        margin_per_unit_usdt=1000.0,
    )

    assert current.known is False


def test_render_market_order_bodies_builds_open_order_body():
    plan = plan_position_delta(
        TargetPosition(side="flat", unit=0.0, known=True, reason="current"),
        TargetPosition(side="long", unit=0.5, known=True, reason="target"),
    )

    orders = render_market_order_bodies(
        plan=plan,
        symbol="ETH-USDT-SWAP",
        mark_price=3000.0,
        metadata=InstrumentMeta(ct_val=0.1, lot_sz=1.0, min_sz=1.0),
        leverage=3,
        margin_per_unit_usdt=1000.0,
        max_new_margin_usdt=500.0,
        max_total_margin_usdt=1000.0,
        client_order_id_prefix="eth-1000",
    )

    assert len(orders) == 1
    assert orders[0].margin_usdt == 500.0
    assert orders[0].body == {
        "instId": "ETH-USDT-SWAP",
        "tdMode": "isolated",
        "side": "buy",
        "posSide": "long",
        "ordType": "market",
        "sz": "5",
        "clOrdId": "eth-1000-1-open",
    }


def test_render_market_order_bodies_builds_reduce_only_close_before_reverse():
    plan = plan_position_delta(
        TargetPosition(side="long", unit=1.0, known=True, reason="current"),
        TargetPosition(side="short", unit=0.5, known=True, reason="target"),
    )

    orders = render_market_order_bodies(
        plan=plan,
        symbol="ETH-USDT-SWAP",
        mark_price=3000.0,
        metadata=InstrumentMeta(ct_val=0.1, lot_sz=1.0, min_sz=1.0),
        leverage=3,
        margin_per_unit_usdt=1000.0,
        max_new_margin_usdt=500.0,
        max_total_margin_usdt=1000.0,
        client_order_id_prefix="eth-2000",
    )

    assert [order.body for order in orders] == [
        {
            "instId": "ETH-USDT-SWAP",
            "tdMode": "isolated",
            "side": "sell",
            "posSide": "long",
            "ordType": "market",
            "sz": "10",
            "clOrdId": "eth-2000-1-close",
            "reduceOnly": "true",
        },
        {
            "instId": "ETH-USDT-SWAP",
            "tdMode": "isolated",
            "side": "sell",
            "posSide": "short",
            "ordType": "market",
            "sz": "5",
            "clOrdId": "eth-2000-2-reverse",
        },
    ]


def test_render_market_order_bodies_enforces_new_margin_cap():
    plan = plan_position_delta(
        TargetPosition(side="flat", unit=0.0, known=True, reason="current"),
        TargetPosition(side="long", unit=1.0, known=True, reason="target"),
    )

    with pytest.raises(ValueError, match="new margin exceeds max_new_margin_usdt"):
        render_market_order_bodies(
            plan=plan,
            symbol="ETH-USDT-SWAP",
            mark_price=3000.0,
            metadata=InstrumentMeta(ct_val=0.1, lot_sz=1.0, min_sz=1.0),
            leverage=3,
            margin_per_unit_usdt=1000.0,
            max_new_margin_usdt=500.0,
            max_total_margin_usdt=1000.0,
            client_order_id_prefix="eth-3000",
        )


def test_render_market_order_bodies_enforces_total_margin_cap():
    plan = plan_position_delta(
        TargetPosition(side="long", unit=0.5, known=True, reason="current"),
        TargetPosition(side="long", unit=1.0, known=True, reason="target"),
    )

    with pytest.raises(ValueError, match="target margin exceeds max_total_margin_usdt"):
        render_market_order_bodies(
            plan=plan,
            symbol="ETH-USDT-SWAP",
            mark_price=3000.0,
            metadata=InstrumentMeta(ct_val=0.1, lot_sz=1.0, min_sz=1.0),
            leverage=3,
            margin_per_unit_usdt=1000.0,
            max_new_margin_usdt=500.0,
            max_total_margin_usdt=500.0,
            client_order_id_prefix="eth-4000",
        )