Add ETH nextgen micro execution planning

docs/live-eth-nextgen-micro-deployment.md

@@ -68,3 +68,22 @@ The executor should process only completed 15m candles. For each candle it shoul
 5. Persist candle timestamp, target position, submitted client order id, fills, fee, and resulting position.
 
 The first live version should use a hard small margin cap and market/reduce-only orders only. Price-distributed TWAP entry can be added after target reconciliation and fill logging are proven live.
+
+## Current implementation boundary
+
+The strategy logic is moving out of ad hoc scripts into package modules:
+
+- `okx_codex_trader.eth_nextgen_micro` builds the ETH nextgen+micro signal payload.
+- `okx_codex_trader.live_execution` maintains runtime strategy state, converts nextgen virtual legs into one net target position, normalizes OKX positions into strategy units, and builds a pure delta plan.
+- `okx_codex_trader.live_execution.render_market_order_bodies` converts a tested delta plan into OKX market order bodies with deterministic client order ids and a hard new-margin cap.
+- `okx_codex_trader.okx_client.submit_market_order_body` can submit a prebuilt market order body, but no deployed service or CLI calls it yet.
+- `scripts/run_eth_nextgen_micro_observer.py` remains read-only. It writes runtime state and target-position diagnostics, but it still submits no orders.
+- `scripts/run_eth_nextgen_micro_executor.py` builds a dry-run execution snapshot. It reads the strategy payload and OKX account state, renders order bodies only when current and target positions are known, and always reports `orders_submitted: 0`.
+
+Live order submission is still intentionally unreachable from the daemon. The next required boundary is an executor command or service that calls the submit method, records client order ids, and verifies fills.
+
+Dry-run executor example:
+
+```bash
+python scripts/run_eth_nextgen_micro_executor.py --margin-per-unit-usdt 5 --max-new-margin-usdt 5
+```

okx_codex_trader/eth_nextgen_micro.py

@@ -0,0 +1,203 @@
+from __future__ import annotations
+
+import json
+import sys
+from datetime import UTC, datetime
+from pathlib import Path
+
+import pandas as pd
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT))
+
+from scripts import build_eth_btc_nextgen_signal_intent as nextgen_intent
+from scripts import search_eth_microstructure_variants as micro
+from scripts import search_eth_nextgen_micro_portfolio as portfolio
+
+
+REPORT_DIR = ROOT / "reports" / "eth-exploration"
+JSON_REPORT = REPORT_DIR / "eth-nextgen-micro-signal-intent.json"
+MARKDOWN_REPORT = REPORT_DIR / "eth-nextgen-micro-signal-intent.md"
+ETH = "ETH-USDT-SWAP"
+BAR = "15m"
+TARGET_NAME = "switch-l30-r96_q0.15_mf0.25_us"
+MICRO_NAME = "atr-compress-expand-r96-q0.15-sl0.008-tp0.016-mf0.25-us"
+LOOKBACK_DAYS = 30
+ROUNDTRIP_COST_ON_MARGIN = 0.0021
+MICRO_PARAMS = {
+    "range_window": 96,
+    "atr_window": 48,
+    "atr_quantile_window": 480,
+    "atr_quantile": 0.15,
+    "stop_loss_pct": 0.008,
+    "take_profit_pct": 0.016,
+    "max_hold_bars": 32,
+    "margin_fraction": 0.25,
+    "session": "us",
+}
+
+
+def iso_text(ts: int) -> str:
+    return datetime.fromtimestamp(ts / 1000, UTC).isoformat().replace("+00:00", "Z")
+
+
+def latest_active_engine() -> dict[str, object]:
+    existing_equity = pd.read_csv(REPORT_DIR / "eth-btc-nextgen-equity.csv")
+    base = existing_equity[
+        (existing_equity["cost_model"] == portfolio.PRIMARY_COST) & (existing_equity["name"] == portfolio.NEXTGEN_BASELINE)
+    ].copy()
+    if base.empty:
+        raise KeyError(f"missing existing nextgen equity for {portfolio.NEXTGEN_BASELINE}")
+    index = pd.DatetimeIndex(pd.to_datetime(base["date"], utc=True))
+    nextgen_series, _ = portfolio.load_nextgen(index, ROUNDTRIP_COST_ON_MARGIN)
+    micro_series = portfolio.load_micro_candidates(index, ROUNDTRIP_COST_ON_MARGIN)[MICRO_NAME][0]
+    nextgen_regime = nextgen_series / nextgen_series.shift(LOOKBACK_DAYS) - 1.0
+    micro_regime = micro_series / micro_series.shift(LOOKBACK_DAYS) - 1.0
+    active = ((nextgen_regime < 0.0) & (micro_regime > 0.0)).shift(1).fillna(False).astype(bool)
+    decision_date = active.index[-1]
+    micro_active = bool(active.iloc[-1])
+    return {
+        "active_engine": "micro" if micro_active else "nextgen",
+        "decision_date": decision_date.strftime("%Y-%m-%d"),
+        "switch_rule": "prior completed daily nextgen 30d return < 0 and micro 30d return > 0",
+        "lookback_days": LOOKBACK_DAYS,
+        "nextgen_30d_return": float(nextgen_regime.iloc[-2]) if len(nextgen_regime) >= 2 else None,
+        "micro_30d_return": float(micro_regime.iloc[-2]) if len(micro_regime) >= 2 else None,
+        "nextgen_equity": float(nextgen_series.iloc[-1]),
+        "micro_equity": float(micro_series.iloc[-1]),
+    }
+
+
+def micro_signal() -> dict[str, object]:
+    candles = micro._load_candles(ETH, BAR)
+    decision_index = len(candles) - 2
+    if decision_index < max(int(MICRO_PARAMS["range_window"]), int(MICRO_PARAMS["atr_window"]), int(MICRO_PARAMS["atr_quantile_window"])):
+        raise ValueError("not enough ETH candles for micro signal")
+
+    highs = pd.Series([c.high for c in candles], dtype=float)
+    lows = pd.Series([c.low for c in candles], dtype=float)
+    closes = pd.Series([c.close for c in candles], dtype=float)
+    prev_close = closes.shift(1)
+    true_range = pd.concat([(highs - lows), (highs - prev_close).abs(), (lows - prev_close).abs()], axis=1).max(axis=1)
+    atr = true_range.rolling(int(MICRO_PARAMS["atr_window"])).mean() / closes
+    atr_limit = atr.rolling(int(MICRO_PARAMS["atr_quantile_window"])).quantile(float(MICRO_PARAMS["atr_quantile"]))
+    range_high = highs.shift(1).rolling(int(MICRO_PARAMS["range_window"])).max()
+    range_low = lows.shift(1).rolling(int(MICRO_PARAMS["range_window"])).min()
+    candle = candles[decision_index]
+    compressed = bool(float(atr.iloc[decision_index - 1]) <= float(atr_limit.iloc[decision_index - 1]))
+    long_signal = compressed and candle.close > float(range_high.iloc[decision_index])
+    short_signal = compressed and candle.close < float(range_low.iloc[decision_index])
+    session_ok = micro._session_ok(candle, str(MICRO_PARAMS["session"]))
+    signal = "long" if session_ok and long_signal else "short" if session_ok and short_signal else "no_signal"
+    return {
+        "engine": "micro",
+        "candidate": MICRO_NAME,
+        "symbol": ETH,
+        "bar": BAR,
+        "decision_candle_ts": candle.ts,
+        "decision_candle_time": iso_text(candle.ts),
+        "latest_local_candle_ts": candles[-1].ts,
+        "latest_local_candle_time": iso_text(candles[-1].ts),
+        "signal": signal,
+        "raw_long_signal": bool(long_signal),
+        "raw_short_signal": bool(short_signal),
+        "session_ok": session_ok,
+        "indicators": {
+            "eth_close": candle.close,
+            "atr_previous": float(atr.iloc[decision_index - 1]),
+            "atr_limit_previous": float(atr_limit.iloc[decision_index - 1]),
+            "range_high": float(range_high.iloc[decision_index]),
+            "range_low": float(range_low.iloc[decision_index]),
+        },
+        "params": MICRO_PARAMS,
+    }
+
+
+def execution_intent(active_engine: str, selected_signal: str, nextgen_payload: dict[str, object]) -> dict[str, object]:
+    if selected_signal == "no_signal":
+        entry_unit = 0.0
+    elif active_engine == "nextgen":
+        entry_unit = float(nextgen_payload["decision"]["active_suggested_weight"])
+    else:
+        entry_unit = 1.0
+    return {
+        "entry_signal": selected_signal,
+        "entry_unit": entry_unit,
+        "target_position_known": False,
+        "target_position": None,
+        "blocker": "persistent strategy position state is required before entry signals can be reconciled to target position",
+    }
+
+
+def build_payload() -> dict[str, object]:
+    switch_state = latest_active_engine()
+    nextgen_payload = nextgen_intent.build_payload()
+    micro_payload = micro_signal()
+    active_engine = str(switch_state["active_engine"])
+    selected_signal = str(nextgen_payload["decision"]["signal"]) if active_engine == "nextgen" else str(micro_payload["signal"])
+    return {
+        "mode": "readonly_signal_intent",
+        "strategy": {
+            "name": TARGET_NAME,
+            "symbol": ETH,
+            "bar": BAR,
+            "direction": "nextgen_long_only_or_micro_observation",
+            "source_report": "reports/eth-exploration/eth-nextgen-micro-portfolio-report.md",
+            "cost_model": "maker_taker",
+            "roundtrip_cost_on_margin": ROUNDTRIP_COST_ON_MARGIN,
+        },
+        "created_at": datetime.now(UTC).isoformat(timespec="seconds").replace("+00:00", "Z"),
+        "submitted_orders": 0,
+        "private_key_required": False,
+        "order_client": None,
+        "switch_state": switch_state,
+        "decision": {
+            "active_engine": active_engine,
+            "selected_signal": selected_signal,
+            "needs_order": False,
+            "needs_cancel": False,
+            "intent": f"observe_{active_engine}_{selected_signal}",
+        },
+        "execution_intent": execution_intent(active_engine, selected_signal, nextgen_payload),
+        "risk_limits": {
+            "no_order_submission": True,
+            "no_cancel_submission": True,
+            "no_position_state_assumed": True,
+            "execution": "intent_only",
+            "blocked_for_live_trading": True,
+            "blocker": "persistent virtual position state is not maintained by this read-only signal builder",
+        },
+        "nextgen": {
+            "decision": nextgen_payload["decision"],
+            "data": nextgen_payload["data"],
+            "legs": nextgen_payload["legs"],
+        },
+        "micro": micro_payload,
+    }
+
+
+def markdown_report(payload: dict[str, object]) -> str:
+    lines = [
+        "# ETH nextgen + micro signal intent",
+        "",
+        "Read-only signal intent. No order or cancel request was submitted.",
+        "",
+        "## Decision",
+        "",
+        f"- Created at: `{payload['created_at']}`",
+        f"- Strategy: `{payload['strategy']['name']}`",
+        f"- Active engine: `{payload['decision']['active_engine']}`",
+        f"- Selected signal: `{payload['decision']['selected_signal']}`",
+        f"- Entry unit: `{payload['execution_intent']['entry_unit']}`",
+        f"- Target position known: `{payload['execution_intent']['target_position_known']}`",
+        f"- Needs order: `{payload['decision']['needs_order']}`",
+        f"- Blocked for live trading: `{payload['risk_limits']['blocked_for_live_trading']}`",
+        f"- Blocker: `{payload['risk_limits']['blocker']}`",
+        "",
+        "## Intent JSON",
+        "",
+        "```json",
+        json.dumps(payload, indent=2, sort_keys=True),
+        "```",
+    ]
+    return "\n".join(lines) + "\n"

