import ccxt import logging from typing import Tuple, Optional import time
logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(name)
class TradingBot: def init(self, api_keys: dict, min_profit_threshold: float = 0.5): """ Initialize trading bot with API keys and minimum profit threshold
Args:
api_keys: Dictionary containing exchange API keys and secrets
min_profit_threshold: Minimum price difference percentage for trade execution
"""
self.api_keys = api_keys
self.min_profit_threshold = min_profit_threshold
self.exchanges = {}
def initialize_exchange(self, exchange_name: str, exchange_class) -> Optional[ccxt.Exchange]:
"""Initialize an exchange with error handling"""
try:
exchange = exchange_class({
"apiKey": self.api_keys[exchange_name][0],
"secret": self.api_keys[exchange_name][1],
"enableRateLimit": True,
"options": {"adjustForTimeDifference": True}
})
exchange.load_markets()
return exchange
except Exception as e:
logger.error(f"Failed to initialize {exchange_name}: {e}")
return None
def get_price(self, exchange: ccxt.Exchange, symbol: str) -> Optional[float]:
"""Fetch current price for a symbol from an exchange"""
try:
ticker = exchange.fetch_ticker(symbol)
price = ticker.get('last')
if price is None:
raise ValueError("Price data not available")
return price
except Exception as e:
logger.error(f"Error fetching price from {exchange.id} for {symbol}: {e}")
return None
def execute_hedged_trade(self, buy_exchange: ccxt.Exchange,
sell_exchange: ccxt.Exchange,
symbol: str,
amount: float) -> Tuple[Optional[dict], Optional[dict]]:
"""Execute simultaneous buy and sell orders"""
try:
logger.info(f"Executing hedged trade: Buy on {buy_exchange.id}, Sell on {sell_exchange.id}")
# Check available balance before trading
buy_balance = buy_exchange.fetch_balance()['USDT']['free']
if buy_balance < amount * self.get_price(buy_exchange, symbol):
raise ValueError(f"Insufficient USDT balance on {buy_exchange.id}")
buy_order = buy_exchange.create_market_buy_order(symbol, amount)
sell_order = sell_exchange.create_market_sell_order(symbol, amount)
logger.info(f"Trade executed - Buy: {buy_order['id']}, Sell: {sell_order['id']}")
return buy_order, sell_order
except Exception as e:
logger.error(f"Trade execution failed: {e}")
return None, None
def calculate_profit_potential(self, price1: float, price2: float) -> float:
"""Calculate potential profit percentage"""
return abs(((price1 - price2) / min(price1, price2)) * 100)
def run(self, symbol: str = "BTC/USDT", amount: float = 0.01):
"""Main trading loop"""
# Initialize exchanges
self.exchanges["binance"] = self.initialize_exchange("binance", ccxt.binance)
self.exchanges["bybit"] = self.initialize_exchange("bybit", ccxt.bybit)
if None in self.exchanges.values():
logger.error("Failed to initialize one or more exchanges")
return
# Main trading logic
binance_price = self.get_price(self.exchanges["binance"], symbol)
bybit_price = self.get_price(self.exchanges["bybit"], symbol)
if None in (binance_price, bybit_price):
logger.error("Failed to fetch prices")
return
logger.info(f"Binance price: {binance_price}, Bybit price: {bybit_price}")
profit_potential = self.calculate_profit_potential(binance_price, bybit_price)
if profit_potential < self.min_profit_threshold:
logger.info(f"Profit potential {profit_potential:.2f}% below threshold {self.min_profit_threshold}%")
return
if binance_price > bybit_price:
self.execute_hedged_trade(self.exchanges["bybit"], self.exchanges["binance"], symbol, amount)
elif bybit_price > binance_price:
self.execute_hedged_trade(self.exchanges["binance"], self.exchanges["bybit"], symbol, amount)
else:
logger.info("No arbitrage opportunity detected")
def main(): api_keys = { "binance": ("your_binance_api_key", "your_binance_api_secret"), "bybit": ("your_bybit_api_key", "your_bybit_api_secret") }
bot = TradingBot(api_keys, min_profit_threshold=0.5)
bot.run(symbol="BTC/USDT", amount=0.01)
if name == "main": main()