feat(stream): add RTDS and Sports WebSocket streams

CMakeLists.txt

@@ -169,6 +169,13 @@ set(POLYMARKET_CLIENT_SOURCES
     src/user_stream_runtime.cpp
     src/user_stream_runtime_events.cpp
     src/user_stream_subscription.cpp
+    src/public_stream.cpp
+    src/rtds_parsing.cpp
+    src/rtds_price_parsing.cpp
+    src/rtds_stream.cpp
+    src/sports_parsing.cpp
+    src/sports_stream.cpp
+    src/stream_time.cpp
     src/order_amounts.cpp
     src/clob_order_execution.cpp
     src/clob_order_submission.cpp

cmake/PolymarketTests.cmake

@@ -265,6 +265,34 @@ if(POLYMARKET_CLIENT_BUILD_TESTS)
     target_link_libraries(test_websocket_resilience PRIVATE polymarket::client)
     add_test(NAME test_websocket_resilience COMMAND test_websocket_resilience)
 
+    add_executable(test_websocket_heartbeat tests/test_websocket_heartbeat.cpp)
+    target_include_directories(test_websocket_heartbeat PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/src)
+    target_link_libraries(test_websocket_heartbeat PRIVATE polymarket::client)
+    add_test(NAME test_websocket_heartbeat COMMAND test_websocket_heartbeat)
+
+    add_executable(test_sports_parsing tests/test_sports_parsing.cpp)
+    target_include_directories(test_sports_parsing PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/src)
+    target_link_libraries(test_sports_parsing PRIVATE polymarket::client)
+    add_test(NAME test_sports_parsing COMMAND test_sports_parsing)
+
+    add_executable(test_sports_stream tests/test_sports_stream.cpp)
+    target_link_libraries(test_sports_stream PRIVATE polymarket::client)
+    add_test(NAME test_sports_stream COMMAND test_sports_stream)
+
+    add_executable(test_rtds_parsing tests/test_rtds_parsing.cpp)
+    target_include_directories(test_rtds_parsing PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/src)
+    target_link_libraries(test_rtds_parsing PRIVATE polymarket::client)
+    add_test(NAME test_rtds_parsing COMMAND test_rtds_parsing)
+
+    add_executable(test_rtds_stream tests/test_rtds_stream.cpp)
+    target_link_libraries(test_rtds_stream PRIVATE polymarket::client)
+    add_test(NAME test_rtds_stream COMMAND test_rtds_stream)
+
+    add_executable(test_rtds_price_parsing tests/test_rtds_price_parsing.cpp)
+    target_include_directories(test_rtds_price_parsing PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/src)
+    target_link_libraries(test_rtds_price_parsing PRIVATE polymarket::client)
+    add_test(NAME test_rtds_price_parsing COMMAND test_rtds_price_parsing)
+
     add_executable(test_network_routing tests/test_network_routing.cpp)
     target_link_libraries(test_network_routing PRIVATE polymarket::client)
     add_test(NAME test_network_routing COMMAND test_network_routing)

include/polymarket/environment.hpp

@@ -26,6 +26,7 @@ namespace polymarket
         std::string data_url;
         std::string relayer_url;
         std::string rtds_ws_url;
+        std::string sports_ws_url;
 
         static Environment production();
         static Environment preproduction();

include/polymarket/rtds_stream.hpp

@@ -0,0 +1,162 @@
+#pragma once
+
+#include "polymarket/gamma_social_types.hpp"
+#include "polymarket/types.hpp"
+
+#include <nlohmann/json.hpp>
+
+#include <cstdint>
+#include <functional>
+#include <memory>
+#include <optional>
+#include <string>
+#include <vector>
+
+namespace polymarket
+{
+    namespace detail
+    {
+        class RtdsStreamRuntime;
+    }
+
+    // Comment events to receive. `types` holds comment_created (the default
+    // when empty), comment_removed, reaction_created or reaction_removed.
+    // A parent filter keeps comments on that event or market and drops
+    // reactions, which carry no parent.
+    struct RtdsCommentFilter
+    {
+        std::vector<std::string> types;
+        std::optional<std::int64_t> parent_entity_id;
+        std::string parent_entity_type; // Event or Market
+    };
+
+    // One comment event, following the official py-sdk models. `comment` is
+    // set for comment_created and comment_removed, `reaction` for the
+    // reaction events. `raw` holds the whole frame as received.
+    struct RtdsCommentEvent
+    {
+        std::string type;
+        std::optional<std::int64_t> timestamp_ms; // Unix milliseconds
+        std::optional<GammaComment> comment;
+        std::optional<GammaReaction> reaction;
+        nlohmann::json raw;
+    };
+
+    // Live crypto prices from one source: binance (USDT pairs such as btcusdt)
+    // or chainlink (pairs such as btc/usd). Every symbol when empty.
+    struct RtdsCryptoPriceFilter
+    {
+        std::string source;
+        std::vector<std::string> symbols;
+    };
+
+    // Chainlink TWAPs over a 30- or 60-second window. Every symbol when empty.
+    struct RtdsCryptoTwapFilter
+    {
+        int window_seconds{60};
+        std::vector<std::string> symbols;
+    };
+
+    // Pyth prices for one symbol, matched case-insensitively. `types` holds
+    // update and/or subscribe (history snapshots); both when empty.
+    struct RtdsEquityPriceFilter
+    {
+        std::string symbol;
+        std::vector<std::string> types;
+    };
+
+    // `exact_value` is the most precise decimal the frame carries
+    // (full_accuracy_value, with Chainlink's 18-decimal integers scaled);
+    // `value` is the same number as a double.
+    struct RtdsPricePoint
+    {
+        std::int64_t timestamp_ms{0}; // Unix milliseconds
+        double value{0.0};
+        std::string exact_value;
+    };
+
+    // One price event, following the official py-sdk models. `topic` is
+    // prices.crypto.binance, prices.crypto.chainlink,
+    // prices.crypto.chainlink.twap or prices.equity.pyth. An update carries
+    // `price`; an equity subscribe event carries `history` instead. Equity
+    // prices use the instrument's quote currency; crypto prices are USD, or
+    // USDT for Binance. `raw` holds the whole frame as received.
+    struct RtdsPriceEvent
+    {
+        std::string topic;
+        std::string type; // update or subscribe
+        std::optional<std::int64_t> timestamp_ms;
+        std::string symbol;
+        RtdsPricePoint price;
+        std::optional<int> window_seconds;       // TWAP
+        std::optional<std::int64_t> received_at; // equity, as sent
+        std::optional<bool> is_carried_forward;  // equity
+        std::vector<RtdsPricePoint> history;     // equity subscribe
+        nlohmann::json raw;
+    };
+
+    using RtdsSubscriptionId = std::uint64_t;
+    using RtdsCommentCallback = std::function<void(const RtdsCommentEvent &event)>;
+    using RtdsPriceCallback = std::function<void(const RtdsPriceEvent &event)>;
+    // Fired whenever events may have been missed: every connect, disconnect,
+    // queue overflow, idle reconnect or malformed event. RTDS does not replay
+    // them.
+    using RtdsStreamGapCallback = std::function<void()>;
+    // Fired once the connection and its subscriptions are back after a gap.
+    using RtdsStreamRecoveredCallback = std::function<void()>;
+    using RtdsStreamErrorCallback = std::function<void(const std::string &error)>;
+
+    // Public real-time data stream (Config::rtds_ws_url). Subscriptions can
+    // be added and removed before or after connect(); the stream sends the
+    // combined server subscription and restores it after every reconnect.
+    // It pings every 5 seconds and reconnects when no frame arrives for 10
+    // minutes. Callbacks run on the stream's own threads. Malformed and
+    // unknown frames are counted in dropped_events(). A malformed frame on a
+    // comment or price topic is also reported to on_error() with its cause
+    // and followed by a reconnect; other frames are skipped.
+    class RtdsStream
+    {
+      public:
+        explicit RtdsStream(const Config &config = Config{});
+        ~RtdsStream();
+
+        RtdsStream(const RtdsStream &) = delete;
+        RtdsStream &operator=(const RtdsStream &) = delete;
+
+        // Throws std::invalid_argument for an unknown type or parent type.
+        RtdsSubscriptionId subscribe_comments(const RtdsCommentFilter &filter,
+                                              RtdsCommentCallback callback);
+        // Price topics are subscribed whole and narrowed here, as the official
+        // SDKs do, because RTDS keeps one filter per topic and connection.
+        // The official SDKs deprecate these topics in favour of the
+        // authenticated price service. Throw std::invalid_argument for an
+        // unknown source, window or type, or an empty symbol.
+        RtdsSubscriptionId subscribe_crypto_prices(const RtdsCryptoPriceFilter &filter,
+                                                   RtdsPriceCallback callback);
+        RtdsSubscriptionId subscribe_crypto_twap(const RtdsCryptoTwapFilter &filter,
+                                                 RtdsPriceCallback callback);
+        RtdsSubscriptionId subscribe_equity_prices(const RtdsEquityPriceFilter &filter,
+                                                   RtdsPriceCallback callback);
+        // Unknown IDs are ignored.
+        void unsubscribe(RtdsSubscriptionId id);
+
+        void on_stream_gap(RtdsStreamGapCallback callback);
+        void on_stream_recovered(RtdsStreamRecoveredCallback callback);
+        void on_error(RtdsStreamErrorCallback callback);
+
+        // Waits up to Config::ws_connect_timeout_ms for the connection.
+        bool connect();
+        void disconnect();
+        bool is_connected() const;
+        // Blocks until stop().
+        void run();
+        void stop();
+
+        std::uint64_t events() const;
+        std::uint64_t dropped_events() const;
+
+      private:
+        std::shared_ptr<detail::RtdsStreamRuntime> runtime_;
+    };
+
+} // namespace polymarket

include/polymarket/sports_stream.hpp

@@ -0,0 +1,87 @@
+#pragma once
+
+#include "polymarket/types.hpp"
+
+#include <nlohmann/json.hpp>
+
+#include <cstdint>
+#include <functional>
+#include <memory>
+#include <optional>
+#include <string>
+
+namespace polymarket
+{
+    namespace detail
+    {
+        class SportsStreamRuntime;
+    }
+
+    // One live game update from the Sports WebSocket, following the official
+    // py-sdk model. Absent or empty text stays empty; `raw` holds the whole
+    // object as received.
+    struct SportsGameResult
+    {
+        std::int64_t game_id{0};
+        std::string sportradar_game_id;
+        std::string slug;
+        std::string league_abbreviation;
+        std::string home_team;
+        std::string away_team;
+        std::string status;
+        bool live{false};
+        bool ended{false};
+        std::string score;
+        std::string period;
+        std::string elapsed;
+        std::string turn;
+        std::optional<std::int64_t> finished_at_ms; // Unix milliseconds
+        nlohmann::json raw;
+    };
+
+    using SportsResultCallback = std::function<void(const SportsGameResult &result)>;
+    // Fired whenever updates may have been missed: every connect, disconnect,
+    // queue overflow, idle reconnect or malformed update. The server does not
+    // replay them.
+    using SportsStreamGapCallback = std::function<void()>;
+    // Fired once the connection is back after a gap; refresh game state from
+    // GammaClient here if it matters.
+    using SportsStreamRecoveredCallback = std::function<void()>;
+    using SportsStreamErrorCallback = std::function<void(const std::string &error)>;
+
+    // Public Sports stream (Config::sports_ws_url). Every connection receives
+    // every game update; there are no filters. The stream answers the
+    // server's heartbeat and reconnects when no frame arrives for 30 seconds.
+    // Callbacks run on the stream's own threads. A malformed update is
+    // counted in dropped_events(), reported to on_error() with its cause, and
+    // followed by a reconnect.
+    class SportsStream
+    {
+      public:
+        explicit SportsStream(const Config &config = Config{});
+        ~SportsStream();
+
+        SportsStream(const SportsStream &) = delete;
+        SportsStream &operator=(const SportsStream &) = delete;
+
+        void on_result(SportsResultCallback callback);
+        void on_stream_gap(SportsStreamGapCallback callback);
+        void on_stream_recovered(SportsStreamRecoveredCallback callback);
+        void on_error(SportsStreamErrorCallback callback);
+
+        // Waits up to Config::ws_connect_timeout_ms for the connection.
+        bool connect();
+        void disconnect();
+        bool is_connected() const;
+        // Blocks until stop().
+        void run();
+        void stop();
+
+        std::uint64_t result_events() const;
+        std::uint64_t dropped_events() const;
+
+      private:
+        std::shared_ptr<detail::SportsStreamRuntime> runtime_;
+    };
+
+} // namespace polymarket

include/polymarket/types.hpp

@@ -264,6 +264,7 @@ namespace polymarket
         std::string clob_ws_url = "wss://ws-subscriptions-clob.polymarket.com/ws/market";
         std::string gamma_api_url = "https://gamma-api.polymarket.com";
         std::string rtds_ws_url = "wss://ws-live-data.polymarket.com";
+        std::string sports_ws_url = "wss://sports-api.polymarket.com/ws";
         std::string clob_user_ws_url = "wss://ws-subscriptions-clob.polymarket.com/ws/user";
 
         // Trading parameters
@@ -283,8 +284,8 @@ namespace polymarket
             "btc", "eth", "xrp", "sol", "doge", "bnb",
             "ada", "avax", "matic", "link", "dot", "ltc"};
 