okx_codex_trader/live_execution.py

@@ -0,0 +1,228 @@
+from __future__ import annotations
+
+import json
+from dataclasses import asdict, dataclass
+from pathlib import Path
+from typing import Literal
+
+from okx_codex_trader.models import InstrumentMeta, Position
+from okx_codex_trader.okx_client import OkxClient, build_contract_size
+
+
+PositionSide = Literal["flat", "long", "short"]
+
+
+@dataclass(frozen=True)
+class TargetPosition:
+    side: PositionSide
+    unit: float
+    known: bool
+    reason: str
+
+
+@dataclass(frozen=True)
+class RuntimeState:
+    last_candle_ts: int | None
+    nextgen_active_legs: tuple[str, ...]
+    micro_side: Literal["long", "short"] | None
+
+
+@dataclass(frozen=True)
+class PlannedAction:
+    action: Literal["noop", "open", "increase", "reduce", "close", "reverse"]
+    side: PositionSide
+    unit: float
+    reduce_only: bool
+
+
+@dataclass(frozen=True)
+class ExecutionPlan:
+    target: TargetPosition
+    current: TargetPosition
+    actions: tuple[PlannedAction, ...]
+
+
+@dataclass(frozen=True)
+class RenderedOrder:
+    action: str
+    margin_usdt: float
+    body: dict[str, str]
+
+
+EMPTY_STATE = RuntimeState(last_candle_ts=None, nextgen_active_legs=(), micro_side=None)
+
+
+def load_runtime_state(path: Path) -> RuntimeState:
+    if not path.exists():
+        return EMPTY_STATE
+    payload = json.loads(path.read_text(encoding="utf-8"))
+    return RuntimeState(
+        last_candle_ts=payload["last_candle_ts"],
+        nextgen_active_legs=tuple(payload["nextgen_active_legs"]),
+        micro_side=payload["micro_side"],
+    )
+
+
+def save_runtime_state(path: Path, state: RuntimeState) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(asdict(state), indent=2, sort_keys=True) + "\n", encoding="utf-8")
+
+
+def _decision_candle_ts(payload: dict[str, object]) -> int:
+    active_engine = str(payload["decision"]["active_engine"])
+    if active_engine == "nextgen":
+        nextgen = payload["nextgen"]
+        if "data" in nextgen:
+            return int(nextgen["data"]["decision_candle_ts"])
+        return int(nextgen["decision"]["decision_candle_ts"])
+    return int(payload["micro"]["decision_candle_ts"])
+
+
+def target_from_signal(payload: dict[str, object], state: RuntimeState) -> tuple[RuntimeState, TargetPosition]:
+    candle_ts = _decision_candle_ts(payload)
+    if state.last_candle_ts is not None and candle_ts <= state.last_candle_ts:
+        return state, target_from_state(payload, state)
+
+    active_engine = str(payload["decision"]["active_engine"])
+    if active_engine == "nextgen":
+        active = set(state.nextgen_active_legs)
+        weights: dict[str, float] = {}
+        for leg in payload["nextgen"]["legs"]:
+            leg_id = str(leg["leg_id"])
+            weights[leg_id] = float(leg["suggested_weight"])
+            if bool(leg["signal"]):
+                active.add(leg_id)
+            elif leg_id in active and bool(leg["exit_signal"]):
+                active.remove(leg_id)
+        next_state = RuntimeState(last_candle_ts=candle_ts, nextgen_active_legs=tuple(sorted(active)), micro_side=None)
+        return next_state, _nextgen_target(next_state, weights)
+
+    return (
+        RuntimeState(last_candle_ts=candle_ts, nextgen_active_legs=(), micro_side=state.micro_side),
+        TargetPosition(
+            side="flat",
+            unit=0.0,
+            known=False,
+            reason="micro target position requires persistent micro exit state before live execution",
+        ),
+    )
+
+
+def target_from_state(payload: dict[str, object], state: RuntimeState) -> TargetPosition:
+    active_engine = str(payload["decision"]["active_engine"])
+    if active_engine != "nextgen":
+        return TargetPosition(
+            side="flat",
+            unit=0.0,
+            known=False,
+            reason="micro target position requires persistent micro exit state before live execution",
+        )
+    weights = {str(leg["leg_id"]): float(leg["suggested_weight"]) for leg in payload["nextgen"]["legs"]}
+    return _nextgen_target(state, weights)
+
+
+def _nextgen_target(state: RuntimeState, weights: dict[str, float]) -> TargetPosition:
+    unit = sum(weights[leg_id] for leg_id in state.nextgen_active_legs)
+    if unit <= 0.0:
+        return TargetPosition(side="flat", unit=0.0, known=True, reason="no active nextgen virtual legs")
+    return TargetPosition(side="long", unit=unit, known=True, reason="active nextgen virtual legs net to one long ETH target")
+
+
+def plan_position_delta(current: TargetPosition, target: TargetPosition) -> ExecutionPlan:
+    if not current.known or not target.known:
+        return ExecutionPlan(target=target, current=current, actions=())
+    if current.side == target.side and current.unit == target.unit:
+        return ExecutionPlan(target=target, current=current, actions=(PlannedAction("noop", target.side, 0.0, False),))
+    if current.side == "flat":
+        return ExecutionPlan(target=target, current=current, actions=(PlannedAction("open", target.side, target.unit, False),))
+    if target.side == "flat":
+        return ExecutionPlan(target=target, current=current, actions=(PlannedAction("close", current.side, current.unit, True),))
+    if current.side == target.side:
+        if target.unit > current.unit:
+            return ExecutionPlan(target=target, current=current, actions=(PlannedAction("increase", target.side, target.unit - current.unit, False),))
+        return ExecutionPlan(target=target, current=current, actions=(PlannedAction("reduce", current.side, current.unit - target.unit, True),))
+    return ExecutionPlan(
+        target=target,
+        current=current,
+        actions=(
+            PlannedAction("close", current.side, current.unit, True),
+            PlannedAction("reverse", target.side, target.unit, False),
+        ),
+    )
+
+
+def current_position_from_okx(
+    *,
+    positions: list[Position],
+    mark_price: float,
+    metadata: InstrumentMeta,
+    leverage: int,
+    margin_per_unit_usdt: float,
+) -> TargetPosition:
+    if leverage <= 0 or margin_per_unit_usdt <= 0.0 or mark_price <= 0.0 or metadata.ct_val <= 0.0:
+        raise ValueError("position normalization inputs are invalid")
+    active = [position for position in positions if position.size > 0.0]
+    if not active:
+        return TargetPosition(side="flat", unit=0.0, known=True, reason="no open OKX position")
+    sides = {position.pos_side for position in active}
+    if len(sides) != 1:
+        return TargetPosition(side="flat", unit=0.0, known=False, reason="both OKX hedge sides are open")
+    side = active[0].pos_side
+    if side not in {"long", "short"}:
+        return TargetPosition(side="flat", unit=0.0, known=False, reason="OKX position side is unsupported")
+    notional = sum(position.size for position in active) * metadata.ct_val * mark_price
+    margin = notional / leverage
+    unit = margin / margin_per_unit_usdt
+    return TargetPosition(side=side, unit=unit, known=True, reason="OKX position normalized by configured strategy unit margin")
+
+
+def render_market_order_bodies(
+    *,
+    plan: ExecutionPlan,
+    symbol: str,
+    mark_price: float,
+    metadata: InstrumentMeta,
+    leverage: int,
+    margin_per_unit_usdt: float,
+    max_new_margin_usdt: float,
+    client_order_id_prefix: str,
+) -> tuple[RenderedOrder, ...]:
+    if leverage <= 0 or margin_per_unit_usdt <= 0.0 or max_new_margin_usdt < 0.0:
+        raise ValueError("order rendering inputs are invalid")
+    rendered: list[RenderedOrder] = []
+    new_margin = 0.0
+    index = 1
+    for action in plan.actions:
+        if action.action == "noop":
+            continue
+        margin = action.unit * margin_per_unit_usdt
+        if margin <= 0.0 or action.side == "flat":
+            raise ValueError("planned action is invalid")
+        if not action.reduce_only:
+            new_margin += margin
+            if new_margin > max_new_margin_usdt:
+                raise ValueError("new margin exceeds max_new_margin_usdt")
+        side = _okx_side(action)
+        size = build_contract_size(margin * leverage, mark_price, metadata)
+        rendered.append(
+            RenderedOrder(
+                action=action.action,
+                margin_usdt=margin,
+                body=OkxClient.build_market_order_body(
+                    symbol=symbol,
+                    side=side,
+                    pos_side=action.side,
+                    size=size,
+                    client_order_id=f"{client_order_id_prefix}-{index}-{action.action}",
+                    reduce_only=action.reduce_only,
+                ),
+            )
+        )
+        index += 1
+    return tuple(rendered)
+
+
+def _okx_side(action: PlannedAction) -> str:
+    if action.reduce_only:
+        return "sell" if action.side == "long" else "buy"
+    return "buy" if action.side == "long" else "sell"

