Fixing PNL for both crypto and non crypto

So it doesn't need to be hardcoded per release

.devcontainer/devcontainer.json

@@ -1,8 +1,8 @@
 {
   "dockerFile": "Dockerfile",
   "hostRequirements": {
-    "memory": "4gb",
-    "storage": "6gb"
+    "memory": "8gb",
+    "storage": "10gb"
   },
   "containerUser": "bts",
   "updateRemoteUserUID": true,
@@ -12,7 +12,7 @@
   // example, if 8000-8003 is taken, go with 9000-9003, then 10000-10003, etc.
   "runArgs": [
     "--shm-size",
-    "2gb",
+    "8gb",
     "-p",
     "8000:8000",
     "-p",
@@ -30,7 +30,8 @@
         "eamodio.gitlens",
         "llvm-vs-code-extensions.vscode-clangd",
         "ms-vscode.cmake-tools",
-        "ms-vscode.cpptools"
+        "ms-vscode.cpptools",
+        "xaver.clang-format"
       ]
     }
   }

src/client/spark/Daemons/MDDaemon.cs

@@ -133,17 +133,31 @@ namespace bts {
 
             // set 'unit quantity' for each sec
             foreach (var smd in exch_secs.Values) {
-               var meta = _security_metadata.get(smd.security);
-               if (meta.HasValue) {
-                  exch_secs[smd.id] = new(
-                     smd.security,
-                     smd.md,
-                     meta.Value.HasPriceDisplay ? meta.Value.PriceDisplay : null,
-                     meta.Value.HasQuantityDisplay ? meta.Value.QuantityDisplay : null);
-               }
+               var (priceDisplay, qtyDisplay) = GetDisplaySettings(smd.security);
+               exch_secs[smd.id] = new(smd.security, smd.md, priceDisplay, qtyDisplay);
             }
             _bus.post(SparkTags.ServerSecuritiesTag, exch_secs);
          }
+
+         (TickDisplaySettingsImmutable? Price, TickDisplaySettingsImmutable? Quantity) GetDisplaySettings(SecurityImmutable sec) {
+            var meta = _security_metadata.get(sec);
+            var priceDisplay = meta?.PriceDisplay ?? Security.Defaults.price_display_settingsi;
+            var qtyDisplay = meta?.QuantityDisplay ?? Security.Defaults.quantity_display_settingsi;
+            if (sec.HasContractSpec) {
+               var pd = priceDisplay.CloneToMutable();
+               pd.TickFactors = TickUtils.get_tick(sec, 0).to_api_tick_factors();
+               priceDisplay = new(pd);
+               if (sec.ContractSpec.QuantityInfos.TryGetFirst(out var qtyInfo)) {
+                  var qd = qtyDisplay.CloneToMutable();
+                  qd.ShowDecimalPart = qtyInfo.TickBase.is_real_non_zero() && !(qtyInfo.TickUnit / qtyInfo.TickBase).is_int();
+                  qd.TickFactors ??= new();
+                  qd.TickFactors.TickBase = qtyInfo.TickBase;
+                  qd.TickFactors.TickUnit = qtyInfo.TickUnit;
+                  qtyDisplay = new(qd);
+               }
+            }
+            return (priceDisplay, qtyDisplay);
+         }
       }
 
       public void Dispose() =>

src/client/spark/Risk/PNL/PNLTradeLogRow.cs

@@ -4,6 +4,9 @@ using bts.api;
 
 using BCG;
 using mbts.co2;
