# This is an end to end example of a very basic adding strategy.
import json
import logging
import threading
import time
import example_utils
from hyperliquid.exchange import Exchange
from hyperliquid.info import Info
from hyperliquid.utils import constants
from hyperliquid.utils.signing import get_timestamp_ms
from hyperliquid.utils.types import (
SIDES,
Dict,
L2BookData,
L2BookMsg,
L2BookSubscription,
Literal,
Optional,
Side,
TypedDict,
Union,
UserEventsMsg,
UserEventsSubscription,
)
# --------------------------- CONFIGURATION ---------------------------
# How far from the best bid and offer this strategy ideally places orders. Currently set to 0.3%.
# i.e., if the best bid is $1000, this strategy will place a resting bid at $997 (1000 - 0.3% of 1000).
DEPTH = 0.003
# How far from the target price a resting order can deviate before the strategy will cancel and replace it.
# i.e., using the same example as above of a best bid of $1000 and targeted depth of 0.3% (a $3 difference).
# Bids within $3 * 0.5 = $1.5 of the ideal price will not be cancelled. Therefore, any bids > $998.5 or < $995.5
# will be cancelled and replaced.
ALLOWABLE_DEVIATION = 0.5
# The maximum absolute position value the strategy can accumulate in units of the coin.
# i.e., the strategy will place orders such that it can long up to 1 ETH or short up to 1 ETH.
MAX_POSITION = 1.0
# The coin to add liquidity on.
COIN = "ETH"
# The interval (in seconds) at which the polling function runs.
POLL_INTERVAL = 10
# The maximum time (in milliseconds) to wait for an in-flight order before treating it as cancelled.
ORDER_TIMEOUT = 10000
# The time (in milliseconds) to keep recently cancelled orders before cleaning them up.
CANCEL_CLEANUP_TIME = 30000
# --------------------------- TYPE DEFINITIONS ---------------------------
InFlightOrder = TypedDict("InFlightOrder", {"type": Literal["in_flight_order"], "time": int})
Resting = TypedDict("Resting", {"type": Literal["resting"], "px": float, "oid": int})
Cancelled = TypedDict("Cancelled", {"type": Literal["cancelled"]})
ProvideState = Union[InFlightOrder, Resting, Cancelled]
def side_to_int(side: Side) -> int:
"""Convert side ('A' for Ask, 'B' for Bid) to an integer multiplier."""
return 1 if side == "A" else -1
def side_to_uint(side: Side) -> int:
"""Convert side ('A' for Ask, 'B' for Bid) to an unsigned integer (0 or 1)."""
return 1 if side == "A" else 0
class BasicAdder:
def __init__(self, address: str, info: Info, exchange: Exchange):
self.info = info
self.exchange = exchange
self.address = address
self.position: Optional[float] = None
self.provide_state: Dict[Side, ProvideState] = {
"A": {"type": "cancelled"},
"B": {"type": "cancelled"},
}
self.recently_cancelled_oid_to_time: Dict[int, int] = {}
# Subscribe to updates
self.subscribe_to_updates()
# Start the polling thread
self.start_poller()
def subscribe_to_updates(self) -> None:
"""Subscribe to order book and user event updates."""
l2_book_subscription: L2BookSubscription = {"type": "l2Book", "coin": COIN}
self.info.subscribe(l2_book_subscription, self.on_book_update)
user_events_subscription: UserEventsSubscription = {"type": "userEvents", "user": self.address}
self.info.subscribe(user_events_subscription, self.on_user_events)
def start_poller(self) -> None:
"""Start the polling thread for checking open orders and positions."""
self.poller = threading.Thread(target=self.poll, daemon=True)
self.poller.start()
def on_book_update(self, book_msg: L2BookMsg) -> None:
"""Callback for order book updates."""
logging.debug(f"Received book message: {book_msg}")
book_data = book_msg["data"]
if book_data["coin"] != COIN:
logging.warning("Unexpected book message, skipping.")
return
for side in SIDES:
self.handle_order_placement(side, book_data)
def handle_order_placement(self, side: Side, book_data: L2BookData) -> None:
"""Handle the placement and cancellation of orders based on the order book update."""
book_price = float(book_data["levels"][side_to_uint(side)][0]["px"])
ideal_distance = book_price * DEPTH
ideal_price = book_price + (ideal_distance * side_to_int(side))
provide_state = self.provide_state[side]
if provide_state["type"] == "resting":
self.maybe_cancel_order(side, provide_state, ideal_price, ideal_distance)
elif provide_state["type"] == "in_flight_order":
self.check_in_flight_order(side, provide_state)
if provide_state["type"] == "cancelled":
self.place_new_order(side, ideal_price)
def maybe_cancel_order(self, side: Side, provide_state: Resting, ideal_price: float, ideal_distance: float) -> None:
"""Cancel the order if it deviates beyond the allowable limit."""
distance = abs(ideal_price - provide_state["px"])
if distance > ALLOWABLE_DEVIATION * ideal_distance:
oid = provide_state["oid"]
print(f"Cancelling order due to deviation: oid:{oid}, side:{side}, ideal_price:{ideal_price}")
response = self.exchange.cancel(COIN, oid)
if response["status"] == "ok":
self.recently_cancelled_oid_to_time[oid] = get_timestamp_ms()
self.provide_state[side] = {"type": "cancelled"}
else:
logging.error(f"Failed to cancel order {oid} for side {side}: {response}")
def check_in_flight_order(self, side: Side, provide_state: InFlightOrder) -> None:
"""Check if the in-flight order has timed out."""
if get_timestamp_ms() - provide_state["time"] > ORDER_TIMEOUT:
print("Order is still in flight after timeout, treating as cancelled.")
self.provide_state[side] = {"type": "cancelled"}
def place_new_order(self, side: Side, ideal_price: float) -> None:
"""Place a new order if conditions are met."""
if self.position is None:
logging.debug("Waiting for position refresh before placing order.")
return
size = MAX_POSITION + self.position * side_to_int(side)
if size * ideal_price < 10:
logging.debug("Order size too small, not placing order.")
return
px = float(f"{ideal_price:.5g}")
print(f"Placing order: size:{size}, price:{px}, side:{side}")
response = self.exchange.order(COIN, side == "B", size, px, {"limit": {"tif": "Alo"}})
if response["status"] == "ok":
status = response["response"]["data"]["statuses"][0]
if "resting" in status:
self.provide_state[side] = {"type": "resting", "px": px, "oid": status["resting"]["oid"]}
def on_user_events(self, user_events: UserEventsMsg) -> None:
"""Callback for user events (e.g., fills)."""
print(user_events)
if "fills" in user_events["data"]:
with open("fills", "a+") as f:
f.write(json.dumps(user_events["data"]["fills"]) + "\n")
# Set the position to None so that we don't place more orders without knowing our position
# You might want to also update provide_state to account for the fill. This could help avoid sending an
# unneeded cancel or failing to send a new order to replace the filled order, but we skipped this logic
# to make the example simpler
self.position = None
def poll(self) -> None:
"""Poll open orders and user positions periodically."""
while True:
# Fetch open orders
open_orders = self.info.open_orders(self.exchange.wallet.address)
print("open_orders", open_orders)
# Collect valid order IDs (from recently cancelled orders and resting orders)
ok_oids = set(self.recently_cancelled_oid_to_time.keys())
for provide_state in self.provide_state.values():
if provide_state["type"] == "resting":
ok_oids.add(provide_state["oid"])
# Cancel any unknown orders
for open_order in open_orders:
if open_order["coin"] == COIN and open_order["oid"] not in ok_oids:
print("Cancelling unknown oid", open_order["oid"])
self.exchange.cancel(open_order["coin"], open_order["oid"])
# Clean up recently cancelled orders after a timeout
current_time = get_timestamp_ms()
self.recently_cancelled_oid_to_time = {
oid: timestamp
for oid, timestamp in self.recently_cancelled_oid_to_time.items()
if current_time - timestamp <= CANCEL_CLEANUP_TIME
}
self.refresh_position()
time.sleep(POLL_INTERVAL)
def refresh_position(self) -> None:
"""Refresh the user’s current position."""
user_state = self.info.user_state(self.address)
for position in user_state.get("assetPositions", []):
if position["position"]["coin"] == COIN:
self.position = float(position["position"]["szi"])
return
self.position = 0.0
def main():
# Setting this to logging.DEBUG can be helpful for debugging websocket callback issues
logging.basicConfig(level=logging.INFO)
address, info, exchange = example_utils.setup(constants.TESTNET_API_URL)
BasicAdder(address, info, exchange)
if __name__ == "__main__":
main()
