GMDL use_given_sec_key set to true to enable replay data to play without copying md and keys table

* Unifies log message structure for improved readability
* Grants us runtime control over log levels via config

src/lib/api/md/DatabentoMarketDataConfig.proto

@@ -15,4 +15,7 @@ message DatabentoMarketDataConfig {
   string dataset = 70;
   uint32 filter_securities_upto_n_months= 80;
   bool load_reference_data = 90;
+  uint32 debug_sec = 100;
+  bool only_securities =  110;
+  string bbo_start_time = 120;
 }

src/lib/io/websocket/websocket_client.cpp

@@ -3,6 +3,7 @@
 #include "lib/core/TimeSpan.h"
 #include "lib/core/TimeStamp.h"
 #include "lib/io/websocket/websocket_client_config.h"
+#include "lib/io/websocket/websocket_client_logger.h"
 #include "lib/io/websocket/websocket_client_state.h"
 #include "lib/io/websocket/websocket_close_status.h"
 #include "lib/io/websocket/websocket_error_code.h"
@@ -33,6 +34,45 @@
 #include <websocketpp/config/core_client.hpp>
 #include <websocketpp/roles/client_endpoint.hpp>
 
+namespace websocketpp::config {
+
+/// Client config with asio transport and TLS enabled
+struct bts_asio_tls_client : public core_client {
+   typedef bts_asio_tls_client type;
+   typedef core_client base;
+
+   typedef base::concurrency_type concurrency_type;
+
+   typedef base::request_type request_type;
+   typedef base::response_type response_type;
+
+   typedef base::message_type message_type;
+   typedef base::con_msg_manager_type con_msg_manager_type;
+   typedef base::endpoint_msg_manager_type endpoint_msg_manager_type;
+
+   typedef websocketpp::log::WebsocketClientLogger<
+       websocketpp::log::ElevelAdapter>
+       elog_type;
+   typedef websocketpp::log::WebsocketClientLogger<
+       websocketpp::log::AlevelAdapter>
+       alog_type;
+
+   typedef base::rng_type rng_type;
+
+   struct transport_config : public base::transport_config {
+      typedef type::concurrency_type concurrency_type;
+      typedef type::alog_type alog_type;
+      typedef type::elog_type elog_type;
+      typedef type::request_type request_type;
+      typedef type::response_type response_type;
+      typedef transport::asio::tls_socket::endpoint socket_type;
+   };
+
+   typedef transport::asio::endpoint<transport_config> transport_type;
+};
+
+} // namespace websocketpp::config
+
 namespace bts::websocket {
 
 struct WebsocketClientSnapshot {
@@ -70,10 +110,9 @@ using std::thread;
 using std::uint32_t;
 using std::chrono::nanoseconds;
 using std::this_thread::sleep_for;
-using websocketpp::config::asio_tls_client;
 
 using WebsocketppClient =
-    websocketpp::client<websocketpp::config::asio_tls_client>;
+    websocketpp::client<websocketpp::config::bts_asio_tls_client>;
 using ConnectionHdl = websocketpp::connection_hdl;
 using ConnectionPtr = WebsocketppClient::connection_ptr;
 using MessageTypePtr = websocketpp::config::core_client::message_type::ptr;
@@ -117,7 +156,10 @@ struct WebsocketClient::PImpl {
    int version = 0;
    int n_reconnects = 0;
 
-   PImpl() {
+   PImpl(WebsocketClientConfig config, boost::asio::io_context *io_context)
+       : config(std::move(config)), io_context(io_context) {
+      client.get_alog().set_name(this->config.name);
+      client.get_elog().set_name(this->config.name);
       client.set_access_channels(websocketpp::log::alevel::all);
       // client.clear_access_channels(websocketpp::log::alevel::frame_payload);
       client.set_error_channels(websocketpp::log::elevel::all);
@@ -404,10 +446,7 @@ struct WebsocketClient::PImpl {
 
 WebsocketClient::WebsocketClient(
     WebsocketClientConfig config, boost::asio::io_context *io_context)
-    : me_(std::make_unique<PImpl>()) {
-   me_->config = std::move(config);
-   me_->io_context = io_context;
-}
+    : me_(std::make_unique<PImpl>(std::move(config), io_context)) {}
 
 WebsocketClient::~WebsocketClient() {
    me_->close(websocketpp::close::status::going_away, "");

src/lib/io/websocket/websocket_client_config.h

@@ -11,6 +11,8 @@ struct WebsocketClientConfig {
    bool reconnect_on_close = false;
    bool reconnect_on_fail = false;
    std::map<std::string, std::string> headers{};
+   // name to use in logs
+   std::string name{};
 };
 
 } // namespace bts::websocket

src/lib/io/websocket/websocket_client_logger.h

@@ -0,0 +1,88 @@
+#pragma once
+
+#include "lib/logging/Common.h"
+#include "lib/logging/logging.h"
+
+#include <websocketpp/concurrency/basic.hpp>
+#include <websocketpp/logger/basic.hpp>
+#include <websocketpp/logger/levels.hpp>
+
+namespace websocketpp::log {
+
+struct ElevelAdapter {
+   typedef elevel LevelT;
+
+   constexpr bts::log::Level operator()(level channel) {
+      switch (channel) {
+      case elevel::devel:
+         return bts::Diag;
+      case elevel::library:
+      case elevel::warn:
+         return bts::Warn;
+      case elevel::rerror:
+         return bts::Error;
+      case elevel::fatal:
+         return bts::Fatal;
+      default:
+         return bts::Info;
+      }
+   }
+};
+
+struct AlevelAdapter {
+   typedef alevel LevelT;
+
+   constexpr bts::log::Level operator()(level channel) {
+      switch (channel) {
+      case alevel::control:
+      case alevel::frame_header:
+      case alevel::frame_payload:
+      case alevel::message_header:
+      case alevel::message_payload:
+      case alevel::debug_handshake:
+      case alevel::debug_close:
+      case alevel::devel:
+         return bts::Diag;
+      case alevel::disconnect:
+         return bts::Warn;
+      case alevel::fail:
+         return bts::Error;
+      default:
+         return bts::Info;
+      }
+   }
+};
+
+template <typename LevelAdapter>
+class WebsocketClientLogger
+    : public basic<concurrency::basic, typename LevelAdapter::LevelT> {
+   typedef basic<concurrency::basic, typename LevelAdapter::LevelT> base;
+
+   std::string name;
+   LevelAdapter level_adapter;
+
+ public:
+   WebsocketClientLogger(
+       level c, channel_type_hint::value h = channel_type_hint::access)
+       : base(c, h) {}
+
+   void set_name(std::string name) {
+      concurrency::basic::scoped_lock_type lock{this->m_lock};
+      this->name = std::move(name);
+   }
+
+   void write(level channel, const std::string &msg) {
+      concurrency::basic::scoped_lock_type lock{this->m_lock};
+      if (!this->dynamic_test(channel)) {
+         return;
+      }
+      LOG(Http, level_adapter(channel), "WebsocketClient-{} {}. {}", name,
+          std::string{LevelAdapter::LevelT::channel_name(channel)}, msg);
+   }
+
+   void write(level channel, const char *msg) {
+      write(channel, std::string{msg});
+   }
+};
+
+} // namespace websocketpp::log

src/lib/logging/Sink.cpp

@@ -9,6 +9,7 @@
 #include <boost/date_time/posix_time/posix_time.hpp>
 #include <boost/filesystem.hpp>
 #include <boost/filesystem/path.hpp>
+#include <cstdint>
 #include <fmt/ostream.h>
 #include <fstream>
 #include <iostream>
@@ -94,7 +95,7 @@ std::ifstream::pos_type filesize(std::string const &filename) {
 Sink file_sink(LoggingConfig const &cfg) {
    static std::ofstream stream;
    static uint64_t bytes_written = 0;
-   const auto max_sz = 120 * 1024 * 1024;
+   const uintmax_t max_sz =  1024 * 1024 * 1024 ;
 
    auto f = formatter(cfg);
    auto path = log_path();

src/lib/md/GeneralMarketDataListener.cpp

@@ -290,7 +290,7 @@ struct GeneralMarketDataListener::PImpl {
    // if this is set to true, then the security key passed on in the
    // on_securities function is used. If this is not set to true, then a new key
    // is generated
-   const bool use_given_sec_key = false;
+   const bool use_given_sec_key = true;
    const std::string cc_name_prefix = "cc_security";
    size_t n_bk_updates = 0;
    bts::persistence::IDGenerator &id_gen =

src/lib/orderman/gateways/OkxExecutionGateway.cpp

@@ -6,6 +6,7 @@
 #include "lib/io/http/http_client.h"
 #include "lib/io/http/http_response.h"
 #include "lib/io/websocket/websocket_client.h"
+#include "lib/io/websocket/websocket_client_config.h"
 #include "lib/logging/logging.h"
 #include "lib/md/SecurityOracle.h"
 #include "lib/orderman/CancelOrderMsg.h"
@@ -77,6 +78,16 @@ void OkxExecutionGateway::send_msgs(
           msg);
    }
 }
+
+static websocket::WebsocketClientConfig get_ws_client_config(
+    const std::string &url,
+    core::TimeSpan connect_timeout,
+    core::TimeSpan pong_timeout) {
+   websocket::WebsocketClientConfig rv{url, connect_timeout, pong_timeout};
+   rv.name = "OkxExecutionGateway";
+   return rv;
+}
+
 struct OkxExecutionGateway::PImpl {
    const std::string api_key = config::get<config::OkxExeCfg>().api_key();
    const std::string secret_key = config::get<config::OkxExeCfg>().secret_key();
@@ -99,8 +110,8 @@ struct OkxExecutionGateway::PImpl {
    md::SecurityOracle o;
    std::atomic<OkxExecutionState> state = OkxExecutionState::OFF;
    PImpl(md::SecurityOracle o)
-       : ws_client(
-             websocket::WebsocketClient({url, connect_timeout, pong_timeout})),
+       : ws_client(websocket::WebsocketClient(
+             get_ws_client_config(url, connect_timeout, pong_timeout))),
          cb(dummy), o(o) {
       ws_client.message_handler().bind<&PImpl::on_message>(*this);
       ws_client.close_handler().bind<&PImpl::on_close>(*this);
@@ -284,25 +295,38 @@ struct OkxExecutionGateway::PImpl {
 
    void start_impl() {
       size_t attempt = 0;
-      for (size_t version = 0; state != OkxExecutionState::OFF; version++) {
+      while (state != OkxExecutionState::OFF) {
          try {
             LOG(OrderMan, Warn,
                 "OkxExecutionAdapter::start Attempting login (Attempt: {})",
                 attempt + 1);
             if (ws_run_thread) {
+               // Join this thread to block until the websocket disconnects (then reconnect)
                if (ws_run_thread->joinable()) {
                   ws_run_thread->join();
                }
                ws_run_thread.reset();
             }
+            // Connect
             ws_client.connect();
             ws_run_thread = threading::make_thread(
-                "okx_execution_ws", [&, orig_version = version]() {
-                   while (version == orig_version) {
-                      ws_client.run_one();
-                   }
-                });
-            sleep(2);
+                "okx_execution_ws", [&]() { ws_client.run(); });
+            // Block until the websocket connects (or times out)
+            {
+               const auto start_time = bts::core::TimeStamp::now();
+               for (auto elapsed = core::TimeSpan::zero();
+                    !ws_client.is_connected() &&
+                    elapsed < core::TimeSpan::from_seconds(20);
+                    elapsed = core::TimeStamp::now() - start_time) {
+                  std::this_thread::sleep_for(std::chrono::milliseconds(500));
+               }
+               if (!ws_client.is_connected()) {
+                  LOG(OrderMan, Warn,
+                      "OkxExecutionAdapter::start. Timed out waiting for "
+                      "websocket to connect");
+                  continue;
+               }
+            }
             if (ws_ping_thread.has_value()) {
                if (ws_ping_thread->joinable()) {
                   ws_ping_thread
@@ -312,7 +336,7 @@ struct OkxExecutionGateway::PImpl {
                    .reset(); // Reset the optional before creating a new thread
             }
 
-            ws_ping_thread.emplace([&, orig_version = version]() {
+            ws_ping_thread.emplace([&]() {
                LOG(OrderMan, Warn,
                    "OkxExecutionAdapter::ws_ping_thread thread starting. url: "
                    "{}",
@@ -323,7 +347,7 @@ struct OkxExecutionGateway::PImpl {
                std::this_thread::sleep_for(
                    std::chrono::milliseconds(ping_interval.millis()));
 
-               while (ws_client.is_connected() && version == orig_version) {
+               while (ws_client.is_connected()) {
                   const auto now = bts::core::TimeStamp::now();
                   if ((now - last_ping_time) > ping_interval) {
                      LOG(OrderMan, Warn,

src/lib/orderman/orders/spreader_orders/SpreaderOrder.cpp

@@ -68,6 +68,8 @@ OrderOpResult SpreaderOrder::on_activate(
       return OrderOpResult{api::orderman::ORDER_ERROR_OP_NOT_SUPPORTED};
    quantity =
        _bookkeeping_data().fqty_helper().in_transform_spreader_qty(quantity);
+   LOG(OrderMan, Info, "SpreaderOrder::on_activate price: {} quantity: {}",
+       price, quantity);
    SpreaderOrderServices spreader_order_services(&services);
    _on_activate(price, quantity, spreader_order_services);
    // the reason why this is immediatly reported (vs in the timer
@@ -82,8 +84,9 @@ OrderOpResult SpreaderOrder::on_update(Price price, Quantity quantity) {
       return OrderOpResult{api::orderman::ORDER_ERROR_OP_NOT_SUPPORTED};
    quantity =
        _bookkeeping_data().fqty_helper().in_transform_spreader_qty(quantity);
-   LOG(OrderMan, Info, "SpreaderOrder.on_update key:{} price:{} quantity:{}",
-       key().value(), price.value(), quantity.value());
+
+   LOG(OrderMan, Info, "SpreaderOrder::on_update price: {} quantity: {}", price,
+       quantity);
    _state_worker->on_update(price, quantity);
    _report_state();
    return OrderOpResult{};
@@ -329,8 +332,11 @@ void SpreaderOrder::_on_activate(
     Price price, Quantity quantity, SpreaderOrderServices &order_services) {
    bts_assert(_inited());
    bts_assert(!_activated());
-   quantity =
-       _bookkeeping_data().fqty_helper().in_transform_spreader_qty(quantity);
+   LOG(OrderMan, Info, "SpreaderOrder::_on_activate price: {} quantity: {}",
+       price, quantity);
+   // quantity has already been fractional_normalized
+   // quantity =
+   //     _bookkeeping_data().fqty_helper().in_transform_spreader_qty(quantity);
    LOG(OrderMan, Info, "SpreaderOrder.on_activate key:{} price:{} quantity:{}",
        key().value(), price.value(), quantity.value());
    _order_services = new SpreaderOrderServices(order_services);

src/lib/orderman/orders/spreader_orders/SpreaderOrder.h

@@ -285,6 +285,9 @@ class SpreaderOrder : public Order, public ISpreaderManager {
    }
    // just sets the price and qty in the bookkeeping data
    void _set_price_quantity(Price input_price, Quantity quantity) {
+      LOG(OrderMan, Info,
+          "SpreaderOrder::_set_price_quantity input_price: {} quantity: {}",
+          input_price, quantity);
       Price effective_price = _compute_effective_price(input_price);
       _bookkeeping_data().update_order(input_price, effective_price, quantity);
    }

src/lib/orderman/orders/spreader_orders/spreader/SpreaderBookkeepingData.h

@@ -1,4 +1,5 @@
 #pragma once
+#include "lib/logging/logging.h"
 #include "lib/orderman/orderman_utils/orderman_common.h"
 #include <boost/container/flat_map.hpp>
 #include <memory>
@@ -76,7 +77,11 @@ class SpreaderBookkeepingData {
    }
    Price input_price() const { return _input_price; }
    Price effective_price() const { return _effective_price; }
-   Quantity quantity() const { return _quantity; }
+   Quantity quantity() const {
+      LOG(OrderMan, Info, "SpreaderBookkeepingData::quantity _quantity: {}",
+          _quantity);
+      return _quantity;
+   }
    NLegsHedgeRecompData &hedge_recomp_data() { return _hedge_recomp_data; }
    // gets leg data from leg's sec key
    SpreaderLegData &get_leg_data(SecurityKey security_key) {
@@ -146,6 +151,10 @@ class SpreaderBookkeepingData {
    }
    void update_order(
        Price input_price, Price effective_price, Quantity quantity) {
+      LOG(OrderMan, Info,
+          "SpreaderBookkeepingData::update_order input_price: {} "
+          "effective_price: {} quantity: {}",
+          input_price, effective_price, quantity.full_value());
       _input_price = input_price;
       _effective_price = effective_price;
       _quantity = quantity;
@@ -157,9 +166,6 @@ class SpreaderBookkeepingData {
    }
    size_t n_quoting_legs() const { return _n_quoting_legs; }
    SpreaderFractionalQtyHelper &fqty_helper() { return _fqty_helper; }
-   // SpreaderFractionalQtyHelper const *spreader_qty_helper() {
-   //    return &_fqty_helper;
-   // }
 };
 } // namespace orderman
 } // namespace bts

src/lib/orderman/orders/spreader_orders/spreader/SpreaderFractionalQtyHelper.cpp

@@ -9,6 +9,7 @@
 #include "lib/orderman/events/BookUpdateEvent.h"
 #include "lib/orderman/events/OrderStateUpdateEvent.h"
 //#include "lib/orderman/orders/spreader_orders/SpreaderOrder.h"
+#include "lib/orderman/orderman_utils/orderman_common.h"
 #include "lib/orderman/orders/spreader_orders/spreader/LegPropertyData.h"
 #include "lib/orderman/orders/spreader_orders/spreader/SpreaderUtils.h"
 #include "lib/os/os.h"
@@ -63,7 +64,9 @@ struct SpreaderFractionalQtyHelper ::PImpl {
    Book op_book = {};
    OrderStateUpdateEvent op_ord = {};
    const bool passthrough = false;
-   //const double in_qty_mult;
+   // we can keep the mult per leg, based on their tick info, but its easier
+   // (and correct) if we do it
+   const double in_qty_mult = 100000;
    const bool is_notional;
    const QtyTickInfo def_qty_tick_info = QtyTickInfo(10000.0, 1.0);
    std::map<bts::SecurityKey, QtyTickInfo> qty_info_map = {};
@@ -92,25 +95,43 @@ struct SpreaderFractionalQtyHelper ::PImpl {
    }
 };
 
+double SpreaderFractionalQtyHelper::_get_mult(size_t) const {
+   return me->in_qty_mult;
+}
+
+Quantity SpreaderFractionalQtyHelper::in_transform_spreader_qty(
+    Quantity qty) const {
+   // nothing needs to be done with spreader qty
+   if (me->passthrough)
+      return qty;
+   double qm = _get_mult(0);
+   double rv = qty.full_value() * qm;
+   auto rv_qty = Quantity::create_new(rv);
+   LOG(OrderMan, Info,
+       "SpreaderFractionalQtyHelper::in_transform_spreader_qty {} => {}", qty,
+       rv_qty);
+   return rv_qty;
+}
+
 std::vector<LegPropertyData> SpreaderFractionalQtyHelper::get_transformed_legs(
     const std::vector<LegPropertyData> &legs) const {
-   std::vector<LegPropertyData> rv = {};
-   for (const auto &l : legs) {
-      bts::api::ContractLeg leg_adj = l.leg_contract();
-      const auto &q_tinfo = get_qty_tick_info(
-          me->qty_info_map, bts::SecurityKey(l.leg_security_key()),
-          me->def_qty_tick_info);
-      double qm = q_tinfo.tick_mult;
-      leg_adj.set_quantity(l.size_multiplier() * qm);
-      double qty_tick_size = q_tinfo.tick_sz;
-      LegPropertyData lpd_adj(
-          l.leg_index(), leg_adj, l.leg_params(), l.trading_side(),
-          bts::api::SyntheticContract::PriceCalc::
-              SyntheticContract_PriceCalc_PRICECALC_STANDARD,
-          qty_tick_size);
-      rv.push_back(lpd_adj);
-   }
-   return rv;
+   // std::vector<LegPropertyData> rv = {};
+   // for (const auto &l : legs) {
+   //    bts::api::ContractLeg leg_adj = l.leg_contract();
+   //    const auto &q_tinfo = get_qty_tick_info(
+   //        me->qty_info_map, bts::SecurityKey(l.leg_security_key()),
+   //    double qm = q_tinfo.tick_mult;
+   //    leg_adj.set_quantity(l.size_multiplier() * qm);
+   //    double qty_tick_size = q_tinfo.tick_sz;
+   //    LegPropertyData lpd_adj(
+   //        l.leg_index(), leg_adj, l.leg_params(), l.trading_side(),
+   //        bts::api::SyntheticContract::PriceCalc::
+   //            SyntheticContract_PriceCalc_PRICECALC_STANDARD,
+   //        qty_tick_size);
+   //    rv.push_back(lpd_adj);
+   // }
+   // return rv;
+   return legs;
 }
 
 SpreaderFractionalQtyHelper::SpreaderFractionalQtyHelper(
@@ -140,9 +161,7 @@ const Book &SpreaderFractionalQtyHelper::in_transform_book(
    } else {
       me->op_book = bk;
    }
-   const auto &q_tinfo = get_qty_tick_info(
-       me->qty_info_map, bk.security_key(), me->def_qty_tick_info);
-   double qm = q_tinfo.tick_mult;
+   double qm = _get_mult(0);
    for (auto &ps : me->op_book.bid()) {
       ps = bts::md::PriceSize(ps.int_price(), ps.size() * qm);
    }
@@ -183,9 +202,7 @@ SpreaderFractionalQtyHelper::in_transform_order_event(
       me->op_ord.filled_at_attach_time_quantity = os::nan();
    }
    // factor mult
-   const auto &q_tinfo =
-       get_qty_tick_info(me->qty_info_map, sec_key, me->def_qty_tick_info);
-   double qmult = q_tinfo.tick_mult;
+   double qmult = _get_mult(sec_key.value());
    me->op_ord.quantity *= qmult;
    me->op_ord.filled_quantity *= qmult;
    me->op_ord.filled_at_attach_time_quantity *= qmult;
@@ -193,14 +210,6 @@ SpreaderFractionalQtyHelper::in_transform_order_event(
    return me->op_ord;
 }
 
-Quantity SpreaderFractionalQtyHelper::in_transform_spreader_qty(
-    Quantity qty) const {
-   // nothing needs to be done with spreader qty
-   if (me->passthrough)
-      return qty;
-   return qty;
-}
-
 Quantity SpreaderFractionalQtyHelper::out_transform_leg_order_qty(
     SecurityKey skey, Price price, Quantity qty) const {
    auto qty_init = qty;
@@ -214,13 +223,10 @@ Quantity SpreaderFractionalQtyHelper::out_transform_leg_order_qty(
              "notional -> physical {} -> {}",
              qty_init, qty);
       }
-      const auto &q_tinfo =
-          get_qty_tick_info(me->qty_info_map, skey, me->def_qty_tick_info);
-      double tick_sz = q_tinfo.tick_sz;
-      double qm = q_tinfo.tick_mult;
+      double qm = _get_mult(skey.value());
       double qty_out = qty.full_value() / qm;
       // rouning it to tick size
-      qty_out = std::floor(qty_out / tick_sz) * tick_sz;
+      //qty_out = std::floor(qty_out / tick_sz) * tick_sz;
       return Quantity::create_new(qty_out);
    };
    auto rv_qty = tr();
@@ -241,9 +247,15 @@ void SpreaderFractionalQtyHelper::out_transform_order_state(
        s_txt, s.filled_quantity(), s.avg_price());
    // todo:
    // this needs to divide the filled qty of the spreader
-   // if (me->passthrough)
-   //    return ord_st;
-   // return ord_st;
+   if (me->passthrough)
+      return;
+   double qm = _get_mult(0);
+   set_quantity(
+       s, StateQuantityType::OrderQuantity,
+       Quantity::create_new(static_cast<double>(s.quantity()) / qm));
+   set_quantity(
+       s, StateQuantityType::FilledQuantity,
+       static_cast<double>(s.filled_quantity()) / qm);
 }
 
 } // namespace bts::orderman

src/lib/orderman/orders/spreader_orders/spreader/SpreaderFractionalQtyHelper.h

@@ -15,31 +15,52 @@ class OrderState;
 namespace bts::orderman {
 class Book;
 class OrderStateUpdateEvent;
-// forinstruments with fractional qtys this sits on the outside layer and does conversions,
-//both for market data and orders
-// this resdies inside SpeaderBookkeepingData
+// for instruments with fractional qtys this sits on the outside layer and does conversions,
+// both for market data and orders
+// it resides inside SpeaderBookkeepingData
+// these are the trasnfo
+// spreader_qty in *mult
+// books coming in *mult
+// order out /mult
+// order cb in * mult
+// report out /mult
 class SpreaderFractionalQtyHelper {
  public:
    SpreaderFractionalQtyHelper(
        OrderKey spreader_ord_key,
        const api::orderman::SpreaderOrderDefinition &spreader_order_def,
        const std::vector<LegPropertyData> &legs);
-   // the leg qtys need to be translated
+   double _get_mult(size_t leg_idx) const;
+   ////////////////////////////////////
+   // 1. qty (in)
+   ////////////////////////////////////
+   Quantity in_transform_spreader_qty(Quantity qty) const;
+   ////////////////////////////////////
+   // 2. leg sizes (in)
+   ////////////////////////////////////
    std::vector<LegPropertyData> get_transformed_legs(
        const std::vector<LegPropertyData> &legs) const;
-   // in_ and out_ signify market data / orders coming in or out.
+   ////////////////////////////////////
+   // 3. books (in)
+   ////////////////////////////////////
    const Book &in_transform_book(const Book &book) const;
    // this is called when the order feedback fomr the outside in to the spreader
+   ////////////////////////////////////
+   // 4. orders event (in)
+   ////////////////////////////////////
    const OrderStateUpdateEvent &in_transform_order_event(
        SecurityKey sec_key, const OrderStateUpdateEvent &) const;
-   Quantity in_transform_spreader_qty(Quantity qty) const;
    // TODO add price
+   ////////////////////////////////////
+   // 5. orders (out)
+   ////////////////////////////////////
    Quantity out_transform_leg_order_qty(
        SecurityKey key, Price price, Quantity qty) const;
    // the state reported ot he rest of the system also has to be converted
+   ////////////////////////////////////
+   // 6. order report (out)
+   ////////////////////////////////////
    void out_transform_order_state(bts::api::orderman::OrderState &) const;
-   // struct QtyTickInfo;
-   // const QtyTickInfo &_get_tickinfo(SecurityKey skey);
 
  private:
    struct PImpl;

src/lib/orderman/orders/spreader_orders/spreader_states/SpreaderStartStateWorker.cpp

@@ -1,4 +1,5 @@
 #include "SpreaderStartStateWorker.h"
+#include "lib/logging/logging.h"
 #include "lib/orderman/OrderBuilder.h"
 #include "lib/orderman/orders/PreActivationOrderServices.h"
 #include "lib/orderman/orders/spreader_orders/SpreaderOrder.h"
@@ -47,6 +48,8 @@ void SpreaderStartStateWorker::on_init_new(
       std::string msg = "Could not resolve securities";
       _spreader->_transform_to_END(&msg);
    }
+   LOG(OrderMan, Info, "SpreaderStartStateWorker::on_init_new qty: {}",
+       _spreader->_bookkeeping_data().quantity().full_value());
 }
 void SpreaderStartStateWorker::on_init_recovered(
     const bts::api::orderman::OrderRecord &order_record,

src/lib/persistence/IDGenerator.cpp

@@ -73,7 +73,7 @@ std::optional<uint64_t> IDGenerator::get_id(
       return std::optional<uint64_t>(rowid);
    } else {
       if (add_if_new) {
-         auto all = me->kv_store->retrieve_all();
+         // auto all = me->kv_store->retrieve_all();
          auto new_id = me->kv_store->insert(ns, key, "");
          LOG(Gen, Diag, "IDGenerator::get_id ns: {} key: id: {}", ns, key,
              new_id);

src/lib_binance_execution/binance_execution/cross_margin/cross_margin_gateway.cpp

@@ -111,8 +111,6 @@ struct CrossMarginGateway::PImpl {
       pos.exchange = VENUE;
    }
 
-   static constexpr const char *CANCEL_REPLACE_CANCEL_SUFFIX = "crc";
-
    bts::md::SecurityOracle sec_oracle;
    GatewayConfig config;
    shared_ptr<bts::config::JetFIX_Connection> jetfix_connection;
@@ -152,16 +150,6 @@ struct CrossMarginGateway::PImpl {
 
    void on_order_update(trade_data_stream::OrderUpdate u) {
       INFO_FN("on_order_update", "\n{:2}", json_formatter(u));
-      if (u.orig_client_order_id.ends_with(CANCEL_REPLACE_CANCEL_SUFFIX)) {
-         // This is the "canceled" order part of a cancel/replace op. This op
-         // generates another update without such a suffix. That's the new
-         // order; that's the one we want.
-         DIAG_FN(
-             "on_order_update",
-             "Ignoring cancel/replace cancel order update. id: {}",
-             u.client_order_id);
-         return;
-      }
       incoming_order_updates.push(std::move(u));
    }
 
@@ -268,7 +256,7 @@ struct CrossMarginGateway::PImpl {
       trade_data_stream::OrderUpdate curr;
       auto &km = *key_map;
       while (incoming_order_updates.pop(curr)) {
-         auto bts_key = km.get_bts_key(curr.client_order_id);
+         const auto bts_key = km.get_bts_key(curr.client_order_id);
          if (!bts_key) {
             ERROR_FN(
                 "dispatch_order_updates",
@@ -277,7 +265,7 @@ struct CrossMarginGateway::PImpl {
                 curr.client_order_id);
             continue;
          }
-         auto exch_key = km.get_exch_key(*bts_key);
+         const auto exch_key = km.get_exch_key(*bts_key);
          if (exch_key != curr.client_order_id) {
             INFO_FN(
                 "dispatch_order_updates",

src/lib_binance_execution/binance_execution/cross_margin/cross_margin_gateway_config.h

@@ -1,26 +0,0 @@
-#pragma once
-
-#include <nlohmann/json.hpp>
-#include <string>
-
-namespace binance_execution::cross_margin {
-
-struct CrossMarginGatewayConfig {
-   std::string api_key;
-   std::string secret_key;
-   std::string account;
-   std::string rest_endpoint;
-   std::string ws_endpoint;
-   std::string ws_stream_endpoint;
-};
-
-NLOHMANN_DEFINE_TYPE_NON_INTRUSIVE(
-    CrossMarginGatewayConfig,
-    api_key,
-    secret_key,
-    account,
-    rest_endpoint,
-    ws_endpoint,
-    ws_stream_endpoint);
-
-} // namespace binance_execution::cross_margin

src/lib_binance_execution/binance_execution/cross_margin/messages/create_oco_request.h

@@ -1,49 +0,0 @@
-// #pragma once
-
-// #include "binance_execution/enums/new_order_response_type.h"
-// #include "binance_execution/enums/order_type.h"
-// #include "binance_execution/enums/side.h"
-// #include "binance_execution/enums/time_in_force.h"
-
-// #include <nlohmann/json.hpp>
-// #include <string>
-
-// namespace bts::orderman {
-// struct CreateOrderMsg;
-// }
-
-// namespace binance_execution::cross_margin {
-
-// /**
-//  * @a https://developers.binance.com/docs/binance-spot-api-docs/rest-api/trading-endpoints#new-order-trade
-//  */
-// struct CreateOcoRequest {
-//    std::string symbol;
-//    std::string listClientOrderId;
-//    Side side = Side::NONE;
-//    double quantity = 0;
-//    std::string limitClientOrderId;
-//    double price = std::numeric_limits<double>::quiet_NaN();
-//    double limitIcebergQty = 0;
-//    std::string stopClientOrderId;
-//    double stopPrice;
-//    double stopLimitPrice;
-//    double stopIcebergQty;
-//    TimeInForce stopLimitTimeInForce;
-//    NewOrderResponseType newOrderRespType;
-//    OrderType type = OrderType::NONE;
-//    std::int64_t timestamp = 0;
-//    std::string newClientOrderId;
-//    TimeInForce timeInForce = TimeInForce::NONE;
-//    std::int64_t recvWindow = 0;
-//    NewOrderResponseType newOrderRespType = NewOrderResponseType::NONE;
-
-//    bool init(const bts::orderman::CreateOrderMsg &m, std::string &error);
-// };
-
-// void to_json(nlohmann::json &j, const CreateOcoRequest &req);
-// void from_json(const nlohmann::json &j, CreateOcoRequest &req);
-
-// void to_query_params(std::string &query_params, const CreateOcoRequest &req);
-
-// } // namespace binance_execution::spots

src/lib_binance_execution/binance_execution/cross_margin/rest_api_client.cpp

@@ -11,7 +11,7 @@ using bts::system::crypto::hmac_digest_hex;
 using std::string;
 
 #define _LOG_FN(level, fn, ...)                                                \
-   LOG(OrderMan, level, "Binance::" fn ". " __VA_ARGS__)
+   LOG(OrderMan, level, "BinanceCrossMargin::" fn ". " __VA_ARGS__)
 #define DIAG_FN(fn, ...) _LOG_FN(bts::Diag, fn, __VA_ARGS__)
 #define INFO_FN(fn, ...) _LOG_FN(bts::Info, fn, __VA_ARGS__)
 #define WARN_FN(fn, ...) _LOG_FN(bts::Warn, fn, __VA_ARGS__)
@@ -52,6 +52,7 @@ void RestApiClient::request_(
     const std::string &path,
     bts::io::http::HttpMethod method,
     const std::string &body,
+    bool sign,
     RestApiResult<ResponseT> &res) {
    // Use thread_local vars to reduce memory allocations. Shrink strings
    // to reduce memory footprint from previous requests.
@@ -69,7 +70,9 @@ void RestApiClient::request_(
    qps.clear();
    to_query_params(qps, req);
    if (!qps.empty()) {
-      sign_query_params_(qps);
+      if (sign) {
+         sign_query_params_(qps);
+      }
       http_req.url.append("?");
       http_req.url.append(qps);
    }
@@ -102,11 +105,12 @@ void RestApiClient::request_(
 #define REQUEST_DEF(ResponseT, RequestT)                                       \
    template void RestApiClient::request_<ResponseT, RequestT>(                 \
        const RequestT &req, const std::string &path,                           \
-       bts::io::http::HttpMethod method, const std::string &body,              \
+       bts::io::http::HttpMethod method, const std::string &body, bool sign,   \
        RestApiResult<ResponseT> &res)
 
 REQUEST_DEF(CreateOrderResponse, CreateOrderRequest);
 REQUEST_DEF(CancelOrderResponse, CancelOrderRequest);
 REQUEST_DEF(ListenKeyResponse, ListenKeyRequest);
+REQUEST_DEF(RefreshListenKeyResponse, RefreshListenKeyRequest);
 
 } // namespace binance_execution::cross_margin

src/lib_binance_execution/binance_execution/cross_margin/rest_api_client.h

@@ -6,12 +6,13 @@
 #include "binance_execution/cross_margin/messages/create_order_response.h"
 #include "binance_execution/messages/listen_key_request.h"
 #include "binance_execution/messages/listen_key_response.h"
+#include "binance_execution/messages/refresh_listen_key_request.h"
+#include "binance_execution/messages/refresh_listen_key_response.h"
 #include "binance_execution/rest_api_result.h"
 #include "lib/io/http/HttpMethod.h"
 
 #include <nlohmann/json.hpp>
 #include <string>
-#include <vector>
 
 namespace binance_execution::cross_margin {
 
@@ -26,8 +27,8 @@ class RestApiClient {
        const CancelOrderRequest &req);
    const RestApiResult<ListenKeyResponse> &get_listen_key(
        const ListenKeyRequest &req = {});
-   const RestApiResult<ListenKeyResponse> &refresh_listen_key(
-       const ListenKeyRequest &req = {});
+   const RestApiResult<RefreshListenKeyResponse> &refresh_listen_key(
+       const RefreshListenKeyRequest &req = {});
 
  private:
    std::string url_;
@@ -40,6 +41,7 @@ class RestApiClient {
        const std::string &path,
        bts::io::http::HttpMethod method,
        const std::string &body,
+       bool sign,
        RestApiResult<ResponseT> &op_res);
 
    bool sign_query_params_(std::string &qp);
@@ -60,7 +62,7 @@ inline const RestApiResult<CreateOrderResponse> &RestApiClient::create_order(
    thread_local RestApiResult<CreateOrderResponse> res;
    res.reset();
    request_<CreateOrderResponse>(
-       req, path, bts::io::http::HttpMethod::POST, "", res);
+       req, path, bts::io::http::HttpMethod::POST, "", true, res);
    return res;
 }
 
@@ -70,7 +72,7 @@ inline const RestApiResult<CancelOrderResponse> &RestApiClient::cancel_order(
    thread_local RestApiResult<CancelOrderResponse> res;
    res.reset();
    request_<CancelOrderResponse>(
-       req, path, bts::io::http::HttpMethod::DELETE, "", res);
+       req, path, bts::io::http::HttpMethod::DELETE, "", true, res);
    return res;
 }
 
@@ -79,16 +81,16 @@ inline const RestApiResult<ListenKeyResponse> &RestApiClient::get_listen_key(
    static const std::string path{"/sapi/v1/userDataStream"};
    thread_local RestApiResult<ListenKeyResponse> res;
    res.reset();
-   request_(req, path, bts::io::http::HttpMethod::POST, "", res);
+   request_(req, path, bts::io::http::HttpMethod::POST, "", false, res);
    return res;
 }
 
-inline const RestApiResult<ListenKeyResponse> &
-RestApiClient::refresh_listen_key(const ListenKeyRequest &req) {
+inline const RestApiResult<RefreshListenKeyResponse> &
+RestApiClient::refresh_listen_key(const RefreshListenKeyRequest &req) {
    static const std::string path{"/sapi/v1/userDataStream"};
-   thread_local RestApiResult<ListenKeyResponse> res;
+   thread_local RestApiResult<RefreshListenKeyResponse> res;
    res.reset();
-   request_(req, path, bts::io::http::HttpMethod::PUT, "", res);
+   request_(req, path, bts::io::http::HttpMethod::PUT, "", false, res);
    return res;
 }
 

src/lib_binance_execution/binance_execution/cross_margin/trade_data_stream/trade_data_subscription.cpp

@@ -16,7 +16,7 @@
 namespace binance_execution::cross_margin::trade_data_stream {
 
 #define _LOG_FN(level, fn, ...)                                                \
-   LOG(OrderMan, level, "Binance::" fn ". " __VA_ARGS__)
+   LOG(OrderMan, level, "BinanceCrossMargin::" fn ". " __VA_ARGS__)
 #define DIAG_FN(fn, ...) _LOG_FN(bts::Diag, fn, __VA_ARGS__)
 #define INFO_FN(fn, ...) _LOG_FN(bts::Info, fn, __VA_ARGS__)
 #define WARN_FN(fn, ...) _LOG_FN(bts::Warn, fn, __VA_ARGS__)
@@ -58,6 +58,7 @@ struct TradeDataSubscription::PImpl {
       LOG(OrderMan, bts::Info, "Creating websocket client. url: {}", url);
       WebsocketClientConfig ws_stream_client_cfg;
       ws_stream_client_cfg.url = url;
+      ws_stream_client_cfg.name = "BinanceCrossMargin_TradeData";
       auto &ws_cli = ws_client.emplace(ws_stream_client_cfg);
       ws_cli.message_handler().bind<&PImpl::on_message>(*this);
       ws_cli.close_handler().bind<&PImpl::on_close>(*this);

src/lib_binance_execution/binance_execution/listen_key_requestor.cpp

@@ -17,7 +17,8 @@ using std::thread;
 struct ListenKeyRequestor::PImpl {
    std::function<RestApiResult<ListenKeyResponse>(const ListenKeyRequest &)>
        get_listen_key;
-   std::function<RestApiResult<ListenKeyResponse>(const ListenKeyRequest &)>
+   std::function<RestApiResult<RefreshListenKeyResponse>(
+       const RefreshListenKeyRequest &)>
        refresh_listen_key;
    string listen_key;
    TimeStamp listen_key_timestamp;
@@ -58,7 +59,7 @@ struct ListenKeyRequestor::PImpl {
             callback(listen_key);
          }
          if ((now - listen_key_timestamp) > refresh_period) {
-            const auto &res = refresh_listen_key({});
+            const auto &res = refresh_listen_key({listen_key});
             bts_assert_msgf(
                 res.ok(), "Failed to refresh listen key. {}",
                 json_formatter(res));
@@ -72,8 +73,8 @@ struct ListenKeyRequestor::PImpl {
 ListenKeyRequestor::ListenKeyRequestor(
     std::function<RestApiResult<ListenKeyResponse>(const ListenKeyRequest &)>
         get_listen_key,
-    std::function<RestApiResult<ListenKeyResponse>(const ListenKeyRequest &)>
-        refresh_listen_key)
+    std::function<RestApiResult<RefreshListenKeyResponse>(
+        const RefreshListenKeyRequest &)> refresh_listen_key)
     : me_(std::make_unique<PImpl>(get_listen_key, refresh_listen_key)) {}
 
 ListenKeyRequestor::~ListenKeyRequestor() = default;

src/lib_binance_execution/binance_execution/listen_key_requestor.h

@@ -2,6 +2,8 @@
 
 #include "binance_execution/messages/listen_key_request.h"
 #include "binance_execution/messages/listen_key_response.h"
+#include "binance_execution/messages/refresh_listen_key_request.h"
+#include "binance_execution/messages/refresh_listen_key_response.h"
 #include "binance_execution/rest_api_result.h"
 #include "lib/utils/Delegate.h"
 
@@ -19,8 +21,8 @@ class ListenKeyRequestor {
    explicit ListenKeyRequestor(
        std::function<RestApiResult<ListenKeyResponse>(const ListenKeyRequest &)>
            get_listen_key,
-       std::function<RestApiResult<ListenKeyResponse>(const ListenKeyRequest &)>
-           refresh_listen_key);
+       std::function<RestApiResult<RefreshListenKeyResponse>(
+           const RefreshListenKeyRequest &)> refresh_listen_key);
    ~ListenKeyRequestor();
 
    void start();

src/lib_binance_execution/binance_execution/messages/refresh_listen_key_request.h

@@ -0,0 +1,31 @@
+#pragma once
+
+#include "lib/io/JsonReader.h"
+
+#include <nlohmann/json.hpp>
+#include <string>
+
+namespace binance_execution {
+
+struct RefreshListenKeyRequest {
+   std::string listenKey;
+};
+
+inline void to_json(nlohmann::json &j, const RefreshListenKeyRequest &req) {
+   j["listenKey"] = req.listenKey;
+}
+
+inline void from_json(const nlohmann::json &j, RefreshListenKeyRequest &req) {
+   const bts::io::JsonReader jr{&j};
+   jr.read_or("listenKey", req.listenKey);
+}
+
+inline void to_query_params(
+    std::string &qps, const RefreshListenKeyRequest &req) {
+   if (!req.listenKey.empty()) {
+      qps.append("listenKey=");
+      qps.append(req.listenKey);
+   }
+}
+
+} // namespace binance_execution

src/lib_binance_execution/binance_execution/messages/refresh_listen_key_response.h

@@ -0,0 +1,25 @@
+#pragma once
+
+#include "lib/io/JsonReader.h"
+
+#include <nlohmann/json.hpp>
+#include <string>
+
+namespace binance_execution {
+
+struct RefreshListenKeyResponse {
+   std::string listenKey;
+};
+
+inline void to_json(nlohmann::json &j, const RefreshListenKeyResponse &res) {
+   if (!res.listenKey.empty()) {
+      j["listenKey"] = res.listenKey;
+   }
+}
+
+inline void from_json(const nlohmann::json &j, RefreshListenKeyResponse &res) {
+   const bts::io::JsonReader jr{&j};
+   jr.read_or("listenKey", res.listenKey);
+}
+
+} // namespace binance_execution

src/lib_binance_execution/binance_execution/spots/rest_api_client.cpp

@@ -54,6 +54,7 @@ void RestApiClient::request_(
     const std::string &path,
     bts::io::http::HttpMethod method,
     const std::string &body,
+    bool sign,
     RestApiResult<ResponseT> &res) {
    // Use thread_local vars to reduce memory allocations. Shrink strings
    // to reduce memory footprint from previous requests.
@@ -71,7 +72,9 @@ void RestApiClient::request_(
    qps.clear();
    to_query_params(qps, req);
    if (!qps.empty()) {
-      sign_query_params_(qps);
+      if (sign) {
+         sign_query_params_(qps);
+      }
       http_req.url.append("?");
       http_req.url.append(qps);
    }
@@ -104,7 +107,7 @@ void RestApiClient::request_(
 #define REQUEST_DEF(ResponseT, RequestT)                                       \
    template void RestApiClient::request_<ResponseT, RequestT>(                 \
        const RequestT &req, const std::string &path,                           \
-       bts::io::http::HttpMethod method, const std::string &body,              \
+       bts::io::http::HttpMethod method, const std::string &body, bool sign,   \
        RestApiResult<ResponseT> &res)
 
 REQUEST_DEF(OrderRecord, CreateOrderRequest);
@@ -112,5 +115,6 @@ REQUEST_DEF(ModifyOrderResponse, ModifyOrderRequest);
 REQUEST_DEF(OrderRecord, CancelOrderRequest);
 REQUEST_DEF(std::vector<OrderRecord>, AllOrdersRequest);
 REQUEST_DEF(ListenKeyResponse, ListenKeyRequest);
+REQUEST_DEF(RefreshListenKeyResponse, RefreshListenKeyRequest);
 
 } // namespace binance_execution::spots

src/lib_binance_execution/binance_execution/spots/rest_api_client.h

@@ -2,6 +2,8 @@
 
 #include "binance_execution/messages/listen_key_request.h"
 #include "binance_execution/messages/listen_key_response.h"
+#include "binance_execution/messages/refresh_listen_key_request.h"
+#include "binance_execution/messages/refresh_listen_key_response.h"
 #include "binance_execution/rest_api_result.h"
 #include "binance_execution/spots/messages/all_orders_request.h"
 #include "binance_execution/spots/messages/cancel_order_request.h"
@@ -34,8 +36,8 @@ class RestApiClient {
        const AllOrdersRequest &req);
    const RestApiResult<ListenKeyResponse> &get_listen_key(
        const ListenKeyRequest &req = {});
-   const RestApiResult<ListenKeyResponse> &refresh_listen_key(
-       const ListenKeyRequest &req = {});
+   const RestApiResult<RefreshListenKeyResponse> &refresh_listen_key(
+       const RefreshListenKeyRequest &req = {});
 
  private:
    std::string url_;
@@ -48,6 +50,7 @@ class RestApiClient {
        const std::string &path,
        bts::io::http::HttpMethod method,
        const std::string &body,
+       bool sign,
        RestApiResult<ResponseT> &op_res);
 
    bool sign_query_params_(std::string &qp);
@@ -67,7 +70,8 @@ inline const RestApiResult<OrderRecord> &RestApiClient::create_order(
    static const std::string path{"/api/v3/order"};
    thread_local RestApiResult<OrderRecord> res;
    res.reset();
-   request_<OrderRecord>(req, path, bts::io::http::HttpMethod::POST, "", res);
+   request_<OrderRecord>(
+       req, path, bts::io::http::HttpMethod::POST, "", true, res);
    return res;
 }
 
@@ -76,7 +80,8 @@ inline const RestApiResult<OrderRecord> &RestApiClient::create_test_order(
    static const std::string path{"/api/v3/order/test"};
    thread_local RestApiResult<OrderRecord> res;
    res.reset();
-   request_<OrderRecord>(req, path, bts::io::http::HttpMethod::POST, "", res);
+   request_<OrderRecord>(
+       req, path, bts::io::http::HttpMethod::POST, "", true, res);
    return res;
 }
 
@@ -86,7 +91,7 @@ inline const RestApiResult<ModifyOrderResponse> &RestApiClient::modify_order(
    thread_local RestApiResult<ModifyOrderResponse> res;
    res.reset();
    request_<ModifyOrderResponse>(
-       req, path, bts::io::http::HttpMethod::POST, "", res);
+       req, path, bts::io::http::HttpMethod::POST, "", true, res);
    return res;
 }
 
@@ -95,7 +100,8 @@ inline const RestApiResult<OrderRecord> &RestApiClient::cancel_order(
    static const std::string path{"/api/v3/order"};
    thread_local RestApiResult<OrderRecord> res;
    res.reset();
-   request_<OrderRecord>(req, path, bts::io::http::HttpMethod::DELETE, "", res);
+   request_<OrderRecord>(
+       req, path, bts::io::http::HttpMethod::DELETE, "", true, res);
    return res;
 }
 
@@ -105,7 +111,7 @@ RestApiClient::get_all_orders(const AllOrdersRequest &req) {
    thread_local RestApiResult<std::vector<OrderRecord>> res;
    res.reset();
    request_<std::vector<OrderRecord>>(
-       req, path, bts::io::http::HttpMethod::GET, "", res);
+       req, path, bts::io::http::HttpMethod::GET, "", true, res);
    return res;
 }
 
@@ -114,16 +120,16 @@ inline const RestApiResult<ListenKeyResponse> &RestApiClient::get_listen_key(
    static const std::string path{"/api/v3/userDataStream"};
    thread_local RestApiResult<ListenKeyResponse> res;
    res.reset();
-   request_(req, path, bts::io::http::HttpMethod::POST, "", res);
+   request_(req, path, bts::io::http::HttpMethod::POST, "", false, res);
    return res;
 }
 
-inline const RestApiResult<ListenKeyResponse> &
-RestApiClient::refresh_listen_key(const ListenKeyRequest &req) {
+inline const RestApiResult<RefreshListenKeyResponse> &
+RestApiClient::refresh_listen_key(const RefreshListenKeyRequest &req) {
    static const std::string path{"/api/v3/userDataStream"};
-   thread_local RestApiResult<ListenKeyResponse> res;
+   thread_local RestApiResult<RefreshListenKeyResponse> res;
    res.reset();
-   request_(req, path, bts::io::http::HttpMethod::PUT, "", res);
+   request_(req, path, bts::io::http::HttpMethod::PUT, "", false, res);
    return res;
 }
 

src/lib_binance_execution/binance_execution/spots/spots_gateway.cpp

@@ -280,7 +280,7 @@ struct SpotsGateway::PImpl {
       user_data_stream::OrderUpdate curr;
       auto &km = *key_map;
       while (incoming_order_updates.pop(curr)) {
-         auto bts_key = km.get_bts_key(curr.client_order_id);
+         const auto bts_key = km.get_bts_key(curr.client_order_id);
          if (!bts_key) {
             ERROR_FN(
                 "dispatch_order_updates",
@@ -289,7 +289,7 @@ struct SpotsGateway::PImpl {
                 curr.client_order_id);
             continue;
          }
-         auto exch_key = km.get_exch_key(*bts_key);
+         const auto exch_key = km.get_exch_key(*bts_key);
          if (exch_key != curr.client_order_id) {
             INFO_FN(
                 "dispatch_order_updates",

src/lib_binance_execution/binance_execution/spots/user_data_stream/user_data_subscription.cpp

@@ -58,6 +58,7 @@ struct UserDataSubscription::PImpl {
       LOG(OrderMan, bts::Info, "Creating websocket client. url: {}", url);
       WebsocketClientConfig ws_stream_client_cfg;
       ws_stream_client_cfg.url = url;
+      ws_stream_client_cfg.name = "Binance_UserData";
       auto &ws_cli = ws_client.emplace(ws_stream_client_cfg);
       ws_cli.message_handler().bind<&PImpl::on_message>(*this);
       ws_cli.close_handler().bind<&PImpl::on_close>(*this);

src/lib_binance_execution/binance_execution/usds_futures/rest_api_client.cpp

@@ -52,6 +52,7 @@ void RestApiClient::request_(
     const std::string &path,
     bts::io::http::HttpMethod method,
     const std::string &body,
+    bool sign,
     RestApiResult<ResponseT> &res) {
    // Use thread_local vars to reduce memory allocations. Shrink strings
    // to reduce memory footprint from previous requests.
@@ -69,7 +70,9 @@ void RestApiClient::request_(
    qps.clear();
    to_query_params(qps, req);
    if (!qps.empty()) {
-      sign_query_params_(qps);
+      if (sign) {
+         sign_query_params_(qps);
+      }
       http_req.url.append("?");
       http_req.url.append(qps);
    }
@@ -102,7 +105,7 @@ void RestApiClient::request_(
 #define REQUEST_DEF(ResponseT, RequestT)                                       \
    template void RestApiClient::request_<ResponseT, RequestT>(                 \
        const RequestT &req, const std::string &path,                           \
-       bts::io::http::HttpMethod method, const std::string &body,              \
+       bts::io::http::HttpMethod method, const std::string &body, bool sign,   \
        RestApiResult<ResponseT> &res)
 
 REQUEST_DEF(CreateOrderResponse, CreateOrderRequest);
@@ -111,5 +114,6 @@ REQUEST_DEF(CancelOrderResponse, CancelOrderRequest);
 REQUEST_DEF(std::vector<OrderRecord>, AllOrdersRequest);
 REQUEST_DEF(std::vector<PositionInfo>, PositionInfoRequest);
 REQUEST_DEF(ListenKeyResponse, ListenKeyRequest);
+REQUEST_DEF(RefreshListenKeyResponse, RefreshListenKeyRequest);
 
 } // namespace binance_execution::usds_futures

src/lib_binance_execution/binance_execution/usds_futures/rest_api_client.h

@@ -2,6 +2,8 @@
 
 #include "binance_execution/messages/listen_key_request.h"
 #include "binance_execution/messages/listen_key_response.h"
+#include "binance_execution/messages/refresh_listen_key_request.h"
+#include "binance_execution/messages/refresh_listen_key_response.h"
 #include "binance_execution/rest_api_result.h"
 #include "binance_execution/usds_futures/messages/all_orders_request.h"
 #include "binance_execution/usds_futures/messages/cancel_order_request.h"
@@ -38,8 +40,8 @@ class RestApiClient {
        const PositionInfoRequest &req);
    const RestApiResult<ListenKeyResponse> &get_listen_key(
        const ListenKeyRequest &req = {});
-   const RestApiResult<ListenKeyResponse> &refresh_listen_key(
-       const ListenKeyRequest &req = {});
+   const RestApiResult<RefreshListenKeyResponse> &refresh_listen_key(
+       const RefreshListenKeyRequest &req = {});
 
  private:
    std::string url_;
@@ -52,6 +54,7 @@ class RestApiClient {
        const std::string &path,
        bts::io::http::HttpMethod method,
        const std::string &body,
+       bool sign,
        RestApiResult<ResponseT> &op_res);
 
    bool sign_query_params_(std::string &qp);
@@ -72,7 +75,7 @@ inline const RestApiResult<CreateOrderResponse> &RestApiClient::create_order(
    thread_local RestApiResult<CreateOrderResponse> res;
    res.reset();
    request_<CreateOrderResponse>(
-       req, path, bts::io::http::HttpMethod::POST, "", res);
+       req, path, bts::io::http::HttpMethod::POST, "", true, res);
    return res;
 }
 
@@ -82,7 +85,7 @@ inline const RestApiResult<ModifyOrderResponse> &RestApiClient::modify_order(
    thread_local RestApiResult<ModifyOrderResponse> res;
    res.reset();
    request_<ModifyOrderResponse>(
-       req, path, bts::io::http::HttpMethod::PUT, "", res);
+       req, path, bts::io::http::HttpMethod::PUT, "", true, res);
    return res;
 }
 
@@ -92,7 +95,7 @@ inline const RestApiResult<CancelOrderResponse> &RestApiClient::cancel_order(
    thread_local RestApiResult<CancelOrderResponse> res;
    res.reset();
    request_<CancelOrderResponse>(
-       req, path, bts::io::http::HttpMethod::DELETE, "", res);
+       req, path, bts::io::http::HttpMethod::DELETE, "", true, res);
    return res;
 }
 
@@ -102,7 +105,7 @@ RestApiClient::get_all_orders(const AllOrdersRequest &req) {
    thread_local RestApiResult<std::vector<OrderRecord>> res;
    res.reset();
    request_<std::vector<OrderRecord>>(
-       req, path, bts::io::http::HttpMethod::GET, "", res);
+       req, path, bts::io::http::HttpMethod::GET, "", true, res);
    return res;
 }
 
@@ -111,7 +114,7 @@ RestApiClient::get_position_info(const PositionInfoRequest &req) {
    static const std::string path{"/fapi/v3/positionRisk"};
    thread_local RestApiResult<std::vector<PositionInfo>> res;
    res.reset();
-   request_(req, path, bts::io::http::HttpMethod::GET, "", res);
+   request_(req, path, bts::io::http::HttpMethod::GET, "", true, res);
    return res;
 }
 
@@ -120,16 +123,16 @@ inline const RestApiResult<ListenKeyResponse> &RestApiClient::get_listen_key(
    static const std::string path{"/fapi/v1/listenKey"};
    thread_local RestApiResult<ListenKeyResponse> res;
    res.reset();
-   request_(req, path, bts::io::http::HttpMethod::POST, "", res);
+   request_(req, path, bts::io::http::HttpMethod::POST, "", false, res);
    return res;
 }
 
-inline const RestApiResult<ListenKeyResponse> &
-RestApiClient::refresh_listen_key(const ListenKeyRequest &req) {
+inline const RestApiResult<RefreshListenKeyResponse> &
+RestApiClient::refresh_listen_key(const RefreshListenKeyRequest &req) {
    static const std::string path{"/fapi/v1/listenKey"};
-   thread_local RestApiResult<ListenKeyResponse> res;
+   thread_local RestApiResult<RefreshListenKeyResponse> res;
    res.reset();
-   request_(req, path, bts::io::http::HttpMethod::PUT, "", res);
+   request_(req, path, bts::io::http::HttpMethod::PUT, "", false, res);
    return res;
 }
 

src/lib_binance_execution/binance_execution/usds_futures/user_data_stream/user_data_subscription.cpp

@@ -54,6 +54,7 @@ struct UserDataSubscription::PImpl {
       LOG(OrderMan, bts::Info, "Creating websocket client. url: {}", url);
       WebsocketClientConfig ws_stream_client_cfg;
       ws_stream_client_cfg.url = url;
+      ws_stream_client_cfg.name = "BinanceUsdsFutures_UserData";
       auto &ws_cli = ws_client.emplace(ws_stream_client_cfg);
       ws_cli.message_handler().bind<&PImpl::on_message>(*this);
       ws_cli.close_handler().bind<&PImpl::on_close>(*this);

src/server/sparkd/configurations/binance_dev/spark.conf

@@ -1,9 +1,5 @@
 ExchangeSubAccounts {
   value {
-    accounts {
-      exchange_subaccount: "binance_dev"
-      bts_account_name: "BINANCE_DEV"
-    }
     accounts{
       exchange_subaccount: "ZmIKL6fb"
       bts_account_name: "MAIN"     
@@ -50,15 +46,15 @@ Logging {
     initial_queue_size: 120001
     max_queue_growth: 4
     console {
-      default_level: "Info"
-      # subsystem {
-      #   name: "OrderMan"
-      #   level: "Diag"
-      # }
-      # subsystem {
-      #   name: "Http"
-      #   level: "Diag"
-      # }
+      default_level: "Warn"
+      subsystem {
+        name: "OrderMan"
+        level: "Diag"
+      }
+      subsystem {
+        name: "Http"
+        level: "Diag"
+      }
     }
     file {
       default_level: "Info"
@@ -101,6 +97,7 @@ MarketDataConfig {
           symbols: "DOGE"
           symbols: "EIGEN"
           symbols: "ETH"
+          name_regexps: "USDC|USDT$"
         }
       }
     }

src/server/sparkd/md/databento/DatabentoDataHandler.cpp

@@ -1,5 +1,6 @@
 #include "DatabentoDataHandler.h"
 #include "databento/datetime.hpp"
+#include "databento/dbn_encoder.hpp"
 #include "databento/enums.hpp"
 #include "databento/exceptions.hpp"
 #include "databento/live_blocking.hpp"
@@ -7,6 +8,7 @@
 #include "databento/record.hpp"
 #include "databento/timeseries.hpp"
 #include "lib/api/domain/Venue.pb.h"
+#include "lib/api/md/MarketDataConfig.pb.h"
 #include "lib/config/config.h"
 #include "lib/config/protos/system_config.pb.h"
 #include "lib/core/TimeStamp.h"
@@ -22,14 +24,21 @@
 #include "lib/utils/TimeStamp.h"
 #include "lib/utils/bts_assert.h"
 #include "lib/utils/bts_chrono.h"
+#include "lib/utils/datetime.h"
 #include "lib/utils/helper_macros.h"
+#include <boost/filesystem/operations.hpp>
 #include <cassert>
 #include <chrono>
+#include <cstddef>
 #include <databento/constants.hpp>
 #include <databento/historical.hpp>
 #include <databento/live.hpp>
 #include <optional>
+#include <ratio>
 #include <string>
+#include <string_view>
+#include <tuple>
+#include <unordered_map>
 #include <vector>
 
 namespace bts::md::DatabentoHandler {
@@ -61,7 +70,87 @@ static SecurityType parse_sec_type(std::string_view str) {
    }
    return SEC_TYPE(UNSPECIFIED);
 }
+void update_file(
+    const std::string &key,
+    const std::string &dataset,
+    const std::vector<std::string> &parent_symbols,
+    const uint time_to_run,
+    std::string file_path) {
 
+   auto client =
+       databento::LiveBuilder{}
+           .SetKey(key)
+           .SetDataset(dataset)
+           .SetUpgradePolicy(databento::VersionUpgradePolicy::UpgradeToV2)
+           .BuildBlocking();
+
+   client.Subscribe(
+       parent_symbols, databento::Schema::Definition, databento::SType::Parent,
+       databento::UnixNanos{});
+   // We start streaming
+   databento::Metadata metadata = client.Start();
+   {
+      // We open a file for writing and encode the metadata
+      databento::OutFileStream out_file{file_path};
+      databento::DbnEncoder encoder{metadata, &out_file};
+      // Wait for at least one record and 5 seconds before closing the connection
+      const std::chrono::seconds timeout{time_to_run};
+      const auto start = std::chrono::steady_clock::now();
+      while (auto *record = client.NextRecord(timeout)) {
+         encoder.EncodeRecord(*record);
+         if (std::chrono::steady_clock::now() - start >= timeout) {
+            break;
+         }
+      }
+   }
+   return;
+}
+void log_time_statistics(
+    const std::vector<std::vector<std::tuple<uint64_t, std::string>>>
+        &all_times) {
+   std::unordered_map<std::string, std::vector<uint64_t>> subpart_times;
+
+   // Collect all timing values per subpart
+   for (const auto &entry : all_times) {
+      for (const auto &[time, label] : entry) {
+         subpart_times[label].push_back(time);
+      }
+   }
+
+   // Function to compute statistics
+   auto compute_statistics = [](std::vector<uint64_t> &values)
+       -> std::tuple<double, double, double, double, double, double, double> {
+      if (values.empty())
+         return std::make_tuple(0.0, 0.0, 0, 0, 0, 0, 0);
+
+      std::sort(values.begin(), values.end());
+
+      double sum = std::accumulate(values.begin(), values.end(), 0.0);
+      double avg = sum / values.size();
+
+      double sq_sum =
+          std::inner_product(values.begin(), values.end(), values.begin(), 0.0);
+      double stddev = std::sqrt(sq_sum / values.size() - avg * avg);
+
+      auto percentile = [&](double p) -> uint64_t {
+         size_t idx = static_cast<size_t>(p * (values.size() - 1));
+         return values[idx];
+      };
+
+      return std::make_tuple(
+          avg, stddev, percentile(0.00), percentile(0.25), percentile(0.50),
+          percentile(0.75), percentile(1.00));
+   };
+
+   // Compute and log statistics
+   for (auto &[label, times] : subpart_times) {
+      auto [avg, stddev, p0, p25, p50, p75, p100] = compute_statistics(times);
+      LOG(md, Sys,
+          "Subpart: {}, Avg: {:.2f} ns, Std: {:.2f} ns, P0: {} ns, P25: {} ns, "
+          "P50: {} ns, P75: {} ns, P100: {} ns",
+          label, avg, stddev, p0, p25, p50, p75, p100);
+   }
+}
 void DatabentoDatahandler::run(
     bool use_file,
     std::shared_ptr<IMarketDataListener> listener,
@@ -74,59 +163,71 @@ void DatabentoDatahandler::run(
     const uint filter_securities_upto_n_months,
     const bool load_reference_data,
     const std::chrono::duration<long double> sleep_duration) {
-
+   uint DEBUG_SEC_ID = 0;
+   if (venue == api::domain::VENUE_CBOE)
+      DEBUG_SEC_ID = config::get<bts::api::md::MarketDataConfig>()
+                         .cboe()
+                         .at(0)
+                         .databento_settings()
+                         .debug_sec();
    auto sleep_duration_updated = sleep_duration;
-
+   std::vector<std::vector<std::tuple<uint64_t, std::string>>> all_times;
+   std::unordered_map<uint32_t, bts::domain::SecurityID> sec_id_map;
    if (!use_file) {
       sleep_duration_updated = std::chrono::duration<
           long double>::zero(); // Set to zero for live data
    }
-   auto process_security = [&](const databento::Record &record) {
-      const auto &msg = record.Get<databento::InstrumentDefMsg>();
-      std::string id = remove_spaces(msg.RawSymbol());
-      std::string symbol;
-
+   auto sec_filter = [&](const core::TimeStamp expiration_time,
+                         const std::string &symbol,
+                         const api::domain::Venue &venue, uint strike) {
+      uint months = filter_securities_upto_n_months;
       switch (venue) {
       case api::domain::Venue::VENUE_CBOE: {
-         symbol = remove_spaces(
-             std::string(msg.underlying.data(), msg.underlying.size()));
+         if (strike < 3800 || strike > 8000) {
+            return false;
+         }
+         if (symbol == "SPX")
+            months = 6;
+         else if (symbol == "SPXW")
+            months = 1;
          break;
       }
       case api::domain::Venue::VENUE_GLBX:
-         if (id.length() > 5)
-            return databento::KeepGoing::Continue;
-         symbol = id.substr(0, 2);
-         //  sec.security_common.cpp = msg.unit_of_measure_qty;
-
-         break;
+         return true;
       default:
          bts_assert_msg(false, "Exchange not supported on databento");
-         return databento::KeepGoing::Stop;
+      }
+
+      auto duration = months * 32 * 24 * 60 * 60;
+      return expiration_time.seconds() <
+             bts::core::TimeStamp::now().seconds() + duration;
+   };
+   auto process_security = [&](const databento::Record &record) -> bool {
+      const auto &msg = record.Get<databento::InstrumentDefMsg>();
+      const auto exp_nanos = msg.expiration;
+      std::string exp_string = databento::ToIso8601(exp_nanos);
+      const auto expiration_time =
+          bts::core::TimeStamp::from_iso_string(exp_string);
+      if (!sec_filter(
+              expiration_time, msg.Underlying(), venue,
+              msg.strike_price / 1000000000)) {
+         return false;
       }
       bts::domain::Security sec;
       sec.security_common = domain::SecurityCommon();
       sec.security_common.venue = venue;
       sec.security_common.exchange_sec_id = msg.hd.instrument_id;
       sec.security_common.desc = msg.RawSymbol();
-      sec.security_common.symbol = msg.Asset();
+      sec.security_common.symbol = msg.Underlying();
       sec.security_common.asset = msg.Asset();
       sec.security_common.group = msg.Group();
       sec.security_common.min_price_increment = 0.01;
       sec.security_common.min_quantity = 0.01;
       sec.security_common.min_quantity_increment = 0.01;
       sec.security_common.currency = {msg.currency.data(), msg.currency.size()};
-      // sec.security_common.cpp = msg.hd.
 
       const SecurityType sec_type =
           parse_sec_type({msg.security_type.data(), msg.security_type.size()});
-      const auto exp_nanos = msg.expiration;
-      std::string exp_string = databento::ToIso8601(exp_nanos);
-      const auto expiration_time =
-          bts::core::TimeStamp::from_iso_string(exp_string);
-      auto duration = filter_securities_upto_n_months*30*24*60*60;
-      if( expiration_time.seconds() >= bts::core::TimeStamp::now().seconds() + duration){
-         return databento::KeepGoing::Continue;
-      }
 
       switch (sec_type) {
       case SEC_TYPE(FUTURE):
@@ -134,11 +235,10 @@ void DatabentoDatahandler::run(
          break;
       case SEC_TYPE(OPTION): {
          auto call_put = CallPut::PUT;
-         if (msg.instrument_class == 'C') {
+         if (msg.instrument_class == 'C')
             call_put = CallPut::CALL;
-         } else {
+         else
             bts_assert(msg.instrument_class == 'P');
-         }
          sec.security.emplace<bts::domain::OptionSecurity>(
              expiration_time, bts::domain::Price(msg.strike_price, -9),
              call_put);
@@ -147,36 +247,62 @@ void DatabentoDatahandler::run(
       default:
          bts_assert_msg(
              false, "Only futures and options supported on databento");
-         return databento::KeepGoing::Stop;
       }
-
       *sec.security_common.id = sec_id_store->get_or_add(
           venue, sec_type, std::to_string(msg.hd.instrument_id));
       listener->on_securities({&sec});
-      return databento::KeepGoing::Continue;
+      sec_id_map.insert({msg.hd.instrument_id, sec.security_common.id});
+      return true;
+   };
+   auto debug_log = [&](const databento::Mbp1Msg &msg) {
+      if (DEBUG_SEC_ID != msg.hd.instrument_id)
+         return;
+      const double denom = 1000000000;
+      // auto delta = std::to_string(std::chrono::duration<double, std::micro>(msg.ts_in_delta).count());
+      auto seq_num = msg.sequence;
+      auto ap = msg.levels[0].ask_px / denom;
+      auto bp = msg.levels[0].bid_px / denom;
+      auto as = msg.levels[0].ask_sz;
+      auto bs = msg.levels[0].bid_sz;
+      LOG(md, Sys, "ID: {}, seq_num: {}, B_P: {}, B_S: {}, A_P: {}, A_S: {}",
+          msg.hd.instrument_id, seq_num, bp, bs, ap, as);
    };
 
    auto process_bbo = [&](const databento::Record &record) {
       const auto &msg = record.Get<databento::Mbp1Msg>();
+      debug_log(msg);
+
+      // std::vector<std::tuple<uint64_t,std::string>>& times = all_times.emplace_back();
+      // times.reserve(8);
+      // return databento::KeepGoing::Continue; //only added so the same functions work with the threaded builder, not really need in the blocking version
       // The security type to is needed to get the id. Since the sec type isn't known,
       // check for an id using both types (OPTION and FUTURE). This is safe to do since
       // options and futures never have common names (instrument ids).
-      const auto name = std::to_string(msg.hd.instrument_id);
-      auto sec_id_opt =
-          sec_id_store->get_id(venue, SEC_TYPE(OPTION), name, false);
-      if (!sec_id_opt.has_value()) {
-         sec_id_opt =
-             sec_id_store->get_id(venue, SEC_TYPE(FUTURE), name, false);
-         if (!sec_id_opt.has_value()) {
-            return databento::KeepGoing::Continue;
-         }
-      }
+      // auto current = bts::datetime::unix_nano_now();
+      // auto prev = current;
+      // times.push_back({current-prev,"Start Process bbo"});
+
+      // prev = current;
+      // current = bts::datetime::unix_nano_now();
+      // times.push_back({current-prev,"MISC to check log time"});
+      if (!sec_id_map.contains(msg.hd.instrument_id))
+         return databento::KeepGoing::Continue;
+      bts::domain::SecurityID sec_id = sec_id_map.at(msg.hd.instrument_id);
+
+      // prev = current;
+      // current = bts::datetime::unix_nano_now();
+      // times.push_back({current-prev,"Sec_ID done"});
+
+      // sleep_for(sleep_duration_updated);
+      // prev = current;
+      // current = bts::datetime::unix_nano_now();
+      // times.push_back({current-prev,"Sleep done"});
 
-      sleep_for(sleep_duration_updated);
-      bts::domain::SecurityID sec_id =
-          bts::domain::SecurityID(sec_id_opt.value());
       domain::Book book;
       book.set_security_id(sec_id);
+      // prev = current;
+      // current = bts::datetime::unix_nano_now();
+      // times.push_back({current-prev,"Empty Book Made"});
 
       auto &bids = book.bids();
       auto &asks = book.asks();
@@ -186,130 +312,215 @@ void DatabentoDatahandler::run(
       asks.begin()->price = bts::domain::Price(msg.levels[0].ask_px, -9);
       asks.begin()->total_quantity =
           bts::domain::Quantity(msg.levels[0].ask_sz);
+      //     prev = current;
+      //     current = bts::datetime::unix_nano_now();
+      // times.push_back({current-prev,"Complete book done"});
 
       auto recv_time = bts::core::TimeStamp::from_iso_string(
           databento::ToIso8601(msg.ts_recv));
+      // prev = current;
+      // current = bts::datetime::unix_nano_now();
+      // times.push_back({current-prev,"Time converted"});
+
       listener->on_book(
           book,
           utils::TimeStamp{datetime::ptime_to_timestamp(recv_time.to_ptime())});
-      return databento::KeepGoing::Continue;
+      // prev = current;
+      // current = bts::datetime::unix_nano_now();
+      // times.push_back({current-prev,"On book done"});
+
+      return databento::KeepGoing::
+          Continue; //only added so the same functions work with the threaded builder, not really need in the blocking version
    };
+
    auto process_trades = [&](const databento::Record &record) {
       const auto &msg = record.Get<databento::TradeMsg>();
       // The security type to is needed to get the id. Since the sec type isn't known,
       // check for an id using both types (OPTION and FUTURE). This is safe to do since
       // options and futures never have common names (instrument ids).
-      const auto name = std::to_string(msg.hd.instrument_id);
-      auto sec_id_opt =
-          sec_id_store->get_id(venue, SEC_TYPE(OPTION), name, false);
-      if (!sec_id_opt.has_value()) {
-         sec_id_opt =
-             sec_id_store->get_id(venue, SEC_TYPE(FUTURE), name, false);
-         if (!sec_id_opt.has_value()) {
-            return databento::KeepGoing::Continue;
-         }
-      }
+      if (!sec_id_map.contains(msg.hd.instrument_id))
+         return databento::KeepGoing::Continue;
+      bts::domain::SecurityID sec_id = sec_id_map.at(msg.hd.instrument_id);
 
-      sleep_for(sleep_duration_updated);
-      bts::domain::SecurityID sec_id =
-          bts::domain::SecurityID(sec_id_opt.value());
+      // sleep_for(sleep_duration_updated);
       domain::TradeRecord trade;
       trade.trade_price = bts::domain::Price(msg.price, -9);
       trade.trade_quantity = bts::domain::Quantity(msg.size);
-      trade.sec_id =  sec_id;
-      switch(msg.side){
-         case databento::Side::Ask:
-            trade.agg_side = domain::AggressorSide::Sell;
-            break;
-         case databento::Side::Bid:
-            trade.agg_side = domain::AggressorSide::Buy;
-            break;
-         case databento::Side::None:
-            trade.agg_side = domain::AggressorSide::None;
-            break;
+      trade.sec_id = sec_id;
+      switch (msg.side) {
+      case databento::Side::Ask:
+         trade.agg_side = domain::AggressorSide::Sell;
+         break;
+      case databento::Side::Bid:
+         trade.agg_side = domain::AggressorSide::Buy;
+         break;
+      case databento::Side::None:
+         trade.agg_side = domain::AggressorSide::None;
+         break;
       }
       trade.time_stamp = bts::core::TimeStamp::from_iso_string(
           databento::ToIso8601(msg.ts_recv));
       std::vector<const domain::TradeRecord *> trade_ptrs;
       trade_ptrs.push_back(&trade);
       listener->on_trades(trade_ptrs);
-     
-      return databento::KeepGoing::Continue;
+
+      return databento::KeepGoing::
+          Continue; //only added so the same functions work with the threaded builder, not really need in the blocking version
    };
    if (use_file) {
       auto client = databento::HistoricalBuilder{}.SetKey(key).Build();
       databento::DbnFileStore dbn_file_store(securities_path);
-      dbn_file_store.Replay(process_security);
+      // dbn_file_store.Replay(process_security); // fix this
       databento::DbnFileStore dbn_file_store2(books_path);
       dbn_file_store2.Replay(process_bbo);
-   }else {
-    bts_assert(dataset!="");
-    // auto client = databento::LiveBuilder{}
-    //               .SetKey(key)
-    //               .SetDataset(dataset)
-    //               .BuildThreaded();
-    auto client = databento::LiveBlocking{
-        databento::ILogReceiver::Default(),
-        key,
-        dataset,
-        true,
-        databento::VersionUpgradePolicy::UpgradeToV2,
-        {}
-    };
-    //hardcoding for now will update later
-    std::vector<std::string> symbols;
-    switch (venue) {
-        case api::domain::VENUE_CBOE: symbols= {"SPX.OPT","SPXW.OPT"}; break;
-        case api::domain::VENUE_GLBX: symbols= {"ES.FUT"}; break;
-        default:bts_assert_msg(false, "ONLY GLBX AND CBOE SUPPORTED FOR NOW");
-    }
-    if(load_reference_data) 
-         client.Subscribe(symbols, databento::Schema::Definition, databento::SType::Parent,databento::UnixNanos{});
-    else client.Subscribe(symbols, databento::Schema::Definition, databento::SType::Parent);
-    client.Subscribe(symbols, databento::Schema::Mbp1, databento::SType::Parent);
-    client.Subscribe(symbols, databento::Schema::Trades, databento::SType::Parent);
-    client.Start();
-    while (true) {
-    try {
-    auto* record = client.NextRecord(std::chrono::seconds{500});
-    if (record == nullptr) {
-        LOG(md, Error, "Timed out waiting for data");
-        break;
-    }
-    if (record->Holds<databento::Mbp1Msg>()) {
-        process_bbo(*record);
-    }
-    if (record->Holds<databento::InstrumentDefMsg>()) {
-        process_security(*record);
-    }
-    if (record->Holds<databento::TradeMsg>()) {
-        process_trades(*record);
-    }
-  } catch (const databento::TcpError& error) {
-    std::string err = error.what();
-    LOG(md, Error, "Caught TCP error: {} \nAttempting reconnect\n",err);
-    client.Reconnect();
-  } catch(const databento::DbnResponseError& err){
-    std::string msg = err.Message();
-    std::string what = err.what();
-    LOG(md, Error, "Caught DbnResponseError what: {} ,msg:{} \nShutting down\n",msg, what);
-    client.Stop();
-    // bts_assert(false);
-  }
-  catch(const databento::Exception& excp){
-    std::string err = excp.Message();
-    std::string err2 = excp.what();
-    LOG(md, Error, "Caught unknown dadtabento::exception what: {}, msg: {} \nShutting down\n",err, err2);
-    client.Stop();
-    // bts_assert(false);
-  }catch (const std::exception& exc) {
-    std::string err = exc.what();
-    LOG(md, Error, "Caught unknown exception: {} \nShutting down\n",err);
-    client.Stop();
-    // bts_assert(false);
-  }
-    
-    }
+   } else {
+      bts_assert(dataset != "");
+
+      //hardcoding for now will update later
+      std::vector<std::string> symbols;
+      switch (venue) {
+      case api::domain::VENUE_CBOE:
+         symbols = {"SPX.OPT", "SPXW.OPT"};
+         break;
+      case api::domain::VENUE_GLBX:
+         symbols = {"ES.FUT"};
+         break;
+      default:
+         bts_assert_msg(false, "ONLY GLBX AND CBOE SUPPORTED FOR NOW");
+      }
+      //ocassionaly run security writer to update security dbn
+      if (!boost::filesystem::exists(securities_path) || load_reference_data) {
+         update_file(key, dataset, symbols, 15 * 60, securities_path);
+      }
+
+      //regular flow
+      //read from security dbn, filter contracts that we need and subscribe to those in batches of two thousand
+      LOG(md, Info, "starting loading symbols");
+      databento::DbnFileStore store{securities_path};
+      std::vector<std::vector<std::string>> all_sec;
+
+      all_sec.emplace_back();
+      all_sec.back().reserve(2000);
+      size_t total_count = 0;
+
+      while (auto *record = store.NextRecord()) {
+         if (auto *def = record->GetIf<databento::InstrumentDefMsg>()) {
+            std::string s = def->RawSymbol();
+            if (process_security(*record)) {
+               ++total_count;
+               // Check if the last vector is full
+               if (all_sec.back().size() == 2000) {
+                  all_sec.emplace_back();
+                  all_sec.back().reserve(2000);
+                  LOG(md, Info, "{} symbols added", total_count);
+               }
+               all_sec.back().push_back(std::move(s)); // Move to avoid copying
+            }
+         }
+      }
+      // std::ostringstream output;
+
+      // for (const auto& sec_group : all_sec) {
+      //    for (size_t i = 0; i < sec_group.size(); ++i) {
+      //       if (i > 0) output << ", ";
+      //       output << sec_group[i];
+      //    }
+      //    output << '\n';  // Newline after each group
+      // }
+
+      // // Store the final output in a stringls
+
+      // std::string final_output = output.str();
+      // size_t size = all_sec.size();
+      LOG(md, Info, "total {} secs loaded", total_count);
+
+
+      bool first = true;
+      auto client = databento::LiveBlocking{
+          databento::ILogReceiver::Default(),           key, dataset, true,
+          databento::VersionUpgradePolicy::UpgradeToV2, {}};
+      std::string start = config::get<bts::api::md::MarketDataConfig>()
+          .cboe()
+          .at(0)
+          .databento_settings()
+          .bbo_start_time();
+      if(start != ""){
+         for (std::vector vec : all_sec) {
+            client.Subscribe(
+                vec, databento::Schema::Mbp1, databento::SType::RawSymbol, start);
+            client.Subscribe(
+                vec, databento::Schema::Trades, databento::SType::RawSymbol,start);
+         }   
+      }else{
+         for (std::vector vec : all_sec) {
+            client.Subscribe(
+                vec, databento::Schema::Mbp1, databento::SType::RawSymbol);
+            client.Subscribe(
+                vec, databento::Schema::Trades, databento::SType::RawSymbol);
+         } 
+      }
+      bool only_securities = config::get<bts::api::md::MarketDataConfig>()
+                         .cboe()
+                         .at(0)
+                         .databento_settings()
+                         .only_securities();
+      client.Start();
+      while (true) {
+         try {
+            auto *record = client.NextRecord(std::chrono::seconds{500});
+            if (first) {
+               LOG(md, Info, "first record");
+               first = false;
+            }
+            if (record == nullptr) {
+               LOG(md, Error, "Timed out waiting for data");
+               break;
+            }
+            if(only_securities)
+               continue;
+            if (record->Holds<databento::Mbp1Msg>()) {
+               process_bbo(*record);
+            }
+            if (all_times.size() == 1001) {
+               log_time_statistics(all_times);
+               client.Stop();
+            }
+            //  if (record->Holds<databento::InstrumentDefMsg>()) {
+            //      process_security(*record);
+            //  }
+            if (record->Holds<databento::TradeMsg>()) {
+               process_trades(*record);
+            }
+         } catch (const databento::TcpError &error) {
+            std::string err = error.what();
+            LOG(md, Error, "Caught TCP error: {} \nAttempting reconnect\n",
+                err);
+            client.Reconnect();
+         } catch (const databento::DbnResponseError &err) {
+            std::string msg = err.Message();
+            std::string what = err.what();
+            LOG(md, Error,
+                "Caught DbnResponseError what: {} ,msg:{} \nShutting down\n",
+                msg, what);
+            client.Stop();
+            // bts_assert(false);
+         } catch (const databento::Exception &excp) {
+            std::string err = excp.Message();
+            std::string err2 = excp.what();
+            LOG(md, Error,
+                "Caught unknown dadtabento::exception what: {}, msg: {} "
+                "\nShutting down\n",
+                err, err2);
+            client.Stop();
+            // bts_assert(false);
+         } catch (const std::exception &exc) {
+            std::string err = exc.what();
+            LOG(md, Error, "Caught unknown exception: {} \nShutting down\n",
+                err);
+            client.Stop();
+            // bts_assert(false);
+         }
+      }
    }
 }
 

tools/binance/binance.env

@@ -74,6 +74,16 @@ BINANCE_USDS_FUTURES_POSITION_INFO_PARAMS='symbol=${SYMBOL:-OMIT}&recvWindow=${R
 BINANCE_USDS_FUTURES_POSITION_INFO_METHOD="GET"
 BINANCE_USDS_FUTURES_POSITION_INFO_IS_SIGNED=1
 
+BINANCE_USDS_FUTURES_LISTEN_KEY_PATH="/fapi/v1/listenKey"
+BINANCE_USDS_FUTURES_LISTEN_KEY_PARAMS=''
+BINANCE_USDS_FUTURES_LISTEN_KEY_METHOD="POST"
+BINANCE_USDS_FUTURES_LISTEN_KEY_IS_KEYED=1
+
+BINANCE_USDS_FUTURES_REFRESH_LISTEN_KEY_PATH="/fapi/v1/listenKey"
+BINANCE_USDS_FUTURES_REFRESH_LISTEN_KEY_PARAMS='listenKey=${LISTEN_KEY:-OMIT}'
+BINANCE_USDS_FUTURES_REFRESH_LISTEN_KEY_METHOD="PUT"
+BINANCE_USDS_FUTURES_REFRESH_LISTEN_KEY_IS_KEYED=1
+
 BINANCE_COIN_FUTURES_PROD_URL="https://dapi.binance.com"
 BINANCE_COIN_FUTURES_DEV_URL="https://testnet.binancefuture.com"
 BINANCE_COIN_FUTURES_EXCHANGE_INFO_PATH="/dapi/v1/exchangeInfo"
@@ -116,3 +126,57 @@ BINANCE_CANCEL_REPLACE_PATH="/api/v3/order/cancelReplace"
 BINANCE_CANCEL_REPLACE_PARAMS='symbol=${SYMBOL}&side=${SIDE:-BUY}&type=${TYPE:-LIMIT}&timeInForce=${TIME_IN_FORCE:-GTC}&quantity=${QUANTITY:-OMIT}&price=${PRICE:-OMIT}&cancelReplaceMode=${CANCEL_REPLACE_MODE}&cancelNewClientOrderId=${CANCEL_NEW_CLIENT_ORDER_ID:-OMIT}&cancelOrigClientOrderId=${CANCEL_ORIG_CLIENT_ORDER_ID:-OMIT}&cancelOrderId=${CANCEL_ORDER_ID:-OMIT}&newClientOrderId=${NEW_CLIENT_ORDER_ID:-OMIT}&recvWindow=${RECV_WINDOW:-6000}'
 BINANCE_CANCEL_REPLACE_METHOD="POST"
 BINANCE_CANCEL_REPLACE_IS_SIGNED=1
+
+BINANCE_LISTEN_KEY_PATH="/api/v3/userDataStream"
+BINANCE_LISTEN_KEY_PARAMS=''
+BINANCE_LISTEN_KEY_METHOD="POST"
+BINANCE_LISTEN_KEY_IS_KEYED=1
+
+BINANCE_REFRESH_LISTEN_KEY_PATH="/api/v3/userDataStream"
+BINANCE_REFRESH_LISTEN_KEY_PARAMS='listenKey=${LISTEN_KEY}'
+BINANCE_REFRESH_LISTEN_KEY_METHOD="PUT"
+BINANCE_REFRESH_LISTEN_KEY_IS_KEYED=1
+
+BINANCE_CROSS_MARGIN_COLLATERAL_RATIO_PATH="/sapi/v1/margin/crossMarginCollateralRatio"
+BINANCE_CROSS_MARGIN_COLLATERAL_RATIO_METHOD="GET"
+BINANCE_CROSS_MARGIN_COLLATERAL_RATIO_IS_SIGNED=1
+
+BINANCE_CROSS_MARGIN_PAIRS_PATH="/sapi/v1/margin/allPairs"
+BINANCE_CROSS_MARGIN_PAIRS_PARAMS='symbol=${SYMBOL:-OMIT}'
+BINANCE_CROSS_MARGIN_PAIRS_METHOD="GET"
+BINANCE_CROSS_MARGIN_PAIRS_IS_SIGNED=1
+
+BINANCE_CROSS_MARGIN_ORDER_PATH="/sapi/v1/margin/order"
+BINANCE_CROSS_MARGIN_ORDER_PARAMS='symbol=${SYMBOL}&side=${SIDE}&type=${TYPE}&quantity=${QUANTITY}&price=${PRICE}&timeInForce=GTC&autoRepayAtCancel=true&sideEffectType=AUTO_BORROW_REPAY&recvWindow=${RECV_WINDOW:-6000}'
+BINANCE_CROSS_MARGIN_ORDER_METHOD="POST"
+BINANCE_CROSS_MARGIN_ORDER_IS_SIGNED=1
+
+BINANCE_CROSS_MARGIN_OPEN_ORDERS_PATH="/sapi/v1/margin/openOrders"
+BINANCE_CROSS_MARGIN_OPEN_ORDERS_PARAMS='symbol=${SYMBOL}&isIsolated=${IS_ISOLATED:-OMIT}&recvWindow=${RECV_WINDOW:-6000}'
+BINANCE_CROSS_MARGIN_OPEN_ORDERS_METHOD="GET"
+BINANCE_CROSS_MARGIN_OPEN_ORDERS_IS_SIGNED=1
+
+BINANCE_CROSS_MARGIN_CANCEL_PATH="/sapi/v1/margin/order"
+BINANCE_CROSS_MARGIN_CANCEL_PARAMS='symbol=${SYMBOL}&isIsolated=${IS_ISOLATED:-OMIT}&orderId=${ORDER_ID:-OMIT}&origClientOrderId=${ORIG_CLIENT_ORDER_ID:-OMIT}&newClientOrderId=${NEW_CLIENT_ORDER_ID:-OMIT}&recvWindow=${RECV_WINDOW:-6000}'
+BINANCE_CROSS_MARGIN_CANCEL_METHOD="DELETE"
+BINANCE_CROSS_MARGIN_CANCEL_IS_SIGNED=1
+
+BINANCE_CROSS_MARGIN_LISTEN_KEY_PATH="/sapi/v1/userDataStream"
+BINANCE_CROSS_MARGIN_LISTEN_KEY_PARAMS=''
+BINANCE_CROSS_MARGIN_LISTEN_KEY_METHOD="POST"
+BINANCE_CROSS_MARGIN_LISTEN_KEY_IS_KEYED=1
+
+BINANCE_CROSS_MARGIN_REFRESH_LISTEN_KEY_PATH="/sapi/v1/userDataStream"
+BINANCE_CROSS_MARGIN_REFRESH_LISTEN_KEY_PARAMS='listenKey=${LISTEN_KEY}'
+BINANCE_CROSS_MARGIN_REFRESH_LISTEN_KEY_METHOD="PUT"
+BINANCE_CROSS_MARGIN_REFRESH_LISTEN_KEY_IS_KEYED=1
+
+BINANCE_ISOLATED_MARGIN_SYMBOL_PATH="/sapi/v1/margin/order"
+BINANCE_ISOLATED_MARGIN_SYMBOL_PARAMS='symbol=${SYMBOL}&side=${SIDE}&type=${TYPE}&quantity=${QUANTITY}&price=${PRICE}&autoRepayAtCancel=true&recvWindow=${RECV_WINDOW:-6000}'
+BINANCE_ISOLATED_MARGIN_SYMBOL_METHOD="GET"
+BINANCE_ISOLATED_MARGIN_SYMBOL_IS_SIGNED=1
+
+BINANCE_ALL_MARGIN_ASSETS_PATH="/sapi/v1/margin/allAssets"
+BINANCE_ALL_MARGIN_ASSETS_PARAMS='asset=${ASSET:-OMIT}'
+BINANCE_ALL_MARGIN_ASSETS_METHOD="GET"
+BINANCE_ALL_MARGIN_ASSETS_IS_SIGNED=1

tools/binance/binance_orders_client.sh

@@ -133,6 +133,8 @@ method_name="${venue}_${op}_METHOD"
 method="${!method_name:-GET}"
 is_signed_name="${venue}_${op}_IS_SIGNED"
 is_signed="${!is_signed_name:-0}"
+is_keyed_name="${venue}_${op}_IS_KEYED"
+is_keyed="${!is_keyed_name:-0}"
 
 # Notes on query params
 # =============================================================================
@@ -168,13 +170,21 @@ params=$(echo "$params" | sed -E 's/(&)?[^&=]+=OMIT//g' | sed 's/^&//;s/&$//')
 # Sign the request (if necessary)
 declare -A headers
 if [ "$is_signed" -eq 1 ]; then
+  if [ -n "$params" ]; then
+    params="$params&"
+  fi
   # Signed requests require a timestamp, and api key header
-  params="$params&timestamp=$(date +%s%3N)"
+  params="${params}timestamp=$(date +%s%3N)"
   signature="$(echo -n "$params" | openssl dgst -sha256 -hmac "$secret_key" | awk '{print $NF}')"
   params="$params&signature=$signature"
   headers["X-MBX-APIKEY"]="$api_key"
+elif [ "$is_keyed" -eq 1 ]; then
+  headers["X-MBX-APIKEY"]="$api_key"
+fi
+url="${!base_url_name}${!path_name}"
+if [ -n "$params" ]; then
+  url="$url?$params"
 fi
-url="${!base_url_name}${!path_name}?${params}"
 
 headers_str=""
 for key in "${!headers[@]}"; do