From 19aba0ac7c8766d10e9c0f75d53e7d53ce11be67 Mon Sep 17 00:00:00 2001 From: Aster Seker Date: Thu, 27 Aug 2026 14:46:32 +0300 Subject: [PATCH] refactor(intrade-bar): use source tick batches internally Parse websocket and polling payloads directly into TickUpdateBatch values and carry those batches through BTC, FX, and polling managers. Keep request_price and PriceSnapshot as explicit SingleTick compatibility adapters, and add parser regression coverage. --- guides/api-and-header-contracts.md | 3 + .../IntradeBarPlatform/BtcPriceManager.hpp | 22 +--- .../FxPriceWebSocketManager.hpp | 13 +- .../IntradeBarPlatform/PriceManager.hpp | 51 ++++---- .../IntradeBarPlatform/RequestManager.hpp | 76 +++++++----- .../IntradeBarPlatform/http_parsers.hpp | 108 ++++++++++------ tests/intrade_bar_api/WORKFLOWS.md | 2 +- .../intrade_bar_api_response_test.cpp | 117 ++++++++++++------ 8 files changed, 237 insertions(+), 155 deletions(-) diff --git a/guides/api-and-header-contracts.md b/guides/api-and-header-contracts.md index c1b84885..99116e93 100644 --- a/guides/api-and-header-contracts.md +++ b/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 diff --git a/include/optionx_cpp/platforms/IntradeBarPlatform/BtcPriceManager.hpp b/include/optionx_cpp/platforms/IntradeBarPlatform/BtcPriceManager.hpp index 023ccf52..730917c3 100644 --- a/include/optionx_cpp/platforms/IntradeBarPlatform/BtcPriceManager.hpp +++ b/include/optionx_cpp/platforms/IntradeBarPlatform/BtcPriceManager.hpp @@ -28,11 +28,6 @@ namespace optionx::platforms::intrade_bar { subscribe(); 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 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 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( 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 { diff --git a/include/optionx_cpp/platforms/IntradeBarPlatform/FxPriceWebSocketManager.hpp b/include/optionx_cpp/platforms/IntradeBarPlatform/FxPriceWebSocketManager.hpp index 61974d77..97265199 100644 --- a/include/optionx_cpp/platforms/IntradeBarPlatform/FxPriceWebSocketManager.hpp +++ b/include/optionx_cpp/platforms/IntradeBarPlatform/FxPriceWebSocketManager.hpp @@ -443,17 +443,12 @@ namespace optionx::platforms::intrade_bar { const std::shared_ptr& 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 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( std::move(batches), MarketDataUpdateSource::WEBSOCKET)); diff --git a/include/optionx_cpp/platforms/IntradeBarPlatform/PriceManager.hpp b/include/optionx_cpp/platforms/IntradeBarPlatform/PriceManager.hpp index f9f82edf..e2a21a42 100644 --- a/include/optionx_cpp/platforms/IntradeBarPlatform/PriceManager.hpp +++ b/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 m_ticks; ///< Stores the latest tick data for each symbol. + std::unordered_map 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 ticks) { + std::vector 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 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)); }); } diff --git a/include/optionx_cpp/platforms/IntradeBarPlatform/RequestManager.hpp b/include/optionx_cpp/platforms/IntradeBarPlatform/RequestManager.hpp index 97f78f12..b5076927 100644 --- a/include/optionx_cpp/platforms/IntradeBarPlatform/RequestManager.hpp +++ b/include/optionx_cpp/platforms/IntradeBarPlatform/RequestManager.hpp @@ -221,13 +221,21 @@ namespace optionx::platforms::intrade_bar { void request_switch_currency_result( std::function 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 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 batches)> price_callback); + /// \brief Typed variant of request_price. void request_price_result( std::function price_callback); @@ -943,6 +951,40 @@ namespace optionx::platforms::intrade_bar { std::function ticks)> price_callback) { + request_price_batches( + [price_callback = std::move(price_callback)]( + bool success, + std::vector 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 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 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 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(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, {}); diff --git a/include/optionx_cpp/platforms/IntradeBarPlatform/http_parsers.hpp b/include/optionx_cpp/platforms/IntradeBarPlatform/http_parsers.hpp index 718d5075..85cbcdd2 100644 --- a/include/optionx_cpp/platforms/IntradeBarPlatform/http_parsers.hpp +++ b/include/optionx_cpp/platforms/IntradeBarPlatform/http_parsers.hpp @@ -1461,33 +1461,69 @@ namespace optionx::platforms::intrade_bar { return sequence; } - /// \brief Parses a BTCUSDT tick message from the WebSocket and updates the provided SingleTick structure. + /// \brief Parses the Intrade `/price_now` response into source tick batches. + /// \param content JSON object keyed by broker symbol. + /// \param received_ms Local receipt timestamp in milliseconds. + /// \return Tick batches grouped by normalized symbol and precision metadata. + inline std::vector parse_price_snapshot_response( + const std::string& content, + std::uint64_t received_ms) { + const auto j = nlohmann::json::parse(content); + std::vector batches; + batches.reserve(j.size()); + + for (const auto& el : j.items()) { + const auto symbol = normalize_symbol_name(el.key()); + Tick tick; + tick.ask = el.value().at("ask").get(); + tick.bid = el.value().at("bid").get(); + tick.time_ms = time_shield::sec_to_ms( + static_cast( + detail::read_json_int64(el.value().at("Updates"), "Updates"))); + tick.received_ms = received_ms; + tick.set_flag(MarketDataFlags::INITIALIZED); + tick.set_flag(MarketDataFlags::REALTIME); + batches.push_back(events::PriceUpdateEvent::make_tick_batch( + std::move(tick), + symbol, + to_str(PlatformType::INTRADE_BAR), + price_digits_for_symbol(symbol), + 0)); + } + + return batches; + } + + /// \brief Parses a BTCUSDT tick message into a source batch. /// \param message The JSON-formatted string containing the tick data. - /// \param tick_data Reference to the SingleTick structure to be updated. + /// \param batch Destination batch to fill. /// \return true if parsing is successful and the symbol matches BTCUSDT; false otherwise. - inline bool parse_btcusdt_tick(const std::string& message, SingleTick& tick_data) { + inline bool parse_btcusdt_tick( + const std::string& message, + events::TickUpdateBatch& batch) { auto j = nlohmann::json::parse(message); if (j.contains("data")) { const auto& j_data = j["data"]; if (j_data.value("s", "") != "BTCUSDT") return false; if (!j_data.contains("T")) return false; - tick_data.symbol = "BTCUSDT"; - tick_data.provider = to_str(PlatformType::INTRADE_BAR); - tick_data.price_digits = 2; - tick_data.volume_digits = 5; - tick_data.tick.flags = 0; - tick_data.tick.ask = 0.0; - tick_data.tick.bid = 0.0; - tick_data.tick.last = std::stod(j_data.value("p", "0.0")); - tick_data.tick.volume = std::stod(j_data.value("q", "0.0")); - tick_data.tick.time_ms = static_cast( + + Tick tick; + tick.last = std::stod(j_data.value("p", "0.0")); + tick.volume = std::stod(j_data.value("q", "0.0")); + tick.time_ms = static_cast( detail::read_json_int64(j_data.at("T"), "T")); - tick_data.tick.received_ms = OPTIONX_TIMESTAMP_MS; - tick_data.tick.set_flag(TickUpdateFlags::LAST_UPDATED); - tick_data.tick.set_flag(TickUpdateFlags::VOLUME_UPDATED); - tick_data.tick.set_flag(MarketDataFlags::INITIALIZED); - tick_data.tick.set_flag(MarketDataFlags::REALTIME); + tick.received_ms = OPTIONX_TIMESTAMP_MS; + tick.set_flag(TickUpdateFlags::LAST_UPDATED); + tick.set_flag(TickUpdateFlags::VOLUME_UPDATED); + tick.set_flag(MarketDataFlags::INITIALIZED); + tick.set_flag(MarketDataFlags::REALTIME); + batch = events::PriceUpdateEvent::make_tick_batch( + std::move(tick), + "BTCUSDT", + to_str(PlatformType::INTRADE_BAR), + 2, + 5); return true; } @@ -1496,9 +1532,11 @@ namespace optionx::platforms::intrade_bar { /// \brief Parses an Intrade `/fxconnect` tick message. /// \param message JSON-formatted message received from the FX websocket. - /// \param tick_data Destination tick object to fill. + /// \param batch Destination source batch to fill. /// \return True when the payload contains a supported FX symbol and bid/ask prices. - inline bool parse_fxconnect_tick(const std::string& message, SingleTick& tick_data) { + inline bool parse_fxconnect_tick( + const std::string& message, + events::TickUpdateBatch& batch) { const auto j = nlohmann::json::parse(message); if (!j.contains("symbol") || !j.contains("ask") || !j.contains("bid")) { @@ -1510,24 +1548,24 @@ namespace optionx::platforms::intrade_bar { return false; } - tick_data.symbol = normalized_symbol; - tick_data.provider = to_str(PlatformType::INTRADE_BAR); - tick_data.price_digits = price_digits_for_symbol(normalized_symbol); - tick_data.volume_digits = 0; - tick_data.tick.flags = 0; - tick_data.tick.ask = detail::read_json_double(j.at("ask"), "ask"); - tick_data.tick.bid = detail::read_json_double(j.at("bid"), "bid"); - tick_data.tick.last = 0.0; - tick_data.tick.volume = 0.0; - tick_data.tick.time_ms = j.contains("Updates") + Tick tick; + tick.ask = detail::read_json_double(j.at("ask"), "ask"); + tick.bid = detail::read_json_double(j.at("bid"), "bid"); + tick.time_ms = j.contains("Updates") ? static_cast( detail::read_json_int64(j.at("Updates"), "Updates")) * time_shield::MS_PER_SEC : 0; - tick_data.tick.received_ms = OPTIONX_TIMESTAMP_MS; - tick_data.tick.set_flag(TickUpdateFlags::ASK_UPDATED); - tick_data.tick.set_flag(TickUpdateFlags::BID_UPDATED); - tick_data.tick.set_flag(MarketDataFlags::INITIALIZED); - tick_data.tick.set_flag(MarketDataFlags::REALTIME); + tick.received_ms = OPTIONX_TIMESTAMP_MS; + tick.set_flag(TickUpdateFlags::ASK_UPDATED); + tick.set_flag(TickUpdateFlags::BID_UPDATED); + tick.set_flag(MarketDataFlags::INITIALIZED); + tick.set_flag(MarketDataFlags::REALTIME); + batch = events::PriceUpdateEvent::make_tick_batch( + std::move(tick), + normalized_symbol, + to_str(PlatformType::INTRADE_BAR), + price_digits_for_symbol(normalized_symbol), + 0); return true; } diff --git a/tests/intrade_bar_api/WORKFLOWS.md b/tests/intrade_bar_api/WORKFLOWS.md index e3c2d140..fc1f850f 100644 --- a/tests/intrade_bar_api/WORKFLOWS.md +++ b/tests/intrade_bar_api/WORKFLOWS.md @@ -105,7 +105,7 @@ counters are event-loop-owned state. Started after `AccountInfoUpdateEvent::CONNECTED`. -1. `PriceManager` calls `request_price` every second. +1. `PriceManager` calls batch-native `request_price_batches` every second. 2. Failed price request slows the task to five minutes. 3. Successful price request restores the one-second period and publishes `PriceUpdateEvent`. diff --git a/tests/intrade_bar_api/intrade_bar_api_response_test.cpp b/tests/intrade_bar_api/intrade_bar_api_response_test.cpp index 4b111bd5..1c963c78 100644 --- a/tests/intrade_bar_api/intrade_bar_api_response_test.cpp +++ b/tests/intrade_bar_api/intrade_bar_api_response_test.cpp @@ -1380,66 +1380,107 @@ TEST(IntradeBarSymbols, FormatsFxConnectSymbols) { EXPECT_EQ(make_fxconnect_symbol(""), ""); } +TEST(IntradeBarApiResponses, ParsesPriceSnapshotIntoSourceBatches) { + const std::string content = + R"({"EUR/USD":{"Updates":1783028728,"ask":1.17234,"bid":1.17231},"BTC/USD":{"Updates":1783028729,"ask":61521.35,"bid":61521.34}})"; + constexpr std::uint64_t received_ms = 1783028728123ULL; + + const auto batches = parse_price_snapshot_response(content, received_ms); + ASSERT_EQ(batches.size(), 2u); + + const auto fx_it = std::find_if( + batches.begin(), + batches.end(), + [](const auto& batch) { return batch.symbol == "EURUSD"; }); + ASSERT_NE(fx_it, batches.end()); + EXPECT_EQ(fx_it->provider, to_str(PlatformType::INTRADE_BAR)); + EXPECT_EQ(fx_it->price_digits, 5u); + EXPECT_EQ(fx_it->volume_digits, 0u); + ASSERT_EQ(fx_it->items.size(), 1u); + EXPECT_DOUBLE_EQ(fx_it->items[0].ask, 1.17234); + EXPECT_DOUBLE_EQ(fx_it->items[0].bid, 1.17231); + EXPECT_EQ(fx_it->items[0].time_ms, 1783028728000ULL); + EXPECT_EQ(fx_it->items[0].received_ms, received_ms); + EXPECT_TRUE(fx_it->items[0].has_flag(MarketDataFlags::INITIALIZED)); + EXPECT_TRUE(fx_it->items[0].has_flag(MarketDataFlags::REALTIME)); + EXPECT_FALSE(fx_it->items[0].has_flag(TickUpdateFlags::ASK_UPDATED)); + EXPECT_FALSE(fx_it->items[0].has_flag(TickUpdateFlags::BID_UPDATED)); + + const auto btc_it = std::find_if( + batches.begin(), + batches.end(), + [](const auto& batch) { return batch.symbol == "BTCUSDT"; }); + ASSERT_NE(btc_it, batches.end()); + EXPECT_EQ(btc_it->price_digits, 2u); + ASSERT_EQ(btc_it->items.size(), 1u); + EXPECT_DOUBLE_EQ(btc_it->items[0].ask, 61521.35); + EXPECT_DOUBLE_EQ(btc_it->items[0].bid, 61521.34); +} + TEST(IntradeBarApiResponses, ParsesFxConnectTickMessage) { const std::string message = R"({"Updates":1783028728,"ask":0.56971,"bid":0.5693,"symbol":"NZD\/USD"})"; - SingleTick tick; - ASSERT_TRUE(parse_fxconnect_tick(message, tick)); - - EXPECT_EQ(tick.symbol, "NZDUSD"); - EXPECT_EQ(tick.provider, to_str(PlatformType::INTRADE_BAR)); - EXPECT_EQ(tick.price_digits, 5u); - EXPECT_EQ(tick.volume_digits, 0u); - EXPECT_DOUBLE_EQ(tick.tick.ask, 0.56971); - EXPECT_DOUBLE_EQ(tick.tick.bid, 0.5693); - EXPECT_DOUBLE_EQ(tick.tick.volume, 0.0); - EXPECT_EQ(tick.tick.time_ms, 1783028728000ULL); - EXPECT_TRUE(tick.tick.has_flag(TickUpdateFlags::ASK_UPDATED)); - EXPECT_TRUE(tick.tick.has_flag(TickUpdateFlags::BID_UPDATED)); - EXPECT_FALSE(tick.tick.has_flag(TickUpdateFlags::VOLUME_UPDATED)); - EXPECT_TRUE(tick.tick.has_flag(MarketDataFlags::INITIALIZED)); - EXPECT_TRUE(tick.tick.has_flag(MarketDataFlags::REALTIME)); + events::TickUpdateBatch batch; + ASSERT_TRUE(parse_fxconnect_tick(message, batch)); + + EXPECT_EQ(batch.symbol, "NZDUSD"); + EXPECT_EQ(batch.provider, to_str(PlatformType::INTRADE_BAR)); + EXPECT_EQ(batch.price_digits, 5u); + EXPECT_EQ(batch.volume_digits, 0u); + ASSERT_EQ(batch.items.size(), 1u); + const auto& tick = batch.items[0]; + EXPECT_DOUBLE_EQ(tick.ask, 0.56971); + EXPECT_DOUBLE_EQ(tick.bid, 0.5693); + EXPECT_DOUBLE_EQ(tick.volume, 0.0); + EXPECT_EQ(tick.time_ms, 1783028728000ULL); + EXPECT_TRUE(tick.has_flag(TickUpdateFlags::ASK_UPDATED)); + EXPECT_TRUE(tick.has_flag(TickUpdateFlags::BID_UPDATED)); + EXPECT_FALSE(tick.has_flag(TickUpdateFlags::VOLUME_UPDATED)); + EXPECT_TRUE(tick.has_flag(MarketDataFlags::INITIALIZED)); + EXPECT_TRUE(tick.has_flag(MarketDataFlags::REALTIME)); } TEST(IntradeBarApiResponses, RejectsFxConnectBtcTickMessage) { const std::string message = R"({"Updates":1783028728,"ask":61521.35,"bid":61521.34,"symbol":"BTC\/USD"})"; - SingleTick tick; - EXPECT_FALSE(parse_fxconnect_tick(message, tick)); + events::TickUpdateBatch batch; + EXPECT_FALSE(parse_fxconnect_tick(message, batch)); } TEST(IntradeBarApiResponses, RejectsFxConnectUnsupportedTickMessage) { const std::string message = R"({"Updates":1783028728,"ask":2300.10,"bid":2299.90,"symbol":"XAU\/USD"})"; - SingleTick tick; - EXPECT_FALSE(parse_fxconnect_tick(message, tick)); + events::TickUpdateBatch batch; + EXPECT_FALSE(parse_fxconnect_tick(message, batch)); } TEST(IntradeBarApiResponses, ParsesBtcusdtWebSocketTickWithEpochMilliseconds) { const std::string message = R"({"stream":"btcusdt@aggTrade","data":{"e":"aggTrade","E":1783028778697,"s":"BTCUSDT","a":4005288360,"p":"61521.34000000","q":"0.00017000","f":6473852503,"l":6473852503,"T":1783028778697,"m":false,"M":true}})"; - SingleTick tick; - ASSERT_TRUE(parse_btcusdt_tick(message, tick)); - - EXPECT_EQ(tick.symbol, "BTCUSDT"); - EXPECT_EQ(tick.provider, to_str(PlatformType::INTRADE_BAR)); - EXPECT_EQ(tick.price_digits, 2u); - EXPECT_EQ(tick.volume_digits, 5u); - EXPECT_DOUBLE_EQ(tick.tick.ask, 0.0); - EXPECT_DOUBLE_EQ(tick.tick.bid, 0.0); - EXPECT_DOUBLE_EQ(tick.tick.last, 61521.34); - EXPECT_DOUBLE_EQ(tick.tick.volume, 0.00017); - EXPECT_EQ(tick.tick.time_ms, 1783028778697ULL); - EXPECT_TRUE(tick.tick.has_flag(TickUpdateFlags::LAST_UPDATED)); - EXPECT_FALSE(tick.tick.has_flag(TickUpdateFlags::ASK_UPDATED)); - EXPECT_FALSE(tick.tick.has_flag(TickUpdateFlags::BID_UPDATED)); - EXPECT_TRUE(tick.tick.has_flag(TickUpdateFlags::VOLUME_UPDATED)); - EXPECT_TRUE(tick.tick.has_flag(MarketDataFlags::INITIALIZED)); - EXPECT_TRUE(tick.tick.has_flag(MarketDataFlags::REALTIME)); + events::TickUpdateBatch batch; + ASSERT_TRUE(parse_btcusdt_tick(message, batch)); + + EXPECT_EQ(batch.symbol, "BTCUSDT"); + EXPECT_EQ(batch.provider, to_str(PlatformType::INTRADE_BAR)); + EXPECT_EQ(batch.price_digits, 2u); + EXPECT_EQ(batch.volume_digits, 5u); + ASSERT_EQ(batch.items.size(), 1u); + const auto& tick = batch.items[0]; + EXPECT_DOUBLE_EQ(tick.ask, 0.0); + EXPECT_DOUBLE_EQ(tick.bid, 0.0); + EXPECT_DOUBLE_EQ(tick.last, 61521.34); + EXPECT_DOUBLE_EQ(tick.volume, 0.00017); + EXPECT_EQ(tick.time_ms, 1783028778697ULL); + EXPECT_TRUE(tick.has_flag(TickUpdateFlags::LAST_UPDATED)); + EXPECT_FALSE(tick.has_flag(TickUpdateFlags::ASK_UPDATED)); + EXPECT_FALSE(tick.has_flag(TickUpdateFlags::BID_UPDATED)); + EXPECT_TRUE(tick.has_flag(TickUpdateFlags::VOLUME_UPDATED)); + EXPECT_TRUE(tick.has_flag(MarketDataFlags::INITIALIZED)); + EXPECT_TRUE(tick.has_flag(MarketDataFlags::REALTIME)); } TEST(IntradeBarApiResponses, BtcWebSocketSubscriptionReportsStatusAndRoutesTick) {