okx_codex_trader/okx_client.py

@@ -159,6 +159,33 @@ class OkxClient:
             body["clOrdId"] = client_order_id
         return body
 
+    @staticmethod
+    def build_market_order_body(
+        *,
+        symbol: str,
+        side: str,
+        pos_side: str,
+        size: object,
+        client_order_id: str,
+        reduce_only: bool,
+    ) -> dict[str, str]:
+        if side not in {"buy", "sell"}:
+            raise ValueError("side is invalid")
+        if pos_side not in {"long", "short"}:
+            raise ValueError("pos_side is invalid")
+        body = {
+            "instId": symbol,
+            "tdMode": "isolated",
+            "side": side,
+            "posSide": pos_side,
+            "ordType": "market",
+            "sz": _format_number(size),
+            "clOrdId": client_order_id,
+        }
+        if reduce_only:
+            body["reduceOnly"] = "true"
+        return body
+
     @staticmethod
     def build_pending_orders_params(*, symbol: str) -> dict[str, str]:
         return {"instType": "SWAP", "instId": symbol}
@@ -398,6 +425,26 @@ class OkxClient:
             size=size,
         )
 
+    def submit_market_order_body(self, body: dict[str, str]) -> OrderResult:
+        required = {"instId", "side", "posSide", "ordType", "sz", "clOrdId"}
+        if any(not body.get(key) for key in required) or body.get("ordType") != "market":
+            raise ValueError("market order body is invalid")
+        _parse_finite_float(body.get("sz"))
+        data = self._request("POST", "/api/v5/trade/order", json_body=body)
+        order = self._first_item(data)
+        order_id = str(order.get("ordId") or "")
+        if not order_id:
+            raise self._invalid_payload()
+        return OrderResult(
+            status="placed",
+            order_id=order_id,
+            symbol=body.get("instId"),
+            side=body.get("side"),
+            pos_side=body.get("posSide"),
+            order_type=body.get("ordType"),
+            size=_parse_finite_float(body.get("sz")),
+        )
+
     def get_positions(self, symbol: str) -> list[Position]:
         requested_symbol = symbol
         data = self._request("GET", "/api/v5/account/positions", params={"instId": requested_symbol})

