okx-codex-trader/tests/test_eth_nextgen_micro_executor.py

import importlib.util
import sys
from pathlib import Path

from okx_codex_trader.live_execution import TargetPosition


def load_executor():
    path = Path(__file__).resolve().parents[1] / "scripts" / "run_eth_nextgen_micro_executor.py"
    spec = importlib.util.spec_from_file_location("run_eth_nextgen_micro_executor", path)
    assert spec is not None
    module = importlib.util.module_from_spec(spec)
    assert spec.loader is not None
    sys.modules[spec.name] = module
    spec.loader.exec_module(module)
    return module


executor = load_executor()


def payload_with_nextgen_long():
    return {
        "decision": {"active_engine": "nextgen", "selected_signal": "long"},
        "execution_intent": {"entry_signal": "long", "entry_unit": 0.5, "target_position_known": False, "target_position": None},
        "nextgen": {
            "data": {"decision_candle_ts": 1000},
            "legs": [
                {"leg_id": "a", "suggested_weight": 0.5, "signal": True, "exit_signal": False},
                {"leg_id": "b", "suggested_weight": 0.5, "signal": False, "exit_signal": False},
            ],
        },
    }


def test_executor_snapshot_does_not_render_orders_when_current_position_is_unknown(monkeypatch, tmp_path):
    monkeypatch.setattr(executor.eth_nextgen_micro, "build_payload", payload_with_nextgen_long)
    monkeypatch.setattr(
        executor,
        "account_current_position",
        lambda _: (TargetPosition(side="flat", unit=0.0, known=False, reason="no credentials"), {"account_error": "no credentials"}),
    )

    snapshot = executor.build_snapshot(state_dir=tmp_path, margin_per_unit_usdt=1000.0, max_new_margin_usdt=500.0, max_total_margin_usdt=1000.0)

    assert snapshot["orders_submitted"] == 0
    assert snapshot["target_position"]["known"] is True
    assert snapshot["target_position"]["unit"] == 0.5
    assert snapshot["current_position"]["known"] is False
    assert snapshot["rendered_orders"] == []


def test_executor_snapshot_renders_order_body_when_positions_are_known(monkeypatch, tmp_path):
    monkeypatch.setattr(executor.eth_nextgen_micro, "build_payload", payload_with_nextgen_long)
    monkeypatch.setattr(
        executor,
        "account_current_position",
        lambda _: (
            TargetPosition(side="flat", unit=0.0, known=True, reason="flat"),
            {"mark_price": 3000.0, "instrument_meta": {"ct_val": 0.1, "lot_sz": 1.0, "min_sz": 1.0}},
        ),
    )

    snapshot = executor.build_snapshot(state_dir=tmp_path, margin_per_unit_usdt=1000.0, max_new_margin_usdt=500.0, max_total_margin_usdt=1000.0)

    assert snapshot["orders_submitted"] == 0
    assert snapshot["rendered_orders"] == [
        {
            "action": "open",
            "margin_usdt": 500.0,
            "body": {
                "instId": "ETH-USDT-SWAP",
                "tdMode": "isolated",
                "side": "buy",
                "posSide": "long",
                "ordType": "market",
                "sz": "5",
                "clOrdId": "ethnm-1000-1-open",
            },
        }
    ]


def test_executor_uses_executor_state_file(monkeypatch, tmp_path):
    monkeypatch.setattr(executor.eth_nextgen_micro, "build_payload", payload_with_nextgen_long)
    monkeypatch.setattr(
        executor,
        "account_current_position",
        lambda _: (
            TargetPosition(side="flat", unit=0.0, known=True, reason="flat"),
            {"mark_price": 3000.0, "instrument_meta": {"ct_val": 0.1, "lot_sz": 1.0, "min_sz": 1.0}},
        ),
    )
    (tmp_path / "runtime-state.json").write_text('{"last_candle_ts":1000,"micro_side":null,"nextgen_active_legs":[]}\n', encoding="utf-8")

    snapshot = executor.build_snapshot(state_dir=tmp_path, margin_per_unit_usdt=1000.0, max_new_margin_usdt=500.0, max_total_margin_usdt=1000.0)

    assert snapshot["target_position"]["unit"] == 0.5