164 lines
6.0 KiB
Python
164 lines
6.0 KiB
Python
"""
|
|
Exchange Factory - Creates exchange interfaces based on configuration
|
|
"""
|
|
import os
|
|
import logging
|
|
from typing import Dict, Any, Optional
|
|
from .exchange_interface import ExchangeInterface
|
|
from .mexc_interface import MEXCInterface
|
|
from .binance_interface import BinanceInterface
|
|
from .deribit_interface import DeribitInterface
|
|
from .bybit_interface import BybitInterface
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class ExchangeFactory:
|
|
"""Factory class for creating exchange interfaces"""
|
|
|
|
SUPPORTED_EXCHANGES = {
|
|
'mexc': MEXCInterface,
|
|
'binance': BinanceInterface,
|
|
'deribit': DeribitInterface,
|
|
'bybit': BybitInterface
|
|
}
|
|
|
|
@classmethod
|
|
def create_exchange(cls, exchange_name: str, config: Dict[str, Any]) -> Optional[ExchangeInterface]:
|
|
"""Create an exchange interface based on the name and configuration.
|
|
|
|
Args:
|
|
exchange_name: Name of the exchange ('mexc', 'deribit', 'binance')
|
|
config: Configuration dictionary for the exchange
|
|
|
|
Returns:
|
|
Configured exchange interface or None if creation fails
|
|
"""
|
|
exchange_name = exchange_name.lower()
|
|
|
|
if exchange_name not in cls.SUPPORTED_EXCHANGES:
|
|
logger.error(f"Unsupported exchange: {exchange_name}")
|
|
return None
|
|
|
|
try:
|
|
# Get API credentials from environment variables
|
|
api_key, api_secret = cls._get_credentials(exchange_name)
|
|
|
|
# Get exchange-specific configuration
|
|
test_mode = config.get('test_mode', True)
|
|
trading_mode = config.get('trading_mode', 'simulation')
|
|
|
|
# Create exchange interface
|
|
exchange_class = cls.SUPPORTED_EXCHANGES[exchange_name]
|
|
|
|
if exchange_name == 'mexc':
|
|
exchange = exchange_class(
|
|
api_key=api_key,
|
|
api_secret=api_secret,
|
|
test_mode=test_mode,
|
|
trading_mode=trading_mode
|
|
)
|
|
elif exchange_name == 'deribit':
|
|
exchange = exchange_class(
|
|
api_key=api_key,
|
|
api_secret=api_secret,
|
|
test_mode=test_mode
|
|
)
|
|
elif exchange_name == 'bybit':
|
|
exchange = exchange_class(
|
|
api_key=api_key,
|
|
api_secret=api_secret,
|
|
test_mode=test_mode
|
|
)
|
|
else: # binance and others
|
|
exchange = exchange_class(
|
|
api_key=api_key,
|
|
api_secret=api_secret,
|
|
test_mode=test_mode
|
|
)
|
|
|
|
# Test connection
|
|
if exchange.connect():
|
|
logger.info(f"Successfully created and connected to {exchange_name} exchange")
|
|
return exchange
|
|
else:
|
|
logger.error(f"Failed to connect to {exchange_name} exchange")
|
|
return None
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error creating {exchange_name} exchange: {e}")
|
|
return None
|
|
|
|
@classmethod
|
|
def _get_credentials(cls, exchange_name: str) -> tuple[str, str]:
|
|
"""Get API credentials from environment variables.
|
|
|
|
Args:
|
|
exchange_name: Name of the exchange
|
|
|
|
Returns:
|
|
Tuple of (api_key, api_secret)
|
|
"""
|
|
if exchange_name == 'mexc':
|
|
api_key = os.getenv('MEXC_API_KEY', '')
|
|
api_secret = os.getenv('MEXC_SECRET_KEY', '')
|
|
elif exchange_name == 'deribit':
|
|
api_key = os.getenv('DERIBIT_API_CLIENTID', '')
|
|
api_secret = os.getenv('DERIBIT_API_SECRET', '')
|
|
elif exchange_name == 'binance':
|
|
api_key = os.getenv('BINANCE_API_KEY', '')
|
|
api_secret = os.getenv('BINANCE_SECRET_KEY', '')
|
|
elif exchange_name == 'bybit':
|
|
api_key = os.getenv('BYBIT_API_KEY', '')
|
|
api_secret = os.getenv('BYBIT_API_SECRET', '')
|
|
else:
|
|
logger.warning(f"Unknown exchange credentials for {exchange_name}")
|
|
api_key = api_secret = ''
|
|
|
|
return api_key, api_secret
|
|
|
|
@classmethod
|
|
def create_multiple_exchanges(cls, exchanges_config: Dict[str, Any]) -> Dict[str, ExchangeInterface]:
|
|
"""Create multiple exchange interfaces from configuration.
|
|
|
|
Args:
|
|
exchanges_config: Configuration dictionary with exchange settings
|
|
|
|
Returns:
|
|
Dictionary mapping exchange names to their interfaces
|
|
"""
|
|
exchanges = {}
|
|
|
|
for exchange_name, config in exchanges_config.items():
|
|
if exchange_name == 'primary':
|
|
continue # Skip the primary exchange indicator
|
|
|
|
if config.get('enabled', False):
|
|
exchange = cls.create_exchange(exchange_name, config)
|
|
if exchange:
|
|
exchanges[exchange_name] = exchange
|
|
else:
|
|
logger.warning(f"Failed to create {exchange_name} exchange, skipping")
|
|
else:
|
|
logger.info(f"Exchange {exchange_name} is disabled, skipping")
|
|
|
|
return exchanges
|
|
|
|
@classmethod
|
|
def get_primary_exchange(cls, exchanges_config: Dict[str, Any]) -> Optional[ExchangeInterface]:
|
|
"""Get the primary exchange interface.
|
|
|
|
Args:
|
|
exchanges_config: Configuration dictionary with exchange settings
|
|
|
|
Returns:
|
|
Primary exchange interface or None
|
|
"""
|
|
primary_name = exchanges_config.get('primary', 'deribit')
|
|
primary_config = exchanges_config.get(primary_name, {})
|
|
|
|
if not primary_config.get('enabled', False):
|
|
logger.error(f"Primary exchange {primary_name} is not enabled")
|
|
return None
|
|
|
|
return cls.create_exchange(primary_name, primary_config) |