scripts/__init__.py

@@ -0,0 +1 @@
+

scripts/build_eth_nextgen_micro_signal_intent.py

@@ -2,186 +2,7 @@ from __future__ import annotations
 
 import argparse
 import json
-import sys
-from datetime import UTC, datetime
-from pathlib import Path
-
-import pandas as pd
-
-sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
-
-from scripts import build_eth_btc_nextgen_signal_intent as nextgen_intent
-from scripts import search_eth_microstructure_variants as micro
-from scripts import search_eth_nextgen_micro_portfolio as portfolio
-
-
-ROOT = Path(__file__).resolve().parents[1]
-REPORT_DIR = ROOT / "reports" / "eth-exploration"
-JSON_REPORT = REPORT_DIR / "eth-nextgen-micro-signal-intent.json"
-MARKDOWN_REPORT = REPORT_DIR / "eth-nextgen-micro-signal-intent.md"
-ETH = "ETH-USDT-SWAP"
-BAR = "15m"
-TARGET_NAME = "switch-l30-r96_q0.15_mf0.25_us"
-MICRO_NAME = "atr-compress-expand-r96-q0.15-sl0.008-tp0.016-mf0.25-us"
-LOOKBACK_DAYS = 30
-ROUNDTRIP_COST_ON_MARGIN = 0.0021
-MICRO_PARAMS = {
-    "range_window": 96,
-    "atr_window": 48,
-    "atr_quantile_window": 480,
-    "atr_quantile": 0.15,
-    "stop_loss_pct": 0.008,
-    "take_profit_pct": 0.016,
-    "max_hold_bars": 32,
-    "margin_fraction": 0.25,
-    "session": "us",
-}
-
-
-def iso_text(ts: int) -> str:
-    return datetime.fromtimestamp(ts / 1000, UTC).isoformat().replace("+00:00", "Z")
-
-
-def latest_active_engine() -> dict[str, object]:
-    existing_equity = pd.read_csv(REPORT_DIR / "eth-btc-nextgen-equity.csv")
-    base = existing_equity[
-        (existing_equity["cost_model"] == portfolio.PRIMARY_COST) & (existing_equity["name"] == portfolio.NEXTGEN_BASELINE)
-    ].copy()
-    if base.empty:
-        raise KeyError(f"missing existing nextgen equity for {portfolio.NEXTGEN_BASELINE}")
-    index = pd.DatetimeIndex(pd.to_datetime(base["date"], utc=True))
-    nextgen_series, _ = portfolio.load_nextgen(index, ROUNDTRIP_COST_ON_MARGIN)
-    micro_series = portfolio.load_micro_candidates(index, ROUNDTRIP_COST_ON_MARGIN)[MICRO_NAME][0]
-    nextgen_regime = nextgen_series / nextgen_series.shift(LOOKBACK_DAYS) - 1.0
-    micro_regime = micro_series / micro_series.shift(LOOKBACK_DAYS) - 1.0
-    active = ((nextgen_regime < 0.0) & (micro_regime > 0.0)).shift(1).fillna(False).astype(bool)
-    decision_date = active.index[-1]
-    micro_active = bool(active.iloc[-1])
-    return {
-        "active_engine": "micro" if micro_active else "nextgen",
-        "decision_date": decision_date.strftime("%Y-%m-%d"),
-        "switch_rule": "prior completed daily nextgen 30d return < 0 and micro 30d return > 0",
-        "lookback_days": LOOKBACK_DAYS,
-        "nextgen_30d_return": float(nextgen_regime.iloc[-2]) if len(nextgen_regime) >= 2 else None,
-        "micro_30d_return": float(micro_regime.iloc[-2]) if len(micro_regime) >= 2 else None,
-        "nextgen_equity": float(nextgen_series.iloc[-1]),
-        "micro_equity": float(micro_series.iloc[-1]),
-    }
-
-
-def micro_signal() -> dict[str, object]:
-    candles = micro._load_candles(ETH, BAR)
-    decision_index = len(candles) - 2
-    if decision_index < max(int(MICRO_PARAMS["range_window"]), int(MICRO_PARAMS["atr_window"]), int(MICRO_PARAMS["atr_quantile_window"])):
-        raise ValueError("not enough ETH candles for micro signal")
-
-    highs = pd.Series([c.high for c in candles], dtype=float)
-    lows = pd.Series([c.low for c in candles], dtype=float)
-    closes = pd.Series([c.close for c in candles], dtype=float)
-    prev_close = closes.shift(1)
-    true_range = pd.concat([(highs - lows), (highs - prev_close).abs(), (lows - prev_close).abs()], axis=1).max(axis=1)
-    atr = true_range.rolling(int(MICRO_PARAMS["atr_window"])).mean() / closes
-    atr_limit = atr.rolling(int(MICRO_PARAMS["atr_quantile_window"])).quantile(float(MICRO_PARAMS["atr_quantile"]))
-    range_high = highs.shift(1).rolling(int(MICRO_PARAMS["range_window"])).max()
-    range_low = lows.shift(1).rolling(int(MICRO_PARAMS["range_window"])).min()
-    candle = candles[decision_index]
-    compressed = bool(float(atr.iloc[decision_index - 1]) <= float(atr_limit.iloc[decision_index - 1]))
-    long_signal = compressed and candle.close > float(range_high.iloc[decision_index])
-    short_signal = compressed and candle.close < float(range_low.iloc[decision_index])
-    session_ok = micro._session_ok(candle, str(MICRO_PARAMS["session"]))
-    signal = "long" if session_ok and long_signal else "short" if session_ok and short_signal else "no_signal"
-    return {
-        "engine": "micro",
-        "candidate": MICRO_NAME,
-        "symbol": ETH,
-        "bar": BAR,
-        "decision_candle_ts": candle.ts,
-        "decision_candle_time": iso_text(candle.ts),
-        "latest_local_candle_ts": candles[-1].ts,
-        "latest_local_candle_time": iso_text(candles[-1].ts),
-        "signal": signal,
-        "raw_long_signal": bool(long_signal),
-        "raw_short_signal": bool(short_signal),
-        "session_ok": session_ok,
-        "indicators": {
-            "eth_close": candle.close,
-            "atr_previous": float(atr.iloc[decision_index - 1]),
-            "atr_limit_previous": float(atr_limit.iloc[decision_index - 1]),
-            "range_high": float(range_high.iloc[decision_index]),
-            "range_low": float(range_low.iloc[decision_index]),
-        },
-        "params": MICRO_PARAMS,
-    }
-
-
-def build_payload() -> dict[str, object]:
-    switch_state = latest_active_engine()
-    nextgen_payload = nextgen_intent.build_payload()
-    micro_payload = micro_signal()
-    active_engine = str(switch_state["active_engine"])
-    selected_signal = str(nextgen_payload["decision"]["signal"]) if active_engine == "nextgen" else str(micro_payload["signal"])
-    return {
-        "mode": "readonly_signal_intent",
-        "strategy": {
-            "name": TARGET_NAME,
-            "symbol": ETH,
-            "bar": BAR,
-            "direction": "nextgen_long_only_or_micro_observation",
-            "source_report": "reports/eth-exploration/eth-nextgen-micro-portfolio-report.md",
-            "cost_model": "maker_taker",
-            "roundtrip_cost_on_margin": ROUNDTRIP_COST_ON_MARGIN,
-        },
-        "created_at": datetime.now(UTC).isoformat(timespec="seconds").replace("+00:00", "Z"),
-        "submitted_orders": 0,
-        "private_key_required": False,
-        "order_client": None,
-        "switch_state": switch_state,
-        "decision": {
-            "active_engine": active_engine,
-            "selected_signal": selected_signal,
-            "needs_order": False,
-            "needs_cancel": False,
-            "intent": f"observe_{active_engine}_{selected_signal}",
-        },
-        "risk_limits": {
-            "no_order_submission": True,
-            "no_cancel_submission": True,
-            "no_position_state_assumed": True,
-            "execution": "intent_only",
-            "blocked_for_live_trading": True,
-            "blocker": "persistent virtual position state is not maintained by this read-only script",
-        },
-        "nextgen": {
-            "decision": nextgen_payload["decision"],
-            "legs": nextgen_payload["legs"],
-        },
-        "micro": micro_payload,
-    }
-
-
-def markdown_report(payload: dict[str, object]) -> str:
-    lines = [
-        "# ETH nextgen + micro signal intent",
-        "",
-        "Read-only signal intent. No order or cancel request was submitted.",
-        "",
-        "## Decision",
-        "",
-        f"- Created at: `{payload['created_at']}`",
-        f"- Strategy: `{payload['strategy']['name']}`",
-        f"- Active engine: `{payload['decision']['active_engine']}`",
-        f"- Selected signal: `{payload['decision']['selected_signal']}`",
-        f"- Needs order: `{payload['decision']['needs_order']}`",
-        f"- Blocked for live trading: `{payload['risk_limits']['blocked_for_live_trading']}`",
-        f"- Blocker: `{payload['risk_limits']['blocker']}`",
-        "",
-        "## Intent JSON",
-        "",
-        "```json",
-        json.dumps(payload, indent=2, sort_keys=True),
-        "```",
-    ]
-    return "\n".join(lines) + "\n"
+from okx_codex_trader.eth_nextgen_micro import JSON_REPORT, MARKDOWN_REPORT, REPORT_DIR, build_payload, markdown_report
 
 
 def main() -> int:

