import unittest from unittest.mock import MagicMock, patch import ccxt from datetime import datetime import pandas as pd from Exchange import Exchange class TestExchange(unittest.TestCase): @patch('ccxt.binance') def setUp(self, mock_exchange): self.mock_client = MagicMock() mock_exchange.return_value = self.mock_client self.api_keys = {'key': 'test_key', 'secret': 'test_secret'} self.mock_client.fetch_markets.return_value = [ {'symbol': 'BTC/USDT', 'active': True}, {'symbol': 'ETH/USDT', 'active': True}, ] self.mock_client.load_markets.return_value = { 'BTC/USDT': { 'limits': { 'amount': {'min': 0.001}, 'cost': {'min': 10.0} # Ensure the cost limit is included }, 'precision': { 'amount': 8, 'price': 8 }, 'active': True, 'symbol': 'BTC/USDT' }, 'ETH/USDT': { 'limits': { 'amount': {'min': 0.01}, 'cost': {'min': 20.0} # Ensure the cost limit is included }, 'precision': { 'amount': 8, 'price': 8 }, 'active': True, 'symbol': 'ETH/USDT' } } self.mock_client.fetch_ticker.return_value = {'last': 30000.0} self.mock_client.fetch_ohlcv.return_value = [ [1609459200000, 29000.0, 29500.0, 28800.0, 29400.0, 1000] ] self.mock_client.fetch_balance.return_value = { 'total': {'BTC': 1.0, 'USDT': 1000.0} } self.mock_client.create_order.return_value = { 'id': 'test_order_id', 'symbol': 'BTC/USDT', 'type': 'limit', 'side': 'buy', 'price': 30000.0, 'amount': 1.0, 'status': 'open' } self.mock_client.fetch_open_orders.return_value = [ {'id': 'test_order_id', 'symbol': 'BTC/USDT', 'side': 'buy', 'amount': 1.0, 'price': 30000.0} ] self.mock_client.fetch_positions.return_value = [ {'symbol': 'BTC/USDT', 'quantity': 1.0, 'entry_price': 29000.0} ] self.mock_client.fetch_order.return_value = { 'id': 'test_order_id', 'symbol': 'BTC/USDT', 'side': 'buy', 'amount': 1.0, 'price': 30000.0, 'status': 'open' } self.mock_client.timeframes = { '1m': '1m', '5m': '5m', '1h': '1h', '1d': '1d', } self.exchange = Exchange(name='Binance', api_keys=self.api_keys, exchange_id='binance') def test_connect_exchange(self): self.assertIsInstance(self.exchange.client, MagicMock) self.mock_client.load_markets.assert_called_once() self.assertEqual(self.exchange.symbols, ['BTC/USDT', 'ETH/USDT']) self.assertIn('BTC/USDT', self.exchange.exchange_info) self.assertIn('ETH/USDT', self.exchange.exchange_info) self.assertEqual(self.exchange.balances, [ {'asset': 'BTC', 'balance': 1.0, 'pnl': 0}, {'asset': 'USDT', 'balance': 1000.0, 'pnl': 0} ]) def test_get_symbols(self): symbols = self.exchange.get_symbols() self.assertEqual(symbols, ['BTC/USDT', 'ETH/USDT']) def test_get_price(self): price = self.exchange.get_price('BTC/USDT') self.assertEqual(price, 30000.0) self.mock_client.fetch_ticker.assert_called_with('BTC/USDT') def test_get_balances(self): balances = self.exchange.get_balances() self.assertEqual(balances, [ {'asset': 'BTC', 'balance': 1.0, 'pnl': 0}, {'asset': 'USDT', 'balance': 1000.0, 'pnl': 0} ]) self.mock_client.fetch_balance.assert_called_once() def test_get_historical_klines(self): start_dt = datetime(2021, 1, 1) end_dt = datetime(2021, 1, 2) klines = self.exchange.get_historical_klines('BTC/USDT', '1d', start_dt, end_dt) expected_df = pd.DataFrame([ {'time': 1609459200, 'open': 29000.0, 'high': 29500.0, 'low': 28800.0, 'close': 29400.0, 'volume': 1000} ]) pd.testing.assert_frame_equal(klines, expected_df) self.mock_client.fetch_ohlcv.assert_called() def test_get_min_qty(self): min_qty = self.exchange.get_min_qty('BTC/USDT') self.assertEqual(min_qty, 0.001) def test_get_min_notional_qty(self): min_notional_qty = self.exchange.get_min_notional_qty('BTC/USDT') self.assertEqual(min_notional_qty, 10.0) def test_get_order(self): order = self.exchange.get_order('BTC/USDT', 'test_order_id') self.assertIsInstance(order, dict) self.assertEqual(order['id'], 'test_order_id') self.mock_client.fetch_order.assert_called_with('test_order_id', 'BTC/USDT') def test_place_order(self): result, order = self.exchange.place_order( symbol='BTC/USDT', side='buy', type='limit', timeInForce='GTC', quantity=1.0, price=30000.0 ) self.assertEqual(result, 'Success') self.assertIsInstance(order, dict) self.assertEqual(order.get('id'), 'test_order_id') self.mock_client.create_order.assert_called_once() def test_get_open_orders(self): open_orders = self.exchange.get_open_orders() self.assertEqual(open_orders, [ {'symbol': 'BTC/USDT', 'side': 'buy', 'quantity': 1.0, 'price': 30000.0} ]) self.mock_client.fetch_open_orders.assert_called_once() def test_get_active_trades(self): active_trades = self.exchange.get_active_trades() self.assertEqual(active_trades, [ {'symbol': 'BTC/USDT', 'side': 'buy', 'quantity': 1.0, 'price': 29000.0} ]) self.mock_client.fetch_positions.assert_called_once() @patch('ccxt.binance') def test_fetch_ohlcv_network_failure(self, mock_exchange): self.mock_client.fetch_ohlcv.side_effect = ccxt.NetworkError('Network error') start_dt = datetime(2021, 1, 1) end_dt = datetime(2021, 1, 2) klines = self.exchange.get_historical_klines('BTC/USDT', '1d', start_dt, end_dt) self.assertTrue(klines.empty) self.mock_client.fetch_ohlcv.assert_called() @patch('ccxt.binance') def test_fetch_ticker_invalid_response(self, mock_exchange): self.mock_client.fetch_ticker.side_effect = ccxt.ExchangeError('Invalid response') price = self.exchange.get_price('BTC/USDT') self.assertEqual(price, 0.0) self.mock_client.fetch_ticker.assert_called_with('BTC/USDT') if __name__ == '__main__': unittest.main()