Added activeAssetData websocket subscription

examples/basic_ws.py

@@ -20,6 +20,7 @@ def main():
     info.subscribe({"type": "bbo", "coin": "ETH"}, print)
     info.subscribe({"type": "activeAssetCtx", "coin": "BTC"}, print)  # Perp
     info.subscribe({"type": "activeAssetCtx", "coin": "@1"}, print)  # Spot
+    info.subscribe({"type": "activeAssetData", "user": address, "coin": "BTC"}, print)  # Perp only
 
 
 if __name__ == "__main__":

hyperliquid/utils/types.py

@@ -50,6 +50,9 @@
 )
 WebData2Subscription = TypedDict("WebData2Subscription", {"type": Literal["webData2"], "user": str})
 ActiveAssetCtxSubscription = TypedDict("ActiveAssetCtxSubscription", {"type": Literal["activeAssetCtx"], "coin": str})
+ActiveAssetDataSubscription = TypedDict(
+    "ActiveAssetDataSubscription", {"type": Literal["activeAssetData"], "user": str, "coin": str}
+)
 # If adding new subscription types that contain coin's don't forget to handle automatically rewrite name to coin in info.subscribe
 Subscription = Union[
     AllMidsSubscription,
@@ -64,6 +67,7 @@
     UserNonFundingLedgerUpdatesSubscription,
     WebData2Subscription,
     ActiveAssetCtxSubscription,
+    ActiveAssetDataSubscription,
 ]
 
 AllMidsData = TypedDict("AllMidsData", {"mids": Dict[str, str]})
@@ -75,6 +79,22 @@
 BboMsg = TypedDict("BboMsg", {"channel": Literal["bbo"], "data": BboData})
 PongMsg = TypedDict("PongMsg", {"channel": Literal["pong"]})
 Trade = TypedDict("Trade", {"coin": str, "side": Side, "px": str, "sz": int, "hash": str, "time": int})
+CrossLeverage = TypedDict(
+    "CrossLeverage",
+    {
+        "type": Literal["cross"],
+        "value": int,
+    },
+)
+IsolatedLeverage = TypedDict(
+    "IsolatedLeverage",
+    {
+        "type": Literal["isolated"],
+        "value": int,
+        "rawUsd": str,
+    },
+)
+Leverage = Union[CrossLeverage, IsolatedLeverage]
 TradesMsg = TypedDict("TradesMsg", {"channel": Literal["trades"], "data": List[Trade]})
 PerpAssetCtx = TypedDict(
     "PerpAssetCtx",
@@ -97,6 +117,18 @@
 ActiveSpotAssetCtxMsg = TypedDict(
     "ActiveSpotAssetCtxMsg", {"channel": Literal["activeSpotAssetCtx"], "data": ActiveSpotAssetCtx}
 )
+ActiveAssetData = TypedDict(
+    "ActiveAssetData",
+    {
+        "user": str,
+        "coin": str,
+        "leverage": Leverage,
+        "maxTradeSzs": Tuple[str, str],
+        "availableToTrade": Tuple[str, str],
+        "markPx": str,
+    },
+)
+ActiveAssetDataMsg = TypedDict("ActiveAssetDataMsg", {"channel": Literal["activeAssetData"], "data": ActiveAssetData})
 Fill = TypedDict(
     "Fill",
     {
@@ -146,6 +178,7 @@
     OtherWsMsg,
     ActiveAssetCtxMsg,
     ActiveSpotAssetCtxMsg,
+    ActiveAssetDataMsg,
 ]
 
 # b is the public address of the builder, f is the amount of the fee in tenths of basis points. e.g. 10 means 1 basis point

hyperliquid/websocket_manager.py

@@ -35,6 +35,8 @@ def subscription_to_identifier(subscription: Subscription) -> str:
         return f'bbo:{subscription["coin"].lower()}'
     elif subscription["type"] == "activeAssetCtx":
         return f'activeAssetCtx:{subscription["coin"].lower()}'
+    elif subscription["type"] == "activeAssetData":
+        return f'activeAssetData:{subscription["coin"].lower()},{subscription["user"]}'
 
 
 def ws_msg_to_identifier(ws_msg: WsMsg) -> Optional[str]:
@@ -68,6 +70,8 @@ def ws_msg_to_identifier(ws_msg: WsMsg) -> Optional[str]:
         return f'bbo:{ws_msg["data"]["coin"].lower()}'
     elif ws_msg["channel"] == "activeAssetCtx" or ws_msg["channel"] == "activeSpotAssetCtx":
         return f'activeAssetCtx:{ws_msg["data"]["coin"].lower()}'
+    elif ws_msg["channel"] == "activeAssetData":
+        return f'activeAssetData:{ws_msg["data"]["coin"].lower()},{ws_msg["data"]["user"]}'
 
 
 class WebsocketManager(threading.Thread):