Refactor infrastructure and improve trading pair support

Key changes:

1. Infrastructure improvements:
- Reordered session member in OrderBookInfraContext for better destruction order
- Replaced jthread with boost::asio::post for IO context execution
- Added proper session stopping in StopAsync()
- Enhanced WebSocket session cleanup with promises/futures for proper shutdown

2. Trading pair enhancements:
- Expanded supported trading pairs (added ETHBTC, BNBUSDT, SOLBNB, XRPUSDT, ADAUSDT, ADAXRP etc.)
- Updated converters and mappings for new trading pairs
- Added validation for trading pair in snapshot processing

3. Performance and stability:
- Added latency measurement for Binance depth updates
- Reduced verbose logging in JSON parsers
- Improved error handling for missing trading pairs
- Added proper cancellation support in WebSocket sessions

4. New features:
- Added base TradeSubscriptionBuilder for spot and futures
- Extended Bybit support with linear tickers channel
- Improved memory pool constants for funding rate events

5. Bug fixes:
- Fixed trading pair initialization in OrderBookEventInterface
- Ensured proper trading pair setting in snapshot listeners
- Added missing trading pair checks in sync logic

The changes improve stability, add support for more trading pairs, enhance performance monitoring, and provide better resource cleanup during shutdown.

Author: linoxoidunix

aoe/aoe.h

@@ -43,6 +43,8 @@
 #include "aoe/binance/parser/json/ws/execution_response/i_execution_event_parser.h"
 #include "aoe/binance/parser/json/ws/order_response/i_order_event_parser.h"
 #include "aoe/binance/parser/json/ws/order_response/order_event_parser.h"
+#include "aoe/binance/parser/json/ws/trade_response/i_trade_event_parser.h"
+#include "aoe/binance/parser/json/ws/trade_response/trade_event_parser.h"
 #include "aoe/binance/request/amend_order/i_request.h"
 #include "aoe/binance/request/amend_order/request.h"
 #include "aoe/binance/request/cancel_order/i_request.h"
@@ -86,6 +88,10 @@
 #include "aoe/bybit/credentials_loader/credentials_loader.h"
 #include "aoe/bybit/enum_printer/enum_printer.h"
 #include "aoe/bybit/enums/enums.h"
+#include "aoe/bybit/event/funding_rate_event/funding_rate_event.h"
+#include "aoe/bybit/event/funding_rate_event/i_funding_rate_event.h"
+#include "aoe/bybit/events_acceptor/funding_rate_event/acceptor.h"
+#include "aoe/bybit/events_acceptor/funding_rate_event/i_acceptor.h"
 #include "aoe/bybit/execution_event/i_types.h"
 #include "aoe/bybit/execution_event/types.h"
 #include "aoe/bybit/execution_watcher/execution_watcher.h"
@@ -119,6 +125,8 @@
 #include "aoe/bybit/parser/json/ws/order_book_response/parser.h"
 #include "aoe/bybit/parser/json/ws/order_response/i_order_event_parser.h"
 #include "aoe/bybit/parser/json/ws/order_response/order_event_parser.h"
+#include "aoe/bybit/parser/json/ws/ticker_response/funding_rate/i_parser.h"
+#include "aoe/bybit/parser/json/ws/ticker_response/funding_rate/parser.h"
 #include "aoe/bybit/ping_manager/for_private_channel/ping_manager.h"
 #include "aoe/bybit/request/amend_order/i_request.h"
 #include "aoe/bybit/request/amend_order/request.h"
@@ -138,10 +146,13 @@
 #include "aoe/bybit/response_queue_listener/json/ws/order_book_response/listener.h"
 #include "aoe/bybit/response_queue_listener/json/ws/order_response/listener_default.h"
 #include "aoe/bybit/response_queue_listener/json/ws/order_response/listener.h"