scripts/run_eth_nextgen_micro_executor.py

@@ -0,0 +1,139 @@
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from dataclasses import asdict
+from datetime import UTC, datetime
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from okx_codex_trader import eth_nextgen_micro
+from okx_codex_trader.config import load_config
+from okx_codex_trader.live_execution import (
+    TargetPosition,
+    current_position_from_okx,
+    load_runtime_state,
+    plan_position_delta,
+    render_market_order_bodies,
+    target_from_signal,
+)
+from okx_codex_trader.okx_client import OkxClient
+
+
+ROOT = Path(__file__).resolve().parents[1]
+STATE_DIR = ROOT / "var" / "eth-nextgen-micro"
+SYMBOL = "ETH-USDT-SWAP"
+LEVERAGE = 3
+
+
+def now_iso() -> str:
+    return datetime.now(UTC).isoformat(timespec="seconds").replace("+00:00", "Z")
+
+
+def unknown_current_position(reason: str) -> TargetPosition:
+    return TargetPosition(side="flat", unit=0.0, known=False, reason=reason)
+
+
+def account_current_position(margin_per_unit_usdt: float) -> tuple[TargetPosition, dict[str, object]]:
+    try:
+        client = OkxClient(load_config())
+    except ValueError as exc:
+        return unknown_current_position(str(exc)), {"account_error": str(exc)}
+    try:
+        positions = client.get_positions(SYMBOL)
+        metadata = client.get_instrument_meta(SYMBOL)
+        mark_price = client.get_last_price(SYMBOL)
+        current = current_position_from_okx(
+            positions=positions,
+            mark_price=mark_price,
+            metadata=metadata,
+            leverage=LEVERAGE,
+            margin_per_unit_usdt=margin_per_unit_usdt,
+        )
+        return current, {
+            "positions": [asdict(position) for position in positions],
+            "instrument_meta": asdict(metadata),
+            "mark_price": mark_price,
+        }
+    except ValueError as exc:
+        return unknown_current_position(str(exc)), {"account_error": str(exc)}
+
+
+def build_snapshot(*, state_dir: Path, margin_per_unit_usdt: float, max_new_margin_usdt: float) -> dict[str, object]:
+    state_dir.mkdir(parents=True, exist_ok=True)
+    payload = eth_nextgen_micro.build_payload()
+    previous_state = load_runtime_state(state_dir / "runtime-state.json")
+    next_state, target = target_from_signal(payload, previous_state)
+    current, account = account_current_position(margin_per_unit_usdt)
+    plan = plan_position_delta(current, target)
+    orders = ()
+    if current.known and target.known:
+        orders = render_market_order_bodies(
+            plan=plan,
+            symbol=SYMBOL,
+            mark_price=float(account["mark_price"]),
+            metadata=client_metadata(account),
+            leverage=LEVERAGE,
+            margin_per_unit_usdt=margin_per_unit_usdt,
+            max_new_margin_usdt=max_new_margin_usdt,
+            client_order_id_prefix=f"ethnm-{target_candle_ts(payload)}",
+        )
+    return {
+        "created_at": now_iso(),
+        "mode": "dry_run_executor",
+        "orders_submitted": 0,
+        "signal": payload["decision"],
+        "execution_intent": payload["execution_intent"],
+        "previous_state": asdict(previous_state),
+        "next_state": asdict(next_state),
+        "current_position": asdict(current),
+        "target_position": asdict(target),
+        "execution_plan": {
+            "current": asdict(plan.current),
+            "target": asdict(plan.target),
+            "actions": [asdict(action) for action in plan.actions],
+        },
+        "rendered_orders": [asdict(order) for order in orders],
+        "account": account,
+        "risk_limits": {
+            "submit_enabled": False,
+            "max_new_margin_usdt": max_new_margin_usdt,
+            "margin_per_unit_usdt": margin_per_unit_usdt,
+            "state_write_required_before_live": True,
+        },
+    }
+
+
+def target_candle_ts(payload: dict[str, object]) -> int:
+    if payload["decision"]["active_engine"] == "nextgen":
+        return int(payload["nextgen"]["data"]["decision_candle_ts"])
+    return int(payload["micro"]["decision_candle_ts"])
+
+
+def client_metadata(account: dict[str, object]):
+    from okx_codex_trader.models import InstrumentMeta
+
+    meta = account["instrument_meta"]
+    return InstrumentMeta(ct_val=float(meta["ct_val"]), lot_sz=float(meta["lot_sz"]), min_sz=float(meta["min_sz"]))
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="Build ETH nextgen+micro live execution dry-run snapshot.")
+    parser.add_argument("--state-dir", type=Path, default=STATE_DIR)
+    parser.add_argument("--margin-per-unit-usdt", type=float, required=True)
+    parser.add_argument("--max-new-margin-usdt", type=float, required=True)
+    args = parser.parse_args()
+
+    snapshot = build_snapshot(
+        state_dir=args.state_dir,
+        margin_per_unit_usdt=args.margin_per_unit_usdt,
+        max_new_margin_usdt=args.max_new_margin_usdt,
+    )
+    print(json.dumps(snapshot, indent=2, sort_keys=True))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/run_eth_nextgen_micro_observer.py

