Add WS support for launchpad

README.md

@@ -83,7 +83,7 @@ ws.run(handle_msg=handle_msg)
 ```
 Check out more detailed examples [here](https://github.com/polygon-io/client-python/tree/master/examples/websocket).
 
-## Launchpad
+## Launchpad REST API Client
 Users of the Launchpad product will need to pass in certain headers in order to make API requests using the RequestOptionBuilder.
 Example can be found [here](./examples/launchpad).
 
@@ -114,6 +114,23 @@ res = c.get_aggs("AAPL", 1, "day", "2022-04-04", "2022-04-04", options=options)
 Checkout Launchpad readme for more details on RequestOptionBuilder [here](./examples/launchpad)
 
 
+## Launchpad WebSocket Client
+
+```python
+from polygon import WebSocketClient
+from polygon.websocket.models import WebSocketMessage
+from polygon.websocket.models.common import Feed, Market
+from typing import List
+
+ws = WebSocketClient(api_key="API_KEY",feed=Feed.Launchpad,market=Market.Stocks, subscriptions=["AM.AAPL"])
+
+def handle_msg(msg: List[WebSocketMessage]):
+    for m in msg:
+        print(m)
+
+ws.run(handle_msg=handle_msg)
+```
+
 ## Contributing
 
 If you found a bug or have an idea for a new feature, please first discuss it with us by

examples/websocket/launchpad-ws.py

@@ -0,0 +1,21 @@
+from polygon import WebSocketClient
+from polygon.websocket.models import WebSocketMessage, Feed, Market
+from typing import List
+
+client = WebSocketClient(
+    api_key="<POLYGON_API_KEY>", feed=Feed.Launchpad, market=Market.Stocks
+)
+
+client.subscribe("AM.*")  # all aggregates
+# client.subscribe("LV.*")  # all aggregates
+# client.subscribe("AM.O:A230616C00070000")  # all aggregates
+# client.subscribe("LV.O:A230616C00070000")  # all aggregates
+
+
+def handle_msg(msgs: List[WebSocketMessage]):
+    for m in msgs:
+        print(m)
+
+
+# print messages
+client.run(handle_msg)

polygon/websocket/models/__init__.py

@@ -28,6 +28,8 @@ def parse_single(data: Dict[str, Any]):
         return Level2Book.from_dict(data)
     elif event_type == EventType.Value.value:
         return IndexValue.from_dict(data)
+    elif event_type == EventType.LaunchpadValue.value:
+        return LaunchpadValue.from_dict(data)
     return None
 
 

polygon/websocket/models/common.py

@@ -8,6 +8,7 @@ class Feed(Enum):
     PolyFeed = "polyfeed.polygon.io"
     PolyFeedPlus = "polyfeedplus.polygon.io"
     StarterFeed = "starterfeed.polygon.io"
+    Launchpad = "launchpad.polygon.io"
 
 
 class Market(Enum):
@@ -32,3 +33,8 @@ class EventType(Enum):
     LimitUpLimitDown = "LULD"
     CryptoL2 = "XL2"
     Value = "V"
+    """Launchpad* EventTypes are only available to Launchpad users. These values are the same across all asset classes (
+    stocks, options, forex, crypto). 
+    """
+    LaunchpadValue = "LV"
+    LaunchpadAggMin = "AM"

polygon/websocket/models/models.py

@@ -325,6 +325,23 @@ def from_dict(d):
         )
 
 
+@modelclass
+class LaunchpadValue:
+    event_type: Optional[Union[str, EventType]] = None
+    value: Optional[float] = None
+    symbol: Optional[str] = None
+    timestamp: Optional[int] = None
+
+    @staticmethod
+    def from_dict(d):
+        return LaunchpadValue(
+            event_type=d.get("ev", None),
+            value=d.get("val", None),
+            symbol=d.get("sym", None),
+            timestamp=d.get("t", None),
+        )
+
+
 WebSocketMessage = NewType(
     "WebSocketMessage",
     List[
@@ -340,6 +357,7 @@ def from_dict(d):
             LimitUpLimitDown,
             Level2Book,
             IndexValue,
+            LaunchpadValue,
         ]
     ],
 )