htfbacktest_very_small.py

import time

import numpy as np
from numba import njit

from hftbacktest import FeedLatency, Linear, HftBacktest, GTX, Stat


@njit
def simple_two_sided_quote(hbt, stat):
    max_position = 200
    half_spread = hbt.tick_size * 2
    skew = 1
    order_qty = 0.1
    last_order_id = -1

    bid_order_price_tick_as_id = -1
    ask_order_price_tick_as_id = -1

    interval_seconds = 0.5

    while hbt.run:
        if not hbt.elapse(interval_seconds * 1e6):
            return False

        # Clear cancelled, filled or expired orders.
        hbt.clear_inactive_orders()

        # Obtain the current mid-price and compute the reservation price.
        mid_price = (hbt.best_bid + hbt.best_ask) / 2.0
        reservation_price = mid_price - skew * hbt.position * hbt.tick_size

        bid_order_price = reservation_price - half_spread
        ask_order_price = reservation_price + half_spread

        # print("ts = ", hbt.current_timestamp, " mid price = ", mid_price, " bid order price = ", bid_order_price,
        #       " ask order price = ", ask_order_price)
        # Cancel the existing bid order.
        existing_bid_order = hbt.orders.get(bid_order_price_tick_as_id)
        # print("existing orders = ", hbt.orders)
        # print(" existing bid order = ", existing_bid_order, " cancellable = ", existing_bid_order.cancellable)
        if existing_bid_order is not None and existing_bid_order.cancellable:
            # print("cancelling bid")
            bid_cancel_result = hbt.cancel(existing_bid_order.order_id, wait=True)
            # print("bid cancel result = ", bid_cancel_result)
            last_order_id = existing_bid_order.order_id

        # Cancel the existing ask order.
        existing_ask_order = hbt.orders.get(ask_order_price_tick_as_id)
        # print(" existing ask order = ", existing_ask_order, " cancellable = ", existing_ask_order.cancellable)
        if existing_ask_order is not None and existing_ask_order.cancellable:
            # print("cancelling ask")
            ask_cancel_result = hbt.cancel(existing_ask_order.order_id, wait=True)
            # print("ask cancel result = ", ask_cancel_result)
            last_order_id = existing_ask_order.order_id

        if hbt.position < max_position:
            # Submit a new post-only limit bid order.
            bid_order_price_tick_as_id = round(bid_order_price / hbt.tick_size)
            hbt.submit_buy_order(
                bid_order_price_tick_as_id,
                bid_order_price,
                order_qty,
                GTX
            )
            last_order_id = bid_order_price_tick_as_id

        if hbt.position > -max_position:
            # Submit a new post-only limit ask order.
            ask_order_price_tick_as_id = round(ask_order_price / hbt.tick_size)
            hbt.submit_sell_order(
                ask_order_price_tick_as_id,
                ask_order_price,
                order_qty,
                GTX
            )
            last_order_id = ask_order_price_tick_as_id

        # All order requests are considered to be requested at the same time.
        # Wait until one of the order responses is received.
        if last_order_id >= 0:
            hbt.wait_order_response(last_order_id)

        # Record the current state for stat calculation.
        stat.record(hbt)
    return True


def run():
    t0 = time.perf_counter()
    data = np.load("data_very_small.npz")
    data = data[list(data.keys())[0]]
    snapshot = np.load("data_snapshot_very_small.npz")
    snapshot = snapshot[list(snapshot.keys())[0]]
    tick_size = 1.0
    lot_size = 0.1
    fee_maker = 0.0
    fee_taker = 0.00001
    hbt = HftBacktest(data,
                      tick_size=tick_size,  # Tick size of a target trading asset
                      lot_size=lot_size,  # Lot size of a target trading asset, minimum trading unit.
                      maker_fee=fee_maker,
                      taker_fee=fee_taker,
                      order_latency=FeedLatency(),  # Latency model: ConstantLatency, FeedLatency.
                      asset_type=Linear,  # Asset type: Linear, Inverse.
                      snapshot=snapshot)
    stat = Stat(hbt)
    simple_two_sided_quote(hbt, stat.recorder)
    t1 = time.perf_counter()
    print(f"time taken = {t1 - t0:,.2f} s")


if __name__ == "__main__":
    run()