> ## Documentation Index > Fetch the complete documentation index at: https://jephalabs.mintlify.site/llms.txt > Use this file to discover all available pages before exploring further. # CEX Lane > Centralized exchange executors — withdrawal, status polling, and deposit observation # CEX Lane Executors Bithumb and Backpack are Pro CEX integrations. Lighter is a Pro on-chain venue, not a CEX adapter; it does not use this three-node CEX lane. > **Source files** > > * `src/qtg/infrastructure/executors/cex/__init__.py` > * `src/qtg/infrastructure/executors/cex/withdrawal_action.py` > * `src/qtg/infrastructure/executors/cex/withdrawal_observe.py` > * `src/qtg/infrastructure/executors/cex/deposit_observe.py` > * `src/qtg/infrastructure/executors/cex/common.py` > * `src/qtg/infrastructure/executors/cex/adapters/` *** ## 1. CEX 3-Node Lane Overview Transfers between CEX venues are a serial pipeline composed of three nodes. ```mermaid theme={null} flowchart LR N1["Node 1: withdrawal_action
withdrawal preflight + submission"] N2["Node 2: withdrawal_observe
withdrawal status polling"] N3["Node 3: deposit_observe
deposit status polling"] N1 --> N2 --> N3 N1 -.- R1["provider_refs:
exchange_withdrawal_id
submitted_amount
asset, network
destination_exchange"] N2 -.- R2["provider_refs:
txid
(+ prior refs retained)"] N3 -.- R3["provider_refs:
exchange_deposit_id
(+ prior refs retained)"] ``` ### Executor key mapping | Node | executor\_key | Class | | ---- | ----------------------------- | ------------------------------ | | 1 | `exec.cex.withdrawal_action` | `CexWithdrawalActionExecutor` | | 2 | `exec.cex.withdrawal_observe` | `CexWithdrawalObserveExecutor` | | 3 | `exec.cex.deposit_observe` | `CexDepositObserveExecutor` | ### Registration `register_builtin_cex_executors()` registers all three executors wrapped in `LocalExecutorAdapter`: ```python theme={null} def register_builtin_cex_executors() -> None: register_executor(LocalExecutorAdapter(executor_key="exec.cex.withdrawal_action", handler=CexWithdrawalActionExecutor())) register_executor(LocalExecutorAdapter(executor_key="exec.cex.withdrawal_observe", handler=CexWithdrawalObserveExecutor())) register_executor(LocalExecutorAdapter(executor_key="exec.cex.deposit_observe", handler=CexDepositObserveExecutor())) ``` *** ## 2. Provider Refs Propagation Flow Data is propagated between nodes through the `provider_refs` dict. The previous node's `provider_refs` are passed to the next node as `ctx.provider_context`. ```mermaid theme={null} flowchart TB A["withdrawal_action.submit()
provider_refs:
exchange_withdrawal_id: abc-123
submitted_amount: 25.0
asset: XRP
network: XRP
destination_exchange: upbit"] B["withdrawal_observe.observe()
provider_refs:
exchange_withdrawal_id: abc-123 (retained)
submitted_amount: 25.0 (retained)
txid: ABCDEF1234... (new)"] C["deposit_observe.observe()
provider_refs:
exchange_withdrawal_id: abc-123 (retained)
txid: ABCDEF1234... (retained)
exchange_deposit_id: dep-456 (new)"] A -->|ctx.provider_context| B B -->|ctx.provider_context| C ``` Core propagated fields: | Field | Produced by | Consumed by | | ------------------------ | ------------------- | ------------------------------------- | | `exchange_withdrawal_id` | withdrawal\_action | withdrawal\_observe | | `txid` | withdrawal\_observe | deposit\_observe | | `asset` | withdrawal\_action | withdrawal\_observe, deposit\_observe | | `destination_exchange` | withdrawal\_action | deposit\_observe | | `exchange_deposit_id` | deposit\_observe | (terminal) | *** ## 3. CexWithdrawalActionExecutor **File**: `cex/withdrawal_action.py` Responsible for pre-validating and then submitting a withdrawal request. ### 3.1 preflight This is the most complex preflight. It performs five validations in order: 1. **Whitelist address check**: call `adapter.get_whitelisted_addresses(asset)` and verify the destination address and memo match 2. **Withdrawal availability**: `adapter.check_withdrawal_available(asset, network)` -> verify `is_available` 3. **Minimum / maximum amount**: validate against the `chance.minimum_amount` and `chance.maximum_amount` range 4. **Wallet service state**: `adapter.check_wallet_service(asset)` -> verify `is_maintenance` 5. **Adapter-level validation** (optional): `adapter.validate_withdrawal_request(WithdrawalRequest(...))` -> called only when that method exists on the adapter Returns `None` if all validations pass. Returns `ExecutionResult` with `FAILED` when validation fails. **Error codes (preflight)**: | Error Code | Condition | | ------------------------- | --------------------------------------------------------------- | | `ADDRESS_NOT_WHITELISTED` | Withdrawal address is not on the exchange whitelist | | `WITHDRAWAL_UNAVAILABLE` | The exchange has disabled withdrawals for that asset or network | | `BELOW_MINIMUM` | Amount is below the minimum withdrawal amount | | `ABOVE_MAXIMUM` | Amount exceeds the maximum withdrawal amount | | `WALLET_MAINTENANCE` | Wallet service is under maintenance | | `AUTHENTICATION_ERROR` | Exchange API authentication failed | | `INSUFFICIENT_BALANCE` | Balance is insufficient | | `NETWORK_UNAVAILABLE` | Network is unavailable | | `TRANSFER_ERROR` | Other exchange error | ### 3.2 prepare Packages the withdrawal request into `PreparedAction`. ```python theme={null} provider_request = { "exchange": source_exchange, "asset": asset, "network": network, "amount": str(amount), "address": address, "memo": memo, "destination_exchange": destination_exchange, } ``` `action_type = "cex_withdrawal"`, `signing_required = False` (CEX withdrawals use API-key authentication, so no separate signer is required). `PreparedAction.prepared_action_id` is derived from the payload's canonical JSON hash in the form `cex_withdrawal:{sha256_prefix}`. ### 3.3 submit Calls `adapter.submit_withdrawal(WithdrawalRequest(...))`. `WithdrawalRequest` shape: ```python theme={null} @dataclass(frozen=True) class WithdrawalRequest: asset: str network: str amount: Decimal address: str memo: str | None = None internal_request_id: str | None = None destination_exchange: str | None = None wallet_type: int | None = None travel_rule_questionnaire: dict | None = None ``` `wallet_type` and `travel_rule_questionnaire` are extracted from per-exchange provider options in `ctx.input_params[exchange_name]`. **Returned on success:** * `next_state = "COMPLETED"` * `provider_refs`: `exchange_withdrawal_id`, `submitted_amount`, `asset`, `network`, `destination_exchange` **Error codes (submit)**: | Error Code | Condition | | -------------- | ------------------------------------------------------------------------- | | `TIMEOUT` | `httpx.TimeoutException` / `httpx.ConnectError` -> `next_state = UNKNOWN` | | (adapter code) | `TransferError` -> `next_state = FAILED` | ### 3.4 observe The action node's `observe` is a simple pass-through: ```python theme={null} async def observe(self, ctx): return ExecutionResult(next_state="COMPLETED", provider_refs=dict(ctx.provider_context)) ``` ### 3.5 recover If `exchange_withdrawal_id` exists in `provider_context`, return `COMPLETED`. Otherwise return `FAILED(RECOVERY_FAILED)`. *** ## 4. CexWithdrawalObserveExecutor **File**: `cex/withdrawal_observe.py` Polls exchange withdrawal status after a withdrawal has been submitted. ### 4.1 preflight ```python theme={null} if withdrawal_id(ctx) is None: return ExecutionResult(next_state="FAILED", error_code="MISSING_WITHDRAWAL_ID") return None ``` `withdrawal_id` is extracted from `ctx.provider_context["exchange_withdrawal_id"]`. ### 4.2 prepare Builds the observe action with `action_type = "cex_withdrawal_status"`. ### 4.3 submit Returns `SUBMITTED` immediately. Actual polling happens in `observe`. ### 4.4 observe Core polling logic: ```python theme={null} adapter = get_exchange_adapter(source_exchange(ctx)) status = await adapter.get_withdrawal_status(withdrawal_id, asset=asset) ``` Branching by `WithdrawalState`: | WithdrawalState | next\_state | Description | | --------------------------------------------------- | ----------- | -------------------------------- | | `COMPLETED` | `COMPLETED` | Withdrawal complete. Adds `txid` | | `PROCESSING` / `WAITING_CONFIRMATION` / `SUBMITTED` | `OBSERVING` | Still in progress | | Other (`CANCELLED`, `REJECTED`, `FAILED`) | `FAILED` | `error_code = state` | When `COMPLETED`, `provider_refs` gains `txid`. That `txid` is used by the next node (`deposit_observe`) to match the deposit. ### 4.5 recover Re-calls `observe` directly. **Error codes**: | Error Code | Condition | | ----------------------------------- | -------------------------------------------------------- | | `MISSING_WITHDRAWAL_ID` | `exchange_withdrawal_id` missing from `provider_context` | | `CANCELLED` / `REJECTED` / `FAILED` | Exchange-reported withdrawal state | | (adapter code) | `TransferError` | *** ## 5. CexDepositObserveExecutor **File**: `cex/deposit_observe.py` Confirms settlement of the deposit on the destination exchange. ### 5.1 preflight ```python theme={null} if txid(ctx) is None: return ExecutionResult(next_state="FAILED", error_code="MISSING_TXID") return None ``` `txid` is extracted from `ctx.provider_context["txid"]`. ### 5.2 prepare Builds the observe action with `action_type = "cex_deposit_status"`. ### 5.3 submit Returns `SUBMITTED` immediately. ### 5.4 observe Core deposit matching logic: ```python theme={null} adapter = get_exchange_adapter(destination_exchange(ctx)) deposits = await adapter.list_deposits(asset=asset, limit=50) matched = next( (item for item in deposits if normalize_txid(item.txid) == current_txid), None, ) ``` Key characteristics: * compares `txid` case-insensitively (`normalize_txid` applies `lower()` + `strip()`) * searches the most recent 50 deposits for a matching `txid` * if the matched deposit is in `CREDITED` or `ACCEPTED`, transitions to `COMPLETED` | Condition | next\_state | Description | | -------------------------------------- | ----------- | -------------------------------------------------------------------------------------------------------- | | `txid` match + `CREDITED` / `ACCEPTED` | `COMPLETED` | Adds `exchange_deposit_id` | | `txid` match + `PENDING` | `OBSERVING` | Annotates `deposit_detected_pending: true` + `exchange_deposit_id` in `provider_refs` (travel rule hold) | | No `txid` match or any other state | `OBSERVING` | Keep polling | ### 5.5 recover Re-calls `observe` directly. **Error codes**: | Error Code | Condition | | -------------- | -------------------------------------- | | `MISSING_TXID` | `txid` missing from `provider_context` | | (adapter code) | `TransferError` | *** ## 6. Common Helpers **File**: `cex/common.py` Shared context extraction functions and utilities used by all CEX executors. ### 6.1 Context extraction functions Each function extracts a specific value from `ExecutionContext`. They use fallback priority. | Function | Extraction path (in priority order) | | -------------------------------- | --------------------------------------------------------------------------------------------------------------- | | `source_exchange(ctx)` | `intent.source.exchange` -> `input_params.source_exchange` | | `destination_exchange(ctx)` | `intent.destination.exchange` -> `provider_context.destination_exchange` -> `input_params.destination_exchange` | | `asset(ctx)` | `input_params.asset` -> `intent.asset` -> `provider_context.asset` | | `network(ctx)` | `input_params.network` -> `provider_context.network` | | `amount(ctx)` | `input_params.amount` -> `intent.amount` -> `"0"` | | `address(ctx)` | `input_params.address` -> `intent.destination.address` -> `provider_context.address` | | `memo(ctx)` | `input_params.memo` -> `provider_context.memo` (empty string becomes `None`) | | `withdrawal_id(ctx)` | `provider_context.exchange_withdrawal_id` | | `txid(ctx)` | `provider_context.txid` | | `wallet_type(ctx)` | `input_params[source_exchange].wallet_type` | | `travel_rule_questionnaire(ctx)` | `input_params[source_exchange].travel_rule.questionnaire` | ### 6.2 build\_provider\_request\_action ```python theme={null} def build_provider_request_action(*, action_type: str, payload: dict) -> PreparedAction: ``` Serializes `payload` as canonical JSON, computes its SHA-256 hash, and creates `PreparedAction`. * `payload_format = "provider_request"` * `signing_required = False` * `prepared_action_id = "{action_type}:{sha256_prefix_12}"` (first 12 chars of the hash) ### 6.3 normalize\_txid ```python theme={null} def normalize_txid(value: str | None) -> str | None: # strip() + lower() ``` Normalizes to lowercase because exchanges may vary in `txid` casing. ### 6.4 get\_exchange\_adapter ```python theme={null} def get_exchange_adapter(exchange: str): return get_configured_adapter(exchange) ``` Returns an adapter instance with configured credentials via `runtime.py:get_configured_adapter`. *** ## 7. CEX Adapter Abstraction ### 7.1 TransferAdapterBase **File**: `cex/adapters/base.py` ABC that every exchange adapter must implement: ```python theme={null} class TransferAdapterBase(ABC): exchange_name: str supported_networks: dict[str, list[str]] supports_withdrawal_cancellation: bool = False @abstractmethod async def check_withdrawal_available(self, asset, network) -> WithdrawalChance: ... @abstractmethod async def get_whitelisted_addresses(self, asset=None) -> list[DepositAddress]: ... @abstractmethod async def submit_withdrawal(self, request: WithdrawalRequest) -> WithdrawalResult: ... @abstractmethod async def get_withdrawal_status(self, withdrawal_id, *, asset=None) -> WithdrawalStatus: ... @abstractmethod async def cancel_withdrawal(self, withdrawal_id) -> bool: ... @abstractmethod async def get_deposit_address(self, asset, network) -> DepositAddress: ... @abstractmethod async def get_deposit_status(self, deposit_id) -> DepositStatus: ... @abstractmethod async def list_deposits(self, asset=None, limit=50) -> list[DepositStatus]: ... @abstractmethod async def check_wallet_service(self, asset=None) -> WalletServiceStatus: ... @abstractmethod async def get_withdrawal_fee(self, asset, network, amount) -> Decimal: ... @abstractmethod async def list_withdrawals(self, asset=None, limit=50) -> list[WithdrawalStatus]: ... async def close(self) -> None: ... ``` ### 7.2 Type definitions **File**: `cex/adapters/types.py` | Type | Purpose | | --------------------------- | ------------------------------------------------------------------------------------------------- | | `WithdrawalState` (StrEnum) | `SUBMITTED`, `PROCESSING`, `WAITING_CONFIRMATION`, `COMPLETED`, `CANCELLED`, `REJECTED`, `FAILED` | | `DepositState` (StrEnum) | `PENDING`, `CREDITED`, `ACCEPTED`, `REJECTED` | | `WithdrawalRequest` | Withdrawal request data (`asset`, `network`, `amount`, `address`, `memo`, ...) | | `WithdrawalResult` | Withdrawal submission result (`exchange_withdrawal_id`, `amount`, `fee`, `state`, `created_at`) | | `WithdrawalStatus` | Withdrawal status lookup result (`state`, `txid`, `confirmations`, ...) | | `WithdrawalChance` | Withdrawal availability and limit information (`is_available`, min/max amount, fee, ...) | | `DepositAddress` | Deposit address (`address`, `memo`, `is_whitelisted`) | | `DepositStatus` | Deposit state (`exchange_deposit_id`, `state`, `txid`, `confirmations`) | | `WalletServiceStatus` | Wallet service state (`withdrawal_enabled`, `deposit_enabled`, `is_maintenance`) | ### 7.3 Error hierarchy **File**: `cex/adapters/errors.py` ```mermaid theme={null} flowchart TB BASE["TransferError (base)"] AUTH["AuthenticationError
code: AUTHENTICATION_ERROR
retryable: false"] RATE["RateLimitError
code: RATE_LIMIT
retryable: true"] BAL["InsufficientBalanceError
code: INSUFFICIENT_BALANCE
retryable: false"] ADDR["AddressNotWhitelistedError
code: ADDRESS_NOT_WHITELISTED
retryable: false"] MAINT["WalletMaintenanceError
code: WALLET_MAINTENANCE
retryable: true"] NET["NetworkUnavailableError
code: NETWORK_UNAVAILABLE
retryable: true"] BASE --> AUTH BASE --> RATE BASE --> BAL BASE --> ADDR BASE --> MAINT BASE --> NET ``` All errors carry `code`, `retryable`, `exchange_code`, and `exchange_message`. ### 7.4 Adapter registry **File**: `cex/adapters/registry.py` Registers adapter factories with a decorator pattern: ```python theme={null} _REGISTRY: dict[str, AdapterFactory] = {} def register_adapter(name: str): def decorator(factory: AdapterFactory): _REGISTRY[name] = factory return factory return decorator def get_adapter(name: str, **kwargs) -> TransferAdapterBase: factory = _REGISTRY[name] return factory(**kwargs) ``` ### 7.5 Built-in adapter list | Name | Registration location | Implementation file | | ---------- | ----------------------------------- | ------------------------------ | | `upbit` | `cex/adapters/upbit/__init__.py` | `upbit_transfer_adapter.py` | | `binance` | `cex/adapters/binance/__init__.py` | `binance_transfer_adapter.py` | | `coinbase` | `cex/adapters/coinbase/__init__.py` | `coinbase_transfer_adapter.py` | | `bybit` | `cex/adapters/bybit/__init__.py` | `bybit_transfer_adapter.py` | | `okx` | `cex/adapters/okx/__init__.py` | `okx_transfer_adapter.py` | The Free distribution ships five adapters: `upbit`, `binance`, `bybit`, `coinbase`, `okx`. Bithumb and Backpack are the additional Pro CEX adapters. Lighter is a Pro on-chain venue outside this CEX lane. Each `__init__.py` registers its adapter with the `register_adapter("name")(AdapterClass)` pattern. ### 7.6 Runtime adapter creation **File**: `cex/adapters/runtime.py` ```python theme={null} def get_configured_adapter(exchange: str) -> TransferAdapterBase: ``` Creates the adapter by reading credentials from `settings` according to the exchange name. Binance applies `base_url` and `recv_window_ms`; Bybit applies `base_url` and `recv_window_ms`; Coinbase uses CDP API key (PEM) and Advanced Trade API; OKX applies `base_url` and also requires `passphrase`. Raises `ValueError` when credentials are missing. *** ## 8. Node Config Requirements CEX executors read most settings from `input_params` and `intent` rather than `node_config`. | Parameter | Extraction path | Used by executor | Required | | -------------------------------------- | ---------------------------------------------- | --------------------------------------- | -------- | | `source_exchange` | `intent.source.exchange` / `input_params` | withdrawal\_action, withdrawal\_observe | Required | | `destination_exchange` | `intent.destination.exchange` / `input_params` | deposit\_observe | Required | | `asset` | `input_params` / `intent` | all | Required | | `network` | `input_params` / `provider_context` | withdrawal\_action | Required | | `amount` | `input_params` / `intent` | withdrawal\_action | Required | | `address` | `input_params` / `intent.destination.address` | withdrawal\_action | Required | | `memo` | `input_params` / `provider_context` | withdrawal\_action | Optional | | `{exchange}.wallet_type` | `input_params[exchange]` | withdrawal\_action | Optional | | `{exchange}.travel_rule.questionnaire` | `input_params[exchange]` | withdrawal\_action | Optional | *** ## 9. CI test coverage CEX lane regressions are caught at three levels: | Level | Test | Scope | | ---------------------- | ----------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | | Adapter unit | `tests/qtg/test_{venue}_adapter.py` | Wire payload shape, response parsing, fixture-first against real exchange responses. | | Drill harness e2e | `tests/qtg/test_e2e_cex_drill_smoke.py` | End-to-end `run_cex_live_drill.run_drill()` with fake CEX adapters + fake balance fetcher + real DB. Runs `seed → approve → dispatch → withdraw_observe → deposit_observe → terminal` with shared `_SubmissionLedger` so the destination adapter's `list_deposits` sees what the source adapter submitted. 3 tests: Bybit→Upbit happy, OKX→Upbit KR happy (`walletType=exchange` + `exchId=-1` + 6 KYC fields), Bybit→Upbit no-allowlist negative (dispatch-time `address_allowlist_gate` HALT). | | Drill harness contract | `tests/qtg/test_run_cex_live_drill_smoke_contract.py` | Pins `run_drill` public API, asserts `_run` removal (pre-1.0 clean cut), asserts `_all_nodes_terminal` treats `BLOCKED` as terminal. | Both drill harness files run in the per-PR CI subset (`.github/workflows/test.yml`) against the `postgres:16` service container; combined wall-clock \< 12s. They catch the "drill harness silently broken by unrelated merges" class — see [Drill Harness CI Guard](#10-drill-harness-ci-guard). ## 10. Drill Harness CI Guard Background: between 2026-05-13 (Bybit drill #5) and 2026-05-28 (OKX drill #1), `run_cex_live_drill` was silently broken by 5 separate unrelated merges (network-class isolation, balance-fetcher account\_type, dispatch invariants, V12 state-transition guard). Each broken merge was only detected at operator-driven live-drill time — 15-day undetected drift, 5+1 latent bugs accumulated. The drill harness CI guard closes this gap: every PR now exercises the same orchestration path that drill harness uses, with fake adapters substituted at the `common.get_exchange_adapter` chokepoint. Fake substitution patches all four import sites used by CEX executors: ```python theme={null} qtg.infrastructure.executors.cex.common.get_exchange_adapter qtg.infrastructure.executors.cex.withdrawal_action.common.get_exchange_adapter qtg.infrastructure.executors.cex.withdrawal_observe.common.get_exchange_adapter qtg.infrastructure.executors.cex.deposit_observe.common.get_exchange_adapter ``` Balance fetcher fakes are registered into `balance_fetcher_registry` via in-place dict mutation (not `setattr`) because `qtg.application.services.balance` imports the dict at module load time — the smoke must mutate the shared dict object so the service-layer binding sees the fakes. Bootstrap's `register_balance_fetcher` is monkeypatched to a no-op so live credential-backed fetchers can't overwrite the fakes during `run_drill()`. Out of scope for the smoke: wire-byte adapter correctness (already covered by unit tests), Pro/Bithumb path (Pro test suite is separate), live HTTP behavior. ## 11. Observe intervals From `application/services/observe.py:DEFAULT_OBSERVE_INTERVALS`: | action\_type | Default interval (seconds) | | ----------------------- | -------------------------- | | `cex_withdrawal_status` | 5 | | `cex_deposit_status` | 10 | If an executor returns `retry_after_seconds`, that value takes precedence. *** ## Related documents * [Executor Overview](/reference/executors/overview) * [Observe Probes](/reference/executors/observe-probes) (for on-chain confirmation after `deposit_observe` if needed) * [CCTP Lane](/reference/executors/bridges/cctp-lane) * [Venue Smoke Test](/guide/venue-smoke-test) (operator-driven live smoke; complementary to the PR-time fake-adapter smoke above)