[codex] Use source tick batches inside Intrade

guides/api-and-header-contracts.md

@@ -183,6 +183,9 @@ Contract rules:
   Shared stream metadata (`symbol`, `timeframe`, digits, subscription handle)
   lives on the batch; individual `Tick`/`Bar` payloads keep only price/time data
   plus compact `flags`.
+- Intrade websocket parsers and polling managers exchange
+  `events::TickUpdateBatch` directly. `SingleTick` remains only in the legacy
+  `request_price()` and typed `PriceSnapshot` compatibility surface.
 - Live data callbacks are flushed from the provider/platform lifecycle
   (`process()` or the worker loop started by `run()`), after queued price events
   are routed and coalesced. Calling `event_bus().drain()` alone is an internal

include/optionx_cpp/platforms/IntradeBarPlatform/BtcPriceManager.hpp

@@ -28,11 +28,6 @@ namespace optionx::platforms::intrade_bar {
             subscribe<events::AutoDomainSelectedEvent>();
             platform.register_component(this);
 
-            m_tick_data.resize(1);
-            m_tick_data[0].price_digits  = 2;
-            m_tick_data[0].volume_digits = 5;
-            m_tick_data[0].symbol = "BTCUSDT";
-            m_tick_data[0].provider = to_str(PlatformType::INTRADE_BAR);
             m_websocket_client.set_url(m_ws_host, "/bapi");
             m_websocket_client.set_user_agent(OPTIONX_DEFAULT_BROWSER_USER_AGENT);
             m_websocket_client.set_accept_language(OPTIONX_DEFAULT_ACCEPT_LANGUAGE);
@@ -45,7 +40,6 @@ namespace optionx::platforms::intrade_bar {
                 switch (event->event_type) {
                 case kurlyk::WebSocketEventType::WS_OPEN:
                     LOGIT_INFO(event->status_code, event->error_code);
-                    m_tick_data[0].tick.flags = 0;
                     m_is_error = false;
                     emit_status(market_data::MarketDataStreamStatus::CONNECTED);
                     // `/bapi` is a fixed BTCUSDT stream; no subscribe frame is needed.
@@ -56,14 +50,12 @@ namespace optionx::platforms::intrade_bar {
                     break;
                 case kurlyk::WebSocketEventType::WS_CLOSE:
                     LOGIT_INFO(event->status_code, event->error_code);
-                    m_tick_data[0].tick.flags = 0;
                     m_is_error = false;
                     emit_status(market_data::MarketDataStreamStatus::DISCONNECTED);
                     break;
                 case kurlyk::WebSocketEventType::WS_ERROR:
                     if (m_is_error) return;
                     LOGIT_ERROR(event->status_code, event->error_code);
-                    m_tick_data[0].tick.flags = 0;
                     m_is_error = true;
                     emit_status(
                         market_data::MarketDataStreamStatus::FAILED,
@@ -111,8 +103,7 @@ namespace optionx::platforms::intrade_bar {
     private:
         kurlyk::WebSocketClient m_websocket_client; ///< WebSocket client for BTCUSDT.
         std::string m_ws_host = make_websocket_host(AuthData{}.host); ///< Websocket host.
-        std::vector<SingleTick>   m_tick_data;        ///< Container for tick data.
-        bool                    m_is_error = false; ///< Flag indicating if an error has occurred.
+        bool m_is_error = false; ///< Flag indicating if an error has occurred.
         std::mutex m_source_mutex; ///< Protects subscription-driven source state.
         std::size_t m_market_data_ref_count = 0; ///< Public market-data subscriptions using BTC ticks.
         bool m_platform_connected = false; ///< Whether trading lifecycle wants the BTC stream connected.
@@ -285,14 +276,10 @@ namespace optionx::platforms::intrade_bar {
     }
 
     inline void BtcPriceManager::handle_message(const std::string& message) {
-        if (parse_btcusdt_tick(message, m_tick_data[0])) {
+        events::TickUpdateBatch batch;
+        if (parse_btcusdt_tick(message, batch)) {
             std::vector<events::TickUpdateBatch> batches;
-            batches.push_back(events::PriceUpdateEvent::make_tick_batch(
-                m_tick_data[0].tick,
-                m_tick_data[0].symbol,
-                m_tick_data[0].provider,
-                m_tick_data[0].price_digits,
-                m_tick_data[0].volume_digits));
+            batches.push_back(std::move(batch));
             notify_async(std::make_unique<events::PriceUpdateEvent>(
                 std::move(batches),
                 MarketDataUpdateSource::WEBSOCKET));
@@ -332,7 +319,6 @@ namespace optionx::platforms::intrade_bar {
             m_platform_connected = false;
         }
         m_websocket_client.disconnect_and_wait();
-        m_tick_data[0].tick.flags = 0;
     }
 
     inline bool BtcPriceManager::should_connect_no_lock() const noexcept {

include/optionx_cpp/platforms/IntradeBarPlatform/FxPriceWebSocketManager.hpp

@@ -443,17 +443,12 @@ namespace optionx::platforms::intrade_bar {
             const std::shared_ptr<FxStreamState>& stream,
             const std::string& message) {
         try {
-            SingleTick tick;
-            if (!parse_fxconnect_tick(message, tick)) return;
-            if (tick.symbol != stream->symbol) return;
+            events::TickUpdateBatch batch;
+            if (!parse_fxconnect_tick(message, batch)) return;
+            if (batch.symbol != stream->symbol) return;
 
             std::vector<events::TickUpdateBatch> batches;
-            batches.push_back(events::PriceUpdateEvent::make_tick_batch(
-                tick.tick,
-                tick.symbol,
-                tick.provider,
-                tick.price_digits,
-                tick.volume_digits));
+            batches.push_back(std::move(batch));
             notify_async(std::make_unique<events::PriceUpdateEvent>(
                 std::move(batches),
                 MarketDataUpdateSource::WEBSOCKET));

include/optionx_cpp/platforms/IntradeBarPlatform/PriceManager.hpp

@@ -45,7 +45,7 @@ namespace optionx::platforms::intrade_bar {
     private:
         RequestManager&    m_request_manager; ///< Reference to the request manager.
         utils::TaskManager m_task_manager;    ///< Task manager for handling asynchronous tasks.
-        std::unordered_map<std::string, SingleTick> m_ticks; ///< Stores the latest tick data for each symbol.
+        std::unordered_map<std::string, Tick> m_ticks; ///< Latest tick payload by symbol.
         bool m_has_price_update = false; ///< Flag indicating whether a price update is in progress.
 
         /// \brief Initiates the process of retrieving price updates.
@@ -127,9 +127,9 @@ namespace optionx::platforms::intrade_bar {
         if (m_has_price_update) return;
         m_has_price_update = true;
         LOGIT_DEBUG("Intrade Bar price: requesting price snapshot.");
-        m_request_manager.request_price([this, task](
+        m_request_manager.request_price_batches([this, task](
                 bool success,
-                std::vector<SingleTick> ticks) {
+                std::vector<events::TickUpdateBatch> batches) {
             m_has_price_update = false;
             if (task->is_shutdown()) {
                 m_ticks.clear();
@@ -142,35 +142,34 @@ namespace optionx::platforms::intrade_bar {
             }
 
             task->set_period(time_shield::MS_PER_SEC);
-            LOGIT_DEBUG("Intrade Bar price: snapshot received. ticks=", ticks.size());
-
-            for (auto& tick : ticks) {
-                auto it = m_ticks.find(tick.symbol);
-                if (it == m_ticks.end()) {
-                    tick.tick.set_flag(TickUpdateFlags::ASK_UPDATED);
-                    tick.tick.set_flag(TickUpdateFlags::BID_UPDATED);
-                    m_ticks[tick.symbol] = tick;
-                } else {
-                    if (!utils::compare_with_precision(it->second.tick.ask, tick.tick.ask, tick.price_digits)) {
-                        tick.tick.set_flag(TickUpdateFlags::ASK_UPDATED);
+            LOGIT_DEBUG("Intrade Bar price: snapshot received. batches=", batches.size());
+
+            for (auto& batch : batches) {
+                for (auto& tick : batch.items) {
+                    auto it = m_ticks.find(batch.symbol);
+                    if (it == m_ticks.end()) {
+                        tick.set_flag(TickUpdateFlags::ASK_UPDATED);
+                        tick.set_flag(TickUpdateFlags::BID_UPDATED);
+                        m_ticks[batch.symbol] = tick;
+                        continue;
                     }
-                    if (!utils::compare_with_precision(it->second.tick.bid, tick.tick.bid, tick.price_digits)) {
-                        tick.tick.set_flag(TickUpdateFlags::BID_UPDATED);
+
+                    if (!utils::compare_with_precision(
+                            it->second.ask,
+                            tick.ask,
+                            batch.price_digits)) {
+                        tick.set_flag(TickUpdateFlags::ASK_UPDATED);
+                    }
+                    if (!utils::compare_with_precision(
+                            it->second.bid,
+                            tick.bid,
+                            batch.price_digits)) {
+                        tick.set_flag(TickUpdateFlags::BID_UPDATED);
                     }
                     it->second = tick;
                 }
             }
 
-            std::vector<events::TickUpdateBatch> batches;
-            batches.reserve(ticks.size());
-            for (const auto& tick : ticks) {
-                batches.push_back(events::PriceUpdateEvent::make_tick_batch(
-                    tick.tick,
-                    tick.symbol,
-                    tick.provider,
-                    tick.price_digits,
-                    tick.volume_digits));
-            }
             notify(events::PriceUpdateEvent(std::move(batches), MarketDataUpdateSource::POLLING));
         });
     }

include/optionx_cpp/platforms/IntradeBarPlatform/RequestManager.hpp

@@ -221,13 +221,21 @@ namespace optionx::platforms::intrade_bar {
         void request_switch_currency_result(
             std::function<void(SettingsSwitchResult)> switch_callback);
 
-        /// \brief Requests the latest price updates.
-        /// \param price_callback Callback function to receive tick data.
+        /// \brief Requests the latest prices through the legacy per-tick DTO API.
+        /// \param price_callback Callback function to receive `SingleTick` values.
+        /// \note New internal consumers should use `request_price_batches()`.
         void request_price(
             std::function<void(
                 bool success,
                 std::vector<SingleTick> ticks)> price_callback);
 
+        /// \brief Requests the latest prices grouped by source metadata.
+        /// \param price_callback Callback function to receive source tick batches.
+        void request_price_batches(
+            std::function<void(
+                bool success,
+                std::vector<events::TickUpdateBatch> batches)> price_callback);
+
         /// \brief Typed variant of request_price.
         void request_price_result(
             std::function<void(PriceSnapshotResult)> price_callback);
@@ -943,6 +951,40 @@ namespace optionx::platforms::intrade_bar {
             std::function<void(
                 bool success,
                 std::vector<SingleTick> ticks)> price_callback) {
+        request_price_batches(
+            [price_callback = std::move(price_callback)](
+                    bool success,
+                    std::vector<events::TickUpdateBatch> batches) mutable {
+                if (!success) {
+                    price_callback(false, {});
+                    return;
+                }
+
+                std::size_t tick_count = 0;
+                for (const auto& batch : batches) {
+                    tick_count += batch.items.size();
+                }
+
+                std::vector<SingleTick> ticks;
+                ticks.reserve(tick_count);
+                for (auto& batch : batches) {
+                    for (auto& tick : batch.items) {
+                        ticks.emplace_back(
+                            std::move(tick),
+                            batch.symbol,
+                            batch.provider,
+                            batch.price_digits,
+                            batch.volume_digits);
+                    }
+                }
+                price_callback(true, std::move(ticks));
+            });
+    }
+
+    inline void RequestManager::request_price_batches(
+            std::function<void(
+                bool success,
+                std::vector<events::TickUpdateBatch> batches)> price_callback) {
         // Отправка GET-запроса
         auto future = get_http_client().get(
             "/price_now",
@@ -958,33 +1000,11 @@ namespace optionx::platforms::intrade_bar {
                 return;
             }
 
-            using json = nlohmann::json;
-            int64_t received_ms = OPTIONX_TIMESTAMP_MS;
-            std::vector<SingleTick> ticks;
             try {
-                json j = json::parse(response->content); // Парсинг JSON
-                for (auto& el : j.items()) {
-                    const std::string symbol_name = el.key();
-                    SingleTick tick;
-                    tick.provider = to_str(PlatformType::INTRADE_BAR);
-                    tick.symbol = normalize_symbol_name(symbol_name);
-                    tick.volume_digits = 0;
-
-                    tick.price_digits = price_digits_for_symbol(tick.symbol);
-
-                    tick.tick.ask = el.value()["ask"];
-                    tick.tick.bid = el.value()["bid"];
-                    tick.tick.last = 0.0;
-                    tick.tick.time_ms = el.value()["Updates"];
-                    tick.tick.time_ms = time_shield::sec_to_ms(tick.tick.time_ms);
-                    tick.tick.received_ms = received_ms;
-                    tick.tick.set_flag(TickUpdateFlags::NONE);
-                    tick.tick.set_flag(MarketDataFlags::INITIALIZED);
-                    tick.tick.set_flag(MarketDataFlags::REALTIME);
-                    ticks.push_back(std::move(tick));
-                }
-
-                price_callback(true, std::move(ticks));
+                auto batches = parse_price_snapshot_response(
+                    response->content,
+                    static_cast<std::uint64_t>(OPTIONX_TIMESTAMP_MS));
+                price_callback(true, std::move(batches));
             } catch (const std::exception& ex) {
                 LOGIT_ERROR("Error parsing price response: ", ex.what());
                 price_callback(false, {});