@@ -12,8 +12,9 @@ from pathlib import Path
 sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
 
 from okx_codex_trader.config import load_config
+from okx_codex_trader import eth_nextgen_micro as intent
+from okx_codex_trader.live_execution import TargetPosition, load_runtime_state, plan_position_delta, save_runtime_state, target_from_signal
 from okx_codex_trader.okx_client import OkxClient
-from scripts import build_eth_nextgen_micro_signal_intent as intent
 from scripts import explore_ultrashort as explore
 
 
@@ -86,7 +87,21 @@ def run_once(state_dir: Path) -> dict[str, object]:
     signal_payload = intent.build_payload()
     intent.JSON_REPORT.write_text(json.dumps(signal_payload, indent=2, sort_keys=True) + "\n", encoding="utf-8")
     intent.MARKDOWN_REPORT.write_text(intent.markdown_report(signal_payload), encoding="utf-8")
+    runtime_state_path = state_dir / "runtime-state.json"
+    previous_state = load_runtime_state(runtime_state_path)
+    next_state, target_position = target_from_signal(signal_payload, previous_state)
+    save_runtime_state(runtime_state_path, next_state)
+    current_position = TargetPosition(side="flat", unit=0.0, known=False, reason="OKX current position is not normalized to strategy units")
+    execution_plan = plan_position_delta(current_position, target_position)
     payload["signal"] = signal_payload["decision"]
+    payload["execution_intent"] = signal_payload["execution_intent"]
+    payload["runtime_state"] = asdict(next_state)
+    payload["target_position"] = asdict(target_position)
+    payload["execution_plan"] = {
+        "current": asdict(execution_plan.current),
+        "target": asdict(execution_plan.target),
+        "actions": [asdict(action) for action in execution_plan.actions],
+    }
     payload["switch_state"] = signal_payload["switch_state"]
     payload["risk_limits"] = signal_payload["risk_limits"]
     payload["account"] = private_snapshot()

tests/test_build_eth_nextgen_micro_signal_intent.py

@@ -1,24 +1,10 @@
-import importlib.util
-import sys
-from pathlib import Path
-
 import pandas as pd
 import pytest
 