+#include "aoe/bybit/response_queue_listener/json/ws/tickers_response/funding_rate/c_listener.h"
+#include "aoe/bybit/response_queue_listener/json/ws/tickers_response/funding_rate/listener.h"
 #include "aoe/bybit/session_setup/web_socket/private/session_setup.h"
 #include "aoe/bybit/session/web_socket/i_web_socket.h"
 #include "aoe/bybit/session/web_socket/web_socket.h"
 #include "aoe/bybit/subscription_builder/subscription_builder.h"
+#include "aoe/bybit/subscription_builder/tickers/subscription_builder.h"
 #include "aoe/credentials/api_key/api_key.h"
 #include "aoe/credentials/api_key/i_api_key.h"
 #include "aoe/credentials/i_credentials.h"

aoe/binance/infrastructure/infrastructure.h

@@ -46,13 +46,13 @@ struct OrderBookInfraContext {
         best_ask_price_notifier;
 
     std::unique_ptr<ListenerT> listener;
-    std::unique_ptr<SessionRWType> session;
     std::unique_ptr<aoe::SubscriptionBuilderInterface>
         diff_subscription_builder;
+    std::unique_ptr<SessionRWType> session;  // it is last
 
-    std::jthread thread;
-    boost::asio::executor_work_guard<boost::asio::io_context::executor_type>
-        work_guard_{boost::asio::make_work_guard(ioc)};
+    // std::jthread thread;
+    // boost::asio::executor_work_guard<boost::asio::io_context::executor_type>
+    //     work_guard_{boost::asio::make_work_guard(ioc)};
 };
 
 template <typename Price, typename Qty,
@@ -116,8 +116,10 @@ class InfrastructureImpl : public InfrastructureInterface,
         context.listener =
             std::make_unique<ListenerT>(pool_, queue, *context.sync);
 
-        context.thread =
-            std::jthread([ioc_ptr = &context.ioc]() { ioc_ptr->run(); });
+        // context.thread =
+        //     std::jthread([ioc_ptr = &context.ioc]() { ioc_ptr->run(); });
+        // boost::asio::post(pool_,
+        //                   [ioc_ptr = &context.ioc]() { ioc_ptr->run(); });
 
         context.session = std::make_unique<SessionRWType>(context.ioc, queue,
                                                           *context.listener);
@@ -193,11 +195,22 @@ class InfrastructureImpl : public InfrastructureInterface,
         for (auto& [pair, context] : contexts_) {
             if (context) {
                 context->session->StartAsync();
+                // context->thread = std::jthread(
+                //     [ioc_ptr = &context->ioc]() { ioc_ptr->run(); });
+                boost::asio::post(
+                    pool_, [ioc_ptr = &context->ioc]() { ioc_ptr->run(); });
             }
         }
     }
 
-    void StopAsync() override { logi("[BINANCE INFRASTRUCTURE] async stop"); }
+    void StopAsync() override {
+        logi("[BINANCE INFRASTRUCTURE] async stop");
+        for (auto& [pair, context] : contexts_) {
+            if (context) {
+                context->session->StopAsync();
+            }
+        }
+    }
 
   private:
     boost::asio::thread_pool& pool_;

aoe/binance/order_book_event/i_order_book_event.h

@@ -45,7 +45,7 @@ class OrderBookEventInterface
 
   protected:
     common::ExchangeId exchange_id_ = common::ExchangeId::kBinance;
-    aos::TradingPair trading_pair_;
+    aos::TradingPair trading_pair_  = aos::TradingPair::kCount;
     std::vector<aos::OrderBookLevelRaw<Price, Qty>> bids_;
     std::vector<aos::OrderBookLevelRaw<Price, Qty>> asks_;
 };

aoe/binance/order_book_sync/order_book_sync.h

@@ -92,6 +92,9 @@ class OrderBookSync
     }
     boost::asio::awaitable<void> RequestNewSnapshot(aos::TradingPair pair) {
         boost::asio::io_context ioc;
+        // response from binance not provide trading pair.
+        // i need set it in manual mode
+        listener_.SetTradingPair(pair);
 
         aoe::binance::impl::main_net::spot::RestSessionRW session(
             ioc, response_queue_, listener_);
@@ -129,6 +132,16 @@ class OrderBookSync
             SetLastDiffUpdateId(diff.FinalUpdateId());
             diffs_.pop();
         }
