277 lines
12 KiB
Python
277 lines
12 KiB
Python
from exchangeinterface import ExchangeInterface
|
|
from trade import Trades
|
|
|
|
|
|
def test_connect_exchange():
|
|
exchange = ExchangeInterface()
|
|
test_trades_obj = Trades()
|
|
test_trades_obj.connect_exchange(exchange)
|
|
print(f'\nConnected to exchange_interface: {test_trades_obj.exchange_connected()}')
|
|
account_info = exchange.get_precision(symbol='ETHUSDT')
|
|
print(account_info)
|
|
assert test_trades_obj.exchange_connected()
|
|
|
|
|
|
def test_get_trades_by_status():
|
|
# Connect to the exchange_interface
|
|
exchange = ExchangeInterface()
|
|
test_trades_obj = Trades()
|
|
test_trades_obj.connect_exchange(exchange)
|
|
print(f'\nConnected to exchange_interface: {test_trades_obj.exchange_connected()}')
|
|
assert test_trades_obj.exchange_connected()
|
|
|
|
# create a trade but not on the exchange_interface
|
|
test_trades_obj.new_trade('backtestor', 'BTCUSDT', 'BUY', 'MARKET', 1.5, price=100, offset=None)
|
|
print('trade 0 created.')
|
|
print(test_trades_obj.active_trades[0].__dict__)
|
|
# If the status of the trade is inactive the trade is created but the order isn't placed.
|
|
assert test_trades_obj.active_trades[0].status is 'inactive'
|
|
|
|
# create a 2nd trade but not on the exchange_interface
|
|
test_trades_obj.new_trade('backtestor', 'BTCUSDT', 'BUY', 'MARKET', 0.4, price=2100, offset=None)
|
|
print('trade 1 created.')
|
|
print(test_trades_obj.active_trades[1].__dict__)
|
|
# If the status of the trade is inactive the trade is created but the order isn't placed.
|
|
assert test_trades_obj.active_trades[1].status is 'inactive'
|
|
|
|
# create a 3rd trade but not on the exchange_interface
|
|
test_trades_obj.new_trade('backtestor', 'BTCUSDT', 'BUY', 'MARKET', 0.4, price=2100, offset=None)
|
|
print('trade 2 created.')
|
|
print(test_trades_obj.active_trades[2].__dict__)
|
|
# If the status of the trade is inactive the trade is created but the order isn't placed.
|
|
assert test_trades_obj.active_trades[1].status is 'inactive'
|
|
|
|
# should be three trades in this list
|
|
print(f'Expecting 3 trades in list: Actual:{len(test_trades_obj.active_trades)}')
|
|
assert len(test_trades_obj.active_trades) is 3
|
|
print(test_trades_obj.active_trades[0].status)
|
|
print(test_trades_obj.active_trades[1].status)
|
|
print(test_trades_obj.active_trades[2].status)
|
|
|
|
# fill trade one
|
|
test_trades_obj.active_trades[1].trade_filled(0.4, 2100)
|
|
print(f'trade 1 filled. status:')
|
|
print(test_trades_obj.active_trades[1].status)
|
|
|
|
# Search for all inactive trades
|
|
result = test_trades_obj.get_trades_by_status('inactive')
|
|
print(f'search for all inactive trades. The result: {result}')
|
|
assert len(result) is 2
|
|
|
|
# Search for all filled trades
|
|
result = test_trades_obj.get_trades_by_status('filled')
|
|
print(f'search for all filled trades. The result: {result}')
|
|
assert len(result) is 1
|
|
|
|
|
|
def test_get_trade_by_id():
|
|
# Connect to the exchange_interface
|
|
exchange = ExchangeInterface()
|
|
test_trades_obj = Trades()
|
|
test_trades_obj.connect_exchange(exchange)
|
|
print(f'\nConnected to exchange_interface: {test_trades_obj.exchange_connected()}')
|
|
assert test_trades_obj.exchange_connected()
|
|
|
|
# create a trade but not on the exchange_interface
|
|
test_trades_obj.new_trade('backtestor', 'BTCUSDT', 'BUY', 'MARKET', 1.5, price=100, offset=None)
|
|
print('trade 0 created.')
|
|
print(test_trades_obj.active_trades[0].__dict__)
|
|
# If the status of the trade is inactive the trade is created but the order isn't placed.
|
|
assert test_trades_obj.active_trades[0].status is 'inactive'
|
|
|
|
# create a 2nd trade but not on the exchange_interface
|
|
test_trades_obj.new_trade('backtestor', 'BTCUSDT', 'BUY', 'MARKET', 0.4, price=2100, offset=None)
|
|
print('trade 1 created.')
|
|
print(test_trades_obj.active_trades[1].__dict__)
|
|
# If the status of the trade is inactive the trade is created but the order isn't placed.
|
|
assert test_trades_obj.active_trades[1].status is 'inactive'
|
|
|
|
id = test_trades_obj.active_trades[0].unique_id
|
|
print(f'the id of trade 0 is{id}')
|
|
result = test_trades_obj.get_trade_by_id(id)
|
|
print(f'here is the result after searching for the id:{result}')
|
|
assert result.unique_id is id
|
|
|
|
|
|
def test_load_trades():
|
|
# Connect to the exchange_interface
|
|
exchange = ExchangeInterface()
|
|
test_trades_obj = Trades()
|
|
test_trades_obj.connect_exchange(exchange)
|
|
print(f'\nConnected to exchange_interface: {test_trades_obj.exchange_connected()}')
|
|
assert test_trades_obj.exchange_connected()
|
|
print(f'Active trades: {test_trades_obj.active_trades}')
|
|
trades = [{
|
|
'order_price': 24595.4,
|
|
'target': 'backtester',
|
|
'base_order_qty': 0.05,
|
|
'order': None,
|
|
'fee': 0.1,
|
|
'order_type': 'MARKET',
|
|
'side': 'buy',
|
|
'stats': {
|
|
'current_price': 24595.4,
|
|
'current_value': 1229.7700000000002,
|
|
'fee_paid': 0,
|
|
'opening_price': 24595.4,
|
|
'opening_value': 1229.7700000000002,
|
|
'profit': 0,
|
|
'profit_pct': 0,
|
|
'qty_filled': 0,
|
|
'qty_settled': 0,
|
|
'settled_price': 0,
|
|
'settled_value': 0
|
|
},
|
|
'status': 'inactive',
|
|
'symbol': 'BTCUSDT',
|
|
'time_in_force': 'GTC',
|
|
'unique_id': '9330afd188474d83b06e19d1916c0474'
|
|
}]
|
|
test_trades_obj.load_trades(trades)
|
|
print(f'Active trades: {test_trades_obj.active_trades[0].__dict__}')
|
|
assert len(test_trades_obj.active_trades) > 0
|
|
|
|
|
|
def test_place_order():
|
|
# Connect to the exchange_interface
|
|
exchange = ExchangeInterface()
|
|
test_trades_obj = Trades()
|
|
test_trades_obj.connect_exchange(exchange)
|
|
print(f'\nConnected to exchange_interface: {test_trades_obj.exchange_connected()}')
|
|
|
|
# Create a new treade on the exchange_interface.
|
|
test_trades_obj.new_trade('exchange_interface', 'BTCUSDT', 'BUY', 'MARKET', 0.1, price=100, offset=None)
|
|
print(test_trades_obj.active_trades[0].__dict__)
|
|
|
|
# If the status of the trade is unfilled the order is placed.
|
|
assert test_trades_obj.active_trades[0].status is 'unfilled'
|
|
|
|
|
|
def test_update():
|
|
# Connect to the exchange_interface
|
|
exchange = ExchangeInterface()
|
|
test_trades_obj = Trades()
|
|
test_trades_obj.connect_exchange(exchange)
|
|
print(f'\nConnected to exchange_interface: {test_trades_obj.exchange_connected()}')
|
|
|
|
# Create a trade but not on the exchange_interface
|
|
test_trades_obj.new_trade('backtestor', 'BTCUSDT', 'BUY', 'MARKET', 0.4, price=100, offset=None)
|
|
print(test_trades_obj.active_trades[0].__dict__)
|
|
# If the status of the trade is inactive the trade is created but the order isn't placed.
|
|
assert test_trades_obj.active_trades[0].status is 'inactive'
|
|
|
|
# create a 2nd trade but not on the exchange_interface
|
|
test_trades_obj.new_trade('backtestor', 'BTCUSDT', 'BUY', 'MARKET', 0.2, price=100, offset=None)
|
|
print(test_trades_obj.active_trades[1].__dict__)
|
|
# If the status of the trade is inactive the trade is created but the order isn't placed.
|
|
assert test_trades_obj.active_trades[1].status is 'inactive'
|
|
|
|
test_trades_obj.active_trades[0].trade_filled(0.4, 100)
|
|
test_trades_obj.active_trades[1].trade_filled(0.2, 100)
|
|
|
|
test_trades_obj.update(200)
|
|
print(test_trades_obj.active_trades[0].__dict__)
|
|
print(test_trades_obj.active_trades[1].__dict__)
|
|
|
|
|
|
def test_new_trade():
|
|
# Connect to the exchange_interface
|
|
exchange = ExchangeInterface()
|
|
test_trades_obj = Trades()
|
|
test_trades_obj.connect_exchange(exchange)
|
|
print(f'\nConnected to exchange_interface: {test_trades_obj.exchange_connected()}')
|
|
|
|
# create an trade but not on the exchange_interface
|
|
test_trades_obj.new_trade('backtestor', 'BTCUSDT', 'BUY', 'MARKET', 0.1, price=100, offset=None)
|
|
print(test_trades_obj.active_trades[0].__dict__)
|
|
# If the status of the trade is inactive the trade is created but the order isn't placed.
|
|
assert test_trades_obj.active_trades[0].status is 'inactive'
|
|
|
|
# create a 2nd trade but not on the exchange_interface
|
|
test_trades_obj.new_trade('backtestor', 'BTCUSDT', 'BUY', 'MARKET', 0.4, price=2100, offset=None)
|
|
print(test_trades_obj.active_trades[1].__dict__)
|
|
# If the status of the trade is inactive the trade is created but the order isn't placed.
|
|
assert test_trades_obj.active_trades[1].status is 'inactive'
|
|
|
|
|
|
def test_close_trade():
|
|
# Connect to the exchange_interface
|
|
exchange = ExchangeInterface()
|
|
test_trades_obj = Trades()
|
|
test_trades_obj.connect_exchange(exchange)
|
|
print(f'\nConnected to exchange_interface: {test_trades_obj.exchange_connected()}')
|
|
|
|
# create a trade but not on the exchange_interface
|
|
test_trades_obj.new_trade('backtestor', 'BTCUSDT', 'BUY', 'MARKET', 0.1, price=100, offset=None)
|
|
print(test_trades_obj.active_trades[0].__dict__)
|
|
# If the status of the trade is inactive the trade is created but the order isn't placed.
|
|
assert test_trades_obj.active_trades[0].status is 'inactive'
|
|
|
|
# create a 2nd trade but not on the exchange_interface
|
|
test_trades_obj.new_trade('backtestor', 'BTCUSDT', 'BUY', 'MARKET', 0.4, price=2100, offset=None)
|
|
print(test_trades_obj.active_trades[1].__dict__)
|
|
# If the status of the trade is inactive the trade is created but the order isn't placed.
|
|
assert test_trades_obj.active_trades[1].status is 'inactive'
|
|
|
|
# should be two trades in this list
|
|
print(len(test_trades_obj.active_trades))
|
|
assert len(test_trades_obj.active_trades) > 1
|
|
|
|
trade_id = test_trades_obj.active_trades[0].unique_id
|
|
test_trades_obj.close_trade(trade_id)
|
|
|
|
# should be 1 trade in this list
|
|
print(len(test_trades_obj.active_trades))
|
|
assert len(test_trades_obj.active_trades) == 1
|
|
|
|
|
|
def test_reduce_trade():
|
|
# Connect to the exchange_interface
|
|
exchange = ExchangeInterface()
|
|
test_trades_obj = Trades()
|
|
test_trades_obj.connect_exchange(exchange)
|
|
print(f'\nConnected to exchange_interface: {test_trades_obj.exchange_connected()}')
|
|
assert test_trades_obj.exchange_connected()
|
|
|
|
# create a trade but not on the exchange_interface
|
|
test_trades_obj.new_trade('backtestor', 'BTCUSDT', 'BUY', 'MARKET', 1.5, price=100, offset=None)
|
|
print('trade 0 created.')
|
|
print(test_trades_obj.active_trades[0].__dict__)
|
|
# If the status of the trade is inactive the trade is created but the order isn't placed.
|
|
assert test_trades_obj.active_trades[0].status is 'inactive'
|
|
|
|
# create a 2nd trade but not on the exchange_interface
|
|
test_trades_obj.new_trade('backtestor', 'BTCUSDT', 'BUY', 'MARKET', 0.4, price=2100, offset=None)
|
|
print('trade 1 created.')
|
|
print(test_trades_obj.active_trades[1].__dict__)
|
|
# If the status of the trade is inactive the trade is created but the order isn't placed.
|
|
assert test_trades_obj.active_trades[1].status is 'inactive'
|
|
|
|
# should be two trades in this list
|
|
print(f'Expecting 2 trades in list: Actual:{len(test_trades_obj.active_trades)}')
|
|
assert len(test_trades_obj.active_trades) > 1
|
|
|
|
# Grab inactive trade 0
|
|
trade_id = test_trades_obj.active_trades[0].unique_id
|
|
# reduce the trade by 0.5
|
|
remaining_qty = test_trades_obj.reduce_trade(trade_id, 0.5)
|
|
# The trade should be 1 (1.5 - 0.5)
|
|
print(f'The remaining quantity of the trade should be 1: Actual: {remaining_qty}')
|
|
assert remaining_qty == 1
|
|
|
|
print('trade 0:')
|
|
print(test_trades_obj.active_trades[0].__dict__)
|
|
|
|
test_trades_obj.active_trades[1].trade_filled(0.4, 2100)
|
|
# Grab filled trade 1
|
|
trade_id = test_trades_obj.active_trades[1].unique_id
|
|
|
|
# reduce the trade by 0.1
|
|
remaining_qty = float(test_trades_obj.reduce_trade(trade_id, 0.1))
|
|
# The trade should be 0.3 (0.4 - 0.1)
|
|
print(f'\nThe remaining quantity of trade 1 should be 0.3: Actual: {remaining_qty}')
|
|
assert remaining_qty == 0.3
|
|
|
|
print('trade 1:')
|
|
print(test_trades_obj.active_trades[1].__dict__)
|