-
-def load_module():
-    path = Path(__file__).resolve().parents[1] / "scripts" / "build_eth_nextgen_micro_signal_intent.py"
-    spec = importlib.util.spec_from_file_location("build_eth_nextgen_micro_signal_intent", 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
+from okx_codex_trader import eth_nextgen_micro as module
 
 
 def test_latest_active_engine_uses_shifted_prior_day_regime(monkeypatch):
-    module = load_module()
     index = pd.date_range("2026-01-01", periods=32, freq="D", tz="UTC")
     existing = pd.DataFrame({"cost_model": ["maker_taker"], "name": ["equal-2-c0003"], "date": [index[0].strftime("%Y-%m-%d")]})
     nextgen = pd.Series([100.0] * 31 + [80.0], index=index)
@@ -34,7 +20,6 @@ def test_latest_active_engine_uses_shifted_prior_day_regime(monkeypatch):
 
 
 def test_latest_active_engine_switches_on_prior_completed_day(monkeypatch):
-    module = load_module()
     index = pd.date_range("2026-01-01", periods=33, freq="D", tz="UTC")
     existing = pd.DataFrame({"cost_model": ["maker_taker"], "name": ["equal-2-c0003"], "date": [index[0].strftime("%Y-%m-%d")]})
     nextgen = pd.Series([100.0] * 31 + [80.0, 80.0], index=index)
@@ -50,12 +35,11 @@ def test_latest_active_engine_switches_on_prior_completed_day(monkeypatch):
 
 
 def test_payload_is_readonly(monkeypatch):
-    module = load_module()
     monkeypatch.setattr(module, "latest_active_engine", lambda: {"active_engine": "nextgen", "decision_date": "2026-01-01"})
     monkeypatch.setattr(
         module.nextgen_intent,
         "build_payload",
-        lambda: {"decision": {"signal": "no_signal"}, "legs": []},
+        lambda: {"decision": {"signal": "no_signal"}, "data": {"decision_candle_ts": 1000}, "legs": []},
     )
     monkeypatch.setattr(module, "micro_signal", lambda: {"signal": "short"})
 
@@ -66,3 +50,23 @@ def test_payload_is_readonly(monkeypatch):
     assert payload["risk_limits"]["no_order_submission"] is True
     assert payload["risk_limits"]["blocked_for_live_trading"] is True
     assert payload["decision"]["selected_signal"] == "no_signal"
+    assert payload["execution_intent"]["entry_signal"] == "no_signal"
+    assert payload["execution_intent"]["entry_unit"] == 0.0
+    assert payload["execution_intent"]["target_position_known"] is False
+
+
+def test_payload_records_nextgen_entry_unit(monkeypatch):
+    monkeypatch.setattr(module, "latest_active_engine", lambda: {"active_engine": "nextgen", "decision_date": "2026-01-01"})
+    monkeypatch.setattr(
+        module.nextgen_intent,
+        "build_payload",
+        lambda: {"decision": {"signal": "long", "active_suggested_weight": 0.5}, "data": {"decision_candle_ts": 1000}, "legs": []},
+    )
+    monkeypatch.setattr(module, "micro_signal", lambda: {"signal": "no_signal"})
+
+    payload = module.build_payload()
+
+    assert payload["decision"]["selected_signal"] == "long"
+    assert payload["execution_intent"]["entry_signal"] == "long"
+    assert payload["execution_intent"]["entry_unit"] == 0.5
+    assert payload["execution_intent"]["target_position"] is None

tests/test_eth_nextgen_micro_executor.py

@@ -0,0 +1,81 @@
+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)
+
+    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)
+
+    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",
+            },
+        }
+    ]

tests/test_live_execution.py

@@ -0,0 +1,205 @@
+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,
+        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,
+        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,
+            client_order_id_prefix="eth-3000",
+        )

tests/test_okx_client.py

@@ -740,6 +740,61 @@ def test_pending_orders_query_params_are_minimal_params():
     }
 
 
+def test_market_order_body_supports_reduce_only_close():
+    assert OkxClient.build_market_order_body(
+        symbol="ETH-USDT-SWAP",
+        side="sell",
+        pos_side="long",
+        size=2,
+        client_order_id="eth-close-1",
+        reduce_only=True,
+    ) == {
+        "instId": "ETH-USDT-SWAP",
+        "tdMode": "isolated",
+        "side": "sell",
+        "posSide": "long",
+        "ordType": "market",
+        "sz": "2",
+        "clOrdId": "eth-close-1",
+        "reduceOnly": "true",
+    }
+
+
+def test_submit_market_order_body_posts_body_and_returns_order_result():
+    session = DummySession([place_order_response()])
+    client = OkxClient(sample_config(), session=session)
+    body = OkxClient.build_market_order_body(
+        symbol="ETH-USDT-SWAP",
+        side="sell",
+        pos_side="long",
+        size=2,
+        client_order_id="eth-close-1",
+        reduce_only=True,
+    )
+
+    result = client.submit_market_order_body(body)
+
+    assert session.request_paths == ["/api/v5/trade/order"]
+    assert session.last_json_body == body
+    assert result.status == "placed"
+    assert result.order_id == "123"
+    assert result.symbol == "ETH-USDT-SWAP"
+    assert result.side == "sell"
+    assert result.pos_side == "long"
+    assert result.order_type == "market"
+    assert result.size == 2.0
+
+
+def test_submit_market_order_body_rejects_invalid_body_before_okx():
+    session = DummySession([place_order_response()])
+    client = OkxClient(sample_config(), session=session)
+
+    with pytest.raises(ValueError, match="market order body is invalid"):
+        client.submit_market_order_body({"instId": "ETH-USDT-SWAP", "ordType": "limit"})
+
+    assert session.request_paths == []
+
+
 def test_fills_query_params_are_minimal_params():
     assert OkxClient.build_fills_params(symbol="ETH-USDT-SWAP") == {
         "instType": "SWAP",