+        // verified snapshot
+        if (ptr->TradingPair() == aos::TradingPair::kCount) {
+            loge(
+                "[PARSER SNAPSHOT] trading pair is not init. current "
+                "trading_pair={}",
+                ptr->TradingPair());
+            need_process_current_snapshot_ = false;
+            need_make_snapshot_            = true;
+            co_return;
+        }
         need_process_current_snapshot_ = true;
         co_return;
     }
@@ -279,7 +292,9 @@ class OrderBookSync
     }
     boost::asio::awaitable<void> RequestNewSnapshot(aos::TradingPair pair) {
         boost::asio::io_context ioc;
-
+        // response from binance not provide trading pair.
+        // i need set it in manual mode
+        listener_.SetTradingPair(pair);
         aoe::binance::impl::main_net::futures::RestSessionRW session(
             ioc, response_queue_, listener_);
         aoe::binance::snapshot::impl::SnapshotRequestSender<MemoryPool> sender(
@@ -300,7 +315,8 @@ class OrderBookSync
         last_snapshot_update_id_    = ptr->UpdateId();
         last_snapshot_              = ptr;
         snapshot_request_in_flight_ = false;
-        logd("Snapshot received, clearing in-flight flag");
+        logd("Snapshot received for {}, clearing in-flight flag",
+             ptr->TradingPair());
         // skip earlier diffs as soon as possible
         while (!diffs_.empty()) {
             auto ptr         = diffs_.front();
@@ -317,6 +333,16 @@ class OrderBookSync
             SetLastDiffUpdateId(diff.FinalUpdateId());
             diffs_.pop();
         }
+        // verified snapshot
+        if (ptr->TradingPair() == aos::TradingPair::kCount) {
+            loge(
+                "[PARSER SNAPSHOT] trading pair is not init. current "
+                "trading_pair={}",
+                ptr->TradingPair());
+            need_process_current_snapshot_ = false;
+            need_make_snapshot_            = true;
+            co_return;
+        }
         need_process_current_snapshot_ = true;
         co_return;
     }

aoe/binance/parser/json/ws/diff_response/parser.h

@@ -43,6 +43,23 @@ class DiffEventParserBase
             event_type != "depthUpdate") {
             return {false, nullptr};
         }
+        // // NEED PARSE E
+        // --- 🔽 Парсим поле "E" (event time)
+        uint64_t event_time = 0;
+        if (doc["E"].get_uint64().get(event_time) != simdjson::SUCCESS) {
+            return {false, nullptr};
+        }
+        // --- 🔽 Получаем текущее время (в мс с эпохи)
+        uint64_t now = std::chrono::duration_cast<std::chrono::milliseconds>(
+                           std::chrono::system_clock::now().time_since_epoch())
+                           .count();
+
+        // --- 🔽 Вычисляем задержку
+        int64_t delay_ms =
+            static_cast<int64_t>(now) - static_cast<int64_t>(event_time);
+
+        // --- 🔽 Логгируем
+        logi("[depthUpdate binance] Delay: {}", delay_ms);
 
         EventPtr ptr = factory_event_();
 

aoe/binance/parser/json/ws/trade_response/i_trade_event_parser.h

@@ -0,0 +1,18 @@
+#pragma once
+#include "aoe/binance/execution_event/i_types.h"
+#include "boost/intrusive_ptr.hpp"
+#include "simdjson.h"  // NOLINT
+
+namespace aoe {
+namespace binance {
+template <template <typename> typename MemoryPool, typename PositionT>
+class ExecutionEventParserInterface {
+  public:
+    virtual ~ExecutionEventParserInterface() = default;
+    using EventPtr =
+        boost::intrusive_ptr<ExecutionEventInterface<MemoryPool, PositionT>>;
+    virtual std::pair<bool, EventPtr> ParseAndCreate(
+        simdjson::ondemand::document& doc) = 0;
+};
+};  // namespace binance
+};  // namespace aoe