-        // CLOB, Gamma and RTDS endpoints from `environment`, which is validated
-        // first; every other field keeps its default.
+        // CLOB, Gamma, RTDS and Sports endpoints from `environment`, which is
+        // validated first; every other field keeps its default.
         static Config for_environment(const Environment &environment);
     };
 

include/polymarket/websocket_client.hpp

@@ -38,6 +38,15 @@ namespace polymarket
         uint32_t min_backoff_ms{250};
         uint32_t max_backoff_ms{10000};
         int ping_interval_ms{10000};
+        // Text sent every ping_interval_ms; empty sends no client pings.
+        std::string ping_message{"PING"};
+        // A text frame equal to server_ping_message is answered with
+        // server_ping_reply and not delivered. Both set, or both empty.
+        std::string server_ping_message;
+        std::string server_ping_reply;
+        // Reconnects, reporting a stream gap, when no frame arrives for this
+        // long while connected; 0 disables.
+        uint32_t idle_timeout_ms{0};
         std::size_t message_queue_limit{1024};
         // Route for this connection; empty fields fall back to the default
         // route in <polymarket/network.hpp>. A set route sends the connection
@@ -55,6 +64,7 @@ namespace polymarket
         uint64_t parse_errors{0};
         uint64_t last_message_time_ns{0};
         uint64_t callback_errors{0};