+using System.Xml.Serialization;
+using ZstdSharp.Unsafe;
+using bts.proto;
 
 namespace mbts.analytics {
    // Pairs ITradeLogRow along with the data (theo, md) that is needed to compute
@@ -18,20 +21,87 @@ namespace mbts.analytics {
 
       public SecurityImmutable security { get { return md.security; } }
 
-      private double _option_pnl(double trade_price, double now_price) {
-         double coin_terms = (now_price - trade_price) * tl_row.size;
-         double usd_terms = coin_terms * driver;
-         return usd_terms;
+      // private double _option_pnl(double trade_price, double now_price) {
+      //    double coin_terms = (now_price - trade_price) * tl_row.size;
+      //    double usd_terms = coin_terms * driver;
+      //    return usd_terms;
+      // }
+
+      private bool _is_notional {
+         get {
+            return false;
+         }
       }
 
-      private double _future_pnl(double trade_price, double now_price) {
-         // the qty is in USD, so this is the initial cash flow
-         double init_asset_value = tl_row.size;
-         double asset_size = init_asset_value / tl_row.price;
-         double now_asset_value = asset_size * now_price;
-         return now_asset_value - init_asset_value;
+      private bool _is_inverse_option_contract {
+         get {
+            return false;
+         }
       }
 
+      private double _cpp {
+         get {
+            return 1.0;
+         }
+      }
+
+      private bool _is_reverse_option_contract {
+         get {
+            // check for Deribit options
+            return false;
+         }
+      }
+
+      private double _option_pnl {
+         get {
+            if (_is_notional) {
+               var now_price = theo;
+               double coin_terms = (now_price - tl_row.price) * tl_row.size;
+               double usd_terms = coin_terms * driver;
+               return usd_terms;
+            } else {
+               return _cpp * tl_row.size * (theo - tl_row.price);
+            }
+         }
+      }
+
+      private double _spot_pnl {
+         get {
+            // both are computed the same way
+            return _future_pnl;
+         }
+      }
+
+      private double _future_pnl {
+         get {
+            if (_is_notional) {
+               // TODO:
+               // let it fall through right now
+               // the qty is in USD, so this is the initial cash flow
+               var now_price = theo;
+               double init_asset_value = tl_row.size;
+               double asset_size = init_asset_value / tl_row.price;
+               double now_asset_value = asset_size * now_price;
+               return now_asset_value - init_asset_value;
+            } else {
+               // TODO: is position pnl, compute it wrt settlement price?
+               return tl_row.size * (theo - tl_row.price);
+            }
+         }
+      }
+
+      // private double _future_pnl(double trade_price, double now_price) {
+      //    if (_is_notional) {
+      //       // the qty is in USD, so this is the initial cash flow
+      //       double init_asset_value = tl_row.size;
+      //       double asset_size = init_asset_value / tl_row.price;
+      //       double now_asset_value = asset_size * now_price;
+      //       return now_asset_value - init_asset_value;
+      //    } else {
+      //       return tl_row.size * (theo - tl_row.price);
+      //    }
+      // }
+
 
       public double driver {
          get {
@@ -46,11 +116,18 @@ namespace mbts.analytics {
       public double theo {
          get {
             switch (tl_row.security_type) {
+               // TODO:
                case SecurityType.Option:
-                  return tl_row.call_put == PutOrCall.Call ?
-                     opt_theo.fwd_normalized_call : opt_theo.fwd_normalized_put;
+                  if (_is_inverse_option_contract) {
+                     return tl_row.call_put == PutOrCall.Call ?
+                        opt_theo.fwd_normalized_call : opt_theo.fwd_normalized_put;
+                  } else {
+                     return tl_row.call_put == PutOrCall.Call ?
+                        opt_theo.call : opt_theo.put;
+                  }
+               // future and options, just get mid market
                case SecurityType.Future:
-                  return fut_theo.theo;
+               //return fut_theo.theo;
                case SecurityType.Spot:
                   var bbo = top_of_book;
                   return (bbo.Item1 + bbo.Item2) / 2;
@@ -62,28 +139,22 @@ namespace mbts.analytics {
       public double pnl {
          get {
             double rv = Double.NaN;
-            if (is_day_trade) {
-               if (tl_row.security_type == SecurityType.Future) {
-                  rv = _future_pnl(tl_row.price, theo);
-               } else if (tl_row.security_type == SecurityType.Option) {
-                  rv = _option_pnl(tl_row.price, theo);
-               } else if (tl_row.security_type == SecurityType.Spot) {
-                  rv = tl_row.size * (theo - tl_row.price);
-               }
-               return rv;
+            switch (tl_row.security_type) {
+               case SecurityType.Option:
+                  rv = _option_pnl;
+                  break;
+               case SecurityType.Future:
+                  rv = _future_pnl;
+                  break;
+               case SecurityType.Spot:
+                  // = tl_row.size * (theo - tl_row.price);
+                  rv = _spot_pnl;
+                  break;
             }
-            // use settle price ?
-            if (tl_row.security_type == SecurityType.Future) {
-               rv = _future_pnl(tl_row.price, theo);
-            } else if (tl_row.security_type == SecurityType.Option) {
-               rv = _option_pnl(tl_row.price, theo);
-            } else if (tl_row.security_type == SecurityType.Spot) {
-               rv = tl_row.size * tl_row.price - settle_price;
-            }
-
             return rv;
          }
       }
+
       public double settle_price {
          get {
             return md.md == null ? Double.NaN : md.md.settlement_price;

src/client/spark_core/utils_api/BTS.cs

@@ -1,5 +1,5 @@
 using System;
-using System.Collections.Generic;
+using System.Reflection;
 using System.Text;
 using System.Threading;
 
@@ -10,9 +10,9 @@ namespace bts
       static int _needs_init_spark_lib = 0;
 
       public static void init(string config_source = null) {
-
+         var version = Assembly.GetExecutingAssembly().GetName().Version;
          Console.WriteLine(
-            $"Configuring BTS v2.2.4-2 with {(config_source is null ? "defaults" : $"'{config_source}'")}");
+            $"Configuring Spark {version} with {(config_source is null ? "defaults" : $"'{config_source}'")}");
          var args = Environment.GetCommandLineArgs();
          utils.BTSNativeMethods.provide_command_line(args, args.Length);
 

src/lib/config/profiles/default/okx_exe_cfg.conf

@@ -0,0 +1,17 @@
+OkxExeCfg {
+    value{
+        account : "ACC A"
+        ops_id : "op_id_bts"
+        secret_key : "DAF6CBB09475D2520B88D2070788A976"
+        api_key : "4492a9ab-5784-454c-acc7-38dc2a503ed0"
+        passphrase : "Pakistan@123"
+        ws_url : "wss://wspap.okx.com:8443"
+        rest_url : "https://www.okx.com"
+        ws_ping_timeout_seconds : 10 
+        ws_ping_interval_seconds : 25
+        ws_reconnect_interval_seconds : 5
+        ws_reconnect_max_count : 5
+        simulated_trading : true
+        tdmode : "cross"
+    }
+}

src/lib/config/protos/okx_exe_cfg.proto

@@ -0,0 +1,18 @@
+syntax = "proto3";
+package bts.config;
+message OkxExeCfg {
+    string account = 1;
+    string ops_id = 2;
+    string secret_key = 3;
+    string api_key = 4;
+    string passphrase = 5;
+    string ws_url = 6;
+    string rest_url = 7;
+    uint32 ws_ping_timeout_seconds = 8;
+    uint32 ws_ping_interval_seconds = 9;
+    uint32 ws_reconnect_interval_seconds = 10;
+    uint32 ws_reconnect_max_count = 11;
+    bool simulated_trading = 12;
+    string tdmode = 13;
+
+}

src/lib/io/http/http_client.cpp

@@ -34,7 +34,31 @@ static size_t write_headers(
     void *ptr, size_t size, size_t nmemb, IHttpResponse *data) {
    return data->write_headers((char *)ptr, size, nmemb);
 }
-
+// static int curl_debug_callback(CURL *handle, curl_infotype type, char *data, size_t size, void *userptr) {
+//    (void)handle;
+//    (void)type;
+//    (void)userptr;
+//     switch (type) {
+//         case CURLINFO_TEXT:
+//             LOG(Http, Info, "CURLINFO_TEXT: {}", std::string(data, size));
+//             break;
+//         case CURLINFO_HEADER_IN:
+//             LOG(Http, Info, "Header In: {}", std::string(data, size));
+//             break;
+//         case CURLINFO_HEADER_OUT:
+//             LOG(Http, Info, "Header Out: {}", std::string(data, size));
+//             break;
+//         case CURLINFO_DATA_IN:
+//             LOG(Http, Info, "Data In: {} bytes, in: {}", size,std::string(data, size));
+//             break;
+//         case CURLINFO_DATA_OUT:
+//             LOG(Http, Info, "Data Out: {} bytes, out: {}", size, std::string(data, size));
+//             break;
+//         default:
+//             break;
+//     }
+//     return 0;
+// }
 void HttpClient::get(
     string url,
     unordered_map<string, string> headers,
@@ -91,6 +115,8 @@ void HttpClient::send(const HttpRequest &req, IHttpResponse *response) {
    CURL *curl = session.curl;
    CurlHeaders curl_headers{req.headers};
    // Prepare req
+   //to debug curl requests
+   // curl_easy_setopt(curl, CURLOPT_DEBUGFUNCTION, curl_debug_callback);
    curl_easy_setopt(curl, CURLOPT_FAILONERROR, true);
    curl_easy_setopt(curl, CURLOPT_URL, req.url.c_str());
    curl_easy_setopt(curl, CURLOPT_HTTPHEADER, curl_headers.data());
@@ -108,6 +134,7 @@ void HttpClient::send(const HttpRequest &req, IHttpResponse *response) {
       curl_easy_setopt(curl, CURLOPT_HTTPPOST, 1L);
       curl_easy_setopt(curl, CURLOPT_POSTFIELDS, req.body.c_str());
       curl_easy_setopt(curl, CURLOPT_POSTFIELDSIZE, req.body.size());
+      // curl_easy_setopt(curl, CURLOPT_VERBOSE, 1L);
       break;
    case io::http::HttpMethod::PUT:
       curl_easy_setopt(curl, CURLOPT_CUSTOMREQUEST, "PUT");

src/lib/io/websocket/websocket_client.cpp

@@ -8,6 +8,7 @@
 #include "lib/io/websocket/websocket_error_code.h"
 #include "lib/io/websocket/websocketpp_exception_formatter.h"
 #include "lib/logging/logging.h"
+#include "lib/utils/bts_assert.h"
 #include "lib/utils/formatters/exception_formatter.h"
 
 #include <atomic>
@@ -201,9 +202,10 @@ struct WebsocketClient::PImpl {
             }
          }
       } catch (const std::exception &ex) {
+         std::string str = ex.what();
          ERROR_FN(
              "connect_no_timeout. Failed to connect. {}; err: {}", snapshot(),
-             ex);
+             str);
          set_state(State::IDLE);
          // Don't wait for connect
          return;
@@ -236,11 +238,16 @@ struct WebsocketClient::PImpl {
          ERROR_FN(
              "io_thread. Websocket io threw websocketpp::exception. {}; "
              "err: {}",
-             snapshot(), ex);
+             snapshot(), std::string(ex.what()));
+             bts_assert_msgf(false, "io_thread. Websocket io threw websocketpp::exception. {}; "
+             "err: {}",
+               snapshot(), std::string(ex.what()));
       } catch (const std::exception &ex) {
          ERROR_FN(
              "io_thread. Websocket io threw std::exception. {}; err: {}",
-             snapshot(), ex);
+             snapshot(), std::string(ex.what()));
+         bts_assert_msgf(false, "io_thread. Websocket io threw std::exception. {}; err: {}",
+             snapshot(), std::string(ex.what()));
       } catch (...) {
          bts_assert_msgf(
              false,
@@ -300,6 +307,7 @@ struct WebsocketClient::PImpl {
       const auto ec = client.get_con_from_hdl(hdl)->get_ec();
       if (ec) {
          ERROR_FN("on_open. {}; err: {}", snapshot(), ec.message());
+         bts_assert(false);
       }
       switch (state.value) {
       case State::OPENING:

src/lib/logging/Sink.cpp

@@ -107,7 +107,7 @@ Sink file_sink(LoggingConfig const &cfg) {
 
    if (boost::filesystem::exists(path)) {
       auto sz = boost::filesystem::file_size(path);
-      if (sz > max_sz && false) {
+      if (sz > max_sz) {
          auto backup = boost::filesystem::path(path.string() + ".old");
          boost::filesystem::rename(path, backup);
       }
@@ -118,8 +118,7 @@ Sink file_sink(LoggingConfig const &cfg) {
    return [=](LogMsg const &m) {
       auto msg = f(m);
       bytes_written += msg.size();
-      if (bytes_written >= max_sz && false) {
-         return;
+      if (bytes_written >= max_sz) {
          bytes_written = 0;
          stream.close();
          auto backup = boost::filesystem::path(path.string() + ".old");

src/lib/md/GeneralMarketDataListener.cpp

@@ -682,6 +682,18 @@ static bool get_qty_info(
       qty_tick_unit = 1;
       qty_tick_base = 1000000;
       break;
+   case 10'000'000:
+      qty_tick_unit = 1;
+      qty_tick_base = 10'000'000;
+      break;
+   case 100'000'000:
+      qty_tick_unit = 1;
+      qty_tick_base = 100'000'000;
+      break;
+   case 1'000'000'000:
+      qty_tick_unit = 1;
+      qty_tick_base = 1'000'000'000;
+      break;
    case 4:
       qty_tick_unit = 25;
       qty_tick_base = 100;

src/lib/md/Shm/FixedSecurity.cpp

@@ -509,6 +509,91 @@ void convert_security(
    }
 }
 
+void to_json(nlohmann::json &j, const FixedSecurity &f) {
+   j = nlohmann::json{
+       {"key", f.key},
+       {"symbol", f.symbol()},
+       {"type", api::SecurityType_Name(f.type)},
+       {"venue", api::Venue_Name(f.exchange)},
+       {"alt_symbol", f.alt_symbol()},
+       {"cfi_code", f.cfi_code()},
+       {"security_desc", f.security_desc()},
+       {"platform_id", f.platform_id()},
+       {"display_factor", f.display_factor},
+       {"numeric_id", f.numeric_id},
+       {"cboe_class_key", f.cboe_class_key},
+       {"main_frac", (int)f.cme_main_fraction},
+       {"sub_frac", (int)f.cme_sub_fraction},
+       {"price_format", (int)f.cme_price_display_format},
+       {"market_segment", (int)f.cme_market_segment_id},
+
+       {"price_break", f.price_break},
+       {"low_tick", f.low_tick},
+       {"high_tick", f.high_tick},
+       {"cpp", f.cash_per_point},
+
+       {"ice_market_description", f.ice_market_description()},
+       {"ice_cleared_alias", f.ice_cleared_alias()},
+       {"ice_product_name", f.ice_product_name()},
+       {"ice_hub_alias", f.ice_hub_alias()},
+       {"ice_strip_name", f.ice_strip_name()},
+       {"mic_code", f.mic_code()},
+   };
+   // Qty Info
+   {
+      auto qty_infos_j = nlohmann::json::array();
+      float min_value, max_value, tick_unit, tick_base, max_price;
+      bool is_notional;
+      for (int8_t i = 0; f.get_quantity_info(
+               i, min_value, max_value, tick_unit, tick_base, max_price,
+               is_notional);
+           i++) {
+         auto qty_info_j = nlohmann::json::object();
+         qty_info_j["min_value"] = min_value;
+         qty_info_j["max_value"] = max_value;
+         qty_info_j["tick_unit"] = tick_unit;
+         qty_info_j["tick_base"] = tick_base;
+         qty_info_j["max_price"] = max_price;
+         qty_info_j["is_notional"] = is_notional;
+         qty_infos_j.push_back(std::move(qty_info_j));
+      }
+      j["qty_infos"] = std::move(qty_infos_j);
+   }
+
+   switch (f.type) {
+   case api::SECURITYTYPE_SPOT:
+      break;
+   case api::SECURITYTYPE_FUTURE: {
+      auto *fut = static_cast<const FixedFutureSecurity *>(&f);
+      core::TimeStamp exp;
+      core::TimeStamp::from_ymd(
+          fut->expiration_year, fut->expiration_month, fut->expiration_day,
+          &exp);
+      j["expiration"] = exp.to_iso_string();
+      break;
+   }
+   case api::SECURITYTYPE_OPTION: {
+      auto *opt = static_cast<const FixedOptionSecurity *>(&f);
+      core::TimeStamp exp;
+      core::TimeStamp::from_ymd(
+          opt->expiration_year, opt->expiration_month, opt->expiration_day,
+          &exp);
+      j["expiration"] = exp.to_iso_string();
+      j["strike"] = opt->strike;
+      j["callput"] = api::PutOrCall_Name(opt->call_put);
+      break;
+   }
+   case api::SECURITYTYPE_SPREAD: {
+      auto *spd = static_cast<const FixedSpreadSecurityXL *>(&f);
+      std::vector<FixedSpreadLeg> legs{spd->legs, spd->legs + spd->leg_count()};
+      j["legs"] = legs;
+      break;
+   }
+   default:
+      break;
+   }
+}
+
 } // namespace md
 } // namespace bts
 

src/lib/md/Shm/FixedSecurity.h

@@ -273,71 +273,6 @@ inline void to_json(nlohmann::json &j, const FixedSpreadLeg &f) {
    };
 }
 
-inline void to_json(nlohmann::json &j, const FixedSecurity &f) {
-   j = nlohmann::json{
-       {"key", f.key},
-       {"symbol", f.symbol()},
-       {"type", api::SecurityType_Name(f.type)},
-       {"venue", api::Venue_Name(f.exchange)},
-       {"alt_symbol", f.alt_symbol()},
-       {"cfi_code", f.cfi_code()},
-       {"security_desc", f.security_desc()},
-       {"platform_id", f.platform_id()},
-       {"display_factor", f.display_factor},
-       {"numeric_id", f.numeric_id},
-       {"cboe_class_key", f.cboe_class_key},
-       {"main_frac", (int)f.cme_main_fraction},
-       {"sub_frac", (int)f.cme_sub_fraction},
-       {"price_format", (int)f.cme_price_display_format},
-       {"market_segment", (int)f.cme_market_segment_id},
-
-       {"price_break", f.price_break},
-       {"low_tick", f.low_tick},
-       {"high_tick", f.high_tick},
-       {"cpp", f.cash_per_point},
-
-       {"ice_market_description", f.ice_market_description()},
-       {"ice_cleared_alias", f.ice_cleared_alias()},
-       {"ice_product_name", f.ice_product_name()},
-       {"ice_hub_alias", f.ice_hub_alias()},
-       {"ice_strip_name", f.ice_strip_name()},
-       {"mic_code", f.mic_code()},
-   };
-
-   switch (f.type) {
-   case api::SECURITYTYPE_SPOT:
-      break;
-   case api::SECURITYTYPE_FUTURE: {
-      auto *fut = static_cast<const FixedFutureSecurity *>(&f);
-      core::TimeStamp exp;
-      core::TimeStamp::from_ymd(
-          fut->expiration_year, fut->expiration_month, fut->expiration_day,
-          &exp);
-      j["expiration"] = exp.to_iso_string();
-      break;
-   }
-   case api::SECURITYTYPE_OPTION: {
-      auto *opt = static_cast<const FixedOptionSecurity *>(&f);
-      core::TimeStamp exp;
-      core::TimeStamp::from_ymd(
-          opt->expiration_year, opt->expiration_month, opt->expiration_day,
-          &exp);
-      j["expiration"] = exp.to_iso_string();
-      j["strike"] = opt->strike;
-      j["callput"] = api::PutOrCall_Name(opt->call_put);
-      break;
-   }
-   case api::SECURITYTYPE_SPREAD: {
-      auto *spd = static_cast<const FixedSpreadSecurityXL *>(&f);
-      std::vector<FixedSpreadLeg> legs{spd->legs, spd->legs + spd->leg_count()};
-      j["legs"] = legs;
-      break;
-   }
-   default:
-      break;
-   }
-}
-
 inline std::ostream &operator<<(
     std::ostream &out, const FixedSecurity &f) noexcept {
    out << io::JsonDumper::dump(f);

src/lib/orderman/actors/order_manager/OrderManager.cpp

@@ -271,13 +271,14 @@ void OrderManager::on_execution_report(
             if (it == me->orders.end()) {
                LOG(OrderMan, Error,
                    "Received an execution report for an unknown order with key "
-                   "{}. Order state will be reconstructed from this report but "
+                   "{}. Order state USED TO be reconstructed from this report but "
                    "some "
                    "information will be lost, including CustomerOrFirm,CTI "
                    "code "
                    "and "
-                   "parent order id",
+                   "parent order id but now these are IGNORED",
                    k.value());
+               continue;
                auto order =
                    &me->services_data.order_factory.recreate_from_exe_report(
                        *report.data);

src/lib/orderman/gateways/ExecutionGatewayFactory.cpp

@@ -9,7 +9,7 @@
 #include "lib/orderman/gateways/GLBXiLink3ExecutionGateway.h"
 #include "lib/orderman/gateways/ICEExecutionGateway.h"
 #include "lib/orderman/gateways/IExecutionGenericAdapter.h"
-#include "lib/orderman/gateways/OkxExecutionAdapter.h"
+#include "lib/orderman/gateways/OkxExecutionGateway.h"
 #include "lib/orderman/gateways/OnixFIXAdapter.h"
 #include "lib/orderman/gateways/sim/SimExecutionGatewayFactory.h"
 #include "lib/orderman/gateways/sim/SimExecutionGatewayUniqueWrapper.h"
@@ -49,9 +49,7 @@ std::unique_ptr<IExecutionGateway> ExecutionGatewayFactory::create_gateway(
             return std::make_unique<GeneralExecutionGateway>(conn.venue(),std::move(adapter),o);
          }
       case bts::api::VENUE_OKX:
-         {
-             std::unique_ptr<IExecutionGenericAdapter> adapter = std::make_unique<OkxExecutionAdapter>(o);
-            return std::make_unique<GeneralExecutionGateway>(conn.venue(),std::move(adapter),o);
+         {  return std::make_unique<OkxExecutionGateway>(o);
          }
       default:
          return std::make_unique<FakeExecutionGateway>(o);

src/lib/orderman/gateways/GeneralExecutionGateway.cpp

@@ -64,7 +64,9 @@ void GeneralExecutionGateway::send_msgs(
           msg);
    }
 }
-void GeneralExecutionGateway::dispatch_incoming_fix_messages() {}
+void GeneralExecutionGateway::dispatch_incoming_fix_messages() {
+   me->_adapter->dispatch_incoming_fix_messages();
+}
 void GeneralExecutionGateway::on_securities(
     const mercury::Update<NewSecuritiesEvent> &) {}
 void GeneralExecutionGateway::on_admin_data(

src/lib/orderman/gateways/IExecutionGenericAdapter.h

@@ -22,7 +22,7 @@ class CancelReject;
 
 using OnExecutionReport = std::function<void(ExecutionReport const &, const trace::CorID &)>;
 
-// fix execution gateway for a single FIX session
+//  execution gateway for a single session
 class IExecutionGenericAdapter {
  public:
    IExecutionGenericAdapter() = default;
@@ -34,6 +34,7 @@ class IExecutionGenericAdapter {
    virtual void launch_new_order(const CreateOrderMsg &) =0;
    virtual void modify_order(const CancelReplaceOrderMsg &) = 0;
    virtual void cancel_order(const CancelOrderMsg &) =0;
+   virtual void dispatch_incoming_fix_messages() = 0;
    // as new securities are added
    // virtual void update_securities(
    //     const mercury::Update<NewSecuritiesEvent> &) = 0;

src/lib/orderman/gateways/OkxExecutionAdapter.cpp

@@ -1,95 +0,0 @@
-#include "OkxExecutionAdapter.h"
-#include "OnixS/FIXEngine/FIX/Engine.h"
-#include "lib/config/protos/marex_fix_cfg.pb.h"
-#include "lib/domain_model/Security.pb.h"
-#include "lib/logging/logging.h"
-#include "lib/orderman/CancelOrderMsg.h"
-#include "lib/orderman/CancelReplaceOrderMsg.h"
-#include "lib/orderman/CreateOrderMsg.h"
-#include "lib/orderman/gateways/OnixFIXListener.h"
-#include "lib/orderman/gateways/TrafixOrderStateHandler.h"
-#include "lib/orderman/orderman_utils/orderman_common.h"
-#include "lib/utils/bts_assert.h"
-#include "lib/utils/helper_macros.h"
-#include <OnixS/FIXEngine/FIX/Exception.h>
-#include <OnixS/FIXEngine/FIX/FIX42.h>
-#include <OnixS/FIXEngine/FIX/Message.h>
-#include <OnixS/FIXEngine/FIX/Numeric.h>
-#include <OnixS/FIXEngine/FIX/ProtocolVersion.h>
-#include <cassert>
-#include <chrono>
-#include <exception>
-#include <memory>
-#include <quickfix/FixFields.h>
-#include <ratio>
-#include <string>
-#include <unistd.h>
-
-namespace bts::orderman {
-
-OkxExecutionAdapter::OkxExecutionAdapter(md::SecurityOracle){}//listener() {}
-OkxExecutionAdapter::~OkxExecutionAdapter() {}
-inline void handleOkException(
-    const std::string &context, const std::string &error) {
-   bts_assert_msgf(false, "{} Exception: {}", context, error);
-}
-void OkxExecutionAdapter::start() {
-    int attempt = 0;
-
-    while (true) {
-        try {
-            LOG(OrderMan, Warn, "OkxExecutionAdapter::start Attempting login (Attempt: {})", attempt + 1);
-            LOG(OrderMan, Warn, "OkxExecutionAdapter::start Login successful");
-            break; // Exit loop on success
-        } catch (const std::exception &ex) {
-            const std::string err = ex.what();
-            LOG(OrderMan, Error, "OkxExecutionAdapter::start Login failed: {}. Retrying in {} seconds.", err, 10);
-
-            // if (maxRetries > 0 && ++attempt >= maxRetries) {
-            //     LOG(OrderMan, Error, "OkxExecutionAdapter::start Exceeded max retries ({}). Aborting.", maxRetries);
-            //     handleOkException("OkxExecutionAdapter::start", err);
-            //     break; // Exit if retries are exhausted
-            // }
-            // std::this_thread::sleep_for(std::chrono::seconds(std::max(60,retryInterval)));
-        }
-    }
-}
-
-void OkxExecutionAdapter::modify_order(
-    const CancelReplaceOrderMsg &msg) { //confirm will have everything
-   try {
-      LOG(OrderMan, Warn, "OkxExecutionAdapter::modify_order modifying order{}",
-          msg);
-   } catch (const std::exception &ex) {
-     const std::string err = ex.what();
-      handleOkException("Modify order", err);
-   }
-}
-
-void OkxExecutionAdapter::launch_new_order(const CreateOrderMsg &msg) {
-   LOG(OrderMan, Warn, "launch order {}", msg);
-   try {
-      LOG(OrderMan, Warn, "OkxExecutionAdapter::launch_new_order creating order");
-   }
-   // Shut down all reactors if they exist.
-   catch (const std::exception &ex) {
-    const std::string err = ex.what();
-      handleOkException("Launch order", err);
-   }
-}
-void OkxExecutionAdapter::register_callback([[maybe_unused]]const OnExecutionReport &cb) {
-   LOG(OrderMan, Warn, "OkxExecutionAdapter::register_callback callback registered");
-   // listener.setCallback(cb);
-}
-void OkxExecutionAdapter::cancel_order(const CancelOrderMsg &msg) {
-   LOG(OrderMan, Warn, "cancel order {}", msg);
-   try {
-      LOG(OrderMan, Warn, "order cancel sent");
-   }
-   // Shut down all reactors if they exist.
-   catch (const std::exception &ex) {
-    const std::string err = ex.what();
-      handleOkException("Cacnel order", err);
-   }
-}
-} // namespace bts::orderman

src/lib/orderman/gateways/OkxExecutionAdapter.h

@@ -1,19 +0,0 @@
-#pragma once
-
-#include "IExecutionGenericAdapter.h"
-#include "lib/md/SecurityOracle.h"
-namespace bts::orderman {
-
-class OkxExecutionAdapter : public IExecutionGenericAdapter {
- public:
-   OkxExecutionAdapter(md::SecurityOracle sec_oracle);
-   ~OkxExecutionAdapter() override;
-   void register_callback(const OnExecutionReport &report) override;
-   void launch_new_order(const CreateOrderMsg &msg) override;
-   void cancel_order(const CancelOrderMsg &msg) override;
-   void modify_order(const CancelReplaceOrderMsg &) override;
-   void start() override;
- private:
-};
-
-} // namespace bts::orderman

src/lib/orderman/gateways/OkxExecutionGateway.cpp

@@ -0,0 +1,722 @@
+#include "OkxExecutionGateway.h"
+#include "lib/api/services/OrderService.pb.h"
+#include "lib/config/protos/okx_exe_cfg.pb.h"
+#include "lib/domain_model/Security.pb.h"
+#include "lib/domain_model/common.h"
+#include "lib/io/http/http_client.h"
+#include "lib/io/http/http_response.h"
+#include "lib/io/websocket/websocket_client.h"
+#include "lib/logging/logging.h"
+#include "lib/md/SecurityOracle.h"
+#include "lib/orderman/CancelOrderMsg.h"
+#include "lib/orderman/CancelReplaceOrderMsg.h"
+#include "lib/orderman/CreateOrderMsg.h"
+#include "lib/orderman/gateways/IExecutionGateway.h"
+#include "lib/orderman/orderman_utils/ExecutionReport.h"
+#include "lib/orderman/orderman_utils/OkxUtils.h"
+#include "lib/orderman/orderman_utils/orderman_common.h"
+#include "lib/utils/Uuid.h"
+#include "lib/utils/bts_assert.h"
+#include "lib/utils/thread_manager.h"
+#include <OnixS/FIXEngine/FIX/Exception.h>
+#include <OnixS/FIXEngine/FIX/FIX42.h>
+#include <OnixS/FIXEngine/FIX/Message.h>
+#include <OnixS/FIXEngine/FIX/Numeric.h>
+#include <OnixS/FIXEngine/FIX/ProtocolVersion.h>
+#include <atomic>
+#include <cassert>
+#include <chrono>
+#include <cstddef>
+#include <ctime>
+#include <exception>
+#include <fmt/core.h>
+#include <map>
+#include <memory>
+#include <openssl/buffer.h>
+#include <openssl/evp.h>
+#include <openssl/hmac.h>
+#include <quickfix/FixFields.h>
+#include <quickfix/Utility.h>
+#include <string>
+#include <unistd.h>
+#include <unordered_map>
+// Example namespace alias for nlohmann::json
+using json = nlohmann::json;
+
+namespace bts::orderman {
+
+OkxExecutionGateway::OkxExecutionGateway(md::SecurityOracle o)
+    : me_(std::make_unique<PImpl>(o)) {
+   LOG(OrderMan, Info, "OkxExecutionAdapter::constructor ");
+} //listener() {}
+OkxExecutionGateway::~OkxExecutionGateway() {}
+
+enum class OkxExecutionState { OFF, STARTED, LOGGED_IN, SUBSCIRBED };
+void OkxExecutionGateway::send_msgs(
+    std::vector<OrderMsg> const &msgs, trace::CorID) {
+   LOG(OrderMan,Info,"ExecutionGateway::send_msgs" );
+   for (const auto &msg : msgs) {
+      std::visit(
+          [this](auto &&order) {
+             using T = std::decay_t<decltype(order)>;
+             if constexpr (std::is_same_v<T, CreateOrderMsg>) {
+                LOG(OrderMan,Warn,"ExecutionGateway::CreateOrderMsg" );
+                launch_new_order(order);
+             
+             } else if constexpr (std::is_same_v<T, CancelOrderMsg>) {
+                LOG(OrderMan,Warn,"ExecutionGateway::CancelOrderMsg" );
+                cancel_order(order);
+             } else if constexpr (std::is_same_v<T, CancelReplaceOrderMsg>) {
+                LOG(OrderMan,Warn,"ExecutionGateway::CancelReplaceOrderMsg" );
+                modify_order(order);
+             } else if constexpr (std::is_same_v<T, RFQMsg>) {
+                 // Handle RFQMsg if necessary
+                 // adapter->handle_rfq(order); // Example placeholder
+             }
+          },
+          msg);
+   }
+}
+struct OkxExecutionGateway::PImpl {
+   const std::string api_key = config::get<config::OkxExeCfg>().api_key();
+   const std::string secret_key = config::get<config::OkxExeCfg>().secret_key();
+   const std::string passphrase = config::get<config::OkxExeCfg>().passphrase();
+   const std::string url =
+       config::get<config::OkxExeCfg>().ws_url() + "/ws/v5/private";
+   const bool simulated_trading = config::get<config::OkxExeCfg>().simulated_trading();
+   const std::string tdmode = config::get<config::OkxExeCfg>().tdmode();
+   const core::TimeSpan connect_timeout = core::TimeSpan::from_seconds(0);
+   const core::TimeSpan pong_timeout = core::TimeSpan::from_seconds(
+       config::get<config::OkxExeCfg>().ws_ping_interval_seconds());
+   websocket::WebsocketClient ws_client;
+   boost::object_pool<ExecutionReport> ws_exe_report_pool;
+   boost::object_pool<ExecutionReport> rest_exe_report_pool;
+   tbb::concurrent_queue<ExecutionReport *> buffered_exec_reps;
+   std::optional<std::thread> ws_thread;
+   OnExeReport cb;
+   OnGwStatusChange on_gw_status_change;
+   OnCancelReject on_cancel_reject;
+   OnExchangePositionState on_exchange_position_state;
+   OnExchangeOpenOrdersState on_exchange_open_orders;
+   md::SecurityOracle o;
+   std::atomic<OkxExecutionState> state = OkxExecutionState::OFF;
+   PImpl(md::SecurityOracle o)
+       : ws_client(
+             websocket::WebsocketClient({url, connect_timeout, pong_timeout})),
+         cb(dummy), o(o) {
+      ws_client.message_handler().bind<&PImpl::on_message>(*this);
+      ws_client.close_handler().bind<&PImpl::on_close>(*this);
+   };
+
+   ~PImpl() {
+      state = OkxExecutionState::OFF;
+      if (ws_run_thread) {
+         ws_run_thread->join();
+      }
+   }
+
+   static void dummy(const ExecutionReport &, const trace::CorID &) {
+      LOG(OrderMan, Error, "OnixFixListener::Dummy fired");
+   }
+   void setCallback(const OnExeReport &callback) {
+      LOG(OrderMan, Info,
+          "OkxExecutionAdapter::setCallback callback registered");
+      cb = callback;
+   }
+
+   std::optional<std::thread> ws_run_thread;
+   std::optional<std::thread> ws_ping_thread;
+
+   void get_exe_report(
+       const nlohmann::basic_json<> &order, ExecutionReport &report) {
+      report.exchange = api::VENUE_OKX;
+      std::string account_name = config::get<config::OkxExeCfg>().account();
+      std::string ops_id = config::get<config::OkxExeCfg>().ops_id();
+
+      std::array<char, 12> account_array = {};
+      std::copy_n(
+          account_name.begin(),
+          std::min(account_name.size(), account_array.size()),
+          account_array.begin());
+      report.account_name = AccountName(account_array);
+
+      to_array(ops_id, &report.operator_id.value());
+
+      report.order_key =
+          OrderKey(std::stoi(order.at("clOrdId").get<std::string>()));
+
+      auto insttype =
+          OkxUtils::getInstType(order.at("instType").get<std::string>());
+      auto venue = api::VENUE_OKX;
+      auto inst_id = order.at("instId").get<std::string>();
+
+      // report.msg_seq_num = /* Generate or map message sequence */;
+      bts_assert(o.get_securities(venue, insttype, inst_id).size() == 1);
+
+      report.security = o.get_securities(venue, insttype, inst_id)[0];
+
+      report.side = (order.at("side").get<std::string>() == "buy")
+                        ? api::Side::SIDE_BUY
+                        : api::Side::SIDE_SELL;
+      report.order_type =
+          OkxUtils::getOrdType(order.at("ordType").get<std::string>());
+      OkxUtils::populateExecTypeAndOrderStatus(
+          report, order.at("state").get<std::string>());
+
+      report.exchange_order_id = ExchangeOrderID(
+          OkxUtils::stringToFixedArray(order.at("ordId").get<std::string>()));
+
+      report.price = Price(std::stod(order.at("px").get<std::string>()));
+
+      report.quantity =
+          Quantity::create_new(std::stod(order.at("sz").get<std::string>()));
+      report.filled_quantity = Quantity::create_new(
+          std::stod(order.at("accFillSz").get<std::string>()));
+      report.last_price =
+          Price(std::stod(order.at("lastPx").get<std::string>()));
+      report.avg_price = Price(std::stod(order.at("avgPx").get<std::string>()));
+      report.last_shares = Quantity::create_new(
+        std::stod(order.at("fillSz").get<std::string>()));
+
+      //there are algos, like spreader
+      // that are super sensitive to messages and cancel themselves. To avoid
+      // this we only pass on the error messages when the order has terminated.
+      // The assumption being that if the order has not terminated it is OK to
+      // ignore the message.
+      // Alternatively we can change the spreader to not do that. Right not this
+      // seems to be the better choice
+      if (!(order.at("code").get<std::string>() == "0")) {
+        report.msg = order.at("msg").get<std::string>();
+     }
+
+
+     //report can be new order, cancel order, order filled, replace order, and partial fill or errror
+     //new order, cancelled and order filled dont need special handling, error gets handled above 
+     //for other three we will look for fields that are only specific to the respectives types
+    //   if (!(order.at("cancelSource").get<std::string>() == "")) {
+
+    //   } 
+      if (!(order.at("amendSource").get<std::string>() == "")) {
+        if((order.at("amendResult").get<std::string>() == "0"))
+            report.exec_type = api::orderman::ExecType::EXEC_TYPE_REPLACED;
+        else{
+            LOG(OrderMan, Warn, "order:{} replace rejected", report.order_key);
+            report.msg = order.at("msg").get<std::string>();
+        }
+      } else if (!(order.at("fillSz").get<std::string>() == "0")) {
+        utils::Uuid guid;
+        std::string fill_id = guid.to_text_string();
+        to_array(fill_id, &report.exchange_fill_id.value());
+      }
+      
+      // report.transact_time = ExchangeTimeStampVar(std::stoll(order.at("uTime").get<std::string>()));
+      ExchangeTimeStamp64 now = ExchangeTimeStamp64(datetime::unix_nano_now());
+      report.transact_time = now;
+      std::string er = report.to_string();
+      LOG(OrderMan, Info, "Created ExecutionReport: {}", er);
+   }
+   void on_close(const std::error_code &ec) {
+      bts_assert_msgf(
+          false, "Websocket stream client disconnected. {}", ec.message());
+   }
+    // fills execution report form basic_json and writes it to
+    // buffered_exec_reps, which is picked up from the other thread
+    // and passed over to the order infrastructure
+    void _push_execution_report(const nlohmann::basic_json<> &order) {
+        ExecutionReport *er_p = ws_exe_report_pool.construct();
+        ExecutionReport &er = *er_p;
+        get_exe_report(order, er);
+        buffered_exec_reps.push(er_p);
+    }
+   void on_message(const std::string &msg_str) {
+      LOG(OrderMan, Info, "okxlistener::on_message. {}", msg_str);
+      nlohmann::json msg;
+      try {
+         msg = nlohmann::json::parse(msg_str);
+      } catch (const std::exception &ex) {
+         std::string s = ex.what();
+         LOG(OrderMan, Error, "Failed to parse json. {}\n{}", s, msg_str);
+         return;
+      }
+      const auto event_type_it = msg.find("event");
+      if (event_type_it != msg.end()) {
+         const auto event_type = event_type_it->get<std::string>();
+         if (event_type == "login") {
+            bts_assert(state == OkxExecutionState::STARTED);
+            const auto code_it = msg.find("code");
+            const auto code = code_it->get<std::string>();
+            if (code == "0") {
+               LOG(OrderMan, Info, "Succefully logged in {}", msg_str);
+               state = OkxExecutionState::LOGGED_IN;
+            }
+
+            return;
+         }
+         if (event_type == "subscribe") {
+            bts_assert(state == OkxExecutionState::LOGGED_IN);
+            state = OkxExecutionState::SUBSCIRBED;
+            return;
+         }
+         if (event_type == "channel-conn-count") { //ignore for now
+            return;
+         }
+      }
+      // Handle order updates
+
+      const auto channel_it = msg.find("arg");
+      if (channel_it == msg.end() ||
+          channel_it->at("channel").get<std::string>() != "orders") {
+         LOG(OrderMan, Error, "Unexpected response id:\n{}", msg_str);
+         return;
+      }
+
+      const auto &data = msg.at("data");
+
+      for (const nlohmann::basic_json<> &order : data) {
+        _push_execution_report(order);
+      }
+   }
+   void start() {
+      bts_assert(!ws_thread);
+      state = OkxExecutionState::STARTED;
+      ws_thread.emplace([this]() {
+         bts::threading::configure_current_thread("okx_execution");
+         start_impl();
+      });
+   }
+
+   void start_impl() {
+      size_t attempt = 0;
+      while (state != OkxExecutionState::OFF) {
+         try {
+            LOG(OrderMan, Warn,
+                "OkxExecutionAdapter::start Attempting login (Attempt: {})",
+                attempt + 1);
+            ws_client.connect({}, true);
+            sleep(2);
+            if (ws_ping_thread.has_value()) {
+               if (ws_ping_thread->joinable()) {
+                  ws_ping_thread
+                      ->join(); // Ensure the previous thread exits cleanly
+               }
+               ws_ping_thread
+                   .reset(); // Reset the optional before creating a new thread
+            }
+
+            ws_ping_thread.emplace([&]() {
+               LOG(OrderMan, Warn,
+                   "OkxExecutionAdapter::ws_ping_thread thread starting. url: "
+                   "{}",
+                   url);
+               auto last_ping_time = core::TimeStamp::now();
+               auto ping_interval = core::TimeSpan::from_seconds(
+                   config::get<config::OkxExeCfg>().ws_ping_interval_seconds());
+               std::this_thread::sleep_for(
+                   std::chrono::milliseconds(ping_interval.millis()));
+
+               while (ws_client.is_connected()) {
+                  const auto now = bts::core::TimeStamp::now();
+                  if ((now - last_ping_time) > ping_interval) {
+                     LOG(OrderMan, Warn,
+                         "OkxExecutionAdapter::ws_ping_thread Pinging "
+                         "websocket. url: {}",
+                         url);
+                     ws_client.ping();
+                     last_ping_time = now;
+                  }
+                  std::this_thread::sleep_for(
+                      std::chrono::seconds(ping_interval.seconds() / 10));
+               }
+               LOG(OrderMan, Warn,
+                   "OkxExecutionAdapter::ws_ping_thread thread exiting. url: "
+                   "{}",
+                   url);
+            });
+
+            auto [timestamp, sign] = OkxUtils::getLoginSignTime(secret_key);
+            static constexpr const char *AUTH_REQ_TEMPLATE =
+                R"({{"op": "login", "args": [{{ "apiKey": "{}", "passphrase": "{}", "timestamp": "{}", "sign": "{}" }}] }})";
+
+            const std::string auth_req = fmt::format(
+                AUTH_REQ_TEMPLATE, api_key, passphrase, timestamp, sign);
+            LOG(OrderMan, Warn,
+                "OkxExecutionAdapter::start Attempting login (req: {})",
+                auth_req);
+            const auto auth_res = ws_client.send_text(auth_req);
+            LOG(OrderMan, Warn,
+                "OkxExecutionAdapter::start Attempting login {}, {}, {}",
+                std::to_string(auth_res.value()),
+                std::string(auth_res.category().name()), auth_res.message());
+            if (auth_res.value()) {
+               LOG(md, Info, "okx authentication failed. {}",
+                   auth_res.message());
+               attempt++;
+               throw std::runtime_error(
+                   "okx authentication failed: " + auth_res.message());
+            }
+            while (state == OkxExecutionState::STARTED) {
+               LOG(OrderMan, Warn,
+                   "OkxExecutionAdapter::start waiting to login");
+               std::this_thread::sleep_for(1s);
+            }
+            LOG(OrderMan, Warn,
+                "OkxExecutionAdapter::start Login successful {}",
+                auth_res.message());
+            std::string subscribe_req = R"({
+                  "op": "subscribe",
+                  "args": [{"channel": "orders", "instType": "ANY"}]
+               })";
+            LOG(OrderMan, Warn,
+                "OkxExecutionAdapter::start AGttempting sub (req: {})",
+                subscribe_req);
+            const auto auth_res2 = ws_client.send_text(subscribe_req);
+            LOG(OrderMan, Warn, "OkxExecutionAdapter::subscribing to orders{}",
+                auth_res2.message());
+            while (state != OkxExecutionState::SUBSCIRBED) {
+               LOG(OrderMan, Warn,
+                   "OkxExecutionAdapter::start waiting to subscribe");
+
+               std::this_thread::sleep_for(1s);
+            }
+            LOG(OrderMan, Warn,
+                "OkxExecutionAdapter::subscribing to orders successful{}",
+                auth_res2.message());
+            on_gw_status_change(true);
+            break; // Exit loop on success
+         } catch (const std::exception &ex) {
+            const std::string err = ex.what();
+            LOG(OrderMan, Error,
+                "OkxExecutionAdapter::start Login failed: {}. Retrying in {} "
+                "seconds.",
+                err,
+                config::get<config::OkxExeCfg>()
+                    .ws_reconnect_interval_seconds());
+
+            size_t maxRetries =
+                config::get<config::OkxExeCfg>().ws_reconnect_max_count();
+            if (maxRetries > 0 && ++attempt >= maxRetries) {
+               LOG(OrderMan, Error,
+                   "OkxExecutionAdapter::start Exceeded max retries ({}). "
+                   "Aborting.",
+                   maxRetries);
+               OkxUtils::handleOkException("OkxExecutionAdapter::start", err);
+               break; // Exit if retries are exhausted
+            }
+            std::this_thread::sleep_for(
+                std::chrono::seconds(config::get<config::OkxExeCfg>()
+                                         .ws_reconnect_interval_seconds()));
+         }
+      }
+   }
+};
+void OkxExecutionGateway::start() { me_->start(); }
+OnGwStatusChange &OkxExecutionGateway::on_gw_status_change() {
+    return me_->on_gw_status_change;
+ }
+void OkxExecutionGateway::modify_order(
+    const CancelReplaceOrderMsg &msg) { //confirm will have everything
+   try {
+      LOG(OrderMan, Warn, "OkxExecutionAdapter::modify_order modifying order{}",
+          msg);
+      http::HttpResponse res;
+      // static constexpr const char *body_template = R"({"instId": "BTC-USDT", "tdMode": "cash", "side": "buy", "ordType": "limit", "px": "1000", "sz": "0.01"})";
+      static constexpr const char *body_template =
+          R"({{"instId": "{}",  "clordId": "{}", "newPx": "{}", "newSz": "{}"}})";
+      const std::string REQUEST_PATH = "/api/v5/trade/amend-order";
+      const std::string body = fmt::format(
+          body_template, msg.create_fields.security->okx().security_desc(),
+          msg.create_fields.key, msg.create_fields.price,
+          msg.create_fields.quantity.full_value());
+      auto headers = OkxUtils::createHeaders(
+          me_->api_key, me_->passphrase, me_->secret_key, "POST", REQUEST_PATH,
+          me_->simulated_trading,body);
+      // Generate and log the curl command
+      std::string curlCommand = OkxUtils::generateCurlCommand(
+          config::get<config::OkxExeCfg>().rest_url() + REQUEST_PATH, headers,
+          body);
+      LOG(OrderMan, Warn, "Curl Command: \n{}", curlCommand);
+      http::HttpClient::post(
+          config::get<config::OkxExeCfg>().rest_url() + REQUEST_PATH, body,
+          headers, &res);
+      const auto &[suc, top_msg] = OkxUtils::orderAccepted(res.body);
+
+      if (suc) {
+         LOG(md, Info, "Http get succeeded {}", res.body);
+      } else {
+        LOG(md, Info, "Http get failed code: {}, errors: {}, body: {}", res.code, res.errors, res.body);
+      }
+   } catch (const std::exception &ex) {
+      const std::string err = ex.what();
+      OkxUtils::handleOkException("Amend order", err);
+   }
+}
+void OkxExecutionGateway::dispatch_incoming_fix_messages() {
+    if (me_->buffered_exec_reps.empty()) {
+        return;
+     }
+     ExecutionReport *er;
+     while (me_->buffered_exec_reps.try_pop(er)) {
+        trace::CorID cid = {};
+        LOG(OrderMan, Info,
+            "OkxExecutionGateway::dispatch_incoming_fix_messages er: {} ",
+            er->to_string());
+        me_->cb(*er, cid);
+     }
+}
+OnExeReport &OkxExecutionGateway::on_exe_report() { return me_->cb; }
+OnCancelReject &OkxExecutionGateway::on_cancel_reject() {
+   static OnCancelReject default_cancel_reject;
+   return default_cancel_reject;
+}
+OnExchangePositionState &OkxExecutionGateway::on_exchange_position_state() {
+   static OnExchangePositionState default_exchange_position_state;
+   return default_exchange_position_state;
+}
+OnExchangeOpenOrdersState &OkxExecutionGateway::on_exchange_open_orders() {
+   static OnExchangeOpenOrdersState default_exchange_open_orders;
+   return default_exchange_open_orders;
+}
+void OkxExecutionGateway::launch_new_order(const CreateOrderMsg &msg) {
+   LOG(OrderMan, Warn, "launch order {}", msg);
+   try {
+      LOG(OrderMan, Warn,
+          "OkxExecutionAdapter::launch_new_order creating order");
+      static constexpr const char *body_template =
+          R"({{"instId": "{}", "tdMode": "{}", "side": "{}", "ordType": "{}", "px": "{}", "sz": "{}", "clordId": "{}"}})";
+      const std::string REQUEST_PATH = "/api/v5/trade/order";
+      const std::string side = msg.side == api::Side::SIDE_BUY ? "buy" : "sell";
+      const std::string quantity = std::to_string(msg.quantity.full_value());
+      const std::string price = std::to_string(msg.price.value());
+      std::string suffix = "ORDER_TYPE_";
+      auto ord_full_name = NativeOrderType_Name(msg.order_type);
+      auto pos = ord_full_name.find(suffix);
+      const std::string ord_type =
+          FIX::string_toLower(ord_full_name.substr(pos + suffix.size()));
+      const std::string body = fmt::format(
+          body_template, msg.security->okx().security_desc(),me_->tdmode, side, ord_type,
+          price, quantity, msg.key);
+      http::HttpResponse res;
+      auto headers = OkxUtils::createHeaders(
+          me_->api_key, me_->passphrase, me_->secret_key, "POST", REQUEST_PATH,
+          me_->simulated_trading,body);
+      // Generate and log the curl command
+      std::string curlCommand = OkxUtils::generateCurlCommand(
+          config::get<config::OkxExeCfg>().rest_url() + REQUEST_PATH, headers,
+          body);
+      LOG(OrderMan, Warn, "Curl Command: \n{}", curlCommand);
+      ExecutionReport *er_p =  me_->rest_exe_report_pool.construct();
+      ExecutionReport &er = *er_p;
+      http::HttpClient::post(
+          config::get<config::OkxExeCfg>().rest_url() + REQUEST_PATH, body,
+          headers, &res);
+      const auto &[suc, top_msg] = OkxUtils::orderAccepted(res.body);
+
+      OkxUtils::populateCreateOrderFields(er, msg);
+      er.price = msg.price;
+      er.quantity = msg.quantity;
+      er.filled_quantity = 0;
+      er.last_shares = 0;
+      er.avg_price = Price(0);
+      ExchangeTimeStamp64 now = ExchangeTimeStamp64(datetime::unix_nano_now());
+      er.transact_time = now;
+
+      er.order_key = msg.key;
+      const auto &[id, ts, ex_msg] = OkxUtils::extractOrderInfo(res.body);
+      er.exchange_order_id = ExchangeOrderID{OkxUtils::stringToFixedArray(id)};
+
+      if (suc) {
+         LOG(md, Info, "Http get succeeded {}", res.body);
+         er.exec_type = api::orderman::ExecType::EXEC_TYPE_PENDING_NEW;
+         er.order_status = api::orderman::NATIVE_ORDER_STATUS_PENDING_NEW;
+      } else {
+         LOG(md, Info, "Http get failed code: {}, errors: {}, body: {}", res.code, res.errors, res.body);
+         er.exec_type = api::orderman::ExecType::EXEC_TYPE_REJECTED;
+         er.order_status = api::orderman::NATIVE_ORDER_STATUS_REJECTED;
+         er.msg = ex_msg + " " + top_msg;
+    }
+    me_->buffered_exec_reps.push(er_p);
+   }
+   // Shut down all reactors if they exist.
+   catch (const std::exception &ex) {
+      const std::string err = ex.what();
+      OkxUtils::handleOkException("Launch order", err);
+   }
+}
+void OkxExecutionGateway::register_callback(
+    [[maybe_unused]] const OnExeReport &cb) {
+   LOG(OrderMan, Warn,
+       "OkxExecutionAdapter::register_callback callback registered");
+   me_->setCallback(cb);
+}
+
+void OkxExecutionGateway::cancel_order(const CancelOrderMsg &msg) {
+   LOG(OrderMan, Warn, "cancel order {}", msg);
+   try {
+      http::HttpResponse res;
+      // static constexpr const char *body_template = R"({"instId": "BTC-USDT", "tdMode": "cash", "side": "buy", "ordType": "limit", "px": "1000", "sz": "0.01"})";
+      static constexpr const char *body_template =
+          R"({{"instId": "{}",  "clordId": "{}"}})";
+      const std::string REQUEST_PATH = "/api/v5/trade/cancel-order";
+      const std::string body = fmt::format(
+          body_template, msg.security->okx().security_desc(), msg.key);
+      auto headers = OkxUtils::createHeaders(
+          me_->api_key, me_->passphrase, me_->secret_key, "POST", REQUEST_PATH,
+          me_->simulated_trading,body);
+      // Generate and log the curl command
+      std::string curlCommand = OkxUtils::generateCurlCommand(
+          config::get<config::OkxExeCfg>().rest_url() + REQUEST_PATH, headers,
+          body);
+      LOG(OrderMan, Warn, "Curl Command: \n{}", curlCommand);
+      http::HttpClient::post(
+          config::get<config::OkxExeCfg>().rest_url() + REQUEST_PATH, body,
+          headers, &res);
+      const auto &[suc, top_msg] = OkxUtils::orderAccepted(res.body);
+      //   ExecutionReport *er_p = me_->exe_report_pool.construct();
+      //   ExecutionReport &er = *er_p;
+      //   er.security = msg.security;
+      //   er.side = msg.side;
+      //   er.order_key = msg.key;
+      //   to_array(msg.routing_record->account(), &er.account_name.value());
+      //   to_array(msg.routing_record->operator_id(), &er.operator_id.value());
+      //   er.exchange = msg.security->exchange();
+      //   er.order_key = msg.key;
+      //   const auto &[id, ts, ex_msg] = OkxUtils::extractOrderInfo(res.body);
+      if (suc) {
+         LOG(md, Info, "Http get succeeded {}", res.body);
+         //  er.exec_type = api::orderman::ExecType::EXEC_TYPE_PENDING_CANCEL;
+         //  er.order_status = api::orderman::NATIVE_ORDER_STATUS_PENDING_CANCEL;
+      } else {
+        LOG(md, Info, "Http get failed code: {}, errors: {}, body: {}", res.code, res.errors, res.body);
+         //  er.exec_type = api::orderman::ExecType::EXEC_TYPE_REJECTED;
+         //  er.order_status = api::orderman::NATIVE_ORDER_STATUS_REJECTED;
+         //  er.msg = ex_msg + " " + top_msg;
+      }
+      //   me_->cb(er, {});
+   }
+   // Shut down all reactors if they exist.
+   catch (const std::exception &ex) {
+      const std::string err = ex.what();
+      OkxUtils::handleOkException("Cacnel order", err);
+   }
+}
+} // namespace bts::orderman::okxexe
+
+
+// ExecutionReport OkxExecutionAdapter::fire_dummy_modify_order_report(CancelReplaceOrderMsg msg) {
+//     ExecutionReport *er_p = me_->create_exe_ptr();
+//     ExecutionReport &er = *er_p;
+//     std::string account_name = config::get<config::OkxExeCfg>().account();
+//     std::string ops_id = config::get<config::OkxExeCfg>().ops_id();
+
+//     std::array<char, 12> account_array = {};
+//     std::copy_n(
+//         account_name.begin(),
+//         std::min(account_name.size(), account_array.size()),
+//         account_array.begin());
+//     er.account_name = AccountName(account_array);
+
+//     to_array(ops_id, &er.operator_id.value());
+//     er.order_key = msg.create_fields.key;
+//     er.msg_seq_num = 0;
+//     er.security = msg.create_fields.security;
+//     er.leg_security = nullptr;
+//     er.side = msg.create_fields.side;
+//     er.time_in_force = msg.create_fields.time_in_force;
+//     er.order_type = msg.create_fields.order_type;
+//     er.exec_type = api::orderman::EXEC_TYPE_REPLACED;
+//     er.order_status = api::orderman::NATIVE_ORDER_STATUS_REPLACED;
+//     er.manual_order_indicator = api::orderman::MANUAL_ORDER_INDICATOR_NO;
+//     auto s = std::to_string(msg.create_fields.key.value());
+//     std::array<char, 17> idAsArray{};
+//     std::copy_n(s.begin(), std::min(s.size()-1, idAsArray.size() - 1), idAsArray.begin());
+//     er.exchange_order_id = ExchangeOrderID{idAsArray};
+//     er.price = msg.create_fields.price;
+//     er.quantity = msg.create_fields.quantity;
+//     er.filled_quantity = Quantity(0);
+//     er.last_shares = Quantity(0);
+//     er.msg = "";
+//     er.exchange = msg.create_fields.security->exchange();
+//     // Set transaction time
+//     ExchangeTimeStamp64 now = ExchangeTimeStamp64(datetime::unix_nano_now());
+//     er.transact_time = now;
+//     me_->cb(er, {});
+//     return er;
+// }
+// ExecutionReport OkxExecutionAdapter::fire_dummy_cancel_order_report(CancelOrderMsg msg) {
+//     ExecutionReport *er_p = me_->create_exe_ptr();
+//     ExecutionReport &er = *er_p;
+//     std::string account_name = config::get<config::OkxExeCfg>().account();
+//     std::string ops_id = config::get<config::OkxExeCfg>().ops_id();
+
+//     std::array<char, 12> account_array = {};
+//     std::copy_n(
+//         account_name.begin(),
+//         std::min(account_name.size(), account_array.size()),
+//         account_array.begin());
+//     er.account_name = AccountName(account_array);
+
+//     to_array(ops_id, &er.operator_id.value());
+//     er.order_key = msg.key;
+//     er.msg_seq_num = 0;
+//     er.security = msg.security;
+//     er.leg_security = nullptr;
+//     er.side = msg.side;
+//     er.time_in_force = api::orderman::TIME_IN_FORCE_GTC;
+//     er.order_type = api::orderman::NATIVE_ORDER_TYPE_LIMIT;
+//     er.exec_type = api::orderman::EXEC_TYPE_TRADE_CANCEL;
+//     er.order_status = api::orderman::NATIVE_ORDER_STATUS_CANCELED;
+//     er.manual_order_indicator = api::orderman::MANUAL_ORDER_INDICATOR_NO;
+//     auto s = std::to_string(msg.key.value());
+//     std::array<char, 17> idAsArray{};
+//     std::copy_n(s.begin(), std::min(s.size()-1, idAsArray.size() - 1), idAsArray.begin());
+//     er.exchange_order_id = ExchangeOrderID{idAsArray};
+//     er.filled_quantity = Quantity(0);
+//     er.last_shares = Quantity(0);
+//     er.msg = "";
+//     er.exchange = msg.security->exchange();
+//     // Set transaction time
+//     ExchangeTimeStamp64 now = ExchangeTimeStamp64(datetime::unix_nano_now());
+//     er.transact_time = now;
+//     me_->cb(er, {});
+//     return er;
+// }
+// ExecutionReport OkxExecutionAdapter::fire_dummy_new_order_report(CreateOrderMsg msg) {
+//     ExecutionReport *er_p = me_->create_exe_ptr();
+//     ExecutionReport &er = *er_p;
+//     std::string account_name = config::get<config::OkxExeCfg>().account();
+//     std::string ops_id = config::get<config::OkxExeCfg>().ops_id();
+
+//     std::array<char, 12> account_array = {};
+//     std::copy_n(
+//         account_name.begin(),
+//         std::min(account_name.size(), account_array.size()),
+//         account_array.begin());
+//     er.account_name = AccountName(account_array);
+
+//     to_array(ops_id, &er.operator_id.value());
+//     er.order_key = msg.key;
+//     er.msg_seq_num = 0;
+//     er.security = msg.security;
+//     er.leg_security = nullptr;
+//     er.side = msg.side;
+//     er.time_in_force = api::orderman::TIME_IN_FORCE_GTC;
+//     er.order_type = msg.order_type;
+//     er.exec_type = api::orderman::EXEC_TYPE_NEW;
+//     er.order_status = api::orderman::NATIVE_ORDER_STATUS_NEW;
+//     er.manual_order_indicator = api::orderman::MANUAL_ORDER_INDICATOR_NO;
+//     auto s = std::to_string(msg.key.value());
+//     std::array<char, 17> idAsArray{};
+//     std::copy_n(s.begin(), std::min(s.size()-1, idAsArray.size() - 1), idAsArray.begin());
+//     er.exchange_order_id = ExchangeOrderID{idAsArray};
+//     er.price = msg.price;
+//     er.quantity = msg.quantity;
+//     er.filled_quantity = Quantity(0);
+//     er.last_shares = Quantity(0);
+//     er.msg = "";
+//     er.exchange = msg.security->exchange();
+//     // Set transaction time
+//     ExchangeTimeStamp64 now = ExchangeTimeStamp64(datetime::unix_nano_now());
+//     er.transact_time = now;
+//     me_->cb(er, {});
+//     return er;
+// }

src/lib/orderman/gateways/OkxExecutionGateway.h

@@ -0,0 +1,39 @@
+#pragma once
+
+#include "lib/logging/logging.h"
+#include "lib/md/SecurityOracle.h"
+#include "lib/orderman/gateways/IExecutionGateway.h"
+namespace bts::orderman {
+
+class OkxExecutionGateway : public IExecutionGateway {
+ public:
+   OkxExecutionGateway(md::SecurityOracle sec_oracle);
+   ~OkxExecutionGateway() override;
+   void register_callback(const OnExeReport &report);
+   void start() override;
+   void dispatch_incoming_fix_messages() override;
+   void on_securities(const mercury::Update<NewSecuritiesEvent> &) override {
+    LOG(md, Info,"Not implemented");
+   };
+   void on_admin_data(const mercury::Update<NewAdminDataEvent> &) override{
+    LOG(md, Info,"Not implemented");
+   };
+   void send_fix(const jetfix::OutFIXPtr &, trace::CorID) override{
+    LOG(md, Info,"Not implemented");
+   }
+   void send_msgs(std::vector<OrderMsg> const &, trace::CorID) override;
+   OnExeReport &on_exe_report() override;
+   OnGwStatusChange &on_gw_status_change() override;
+   OnCancelReject &on_cancel_reject() override;
+   OnExchangePositionState &on_exchange_position_state() override;
+   OnExchangeOpenOrdersState &on_exchange_open_orders() override;
+
+ private:
+   struct PImpl;
+   std::unique_ptr<PImpl> me_;
+   void launch_new_order(const CreateOrderMsg &msg);
+   void cancel_order(const CancelOrderMsg &msg);
+   void modify_order(const CancelReplaceOrderMsg &);
+};
+
+} // namespace bts::orderman

src/lib/orderman/gateways/OnixFIXAdapter.cpp

@@ -238,6 +238,9 @@ void populate_common_order_fields(Message &order, const CreateOrderMsg &msg) {
           .set(OnixS::FIX::FIX42::Tags::Currency, msg.security->currency());
    }
 }
+void OnixFIXAdapter::dispatch_incoming_fix_messages() {
+    listener.dispatch_incoming_fix_messages();
+}
 void OnixFIXAdapter::modify_order(
     const CancelReplaceOrderMsg &msg) { //confirm will have everything
    try {

src/lib/orderman/gateways/OnixFIXAdapter.h

@@ -12,6 +12,7 @@ class OnixFIXAdapter : public IExecutionGenericAdapter {
  public:
    OnixFIXAdapter(md::SecurityOracle sec_oracle);
    ~OnixFIXAdapter() override;
+   void dispatch_incoming_fix_messages() override;
    void register_callback(const OnExecutionReport &report) override;
    void launch_new_order(const CreateOrderMsg &msg) override;
    void cancel_order(const CancelOrderMsg &msg) override;

src/lib/orderman/gateways/OnixFIXListener.cpp

@@ -15,7 +15,7 @@
 #include <string>
 
 namespace bts::orderman {
-ExecutionReport OnixFixListener::fire_dummy_report(CreateOrderMsg msg) {
+void OnixFixListener::fire_dummy_report(CreateOrderMsg msg) {
     ExecutionReport *er_p = exe_report_pool.construct();
     ExecutionReport &er = *er_p;
     std::string account_name = "BTSIM";
@@ -51,13 +51,18 @@ ExecutionReport OnixFixListener::fire_dummy_report(CreateOrderMsg msg) {
     // Set transaction time
     ExchangeTimeStamp64 now = ExchangeTimeStamp64(datetime::unix_nano_now());
     er.transact_time = now;
-    cb(er,{});
-    return er;
+    if(er.security==nullptr){
+        LOG(OrderMan, Error,"DID not have sec"); 
+        exe_report_pool.destroy(&er);
+        return;
+    }
+    LOG(OrderMan, Info,"msg {}",er.msg); 
+    buffered_exec_reps.push(er_p);
+    return;
 }
 
 namespace {
-    ExecutionReport createDefaultExecutionReport() {
-        ExecutionReport er;
+    void createDefaultExecutionReport(ExecutionReport &er) {
         er.exchange = api::VENUE_CBOE;  
         std::string account_name = "BTSIM";
         std::string ops_id = "op_id_bts";
@@ -67,7 +72,6 @@ namespace {
         er.account_name = AccountName(account_array);
 
         to_array(ops_id, &er.operator_id.value());
-        return er;
     }
 
     void populateExecTypeAndOrderStatus(ExecutionReport &er, int exec_type_value) {
@@ -120,9 +124,7 @@ namespace {
         return idAsArray;
     }
 }
-ExecutionReport OnixFixListener::get_exe_report(Message msg) {
-    ExecutionReport *er_p = exe_report_pool.construct();
-    ExecutionReport &er = *er_p;
+void OnixFixListener::get_exe_report(Message msg, ExecutionReport &er) {
 
     auto msg_str = msg.toString();
     LOG(OrderMan,Info, "OnixFixListener::get_exe_report {}", msg_str);
@@ -135,7 +137,7 @@ ExecutionReport OnixFixListener::get_exe_report(Message msg) {
         populateCreateOrderFields(er, m.value());
     }else{ //handle this later
         LOG(OrderMan,Error, "OnixFixListener::get_exe_report {} could not find create order message", msg_str);
-        er = createDefaultExecutionReport();
+        createDefaultExecutionReport(er);
     }  
     // Populate fields from the message
 
@@ -186,8 +188,6 @@ ExecutionReport OnixFixListener::get_exe_report(Message msg) {
     // Set transaction time
     ExchangeTimeStamp64 now = ExchangeTimeStamp64(datetime::unix_nano_now()); 
     er.transact_time = now;
-
-    return er;
 }
 
 void OnixFixListener::onInboundApplicationMsg(Message &msg, Session * /*sn*/) {
@@ -210,13 +210,16 @@ void OnixFixListener::onInboundApplicationMsg(Message &msg, Session * /*sn*/) {
             case '8':
                 {
                     //exection report
-                    auto er = get_exe_report(msg);
+                    ExecutionReport *er_p = exe_report_pool.construct();
+                    ExecutionReport &er = *er_p;
+                    get_exe_report(msg,er);
                     if(er.security==nullptr){
-                    LOG(OrderMan, Error,"DID not have sec"); 
-                    return;
+                        LOG(OrderMan, Error,"DID not have sec"); 
+                        exe_report_pool.destroy(&er);
+                        return;
                     }
-                    LOG(OrderMan, Error,"msg {}",er.msg); 
-                    cb(er,{});
+                    LOG(OrderMan, Info,"msg {}",er.msg); 
+                    buffered_exec_reps.push(er_p);
                 }
         }
 
@@ -226,7 +229,20 @@ void OnixFixListener::onInboundApplicationMsg(Message &msg, Session * /*sn*/) {
 
     // cb(*er, {});
 }
-
+void OnixFixListener::dispatch_incoming_fix_messages() {
+    if (buffered_exec_reps.empty()) {
+        return;
+     }
+     ExecutionReport *er;
+     while (buffered_exec_reps.try_pop(er)) {
+        trace::CorID cid = {};
+        LOG(OrderMan, Info,
+            "OnixFixListener::dispatch_incoming_fix_messages er: {} ",
+            er->to_string());
+        cb(*er, cid);
+     }
+    
+ }
 
 void OnixFixListener::onInboundSessionMsg(Message & msg, Session * /*sn*/)
 {

src/lib/orderman/gateways/OnixFIXListener.h

@@ -26,6 +26,7 @@ class OnixFixListener : public OnixS::FIX::ISessionListener {
       LOG(OrderMan, Info, "OnixFixListener::setCallback callback registered");
       cb = callback;
    }
+   void dispatch_incoming_fix_messages();
    /// Is called when the session-level message is received from the counterparty.
    void onInboundSessionMsg(Message &msg, Session *sn) ONIXS_FIXENGINE_FINAL;
    void onStateChange(
@@ -33,8 +34,9 @@ class OnixFixListener : public OnixS::FIX::ISessionListener {
        SessionState::Enum prevState,
        Session *sn) ONIXS_FIXENGINE_FINAL;
    OnExeReport cb;
-   ExecutionReport fire_dummy_report(CreateOrderMsg msg);
-   ExecutionReport get_exe_report(Message msg);
+   tbb::concurrent_queue<ExecutionReport *> buffered_exec_reps;
+   void fire_dummy_report(CreateOrderMsg msg);
+   void get_exe_report(Message msg, ExecutionReport &er);
    static void dummy(const ExecutionReport &, const trace::CorID &) {
       LOG(OrderMan, Error, "OnixFixListener::Dummy fired");
    }

src/lib/orderman/orderman_utils/OkxUtils.h

@@ -0,0 +1,270 @@
+#include "lib/api/services/OrderService.pb.h"
+#include "lib/domain_model/Security.pb.h"
+#include "lib/logging/logging.h"
+#include "lib/md/SecurityOracle.h"
+#include "lib/orderman/CreateOrderMsg.h"
+#include "lib/orderman/gateways/IExecutionGateway.h"
+#include "lib/orderman/orderman_utils/ExecutionReport.h"
+#include "lib/utils/bts_assert.h"
+#include <OnixS/FIXEngine/FIX/Exception.h>
+#include <OnixS/FIXEngine/FIX/FIX42.h>
+#include <OnixS/FIXEngine/FIX/Message.h>
+#include <OnixS/FIXEngine/FIX/Numeric.h>
+#include <OnixS/FIXEngine/FIX/ProtocolVersion.h>
+#include <boost/pool/object_pool.hpp>
+#include <cassert>
+#include <cstddef>
+#include <ctime>
+#include <exception>
+#include <fmt/core.h>
+#include <map>
+#include <nlohmann/json_fwd.hpp>
+#include <openssl/buffer.h>
+#include <openssl/evp.h>
+#include <openssl/hmac.h>
+#include <quickfix/FixFields.h>
+#include <quickfix/Utility.h>
+#include <regex>
+#include <sstream>
+#include <string>
+#include <tuple>
+#include <unistd.h>
+#include <unordered_map>
+namespace bts::orderman {
+using json = nlohmann::json;
+class OkxUtils {
+ public:
+   // Function template to get a value from a JSON object by key
+   template <typename T>
+   static inline T getJsonValue(
+       const json &jsonObject,
+       const std::string &key,
+       const T &defaultValue = T()) {
+      if (jsonObject.contains(key) && jsonObject[key].is_primitive()) {
+         return jsonObject.value(key, defaultValue);
+      }
+      return defaultValue;
+   }
+
+   // Function to parse JSON safely
+   static inline std::optional<json> parseJsonSafely(
+       const std::string &jsonString) {
+      try {
+         return json::parse(jsonString);
+      } catch (const std::exception &e) {
+         std::string s = e.what();
+         LOG(OrderMan, bts::Error, "Failed to parse JSON: {}", s);
+         return std::nullopt;
+      }
+   }
+
+   struct OrderResponse {
+      bool accepted;
+      std::string message;
+   };
+
+   static inline OrderResponse orderAccepted(const std::string &jsonString) {
+      auto body = parseJsonSafely(jsonString);
+      if (!body)
+         return {false, "NO Body"};
+      auto &jsonResponse = *body;
+
+      return {
+          jsonResponse.value("code", "1") == "0",
+          jsonResponse.value("msg", "NO code or message")};
+   }
+
+   // Function to extract order information from JSON
+   static inline std::tuple<std::string, std::string, std::string> extractOrderInfo(
+       const std::string &jsonString) {
+      auto body = parseJsonSafely(jsonString);
+      if (!body.has_value()) {
+         return {"", "", "json parsing error no body"};
+      }
+      auto jsonResponse = body.value();
+      if (jsonResponse.contains("data") && jsonResponse["data"].is_array()) {
+         const auto &dataArray = jsonResponse["data"];
+         if (!dataArray.empty() && dataArray[0].is_object()) {
+            const auto &firstItem = dataArray[0];
+
+            // Extract values using the templated function
+            std::string ordId =
+                OkxUtils::getJsonValue(firstItem, "ordId", std::string());
+            std::string timestamp =
+                OkxUtils::getJsonValue(firstItem, "ts", std::string());
+            std::string sMsg =
+                OkxUtils::getJsonValue(firstItem, "sMsg", std::string());
+
+            return {ordId, timestamp, sMsg};
+         }
+      }
+      return {"", "", "json parsing error"};
+   }
+   // Function to encode a byte array to Base64
+   static std::string base64Encode(const unsigned char *buffer, size_t length) {
+      BIO *bmem = BIO_new(BIO_s_mem());
+      BIO *b64 = BIO_new(BIO_f_base64());
+      BIO_set_flags(b64, BIO_FLAGS_BASE64_NO_NL); // No newlines
+      BIO_push(b64, bmem);
+      BIO_write(b64, buffer, static_cast<int>(length));
+      BIO_flush(b64);
+
+      BUF_MEM *bptr;
+      BIO_get_mem_ptr(b64, &bptr);
+      std::string result(bptr->data, bptr->length);
+
+      BIO_free_all(b64);
+      return result;
+   }
+
+   static inline std::vector<unsigned char> generateSignature(
+       const std::string &prehash, const std::string &secretKey) {
+      unsigned int len = EVP_MAX_MD_SIZE;
+      std::vector<unsigned char> result(len);
+      HMAC(
+          EVP_sha256(), secretKey.c_str(), static_cast<int>(secretKey.length()),
+          reinterpret_cast<const unsigned char *>(prehash.c_str()),
+          prehash.length(), result.data(), &len);
+      result.resize(len);
+      return result;
+   }
+   static inline std::string getSign(
+       std::string message, const std::string &secret_key = "your_secret_key") {
+      auto signature = generateSignature(message, secret_key);
+      std::string sign = base64Encode(signature.data(), signature.size());
+
+      return sign;
+   }
+   static std::tuple<long, std::string> getLoginSignTime(
+       const std::string &secret_key = "your_secret_key",
+       const std::string &method = "GET",
+       const std::string &path = "/users/self/verify") {
+      auto seconds = bts::core::TimeStamp::now().seconds();
+      std::string message = std::to_string(seconds) + method + path;
+      return {seconds, getSign(message, secret_key)};
+   }
+   static std::string getUTCTimestamp(bts::core::TimeStamp t) {
+      auto rawTimestamp =
+          t.to_iso_string() +
+          "Z"; //gives 20241213T175033.840123Z should be  2024-12-13T17:50:33.840Z
+               // Use regex to parse and reformat
+      std::regex re(
+          R"((\d{4})(\d{2})(\d{2})T(\d{2})(\d{2})(\d{2})\.(\d{3})\d*Z)");
+      std::string formattedTimestamp =
+          std::regex_replace(rawTimestamp, re, "$1-$2-$3T$4:$5:$6.$7Z");
+
+      return formattedTimestamp;
+   }
+   static std::array<char, 17> stringToFixedArray(const std::string &s) {
+      std::array<char, 17> idAsArray{};
+      std::copy_n(
+          s.begin(), std::min(s.size(), idAsArray.size() - 1),
+          idAsArray.begin());
+      return idAsArray;
+   }
+   static std::unordered_map<std::string, std::string> createHeaders(
+       const std::string &apiKey,
+       const std::string &passphrase,
+       const std::string &secretKey,
+       const std::string &method,
+       const std::string &requestPath,
+       const bool simulated_trading,
+       const std::string &body = "") {
+      auto ts = bts::core::TimeStamp::now();
+      std::string timestamp = getUTCTimestamp(ts);
+      std::string prehash = timestamp + method + requestPath + body;
+      auto signature = getSign(prehash, secretKey);
+      LOG(OrderMan, bts::Info, "prehash '{}' , key '{}'", prehash, signature);
+      // Prepare headers
+      std::unordered_map<std::string, std::string> headers = {
+          {"accept", "application/json"},
+          {"Content-Type", "application/json"},
+          {"OK-ACCESS-KEY", apiKey},
+          {"OK-ACCESS-SIGN", signature},
+          {"OK-ACCESS-TIMESTAMP", timestamp},
+          {"OK-ACCESS-PASSPHRASE", passphrase}};
+      if (simulated_trading){
+         headers.insert({"x-simulated-trading", "1"});
+      }
+
+      return headers;
+   }
+   static std::string generateCurlCommand(
+       const std::string &url,
+       const std::unordered_map<std::string, std::string> &headers,
+       const std::string &body) {
+      std::ostringstream oss;
+
+      // Start with the curl command
+      std::string curlCommand = "curl -X POST " + url + " \\\n";
+
+      // Add headers to the curl command
+      for (const auto &header : headers) {
+         curlCommand +=
+             "  -H \"" + header.first + ": " + header.second + "\" \\\n";
+      }
+
+      // Add bodyl
+      curlCommand += "  -d '" + body + "'";
+
+      return curlCommand;
+   }
+   static inline api::SecurityType getInstType(const std::string &type) {
+      static const std::unordered_map<std::string, api::SecurityType> type_map =
+          {{"SPOT", api::SECURITYTYPE_SPOT},
+           {"OPTION", api::SECURITYTYPE_OPTION},
+           {"FUTURES", api::SECURITYTYPE_FUTURE},
+           {"SWAP", api::SECURITYTYPE_FUTURE}};
+
+      auto it = type_map.find(type);
+      return it != type_map.end() ? it->second : api::SECURITYTYPE_UNSET;
+   }
+   static inline api::orderman::NativeOrderType getOrdType(const std::string &type) {
+      static const std::unordered_map<std::string, api::orderman::NativeOrderType> type_map =
+          {{"limit", api::orderman::NATIVE_ORDER_TYPE_LIMIT},
+           {"market", api::orderman::NATIVE_ORDER_TYPE_MARKET}};
+
+      auto it = type_map.find(type);
+      return it != type_map.end() ? it->second : api::orderman::NativeOrderType::NATIVE_ORDER_TYPE_UNDEFINED;
+   }
+   static void populateCreateOrderFields(
+       ExecutionReport &er, const CreateOrderMsg &create_order_msg) {
+      er.security = create_order_msg.security;
+      er.order_type = create_order_msg.order_type;
+      er.time_in_force = create_order_msg.time_in_force;
+      er.manual_order_indicator = create_order_msg.manual_order_indicator;
+      er.side = create_order_msg.side;
+      er.order_key = create_order_msg.key;
+      to_array(
+          create_order_msg.routing_record->account(), &er.account_name.value());
+      to_array(
+          create_order_msg.routing_record->operator_id(),
+          &er.operator_id.value());
+      er.exchange = create_order_msg.security->exchange();
+   }
+   static void populateExecTypeAndOrderStatus(
+       ExecutionReport &er, const std::string &state) {
+      if (state == "canceled") {
+         er.exec_type = api::orderman::ExecType::EXEC_TYPE_CANCELED;
+         er.order_status = api::orderman::NATIVE_ORDER_STATUS_CANCELED;
+      } else if (state == "live") {
+         er.exec_type = api::orderman::ExecType::EXEC_TYPE_NEW;
+         er.order_status = api::orderman::NATIVE_ORDER_STATUS_NEW;
+      } else if (state == "partially_filled") {
+         er.exec_type = api::orderman::ExecType::EXEC_TYPE_PARTIAL_FILL;
+         er.order_status = api::orderman::NATIVE_ORDER_STATUS_PARTIALLY_FILLED;
+      } else if (state == "filled") {
+         er.exec_type = api::orderman::ExecType::EXEC_TYPE_FILL;
+         er.order_status = api::orderman::NATIVE_ORDER_STATUS_FILLED;
+      } else {
+         er.exec_type = api::orderman::ExecType::EXEC_TYPE_UNDEFINED;
+         er.order_status = api::orderman::NATIVE_ORDER_STATUS_UNDEFINED;
+      }
+   }
+   static inline void handleOkException(
+       const std::string &context, const std::string &error) {
+      bts_assert_msgf(false, "{} Exception: {}", context, error);
+   }
+};
+
+} // namespace bts::orderman::okxexe

src/lib/utils/Dispatcher.cpp

@@ -1,21 +1,15 @@
 #ifndef NO_CPP0X
 
 #include "Dispatcher.h"
-#include "../os/os.h"
 #include "defer.h"
-#include "lib/logging/logging.h"
 #include "lib/utils/bts_assert.h"
-#include "lib/utils/bts_chrono.h"
 #include "lib/utils/thread_manager.h"
 #include "lib/utils/type_name.h"
+
 #include <boost/asio.hpp>
-
-#include <memory>
-#include <stdexcept>
-
 #include <boost/thread.hpp>
-#include <chrono>
 #include <condition_variable>
+#include <memory>
 #include <mutex>
 #include <thread>
 

src/server/sparkd/cmds/Run.cpp

@@ -82,7 +82,7 @@ static OrderServiceServer create_order_service_server(
 static void execute() {
    using namespace bts;
    LOG(Gen, Warn,
-       "------------------------------------ Sparkd v2.2.4-2 launched "
+       "------------------------------------ Sparkd v2.2.5.4 launched "
        "--------------------------------");
 
    auto qb_cfg = config::get<config::QuickbusCfg>().node();

src/server/sparkd/md/okx/OkxInstrumentStore.cpp

@@ -41,8 +41,8 @@ static void instrument_to_sec(const OkxInstrument &oi, Security &sec) {
    scomm.exchange_sec_id = oi.instId;
    scomm.desc = oi.instId;
    scomm.min_price_increment = oi.tickSz;
-   scomm.min_quantity = oi.lotSz;
-   scomm.min_quantity_increment = oi.minSz;
+   scomm.min_quantity = oi.minSz;
+   scomm.min_quantity_increment = oi.lotSz;
    scomm.currency = oi.quoteCcy;
 
    switch (oi.instType.value) {
@@ -51,9 +51,13 @@ static void instrument_to_sec(const OkxInstrument &oi, Security &sec) {
       sec.security.emplace<SpotSecurity>();
       break;
    case bts::md::okx::OkxInstrumentType::SWAP:{
-      sec.security_common.symbol = oi.settleCcy;
       auto &fut = sec.security.emplace<FutureSecurity>();
       core::TimeStamp::from_ymd(3000, 1, 1,&fut.expiration);
+      if(oi.instId.find("USDT")!=std::string::npos || oi.instId.find("USDC")!=std::string::npos){
+         sec.security_common.symbol = oi.instId.substr(0,oi.instId.find_first_of("-"));
+         break;
+      }
+      sec.security_common.symbol = oi.settleCcy;
       break;
    }
    case OkxInstrumentType::FUTURES: {

tools/binance/binance.env

@@ -0,0 +1,43 @@
+BINANCE_USDS_FUTURES_PROD_URL="https://fapi.binance.com"
+BINANCE_USDS_FUTURES_DEV_URL="https://testnet.binancefuture.com"
+BINANCE_USDS_FUTURES_EXCHANGE_INFO_PATH="/fapi/v1/exchangeInfo"
+
+BINANCE_USDS_FUTURES_DEPTH_PATH="/fapi/v1/depth"
+BINANCE_USDS_FUTURES_DEPTH_PARAMS='symbol=${SYMBOL}&limit=${LIMIT:-5}'
+
+BINANCE_USDS_FUTURES_RECENT_TRADES_PATH="/fapi/v1/trades"
+BINANCE_USDS_FUTURES_RECENT_TRADES_PARAMS='symbol=${SYMBOL}&limit=${LIMIT:-500}'
+
+BINANCE_USDS_FUTURES_HISTORICAL_TRADES_PATH="/fapi/v1/historicalTrades"
+BINANCE_USDS_FUTURES_HISTORICAL_TRADES_PARAMS='symbol=${SYMBOL}&limit=${LIMIT:-500}'
+
+BINANCE_USDS_FUTURES_MARK_PRICE_PATH="/fapi/v1/premiumIndex"
+BINANCE_USDS_FUTURES_MARK_PRICE_PARAMS='symbol=${SYMBOL}'
+
+BINANCE_USDS_FUTURES_PRICE_TICKER_PATH="/fapi/v1/ticker/price"
+BINANCE_USDS_FUTURES_PRICE_TICKER_PARAMS='symbol=${SYMBOL}'
+
+BINANCE_USDS_FUTURES_BBO_PATH="/fapi/v1/ticker/bookTicker"
+BINANCE_USDS_FUTURES_BBO_PARAMS='symbol=${SYMBOL}'
+
+BINANCE_USDS_FUTURES_OPEN_INTEREST_PATH="/dapi/v1/openInterest"
+BINANCE_USDS_FUTURES_OPEN_INTEREST_PARAMS='symbol=${SYMBOL}'
+
+BINANCE_USDS_FUTURES_LONG_SHORT_RATIO_PATH="/futures/data/globalLongShortAccountRatio"
+BINANCE_USDS_FUTURES_LONG_SHORT_RATIO_PATH='symbol=${SYMBOL}&period=${PERIOD:-30}&limit=${LIMIT:-10}'
+
+BINANCE_COIN_FUTURES_PROD_URL="https://dapi.binance.com"
+BINANCE_COIN_FUTURES_DEV_URL="https://testnet.binancefuture.com"
+BINANCE_COIN_FUTURES_EXCHANGE_INFO_PATH="/dapi/v1/exchangeInfo"
+BINANCE_COIN_FUTURES_DEPTH_PATH="/dapi/v1/depth"
+BINANCE_COIN_FUTURES_RECENT_TRADES_PATH="/dapi/v1/trades"
+BINANCE_COIN_FUTURES_HISTORICAL_TRADES_PATH="/dapi/v1/historicalTrades"
+BINANCE_COIN_FUTURES_MARK_PRICE_PATH="/dapi/v1/premiumIndex"
+BINANCE_COIN_FUTURES_PRICE_TICKER_PATH="/dapi/v1/ticker/price"
+BINANCE_COIN_FUTURES_BBO_PATH="/dapi/v1/ticker/bookTicker"
+BINANCE_COIN_FUTURES_OPEN_INTEREST_PATH="/dapi/v1/openInterest"
+BINANCE_COIN_FUTURES_LONG_SHORT_RATIO_PATH="/futures/data/globalLongShortAccountRatio"
+
+BINANCE_PROD_URL="https://api.binance.com"
+BINANCE_DEV_URL="https://testnet.binance.vision"
+BINANCE_PATH="/api/v3/depth"

tools/binance/get_binance_md.sh

@@ -0,0 +1,141 @@
+#!/usr/bin/env bash
+
+help_text="\
+Get Binance market data.
+
+Usage: get_binance_md.sh [OPTIONS]
+
+Options:
+  --usds-future     Indicates the symbol is a USDS Future
+  --coin-future     Indicates the symbol is a Coin Future
+  --spot            Indicates the symbol is a Spot
+  --us              Indicates the symbol is a Binance US Spot
+  --prod            Use production url
+  --dev             Use development url
+  --dry             Print the command to console without actually running it
+  --data            DEPTH | RECENT_TRADES | HISTORICAL_TRADES | MARK_PRICE | PRICE_TICKER
+  --<key>=<value>   Query parameter (e.g., --symbol=btcusdt, --limit=5)
+
+"
+
+function print_help_and_exit() {
+  echo "$help_text"
+  exit "$1"
+}
+
+declare this_dir="$(readlink -f $(dirname ${BASH_SOURCE[0]}))"
+source "$this_dir/binance.env"
+
+env_name="binance"
+venue="BINANCE"
+declare -u mode
+declare -u data
+declare -A query_params
+declare -i dry
+dry=0
+
+while [[ $# -gt 0 ]]; do
+  case "$1" in
+    --usds-future)
+      venue="BINANCE_USDS_FUTURES"
+      shift 1
+      ;;
+    --coin-future)
+      venue="BINANCE_COIN_FUTURES"
+      shift 1
+      ;;
+    --spot)
+      venue="BINANCE"
+      shift 1
+      ;;
+    --us)
+      venue="BINANCE_US"
+      shift 1
+      ;;
+    --prod)
+      mode="PROD"
+      shift 1
+      ;;
+    --dev)
+      mode="DEV"
+      shift 1
+      ;;
+    --data)
+      if [[ -n "$data" ]]; then
+        echo "Unexpected argument '$1'"
+        print_help_and_exit 1
+      elif [[ -z "$2" ]]; then
+        echo "Data (--data) must be non-empty"
+        print_help_and_exit 1
+      fi
+      data="$2"
+      shift 2
+      ;;
+    --*=*)
+      key="${1%%=*}"
+      key="${key#--}"
+      value="${1#*=}"
+      if [[ -z "$key" ]]; then
+        echo "Invalid --key=value arg (key is empty). $1"
+        print_help_and_exit 1
+      fi
+      query_params["$key"]="$value"
+      shift 1
+      ;;
+    --dry)
+      dry=1
+      shift 1
+      ;;
+    -h)
+      ;&
+    --help)
+      print_help_and_exit 0
+      ;;
+    *)
+      if [[ -z "$1" ]]; then
+        echo "Invalid positional argument '$1'"
+        print_help_and_exit 1
+      elif [[ -z "$symbol" ]]; then
+        symbol="$1"
+      else
+        echo "Unexpected argument '$1'"
+        print_help_and_exit 1
+      fi
+      shift 1
+    ;;
+  esac
+done
+
+if [ -z "$mode" ]; then
+  mode="DEV"
+fi
+if [ -z "$data" ]; then
+  data="DEPTH"
+fi
+
+base_url_name="${venue}_${mode}_URL"
+base_url_name="${venue}_${mode}_URL"
+path_name="${venue}_${data}_PATH"
+params_name="${venue}_${data}_PARAMS"
+if [ ! -f "$this_dir/$env_name.env" ]; then
+  echo "Env file '$this_dir/$env_name.env' does not exist"
+  print_help_and_exit 1
+fi
+
+for key in "${!query_params[@]}"; do
+  val=${query_params[$key]}
+  declare -u keyu
+  keyu="$key"
+  eval "$keyu=$val"
+done
+
+source "$this_dir/$env_name.env"
+
+params_template=${!params_name}
+eval "params=\"$params_template\""
+url="${!base_url_name}${!path_name}?${params}"
+if [ $dry -ne 0 ]; then
+  echo curl --get $url
+else
+  curl --get $url
+fi