Managing Cross-Exchange Arbitrage Executors with Quote and Gas Conversion
Article Strategy library · Author: hummingbot
Summary
This controller coordinates arbitrage executors for two configured trading venues and a shared base asset. It initializes market-rate sources for each pair, quote-currency conversion, and, for automated market maker venues, the native gas token. Before creating an executor, it converts the configured quote budget into an order amount using the base asset’s conversion rate, quantizes that amount for the buying venue, and passes a minimum profitability threshold and gas conversion price into the executor configuration.
The controller considers both directions: buy on the first venue and sell on the second, then reverse. It limits new activity based on active executor counts, the imbalance between completed directions, and a delay after closures. These are coordination and launch rules, not a complete arbitrage execution or profitability model. The source shows example venue and pair defaults, but reports no trading results. It does not establish how the executor handles fees, slippage, latency, inventory, or failed cross-venue legs; those factors can determine whether an apparent price spread is realizable. Gas-token and exchange-rate lookups can also fail, in which case executor creation may be skipped or lack a gas conversion price.
Key ideas
- The controller configures two venue pairs and can launch arbitrage executors in either buy-sell direction.
- It converts a quote-denominated budget into a quantized order amount using a base-asset exchange rate.
- For automated market maker venues, it attempts to obtain a native gas token and conversion price.
- Executor creation is gated by directional activity, completed-trade imbalance, and a post-closure delay.
- The code provides no performance evidence or full accounting for execution costs and cross-venue leg risk.
Tags
Full text
# ArbitrageController
# ArbitrageController
## Source (Apache-2.0)
```python
from decimal import Decimal
from typing import List, Optional
import pandas as pd
from hummingbot.client.ui.interface_utils import format_df_for_printout
from hummingbot.core.data_type.common import MarketDict
from hummingbot.core.gateway.gateway_http_client import GatewayHttpClient
from hummingbot.strategy_v2.controllers.controller_base import ControllerBase, ControllerConfigBase
from hummingbot.strategy_v2.executors.arbitrage_executor.data_types import ArbitrageExecutorConfig
from hummingbot.strategy_v2.executors.data_types import ConnectorPair
from hummingbot.strategy_v2.models.base import RunnableStatus
from hummingbot.strategy_v2.models.executor_actions import CreateExecutorAction, ExecutorAction
class ArbitrageControllerConfig(ControllerConfigBase):
controller_name: str = "arbitrage_controller"
exchange_pair_1: ConnectorPair = ConnectorPair(connector_name="binance", trading_pair="SOL-USDT")
exchange_pair_2: ConnectorPair = ConnectorPair(connector_name="jupiter/router", trading_pair="SOL-USDC")
min_profitability: Decimal = Decimal("0.01")
delay_between_executors: int = 10 # in seconds
max_executors_imbalance: int = 1
rate_connector: str = "binance"
quote_conversion_asset: str = "USDT"
def update_markets(self, markets: MarketDict) -> MarketDict:
return [markets.add_or_update(cp.connector_name, cp.trading_pair) for cp in [self.exchange_pair_1, self.exchange_pair_2]][-1]
class ArbitrageController(ControllerBase):
def __init__(self, config: ArbitrageControllerConfig, *args, **kwargs):
self.config = config
super().__init__(config, *args, **kwargs)
self._imbalance = 0
self._last_buy_closed_timestamp = 0
self._last_sell_closed_timestamp = 0
self._len_active_buy_arbitrages = 0
self._len_active_sell_arbitrages = 0
self.base_asset = self.config.exchange_pair_1.trading_pair.split("-")[0]
self._gas_token_cache = {} # Cache for gas tokens by connector
self._initialize_gas_tokens() # Fetch gas tokens during init
self.initialize_rate_sources()
def initialize_rate_sources(self):
rates_required = []
for connector_pair in [self.config.exchange_pair_1, self.config.exchange_pair_2]:
base, quote = connector_pair.trading_pair.split("-")
# Add rate source for gas token if it's an AMM connector
if connector_pair.is_amm_connector():
gas_token = self.get_gas_token(connector_pair.connector_name)
if gas_token and gas_token != quote:
rates_required.append(ConnectorPair(connector_name=self.config.rate_connector,
trading_pair=f"{gas_token}-{quote}"))
# Add rate source for quote conversion asset
if quote != self.config.quote_conversion_asset:
rates_required.append(ConnectorPair(connector_name=self.config.rate_connector,
trading_pair=f"{quote}-{self.config.quote_conversion_asset}"))
# Add rate source for trading pairs
rates_required.append(ConnectorPair(connector_name=connector_pair.connector_name,
trading_pair=connector_pair.trading_pair))
if len(rates_required) > 0:
self.market_data_provider.initialize_rate_sources(rates_required)
def _initialize_gas_tokens(self):
"""Initialize gas tokens for AMM connectors during controller initialization."""
import asyncio
async def fetch_gas_tokens():
for connector_pair in [self.config.exchange_pair_1, self.config.exchange_pair_2]:
if connector_pair.is_amm_connector():
connector_name = connector_pair.connector_name
if connector_name not in self._gas_token_cache:
try:
gateway_client = GatewayHttpClient.get_instance()
# Get chain and network for the connector
chain, network, error = await gateway_client.get_connector_chain_network(
connector_name
)
if error:
self.logger().warning(f"Failed to get chain info for {connector_name}: {error}")
continue
# Get native currency symbol
native_currency = await gateway_client.get_native_currency_symbol(chain, network)
if native_currency:
self._gas_token_cache[connector_name] = native_currency
self.logger().info(f"Gas token for {connector_name}: {native_currency}")
else:
self.logger().warning(f"Failed to get native currency for {connector_name}")
except Exception as e:
self.logger().error(f"Error getting gas token for {connector_name}: {e}")
# Run the async function to fetch gas tokens
loop = asyncio.get_event_loop()
if loop.is_running():
asyncio.create_task(fetch_gas_tokens())
else:
loop.run_until_complete(fetch_gas_tokens())
def get_gas_token(self, connector_name: str) -> Optional[str]:
"""Get the cached gas token for a connector."""
return self._gas_token_cache.get(connector_name)
async def update_processed_data(self):
pass
def determine_executor_actions(self) -> List[ExecutorAction]:
self.update_arbitrage_stats()
executor_actions = []
current_time = self.market_data_provider.time()
if (abs(self._imbalance) >= self.config.max_executors_imbalance or
self._last_buy_closed_timestamp + self.config.delay_between_executors > current_time or
self._last_sell_closed_timestamp + self.config.delay_between_executors > current_time):
return executor_actions
if self._len_active_buy_arbitrages == 0:
executor_actions.append(self.create_arbitrage_executor_action(self.config.exchange_pair_1,
self.config.exchange_pair_2))
if self._len_active_sell_arbitrages == 0:
executor_actions.append(self.create_arbitrage_executor_action(self.config.exchange_pair_2,
self.config.exchange_pair_1))
return [action for action in executor_actions if action is not None]
def create_arbitrage_executor_action(self, buying_exchange_pair: ConnectorPair,
selling_exchange_pair: ConnectorPair):
try:
if buying_exchange_pair.is_amm_connector():
gas_token = self.get_gas_token(buying_exchange_pair.connector_name)
if gas_token:
pair = buying_exchange_pair.trading_pair.split("-")[0] + "-" + gas_token
gas_conversion_price = self.market_data_provider.get_rate(pair)
else:
gas_conversion_price = None
elif selling_exchange_pair.is_amm_connector():
gas_token = self.get_gas_token(selling_exchange_pair.connector_name)
if gas_token:
pair = selling_exchange_pair.trading_pair.split("-")[0] + "-" + gas_token
gas_conversion_price = self.market_data_provider.get_rate(pair)
else:
gas_conversion_price = None
else:
gas_conversion_price = None
rate = self.market_data_provider.get_rate(self.base_asset + "-" + self.config.quote_conversion_asset)
if not rate:
self.logger().warning(
f"Cannot get conversion rate for {self.base_asset}-{self.config.quote_conversion_asset}. "
f"Skipping executor creation.")
return None
amount_quantized = self.market_data_provider.quantize_order_amount(
buying_exchange_pair.connector_name, buying_exchange_pair.trading_pair,
self.config.total_amount_quote / rate)
arbitrage_config = ArbitrageExecutorConfig(
timestamp=self.market_data_provider.time(),
buying_market=buying_exchange_pair,
selling_market=selling_exchange_pair,
order_amount=amount_quantized,
min_profitability=self.config.min_profitability,
gas_conversion_price=gas_conversion_price,
)
return CreateExecutorAction(
executor_config=arbitrage_config,
controller_id=self.config.id)
except Exception as e:
self.logger().error(
f"Error creating executor to buy on {buying_exchange_pair.connector_name} and sell on {selling_exchange_pair.connector_name}, {e}")
def update_arbitrage_stats(self):
closed_executors = [e for e in self.executors_info if e.status == RunnableStatus.TERMINATED]
active_executors = [e for e in self.executors_info if e.status != RunnableStatus.TERMINATED]
buy_arbitrages = [arbitrage for arbitrage in closed_executors if
arbitrage.config.buying_market == self.config.exchange_pair_1]
sell_arbitrages = [arbitrage for arbitrage in closed_executors if
arbitrage.config.buying_market == self.config.exchange_pair_2]
self._imbalance = len(buy_arbitrages) - len(sell_arbitrages)
self._last_buy_closed_timestamp = max([arbitrage.close_timestamp for arbitrage in buy_arbitrages]) if len(
buy_arbitrages) > 0 else 0
self._last_sell_closed_timestamp = max([arbitrage.close_timestamp for arbitrage in sell_arbitrages]) if len(
sell_arbitrages) > 0 else 0
self._len_active_buy_arbitrages = len([arbitrage for arbitrage in active_executors if
arbitrage.config.buying_market == self.config.exchange_pair_1])
self._len_active_sell_arbitrages = len([arbitrage for arbitrage in active_executors if
arbitrage.config.buying_market == self.config.exchange_pair_2])
def to_format_status(self) -> List[str]:
all_executors_custom_info = pd.DataFrame(e.custom_info for e in self.executors_info)
return [format_df_for_printout(all_executors_custom_info, table_format="psql", )]
```Shown in full with attribution under the source's licence. Licence: Apache-2.0
This summary was written by Stratmill's research agent from the original; it is not a copy of the source.