+        uint64_t idle_timeouts{0};
     };
 
     struct TypedWebSocketMessage

src/environment.cpp

@@ -38,6 +38,7 @@ namespace polymarket
         env.data_url = "https://data-api.polymarket.com";
         env.relayer_url = "https://relayer-v2.polymarket.com";
         env.rtds_ws_url = "wss://ws-live-data.polymarket.com";
+        env.sports_ws_url = "wss://sports-api.polymarket.com/ws";
         return env;
     }
 
@@ -75,6 +76,7 @@ namespace polymarket
             {"clob_market_ws_url", &clob_market_ws_url},
             {"clob_user_ws_url", &clob_user_ws_url},
             {"rtds_ws_url", &rtds_ws_url},
+            {"sports_ws_url", &sports_ws_url},
         };
         for (const auto &[field, url] : ws_urls)
             require_url(field, *url, "wss://", "ws://");
@@ -91,6 +93,7 @@ namespace polymarket
         config.clob_user_ws_url = environment.clob_user_ws_url;
         config.gamma_api_url = environment.gamma_url;
         config.rtds_ws_url = environment.rtds_ws_url;
+        config.sports_ws_url = environment.sports_ws_url;
         return config;
     }
 

src/gamma_market_parsing.cpp

@@ -30,14 +30,20 @@ namespace polymarket::detail
 
         GammaEventReference event_reference(const json &item)
         {
-            return {field_required_id(item, "id"), field_string(item, "slug"),
-                    field_string(item, "title")};
+            GammaEventReference reference;
+            reference.id = field_required_id(item, "id");
+            reference.slug = field_string(item, "slug");
+            reference.title = field_string(item, "title");
+            return reference;
         }
 
         GammaSeriesReference series_reference(const json &item)
         {
-            return {field_required_id(item, "id"), field_string(item, "slug"),
-                    field_string(item, "title")};
+            GammaSeriesReference reference;
+            reference.id = field_required_id(item, "id");
+            reference.slug = field_string(item, "slug");
+            reference.title = field_string(item, "title");
+            return reference;
         }
     } // namespace