""" Trading Agent Test Script This script demonstrates how to use the swappable exchange modules to connect to and interact with different cryptocurrency exchanges. Usage: python -m NN.exchanges.trading_agent_test --exchange binance --test-mode """ import argparse import logging import os import sys import time # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler("exchange_test.log"), logging.StreamHandler() ] ) logger = logging.getLogger("exchange_test") # Import exchange interfaces try: from .exchange_interface import ExchangeInterface from .binance_interface import BinanceInterface from .mexc_interface import MEXCInterface except ImportError: # When running as standalone script from exchange_interface import ExchangeInterface from binance_interface import BinanceInterface from mexc_interface import MEXCInterface def create_exchange(exchange_name: str, api_key: str = None, api_secret: str = None, test_mode: bool = True) -> ExchangeInterface: """Create an exchange interface instance. Args: exchange_name: Name of the exchange ('binance' or 'mexc') api_key: API key for the exchange api_secret: API secret for the exchange test_mode: If True, use test/sandbox environment Returns: ExchangeInterface: The exchange interface instance """ exchange_name = exchange_name.lower() if exchange_name == 'binance': return BinanceInterface(api_key, api_secret, test_mode) elif exchange_name == 'mexc': return MEXCInterface(api_key, api_secret, test_mode) else: raise ValueError(f"Unsupported exchange: {exchange_name}. Supported exchanges: binance, mexc") def test_exchange(exchange: ExchangeInterface, symbols: list = None): """Test the exchange interface. Args: exchange: Exchange interface instance symbols: List of symbols to test with (e.g., ['BTC/USDT', 'ETH/USDT']) """ if symbols is None: symbols = ['BTC/USDT', 'ETH/USDT'] # Test connection logger.info(f"Testing connection to exchange...") connected = exchange.connect() if not connected and hasattr(exchange, 'api_key') and exchange.api_key: logger.error("Failed to connect to exchange. Make sure your API credentials are correct.") return False elif not connected: logger.warning("Running in read-only mode without API credentials.") else: logger.info("Connection successful with API credentials!") # Test getting ticker data ticker_success = True for symbol in symbols: try: logger.info(f"Getting ticker data for {symbol}...") ticker = exchange.get_ticker(symbol) logger.info(f"Ticker for {symbol}: Last price: {ticker['last']}, Volume: {ticker['volume']}") except Exception as e: logger.error(f"Error getting ticker for {symbol}: {str(e)}") ticker_success = False if not ticker_success: logger.error("Failed to get ticker data. Exchange interface test failed.") return False # Test getting account balances if API keys are provided if hasattr(exchange, 'api_key') and exchange.api_key: logger.info("Testing account balance retrieval...") try: for base_asset in ['BTC', 'ETH', 'USDT']: balance = exchange.get_balance(base_asset) logger.info(f"Balance for {base_asset}: {balance}") except Exception as e: logger.error(f"Error getting account balances: {str(e)}") logger.warning("Balance retrieval failed, but this is not critical if ticker data works.") else: logger.warning("API keys not provided. Skipping balance checks.") logger.info("Exchange interface test completed successfully in read-only mode.") return True def execute_test_trades(exchange: ExchangeInterface, symbol: str, test_trade_amount: float = 0.001): """Execute test trades. Args: exchange: Exchange interface instance symbol: Symbol to trade (e.g., 'BTC/USDT') test_trade_amount: Amount to use for test trades """ if not hasattr(exchange, 'api_key') or not exchange.api_key: logger.warning("API keys not provided. Skipping test trades.") return logger.info(f"Executing test trades for {symbol} with amount {test_trade_amount}...") # Get current ticker for the symbol try: ticker = exchange.get_ticker(symbol) logger.info(f"Current price for {symbol}: {ticker['last']}") except Exception as e: logger.error(f"Error getting ticker for {symbol}: {str(e)}") return # Execute a buy order try: logger.info(f"Placing a test BUY order for {test_trade_amount} {symbol}...") buy_order = exchange.execute_trade(symbol, 'BUY', quantity=test_trade_amount) if buy_order: logger.info(f"BUY order executed: {buy_order}") order_id = buy_order.get('orderId') # Get order status if order_id: time.sleep(2) # Wait for order to process status = exchange.get_order_status(symbol, order_id) logger.info(f"Order status: {status}") else: logger.error("BUY order failed.") except Exception as e: logger.error(f"Error executing BUY order: {str(e)}") # Wait before selling time.sleep(5) # Execute a sell order try: logger.info(f"Placing a test SELL order for {test_trade_amount} {symbol}...") sell_order = exchange.execute_trade(symbol, 'SELL', quantity=test_trade_amount) if sell_order: logger.info(f"SELL order executed: {sell_order}") order_id = sell_order.get('orderId') # Get order status if order_id: time.sleep(2) # Wait for order to process status = exchange.get_order_status(symbol, order_id) logger.info(f"Order status: {status}") else: logger.error("SELL order failed.") except Exception as e: logger.error(f"Error executing SELL order: {str(e)}") # Get open orders try: logger.info("Getting open orders...") open_orders = exchange.get_open_orders(symbol) if open_orders: logger.info(f"Open orders: {open_orders}") # Cancel any open orders for order in open_orders: order_id = order.get('orderId') if order_id: logger.info(f"Cancelling order {order_id}...") cancelled = exchange.cancel_order(symbol, order_id) logger.info(f"Order cancelled: {cancelled}") else: logger.info("No open orders.") except Exception as e: logger.error(f"Error getting/cancelling open orders: {str(e)}") def main(): """Main function for testing exchange interfaces.""" # Parse command-line arguments parser = argparse.ArgumentParser(description="Test exchange interfaces") parser.add_argument('--exchange', type=str, default='binance', choices=['binance', 'mexc'], help='Exchange to test') parser.add_argument('--api-key', type=str, default=None, help='API key for the exchange') parser.add_argument('--api-secret', type=str, default=None, help='API secret for the exchange') parser.add_argument('--test-mode', action='store_true', help='Use test/sandbox environment') parser.add_argument('--symbols', nargs='+', default=['BTC/USDT', 'ETH/USDT'], help='Symbols to test with') parser.add_argument('--execute-trades', action='store_true', help='Execute test trades (use with caution!)') parser.add_argument('--test-trade-amount', type=float, default=0.001, help='Amount to use for test trades') args = parser.parse_args() # Use environment variables for API keys if not provided api_key = args.api_key or os.environ.get(f"{args.exchange.upper()}_API_KEY") api_secret = args.api_secret or os.environ.get(f"{args.exchange.upper()}_API_SECRET") # Create exchange interface try: exchange = create_exchange( exchange_name=args.exchange, api_key=api_key, api_secret=api_secret, test_mode=args.test_mode ) logger.info(f"Created {args.exchange} exchange interface") logger.info(f"Test mode: {args.test_mode}") # Test exchange if test_exchange(exchange, args.symbols): logger.info("Exchange interface test passed!") # Execute test trades if requested if args.execute_trades: logger.warning("Executing test trades. This will use real funds!") execute_test_trades( exchange=exchange, symbol=args.symbols[0], test_trade_amount=args.test_trade_amount ) else: logger.error("Exchange interface test failed!") except Exception as e: logger.error(f"Error testing exchange interface: {str(e)}") import traceback logger.error(traceback.format_exc()) return 1 return 0 if __name__ == "__main__": sys.exit(main())