Added activeAssetCtx websocket subscription (#165)

Author: akpav

examples/basic_ws.py

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

hyperliquid/info.py

@@ -604,6 +604,7 @@ def _remap_coin_subscription(self, subscription: Subscription) -> None:
             or subscription["type"] == "trades"
             or subscription["type"] == "candle"
             or subscription["type"] == "bbo"
+            or subscription["type"] == "activeAssetCtx"
         ):
             subscription["coin"] = self.name_to_coin[subscription["coin"]]
 

hyperliquid/utils/types.py

@@ -49,6 +49,7 @@
     "UserNonFundingLedgerUpdatesSubscription", {"type": Literal["userNonFundingLedgerUpdates"], "user": str}
 )
 WebData2Subscription = TypedDict("WebData2Subscription", {"type": Literal["webData2"], "user": str})
+ActiveAssetCtxSubscription = TypedDict("ActiveAssetCtxSubscription", {"type": Literal["activeAssetCtx"], "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,
@@ -62,6 +63,7 @@
     UserFundingsSubscription,
     UserNonFundingLedgerUpdatesSubscription,
     WebData2Subscription,
+    ActiveAssetCtxSubscription,
 ]
 
 AllMidsData = TypedDict("AllMidsData", {"mids": Dict[str, str]})
@@ -74,6 +76,27 @@
 PongMsg = TypedDict("PongMsg", {"channel": Literal["pong"]})
 Trade = TypedDict("Trade", {"coin": str, "side": Side, "px": str, "sz": int, "hash": str, "time": int})
 TradesMsg = TypedDict("TradesMsg", {"channel": Literal["trades"], "data": List[Trade]})
+PerpAssetCtx = TypedDict(
+    "PerpAssetCtx",
+    {
+        "funding": str,
+        "openInterest": str,
+        "prevDayPx": str,
+        "dayNtlVlm": str,
+        "premium": str,
+        "oraclePx": str,
+        "markPx": str,
+        "midPx": Optional[str],
+        "impactPxs": Optional[Tuple[str, str]],
+        "dayBaseVlm": str,
+    },
+)
+ActiveAssetCtx = TypedDict("ActiveAssetCtx", {"coin": str, "ctx": PerpAssetCtx})
+ActiveSpotAssetCtx = TypedDict("ActiveSpotAssetCtx", {"coin": str, "ctx": SpotAssetCtx})
+ActiveAssetCtxMsg = TypedDict("ActiveAssetCtxMsg", {"channel": Literal["activeAssetCtx"], "data": ActiveAssetCtx})
+ActiveSpotAssetCtxMsg = TypedDict(
+    "ActiveSpotAssetCtxMsg", {"channel": Literal["activeSpotAssetCtx"], "data": ActiveSpotAssetCtx}
+)
 Fill = TypedDict(
     "Fill",
     {
@@ -112,7 +135,18 @@
     },
     total=False,
 )
-WsMsg = Union[AllMidsMsg, BboMsg, L2BookMsg, TradesMsg, UserEventsMsg, PongMsg, UserFillsMsg, OtherWsMsg]
+WsMsg = Union[
+    AllMidsMsg,
+    BboMsg,
+    L2BookMsg,
+    TradesMsg,
+    UserEventsMsg,
+    PongMsg,
+    UserFillsMsg,
+    OtherWsMsg,
+    ActiveAssetCtxMsg,
+    ActiveSpotAssetCtxMsg,
+]
 
 # 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
 BuilderInfo = TypedDict("BuilderInfo", {"b": str, "f": int})

hyperliquid/websocket_manager.py

@@ -33,6 +33,8 @@ def subscription_to_identifier(subscription: Subscription) -> str:
         return f'webData2:{subscription["user"].lower()}'
     elif subscription["type"] == "bbo":
         return f'bbo:{subscription["coin"].lower()}'
+    elif subscription["type"] == "activeAssetCtx":
+        return f'activeAssetCtx:{subscription["coin"].lower()}'
 
 
 def ws_msg_to_identifier(ws_msg: WsMsg) -> Optional[str]:
@@ -64,6 +66,8 @@ def ws_msg_to_identifier(ws_msg: WsMsg) -> Optional[str]:
         return f'webData2:{ws_msg["data"]["user"].lower()}'
     elif ws_msg["channel"] == "bbo":
         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()}'
 
 
 class WebsocketManager(threading.Thread):