fix: Disable ccapi md provider if venue has empty symbols config

Previously, ccapi exchanges (e.g., binance) would subscribe to EVERYTHING if
symbols config was empty. There are way too many symbols for that.

src/app_sparkd/configurations/bybit_dev/py_bybit_md_provider.json

@@ -6,7 +6,9 @@
     "ADA",
     "BNB",
     "BTC",
+    "DUCK",
     "ETH",
+    "MILK",
     "PRIME",
     "XRP"
   ],

src/app_sparkd/configurations/bybit_dev/symbols.conf

@@ -5,7 +5,9 @@ SymbolLists {
       symbol: "ADA"
       symbol: "BNB"
       symbol: "BTC"
+      symbol: "DUCK"
       symbol: "ETH"    
+      symbol: "MILK"
       symbol: "PRIME"
       symbol: "XRP"
     }

src/app_sparkd/sparkd/execution/py_execution_gateway_adapter.cpp

@@ -362,6 +362,7 @@ struct PyExecutionGatewayAdapter::PImpl {
       bts_assert(tx_id.size() < er.exchange_fill_id.value().max_size());
       bts_assert(!py_er.transact_id().empty());
       er.exchange_fill_id->copy_from(py_er.transact_id());
+      er.msg = py_er.error_msg();
    }
 
    void to_py(
@@ -442,6 +443,8 @@ struct PyExecutionGatewayAdapter::PImpl {
       py_req.set_key(msg.key.value());
       py_req.set_exchange_security_id(msg.security_name());
       py_req.set_venue(bts::api::venue_name_pretty(msg.venue()));
+      py_req.set_security_type(
+          bts::api::security_type_name_pretty(msg.security_type()));
       py_req.set_is_buy(msg.side == bts::api::SIDE_BUY);
       py_req.set_price(msg.price.value());
       py_req.set_quantity(msg.quantity.full_value_float());
@@ -460,6 +463,8 @@ struct PyExecutionGatewayAdapter::PImpl {
           std::to_string(msg.exchange_order_id.value()));
       py_req.set_exchange_security_id(cf.security_name());
       py_req.set_venue(bts::api::venue_name_pretty(cf.venue()));
+      py_req.set_security_type(
+          bts::api::security_type_name_pretty(cf.security_type()));
       py_req.set_is_buy(cf.side == bts::api::SIDE_BUY);
       py_req.set_price(cf.price.value());
       py_req.set_quantity(cf.quantity.full_value_float());
@@ -471,6 +476,8 @@ struct PyExecutionGatewayAdapter::PImpl {
           std::to_string(msg.exchange_order_id.value()));
       py_req.set_exchange_security_id(msg.security_name());
       py_req.set_venue(bts::api::venue_name_pretty(msg.venue()));
+      py_req.set_security_type(
+          bts::api::security_type_name_pretty(msg.security_type()));
       py_req.set_is_buy(msg.side == bts::api::SIDE_BUY);
    }
 

src/app_sparkd/sparkd/md/PyMarketDataSubscription.cpp

@@ -336,6 +336,10 @@ struct PyMarketDataSubscription::PImpl {
          sec_comm.min_quantity =
              std::max(qty_info.min_value(), sec_comm.min_quantity_increment);
       }
+      if (spec.notional_quantity_infos().size() > 0) {
+         const auto &notional_qty_info = spec.notional_quantity_infos()[0];
+         sec_comm.min_notional_value = notional_qty_info.min_value();
+      }
       sec_comm.desc = py_sec.name();
       if (py_sec.is_spot()) {
          sec.security = SpotSecurity{};

src/app_sparkd/sparkd/md/ccapi/CCApiMarketDataProvider.cpp

@@ -7,6 +7,8 @@
 #include "spark/md/GeneralMarketDataListener.h"
 #include "spark/md/GeneralSecurityIndex.h"
 #include "spark/md/MDProducer.h"
+#include "spark/md/SymbolLoader.h"
+#include "spark/md/UserSymbol.h"
 #include "spark/persistence/IDGenerator.h"
 #include "sparkd/md/binance/BinanceSubscription.h"
 #include "sparkd/md/binance_us/BinanceUsSubscription.h"
@@ -82,9 +84,23 @@ CCApiMarketDataProvider::get_sources() {
           "CCApiMarketDataProvider::get_sources failed. Invalid config: {}",
           fmt::to_string(fmt::join(cfg_errs, "\n")));
    }
-
+   std::set<UserSymbol> syms;
+   auto is_venue_enabled = [&syms](api::domain::Venue venue) {
+      syms.clear();
+      SymbolLoader::load(syms, domain::SecurityCriteria::create({venue}));
+      if (syms.empty()) {
+         LOG(md, Warn,
+             R"([ccapi_md_pro_disabled.json]{{ "venue": "{}", "rsn": "No symbols config" }})",
+             venue);
+         return false;
+      }
+      return true;
+   };
    // Binance
    for (const auto &binance : cfg.binance()) {
+      if (!is_venue_enabled(api::domain::VENUE_BINANCE)) {
+         continue;
+      }
       me_->sources.emplace_back(
           api::domain::VENUE_BINANCE,
           std::make_shared<binance::BinanceSubscription>(
@@ -92,6 +108,9 @@ CCApiMarketDataProvider::get_sources() {
    }
    // Binance US
    for (const auto &binance_us : cfg.binance_us()) {
+      if (!is_venue_enabled(api::domain::VENUE_BINANCE_US)) {
+         continue;
+      }
       me_->sources.emplace_back(
           api::domain::VENUE_BINANCE_US,
           std::make_shared<binance_us::BinanceUsSubscription>(
@@ -99,6 +118,9 @@ CCApiMarketDataProvider::get_sources() {
    }
    // Binance USDS Futures
    for (const auto &binance_ufs : cfg.binance_usds_futures()) {
+      if (!is_venue_enabled(api::domain::VENUE_BINANCE_USDS_FUTURES)) {
+         continue;
+      }
       me_->sources.emplace_back(
           api::domain::VENUE_BINANCE_USDS_FUTURES,
           std::make_shared<
@@ -107,6 +129,9 @@ CCApiMarketDataProvider::get_sources() {
    }
    // Coinbase
    for (const auto &coinbase : cfg.coinbase()) {
+      if (!is_venue_enabled(api::domain::VENUE_COINBASE)) {
+         continue;
+      }
       me_->sources.emplace_back(
           api::domain::VENUE_COINBASE,
           std::make_shared<coinbase::CoinbaseSubscription>(
@@ -114,6 +139,9 @@ CCApiMarketDataProvider::get_sources() {
    }
    // Deribit
    for (const auto &deribit : cfg.deribit()) {
+      if (!is_venue_enabled(api::domain::VENUE_DERIBIT)) {
+         continue;
+      }
       me_->sources.emplace_back(
           api::domain::VENUE_DERIBIT,
           std::make_shared<deribit::DeribitSubscription>(
@@ -121,6 +149,9 @@ CCApiMarketDataProvider::get_sources() {
    }
    // FTX
    for (const auto &ftx : cfg.ftx()) {
+      if (!is_venue_enabled(api::domain::VENUE_FTX)) {
+         continue;
+      }
       me_->sources.emplace_back(
           api::domain::VENUE_FTX,
           std::make_shared<ftx::FTXSubscription>(
@@ -128,6 +159,9 @@ CCApiMarketDataProvider::get_sources() {
    }
    // Okx
    for (const auto &okx : cfg.okx()) {
+      if (!is_venue_enabled(api::domain::VENUE_OKX)) {
+         continue;
+      }
       me_->sources.emplace_back(
           api::domain::VENUE_OKX,
           std::make_shared<okx::OkxSubscription>(

src/client/spark/Daemons/MDDaemon.cs

@@ -110,8 +110,8 @@ namespace bts {
             ConcurrentDictionary<SecurityKey, SecurityMD> exch_secs,
             Dictionary<SecurityKey, ExecutionStrategyImmutable> synth_strategies
          ) {
-            synth_strategies.Where(kv => kv.Value.Deleted).difference_hard(
-               last_synth_strategies.Where(kv => kv.Value.Deleted),
+            last_synth_strategies.Where(kv => kv.Value.Deleted).difference_hard(
+               synth_strategies.Where(kv => kv.Value.Deleted),
                out var deleted_synths,
                out var undeleted_synths,
                synth_id_comparer);

src/client/spark/Modules/Market/MarketView/MarketViewColumnType.cs

@@ -40,5 +40,7 @@ namespace mbts.spark {
       Rho,
       Currency,
       CashPerPoint,
+      MinOrderQuantity,
+      MinOrderNotional
    }
 }

src/client/spark/Modules/Market/MarketView/MarketViewFilterStates.cs

@@ -130,6 +130,10 @@ namespace bts.shell.market_view {
       public DoubleMinMaxState rho { get; set; }
       [ProtoMember(39)]
       public DoubleMinMaxState cash_per_point { get; set; }
+      [ProtoMember(40)]
+      public DoubleMinMaxState min_order_quantity { get; set; }
+      [ProtoMember(41)]
+      public DoubleMinMaxState min_order_notional { get; set; }
 
       public Func<MarketViewVM.Row, bool> build_predicate() {
          var pb = new ColumnPredicateBuilder<MarketViewVM.Row>();
@@ -188,6 +192,8 @@ namespace bts.shell.market_view {
          pb.add_numeric(theta, r => r.theta);
          pb.add_numeric(rho, r => r.rho);
          pb.add_numeric(cash_per_point, r => r.cash_per_point);
+         pb.add_numeric(min_order_quantity, r => r.min_order_quantity);
+         pb.add_numeric(min_order_notional, r => r.min_order_notional);
          return pb.to_predicate();
       }
    }

src/client/spark/Modules/Market/MarketView/MarketViewRow.cs

@@ -210,6 +210,12 @@ namespace mbts.spark {
          public uint security_key => security_md.security.Id;
          public string currency => security_md.security.Currency;
          public double cash_per_point => security_md.security.ContractSpec.CashPerPoint;
+         public double min_order_quantity => _spec.QuantityInfos.TryGetFirst(out var qi)
+            ? qi.TickUnit.safe_divide(qi.TickBase)
+            : security_md.security.tick_factors().tick_unit_frac();
+         public double min_order_notional => _spec.NotionalQuantityInfos.TryGetFirst(out var qi)
+            ? qi.MinValue
+            : security_md.security.tick_factors().tick_unit_frac();
 
          public bool inside_bids => _inside_bids.value;
          public bool inside_asks => _inside_asks.value;
@@ -345,6 +351,9 @@ namespace mbts.spark {
          TickDisplaySettingsImmutable _qty_display =>
             security_md.quantity_display;
 
+         ContractSpecificationImmutable _spec =>
+            security_md.security.spec_or_default();
+
          public string format_price(double price) =>
             security.ContractSpec.format(price);
 
@@ -448,9 +457,9 @@ namespace mbts.spark {
                   _bus.post(SparkTags.ShowOrderTicketTag, get_params(last_price, last_qty));
                   break;
             }
-            
+
             OrderTicketParams get_params(double price, double quantity = 0) =>
-               new(new(new() {SecurityKey = security_key }), price, quantity);
+               new(new(new() { SecurityKey = security_key }), price, quantity);
          }
 
          void _refresh_orders(ref bool dirty) {

src/client/spark/Modules/Market/MarketView/MarketViewSettings.cs

@@ -147,6 +147,8 @@ namespace bts.shell.market_view {
          public static GridColumnSettings theta => _mk(Columns.Theta, "Theta", ColWidth.Greeks, false);
          public static GridColumnSettings rho => _mk(Columns.Rho, "Rho", ColWidth.Greeks, false);
          public static GridColumnSettings cash_per_point => _mk(Columns.CashPerPoint, "CPP", ColWidth.Greeks, false);
+         public static GridColumnSettings min_order_quantity => _mk(Columns.MinOrderQuantity, "Min Order Qty", 60, false);
+         public static GridColumnSettings min_notional_value => _mk(Columns.MinOrderNotional, "Min Order Notional", 60, false);
 
          public static IEnumerable<SortingCriteria> default_sorting_criteria {
             get {

src/client/spark/Modules/Market/MarketView/MarketViewVM.cs

@@ -198,7 +198,9 @@ namespace mbts.spark {
          add_col(ColKey.Vega, r => fmt_greek(r.vega), nameof(Row.vega), vega_filter, FilterType.DoubleMinMax, ColWidth.Greeks);
          add_col(ColKey.Theta, r => fmt_greek(r.theta), nameof(Row.theta), theta_filter, FilterType.DoubleMinMax, ColWidth.Greeks);
          add_col(ColKey.Rho, r => fmt_greek(r.rho), nameof(Row.rho), rho_filter, FilterType.DoubleMinMax, ColWidth.Greeks);
-         add_col(ColKey.CashPerPoint, r => r.cash_per_point.ToString("0.00#####"), nameof(Row.cash_per_point), currency_filter, FilterType.DoubleMinMax, ColWidth.Greeks);
+         add_col(ColKey.CashPerPoint, r => r.cash_per_point.ToString("0.00#####"), nameof(Row.cash_per_point), cash_per_point_filter, FilterType.DoubleMinMax, ColWidth.Greeks);
+         add_col(ColKey.MinOrderQuantity, r => r.min_order_quantity.ToString("0.00#####"), nameof(Row.min_order_quantity), min_order_quantity_filter, FilterType.DoubleMinMax, ColWidth.Qty);
+         add_col(ColKey.MinOrderNotional, r => r.min_order_notional.ToString("0.00#####"), nameof(Row.min_order_notional), min_order_notional_filter, FilterType.DoubleMinMax, ColWidth.Qty);
 
          // row vm filter
          // Althoguh securities are filtered at the row 'model' level (SecurityMD), the column base filters have to be applied at the row vm level,
@@ -438,6 +440,8 @@ namespace mbts.spark {
       public IDataGridColumnPersistentFilter<Row, DoubleMinMaxState> theta_filter { get; } = new DoubleMinMax<Row>(r => r.theta, nameof(Row.theta));
       public IDataGridColumnPersistentFilter<Row, DoubleMinMaxState> rho_filter { get; } = new DoubleMinMax<Row>(r => r.rho, nameof(Row.rho));
       public IDataGridColumnPersistentFilter<Row, DoubleMinMaxState> cash_per_point_filter { get; } = new DoubleMinMax<Row>(r => r.cash_per_point, nameof(Row.cash_per_point));
+      public IDataGridColumnPersistentFilter<Row, DoubleMinMaxState> min_order_quantity_filter { get; } = new DoubleMinMax<Row>(r => r.min_order_quantity, nameof(Row.min_order_quantity));
+      public IDataGridColumnPersistentFilter<Row, DoubleMinMaxState> min_order_notional_filter { get; } = new DoubleMinMax<Row>(r => r.min_order_notional, nameof(Row.min_order_notional));
 
       IEnumerable<IDataGridColumnFilter<Row>> _all_filters =>
          _filter_getters.Select(g => g(this));
@@ -480,6 +484,8 @@ namespace mbts.spark {
             ColKey.Theta => wrap(r => r.theta),
             ColKey.Rho => wrap(r => r.rho),
             ColKey.CashPerPoint => wrap(r => r.cash_per_point),
+            ColKey.MinOrderQuantity => wrap(r => r.min_order_quantity),
+            ColKey.MinOrderNotional => wrap(r => r.min_order_notional),
             _ => null,
          };
 
@@ -587,6 +593,8 @@ namespace mbts.spark {
             theta = theta_filter.get_state(),
             rho = rho_filter.get_state(),
             cash_per_point = cash_per_point_filter.get_state(),
+            min_order_quantity = min_order_quantity_filter.get_state(),
+            min_order_notional = min_order_notional_filter.get_state(),
          };
 
       public void load_filters(MarketViewFilterStates filters) {

src/client/spark/Modules/Trading/Ladder/LadderSettings.cs

@@ -647,7 +647,7 @@ namespace mbts.spark {
                volume_column = VolumeColumnSettings.Default,
                calculator_buttons = new double[] { 1, 5, 10, 25, 50, 100 },
                scroll_bar_visible = false,
-               default_time_in_force = TimeInForce.Day,
+               default_time_in_force = TimeInForce.Gtc,
                default_algo = new AlgoID(CustomOrderDefinition.CustomOneofCase.Limit),
                animate_recenter = true,
                badge_style = BadgeStyle.Normal,

src/client/spark/Orders/OrderFactory.cs

@@ -129,7 +129,7 @@ namespace mbts.spark.utilities {
          };
 
       public TimeInForce default_time_in_force() =>
-         TimeInForce.Day;
+         TimeInForce.Gtc;
 
       public TimeInForce default_time_in_force(SecurityImmutable security) =>
          default_time_in_force();
@@ -167,7 +167,7 @@ namespace mbts.spark.utilities {
          var rv = new OrderDefinition {
             SecurityKey = security.Id,
             Side = Side.Buy,
-            TimeInForce = TimeInForce.Day,
+            TimeInForce = default_time_in_force(security),
             CustomDef = OrderDataHelper.Defaults.CustomOrderDefinition(),
             CtiCode = CTICode.CtiCode1,
             CustomerOrFirm = CustomerOrFirm.Firm,

src/client/spark_websocket/Core/WebsocketRestHandler.cs

@@ -424,7 +424,7 @@ namespace bts.websocket_core {
 
 
          return (true, new OrderDefinition {
-            TimeInForce = TimeInForce.Day,
+            TimeInForce = TimeInForce.Gtc,
             SecurityKey = sc_md.id,
             Side = (Side)side,
             UserId = user_id,

src/lib_python_interop/python_interop/execution/py_cancel_order_request.cpp

@@ -31,6 +31,13 @@ void PyCancelOrderRequest::set_venue(string value) {
    venue_ = std::move(value);
 }
 
+const string &PyCancelOrderRequest::security_type() const {
+   return security_type_;
+}
+void PyCancelOrderRequest::set_security_type(string value) {
+   security_type_ = std::move(value);
+}
+
 bool PyCancelOrderRequest::is_buy() const { return is_buy_; }
 void PyCancelOrderRequest::set_is_buy(bool value) {
    is_buy_ = std::move(value);
@@ -41,6 +48,7 @@ void to_json(nlohmann::json &j, const PyCancelOrderRequest &req) {
    j["eoid"] = req.exchange_order_id();
    j["esid"] = req.exchange_security_id();
    j["venue"] = req.venue();
+   j["styp"] = req.security_type();
    j["side"] = req.is_buy() ? 'B' : 'S';
 }
 
@@ -49,6 +57,7 @@ void from_json(const nlohmann::json &j, PyCancelOrderRequest &req) {
    j.at("eoid").get_to(req.exchange_order_id_);
    j.at("esid").get_to(req.exchange_security_id_);
    j.at("venue").get_to(req.venue_);
+   j.at("styp").get_to(req.security_type_);
    const auto str = j.at("side").get<string>();
    if (!str.empty()) {
       switch (str[0]) {

src/lib_python_interop/python_interop/execution/py_cancel_order_request.h

@@ -26,6 +26,9 @@ class PyCancelOrderRequest {
    const std::string &venue() const;
    void set_venue(std::string value);
 
+   const std::string &security_type() const;
+   void set_security_type(std::string value);
+
    bool is_buy() const;
    void set_is_buy(bool value);
 
@@ -34,6 +37,7 @@ class PyCancelOrderRequest {
        std::string exchange_order_id,
        std::string exchange_security_id,
        std::string venue,
+       std::string security_type,
        bool is_buy);
 
    std::string to_string() const;
@@ -43,6 +47,7 @@ class PyCancelOrderRequest {
    std::string exchange_order_id_;
    std::string exchange_security_id_;
    std::string venue_;
+   std::string security_type_;
    bool is_buy_;
 
    friend void from_json(const nlohmann::json &j, PyCancelOrderRequest &req);
@@ -73,11 +78,13 @@ inline void PyCancelOrderRequest::set_data(
     std::string exchange_order_id,
     std::string exchange_security_id,
     std::string venue,
+    std::string security_type,
     bool is_buy) {
    key_ = key;
    exchange_order_id_ = std::move(exchange_order_id);
    exchange_security_id_ = std::move(exchange_security_id);
    venue_ = std::move(venue);
+   security_type_ = std::move(security_type);
    is_buy_ = is_buy;
 }
 

src/lib_python_interop/python_interop/execution/py_cancel_replace_order_request.cpp

@@ -31,6 +31,13 @@ void PyCancelReplaceOrderRequest::set_venue(string value) {
    venue_ = std::move(value);
 }
 
+const string &PyCancelReplaceOrderRequest::security_type() const {
+   return security_type_;
+}
+void PyCancelReplaceOrderRequest::set_security_type(string value) {
+   security_type_ = std::move(value);
+}
+
 bool PyCancelReplaceOrderRequest::is_buy() const { return is_buy_; }
 void PyCancelReplaceOrderRequest::set_is_buy(bool value) {
    is_buy_ = std::move(value);
@@ -49,6 +56,7 @@ void to_json(nlohmann::json &j, const PyCancelReplaceOrderRequest &req) {
    j["eoid"] = req.exchange_order_id();
    j["esid"] = req.exchange_security_id();
    j["venue"] = req.venue();
+   j["styp"] = req.security_type();
    j["side"] = req.is_buy() ? 'B' : 'S';
    j["px"] = req.price();
    j["qty"] = req.quantity();
@@ -59,6 +67,7 @@ void from_json(const nlohmann::json &j, PyCancelReplaceOrderRequest &req) {
    j.at("eoid").get_to(req.exchange_order_id_);
    j.at("esid").get_to(req.exchange_security_id_);
    j.at("venue").get_to(req.venue_);
+   j.at("styp").get_to(req.security_type_);
    const auto str = j.at("side").get<string>();
    if (!str.empty()) {
       switch (str[0]) {

src/lib_python_interop/python_interop/execution/py_cancel_replace_order_request.h

@@ -27,6 +27,9 @@ class PyCancelReplaceOrderRequest {
    const std::string &venue() const;
    void set_venue(std::string value);
 
+   const std::string &security_type() const;
+   void set_security_type(std::string value);
+
    bool is_buy() const;
    void set_is_buy(bool value);
 
@@ -41,6 +44,7 @@ class PyCancelReplaceOrderRequest {
        std::string exchange_order_id,
        std::string exchange_security_id,
        std::string venue,
+       std::string security_type,
        bool is_buy,
        double price,
        double quantity);
@@ -52,6 +56,7 @@ class PyCancelReplaceOrderRequest {
    std::string exchange_order_id_;
    std::string exchange_security_id_;
    std::string venue_;
+   std::string security_type_;
    float price_;
    float quantity_;
    bool is_buy_;
@@ -85,6 +90,7 @@ inline void PyCancelReplaceOrderRequest::set_data(
     std::string exchange_order_id,
     std::string exchange_security_id,
     std::string venue,
+    std::string security_type,
     bool is_buy,
     double price,
     double quantity) {
@@ -92,6 +98,7 @@ inline void PyCancelReplaceOrderRequest::set_data(
    exchange_order_id_ = std::move(exchange_order_id);
    exchange_security_id_ = std::move(exchange_security_id);
    venue_ = std::move(venue);
+   security_type_ = std::move(security_type);
    is_buy_ = is_buy;
    price_ = price;
    quantity_ = quantity;

src/lib_python_interop/python_interop/execution/py_create_order_request.cpp

@@ -2,6 +2,7 @@
 #include "spark/api/services/native_order_type.h"
 #include "spark/api/services/time_in_force.h"
 #include "spark/domain_model/Security.pb.h"
+#include "spark/domain_model/SecurityType.h"
 #include "spark/domain_model/Venue.h"
 #include "spark/orderman/CreateOrderMsg.h"
 
@@ -27,6 +28,13 @@ void PyCreateOrderRequest::set_venue(string value) {
    venue_ = std::move(value);
 }
 
+const string &PyCreateOrderRequest::security_type() const {
+   return security_type_;
+}
+void PyCreateOrderRequest::set_security_type(string value) {
+   security_type_ = std::move(value);
+}
+
 float PyCreateOrderRequest::price() const { return price_; }
 void PyCreateOrderRequest::set_price(float value) { price_ = value; }
 
@@ -50,23 +58,11 @@ void PyCreateOrderRequest::set_is_buy(bool value) {
    is_buy_ = std::move(value);
 }
 
-void PyCreateOrderRequest::copy_from(const bts::orderman::CreateOrderMsg &msg) {
-   key_ = msg.key.value();
-   exchange_security_id_ = msg.security_name();
-   venue_ = bts::api::venue_name_pretty(msg.venue());
-   is_buy_ = msg.side == bts::api::SIDE_BUY;
-   price_ = msg.price.value();
-   quantity_ = msg.quantity.full_value_float();
-   order_type_ =
-       bts::api::orderman::native_order_type_name_pretty(msg.order_type);
-   time_in_force_ =
-       bts::api::orderman::time_in_force_name_pretty(msg.time_in_force);
-}
-
 void to_json(nlohmann::json &j, const PyCreateOrderRequest &req) {
    j["key"] = req.key();
    j["esid"] = req.exchange_security_id();
    j["venue"] = req.venue();
+   j["styp"] = req.security_type();
    j["side"] = req.is_buy() ? "B" : "S";
    j["px"] = req.price();
    j["qty"] = req.quantity();
@@ -78,6 +74,7 @@ void from_json(const nlohmann::json &j, PyCreateOrderRequest &req) {
    j.at("key").get_to(req.key_);
    j.at("eid").get_to(req.exchange_security_id_);
    j.at("venue").get_to(req.venue_);
+   j.at("styp").get_to(req.security_type_);
    const auto str = j.at("side").get<string>();
    if (!str.empty()) {
       switch (str[0]) {

src/lib_python_interop/python_interop/execution/py_create_order_request.h

@@ -24,6 +24,9 @@ class PyCreateOrderRequest {
    const std::string &venue() const;
    void set_venue(std::string value);
 
+   const std::string &security_type() const;
+   void set_security_type(std::string value);
+
    bool is_buy() const;
    void set_is_buy(bool value);
 
@@ -43,13 +46,13 @@ class PyCreateOrderRequest {
        std::uint32_t key,
        std::string exchange_security_id,
        std::string venue,
+       std::string security_type,
        bool is_buy,
        double price,
        double quantity,
        std::string order_type,
        std::string time_in_force);
 
-   void copy_from(const bts::orderman::CreateOrderMsg &msg);
    std::string to_string() const;
 
  private:
@@ -58,6 +61,7 @@ class PyCreateOrderRequest {
    float quantity_;
    std::string exchange_security_id_;
    std::string venue_;
+   std::string security_type_;
    std::string order_type_;
    std::string time_in_force_;
    bool is_buy_;
@@ -89,6 +93,7 @@ inline void PyCreateOrderRequest::set_data(
     std::uint32_t key,
     std::string exchange_security_id,
     std::string venue,
+    std::string security_type,
     bool is_buy,
     double price,
     double quantity,
@@ -97,6 +102,7 @@ inline void PyCreateOrderRequest::set_data(
    key_ = key;
    exchange_security_id_ = std::move(exchange_security_id);
    venue_ = std::move(venue);
+   security_type_ = std::move(security_type);
    is_buy_ = is_buy;
    price_ = price;
    quantity_ = quantity;

src/lib_python_interop/python_interop/execution/py_execution_report.h

@@ -74,6 +74,9 @@ class PyExecutionReport {
    std::int64_t transact_time() const;
    void set_transact_time(std::int64_t value);
 
+   const std::string &error_msg() const;
+   void set_error_msg(std::string value);
+
    /**
     * Sets all data at once (more performant than setting props one at a time
     * when calling from python).
@@ -127,6 +130,7 @@ class PyExecutionReport {
    double avg_filled_price_;
    std::string transact_id_;
    std::int64_t transact_time_;
+   std::string error_msg_;
 
    friend void from_json(const nlohmann::json &j, PyExecutionReport &req);
 };
@@ -263,6 +267,13 @@ inline void PyExecutionReport::set_transact_time(std::int64_t value) {
    transact_time_ = value;
 }
 
+inline const std::string &PyExecutionReport::error_msg() const {
+   return error_msg_;
+}
+inline void PyExecutionReport::set_error_msg(std::string value) {
+   error_msg_ = std::move(value);
+}
+
 inline std::string PyExecutionReport::to_string() const {
    return fmt::format("{}", *this);
 }

src/lib_python_interop/python_interop/md/py_book.h

@@ -145,6 +145,9 @@ inline std::string PyBook::to_string() const {
 }
 
 inline void to_json(nlohmann::json &j, const PyBook &bk) {
+   j["sec"] = bk.exchange_security_id();
+   j["venue"] = bk.venue();
+   j["typ"] = bk.security_type();
    j["bids"] = bk.bids();
    j["asks"] = bk.asks();
 }

src/lib_python_interop/python_interop/plugin_manager.cpp

@@ -99,8 +99,8 @@ struct __attribute__((visibility("hidden"))) PluginManager_Impl {
          session_data = &sd;
          mx.unlock();
          try {
-            sd.import("sys").attr("path").attr("append")(
-                "python_interop_plugins");
+            sd.import("sys").attr("path").attr("insert")(
+                0, "python_interop_plugins");
             sd.import("spark_core").attr("configure_default_logger")();
             sd.import("bybit_md_provider");
             sd.import("bybit_execution_gateway");

src/lib_python_interop/python_interop/python_interop_module.cpp

@@ -63,6 +63,24 @@ namespace py = pybind11;
    def(_STR(name), &type::name)                                                \
        .def(_STR(set_##name), &type::set_##name, py::arg(#name))
 
+#define PY_LIST_TO_STRING(type)                                                \
+   def("__str__", [](const std::vector<shared_ptr<type>> &py_list) {           \
+      std::ostringstream oss;                                                  \
+      oss << "[";                                                              \
+      for (size_t i = 0; i < py_list.size(); ++i) {                            \
+         if (i > 0) {                                                          \
+            oss << ", ";                                                       \
+         }                                                                     \
+         if (py_list[i]) {                                                     \
+            oss << py_list[i]->to_string();                                    \
+         } else {                                                              \
+            oss << "None";                                                     \
+         }                                                                     \
+      }                                                                        \
+      oss << "]";                                                              \
+      return oss.str();                                                        \
+   })
+
 // Define python module with c++ classes.
 PYBIND11_MODULE(python_interop_module, m) {
    // price info
@@ -75,7 +93,8 @@ PYBIND11_MODULE(python_interop_module, m) {
        .PROP(PyContractPriceInfo, max_price)
        .def("__str__", &PyContractPriceInfo::to_string);
    py::bind_vector<std::vector<PyContractPriceInfo>>(
-       m, "PyContractPriceInfoList");
+       m, "PyContractPriceInfoList")
+       .PY_LIST_TO_STRING(PyContractPriceInfo);
    // quantity info
    py::class_<PyContractQuantityInfo>(m, "PyContractQuantityInfo")
        .def(py::init<>())
@@ -87,7 +106,8 @@ PYBIND11_MODULE(python_interop_module, m) {
        .PROP(PyContractQuantityInfo, max_price)
        .def("__str__", &PyContractQuantityInfo::to_string);
    py::bind_vector<std::vector<PyContractQuantityInfo>>(
-       m, "PyContractQuantityInfoList");
+       m, "PyContractQuantityInfoList")
+       .PY_LIST_TO_STRING(PyContractQuantityInfo);
    // contract spec
    py::class_<PyContractSpecification>(m, "PyContractSpecification")
        .def(py::init<>())
@@ -124,7 +144,8 @@ PYBIND11_MODULE(python_interop_module, m) {
        .PROP(PySecurity, option)
        .def("__str__", &PySecurity::to_string)
        .def("__format__", &PySecurity::format);
-   py::bind_vector<std::vector<shared_ptr<PySecurity>>>(m, "PySecurityList");
+   py::bind_vector<std::vector<shared_ptr<PySecurity>>>(m, "PySecurityList")
+       .PY_LIST_TO_STRING(PySecurity);
 
    //////////////////////////////////////////////////
    // Md
@@ -158,7 +179,8 @@ PYBIND11_MODULE(python_interop_module, m) {
            py::arg("asks"))
        .def("__len__", &PyBook::size)
        .def("__str__", &PyBook::to_string);
-   py::bind_vector<std::vector<shared_ptr<PyBook>>>(m, "PyBookList");
+   py::bind_vector<std::vector<shared_ptr<PyBook>>>(m, "PyBookList")
+       .PY_LIST_TO_STRING(PyBook);
    // trade
    py::class_<PyTrade, shared_ptr<PyTrade>>(m, "PyTrade")
        .def(py::init<>())
@@ -169,7 +191,8 @@ PYBIND11_MODULE(python_interop_module, m) {
        .PROP(PyTrade, quantity)
        .PROP(PyTrade, aggressor_side)
        .PROP(PyTrade, timestamp);
-   py::bind_vector<std::vector<shared_ptr<PyTrade>>>(m, "PyTradeList");
+   py::bind_vector<std::vector<shared_ptr<PyTrade>>>(m, "PyTradeList")
+       .PY_LIST_TO_STRING(PyTrade);
    // md bridge
    py::class_<MdBridge>(m, "MdBridge")
        .DEF_ARGS(MdBridge, on_securities, py::arg("securities"))
@@ -190,6 +213,7 @@ PYBIND11_MODULE(python_interop_module, m) {
        .PROP(PyCreateOrderRequest, key)
        .PROP(PyCreateOrderRequest, exchange_security_id)
        .PROP(PyCreateOrderRequest, venue)
+       .PROP(PyCreateOrderRequest, security_type)
        .PROP(PyCreateOrderRequest, is_buy)
        .PROP(PyCreateOrderRequest, price)
        .PROP(PyCreateOrderRequest, quantity)
@@ -203,6 +227,7 @@ PYBIND11_MODULE(python_interop_module, m) {
        .PROP(PyCancelOrderRequest, exchange_order_id)
        .PROP(PyCancelOrderRequest, exchange_security_id)
        .PROP(PyCancelOrderRequest, venue)
+       .PROP(PyCancelOrderRequest, security_type)
        .PROP(PyCancelOrderRequest, is_buy)
        .def("__str__", &PyCancelOrderRequest::to_string);
    py::class_<
@@ -213,6 +238,7 @@ PYBIND11_MODULE(python_interop_module, m) {
        .PROP(PyCancelReplaceOrderRequest, exchange_order_id)
        .PROP(PyCancelReplaceOrderRequest, exchange_security_id)
        .PROP(PyCancelReplaceOrderRequest, venue)
+       .PROP(PyCancelReplaceOrderRequest, security_type)
        .PROP(PyCancelReplaceOrderRequest, is_buy)
        .PROP(PyCancelReplaceOrderRequest, price)
        .PROP(PyCancelReplaceOrderRequest, quantity)
@@ -238,6 +264,7 @@ PYBIND11_MODULE(python_interop_module, m) {
        .PROP(PyExecutionReport, avg_filled_price)
        .PROP(PyExecutionReport, order_type)
        .PROP(PyExecutionReport, transact_time)
+       .PROP(PyExecutionReport, error_msg)
        .def(
            "set_data", &PyExecutionReport::set_data, py::arg("key"),
            py::arg("account"), py::arg("exchange_order_id"),

src/lib_python_interop/python_interop_plugins/bybit_execution_gateway.py

@@ -2,7 +2,7 @@ import json
 import time
 from typing import Optional
 import uuid
-from pybit.unified_trading import WebSocket, WebSocketTrading
+from pybit_unified_trading import WebSocket, WebSocketTrading
 from spark_core import (
     JsonElement,
     as_bool,
@@ -24,14 +24,32 @@ from python_interop_module import (
     PySecurity,
     PySecurityList,
 )
-from spark_enums import OrderStatus, ExecType
+from spark_enums import OrderStatus, ExecType, OrderType, SecurityType, TimeInForce
 import threading
 
 
 class OrderData:
-    def __init__(self, bts_id: int, exch_id: str, price: float, quantity: float):
+    def __init__(
+        self,
+        account: str,
+        bts_id: int,
+        exch_id: str,
+        exch_sec_id: str,
+        sec_type: SecurityType,
+        is_buy: bool,
+        order_type: OrderType,
+        time_in_force: TimeInForce,
+        price: float,
+        quantity: float,
+    ):
+        self.account = account
         self.bts_id = bts_id
         self.exch_id = exch_id
+        self.exch_sec_id = exch_sec_id
+        self.sec_type = sec_type
+        self.is_buy = is_buy
+        self.order_type = order_type
+        self.time_in_force = time_in_force
         self.price = price
         self.quantity = quantity
         self.filled_quantity: float = 0
@@ -41,6 +59,53 @@ class OrderData:
         self.exec_type: ExecType = ExecType.PENDING_NEW
         self.status: OrderStatus = OrderStatus.PENDING_NEW
 
+    def to_json(self):
+        return {
+            "account": self.account,
+            "bts_id": self.bts_id,
+            "exch_id": self.exch_id,
+            "exch_sec_id": self.exch_sec_id,
+            "sec_type": self.sec_type.name,
+            "is_buy": self.is_buy,
+            "order_type": self.order_type.name,
+            "time_in_force": self.time_in_force.name,
+            "price": self.price,
+            "quantity": self.quantity,
+            "filled_quantity": self.filled_quantity,
+            "last_filled_quantity": self.last_filled_quantity,
+            "avg_filled_price": self.avg_filled_price,
+            "last_filled_price": self.last_filled_price,
+            "exec_type": self.exec_type.name,
+            "status": self.status.name,
+        }
+
+    def to_execution_report(self):
+        transact_time = time.time_ns()
+        er = PyExecutionReport()
+        er.set_data(
+            key=self.bts_id,
+            exchange_order_id=self.exch_id,
+            venue="BYBIT",
+            security_type=self.sec_type.name,
+            exchange_security_id=self.exch_sec_id,
+            account=self.account,
+            order_type=self.order_type.name,
+            is_buy=self.is_buy,
+            price=self.price,
+            quantity=self.quantity,
+            time_in_force=self.time_in_force.name,
+            exec_type=self.exec_type.name,
+            order_status=self.status.name,
+            last_filled_price=self.last_filled_price,
+            last_filled_quantity=self.last_filled_quantity,
+            filled_quantity=self.filled_quantity,
+            avg_filled_price=self.avg_filled_price,
+            # There is no transact_id in Bybit, so we use updatedTime
+            transact_id=str(transact_time),
+            transact_time=transact_time,
+        )
+        return er
+
     def on_update(
         self,
         price: float,
@@ -83,6 +148,7 @@ class OrderData:
                 self.status = OrderStatus.REJECTED
             case _:
                 raise ValueError(f"Unexpected order status '{status}'")
+        return self.to_execution_report()
 
 
 class BybitExecutionGateway(ExecutionGateway):
@@ -113,10 +179,53 @@ class BybitExecutionGateway(ExecutionGateway):
                 return v
         return None
 
-    def _on_order_ack(self, typ, msg: object):
-        self.logger.info(
-            '[bybit_gw_ord_ack.json]{ "typ": "%s", "msg": %s}' % (typ, msg)
-        )
+    def _order_handlers(self, typ: str, exch_id: str):
+        def get_error_desc(msg: dict):
+            return "%s (code %s)" % (
+                (msg.get("retMsg") or "Unknown", msg.get("retCode") or "Unknown")
+            )
+
+        def on_err(msg: dict):
+            self.logger.error(
+                '[bybit_gw_ord_ack_err.json]{ "typ": "%s", "exch_id": "%s", "err": %s, "known_orders": %s }'
+                % (
+                    typ,
+                    exch_id,
+                    JsonElement(msg),
+                    JsonElement({k: v.to_json() for k, v in self.orders.items()}),
+                )
+            )
+            req_id = msg.get("reqId")
+            if not req_id:
+                self.logger.error(
+                    '[bybit_gw_ord_ack_err.json]{ "typ": "%s", "exch_id": "%s", "err": "%s" }'
+                    % (typ, exch_id, msg)
+                )
+            ord_data = self._get_order_data(exch_id)
+            if ord_data:
+                er = ord_data.to_execution_report()
+                match er.order_status():
+                    case OrderStatus.PENDING_NEW.name:
+                        er.set_order_status(OrderStatus.REJECTED.name)
+                        er.set_exec_type(ExecType.REJECTED.name)
+                        er.set_error_msg(get_error_desc(msg))
+                        self.bridge.on_execution_report(er)
+                    case _:
+                        pass
+            else:
+                self.logger.error(
+                    '[bybit_gw_ord_ack_err.json]{"typ": "%s", "exch_id": "%s", "err": "Unknown exch id" }'
+                )
+
+        def on_ack(msg: dict):
+            ord_data = self._get_order_data(exch_id)
+            if ord_data is None:
+                self.logger.error(
+                    '[bybit_gw_ord_ack_err.json]{ "typ": "%s", "exch_id": "%s", "err": "No order data found with exch id", "msg": "%s" }'
+                    % (typ, exch_id, msg)
+                )
+
+        return (on_ack, on_err)
 
     def _on_abort(
         self,
@@ -194,6 +303,7 @@ class BybitExecutionGateway(ExecutionGateway):
             "api_secret": self.secret_key,
             "testnet": self.use_testnet,
             "channel_type": "private",
+            "private_auth_expire": 5,
             "trace_logging": False,
         }
         self.logger.info("[bybit_gw_stream_sub.json]%s" % JsonElement(params))
@@ -205,12 +315,10 @@ class BybitExecutionGateway(ExecutionGateway):
             "api_key": self.api_key,
             "api_secret": self.secret_key,
             "testnet": self.use_testnet,
+            "private_auth_expire": 5,
         }
         self.logger.info("[bybit_gw_start_trd_cli.json]%s" % JsonElement(params))
         self.trade_cli = WebSocketTrading(**params)
-        params = {
-            "category": "linear",
-        }
         self.bridge.on_is_ready(True)
         # Sleep loop to keep thread alive and detect connection status changes
         conn_statuses = {
@@ -225,12 +333,12 @@ class BybitExecutionGateway(ExecutionGateway):
             }
             if conn_statuses != new_conn_statuses:
                 conn_statuses = new_conn_statuses
+                all_connected = all(map(lambda kv: kv[1], conn_statuses.items()))
                 self.logger.log(
-                    logging.INFO
-                    if all(map(lambda kv: kv[1], conn_statuses.items()))
-                    else logging.WARNING,
+                    logging.INFO if all_connected else logging.WARNING,
                     "[bybit_gw_connection_status.json]%s" % SLog(**conn_statuses),
                 )
+                self.bridge.on_is_ready(all_connected)
 
     def start(self, bridge: ExecutionGatewayBridge):
         def start_wrapper():
@@ -267,24 +375,35 @@ class BybitExecutionGateway(ExecutionGateway):
         bts_id = req.key()
         px = req.price()
         qty = req.quantity()
-        self.orders[req.key()] = OrderData(
-            bts_id=bts_id, exch_id=exch_ord_id, price=px, quantity=qty
-        )
         sec_name = req.exchange_security_id()
+        self.orders[req.key()] = OrderData(
+            account=self.account,
+            bts_id=bts_id,
+            exch_id=exch_ord_id,
+            exch_sec_id=req.exchange_security_id(),
+            sec_type=SecurityType.parse(req.security_type()),
+            is_buy=req.is_buy(),
+            order_type=OrderType.parse(req.order_type()),
+            time_in_force=TimeInForce.parse(req.time_in_force()),
+            price=px,
+            quantity=qty,
+        )
         sec = self.securities.get(sec_name)
         if not sec:
-            return
+            raise ValueError(f"Unknown security '{sec_name}'")
         params = {
             "orderLinkId": exch_ord_id,
             "category": "linear",
             "symbol": sec_name,
             "side": "Buy" if req.is_buy() else "Sell",
             "orderType": req.order_type().title(),
+            "timeInForce": req.time_in_force().upper(),
             "price": format_price(sec, px),
             "qty": format_quantity(sec, qty),
         }
+        on_ack, on_err = self._order_handlers("crt", exch_ord_id)
         self.logger.info("[bybit_gw_crt_ord_send.json]%s" % SLog(params=params))
-        self.trade_cli.place_order(lambda msg: self._on_order_ack("crt", msg), **params)
+        self.trade_cli.place_order(on_ack, **params, error_callback=on_err)
 
     def cancel_replace_order(self, req: PyCancelReplaceOrderRequest):
         self.logger.info("[bybit_gw_cxlrep_ord.json]%s" % req)
@@ -307,10 +426,9 @@ class BybitExecutionGateway(ExecutionGateway):
             "price": format_price(sec, req.price()),
             "qty": format_quantity(sec, req.quantity()),
         }
+        on_ack, on_err = self._order_handlers("cxlrep", ord_data.exch_id)
         self.logger.info("[bybit_gw_cxlrep_ord_send.json]%s" % SLog(params=params))
-        self.trade_cli.amend_order(
-            lambda msg: self._on_order_ack("cxlrep", msg), **params
-        )
+        self.trade_cli.amend_order(on_ack, **params, error_callback=on_err)
 
     def cancel_order(self, req: PyCancelOrderRequest):
         self.logger.info("[bybit_gw_cxl_ord.json]%s" % req)
@@ -327,10 +445,9 @@ class BybitExecutionGateway(ExecutionGateway):
             "category": "linear",
             "symbol": req.exchange_security_id(),
         }
+        on_ack, on_err = self._order_handlers("cxl", ord_data.exch_id)
         self.logger.info("[bybit_gw_cxl_ord_send.json]%s" % SLog(params=params))
-        self.trade_cli.cancel_order(
-            lambda msg: self._on_order_ack("cxl", msg), **params
-        )
+        self.trade_cli.cancel_order(on_ack, **params, error_callback=on_err)
 
     def on_securities(self, securities: PySecurityList):
         self.logger.info("[bybit_gw_sec.json]%s" % securities)

src/lib_python_interop/python_interop_plugins/bybit_md_provider.py

@@ -1,5 +1,4 @@
 import json
-import numpy as np
 import time
 import logging
 import sys
@@ -15,15 +14,18 @@ from python_interop_module import (
     PySecurityList,
 )
 from spark_core import (
+    JsonElement,
     configure_logger,
     get_contract_spec,
     get_price_info,
     get_quantity_info,
     get_security,
+    get_tick_factors,
     load_config,
 )
 from py_md_utils import PERP_EXP_STRING
 import requests
+import numpy as np
 
 logger = logging.getLogger("bybit_md_provider")
 
@@ -32,46 +34,74 @@ def normalize_contract_name(contract: str) -> str:
         return contract[:-4] + "_" + contract[-4:]
     return contract
 
-def fetch_contract_specs(contracts: Iterable[str], use_testnet: bool = True) -> dict:
-    default_specs = {
-        "tick_unit": 1,
-        "tick_base": 10,
-        "cash_per_point": 0.1,
-        "qty_tick_unit": 1,
-        "qty_tick_base": 10000
-    }
-    specs_dict = {contract: default_specs for contract in contracts}
+def fetch_futures_specs(contracts: Iterable[str], use_testnet: bool = True) -> dict:
+    # Fetch specs for futures only, explicitly, handling pagination
+    specs_dict = {}
     try:
         host = "api-testnet.bybit.com" if use_testnet else "api.bybit.com"
-        url = f"https://{host}/v5/market/instruments-info?category=linear"
-        response = requests.get(url, timeout=5)
-        response.raise_for_status()
-        data = response.json()
+        base_url = f"https://{host}/v5/market/instruments-info?category=linear&limit=500"
+        logger.info(f"Querying Bybit API for futures specs: {base_url}")
+        
         found_contracts = set()
-        for item in data.get("result", {}).get("list", []):
-            contract = item.get("symbol")
-            if contract in contracts:
-                price_tick = float(item.get("priceFilter", {}).get("tickSize", "0.1"))
-                lot_size = float(item.get("lotSizeFilter", {}).get("qtyStep", "0.001"))
-                tick_base = int(1 / price_tick) if price_tick != 0 else 10
-                if lot_size >= 1:
-                    qty_tick_base = int(lot_size)
-                else:
-                    qty_tick_base = int(1 / lot_size) if lot_size != 0 else 10000
-                specs_dict[contract] = {
-                    "tick_unit": 1,
-                    "tick_base": tick_base,
-                    "cash_per_point": price_tick,
-                    "qty_tick_unit": 1,
-                    "qty_tick_base": qty_tick_base
-                }
-                found_contracts.add(contract)
-        for contract in contracts:
-            if contract not in found_contracts:
-                logger.warning(f"Contract {contract} not found in API response, using default specs")
+        cursor = ""
+        
+        while True:
+            url = f"{base_url}&cursor={cursor}" if cursor else base_url
+            logger.info(f"Fetching page with cursor: {cursor or 'initial'}")
+            response = requests.get(url, timeout=5)
+            response.raise_for_status()
+            data = response.json()
+            
+            if data.get("retCode") != 0:
+                logger.error(f"Futures API error: {data.get('retMsg', 'Unknown error')}")
+                return specs_dict
+            
+            result = data.get("result", {})
+            symbols = result.get("list", [])
+            logger.info(f"Loaded page with cursor: {cursor or 'initial'}, symbols: {len(symbols)}")
+            
+            for item in symbols:
+                contract = item.get("symbol")
+                if contract in contracts:
+                    try:
+                        price_tick = float(item.get("priceFilter", {}).get("tickSize", 0.1))
+                        lot_size = item.get("lotSizeFilter", {})
+                        qty_step = float(lot_size.get("qtyStep", 0.001))
+                        
+                        if price_tick == 0 or qty_step == 0:
+                            logger.error(f"Invalid tick or lot size for {contract}: tickSize={price_tick}, qtyStep={qty_step}")
+                            continue
+                        
+                        tick_unit, tick_base = get_tick_factors(price_tick)
+                        qty_tick_unit, qty_tick_base = get_tick_factors(qty_step)
+                        
+                        specs_dict[contract] = {
+                            "tick_unit": tick_unit,
+                            "tick_base": tick_base,
+                            "cash_per_point": 1,
+                            "qty_tick_unit": qty_tick_unit,
+                            "qty_tick_base": qty_tick_base,
+                            "min_notional_value": float(
+                                lot_size.get("minNotionalValue", tick_unit / tick_base)
+                            ),
+                        }
+                        found_contracts.add(contract)
+                        logger.info(f"Fetched futures specs for {contract}: {specs_dict[contract]}")
+                    except (ValueError, TypeError) as e:
+                        logger.error(f"Failed to parse futures specs for {contract}: {e}")
+                        continue
+            
+            cursor = result.get("nextPageCursor", "")
+            if not cursor:
+                logger.info("No more pages to fetch; pagination complete")
+                break
+
+        logger.info(f"Returning specs for contracts: {list(specs_dict.keys())}")
+        logger.debug(f"Final specs: {specs_dict}")
         return specs_dict
+    
     except Exception as e:
-        logger.error(f"Failed to fetch contract specs: {e}", exc_info=True)
+        logger.error(f"Failed to fetch futures contract specs: {e}", exc_info=True)
         return specs_dict
 
 class BybitMdProvider(MdProvider):
@@ -113,9 +143,14 @@ class BybitMdProvider(MdProvider):
         contracts = list(
             chain(*[map(lambda f: f"{f}{c}", self.futures) for c in self.currencies])
         )
-        contract_specs = fetch_contract_specs(contracts, use_testnet=self.use_testnet)
+        contract_specs = fetch_futures_specs(contracts, use_testnet=self.use_testnet)
         sec_list = PySecurityList()
-
+        missing_contracts = list(filter(lambda c: c not in contract_specs, contracts))
+        if missing_contracts:
+            logger.error(
+                "Did not find instrument info for %s" % JsonElement(missing_contracts)
+            )
+            contracts = list(filter(lambda c: c in contract_specs, contracts))
         for contract in contracts:
             normalized_contract = normalize_contract_name(contract)
             asset, quote = normalized_contract.split("_")
@@ -131,13 +166,20 @@ class BybitMdProvider(MdProvider):
                 continue
 
             try:
+                price_tick_unit = specs.get("tick_unit", 1)
+                price_tick_base=specs.get("tick_base", 10)
                 spec = get_contract_spec(
                     price_info=get_price_info(
-                        tick_unit=specs.get("tick_unit", 1),
-                        tick_base=specs.get("tick_base", 10),
-                        cash_per_point=specs.get("cash_per_point", 0.1)
+                        tick_unit=price_tick_unit,
+                        tick_base=price_tick_base,
+                        cash_per_point=specs.get("cash_per_point", 0.1),
+                    ),
+                    qty_info=qty_info,
+                    notional_qty_info=get_quantity_info(
+                        tick_unit=price_tick_unit,
+                        tick_base=price_tick_base,
+                        min_value=specs.get("min_notional_value"),
                     ),
-                    qty_info=qty_info
                 )
             except Exception as e:
                 logger.error(f"Failed to get contract spec for contract: {contract}, error: {e}")
@@ -277,3 +319,47 @@ class BybitMdProvider(MdProvider):
 
         import os
         os._exit(0)
+
+
+if __name__ == "__main__":
+    import argparse
+    from fake_md_bridge import FakeMdBridge
+
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--currencies", nargs="+")
+    ap.add_argument("--futures", nargs="+")
+    log_levels = list(logging._nameToLevel)
+    ap.add_argument("--console-level", choices=log_levels)
+    ap.add_argument("--file-level", choices=log_levels)
+    ap.add_argument("--dev", action="store_true", dest="use_testnet")
+    ap.add_argument("--prod", action="store_false", dest="use_testnet")
+    ap.set_defaults(
+        currencies=None,
+        futures=None,
+        console_level=None,
+        file_level=None,
+        use_testnet=None,
+    )
+    args = ap.parse_args()
+    print(args)
+    logger = logging.getLogger("bybit_md_provider_app")
+    md_pro = BybitMdProvider()
+    config = md_pro.config
+    if args.currencies:
+        config["currencies"] = args.currencies
+    if args.futures:
+        config["futures"] = args.futures
+    logging_overrides = {}
+    if args.console_level:
+        logging_overrides["console_level"] = args.console_level
+    if args.file_level:
+        logging_overrides["file_level"] = args.file_level
+    if logging_overrides:
+        config["logging"].update(logging_overrides)
+    if args.use_testnet is not None:
+        config["use_testnet"] = args.use_testnet
+    # print(json.dumps(config))
+    md_bridge = FakeMdBridge(logger)
+    md_pro.start(md_bridge)  # type: ignore
+    while True:
+        time.sleep(1)

src/lib_python_interop/python_interop_plugins/console_table.py

@@ -0,0 +1,102 @@
+import sys
+import threading
+import time
+import traceback
+
+
+class ConsoleTableColumn:
+    def __init__(self, id, header: str, data_format=None, width=None):
+        self.id = id
+        self.header = header
+        self.data_format = data_format
+        self.width = width or len(header)
+
+
+class ConsoleTable:
+    ESC = "\033["
+
+    def __init__(self, columns: list[ConsoleTableColumn], col_separator=" "):
+        self.columns = columns
+        self.col_separator = col_separator
+        self.header_row = col_separator.join(map(lambda c: c.header, columns))
+        self._pending_data = None
+        self._is_running = False
+
+    def move_cursor(self, row, col):
+        sys.stdout.write(f"{ConsoleTable.ESC}{row};{col}H")
+
+    def clear_screen(self):
+        sys.stdout.write(f"{ConsoleTable.ESC}2J")
+
+    def hide_cursor(self):
+        sys.stdout.write(f"{ConsoleTable.ESC}?25l")
+
+    def show_cursor(self):
+        sys.stdout.write(f"{ConsoleTable.ESC}?25h")
+
+    def print_table(self, rows):
+        print(self.header_row)
+        print("-" * len(self.header_row))
+        for row in rows:
+            cells = []
+            for col in self.columns:
+                val = row[col.id]
+                cell = f"{val:{col.data_format}}" if col.data_format else str(val)
+                len_diff = col.width - len(cell)
+                if len_diff > 0:
+                    cell += " " * len_diff
+                else:
+                    leading_ws_count = 0
+                    while (
+                        leading_ws_count < -len_diff and cell[leading_ws_count] == " "
+                    ):
+                        leading_ws_count += 1
+                    if leading_ws_count > 0:
+                        cell = cell[leading_ws_count:]
+                    len_diff = col.width - len(cell)
+                    if len_diff < 0:
+                        cell = cell[:len_diff]
+                cells.append(cell)
+            print(self.col_separator.join(cells))
+
+    def set_data(self, data):
+        self._pending_data = data
+
+    def run(self):
+        try:
+            self.clear_screen()
+            self.hide_cursor()
+            while self._is_running:
+                if self._pending_data:
+                    # move to bottom of screen
+                    self.move_cursor(1, 1)
+                    self.print_table(self._pending_data)
+                    self._pending_data = None
+                    sys.stdout.flush()
+                time.sleep(0.5)
+        finally:
+            self.show_cursor()
+
+    def run_in_background(self):
+        def run_wrapper():
+            self.run()
+
+        self._is_running = True
+        threading.Thread(target=run_wrapper, daemon=True)
+
+
+if __name__ == "__main__":
+    cols = [
+        ConsoleTableColumn("symbol", f"{'Symbol':<10}"),
+        ConsoleTableColumn("bid", f"{'Bid':>10}", data_format=">10"),
+        ConsoleTableColumn("ask", f"{'Ask':>10}", data_format=">10"),
+    ]
+    bbo_data = [
+        {"symbol": "BTCUSDT", "bid": 104500.00, "ask": 105000.00},
+        {"symbol": "BNBUSDT", "bid": 645.00, "ask": 650.00},
+    ]
+    try:
+        table = ConsoleTable(cols)
+        table.print_table(bbo_data)
+    except Exception:
+        traceback.print_exc()

src/lib_python_interop/python_interop_plugins/fake_md_bridge.py

@@ -0,0 +1,19 @@
+import logging
+from typing import Optional
+
+
+class FakeMdBridge:
+    def __init__(self, logger: Optional[logging.Logger] = logging.getLogger()):
+        self.logger = logger
+
+    def on_books(self, v):
+        if self.logger:
+            self.logger.info("[books.json]%s" % v)
+
+    def on_securities(self, v):
+        if self.logger:
+            self.logger.info("[securities.json]%s" % v)
+
+    def on_trades(self, v):
+        if self.logger:
+            self.logger.info("[trades.json]%s" % v)

src/lib_python_interop/python_interop_plugins/pybit_unified_trading.py

@@ -0,0 +1,463 @@
+import logging
+from dataclasses import dataclass
+from typing import Any
+from pybit.exceptions import (
+    InvalidChannelTypeError,
+    TopicMismatchError,
+    UnauthorizedExceptionError,
+)
+from pybit._v5_misc import MiscHTTP
+from pybit._v5_market import MarketHTTP
+from pybit._v5_trade import TradeHTTP
+from pybit._v5_account import AccountHTTP
+from pybit._v5_asset import AssetHTTP
+from pybit._v5_position import PositionHTTP
+from pybit._v5_pre_upgrade import PreUpgradeHTTP
+from pybit._v5_spot_leverage_token import SpotLeverageHTTP
+from pybit._v5_spot_margin_trade import SpotMarginTradeHTTP
+from pybit._v5_user import UserHTTP
+from pybit._v5_broker import BrokerHTTP
+from pybit._v5_institutional_loan import InstitutionalLoanHTTP
+from pybit._v5_crypto_loan import CryptoLoanHTTP
+from pybit._v5_earn import EarnHTTP
+from pybit_websocket_stream import _V5WebSocketManager
+from pybit_websocket_trading import _V5TradeWebSocketManager
+from pybit._v5_spread import (
+    SpreadHTTP,
+    _V5WebSocketSpreadTrading,
+)
+
+
+logger = logging.getLogger(__name__)
+
+WSS_NAME = "Unified V5"
+PRIVATE_WSS = "wss://{SUBDOMAIN}.{DOMAIN}.com/v5/private"
+PUBLIC_WSS = "wss://{SUBDOMAIN}.{DOMAIN}.com/v5/public/{CHANNEL_TYPE}"
+AVAILABLE_CHANNEL_TYPES = [
+    "inverse",
+    "linear",
+    "spot",
+    "option",
+    "private",
+]
+
+
+@dataclass
+class HTTP(
+    MiscHTTP,
+    MarketHTTP,
+    TradeHTTP,
+    AccountHTTP,
+    AssetHTTP,
+    PositionHTTP,
+    PreUpgradeHTTP,
+    SpotLeverageHTTP,
+    SpotMarginTradeHTTP,
+    UserHTTP,
+    BrokerHTTP,
+    InstitutionalLoanHTTP,
+    CryptoLoanHTTP,
+    EarnHTTP,
+):
+    def __init__(self, **args):
+        super().__init__(**args)
+
+
+class WebSocket(_V5WebSocketManager):
+    def _validate_public_topic(self):
+        if "/v5/public" not in self.WS_URL:
+            raise TopicMismatchError(
+                "Requested topic does not match channel_type"
+            )
+
+    def _validate_private_topic(self):
+        if not self.WS_URL.endswith("/private"):
+            raise TopicMismatchError(
+                "Requested topic does not match channel_type"
+            )
+
+    def __init__(
+        self,
+        channel_type: str,
+        **kwargs,
+    ):
+        super().__init__(WSS_NAME, **kwargs)
+        if channel_type not in AVAILABLE_CHANNEL_TYPES:
+            raise InvalidChannelTypeError(
+                f"Channel type is not correct. Available: {AVAILABLE_CHANNEL_TYPES}"
+            )
+
+        if channel_type == "private":
+            self.WS_URL = PRIVATE_WSS
+        else:
+            self.WS_URL = PUBLIC_WSS.replace("{CHANNEL_TYPE}", channel_type)
+            # Do not pass keys and attempt authentication on a public connection
+            self.api_key = None
+            self.api_secret = None
+
+        if (
+            self.api_key is None or self.api_secret is None
+        ) and channel_type == "private":
+            raise UnauthorizedExceptionError(
+                "API_KEY or API_SECRET is not set. They both are needed in order to access private topics"
+            )
+
+        self._connect(self.WS_URL)
+
+    # Private topics
+
+    def position_stream(self, callback):
+        """Subscribe to the position stream to see changes to your position data in real-time.
+
+        Push frequency: real-time
+
+        Additional information:
+            https://bybit-exchange.github.io/docs/v5/websocket/private/position
+        """
+        self._validate_private_topic()
+        topic = "position"
+        self.subscribe(topic, callback)
+
+    def order_stream(self, callback):
+        """Subscribe to the order stream to see changes to your orders in real-time.
+
+        Push frequency: real-time
+
+        Additional information:
+            https://bybit-exchange.github.io/docs/v5/websocket/private/order
+        """
+        self._validate_private_topic()
+        topic = "order"
+        self.subscribe(topic, callback)
+
+    def execution_stream(self, callback):
+        """Subscribe to the execution stream to see your executions in real-time.
+
+        Push frequency: real-time
+
+        Additional information:
+            https://bybit-exchange.github.io/docs/v5/websocket/private/execution
+        """
+        self._validate_private_topic()
+        topic = "execution"
+        self.subscribe(topic, callback)
+
+    def fast_execution_stream(self, callback, categorised_topic=""):
+        """Fast execution stream significantly reduces data latency compared
+        original "execution" stream. However, it pushes limited execution type
+        of trades, and fewer data fields.
+        Use categorised_topic as a filter for a certain `category`. See docs.
+
+        Push frequency: real-time
+
+        Additional information:
+            https://bybit-exchange.github.io/docs/v5/websocket/private/fast-execution
+        """
+        self._validate_private_topic()
+        topic = "execution.fast"
+        if categorised_topic:
+            topic += "." + categorised_topic
+        self.subscribe(topic, callback)
+
+    def wallet_stream(self, callback):
+        """Subscribe to the wallet stream to see changes to your wallet in real-time.
+
+        Push frequency: real-time
+
+        Additional information:
+            https://bybit-exchange.github.io/docs/v5/websocket/private/wallet
+        """
+        self._validate_private_topic()
+        topic = "wallet"
+        self.subscribe(topic, callback)
+
+    def greek_stream(self, callback):
+        """Subscribe to the greeks stream to see changes to your greeks data in real-time. option only.
+
+        Push frequency: real-time
+
+        Additional information:
+            https://bybit-exchange.github.io/docs/v5/websocket/private/greek
+        """
+        self._validate_private_topic()
+        topic = "greeks"
+        self.subscribe(topic, callback)
+
+    def spread_order_stream(self, callback):
+        """Subscribe to the spread trading order stream to see changes to your orders in real-time.
+
+        Push frequency: real-time
+
+        Additional information:
+            https://bybit-exchange.github.io/docs/v5/spread/websocket/private/order
+        """
+        self._validate_private_topic()
+        topic = "spread.order"
+        self.subscribe(topic, callback)
+
+    def spread_execution_stream(self, callback):
+        """Subscribe to the spread trading execution stream to see your executions in real-time.
+
+        Push frequency: real-time
+
+        Additional information:
+            https://bybit-exchange.github.io/docs/v5/spread/websocket/private/execution
+        """
+        self._validate_private_topic()
+        topic = "spread.execution"
+        self.subscribe(topic, callback)
+
+    # Public topics
+
+    def orderbook_stream(self, depth: int, symbol: (str, list), callback):
+        """Subscribe to the orderbook stream. Supports different depths.
+
+        Linear & inverse:
+        Level 1 data, push frequency: 10ms
+        Level 50 data, push frequency: 20ms
+        Level 200 data, push frequency: 100ms
+        Level 500 data, push frequency: 100ms
+
+        Spot:
+        Level 1 data, push frequency: 10ms
+        Level 50 data, push frequency: 20ms
+
+        Option:
+        Level 25 data, push frequency: 20ms
+        Level 100 data, push frequency: 100ms
+
+        Required args:
+            symbol (string/list): Symbol name(s)
+            depth (int): Orderbook depth
+
+        Additional information:
+            https://bybit-exchange.github.io/docs/v5/websocket/public/orderbook
+        """
+        self._validate_public_topic()
+        topic = f"orderbook.{depth}." + "{symbol}"
+        self.subscribe(topic, callback, symbol)
+
+    def trade_stream(self, symbol: (str, list), callback):
+        """
+        Subscribe to the recent trades stream.
+        After subscription, you will be pushed trade messages in real-time.
+
+        Push frequency: real-time
+
+        Required args:
+            symbol (string/list): Symbol name(s)
+
+         Additional information:
+            https://bybit-exchange.github.io/docs/v5/websocket/public/trade
+        """
+        self._validate_public_topic()
+        topic = f"publicTrade." + "{symbol}"
+        self.subscribe(topic, callback, symbol)
+
+    def ticker_stream(self, symbol: (str, list), callback):
+        """Subscribe to the ticker stream.
+
+        Push frequency: 100ms
+
+        Required args:
+            symbol (string/list): Symbol name(s)
+
+         Additional information:
+            https://bybit-exchange.github.io/docs/v5/websocket/public/ticker
+        """
+        self._validate_public_topic()
+        topic = "tickers.{symbol}"
+        self.subscribe(topic, callback, symbol)
+
+    def kline_stream(self, interval: int, symbol: (str, list), callback):
+        """Subscribe to the klines stream.
+
+        Push frequency: 1-60s
+
+        Required args:
+            symbol (string/list): Symbol name(s)
+            interval (int): Kline interval
+
+         Additional information:
+            https://bybit-exchange.github.io/docs/v5/websocket/public/kline
+        """
+        self._validate_public_topic()
+        topic = f"kline.{interval}." + "{symbol}"
+        self.subscribe(topic, callback, symbol)
+
+    def liquidation_stream(self, symbol: (str, list), callback):
+        """
+        Pushes at most one order per second per symbol.
+        As such, this feed does not push all liquidations that occur on Bybit.
+
+        Push frequency: 1s
+
+        Required args:
+            symbol (string/list): Symbol name(s)
+
+         Additional information:
+            https://bybit-exchange.github.io/docs/v5/websocket/public/liquidation
+        """
+        logger.warning("liquidation_stream() is deprecated. Please use "
+                       "all_liquidation_stream().")
+        self._validate_public_topic()
+        topic = "liquidation.{symbol}"
+        self.subscribe(topic, callback, symbol)
+
+    def all_liquidation_stream(self, symbol: (str, list), callback):
+        """Subscribe to the liquidation stream, push all liquidations that
+        occur on Bybit.
+
+        Push frequency: 500ms
+
+        Required args:
+            symbol (string/list): Symbol name(s)
+
+         Additional information:
+            https://bybit-exchange.github.io/docs/v5/websocket/public/all-liquidation
+        """
+        self._validate_public_topic()
+        topic = "allLiquidation.{symbol}"
+        self.subscribe(topic, callback, symbol)
+
+    def lt_kline_stream(self, interval: int, symbol: (str, list), callback):
+        """Subscribe to the leveraged token kline stream.
+
+        Push frequency: 1-60s
+
+        Required args:
+            symbol (string/list): Symbol name(s)
+            interval (int): Leveraged token Kline stream interval
+
+         Additional information:
+            https://bybit-exchange.github.io/docs/v5/websocket/public/etp-kline
+        """
+        self._validate_public_topic()
+        topic = f"kline_lt.{interval}." + "{symbol}"
+        self.subscribe(topic, callback, symbol)
+
+    def lt_ticker_stream(self, symbol: (str, list), callback):
+        """Subscribe to the leveraged token ticker stream.
+
+        Push frequency: 300ms
+
+        Required args:
+            symbol (string/list): Symbol name(s)
+
+         Additional information:
+            https://bybit-exchange.github.io/docs/v5/websocket/public/etp-ticker
+        """
+        self._validate_public_topic()
+        topic = "tickers_lt.{symbol}"
+        self.subscribe(topic, callback, symbol)
+
+    def lt_nav_stream(self, symbol: (str, list), callback):
+        """Subscribe to the leveraged token nav stream.
+
+        Push frequency: 300ms
+
+        Required args:
+            symbol (string/list): Symbol name(s)
+
+         Additional information:
+            https://bybit-exchange.github.io/docs/v5/websocket/public/etp-nav
+        """
+        self._validate_public_topic()
+        topic = "lt.{symbol}"
+        self.subscribe(topic, callback, symbol)
+
+
+class WebSocketTrading(_V5TradeWebSocketManager):
+    def __init__(self, recv_window=0, referral_id="", **kwargs):
+        super().__init__(recv_window, referral_id, **kwargs)
+
+    @staticmethod
+    def _extract_error_callback(req_params: dict[str, Any]):
+        err_cb = req_params.pop("error_callback") if "error_callback" in req_params else None
+        return (req_params, err_cb)
+
+    def place_order(self, callback, **kwargs):
+        operation = "order.create"
+        self._send_order_operation(
+            operation, callback, *self._extract_error_callback(kwargs)
+        )
+
+    def amend_order(self, callback, **kwargs):
+        operation = "order.amend"
+        self._send_order_operation(
+            operation, callback, *self._extract_error_callback(kwargs)
+        )
+
+    def cancel_order(self, callback, **kwargs):
+        operation = "order.cancel"
+        self._send_order_operation(
+            operation, callback, *self._extract_error_callback(kwargs)
+        )
+
+    def place_batch_order(self, callback, **kwargs):
+        operation = "order.create-batch"
+        self._send_order_operation(
+            operation, callback, *self._extract_error_callback(kwargs)
+        )
+
+    def amend_batch_order(self, callback, **kwargs):
+        operation = "order.amend-batch"
+        self._send_order_operation(
+            operation, callback, *self._extract_error_callback(kwargs)
+        )
+
+    def cancel_batch_order(self, callback, **kwargs):
+        operation = "order.cancel-batch"
+        self._send_order_operation(
+            operation, callback, *self._extract_error_callback(kwargs)
+        )
+
+
+class WebsocketSpreadTrading(_V5WebSocketSpreadTrading):
+    def __init__(self,  **kwargs):
+        super().__init__(**kwargs)
+
+    def orderbook_stream(self, depth: int, symbol: (str, list), callback):
+        """Subscribe to the orderbook stream. Supports different depths.
+
+        Level 25 data, push frequency: 20ms
+
+        Required args:
+            symbol (string/list): Symbol name(s)
+            depth (int): Orderbook depth
+
+        Additional information:
+            https://bybit-exchange.github.io/docs/v5/spread/websocket/public/orderbook
+        """
+        topic = f"orderbook.{depth}." + "{symbol}"
+        self.subscribe(topic, callback, symbol)
+
+    def trade_stream(self, symbol: (str, list), callback):
+        """
+        Subscribe to the recent trades stream.
+        After subscription, you will be pushed trade messages in real-time.
+
+        Push frequency: real-time
+
+        Required args:
+            symbol (string/list): Symbol name(s)
+
+         Additional information:
+            https://bybit-exchange.github.io/docs/v5/spread/websocket/public/trade
+        """
+        topic = f"publicTrade." + "{symbol}"
+        self.subscribe(topic, callback, symbol)
+
+    def ticker_stream(self, symbol: (str, list), callback):
+        """Subscribe to the ticker stream.
+
+        Push frequency: 100ms
+
+        Required args:
+            symbol (string/list): Symbol name(s)
+
+         Additional information:
+            https://bybit-exchange.github.io/docs/v5/spread/websocket/public/ticker
+        """
+        topic = "tickers.{symbol}"
+        self.subscribe(topic, callback, symbol)

src/lib_python_interop/python_interop_plugins/pybit_websocket_stream.py

@@ -0,0 +1,514 @@
+import websocket
+import threading
+import time
+import json
+from pybit._http_manager import generate_signature
+import logging
+import copy
+from uuid import uuid4
+from pybit import _helpers
+
+
+logger = logging.getLogger(__name__)
+
+
+SUBDOMAIN_TESTNET = "stream-testnet"
+SUBDOMAIN_MAINNET = "stream"
+DEMO_SUBDOMAIN_TESTNET = "stream-demo-testnet"
+DEMO_SUBDOMAIN_MAINNET = "stream-demo"
+DOMAIN_MAIN = "bybit"
+DOMAIN_ALT = "bytick"
+TLD_MAIN = "com"
+
+
+class _WebSocketManager:
+    def __init__(
+        self,
+        _callback_function,
+        ws_name,
+        testnet,
+        tld="",
+        domain="",
+        demo=False,
+        rsa_authentication=False,
+        api_key=None,
+        api_secret=None,
+        ping_interval=20,
+        ping_timeout=10,
+        retries=10,
+        restart_on_error=True,
+        trace_logging=False,
+        private_auth_expire=1,
+    ):
+        self.testnet = testnet
+        self.domain = domain
+        self.tld = tld
+        self.rsa_authentication = rsa_authentication
+        self.demo = demo
+        # Set API keys.
+        self.api_key = api_key
+        self.api_secret = api_secret
+
+        self.callback = _callback_function
+        self.ws_name = ws_name
+        if api_key:
+            self.ws_name += " (Auth)"
+        
+        # Delta time for private auth expiration in seconds
+        self.private_auth_expire = private_auth_expire
+
+        # Setup the callback directory following the format:
+        #   {
+        #       "topic_name": function
+        #   }
+        self.callback_directory = {}
+        self.error_callback_directory = {}
+
+        # Record the subscriptions made so that we can resubscribe if the WSS
+        # connection is broken.
+        self.subscriptions = []
+
+        # Set ping settings.
+        self.ping_interval = ping_interval
+        self.ping_timeout = ping_timeout
+        self.custom_ping_message = json.dumps({"op": "ping"})
+        self.retries = retries
+
+        # Other optional data handling settings.
+        self.handle_error = restart_on_error
+
+        # Enable websocket-client's trace logging for extra debug information
+        # on the websocket connection, including the raw sent & recv messages
+        websocket.enableTrace(trace_logging)
+
+        # Set initial state, initialize dictionary and connect.
+        self._reset()
+        self.attempting_connection = False
+
+    def _on_open(self):
+        """
+        Log WS open.
+        """
+        logger.debug(f"WebSocket {self.ws_name} opened.")
+
+    def _on_message(self, message):
+        """
+        Parse incoming messages.
+        """
+        message = json.loads(message)
+        if self._is_custom_pong(message):
+            return
+        else:
+            self.callback(message)
+
+    def is_connected(self):
+        try:
+            if self.ws.sock.connected:
+                return True
+            else:
+                return False
+        except AttributeError:
+            return False
+
+    def _connect(self, url):
+        """
+        Open websocket in a thread.
+        """
+
+        def resubscribe_to_topics():
+            if not self.subscriptions:
+                # There are no subscriptions to resubscribe to, probably
+                # because this is a brand new WSS initialisation so there was
+                # no previous WSS connection.
+                return
+
+            for req_id, subscription_message in self.subscriptions.items():
+                self.ws.send(subscription_message)
+
+        self.attempting_connection = True
+
+        # Set endpoint.
+        subdomain = SUBDOMAIN_TESTNET if self.testnet else SUBDOMAIN_MAINNET
+        domain = DOMAIN_MAIN if not self.domain else self.domain
+        tld = TLD_MAIN if not self.tld else self.tld
+        if self.demo:
+            if self.testnet:
+                subdomain = DEMO_SUBDOMAIN_TESTNET
+            else:
+                subdomain = DEMO_SUBDOMAIN_MAINNET
+        url = url.format(SUBDOMAIN=subdomain, DOMAIN=domain, TLD=tld)
+        self.endpoint = url
+
+        # Attempt to connect for X seconds.
+        retries = self.retries
+        if retries == 0:
+            infinitely_reconnect = True
+        else:
+            infinitely_reconnect = False
+
+        while (
+            infinitely_reconnect or retries > 0
+        ) and not self.is_connected():
+            logger.info(f"WebSocket {self.ws_name} attempting connection...")
+            self.ws = websocket.WebSocketApp(
+                url=url,
+                on_message=lambda ws, msg: self._on_message(msg),
+                on_close=lambda ws, *args: self._on_close(),
+                on_open=lambda ws, *args: self._on_open(),
+                on_error=lambda ws, err: self._on_error(err),
+                on_pong=lambda ws, *args: self._on_pong(),
+            )
+
+            # Setup the thread running WebSocketApp.
+            self.wst = threading.Thread(
+                target=lambda: self.ws.run_forever(
+                    ping_interval=self.ping_interval,
+                    ping_timeout=self.ping_timeout,
+                )
+            )
+
+            # Configure as daemon; start.
+            self.wst.daemon = True
+            self.wst.start()
+
+            retries -= 1
+            while self.wst.is_alive():
+                if self.ws.sock and self.is_connected():
+                    break
+
+            # If connection was not successful, raise error.
+            if not infinitely_reconnect and retries <= 0:
+                self.exit()
+                raise websocket.WebSocketTimeoutException(
+                    f"WebSocket {self.ws_name} ({self.endpoint}) connection "
+                    f"failed. Too many connection attempts. pybit will no "
+                    f"longer try to reconnect."
+                )
+
+        logger.info(f"WebSocket {self.ws_name} connected")
+
+        # If given an api_key, authenticate.
+        if self.api_key and self.api_secret:
+            self._auth()
+
+        resubscribe_to_topics()
+        self._send_initial_ping()
+
+        self.attempting_connection = False
+
+    def _auth(self):
+        """
+        Prepares authentication signature per Bybit API specifications.
+        """
+
+        expires = _helpers.generate_timestamp() + (self.private_auth_expire * 1000)
+
+        param_str = f"GET/realtime{expires}"
+
+        signature = generate_signature(
+            self.rsa_authentication, self.api_secret, param_str
+        )
+
+        # Authenticate with API.
+        self.ws.send(
+            json.dumps(
+                {"op": "auth", "args": [self.api_key, expires, signature]}
+            )
+        )
+
+    def _on_error(self, error):
+        """
+        Exit on errors and raise exception, or attempt reconnect.
+        """
+        if type(error).__name__ not in [
+            "WebSocketConnectionClosedException",
+            "ConnectionResetError",
+            "WebSocketTimeoutException",
+        ]:
+            # Raises errors not related to websocket disconnection.
+            self.exit()
+            raise error
+
+        if not self.exited:
+            logger.error(
+                f"WebSocket {self.ws_name} ({self.endpoint}) "
+                f"encountered error: {error}."
+            )
+            self.exit()
+
+        # Reconnect.
+        if self.handle_error and not self.attempting_connection:
+            self._reset()
+            self._connect(self.endpoint)
+
+    def _on_close(self):
+        """
+        Log WS close.
+        """
+        logger.debug(f"WebSocket {self.ws_name} closed.")
+
+    def _on_pong(self):
+        """
+        Sends a custom ping upon the receipt of the pong frame.
+
+        The websocket library will automatically send ping frames. However, to
+        ensure the connection to Bybit stays open, we need to send a custom
+        ping message separately from this. When we receive the response to the
+        ping frame, this method is called, and we will send the custom ping as
+        a normal OPCODE_TEXT message and not an OPCODE_PING.
+        """
+        self._send_custom_ping()
+
+    def _send_custom_ping(self):
+        self.ws.send(self.custom_ping_message)
+
+    def _send_initial_ping(self):
+        """https://github.com/bybit-exchange/pybit/issues/164"""
+        timer = threading.Timer(
+            self.ping_interval, self._send_custom_ping
+        )
+        timer.start()
+
+    @staticmethod
+    def _is_custom_pong(message):
+        """
+        Referring to OPCODE_TEXT pongs from Bybit, not OPCODE_PONG.
+        """
+        if message.get("ret_msg") == "pong" or message.get("op") == "pong":
+            return True
+
+    def _reset(self):
+        """
+        Set state booleans and initialize dictionary.
+        """
+        self.exited = False
+        self.auth = False
+        self.data = {}
+
+    def exit(self):
+        """
+        Closes the websocket connection.
+        """
+
+        self.ws.close()
+        while self.ws.sock:
+            continue
+        self.exited = True
+
+
+class _V5WebSocketManager(_WebSocketManager):
+    def __init__(self, ws_name, **kwargs):
+        callback_function = (
+            kwargs.pop("callback_function")
+            if kwargs.get("callback_function")
+            else self._handle_incoming_message
+        )
+        super().__init__(callback_function, ws_name, **kwargs)
+
+        self.subscriptions = {}
+
+        self.standard_private_topics = [
+            "position",
+            "execution",
+            "order",
+            "wallet",
+            "greeks",
+            "spread.order",
+            "spread.execution",
+        ]
+
+        self.other_private_topics = [
+            "execution.fast"
+        ]
+
+    def subscribe(
+            self,
+            topic: str,
+            callback,
+            symbol: (str, list) = False
+    ):
+
+        def prepare_subscription_args(list_of_symbols):
+            """
+            Prepares the topic for subscription by formatting it with the
+            desired symbols.
+            """
+
+            if topic in self.standard_private_topics:
+                # private topics do not support filters
+                return [topic]
+
+            topics = []
+            for single_symbol in list_of_symbols:
+                topics.append(topic.format(symbol=single_symbol))
+            return topics
+
+        if type(symbol) == str:
+            symbol = [symbol]
+
+        subscription_args = prepare_subscription_args(symbol)
+        self._check_callback_directory(subscription_args)
+
+        req_id = str(uuid4())
+
+        subscription_message = json.dumps(
+            {"op": "subscribe", "req_id": req_id, "args": subscription_args}
+        )
+        while not self.is_connected():
+            # Wait until the connection is open before subscribing.
+            time.sleep(0.1)
+        self.ws.send(subscription_message)
+        self.subscriptions[req_id] = subscription_message
+        for topic in subscription_args:
+            self._set_callback(topic, callback)
+
+    def _initialise_local_data(self, topic):
+        # Create self.data
+        try:
+            self.data[topic]
+        except KeyError:
+            self.data[topic] = []
+
+    def _process_delta_orderbook(self, message, topic):
+        self._initialise_local_data(topic)
+
+        # Record the initial snapshot.
+        if "snapshot" in message["type"]:
+            self.data[topic] = message["data"]
+            return
+
+        # Make updates according to delta response.
+        book_sides = {"b": message["data"]["b"], "a": message["data"]["a"]}
+        self.data[topic]["u"]=message["data"]["u"]
+        self.data[topic]["seq"]=message["data"]["seq"]
+
+        for side, entries in book_sides.items():
+            for entry in entries:
+                # Delete.
+                if float(entry[1]) == 0:
+                    index = _helpers.find_index(
+                        self.data[topic][side], entry, 0
+                    )
+                    self.data[topic][side].pop(index)
+                    continue
+
+                # Insert.
+                price_level_exists = entry[0] in [
+                    level[0] for level in self.data[topic][side]
+                ]
+                if not price_level_exists:
+                    self.data[topic][side].append(entry)
+                    continue
+
+                # Update.
+                qty_changed = entry[1] != next(
+                    level[1]
+                    for level in self.data[topic][side]
+                    if level[0] == entry[0]
+                )
+                if price_level_exists and qty_changed:
+                    index = _helpers.find_index(
+                        self.data[topic][side], entry, 0
+                    )
+                    self.data[topic][side][index] = entry
+                    continue
+
+    def _process_delta_ticker(self, message, topic):
+        self._initialise_local_data(topic)
+
+        # Record the initial snapshot.
+        if "snapshot" in message["type"]:
+            self.data[topic] = message["data"]
+
+        # Make updates according to delta response.
+        elif "delta" in message["type"]:
+            for key, value in message["data"].items():
+                self.data[topic][key] = value
+
+    def _process_auth_message(self, message):
+        # If we get successful futures auth, notify user
+        if message.get("success") is True:
+            logger.debug(f"Authorization for {self.ws_name} successful.")
+            self.auth = True
+        # If we get unsuccessful auth, notify user.
+        elif message.get("success") is False or message.get("type") == "error":
+            raise Exception(
+                f"Authorization for {self.ws_name} failed. Please check your "
+                f"API keys and resync your system time. Raw error: {message}"
+            )
+
+    def _process_subscription_message(self, message):
+        if message.get("req_id"):
+            sub = self.subscriptions[message["req_id"]]
+        else:
+            # if req_id is not supported, guess that the last subscription
+            # sent was successful
+            sub = json.loads(list(self.subscriptions.items())[0][1])["args"][0]
+
+        # If we get successful futures subscription, notify user
+        if message.get("success") is True:
+            logger.debug(f"Subscription to {sub} successful.")
+        # Futures subscription fail
+        elif message.get("success") is False:
+            response = message["ret_msg"]
+            logger.error("Couldn't subscribe to topic." f"Error: {response}.")
+            self._pop_callback(sub[0])
+
+    def _process_normal_message(self, message):
+        topic = message["topic"]
+        if "orderbook" in topic:
+            self._process_delta_orderbook(message, topic)
+            callback_data = copy.deepcopy(message)
+            callback_data["type"] = "snapshot"
+            callback_data["data"] = self.data[topic]
+        elif "tickers" in topic:
+            self._process_delta_ticker(message, topic)
+            callback_data = copy.deepcopy(message)
+            callback_data["type"] = "snapshot"
+            callback_data["data"] = self.data[topic]
+        else:
+            callback_data = message
+        callback_function = self._get_callback(topic)
+        callback_function(callback_data)
+
+    def _handle_incoming_message(self, message):
+        def is_auth_message():
+            if (
+                message.get("op") == "auth"
+                or message.get("type") == "AUTH_RESP"
+            ):
+                return True
+            else:
+                return False
+
+        def is_subscription_message():
+            if (
+                message.get("op") == "subscribe"
+                or message.get("type") == "COMMAND_RESP"
+            ):
+                return True
+            else:
+                return False
+
+        if is_auth_message():
+            self._process_auth_message(message)
+        elif is_subscription_message():
+            self._process_subscription_message(message)
+        else:
+            self._process_normal_message(message)
+
+    def _check_callback_directory(self, topics):
+        for topic in topics:
+            if topic in self.callback_directory:
+                raise Exception(
+                    f"You have already subscribed to this topic: " f"{topic}"
+                )
+
+    def _set_callback(self, topic, callback_function):
+        self.callback_directory[topic] = callback_function
+
+    def _get_callback(self, topic):
+        return self.callback_directory[topic]
+
+    def _pop_callback(self, topic):
+        self.callback_directory.pop(topic)

src/lib_python_interop/python_interop_plugins/pybit_websocket_trading.py

@@ -0,0 +1,100 @@
+from dataclasses import dataclass, field
+import json
+import uuid
+import logging
+from pybit_websocket_stream import _WebSocketManager
+from pybit import _helpers
+
+
+logger = logging.getLogger(__name__)
+
+
+WSS_NAME = "WebSocket Trading"
+TRADE_WSS = "wss://{SUBDOMAIN}.{DOMAIN}.{TLD}/v5/trade"
+
+
+class _V5TradeWebSocketManager(_WebSocketManager):
+    def __init__(self, recv_window, referral_id, **kwargs):
+        super().__init__(self._handle_incoming_message, WSS_NAME, **kwargs)
+        self.recv_window = recv_window
+        self.referral_id = referral_id
+        self._connect(TRADE_WSS)
+
+    def _process_auth_message(self, message):
+        # If we get successful auth, notify user
+        if message.get("retCode") == 0:
+            logger.debug(f"Authorization for {self.ws_name} successful.")
+            self.auth = True
+        # If we get unsuccessful auth, notify user.
+        else:
+            raise Exception(
+                f"Authorization for {self.ws_name} failed. Please check your "
+                f"API keys and resync your system time. Raw error: {message}"
+            )
+
+    def _process_error_message(self, message):
+        logger.error(
+            f"WebSocket request {message['reqId']} hit an error. Enabling "
+            f"traceLogging to reproduce the issue. Raw error: {message}"
+        )
+        req_id = message["reqId"]
+        self._pop_callback(req_id)
+        if req_id in self.error_callback_directory:
+            self.error_callback_directory.pop(req_id)(message)
+
+    def _handle_incoming_message(self, message):
+        def is_auth_message():
+            if message.get("op") == "auth":
+                return True
+            else:
+                return False
+
+        def is_error_message():
+            if message.get("retCode") != 0:
+                return True
+            else:
+                return False
+
+        if is_auth_message():
+            self._process_auth_message(message)
+        elif is_error_message():
+            self._process_error_message(message)
+        else:
+            callback_function = self._pop_callback(message["reqId"])
+            callback_function(message)
+
+    def _set_callback(self, topic, callback_function):
+        self.callback_directory[topic] = callback_function
+
+    def _set_error_callback(self, topic, callback_function):
+        self.error_callback_directory[topic] = callback_function
+
+    def _pop_callback(self, topic):
+        return self.callback_directory.pop(topic)
+
+    def _pop_error_callback(self, topic):
+        return self.error_callback_directory.pop(topic)
+
+    def _send_order_operation(self, operation, callback, request, error_callback=None):
+        request_id = str(uuid.uuid4())
+
+        message = {
+            "reqId": request_id,
+            "header": {
+                "X-BAPI-TIMESTAMP": _helpers.generate_timestamp(),
+            },
+            "op": operation,
+            "args": [
+                request
+            ],
+        }
+
+        if self.recv_window:
+            message["header"]["X-BAPI-RECV-WINDOW"] = self.recv_window
+        if self.referral_id:
+            message["header"]["Referer"] = self.referral_id
+
+        self.ws.send(json.dumps(message))
+        self._set_callback(request_id, callback)
+        if error_callback:
+            self._set_error_callback(request_id, error_callback)

src/lib_python_interop/python_interop_plugins/spark_core.py

@@ -21,6 +21,8 @@ from python_interop_module import (
     PySpot,
 )
 
+FLOAT_RE = r"[+-]?(?:(?:\d+\.\d*)|(?:\.\d+)|(?:\d+))(?:[eE][+-]?\d+)?"
+
 
 class JsonElement:
     def __init__(self, value):
@@ -217,6 +219,9 @@ def get_tick_factors(tick_sz: float):
     """
     if not (tick_sz > 0):
         raise ValueError(f"Invalid tick size {tick_sz}")
+    if tick_sz > 1:
+        # integer tick
+        return (tick_sz, 1)
     tick_denom = 1.0 / tick_sz
     match tick_denom:
         case 1:
@@ -256,7 +261,15 @@ def get_tick_factors(tick_sz: float):
         case 20_000:
             return (5, 100_000)
         case _:
-            raise ValueError(f"Invalid tick size {tick_sz}")
+            tick_base = min(int(1 / tick_sz), 1000000) if tick_sz != 0 else 10
+            tick_base_log10 = math.log10(tick_base)
+            tick_base_log10_ceil = math.ceil(tick_base_log10)
+            tick_base_log10_floor = math.floor(tick_base_log10)
+            if (tick_base_log10_ceil - tick_base_log10 <= 0.1):
+                tick_base = int(math.pow(10, tick_base_log10_ceil))
+            elif (tick_base - tick_base_log10_floor >= 0.9):
+                tick_base = int(math.pow(10, tick_base_log10_floor))
+            return (1, tick_base)
 
 
 def get_num_decimal_places(tick_sz: float):
@@ -325,7 +338,7 @@ def get_quantity_info(
     max_price: Optional[float] = None,
 ):
     # Use 'not' checks to account for nans (for which every op evaluates to false)
-    if not (tick_unit > 0) or not (tick_base >= tick_unit):
+    if not (tick_unit > 0):
         raise ValueError(f"Invalid tick factors {tick_unit}/{tick_base}")
     if not min_value:
         min_value = tick_unit / tick_base

src/lib_python_interop/python_interop_plugins/spark_enums.py

@@ -17,3 +17,35 @@ class ExecType(Enum):
     FILL = 4
     CANCELED = 5
     REJECTED = 6
+
+
+class OrderType(Enum):
+    LIMIT = 1
+    MARKET = 2
+    ICEBERG = 3
+    SPREADER = 4
+
+    @staticmethod
+    def parse(s: str):
+        return OrderType.__members__[s]
+
+
+class TimeInForce(Enum):
+    FOK = 1
+    GTC = 2
+    IOC = 3
+    DAY = 4
+
+    @staticmethod
+    def parse(s: str):
+        return TimeInForce.__members__[s]
+
+
+class SecurityType(Enum):
+    SPOT = 1
+    FUTURE = 2
+    OPTION = 3
+
+    @staticmethod
+    def parse(s: str):
+        return SecurityType.__members__[s]

src/lib_python_interop/python_interop_plugins/trade_client_cli.py

@@ -0,0 +1,334 @@
+import argparse
+import logging
+import re
+import sys
+import time
+import traceback
+from rich.console import Console
+from typing import Optional
+
+from bybit_execution_gateway import BybitExecutionGateway
+from prompt_toolkit import PromptSession
+from prompt_toolkit.completion import Completer, Completion
+from prompt_toolkit.patch_stdout import patch_stdout
+from python_interop_module import (
+    ExecutionGateway,
+    MdProvider,
+    PyBookLevel,
+    PyBookList,
+    PyCreateOrderRequest,
+)
+from console_table import ConsoleTable, ConsoleTableColumn
+from fake_execution_gateway_bridge import FakeExecutionGatewayBridge
+from fake_md_bridge import FakeMdBridge
+from bybit_md_provider import BybitMdProvider
+from spark_core import (
+    ConsoleLogFormatter,
+    FLOAT_RE,
+)
+
+# Simulated trading data
+symbols = ["BTCUSD", "ETHUSD", "AAPL", "GOOG"]
+orders = ["12345", "67890", "24680"]
+
+
+class TradingCompleter(Completer):
+    def get_completions(self, document, complete_event):
+        words = document.text_before_cursor.strip().split()
+        if not words:
+            # First word: suggest top-level commands
+            for cmd in ["buy", "sell", "cancel", "status", "exit"]:
+                yield Completion(cmd, start_position=0)
+        elif words[0] in ["buy", "sell"]:
+            if len(words) == 1:
+                # After 'buy' or 'sell', suggest symbols
+                for sym in symbols:
+                    yield Completion(sym, start_position=0)
+            elif len(words) == 2:
+                # Suggest example quantities
+                for qty in ["0.1", "1.0", "10.0"]:
+                    yield Completion(qty, start_position=0)
+        elif words[0] == "cancel":
+            if len(words) == 1:
+                for oid in orders:
+                    yield Completion(oid, start_position=0)
+
+
+# Logging setup
+class PromptToolkitHandler(logging.Handler):
+    def emit(self, record):
+        from prompt_toolkit.shortcuts import print_formatted_text
+
+        msg = self.format(record)
+        print_formatted_text(msg)
+
+
+def trading_client(logger):
+    i = 0
+    while True:
+        time.sleep(1.5)
+        logger.info(f"[MarketData] Tick {i}")
+        i += 1
+
+
+# TRADE_REQ_RE = re.compile(f"\\[a-zA-Z0-9-\\](buy|sell) ([a-zA-Z]{{3,8}}) ({FLOAT_RE}) @ ({FLOAT_RE})")
+CREATE_ORDER_RE = re.compile(
+    f"(buy|sell) ([a-zA-Z]{{3,8}}) ({FLOAT_RE}) @ ({FLOAT_RE})"
+)
+CHANGE_ORDER_PRICE_RE = re.compile(
+    f"(ch|change|modify) ([a-zA-Z0-9]+) (p|px|price) to ({FLOAT_RE})"
+)
+CHANGE_ORDER_QUANTITY_RE = re.compile(
+    f"(ch|change|modify) ([a-zA-Z0-9]+) (q|qty|quantity) to ({FLOAT_RE})"
+)
+CHANGE_ORDER_RE = re.compile(
+    f"(ch|change|modify) ([a-zA-Z0-9]+) to ({FLOAT_RE}) @ ({FLOAT_RE})"
+)
+CANCEL_SYMBOL_ORDERS_RE = re.compile("(cancel|cxl) all ([a-zA-Z0-9]+)")
+CANCEL_ORDER_RE = re.compile("(cancel|cxl) ([a-zA-Z0-9]+)")
+
+
+class OrderReqParser:
+    def __init__(self, venue: str):
+        self.id_counter = 1
+        self.venue = venue
+        pass
+
+    def parse_create_req(self, input_str: str):
+        m = re.match(CREATE_ORDER_RE, input_str)
+        if not m:
+            return None
+        grps = m.groups()
+        if len(grps) != 4:
+            return None
+        side, sec, qty_s, px_s = grps
+        qty = float(qty_s)
+        px = float(px_s)
+        req = PyCreateOrderRequest()
+        req.set_exchange_security_id(sec)
+        req.set_is_buy(side.startswith("b"))
+        req.set_key(self.id_counter)
+        req.set_order_type("LIMIT")
+        req.set_price(px)
+        req.set_quantity(qty)
+        req.set_time_in_force("GTC")
+        req.set_venue(self.venue)
+        self.id_counter += 1
+        return req
+
+    def parse_modify_req(self, input_str: str):
+        m = re.match(CHANGE_ORDER_RE, input_str)
+        if not m:
+            return None
+        grps = m.groups()
+        if len(grps) != 4:
+            return None
+
+
+class VenueService:
+    def __init__(
+        self,
+        venue: str,
+        logger: logging.Logger,
+        is_execution_enabled=False,
+        is_md_enabled=False,
+        is_md_ticker_enabled=False,
+    ):
+        self.venue = venue
+        self.logger = logger
+        self.md_provider: Optional[MdProvider] = None
+        self.execution_gw: Optional[ExecutionGateway] = None
+        if is_md_enabled:
+            match venue.lower():
+                case "bybit":
+                    self.md_provider = BybitMdProvider()
+                case _:
+                    pass
+        if is_execution_enabled:
+            match venue.lower():
+                case "bybit":
+                    self.execution_gw = BybitExecutionGateway()
+                case _:
+                    pass
+        if self.md_provider:
+            self.md_provider_bridge = (
+                BboSink(venue) if is_md_ticker_enabled else FakeMdBridge(logger)
+            )
+        if self.execution_gw:
+            self.execution_gw_bridge = FakeExecutionGatewayBridge(logger)
+
+    def create_order(self, req):
+        if self.execution_gw:
+            self.execution_gw.create_order(req)
+        else:
+            self.logger.warning("Execution not enabled for venue '%s'" % self.venue)
+
+    def start(self):
+        if self.md_provider:
+            self.md_provider.start(self.md_provider_bridge)  # type: ignore
+        if self.execution_gw:
+            self.execution_gw.start(self.execution_gw_bridge)  # type: ignore
+
+    def stop(self):
+        if self.md_provider:
+            self.md_provider.stop()
+        if self.execution_gw:
+            self.execution_gw.stop()
+
+    def __enter__(self):
+        self.start()
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        self.stop()
+
+
+class BboSink:
+    def __init__(self, name: str):
+        self.name = name
+        self.bbo = {}
+
+    def on_books(self, bks: PyBookList):
+        for bk in bks:
+            bids = bk.bids()
+            asks = bk.asks()
+            b = bids.get_level(0) if len(bids) > 0 else PyBookLevel()
+            a = asks.get_level(0) if len(asks) > 0 else PyBookLevel()
+            self.bbo[bk.exchange_security_id()] = b, a
+
+    def on_securities(self, v):
+        pass
+
+    def on_trades(self, v):
+        pass
+
+
+def run(
+    venues: list[str],
+    is_md_enabled=False,
+    is_execution_enabled=False,
+    is_md_ticker_enabled=False,
+):
+    logger = logging.getLogger()
+    # In md ticker mode, we print bbo to a static table in the console. We only log errors in
+    # this mode to reduce disruptions in the table.
+    logger.setLevel(logging.ERROR if is_md_ticker_enabled else logging.INFO)
+    handler = PromptToolkitHandler()
+    handler.setFormatter(ConsoleLogFormatter("Gen"))
+    logger.addHandler(handler)
+
+    svcs = {
+        v: VenueService(
+            v,
+            logger,
+            is_execution_enabled=is_execution_enabled,
+            is_md_enabled=is_md_enabled,
+            is_md_ticker_enabled=is_md_ticker_enabled,
+        )
+        for v in venues
+    }
+
+    for _, svc in svcs.items():
+        svc.start()
+
+    if is_execution_enabled:
+        session = PromptSession(
+            completer=TradingCompleter(),
+            bottom_toolbar=lambda: "Type a command like 'buy BTCUSD 1.0' or 'cancel <id>'",
+        )
+
+        def parse_req(cmd):
+            try:
+                req = req_parser.parse_create_req(cmd)
+                if req is None:
+                    logger.error("Failed to parse OrderReq")
+                return req
+            except Exception:
+                logger.error("[UserInputParseError] %s", exc_info=True)
+            return None
+
+        with patch_stdout():
+            req_parser = OrderReqParser("BYBIT")
+            while True:
+                try:
+                    cmd = session.prompt("> ")
+                    if cmd.strip().lower() == "exit":
+                        break
+                    req = parse_req(cmd)
+                    if not req:
+                        continue
+                    svc = svcs.get(req.venue().lower())
+                    if svc:
+                        svc.create_order(req)
+
+                except KeyboardInterrupt:
+                    # Allows Ctrl+C to cancel current input
+                    continue
+                except EOFError:
+                    # Allows Ctrl+D to exit
+                    break
+    else:
+        ctable = ConsoleTable(
+            [
+                ConsoleTableColumn("symbol", f"{'Symbol':<10}"),
+                ConsoleTableColumn("bid", f"{'Bid':>10}", data_format=">10"),
+                ConsoleTableColumn("ask", f"{'Ask':>10}", data_format=">10"),
+            ]
+        )
+        ctable.clear_screen()
+        ctable.hide_cursor()
+        console = Console()
+        with console.screen():
+            while True:
+                curr_data = {}
+                for _, svc in svcs.items():
+                    if isinstance(svc.md_provider_bridge, BboSink):
+                        curr_data = [
+                            {
+                                "symbol": sym,
+                                "bid": bbo[0].price(),
+                                "ask": bbo[1].price(),
+                            }
+                            for sym, bbo in svc.md_provider_bridge.bbo.items()
+                        ]
+                        ctable.move_cursor(1, 1)
+                        ctable.print_table(curr_data)
+                sys.stdout.flush()
+                time.sleep(1)
+
+
+def _get_arg_parser():
+    p = argparse.ArgumentParser(description="Run python md and execution plugins")
+    p.add_argument(
+        "--md", dest="is_md_enabled", action="store_true", help="Run md plugin(s)"
+    )
+    p.add_argument(
+        "--md-ticker",
+        dest="is_md_ticker_enabled",
+        action="store_true",
+        help="Run md plugin(s), display ticker data in table",
+    )
+    p.add_argument(
+        "--exe",
+        dest="is_execution_enabled",
+        action="store_true",
+        help="Run execution gateway plugin(s)",
+    )
+    p.add_argument(
+        "venues", nargs="+", help="Venues to run (e.g., 'Bybit', 'Gateio', etc.)"
+    )
+    p.set_defaults(
+        venues=[],
+        is_md_enabled=False,
+        is_execution_enabled=False,
+        is_md_ticker_enabled=False,
+    )
+    return p
+
+
+if __name__ == "__main__":
+    args = _get_arg_parser().parse_args()
+    try:
+        run(**vars(args))
+    except Exception:
+        traceback.print_exc()

src/lib_spark/spark/domain/SecurityCommon.h

@@ -30,6 +30,7 @@ class SecurityCommon {
    double min_quantity_increment = 0;
    double min_quantity = 0;
    double max_quantity = std::numeric_limits<double>::max();
+   double min_notional_value = std::numeric_limits<double>::quiet_NaN();
    double cash_per_point = 1;
 
    SecurityCommon() {}

src/lib_spark/spark/domain/tick_utils.cpp

@@ -6,6 +6,12 @@ namespace bts::domain {
 
 template <concepts::FloatingPoint T>
 bool get_tick_factors(double tick_sz, T &tick_unit, T &tick_base) {
+   if (tick_sz > 1) {
+      // integer tick
+      tick_unit = tick_sz;
+      tick_base = 1;
+      return true;
+   }
    float qty_tick_denom = 1.0 / tick_sz;
    int qty_tick_denomi = static_cast<int>(qty_tick_denom);
    if (qty_tick_denom != qty_tick_denomi) {

src/lib_spark/spark/domain_model/SecurityType.h

@@ -16,6 +16,7 @@ namespace bts::api {
  * insensitive, and makes the 'SECURITYTYPE_' prefix optional.
  */
 bool security_type_parse_smart(const std::string &str, SecurityType *result);
+std::string security_type_name_pretty(SecurityType type);
 
 PB_ENUM_FROM_JSON(SecurityType)
 PB_ENUM_TO_JSON(SecurityType)
@@ -42,6 +43,14 @@ inline bool security_type_parse_smart(
    return SecurityType_Parse(s, result);
 }
 
+inline std::string security_type_name_pretty(SecurityType type) {
+   auto name = SecurityType_Name(type);
+   if (name.starts_with("SECURITYTYPE_")) {
+      return name.substr(strlen("SECURITYTYPE_"));
+   }
+   return name;
+}
+
 inline std::ostream &operator<<(
     std::ostream &out, const SecurityType &type) noexcept {
    out << SecurityType_Name(type);

src/lib_spark/spark/md/GeneralMarketDataListener.cpp

@@ -842,6 +842,16 @@ void GeneralMarketDataListener::_copy_security_common(
           min_qty, max_qty, qty_tick_unit, qty_tick_base,
           std::numeric_limits<float>::max(), false);
    }
+   if (!std::isnan(sc.min_notional_value)) {
+      float price_tick_unit;
+      float price_tick_base;
+      domain::get_tick_factors(
+          sc.min_price_increment, price_tick_unit, price_tick_base);
+      to.add_quantity_info(
+          sc.min_notional_value, std::numeric_limits<float>::max(),
+          price_tick_unit, price_tick_base, std::numeric_limits<float>::max(),
+          true);
+   }
 }
 
 uint64_t GeneralMarketDataListener::_generate_security_key(

src/lib_spark/spark/orderman/CancelOrderMsg.cpp

@@ -7,6 +7,12 @@ api::Venue CancelOrderMsg::venue() const {
    return security == nullptr ? api::VENUE_UNSET : security->exchange();
 }
 
+api::SecurityType CancelOrderMsg::security_type() const {
+   return security == nullptr
+              ? api::SECURITYTYPE_UNSET
+              : bts::securities::security_type(security->Contract_case());
+}
+
 std::uint32_t CancelOrderMsg::security_id() const {
    return security == nullptr ? 0 : security->id();
 }

src/lib_spark/spark/orderman/CancelOrderMsg.h

@@ -17,6 +17,7 @@ struct CancelOrderMsg {
    api::Side side;
 
    api::Venue venue() const;
+   api::SecurityType security_type() const;
    std::uint32_t security_id() const;
    const std::string &security_name() const;
 };

src/lib_spark/spark/orderman/CreateOrderMsg.cpp

@@ -13,6 +13,12 @@ api::Venue CreateOrderMsg::venue() const {
    return security == nullptr ? api::VENUE_UNSET : security->exchange();
 }
 
+api::SecurityType CreateOrderMsg::security_type() const {
+   return security == nullptr
+              ? api::SECURITYTYPE_UNSET
+              : bts::securities::security_type(security->Contract_case());
+}
+
 std::uint32_t CreateOrderMsg::security_id() const {
    return security == nullptr ? 0 : security->id();
 }

src/lib_spark/spark/orderman/CreateOrderMsg.h

@@ -24,6 +24,7 @@ struct CreateOrderMsg {
    api::orderman::RoutingRecord const *routing_record;
 
    api::Venue venue() const;
+   api::SecurityType security_type() const;
    std::uint32_t security_id() const;
    const std::string &security_name() const;
 };

src/lib_spark/spark/orderman/actors/ServiceBridge.cpp

@@ -396,19 +396,15 @@ void ServiceBridge::do_work() {
    });
 }
 
-std::optional<FilledPositionEvent> make_pos_event(
-    api::position::PositionUnit const &pu, securities::MapSecurityDB const &o) {
-   auto sec = o.resolve(SecurityKey(pu.security_id()));
-   if (sec) {
-      uint32_t total_long_pos, total_short_pos;
-      bts::position::PositionUtilities::total_position(
-          pu, total_long_pos, total_short_pos);
-      FilledPositionEvent ev{
-          *sec, AccountID(pu.account_id()), Quantity(total_long_pos),
-          Quantity(total_short_pos)};
-      return {ev};
-   } else
-      return {};
+static FilledPositionEvent make_pos_event(
+    api::position::PositionUnit const &pu) {
+   uint32_t total_long_pos, total_short_pos;
+   bts::position::PositionUtilities::total_position(
+       pu, total_long_pos, total_short_pos);
+   FilledPositionEvent ev{
+       pu.security_id(), AccountID(pu.account_id()), Quantity(total_long_pos),
+       Quantity(total_short_pos)};
+   return {ev};
 }
 
 void ServiceBridge::update_position(
@@ -420,25 +416,11 @@ void ServiceBridge::update_position(
        pos.all().size());
    std::vector<FilledPositionEvent> evs;
    for (auto const &pu : pos.added()) {
-      auto ev = make_pos_event(*pu, me->sec_db);
-      if (ev.has_value())
-         evs.push_back(ev.value());
-      else
-         LOG(OrderMan, Error,
-             "Could not update position limiter for unknown security {}. This "
-             "should not happen",
-             pu->security_id());
+      evs.push_back(make_pos_event(*pu));
    }
 
    for (auto const &pu_change : pos.changed()) {
-      auto ev = make_pos_event(*pu_change.current(), me->sec_db);
-      if (ev.has_value())
-         evs.push_back(ev.value());
-      else
-         LOG(OrderMan, Error,
-             "Could not update position limiter for unknown security {}. This "
-             "should not happen",
-             pu_change.current()->security_id());
+      evs.push_back(make_pos_event(*pu_change.current()));
    }
 
    me->service_call_marshaller.schedule_action(

src/lib_spark/spark/orderman/actors/order_manager/OrderManager.cpp

@@ -267,7 +267,7 @@ void OrderManager::on_filled_position(
    for (auto const &ev : *(update.events)) {
       me->services_data.position.update(ev);
       me->services_data.limiter.set_filled_position(
-          *ev.security, ev.account_id, ev.long_quantity, ev.short_quantity);
+          *ev.security_id, ev.account_id, ev.long_quantity, ev.short_quantity);
    }
 }
 

src/lib_spark/spark/orderman/actors/order_manager/PositionIndex.cpp

@@ -13,14 +13,13 @@ struct PositionIndex::PImpl {
 };
 
 void PositionIndex::update(FilledPositionEvent const &ev) {
-   auto &idx =
-       me->position_idx[{SecurityKey(ev.security->id()), ev.account_id}];
+   auto &idx = me->position_idx[{ev.security_id, ev.account_id}];
    if (idx == 0) {
       me->current_positions.push_back({});
       idx = me->current_positions.size();
    }
    me->current_positions[idx - 1] = {
-       ev.security, ev.account_id, ev.long_quantity, ev.short_quantity};
+       *ev.security_id, ev.account_id, ev.long_quantity, ev.short_quantity};
 }
 
 std::vector<PositionUpdate> const &PositionIndex::current_positions() const {

src/lib_spark/spark/orderman/events/FilledPositionEvent.h

@@ -14,21 +14,21 @@ class FilledPositionEvent : public mercury::Event<FilledPositionEvent> {
    using SynopsisPolicy = mercury::MergeByKey<FilledPositionEvent>;
 
    FilledPositionEvent(
-       api::Security const &sec,
+       std::uint32_t security_id,
        AccountID id,
        Quantity long_quantity,
        Quantity short_quantity)
-       : security(&sec), account_id(id), long_quantity(long_quantity),
+       : security_id(security_id), account_id(id), long_quantity(long_quantity),
          short_quantity(short_quantity) {}
 
    uint64_t key() const {
-      uint64_t result = security->id();
+      uint64_t result = *security_id;
       result <<= 32;
       result += account_id.value();
       return result;
    }
 
-   api::Security const *security;
+   SecurityKey security_id;
    AccountID account_id;
    Quantity long_quantity;
    Quantity short_quantity;

src/lib_spark/spark/orderman/limiter/LimitStore.h

@@ -141,11 +141,11 @@ class LimitStore {
       }
    }
    void set_filled_position(
-       const api::Security &security,
+       std::uint32_t security_id,
        AccountID account_id,
        Quantity long_pos,
        Quantity short_pos) {
-      SecurityLimitTree *sec_tree = _find_security(security.id());
+      SecurityLimitTree *sec_tree = _find_security(security_id);
       bts_assert(sec_tree != nullptr);
       sec_tree->update_position(account_id, long_pos, short_pos);
    }

src/lib_spark/spark/orderman/limiter/Limiter.cpp

@@ -78,7 +78,7 @@ void Limiter::add_security(api::Security const &security) {
    _reset(me->limit_store->admin_data());
 }
 void Limiter::set_filled_position(
-    api::Security const &security,
+    std::uint32_t security_id,
     AccountID account_id,
     Quantity long_pos,
     Quantity short_pos) {
@@ -86,17 +86,17 @@ void Limiter::set_filled_position(
    // add_securities was called or before add was called
    // (we can put it in a queue and examine it on the
    // add_securities call)
-   if (!_contains_security(&security))
+   if (!_contains_security(security_id))
       LOG(OrderMan, Error,
           "Limiter::set_filled_position could not find sec_key:{} this should "
           "be investigated",
-          security.id());
+          security_id);
    else {
       if (!me->limit_store->has_account(account_id))
          // add a temporary account, so trades can go in there
          _add_account_and_reset(account_id);
       me->limit_store->set_filled_position(
-          security, account_id, long_pos, short_pos);
+          security_id, account_id, long_pos, short_pos);
    }
 }
 

src/lib_spark/spark/orderman/limiter/Limiter.h

@@ -37,7 +37,10 @@ class Limiter {
    // not recursively deconstructed into their legs, i.e this sets the
    // position only for the passed security.
    void set_filled_position(
-       api::Security const &, AccountID, Quantity long_pos, Quantity short_pos);
+       std::uint32_t security_id,
+       AccountID account_id,
+       Quantity long_pos,
+       Quantity short_pos);
 
    // Increment/decrement the working position, i.e. the quantity open
    // in the market, recursively updating leg positions in the case of

src/lib_spark/spark/orderman/orders/algo/AlgoWrapperOrder.cpp

@@ -569,7 +569,8 @@ void AlgoWrapperOrder::subscribe_position() {
    me->services->subscribe_to_position_updates([this](auto const &ev) {
       ReportStateGuard _(*this);
       me->algo->on_position(
-          {ev.security, ev.account_id, ev.long_quantity, ev.short_quantity});
+          {*ev.security_id, ev.account_id, ev.long_quantity,
+           ev.short_quantity});
    });
 }
 

src/lib_spark/spark/orderman/orders/algo/PositionUpdate.cpp

@@ -5,7 +5,7 @@ namespace bts {
 namespace orderman {
 
 std::ostream &operator<<(std::ostream &out, const PositionUpdate &update) {
-   out << "Security:  " << update.security->id()
+   out << "Security:  " << update.security_id
        << " Account: " << update.account_id.value()
        << " Long Qty: " << update.long_quantity.value()
        << " Short Qty: " << update.short_quantity.value();

src/lib_spark/spark/orderman/orders/algo/PositionUpdate.h

@@ -11,7 +11,7 @@ class Security;
 namespace orderman {
 
 struct PositionUpdate {
-   api::Security const *security;
+   std::uint32_t security_id;
    AccountID account_id;
    Quantity long_quantity;
    Quantity short_quantity;

src/lib_spark/spark/orderman/orders/algo/algo_orders/TaskingEyeAlgo.cpp

@@ -331,7 +331,7 @@ void TaskingEyeAlgo::on_theos(
 
 void TaskingEyeAlgo::on_position(PositionUpdate const &pu) {
    LOG(OrderMan, Info, "TaskingEyeAlgo::on_position order_key: {} sec_key: {}",
-       me->service_ptr->order_key(), pu.security->id());
+       me->service_ptr->order_key(), pu.security_id);
 }
 
 void TaskingEyeAlgo::_evaluate_and_shoot() {

tools/fish/envsource.fish

@@ -3,7 +3,14 @@
 function envsource --description "Source .env file"
   for line in (cat $argv | grep -v '^#')
     set item (string split -m 1 '=' $line)
-    set -gx $item[1] $item[2]
-    echo "Exported key $item[1]"
+    if string match -rq 'PATH$' $item[1]
+        # Treat vars that end in "PATH" as semi-colon delimited lists.
+        set values (string split ';' $item[2])
+        set -gx $item[1] $values
+        echo "Exported key $item[1] ("(count $values)" values delimited by ';')"
+    else
+        set -gx $item[1] $item[2]
+        echo "Exported key $item[1]"
+    end
   end
 end