popov/gogo2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

bybit REST api

Browse Source
This commit is contained in:
Dobromir Popov
2025-07-14 22:57:02 +03:00
parent ee2e6478d8
commit 02804ee64f
4 changed files with 476 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

197
NN/exchanges/bybit_interface.py
View File

@@ -1,3 +1,9 @@
"""
Bybit Interface
"""
import logging
import time
from typing import Dict, Any, List, Optional, Tuple
@@ -12,6 +18,7 @@ except ImportError:
logging.warning("pybit not installed. Run: pip install pybit")
from .exchange_interface import ExchangeInterface
from .bybit_rest_client import BybitRestClient
logger = logging.getLogger(__name__)
@@ -35,8 +42,15 @@ class BybitInterface(ExchangeInterface):
# Bybit-specific settings
self.session = None
self.rest_client = None # Raw REST client fallback
self.category = "linear" # Default to USDT perpetuals
self.supported_symbols = set()
self.use_fallback = False # Track if we should use REST client
# Caching to reduce API calls and avoid rate limiting
self._open_orders_cache = {}
self._open_orders_cache_time = 0
self._cache_timeout = 5 # 5 seconds cache timeout
# Load credentials from environment if not provided
if not api_key:
@@ -61,22 +75,43 @@ class BybitInterface(ExchangeInterface):
logger.error("API key and secret required for Bybit connection")
return False
# Create HTTP session
# Initialize pybit session
self.session = HTTP(
testnet=self.test_mode,
api_key=self.api_key,
api_secret=self.api_secret,
)
# Test connection by getting account info
account_info = self.session.get_wallet_balance(accountType="UNIFIED")
if account_info.get('retCode') == 0:
logger.info(f"Successfully connected to Bybit (testnet: {self.test_mode})")
self._load_instruments()
return True
else:
logger.error(f"Failed to connect to Bybit: {account_info}")
return False
# Initialize raw REST client as fallback
self.rest_client = BybitRestClient(
api_key=self.api_key,
api_secret=self.api_secret,
testnet=self.test_mode
)
# Test pybit connection first
try:
account_info = self.session.get_wallet_balance(accountType="UNIFIED")
if account_info.get('retCode') == 0:
logger.info(f"Successfully connected to Bybit via pybit (testnet: {self.test_mode})")
self.use_fallback = False
self._load_instruments()
return True
else:
logger.warning(f"pybit connection failed: {account_info}")
raise Exception("pybit connection failed")
except Exception as e:
logger.warning(f"pybit failed, trying REST client fallback: {e}")
# Test REST client fallback
if self.rest_client.test_connectivity() and self.rest_client.test_authentication():
logger.info(f"Successfully connected to Bybit via REST client fallback (testnet: {self.test_mode})")
self.use_fallback = True
self._load_instruments()
return True
else:
logger.error("Both pybit and REST client failed")
return False
except Exception as e:
logger.error(f"Error connecting to Bybit: {e}")
@@ -164,6 +199,43 @@ class BybitInterface(ExchangeInterface):
logger.error(f"Error getting account summary: {e}")
return {}
def get_all_balances(self) -> Dict[str, Dict[str, float]]:
"""Get all account balances in the format expected by trading executor.
Returns:
Dictionary with asset balances in format: {asset: {'free': float, 'locked': float}}
"""
try:
account_info = self.session.get_wallet_balance(accountType="UNIFIED")
if account_info.get('retCode') == 0:
balances = {}
accounts = account_info.get('result', {}).get('list', [])
for account in accounts:
coins = account.get('coin', [])
for coin in coins:
asset = coin.get('coin', '')
if asset:
# Convert Bybit balance format to MEXC-compatible format
available = float(coin.get('availableToWithdraw', 0))
locked = float(coin.get('locked', 0))
balances[asset] = {
'free': available,
'locked': locked,
'total': available + locked
}
logger.debug(f"Retrieved balances for {len(balances)} assets")
return balances
else:
logger.error(f"Failed to get all balances: {account_info}")
return {}
except Exception as e:
logger.error(f"Error getting all balances: {e}")
return {}
def get_ticker(self, symbol: str) -> Dict[str, Any]:
"""Get ticker information for a symbol.
@@ -269,6 +341,29 @@ class BybitInterface(ExchangeInterface):
logger.error(f"Error placing order: {e}")
return {'error': str(e)}
def _process_pybit_orders(self, orders_list: List[Dict]) -> List[Dict[str, Any]]:
"""Process orders from pybit response format."""
open_orders = []
for order in orders_list:
order_info = {
'order_id': order.get('orderId'),
'symbol': order.get('symbol'),
'side': order.get('side', '').lower(),
'type': order.get('orderType', '').lower(),
'quantity': float(order.get('qty', 0)),
'filled_quantity': float(order.get('cumExecQty', 0)),
'price': float(order.get('price', 0)),
'status': self._map_order_status(order.get('orderStatus', '')),
'timestamp': int(order.get('createdTime', 0))
}
open_orders.append(order_info)
return open_orders
def _process_rest_orders(self, orders_list: List[Dict]) -> List[Dict[str, Any]]:
"""Process orders from REST client response format."""
# REST client returns same format as pybit, so we can reuse the method
return self._process_pybit_orders(orders_list)
def cancel_order(self, symbol: str, order_id: str) -> bool:
"""Cancel an order.
@@ -380,7 +475,7 @@ class BybitInterface(ExchangeInterface):
return {}
def get_open_orders(self, symbol: str = None) -> List[Dict[str, Any]]:
"""Get open orders.
"""Get open orders with caching and fallback to REST client.
Args:
symbol: Trading symbol (optional, gets all if None)
@@ -389,38 +484,64 @@ class BybitInterface(ExchangeInterface):
List of open order dictionaries
"""
try:
params = {
'category': self.category,
'openOnly': True
}
import time
current_time = time.time()
cache_key = symbol or 'all'
if symbol:
params['symbol'] = self._format_symbol(symbol)
# Check if we have fresh cached data
if (cache_key in self._open_orders_cache and
current_time - self._open_orders_cache_time < self._cache_timeout):
logger.debug(f"Returning cached open orders for {cache_key}")
return self._open_orders_cache[cache_key]
response = self.session.get_open_orders(**params)
if response.get('retCode') == 0:
orders = response.get('result', {}).get('list', [])
open_orders = []
for order in orders:
order_info = {
'order_id': order.get('orderId'),
'symbol': order.get('symbol'),
'side': order.get('side', '').lower(),
'type': order.get('orderType', '').lower(),
'quantity': float(order.get('qty', 0)),
'filled_quantity': float(order.get('cumExecQty', 0)),
'price': float(order.get('price', 0)),
'status': self._map_order_status(order.get('orderStatus', '')),
'timestamp': int(order.get('createdTime', 0))
# Try pybit first if not using fallback
if not self.use_fallback and self.session:
try:
params = {
'category': self.category,
'openOnly': True
}
open_orders.append(order_info)
if symbol:
params['symbol'] = self._format_symbol(symbol)
response = self.session.get_open_orders(**params)
# Process pybit response
if response.get('retCode') == 0:
orders = self._process_pybit_orders(response.get('result', {}).get('list', []))
# Cache the result
self._open_orders_cache[cache_key] = orders
self._open_orders_cache_time = current_time
logger.debug(f"Found {len(orders)} open orders via pybit, cached for {self._cache_timeout}s")
return orders
else:
logger.warning(f"pybit get_open_orders failed: {response}")
raise Exception("pybit failed")
except Exception as e:
error_str = str(e)
if "10016" in error_str or "System error" in error_str:
logger.warning(f"pybit rate limited (Error 10016), switching to REST fallback: {e}")
self.use_fallback = True
else:
logger.warning(f"pybit get_open_orders error, trying REST fallback: {e}")
# Use REST client (either as primary or fallback)
if self.rest_client:
formatted_symbol = self._format_symbol(symbol) if symbol else None
response = self.rest_client.get_open_orders(self.category, formatted_symbol)
logger.debug(f"Found {len(open_orders)} open orders")
return open_orders
orders = self._process_rest_orders(response.get('result', {}).get('list', []))
# Cache the result
self._open_orders_cache[cache_key] = orders
self._open_orders_cache_time = current_time
logger.debug(f"Found {len(orders)} open orders via REST client, cached for {self._cache_timeout}s")
return orders
else:
logger.error(f"Failed to get open orders: {response}")
logger.error("No available API client (pybit or REST)")
return []
except Exception as e: