brighter-trading/tests/test_trade.py

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,
        'exchange_name': '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__)