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.

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:

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.

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.

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:

@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:

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

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:

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

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:

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

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

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

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:

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

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:

_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

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

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:

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
• Observe Probes (for on-chain confirmation after deposit_observe if needed)
• CCTP Lane
• Venue Smoke Test (operator-driven live smoke; complementary to the PR-time fake-adapter smoke above)