146 lines
5.5 KiB
Python
146 lines
5.5 KiB
Python
import unittest
|
|
import ccxt
|
|
from datetime import datetime
|
|
import pandas as pd
|
|
import pytest
|
|
from Exchange import Exchange
|
|
from typing import Dict, Optional
|
|
|
|
|
|
@pytest.mark.integration
|
|
class TestExchange(unittest.TestCase):
|
|
"""
|
|
Integration tests that connect to real exchanges.
|
|
Run with: pytest -m integration tests/test_live_exchange_integration.py
|
|
"""
|
|
api_keys: Optional[Dict[str, str]]
|
|
exchange: Exchange
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
exchange_name = 'kraken'
|
|
cls.api_keys = None
|
|
"""Uncomment and Provide api keys to connect to exchange."""
|
|
# cls.api_keys = {'key': 'EXCHANGE_API_KEY', 'secret': 'EXCHANGE_API_SECRET'}
|
|
|
|
cls.exchange = Exchange(name=exchange_name, api_keys=cls.api_keys, exchange_id=exchange_name)
|
|
|
|
def test_connect_exchange(self):
|
|
print("\nRunning test_connect_exchange...")
|
|
exchange_class = getattr(ccxt, self.exchange.name)
|
|
self.assertIsInstance(self.exchange.client, exchange_class)
|
|
print("Exchange client instance test passed.")
|
|
|
|
self.assertIn('BTC/USDT', self.exchange.symbols)
|
|
self.assertIn('ETH/USDT', self.exchange.symbols)
|
|
print("Symbols test passed.")
|
|
|
|
self.assertIn('BTC/USDT', self.exchange.exchange_info)
|
|
self.assertIn('ETH/USDT', self.exchange.exchange_info)
|
|
print("Exchange info test passed.")
|
|
|
|
def test_get_symbols(self):
|
|
print("\nRunning test_get_symbols...")
|
|
symbols = self.exchange.get_symbols()
|
|
self.assertIn('BTC/USDT', symbols)
|
|
self.assertIn('ETH/USDT', symbols)
|
|
print("Get symbols test passed.")
|
|
|
|
def test_get_price(self):
|
|
print("\nRunning test_get_price...")
|
|
price = self.exchange.get_price('BTC/USDT')
|
|
self.assertGreater(price, 0)
|
|
print(f"Get price test passed. Price: {price}")
|
|
|
|
def test_get_balances(self):
|
|
print("\nRunning test_get_balances...")
|
|
if self.api_keys is not None:
|
|
balances = self.exchange.get_balances()
|
|
self.assertTrue(any(balance['asset'] == 'BTC' for balance in balances))
|
|
self.assertTrue(any(balance['asset'] == 'USDT' for balance in balances))
|
|
print("Get balances test passed.")
|
|
else:
|
|
print("test_get_balances(): Can not test without providing API keys")
|
|
|
|
def test_get_historical_klines(self):
|
|
print("\nRunning test_get_historical_klines...")
|
|
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.assertIsInstance(klines, pd.DataFrame)
|
|
self.assertFalse(klines.empty)
|
|
print("Get historical klines test passed.")
|
|
print(klines.head())
|
|
|
|
def test_get_min_qty(self):
|
|
print("\nRunning test_get_min_qty...")
|
|
min_qty = self.exchange.get_min_qty('BTC/USDT')
|
|
self.assertGreater(min_qty, 0)
|
|
print(f"Get min qty test passed. Min qty: {min_qty}")
|
|
|
|
def test_get_min_notional_qty(self):
|
|
print("\nRunning test_get_min_notional_qty...")
|
|
min_notional_qty = self.exchange.get_min_notional_qty('BTC/USDT')
|
|
self.assertGreater(min_notional_qty, 0)
|
|
print(f"Get min notional qty test passed. Min notional qty: {min_notional_qty}")
|
|
|
|
def test_get_order(self):
|
|
print("\nRunning test_get_order...")
|
|
if self.api_keys is not None:
|
|
# You need to create an order manually on exchange to test this
|
|
order_id = 'your_order_id'
|
|
order = self.exchange.get_order('BTC/USDT', order_id)
|
|
self.assertIsInstance(order, dict)
|
|
self.assertEqual(order['id'], order_id)
|
|
print("Get order test passed.")
|
|
else:
|
|
print("test_get_order(): Can not test without providing API keys")
|
|
|
|
def test_place_order(self):
|
|
print("\nRunning test_place_order...")
|
|
if self.api_keys is not None:
|
|
# Be cautious with creating real orders. This is for testing purposes.
|
|
symbol = 'BTC/USDT'
|
|
order_type = 'limit'
|
|
side = 'buy'
|
|
amount = 0.001
|
|
price = 30000 # Adjust accordingly
|
|
|
|
result, order = self.exchange.place_order(
|
|
symbol=symbol,
|
|
side=side,
|
|
type=order_type,
|
|
timeInForce='GTC',
|
|
quantity=amount,
|
|
price=price
|
|
)
|
|
|
|
self.assertEqual(result, 'Success')
|
|
self.assertIsInstance(order, dict)
|
|
self.assertIn('id', order)
|
|
print("Place order test passed.")
|
|
else:
|
|
print("test_place_order(): Can not test without providing API keys")
|
|
|
|
def test_get_open_orders(self):
|
|
print("\nRunning test_get_open_orders...")
|
|
if self.api_keys is not None:
|
|
open_orders = self.exchange.get_open_orders()
|
|
self.assertIsInstance(open_orders, list)
|
|
print("Get open orders test passed.")
|
|
else:
|
|
print("test_get_open_orders(): Can not test without providing API keys")
|
|
|
|
def test_get_active_trades(self):
|
|
print("\nRunning test_get_active_trades...")
|
|
if self.api_keys is not None:
|
|
active_trades = self.exchange.get_active_trades()
|
|
self.assertIsInstance(active_trades, list)
|
|
print("Get active trades test passed.")
|
|
else:
|
|
print("test_get_active_trades(): Can not test without providing API keys")
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|