fix (ui): Change qty to double in Manual Fills Import dialog

Previously, these display settings were only available via SecurityMD objs.
However, many data sources (e.g., orders, fills, etc.) are joined with regular
Security objs (not SecurityMD). This saves us from doing awkward joins to get
the SecurityMD to get the display settings.

.devcontainer/devcontainer.json

@@ -22,6 +22,7 @@
       "extensions": [
         "eamodio.gitlens",
         "llvm-vs-code-extensions.vscode-clangd",
+        "ms-python.python",
         "ms-vscode.cmake-tools",
         "ms-vscode.cpptools",
         "xaver.clang-format"

src/app_sparkd/configurations/bybit_algo_test/.env

@@ -0,0 +1,6 @@
+SPARKD__CONFIG_DIR=src/app_sparkd/configurations/bybit_algo_test
+SPARKD__RUNTIME_CONF_DIR=/home/bts/8000/sparkdata/conf
+SPARKD__PYTHON_REQUIREMENTS=src/app_sparkd/configurations/alphaworx_zen_hk_live-preview/requirements.txt
+SPARKD__SHM_SIZE=8gb
+SPARKD__MEMORY=8gb
+SPARKD__IMAGE=sparkd_symbols

src/app_sparkd/configurations/bybit_algo_test/py_bybit_md_provider.json

@@ -0,0 +1,18 @@
+{
+  "currencies": [
+    "USDT"
+  ],
+  "futures": [
+    "MOVE",
+    "DUCK",
+    "ETH",
+    "BTC",
+    "SOL",
+    "KAITO"
+  ],
+  "use_testnet": false,
+  "logging": {
+    "console_level": "WARNING",
+    "file_level": "INFO"
+  }
+}

src/app_sparkd/configurations/bybit_algo_test/requirements.txt

@@ -0,0 +1,10 @@
+certifi==2025.4.26
+charset-normalizer==3.4.2
+dotenv==0.9.9
+idna==3.10
+pybit==5.11.0
+pycryptodome==3.23.0
+python-dotenv==1.1.0
+requests==2.32.4
+urllib3==2.4.0
+websocket-client==1.8.0

src/app_sparkd/configurations/bybit_algo_test/spark.conf

@@ -0,0 +1,58 @@
+QuickbusCfg {
+  value {
+    node {
+      reconnect_period_secs: 1
+      hb_period_msecs: 15000
+      max_missed_hbs: 5
+      parent_host: ""
+      parent_port: 0
+      my_port: 0
+    }
+    client {
+      reconnect_period_secs: 1
+      hb_period_msecs: 15000
+      max_missed_hbs: 5
+      parent_host: @SystemConfig.config_ip
+      parent_port: @Ports.qbus_port
+      my_port: 0
+    }
+  }
+}
+
+JetFIX {
+  value {
+    connection {
+      name: "bybit_1"
+      venue: VENUE_BYBIT
+      max_order_msgs_per_window: 300
+      window_duration_msecs: 1000
+    }
+  }
+}
+
+MarketDataConfig {
+  value {
+    python_md_providers {
+      venue: VENUE_BYBIT
+      module_path: "python_interop_plugins"
+      provider_type: "bybit_md_provider.BybitMdProvider"
+    }
+  }
+}
+
+Logging {
+  value {
+    initial_queue_size: 120001
+    max_queue_growth: 4
+    file {
+      default_level: "Info"
+    }
+  }
+}
+
+SoundNotifications {
+  value {
+    enabled: true
+  }
+}
+

src/app_sparkd/configurations/bybit_algo_test/symbols.conf

@@ -0,0 +1,32 @@
+SymbolLists {
+  value {
+    list {
+      key: "BybitSpots"
+      symbol: "MOVE"
+      symbol: "DUCK"
+      symbol: "ETH"
+      symbol: "BTC"
+      symbol: "SOL"
+      symbol: "KAITO"
+    }
+    list {
+      key: "BybitFutures"
+      symbol: "MOVE"
+      symbol: "DUCK"
+      symbol: "ETH"
+      symbol: "BTC"
+      symbol: "SOL"
+      symbol: "KAITO"
+    }
+  }
+}
+
+Symbols {
+  value {
+    venues {
+      name: "VENUE_BYBIT"
+      spot: @SymbolLists.list[BybitSpots].symbol
+      future: @SymbolLists.list[BybitFutures].symbol
+    }
+  }
+}

src/app_sparkd/configurations/bybit_dev/py_bybit_1.json

@@ -9,7 +9,7 @@
   "secret_key": "${BYBIT_SECRET_KEY}",
   "use_testnet": true,
   "logging": {
-    "console_level": "WARNING",
+    "console_level": "INFO",
     "file_level": "INFO"
   }
 }

src/app_sparkd/sparkd/execution/py_execution_gateway_adapter.cpp

@@ -7,6 +7,7 @@
 #include "python_interop/execution/py_create_order_request.h"
 #include "python_interop/execution/py_execution_gateway.h"
 #include "python_interop/execution/py_execution_report.h"
+#include "python_interop/execution/py_position_info.h"
 #include "python_interop/plugin_manager.h"
 #include "python_interop/py_contract_price_info.h"
 #include "python_interop/py_contract_quantity_info.h"
@@ -60,6 +61,7 @@
 
 #include <fmt/format.h>
 #include <future>
+#include <limits>
 #include <nlohmann/json.hpp>
 #include <string>
 #include <variant>
@@ -92,6 +94,8 @@ namespace bts::orderman {
        params)
 #define DIAG_TRACE_PARAMS(name, params)                                        \
    _LOG_TRACE_PARAMS(bts::Diag, name, params)
+#define WARN_TRACE_PARAMS(name, params)                                        \
+   _LOG_TRACE_PARAMS(bts::Warn, name, params)
 #define ERROR_TRACE_PARAMS(name, params)                                       \
    _LOG_TRACE_PARAMS(bts::Error, name, params)
 
@@ -132,6 +136,7 @@ using python_execution::ExecutionGatewayBridgeImpl;
 using python_execution::PyCancelOrderRequest;
 using python_execution::PyCancelReplaceOrderRequest;
 using python_execution::PyCreateOrderRequest;
+using python_execution::PyPositionInfo;
 using python_interop::PluginManager;
 using python_interop::PythonObjectT;
 using std::int64_t;
@@ -195,6 +200,7 @@ struct PyExecutionGatewayAdapter::PImpl {
    bool is_ready_dirty = false;
    volatile bool py_keep_going = false;
    uint32_t n_active = 0;
+   core::TimeStamp last_unresolved_account_log_time;
 
    StartStopStateMachine sm;
    // SimpleOrderKeyMap key_map;
@@ -312,12 +318,6 @@ struct PyExecutionGatewayAdapter::PImpl {
       bridge.on_securities(std::move(py_secs));
    }
 
-   void get_position_info() {
-      sm.invoke_if_active([&](auto) {
-         //
-      });
-   }
-
    void dispatch_is_ready() {
       BTS_LOCK_GUARD(sm.mutex());
       if (is_ready_dirty) {
@@ -385,7 +385,8 @@ struct PyExecutionGatewayAdapter::PImpl {
       if (!domain::get_tick_factors(spec, tick_unit, tick_base)) {
          WARN_FN(
              "on_securities", "Failed to get price tick from {} {}. spec: {}. ",
-             sec.exchange(), name, utils::pb::to_string(sec.contract_spec()));
+             sec.exchange(), py_sec.name(),
+             utils::pb::to_string(sec.contract_spec()));
          tick_unit = 0.01f;
          tick_base = 0.01f;
       }
@@ -478,6 +479,38 @@ struct PyExecutionGatewayAdapter::PImpl {
       py_req.set_is_buy(msg.side == bts::api::SIDE_BUY);
    }
 
+   void from_py(
+       const python_execution::PyPositionInfo &py_pos_info,
+       ExchangePosition::UnitPosition &upos) {
+      upos.exchange_sec_desc = py_pos_info.exchange_security_id();
+      double long_qty = py_pos_info.long_quantity();
+      if (std::isnan(long_qty)) {
+         long_qty = 0;
+      }
+      double long_avg_cost = py_pos_info.long_avg_cost();
+      if (std::isnan(long_avg_cost)) {
+         long_avg_cost = 0;
+      }
+      double short_qty = py_pos_info.short_quantity();
+      if (std::isnan(short_qty)) {
+         short_qty = 0;
+      }
+      double short_avg_cost = py_pos_info.short_avg_cost();
+      if (std::isnan(short_avg_cost)) {
+         short_avg_cost = 0;
+      }
+      upos.position = long_qty - short_qty;
+      // We shouldn't combine long/short positions like this. Avg cost is
+      // lost/meaningless when the net qty is 0.
+      if (upos.position == 0) {
+         *upos.avg_filled_price = std::numeric_limits<double>::quiet_NaN();
+      } else {
+         *upos.avg_filled_price =
+             (long_avg_cost * long_qty - short_avg_cost * short_qty) /
+             (long_qty - short_qty);
+      }
+   }
+
    void dispatch_order_updates() {
       BTS_LOCK_GUARD(sm.mutex());
       std::shared_ptr<const python_execution::PyExecutionReport> py_er_p;
@@ -531,6 +564,65 @@ struct PyExecutionGatewayAdapter::PImpl {
              TimeStamp::now_ns());
       }
    }
+
+   void dispatch_position_updates() {
+      BTS_LOCK_GUARD(sm.mutex());
+      if (account.id() == 0 || account.name().empty()) {
+         const auto now = core::TimeStamp::now();
+         if ((now - last_unresolved_account_log_time) >
+             core::TimeSpan::from_seconds(5)) {
+            WARN_TRACE_PARAMS(
+                "dispatch_position_updates",
+                fmt::format(
+                    R"({{ "venue": "{}", "cfg_acc": "{}", "msg": "Waiting for admin data to get account id" }})",
+                    api::to_pretty(pos.exchange), this->config.account));
+            last_unresolved_account_log_time = now;
+         }
+         return;
+      }
+      vector<shared_ptr<PyPositionInfo>> py_pos_infos;
+      utils::Stopwatch sw;
+      uint32_t n_pos_changes = 0;
+      sw.start();
+      while (bridge.pop_position_info(py_pos_infos)) {
+         for (const shared_ptr<PyPositionInfo> &py_pos_info_ptr :
+              py_pos_infos) {
+            PyPositionInfo &py_pos_info = *py_pos_info_ptr;
+            const string &exch_sec_id = py_pos_info.exchange_security_id();
+            const api::Security *sec_ptr =
+                sec_oracle.get_security(pos.exchange, exch_sec_id);
+            if (sec_ptr == nullptr) {
+               WARN_TRACE_PARAMS(
+                   "dispatch_position_updates",
+                   fmt::format(
+                       R"({{ "venue": "{}", "acc": "{}", "accid": "{}", "esid": "{}", "msg": "Unresolved security" }})",
+                       api::to_pretty(pos.exchange), account.name(),
+                       account.id(), exch_sec_id));
+               continue;
+            }
+            ExchangePosition::UnitPosition *upos_ptr = nullptr;
+            for (auto &upos : pos.total_exchange_positions) {
+               bts_assert(upos.security != nullptr);
+               if (upos.security->id() == sec_ptr->id()) {
+                  upos_ptr = &upos;
+                  break;
+               }
+            }
+            if (upos_ptr == nullptr) {
+               upos_ptr = &pos.total_exchange_positions.emplace_back();
+               upos_ptr->security = sec_ptr;
+               upos_ptr->exchange_sec_desc = exch_sec_id;
+               upos_ptr->account_id = account.id();
+               upos_ptr->account_name = account.name();
+            }
+            from_py(py_pos_info, *upos_ptr);
+            n_pos_changes++;
+         }
+      }
+      if (n_pos_changes > 0) {
+         callbacks.exchange_position_state(pos);
+      }
+   }
 };
 
 PyExecutionGatewayAdapter::PyExecutionGatewayAdapter(
@@ -616,6 +708,7 @@ void PyExecutionGatewayAdapter::send_fix(
 void PyExecutionGatewayAdapter::dispatch_incoming_fix_messages() {
    me->dispatch_is_ready();
    me->dispatch_order_updates();
+   me->dispatch_position_updates();
 }
 
 // Callbacks

src/client/spark/Daemons/MDDaemon.cs

@@ -1,7 +1,6 @@
 #nullable enable
 namespace bts {
    using bts.api;
-   using bts.api.domain;
    using bts.api.synthetic;
    using bts.md;
    using bts.spark;
@@ -133,31 +132,13 @@ namespace bts {
 
             // set 'unit quantity' for each sec
             foreach (var smd in exch_secs.Values) {
-               var (priceDisplay, qtyDisplay) = GetDisplaySettings(smd.security);
+               var meta = _security_metadata.get(smd.security);
+               var qtyDisplay = smd.security.QuantityDisplaySettings(meta.HasValue && meta.Value.HasQuantityDisplay ? meta.Value.QuantityDisplay : null);
+               var priceDisplay = smd.security.PriceDisplaySettings(meta.HasValue && meta.Value.HasPriceDisplay ? meta.Value.PriceDisplay : null);
                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/Model/Settings/SettingsPersistenceModel.cs

@@ -103,9 +103,9 @@ namespace mbts.persist {
          var re = new Regex(pattern);
          return _persist
             .subscribe_ns(_user_ns(ns))
+            .Filter(r => re.IsMatch(r.Key))
             .Select(update =>
-               update.added
-                  .Where(item => re.IsMatch(item.Key))
+               update.all
                   .Select(item => new PersistResult<T>() {
                      user = _user,
                      ns = ns,

src/client/spark/Modules/Trading/Fills/FillsColumn.cs

@@ -36,6 +36,7 @@ namespace mbts.spark {
    class FillsColumn<TColFilter> : DIYGrid.RenderTextColumn<FillsRowVM, ColKey>, IFillsColumn, IFilterableDIYGridColumn<FillTuple, ColKey, FillsFilterStates> {
       readonly Func<FillsFilterStates, TColFilter> _get_col_filter;
       readonly Action<FillsFilterStates, TColFilter> _set_col_filter;
+      readonly Func<FillsRowVM, string> getText;
 
       public FillsColumn(
          ColKey key,
@@ -57,16 +58,20 @@ namespace mbts.spark {
          _get_col_filter = get_col_filter;
          _set_col_filter = set_col_filter;
          this.filter_model = filter_model;
+         getText = get_text;
       }
 
       public new IDataGridColumnPersistentFilter<FillTuple, TColFilter> filter_model { get; }
 
-      IDataGridColumnFilter<FillTuple> IFilterableDIYGridColumn<FillTuple, ColKey>.filter_model => filter_model;
+      IDataGridColumnFilter <FillTuple> IFilterableDIYGridColumn<FillTuple, ColKey>.filter_model => filter_model;
       IDataGridColumnFilter IFilterableDIYGridColumn.filter_model => filter_model;
       ColKey IDIYGridColumn<ColKey>.id => key;
 
       DIYGrid.ColumnBase IFillsColumn.column_view => this;
 
+      public string GetText(FillsRowVM row) =>
+         getText(row);
+
       public FillsFilterStates assign_filter_settings(FillsFilterStates filter_settings) {
          _set_col_filter(filter_settings, filter_model.get_state());
          return filter_settings;

src/client/spark/Modules/Trading/Fills/FillsOrderTypeColumn.cs

@@ -17,9 +17,11 @@ namespace mbts.spark {
    class FillsOrderTypeColumn<TFilter> : DIYGrid.OrderTypeColumn<ColKey>, IFillsColumn {
       readonly Func<FillsFilterStates, TFilter> _get_col_filter;
       readonly Action<FillsFilterStates, TFilter> _set_col_filter;
+      readonly Func<FillsRowVM, string> getText;
 
       public FillsOrderTypeColumn(
          ColKey key,
+         Func<FillsRowVM, string> getText,
          BindingBase badge_fill_binding,
          Binding display_name_binding,
          IDataGridColumnPersistentFilter<FillTuple, TFilter> filter_model,
@@ -27,6 +29,7 @@ namespace mbts.spark {
          Action<FillsFilterStates, TFilter> set_col_filter,
          string header = "Order Type")
          : base(key, badge_fill_binding, display_name_binding, filter_model, header) {
+         this.getText = getText;
          this.filter_model = filter_model;
          _get_col_filter = get_col_filter;
          _set_col_filter = set_col_filter;
@@ -40,6 +43,9 @@ namespace mbts.spark {
       IDataGridColumnFilter IFilterableDIYGridColumn.filter_model => filter_model;
       ColKey IDIYGridColumn<ColKey>.id => key;
 
+      public string GetText(FillsRowVM r) =>
+         getText(r);
+
       public FillsFilterStates assign_filter_settings(FillsFilterStates filter_settings) {
          _set_col_filter(filter_settings, filter_model.get_state());
          return filter_settings;

src/client/spark/Modules/Trading/Fills/FillsSideColumn.cs

@@ -24,6 +24,9 @@ namespace mbts.spark {
 
       FillsView.Columns IDIYGridColumn<FillsView.Columns>.id => key;
 
+      public string GetText(FillsRowVM row) =>
+         row.side.ToString();
+
       public FillsFilterStates assign_filter_settings(FillsFilterStates filter_settings) {
          filter_settings.side = filter_model.get_state();
          return filter_settings;

src/client/spark/Modules/Trading/Fills/FillsVM.cs

@@ -30,6 +30,7 @@ namespace mbts.spark {
    using System.Collections.Specialized;
    using System.Linq;
    using System.Reactive.Linq;
+   using System.Text;
    using System.Windows.Data;
    using System.Windows.Media;
 
@@ -138,6 +139,7 @@ namespace mbts.spark {
          _columns.Add(
             new FillsOrderTypeColumn<OrderType[]>(
                Columns.OrderType,
+               r => r.order_type.ToString(),
                new Binding(nameof(FillsRowVM.badge_fill)),
                new Binding(nameof(FillsRowVM.display_name)),
                order_type_filter,
@@ -161,6 +163,7 @@ namespace mbts.spark {
          _columns.Add(
             new FillsOrderTypeColumn<AlgoID[]>(
                Columns.RootAlgoType,
+               r => r.root_algo_order_type.ToString(),
                new Binding(nameof(FillsRowVM.root_algo_badge_color)) {
                   Converter = new ui.converters.ColorToBrushConverter()
                },
@@ -474,6 +477,33 @@ namespace mbts.spark {
       public void on_sorting(object sender, DIYGridSortingEventArgs e) =>
          _rows_model.on_sorting(e);
 
+      public IEnumerable<string> CsvRows(bool includeHeaderRow) {
+         var cols = _columns.AsCollection().ToArray();
+         if (cols.Length == 0) {
+            yield break;
+         }
+         var sb = new StringBuilder();
+         if (includeHeaderRow) {
+            for (int i = 0; i < cols.Length; i++) {
+               if (i > 0) {
+                  sb.Append(',');
+               }
+               sb.Append(cols[i].id.ToString());
+            }
+            yield return sb.ToString();
+         }
+         foreach (var r in rows) {
+            sb.Clear();
+            for (int i = 0; i < cols.Length; i++) {
+               if (i > 0) {
+                  sb.Append(',');
+               }
+               sb.Append(cols[i].GetText(r));
+            }
+            yield return sb.ToString();
+         }
+      }
+
       static IObservable<TableUpdate<FillTuple, string>> _get_observable(
          OrdersModel model, FillsSettings state) {
          var orders_hash = state.orders?.ToHashSet();

src/client/spark/Modules/Trading/Fills/FillsWindow.cs

@@ -17,6 +17,7 @@ using Spark.Core.Disposables;
 using System;
 using System.Collections.Generic;
 using System.Collections.Immutable;
+using System.IO;
 using System.Linq;
 using System.Reactive.Linq;
 using System.Windows;
@@ -91,6 +92,7 @@ namespace mbts.spark {
                has_selected_rows_src,
                _ => _,
                _ => _vm.selected_rows.Select(r => r.security_key).ToImmutableHashSet()),
+            new SparkCommandCommand("Export to CSV...", Icons.Disk, FileDialogCommand.ExportToCsv(ExportToCsv)),
             // wrap the grid settings view factory in a dialogCommand here since the vm is destroyed/recreated when the security criteria changes
             SparkCommands.window_settings_command(new DialogCommand(() => {
                var vm = new ShellSettingsDialogVM<GridSettings, SimpleGridSettingsVM>(
@@ -128,6 +130,14 @@ namespace mbts.spark {
          //       show_ladder_cmd,
          //       SparkCommands.KeyGestures.ShowLadder);
          // }
+
+         void ExportToCsv(string filePath) {
+            using var fs = File.OpenWrite(filePath);
+            using var sw = new StreamWriter(fs);
+            foreach (var row in _vm.CsvRows(true)) {
+               sw.WriteLine(row);
+            }
+         }
       }
 
       protected override void OnParentViewCreated(SparkWindowView parent_view) =>

src/client/spark/Modules/Trading/Fills/IFillsColumn.cs

@@ -8,6 +8,7 @@ namespace mbts.spark {
 
    interface IFillsColumn : IFilterableDIYGridColumn<FillTuple, FillsView.Columns, FillsFilterStates>, ISortableDIYGridColumn<FillsView.Columns> {
       DIYGrid.ColumnBase column_view { get; }
+      string GetText(FillsRowVM row);
    }
 
    static class FillsColumnExtensions {

src/client/spark/Modules/Trading/ManualFills/ManualFillRow.cs

@@ -58,12 +58,8 @@ namespace bts.shell.manual_fills {
          _quantity.actual_value = data.value.manual_fill.Quantity;
          _quantity_display = backing(
             nameof(quantity_display),
-            () => {
-                  return Security.Defaults.quantity_display_settingsi.with(tick_factors: new() {
-                  TickBase = 1000000,
-                  TickUnit = 1
-               });
-         });
+            () => Security.Defaults.quantity_display_settingsi.with(
+               show_decimal_part: true, tick_factors: new() { TickBase = 1000000, TickUnit = 1 }));
          // TODO (prob): remove conversion_data.  Replace get_qty_display_divisor delegates w/
          // TickDisplaySettingsImmutable for qty, obtained via ISecurityMetadataStore.
          _quantity.conversion_data = new(1, QuantityConversionOpType.DivideActual);

src/client/spark/Modules/Trading/ManualFills/ManualFillsImportConflictsView.xaml

@@ -61,7 +61,7 @@
       </u:TemplateSelectorKeyValue>
       <u:TemplateSelectorKeyValue Key="{x:Static Quantity, MemberType={u:Nested usp:ManualFillsImportConflictsView.ManualFillComponentNames}}">
         <DataTemplate>
-          <c:TickUpDownUInt Value="{Binding value, UpdateSourceTrigger=PropertyChanged}"/>
+          <c:TickUpDownDouble Value="{Binding value, UpdateSourceTrigger=PropertyChanged}"/>
         </DataTemplate>
       </u:TemplateSelectorKeyValue>
       <u:TemplateSelectorKeyValue Key="{x:Static Price, MemberType={u:Nested usp:ManualFillsImportConflictsView.ManualFillComponentNames}}">

src/client/spark/Modules/Trading/ManualFills/ManualFillsParser.cs

@@ -39,7 +39,7 @@ namespace bts.shell.manual_fills {
             case ManualFillComponent.Account: return (mf, val) => mf.AccountId = ConvertUtils.try_convert<AccountTuple?>(val)?.id ?? 0;
             case ManualFillComponent.User: return (mf, val) => mf.UserId = ConvertUtils.try_convert<UserTuple?>(val)?.id ?? 0;
             case ManualFillComponent.Side: return (mf, val) => mf.Side = ConvertUtils.try_convert<Side>(val, Side.Undefined);
-            case ManualFillComponent.Quantity: return (mf, val) => mf.Quantity = ConvertUtils.try_convert<uint>(val);
+            case ManualFillComponent.Quantity: return (mf, val) => mf.Quantity = ConvertUtils.try_convert<double>(val);
             case ManualFillComponent.Price: return (mf, val) => mf.Price = ConvertUtils.try_convert(val, double.NaN);
             default: throw new NotImplementedException();
          }
@@ -130,7 +130,7 @@ namespace bts.shell.manual_fills {
          }
 
          public static void quantity(ref ParsedManualFill pmf, string cell_text) {
-            if (uint.TryParse(cell_text, out uint v)) {
+            if (double.TryParse(cell_text, out var v)) {
                pmf.result.Quantity = v;
                pmf.quantity = (cell_text, true);
             } else {

src/client/spark/Modules/Trading/PNL/VM/PNLCellEmbeddedAnimator.cs

@@ -28,7 +28,9 @@ namespace bts.shell.pnl {
          foreach (var t in _triggers) {
             foreach (var kv in t.trigger(data)) {
                if (!bd.TryGetValue(kv.Key, out var app)) {
-                  app = _defaults[kv.Key];
+                  if (!_defaults.TryGetValue(kv.Key, out app)) {
+                     continue;
+                  }
                }
                app = kv.Value.merge(app);
                bd[kv.Key] = app;

src/client/spark/view/Extensions/Command/FileDialogCommand.cs

@@ -15,6 +15,19 @@ namespace mbts.ui {
       public static IDialogCommand open_file(Action<string> execute, Func<bool> can_execute, string initial_dir, params (string name, string[] extensions)[] extensions) =>
          new AnonymousFileDialogCommand<object>(_ => _create_open_file_dialog(null, extensions), (_, file) => execute(file), _ => can_execute());
 
+      public static IDialogCommand ExportToCsv(Action<string> execute, Func<bool> canExecute = null, string initialDir = null, string initialName = null) {
+         return new AnonymousFileDialogCommand<object>(DialogFactory, Execute, CanExecute);
+
+         SaveFileDialog DialogFactory(object _) =>
+            _create_save_file_dialog(initialDir, initialName, new[] { ("Excel", new[] { "csv" }) });
+
+         void Execute(object _, string file) =>
+            execute(file);
+
+         bool CanExecute(object _) =>
+            canExecute?.Invoke() ?? true;
+      }
+
       public static IDialogCommand export_proto<TProto>(
          Func<TProto> get_proto,
          Func<bool> can_execute = null,
@@ -66,7 +79,7 @@ namespace mbts.ui {
             }
          }, can_execute == null ? null : new Func<object, bool>(_ => can_execute()));
 
-      static TDialog _create_file_dialog<TDialog>(string initial_dir = null, IReadOnlyList<(string name, string[] extensions)> filter_items = null) where TDialog : FileDialog, new() {
+      public static TDialog CreateFileDialog<TDialog>(string initial_dir = null, IReadOnlyList<(string name, string[] extensions)> filter_items = null) where TDialog : FileDialog, new() {
          var rv = new TDialog();
          if (!string.IsNullOrEmpty(initial_dir))
             rv.InitialDirectory = initial_dir;
@@ -78,7 +91,7 @@ namespace mbts.ui {
       }
 
       static SaveFileDialog _create_save_file_dialog(string initial_dir = null, string initial_name = null, IReadOnlyList<(string name, string[] extensions)> filter_items = null) {
-         var rv = _create_file_dialog<SaveFileDialog>(initial_dir, filter_items);
+         var rv = CreateFileDialog<SaveFileDialog>(initial_dir, filter_items);
          rv.OverwritePrompt = true;
          rv.AddExtension = true;
          if (!string.IsNullOrEmpty(initial_name))
@@ -87,7 +100,7 @@ namespace mbts.ui {
       }
 
       static OpenFileDialog _create_open_file_dialog(string initial_dir = null, IReadOnlyList<(string name, string[] extensions)> filter_items = null) =>
-         _create_file_dialog<OpenFileDialog>(initial_dir, filter_items);
+         CreateFileDialog<OpenFileDialog>(initial_dir, filter_items);
 
       static string _build_filter(IEnumerable<(string name, string[] extensions)> items) =>
          string.Join("|", items.Select(i => {

src/client/spark/view/Utility/RXCollection.cs

@@ -240,16 +240,18 @@ namespace mbts.ui.utility {
                      wrapped_models.InsertSorted(table.state.Values, comp);
                      return;
                   } else {
-                     var view_item = update(target[pos], ch.New);
                      int new_pos = CollectionHelper.binary_search(wrapped_models, ch.New, comp);
-                     var view_item_state = update(target[pos], ch.New);
                      if (new_pos < 0)
                         new_pos = ~new_pos;
+                     var view_item = update(target[pos], ch.New);
                      if (Math.Abs(new_pos - pos) > 1) {
                         // If the abs diff between old & new positions is 1 (or 0),
                         // the item does not need to be moved because its position wrt the other elements is unchanged.
                         var state = on_moving == null ? default : on_moving(view_item);
                         target.RemoveAt(pos);
+                        if (new_pos > pos) {
+                           new_pos--;
+                        }
                         target.Insert(new_pos, view_item);
                         on_moved?.Invoke(view_item, state);
                      }

src/client/spark_api/MarketData/TimeStamp.cs

@@ -2,7 +2,9 @@ namespace mbts.co2 {
    using System.Threading;
 
    struct TimeStamp {
+#pragma warning disable CS0649
       long _value;
+#pragma warning restore CS0649
 
       public static ulong invalid_value => ulong.MaxValue;
       public ulong time => (ulong)Interlocked.Read(ref _value);

src/client/spark_api/security/SecurityExtensions.cs

@@ -2,6 +2,7 @@ namespace bts.api {
    using bts.api.domain;
    using bts.utils;
 
+   using mbts;
    using mbts.spark;
 
    using Spark.Api.Data;
@@ -124,6 +125,29 @@ namespace bts.api {
          return double.NaN;
       }
 
+      public static TickDisplaySettingsImmutable PriceDisplaySettings(this SecurityImmutable sec, TickDisplaySettingsImmutable? defaults = null) {
+         var priceDisplay = defaults ?? Security.Defaults.price_display_settingsi;
+         if (sec.HasContractSpec) {
+            var pd = priceDisplay.CloneToMutable();
+            pd.TickFactors = TickUtils.get_tick(sec, 0).to_api_tick_factors();
+            priceDisplay = new(pd);
+         }
+         return priceDisplay;
+      }
+
+      public static TickDisplaySettingsImmutable QuantityDisplaySettings(this SecurityImmutable sec, TickDisplaySettingsImmutable? defaults = null) {
+         var qtyDisplay = defaults ?? Security.Defaults.quantity_display_settingsi;
+         if (sec.HasContractSpec && 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 qtyDisplay;
+      }
+
       internal static ExpirationDate earliest_expiration(this SecurityImmutable security) =>
          security.normal_expirations().Min();
 

src/lib_python_interop/python_interop/execution/execution_gateway_bridge.h

@@ -10,7 +10,8 @@ class ExecutionGatewayBridge {
    virtual ~ExecutionGatewayBridge();
    // Python interface
    virtual void on_execution_report(std::shared_ptr<PyExecutionReport>) {}
-   virtual void on_position_info(std::shared_ptr<PyPositionInfo>) {}
+   virtual void on_position_info(std::vector<std::shared_ptr<PyPositionInfo>>) {
+   }
    virtual void on_is_ready(bool) {}
 };
 

src/lib_python_interop/python_interop/execution/execution_gateway_bridge_impl.cpp

@@ -4,8 +4,10 @@
 #include "python_interop/execution/py_create_order_request.h"
 #include "python_interop/execution/py_position_info.h"
 #include "python_interop/python_object.h"
+#include "spark/core/FMT.h"
 #include "spark/core/TimeSpan.h"
 #include "spark/domain_model/Venue.h"
+#include "spark/io/JsonDumper.h"
 #include "spark/io/json_formatter.h"
 #include "spark/logging/logging.h"
 #include "spark/logging/trace_log.h"
@@ -69,9 +71,13 @@ void ExecutionGatewayBridgeImpl::on_execution_report(
 }
 
 void ExecutionGatewayBridgeImpl::on_position_info(
-    std::shared_ptr<PyPositionInfo> pi) {
-   INFO_TRACE_PARAMS("on_execution_report", pi->to_string());
-   incoming_pos_infos.push(std::move(pi));
+    std::vector<std::shared_ptr<PyPositionInfo>> pos_infos) {
+   std::string pos_info_str = FMT_JOIN_FN(
+       pos_infos, ", ", [&](const std::shared_ptr<PyPositionInfo> &pi) {
+          return pi->to_string();
+       });
+   INFO_TRACE_PARAMS("on_position_info", fmt::format("[{}]", pos_info_str));
+   incoming_pos_infos.push(std::move(pos_infos));
 }
 
 void ExecutionGatewayBridgeImpl::on_is_ready(bool value) {
@@ -95,6 +101,17 @@ bool ExecutionGatewayBridgeImpl::pop_exe_report(
    return false;
 }
 
+bool ExecutionGatewayBridgeImpl::pop_position_info(
+    vector<shared_ptr<PyPositionInfo>> &result) {
+   auto res = incoming_pos_infos.pop();
+   if (res) {
+      result = std::move(*res);
+      return true;
+   }
+   result = {};
+   return false;
+}
+
 void ExecutionGatewayBridgeImpl::on_securities(
     std::vector<std::shared_ptr<bts::md::py_interop::PySecurity>> securities) {
    INFO_TRACE_PARAMS(

src/lib_python_interop/python_interop/execution/execution_gateway_bridge_impl.h

@@ -20,7 +20,7 @@ class ExecutionGatewayBridgeImpl : public ExecutionGatewayBridge {
 
    // Python interface
    void on_execution_report(std::shared_ptr<PyExecutionReport>) override;
-   void on_position_info(std::shared_ptr<PyPositionInfo>) override;
+   void on_position_info(std::vector<std::shared_ptr<PyPositionInfo>>) override;
    void on_is_ready(bool) override;
 
    // C++ interface
@@ -32,6 +32,7 @@ class ExecutionGatewayBridgeImpl : public ExecutionGatewayBridge {
            securities);
    bool wait_for_ready(bts::core::TimeSpan timeout);
    bool pop_exe_report(std::shared_ptr<const PyExecutionReport> &result);
+   bool pop_position_info(std::vector<std::shared_ptr<PyPositionInfo>> &result);
    std::shared_ptr<const PyExecutionReport> pop_exe_report();
    int run_py(
        volatile bool *keep_going,
@@ -43,7 +44,8 @@ class ExecutionGatewayBridgeImpl : public ExecutionGatewayBridge {
    bts::utils::ConcurrentValueStore<bool> is_ready;
    bts::collections::ConcurrentQueue<std::shared_ptr<PyExecutionReport>>
        incoming_exe_reports;
-   bts::collections::ConcurrentQueue<std::shared_ptr<PyPositionInfo>>
+   bts::collections::ConcurrentQueue<
+       std::vector<std::shared_ptr<PyPositionInfo>>>
        incoming_pos_infos;
    bts::collections::ConcurrentQueue<PyOrderRequest> outgoing_reqs;
    std::mutex securities_mx;

src/lib_python_interop/python_interop/execution/py_execution_report.cpp

@@ -119,6 +119,7 @@ void to_json(nlohmann::json &j, const PyExecutionReport &req) {
    j["typ"] = req.order_type();
    j["tt"] = req.transact_time();
    j["tid"] = req.transact_id();
+   j["msg"] = req.error_msg();
 }
 
 void from_json(const nlohmann::json &j, PyExecutionReport &req) {
@@ -151,6 +152,7 @@ void from_json(const nlohmann::json &j, PyExecutionReport &req) {
    j.at("afpx").get_to(req.avg_filled_price_);
    j.at("typ").get_to(req.order_type_);
    j.at("tt").get_to(req.transact_time_);
+   j.at("msg").get_to(req.error_msg_);
 }
 
 } // namespace python_execution

src/lib_python_interop/python_interop/execution/py_position_info.h

@@ -5,6 +5,7 @@
 
 #include <cstdint>
 #include <fmt/format.h>
+#include <limits>
 #include <nlohmann/json.hpp>
 #include <string>
 
@@ -46,11 +47,11 @@ class PyPositionInfo {
    std::string exchange_security_id_;
    std::string venue_;
    std::string security_type_;
-   float long_quantity_;
-   float long_avg_cost_;
-   float short_quantity_;
-   float short_avg_cost_;
-   std::int64_t transact_time_;
+   float long_quantity_ = 0;
+   float long_avg_cost_ = std::numeric_limits<float>::quiet_NaN();
+   float short_quantity_ = 0;
+   float short_avg_cost_ = std::numeric_limits<float>::quiet_NaN();
+   std::int64_t transact_time_ = 0;
 
    friend void from_json(const nlohmann::json &j, PyPositionInfo &req);
 };

src/lib_python_interop/python_interop/python_interop_module.cpp

@@ -289,6 +289,8 @@ PYBIND11_MODULE(python_interop_module, m) {
        .PROP(PyPositionInfo, short_avg_cost)
        .PROP(PyPositionInfo, transact_time)
        .def("__str__", &PyPositionInfo::to_string);
+   py::bind_vector<std::vector<shared_ptr<PyPositionInfo>>>(m, "PyPositionInfoList")
+       .PY_LIST_TO_STRING(PyPositionInfo);
    py::class_<ExecutionGatewayBridge>(m, "ExecutionGatewayBridge")
        .def(py::init<>())
        .DEF(ExecutionGatewayBridge, on_execution_report)

src/lib_python_interop/python_interop_plugins/bybit_execution_gateway.py

@@ -1,8 +1,14 @@
+import datetime
+import enum
 import json
 import time
-from typing import Optional
+from typing import Callable, Optional
 import uuid
-from pybit_unified_trading import WebSocket, WebSocketTrading
+
+import websocket
+
+from pybit_unified_trading import WebSocket, WebSocketTrading, HTTP
+from pybit_websocket_stream import WebSocketError
 from spark_core import (
     JsonElement,
     as_bool,
@@ -21,6 +27,8 @@ from python_interop_module import (
     PyCancelReplaceOrderRequest,
     PyCancelOrderRequest,
     PyExecutionReport,
+    PyPositionInfo,
+    PyPositionInfoList,
     PySecurity,
     PySecurityList,
 )
@@ -28,6 +36,19 @@ from spark_enums import OrderStatus, ExecType, OrderType, SecurityType, TimeInFo
 import threading
 
 
+class BybitGatewayStatus(enum.Enum):
+    IDLE = 0
+    CONNECTING = 1
+    SUBSCRIBING = 2
+    READY = 3
+    STOPPING = 4
+
+
+PyOrderRequest = (
+    PyCreateOrderRequest | PyCancelReplaceOrderRequest | PyCancelReplaceOrderRequest
+)
+
+
 class OrderData:
     def __init__(
         self,
@@ -58,6 +79,20 @@ class OrderData:
         self.last_filled_price: float = 0
         self.exec_type: ExecType = ExecType.PENDING_NEW
         self.status: OrderStatus = OrderStatus.PENDING_NEW
+        # nanos
+        self.timestamp: int = 0
+        self.version: int = 0
+
+    @property
+    def is_active(self):
+        match self.status:
+            case OrderStatus.PENDING_NEW:
+                return True
+            case OrderStatus.NEW:
+                return True
+            case OrderStatus.PARTIALLY_FILLED:
+                return True
+        return False
 
     def to_json(self):
         return {
@@ -80,7 +115,7 @@ class OrderData:
         }
 
     def to_execution_report(self):
-        transact_time = time.time_ns()
+        self.version += 1
         er = PyExecutionReport()
         er.set_data(
             key=self.bts_id,
@@ -100,9 +135,9 @@ class OrderData:
             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,
+            # There is no transact_id in Bybit, so we generate one.
+            transact_id=str(self.version),
+            transact_time=self.timestamp,
         )
         return er
 
@@ -115,9 +150,9 @@ class OrderData:
         status: str,
     ):
         self.price = price
-        self.filled_quantity = 0
         self.quantity = quantity
         lfq = filled_quantity - self.filled_quantity
+        lfq = round(lfq, 7)
         if lfq > 0:
             if self.filled_quantity == 0:
                 self.last_filled_quantity = filled_quantity
@@ -126,6 +161,7 @@ class OrderData:
                 price_diff = avg_filled_price - self.avg_filled_price
                 lfq_weight = filled_quantity / (self.filled_quantity + filled_quantity)
                 lfp = self.avg_filled_price + (price_diff / lfq_weight)
+                lfp = round(lfp, 7)
                 self.last_filled_price = lfp
                 self.last_filled_quantity = lfq
         self.filled_quantity = filled_quantity
@@ -148,7 +184,87 @@ class OrderData:
                 self.status = OrderStatus.REJECTED
             case _:
                 raise ValueError(f"Unexpected order status '{status}'")
-        return self.to_execution_report()
+
+    def on_update_data(self, data):
+        transact_time = int(data["updatedTime"]) * 1_000_000
+        if self.timestamp > transact_time:
+            return False
+        self.timestamp = transact_time
+        total_filled_qty = parse_float(data["cumExecQty"])
+        avg_filled_price = parse_float(data["avgPrice"])
+        status: str = data["orderStatus"]
+        px = parse_float(data["price"])
+        qty = parse_float(data["qty"])
+        self.on_update(px, qty, total_filled_qty, avg_filled_price, status)
+        return True
+
+    def to_exe_report(self, o: dict, security_type: str = "FUTURE"):
+        er = PyExecutionReport()
+        er.set_data(
+            key=self.bts_id,
+            exchange_order_id=o["orderLinkId"],
+            venue="BYBIT",
+            security_type=security_type,
+            exchange_security_id=o["symbol"],
+            account=self.account,
+            order_type=o["orderType"],
+            is_buy=o["side"] == "Buy",
+            price=self.price,
+            quantity=self.quantity,
+            time_in_force=o["timeInForce"],
+            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(time.time_ns()),
+            transact_time=self.timestamp,
+        )
+        return er
+
+
+def merge_order_req(
+    old_req: PyOrderRequest, new_req: PyOrderRequest
+) -> Optional[PyOrderRequest]:
+    if old_req.key() != new_req.key():
+        raise ValueError("Cannot merge PyOrderRequests with different keys")
+    if isinstance(old_req, PyCreateOrderRequest):
+        if isinstance(new_req, PyCancelOrderRequest):
+            # This is a no op
+            return None
+        elif isinstance(new_req, PyCancelReplaceOrderRequest):
+            res = PyCreateOrderRequest()
+            res.set_exchange_security_id(new_req.exchange_security_id())
+            res.set_is_buy(new_req.is_buy())
+            res.set_key(new_req.key())
+            res.set_order_type(old_req.order_type())
+            res.set_price(new_req.price())
+            res.set_quantity(new_req.quantity())
+            res.set_security_type(new_req.security_type())
+            res.set_time_in_force(old_req.time_in_force())
+            res.set_venue(new_req.venue())
+            return res
+    elif isinstance(old_req, PyCancelOrderRequest):
+        if isinstance(new_req, PyCreateOrderRequest):
+            res = PyCancelReplaceOrderRequest()
+            res.set_exchange_order_id(old_req.exchange_order_id())
+            res.set_exchange_security_id(old_req.exchange_security_id())
+            res.set_is_buy(new_req.is_buy())
+            res.set_key(old_req.key())
+            res.set_price(new_req.price())
+            res.set_quantity(new_req.quantity())
+            res.set_security_type(new_req.security_type())
+            res.set_venue(new_req.venue())
+        elif isinstance(new_req, PyCancelOrderRequest):
+            return old_req
+    elif isinstance(old_req, PyCancelReplaceOrderRequest):
+        if isinstance(new_req, PyCancelOrderRequest | PyCancelReplaceOrderRequest):
+            return new_req
+    raise ValueError(
+        f"Cannot merge PyOrderRequest of type {type(old_req)} (old) with {type(new_req)} (new)"
+    )
 
 
 class BybitExecutionGateway(ExecutionGateway):
@@ -157,6 +273,7 @@ class BybitExecutionGateway(ExecutionGateway):
         name = "bybit_1"
         self.name = name
         cfg = load_config([f"py_{name}"])
+        self.config = cfg
         self.logger = configure_logger(
             logging.getLogger(name), "OrderMan", **(cfg.get("logging") or {})
         )
@@ -167,17 +284,40 @@ class BybitExecutionGateway(ExecutionGateway):
         self.use_testnet: bool = cfg.get("use_testnet", True)
         self.use_trace_logging: bool = cfg.get("use_trace_logging", False)
         self.stop_event = threading.Event()
+        self.stream_cli_disconnect_time: float = 0
+        self.trade_cli_disconnect_time: float = 0
         self.stream_cli: Optional[WebSocket] = None
         self.trade_cli: Optional[WebSocketTrading] = None
-        # bts id -> exch id
-        self.orders: dict[int, OrderData] = {}
+        self.trade_http_cli: Optional[HTTP] = None
+        self.trade_cli_connected: bool = False
+        self.stream_cli_connected: bool = False
+        self.stream_subscriptions: set[str] = set()
+        # exch id -> OrderData
+        self.orders: dict[str, OrderData] = {}
         self.securities: dict[str, PySecurity] = {}
+        self.order_ack_callback: Optional[
+            Callable[[dict, Optional[OrderData]], None]
+        ] = None
+        self.status: BybitGatewayStatus = BybitGatewayStatus.IDLE
+        self.lock = threading.RLock()
+        self.buffered_requests: dict[int, PyOrderRequest] = {}
+        self.pos_poll_thread: Optional[threading.Thread] = None
 
-    def _get_order_data(self, exch_id: str):
-        for _, v in self.orders.items():
-            if v.exch_id == exch_id:
-                return v
-        return None
+    @property
+    def is_connected(self):
+        return self.trade_cli_connected and self.stream_cli_connected
+
+    @property
+    def is_subscribed(self):
+        return "order" in self.stream_subscriptions
+
+    def _get_order_data(self, exch_id=None, bts_id=None, require_exists=False):
+        with self.lock:
+            if bts_id is not None:
+                for o in self.orders.values():
+                    if o.bts_id == bts_id:
+                        return o
+            return self.orders[exch_id] if require_exists else self.orders.get(exch_id)
 
     def _order_handlers(self, typ: str, exch_id: str):
         def get_error_desc(msg: dict):
@@ -201,7 +341,13 @@ class BybitExecutionGateway(ExecutionGateway):
                     '[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 not self.bridge:
+                self.logger.error(
+                    "[bybit_gw_ord_ack_err_err.json]%s"
+                    % SLog(typ=typ, exch_id=exch_id, err="bridge is None")
+                )
+                return
+            ord_data = self._get_order_data(exch_id=exch_id)
             if ord_data:
                 er = ord_data.to_execution_report()
                 match er.order_status():
@@ -224,6 +370,8 @@ class BybitExecutionGateway(ExecutionGateway):
                     '[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)
                 )
+            if self.order_ack_callback:
+                self.order_ack_callback(msg, ord_data)
 
         return (on_ack, on_err)
 
@@ -238,65 +386,211 @@ class BybitExecutionGateway(ExecutionGateway):
             % (key, req_type, rsn, time.time_ns())
         )
 
+    def get_order_base_coins(self):
+        # TODO: get base coins for working orders
+        rv = set()
+        rv.add("USDT")
+        return rv
+
+    def on_stream_open(self, ws):
+        with self.lock:
+            self.logger.info("[bybit_gw_stream_open.json]%s" % SLog())
+            self.stream_cli_connected = True
+            self.stream_cli_disconnect_time = 0
+
+    def on_stream_subscription_ack(self, subs: list[str], ok: bool):
+        self.logger.info("[bybit_gw_sub_ack.json]%s" % SLog(subs=subs, ok=ok))
+        with self.lock:
+            for s in subs:
+                if ok:
+                    self.stream_subscriptions.add(s)
+                else:
+                    self.stream_subscriptions.remove(s)
+            if self.is_subscribed:
+                self.sync_orders()
+                if self.is_connected:
+                    self.status = BybitGatewayStatus.READY
+
+    def on_stream_error(self, ws, err: WebSocketError):
+        self.logger.error("[bybit_gw_stream_err.json]%s" % SLog(err=str(err)))
+        with self.lock:
+            self.stream_cli_connected = False
+            self.stream_cli_disconnect_time = time.time()
+            self.stream_subscriptions.clear()
+            self.status = BybitGatewayStatus.CONNECTING
+        return True
+
+    def on_stream_close(self, ws, code: Optional[int], rsn: Optional[str]):
+        self.logger.error("[bybit_gw_stream_close.json]%s" % SLog(code=code, rsn=rsn))
+        with self.lock:
+            self.stream_cli_connected = False
+            self.stream_cli_disconnect_time = time.time()
+            self.stream_subscriptions.clear()
+            self.status = BybitGatewayStatus.CONNECTING
+        return True
+
+    def on_trade_cli_open(self, ws):
+        self.logger.info("[bybit_gw_trd_cli_open.json]%s" % SLog())
+        with self.lock:
+            self.trade_cli_connected = True
+            self.trade_cli_disconnect_time = 0
+            if self.is_subscribed:
+                self.status = BybitGatewayStatus.READY
+
+    def on_trade_cli_error(self, ws, err: WebSocketError):
+        self.logger.error("[bybit_gw_trd_cli_err.json]%s" % SLog(err=str(err)))
+        with self.lock:
+            self.trade_cli_connected = False
+            self.trade_cli_disconnect_time = time.time()
+            return True
+
+    def on_trade_cli_close(self, ws, code: Optional[int], rsn: Optional[str]):
+        self.logger.error("[bybit_gw_trd_cli_close.json]%s" % SLog(code=code, rsn=rsn))
+        with self.lock:
+            self.trade_cli_connected = False
+            self.trade_cli_disconnect_time = time.time()
+            return True
+
+    def sync_orders(self, category: str = "linear", settle_coin: str = "USDT"):
+        self.logger.info("[bybit_gw_sync_ords.json]%s" % SLog())
+        next_pg = None
+        MAX_REQUESTS = 10
+        request_count = 0
+        params = {"category": category, "settleCoin": settle_coin}
+        try:
+            while True:
+                if request_count == MAX_REQUESTS:
+                    raise ValueError(
+                        f"BybitExecutionGateway.sync_orders performed too many requests in loop ({MAX_REQUESTS})"
+                    )
+                request_count += 1
+                if next_pg:
+                    params["cursor"] = next_pg
+                http_res = self.trade_http_cli.get_open_orders(**params)
+                self.logger.info(
+                    "[bybit_gw_open_ords_res.json]%s" % JsonElement(http_res)
+                )
+                if type(http_res) != dict or http_res["retCode"] != 0:
+                    break
+                for o in http_res["result"]["list"]:
+                    od = self.orders.get(o.get("orderLinkId"))
+                    if not od:
+                        continue
+                    if (not od.is_active) and (od.bts_id in self.buffered_requests):
+                        del self.buffered_requests[od.bts_id]
+                    if od.on_update_data(o) and self.exe_report_callback:
+                        self.exe_report_callback(od.to_exe_report(od))
+                next_pg = http_res["result"].get("nextPageCursor")
+                if not next_pg:
+                    break
+        except Exception as e:
+            self.logger.error("[bybit_gw_open_ords_err.json]%s" % SLog(err=str(e)))
+        # Fire buffered requests
+        for req in self.buffered_requests.values():
+            if isinstance(req, PyCreateOrderRequest):
+                self.create_order(req)
+            elif isinstance(req, PyCancelReplaceOrderRequest):
+                self.cancel_replace_order(req)
+            elif isinstance(req, PyCancelOrderRequest):
+                self.cancel_order(req)
+
+    def get_positions(self) -> Optional[PyPositionInfoList]:
+        try:
+            next_pg = None
+            pos_dict: dict[str, PyPositionInfo] = {}
+            # Safe guard against spamming requests in endless loop.
+            MAX_REQUESTS = 10
+            request_count = 0
+            while True:
+                request_count += 1
+                if request_count == MAX_REQUESTS:
+                    raise ValueError(
+                        f"BybitExecutionGateway.get_positions performed too many requests in loop ({MAX_REQUESTS})"
+                    )
+                # Just get USDT positions for now.
+                # TODO: support other settleCoins
+                params = {"category": "linear", "settleCoin": "USDT"}
+                if next_pg:
+                    params["cursor"] = next_pg
+                http_res = self.trade_http_cli.get_positions(**params)
+                self.logger.info("[bybit_gw_pos_res.json]%s" % JsonElement(http_res))
+                if type(http_res) != dict or http_res["retCode"] != 0:
+                    break
+                res = http_res["result"]
+                posns: list = res["list"]
+                for p in posns:
+                    exch_sec_id = p["symbol"]
+                    pi: Optional[PyPositionInfo] = pos_dict.get(exch_sec_id)
+                    # Convert ms to ns
+                    ts = int(p["updatedTime"]) * 1_000_000
+                    if not pi:
+                        pi = PyPositionInfo()
+                        pos_dict[exch_sec_id] = pi
+                        pi.set_venue("BYBIT")
+                        pi.set_exchange_security_id(exch_sec_id)
+                        pi.set_security_type(SecurityType.FUTURE.name)
+                        pi.set_account(self.account)
+                        pi.set_transact_time(ts)
+                    elif pi.transact_time() < ts:
+                        pi.set_transact_time(ts)
+                    else:
+                        continue
+                    if p["side"] == "Buy":
+                        pi.set_long_avg_cost(float(p["avgPrice"]))
+                        pi.set_long_quantity(float(p["size"]))
+                    else:
+                        pi.set_short_avg_cost(float(p["avgPrice"]))
+                        pi.set_short_quantity(float(p["size"]))
+                next_pg = res.get("nextPageCursor")
+                if not next_pg:
+                    break
+            pos_infos = PyPositionInfoList()
+            for p in pos_dict.values():
+                pos_infos.append(p)
+            return pos_infos
+        except Exception as e:
+            self.logger.error("[bybit_gw_pos_err.json]%s" % SLog(err=str(e)))
+        return None
+
     def start_impl(self, bridge: ExecutionGatewayBridge):
         def on_order_update(msg: dict):
-            def on_error(err: str):
-                self.logger.error(
-                    '[bybit_gw_ord_upd_err.json]{ "err": "%s", "ts": %d }'
-                    % (err, time.time_ns())
-                )
-
             self.logger.info("[bybit_gw_ord_upd.json]%s" % msg)
             try:
                 upds: list[dict] = msg["data"]
                 for d in upds:
-                    er = PyExecutionReport()
                     exch_id: str = d["orderLinkId"]
                     ord_data: Optional[OrderData] = self._get_order_data(exch_id)
                     if not ord_data:
-                        on_error("Did not resolve exch order id %s" % exch_id)
+                        self.logger.error(
+                            "[bybit_gw_ord_upd_err.json]%s"
+                            % SLog(err=f"Unresolved exch order id {exch_id}")
+                        )
                         return
-                    total_filled_qty = parse_float(d["cumExecQty"])
-                    avg_filled_price = parse_float(d["avgPrice"])
-                    status: str = d["orderStatus"]
-                    px = parse_float(d["price"])
-                    qty = parse_float(d["qty"])
-                    ord_data.on_update(
-                        px, qty, total_filled_qty, avg_filled_price, status
-                    )
-                    transact_time = int(d["updatedTime"]) * 1_000_000
-                    er.set_data(
-                        key=ord_data.bts_id,
-                        exchange_order_id=d["orderLinkId"],
-                        venue="BYBIT",
-                        security_type="FUTURE",
-                        exchange_security_id=d["symbol"],
-                        account=self.account,
-                        order_type=d["orderType"],
-                        is_buy=d["side"] == "Buy",
-                        price=px,
-                        quantity=qty,
-                        time_in_force=d["timeInForce"],
-                        exec_type=ord_data.exec_type.name,
-                        order_status=ord_data.status.name,
-                        last_filled_price=ord_data.last_filled_price,
-                        last_filled_quantity=ord_data.last_filled_quantity,
-                        filled_quantity=total_filled_qty,
-                        avg_filled_price=avg_filled_price,
-                        # There is no transact_id in Bybit, so we use updatedTime
-                        transact_id=str(transact_time),
-                        transact_time=transact_time,
-                    )
+                    if not ord_data.on_update_data(d):
+                        continue
+                    if not ord_data.is_active:
+                        del self.orders[ord_data.exch_id]
+                    er = ord_data.to_exe_report(d)
                     self.logger.info("[bybit_gw_exe_rep_out.json]%s" % er)
-                    bridge.on_execution_report(er)
+                    if self.bridge:
+                        self.bridge.on_execution_report(er)
             except Exception as e:
-                self.logger.error(
-                    '[bybit_gw_ord_upd_err.json]{ "err": %s }' % e, exc_info=True
-                )
+                self.logger.error("[bybit_gw_ord_upd_err.json]%s" % SLog(err=str(e)))
 
         def on_exe_report(msg):
             self.logger.info("[bybit_gw_exe_rep.json]%s" % msg)
 
+        with self.lock:
+            self.status = BybitGatewayStatus.CONNECTING
+        self.bridge = bridge
+        # Http client (gets position snapshots)
+        params = {
+            "api_key": self.api_key,
+            "api_secret": self.secret_key,
+            "testnet": self.use_testnet,
+        }
+        self.logger.info("[trade_http_cli.json]%s" % JsonElement(params))
+        self.trade_http_cli = HTTP(**params)
         # Stream client (receives order updates)
         params = {
             "api_key": self.api_key,
@@ -307,7 +601,13 @@ class BybitExecutionGateway(ExecutionGateway):
             "trace_logging": False,
         }
         self.logger.info("[bybit_gw_stream_sub.json]%s" % JsonElement(params))
-        self.stream_cli = WebSocket(**params)
+        self.stream_cli = WebSocket(
+            **params,
+            on_open=self.on_stream_open,
+            on_error=self.on_stream_error,
+            on_close=self.on_stream_close,
+            subscription_ack_callback=self.on_stream_subscription_ack,
+        )
         self.stream_cli.order_stream(on_order_update)
         self.stream_cli.execution_stream(on_exe_report)
         # Trade client (sends orders)
@@ -318,27 +618,29 @@ class BybitExecutionGateway(ExecutionGateway):
             "private_auth_expire": 5,
         }
         self.logger.info("[bybit_gw_start_trd_cli.json]%s" % JsonElement(params))
-        self.trade_cli = WebSocketTrading(**params)
-        self.bridge.on_is_ready(True)
-        # Sleep loop to keep thread alive and detect connection status changes
-        conn_statuses = {
-            "stream": self.stream_cli.is_connected(),
-            "trade": self.trade_cli.is_connected(),
-        }
-        while not self.stop_event.is_set():
-            time.sleep(1)
-            new_conn_statuses = {
-                "stream": self.stream_cli.is_connected(),
-                "trade": self.trade_cli.is_connected(),
-            }
-            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_connected else logging.WARNING,
-                    "[bybit_gw_connection_status.json]%s" % SLog(**conn_statuses),
-                )
-                self.bridge.on_is_ready(all_connected)
+        self.trade_cli = WebSocketTrading(
+            **params,
+            on_open=self.on_trade_cli_open,
+            on_error=self.on_trade_cli_error,
+            on_close=self.on_trade_cli_close,
+        )
+        bridge.on_is_ready(True)
+
+        POS_SNAPSHOT_INTERVAL = 10
+        next_pos_snapshot_time = time.time() + POS_SNAPSHOT_INTERVAL
+        while not self.stop_event.wait(1):
+            if time.time() > next_pos_snapshot_time:
+                positions = self.get_positions()
+                next_pos_snapshot_time = time.time() + POS_SNAPSHOT_INTERVAL
+                if positions:
+                    self.logger.info("[bybit_gw_pos_infos.json]%s" % positions)
+                    bridge.on_position_info(positions)
+                else:
+                    self.logger.warning(
+                        "[bybit_gw_pos_infos_err.json]%s"
+                        % SLog(err="Failed to get positions")
+                    )
+        self.logger.info("[bybit_gw_poll_pos_done.json]%s" % SLog())
 
     def start(self, bridge: ExecutionGatewayBridge):
         def start_wrapper():
@@ -360,41 +662,68 @@ class BybitExecutionGateway(ExecutionGateway):
     def stop(self):
         self.logger.info("[bybit_gw_stop.json]%s" % SLog())
         self.stop_event.set()
+        if self.pos_poll_thread:
+            self.pos_poll_thread.join()
+        self.stop_event.clear()
+
+    def _buffer_request(self, req: PyOrderRequest):
+        bts_id = req.key()
+        old_req = self.buffered_requests.get(bts_id)
+        self.logger.warning(
+            "[bybit_gw_buffer_req.json]%s"
+            % SLog(
+                bts_id=bts_id,
+                old_req_type=str(type(old_req)),
+                new_req_type=str(type(req)),
+            )
+        )
+        if old_req:
+            new_req = merge_order_req(old_req, new_req)
+            if new_req:
+                self.buffered_requests[bts_id] = new_req
+            else:
+                del self.buffered_requests[bts_id]
+        else:
+            self.buffered_requests[bts_id] = req
 
     def create_order(self, req: PyCreateOrderRequest):
-        self.logger.info("[bybit_gw_crt_ord.json]%s" % req)
-        if not self.stream_cli or not self.trade_cli or not self.bridge:
-            self._on_abort(
-                req.key(),
-                "crt",
-                "WebSocket not initialized (stream_cli: %s, trade_cli: %s, bridge: %s)"
-                % as_bool(self.stream_cli, self.trade_cli, self.bridge),
+        self.logger.info("[bybit_gw_crt.json]%s" % req)
+        if self.status != BybitGatewayStatus.READY:
+            self.logger.warning(
+                "[bybit_gw_crt_err.json]%s", SLog(err=f"Status is {self.status.name}")
             )
+            self._buffer_request(req)
             return
         exch_ord_id = uuid.uuid1().hex
         bts_id = req.key()
         px = req.price()
         qty = req.quantity()
-        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)
+        exch_sec_id = req.exchange_security_id()
+        with self.lock:
+            self.orders[exch_ord_id] = OrderData(
+                account=self.account,
+                bts_id=bts_id,
+                exch_id=exch_ord_id,
+                exch_sec_id=exch_sec_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(exch_sec_id)
         if not sec:
-            raise ValueError(f"Unknown security '{sec_name}'")
+            self.logger.error(
+                "[bybit_gw_crt_err.json]%s"
+                % SLog(err=f"Unknown security '{exch_sec_id}'")
+            )
+            self.buffered_requests[req.key()] = req
+            return
         params = {
             "orderLinkId": exch_ord_id,
             "category": "linear",
-            "symbol": sec_name,
+            "symbol": exch_sec_id,
             "side": "Buy" if req.is_buy() else "Sell",
             "orderType": req.order_type().title(),
             "timeInForce": req.time_in_force().upper(),
@@ -402,22 +731,56 @@ class BybitExecutionGateway(ExecutionGateway):
             "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(on_ack, **params, error_callback=on_err)
+        self.logger.info("[bybit_gw_crt_send.json]%s" % SLog(params=params))
+        try:
+            self.trade_cli.place_order(on_ack, **params, error_callback=on_err)
+            return
+        except websocket.WebSocketConnectionClosedException as e:
+            self.logger.error("[bybit_gw_crt_err.json]%s" % SLog(err=str(e)))
+        # Handle websocket closed, signal to algos we are reconnecting
+        if self.bridge:
+            er = PyExecutionReport()
+            ts = time.time_ns()
+            er.set_data(
+                key=req.key(),
+                account=self.account,
+                exchange_order_id="",
+                exchange_security_id=req.exchange_security_id(),
+                venue=req.venue(),
+                security_type=req.security_type(),
+                order_type=req.order_type(),
+                order_status=OrderStatus.REJECTED.name,
+                time_in_force=req.time_in_force(),
+                exec_type=ExecType.REJECTED.name,
+                is_buy=req.is_buy(),
+                price=req.price(),
+                quantity=req.quantity(),
+                filled_quantity=0.0,
+                last_filled_quantity=0.0,
+                last_filled_price=0.0,
+                avg_filled_price=0.0,
+                transact_id=ts,
+                transact_time=ts,
+            )
+            er.set_error_msg("RECONNECTING")
+            self.bridge.on_execution_report(er)
 
     def cancel_replace_order(self, req: PyCancelReplaceOrderRequest):
         self.logger.info("[bybit_gw_cxlrep_ord.json]%s" % req)
-        if not self.stream_cli or not self.trade_cli:
-            self._on_abort(req.key(), "cxlrep", "WebSocket client not created.")
+        if self.status != BybitGatewayStatus.READY:
+            self._buffer_request(req)
             return
         bts_id = req.key()
-        ord_data = self.orders.get(bts_id)
+        ord_data = self._get_order_data(bts_id=bts_id)
         if not ord_data:
             self._on_abort(req.key(), "cxlrep", "Did not resolve exch order id")
             return
         sec_name = req.exchange_security_id()
         sec = self.securities.get(sec_name)
         if not sec:
+            self._on_abort(
+                req.key(), "cxlrep", f"Did not resolve id for sec '{sec_name}'"
+            )
             return
         params = {
             "orderLinkId": ord_data.exch_id,
@@ -427,18 +790,42 @@ class BybitExecutionGateway(ExecutionGateway):
             "qty": format_quantity(sec, req.quantity()),
         }
         on_ack, on_err = self._order_handlers("cxlrep", ord_data.exch_id)
+        if self.status != BybitGatewayStatus.READY:
+            self.buffered_requests[req.key()] = req
+            return
         self.logger.info("[bybit_gw_cxlrep_ord_send.json]%s" % SLog(params=params))
-        self.trade_cli.amend_order(on_ack, **params, error_callback=on_err)
+        try:
+            self.trade_cli.amend_order(on_ack, **params, error_callback=on_err)
+            return
+        except websocket.WebSocketConnectionClosedException as e:
+            self.logger.error("[bybit_cxlrep_err.json]%s" % SLog(err=str(e)))
+        # Handle websocket closed, signal to algos we are reconnecting
+        ord_data = self.orders.get(bts_id=bts_id)
+        if not ord_data:
+            self.logger.error(
+                "[bybit_cxlrep_err.json]%s"
+                % SLog(err="Order not found with id %s" % req.exchange_order_id())
+            )
+            return
+        ord_data.status = OrderStatus.REJECTED
+        if self.exe_report_callback:
+            er = ord_data.to_execution_report()
+            er.set_error_msg("RECONNECTING")
+            self.exe_report_callback(er)
 
     def cancel_order(self, req: PyCancelOrderRequest):
         self.logger.info("[bybit_gw_cxl_ord.json]%s" % req)
-        if not self.stream_cli or not self.trade_cli:
-            self._on_abort(req.key(), "cxl", "WebSocket client not created.")
-            return
         bts_id = req.key()
-        ord_data = self.orders.get(bts_id)
+        if self.status != BybitGatewayStatus.READY:
+            self.buffered_requests[req.key()] = req
+            self._on_abort(bts_id, "cxl", "Status is %s" % self.status.name)
+            return
+        if not self.stream_cli or not self.trade_cli:
+            self._on_abort(bts_id, "cxl", "WebSocket client not created.")
+            return
+        ord_data = self._get_order_data(bts_id=bts_id)
         if not ord_data:
-            self._on_abort(req.key(), "cxl", "Did not resolve exch order id")
+            self._on_abort(bts_id, "cxl", "Did not resolve exch order id")
             return
         params = {
             "orderLinkId": ord_data.exch_id,
@@ -446,9 +833,59 @@ class BybitExecutionGateway(ExecutionGateway):
             "symbol": req.exchange_security_id(),
         }
         on_ack, on_err = self._order_handlers("cxl", ord_data.exch_id)
+        if self.status != BybitGatewayStatus.READY:
+            self.buffered_requests[req.key()] = req
+            return
         self.logger.info("[bybit_gw_cxl_ord_send.json]%s" % SLog(params=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)
         self.securities = {sec.name(): sec for sec in securities}
+
+
+if __name__ == "__main__":
+    import argparse
+    from fake_execution_gateway_bridge import FakeExecutionGatewayBridge
+
+    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.add_argument("--show-config", action="store_true", dest="show_config")
+    ap.set_defaults(
+        currencies=None,
+        futures=None,
+        console_level=None,
+        file_level=None,
+        use_testnet=None,
+        show_config=False,
+    )
+    args = ap.parse_args()
+    gw = BybitExecutionGateway()
+    config = gw.config
+    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
+    logger = configure_logger(
+        logging.getLogger("bybit_execution_gateway_app"),
+        "OrderMan",
+        **(config.get("logging") or {}),
+    )
+    if args.show_config:
+        print(json.dumps(config, indent=2))
+        exit()
+    gw_bridge = FakeExecutionGatewayBridge(logger)
+    gw.start(gw_bridge)  # type: ignore
+    while True:
+        time.sleep(1)

src/lib_python_interop/python_interop_plugins/bybit_order_ticket_app.py

@@ -0,0 +1,420 @@
+#!/usr/bin/env python3
+"""
+Simple Bybit Order Ticket (Flask)
+---------------------------------
+A minimal web UI to place/cancel orders. All actual exchange actions are
+abstracted behind the TradingBackend class — replace its placeholder methods
+with calls into your own module.
+
+Run:
+  pip install flask pydantic
+  python bybit_order_ticket_app.py
+Then open http://127.0.0.1:5000
+
+Notes:
+- This app does NO auth/session management and is meant for a local, trusted
+  environment. Add authentication and CSRF protection before exposing it.
+- Input validation is performed with Pydantic on the API boundary.
+- The UI toggles Limit/Market inputs and shows basic responses.
+"""
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import logging
+import os
+from dataclasses import dataclass
+import threading
+import time
+import traceback
+import flask_socketio
+# from flask_socketio import SocketIO, send, emit, join_room, leave_room
+from typing import Any, Dict, List, Optional, Tuple
+
+from flask import Flask, jsonify, render_template_string, request
+from pydantic import BaseModel, Field, validator
+
+from bybit_execution_gateway import BybitExecutionGateway, OrderData
+from bybit_md_provider import BybitMdProvider
+from python_interop_module import PyBookList, PyContractQuantityInfo, PyContractQuantityInfoList, PyContractSpecification, PyCreateOrderRequest, PyExecutionReport, PyFuture, PyPositionInfo, PyPositionInfoList, PySecurity, PySecurityList, PyTradeList
+from fake_execution_gateway_bridge import FakeExecutionGatewayBridge
+from spark_core import StructuredLog as SLog, configure_logger
+
+logger = logging.getLogger()
+
+# -------------- Backend integration points -----------------
+
+class TradingBackend:
+    """Hook these methods up to your Bybit wrapper.
+
+    Replace bodies with calls into your own module. Keep the signatures.
+    Raise exceptions when the exchange reports an error; they'll be surfaced
+    as JSON {"ok": false, "error": "..."} to the UI.
+    """
+
+    def __init__(self, gw:BybitExecutionGateway, md: BybitMdProvider, logger: logging.Logger):
+        self.gw = gw
+        self.gw.order_ack_callback = self.on_order_ack
+        self.logger = logger
+        self.order_ack_event = threading.Event()
+        self.order_acks: list[tuple[dict, Optional[OrderData]]] = []
+        self.books: dict[str, PyBookList] = {}
+        self.trades: list[PyTradeList] = []
+        self.securities: dict[str, PySecurity] = {}
+
+    def disconnect_order_stream(self, req = None):
+        self.gw.stream_cli.ws.close()
+
+    def on_order_ack(self, msg: dict, ord: Optional[OrderData]):
+        self.order_acks.append((msg, ord))
+        self.order_ack_event.set()
+
+    def list_symbols(self) -> List[str]:
+        # TODO: Replace with your symbol discovery (e.g., via /v5/market/instruments)
+        return [
+            "BTCUSDT",
+            "ETHUSDT",
+            "SOLUSDT",
+            "XRPUSDT",
+        ]
+
+    def place_order(
+        self,
+        symbol: str,
+        side: str,  # "Buy" or "Sell"
+        order_type: str,  # "Limit" or "Market"
+        qty: float,
+        price: Optional[float],
+        tif: str,  # GTC/IOC/FOK/PostOnly
+        reduce_only: bool = False,
+        post_only: bool = False,
+        client_order_id: Optional[str] = None,
+        extra: Optional[dict[str, Any]] = None,
+    ) -> dict[str, Any]:
+        req = PyCreateOrderRequest()
+        if client_order_id:
+            req.set_key(int(client_order_id))
+        req.set_venue("BYBIT")
+        req.set_order_type(order_type.upper())
+        req.set_exchange_security_id(symbol)
+        req.set_security_type("FUTURE")
+        req.set_quantity(qty)
+        req.set_is_buy(side == "Buy")
+        if price is not None:
+            req.set_price(price)
+        req.set_time_in_force(tif)
+        self.order_ack_event.clear()
+        try:
+            self.gw.create_order(req)
+        except Exception as e:
+            self.logger.error("[bybit_order_ticket_ord_err.json]%s" % SLog(err=str(e)))
+        if not self.order_ack_event:
+            self.logger.error("[bybit_order_ticket_ord_err.json]%s" % SLog(err="order_ack_event is None"))
+            return {}
+        got_ack = self.order_ack_event.wait(5)
+        if not got_ack:
+            self.logger.error("[bybit_order_ticket_ord_err.json]%s" % SLog(err="Timed out waiting for ack"))
+            return {}
+        ack, ord = self.order_acks[-1]
+        del self.order_acks[-1]
+        return ack
+        # return {
+        #     "status": "accepted",
+        #     "symbol": symbol,
+        #     "side": side,
+        #     "type": order_type,
+        #     "qty": qty,
+        #     "price": price,
+        #     "tif": tif,
+        #     "reduce_only": reduce_only,
+        #     "post_only": post_only,
+        #     "client_order_id": client_order_id,
+        #     "order_id": "test-oid-12345",
+        # }
+
+    def cancel_order(
+        self,
+        *,
+        symbol: str,
+        order_id: Optional[str] = None,
+        client_order_id: Optional[str] = None,
+    ) -> Dict[str, Any]:
+        # TODO: Call your module to cancel; return the exchange response
+        return {
+            "status": "cancel_requested",
+            "symbol": symbol,
+            "order_id": order_id,
+            "client_order_id": client_order_id,
+        }
+
+    def get_balances(self) -> dict[str, Any]:
+        # TODO: Plug into your account/balance endpoint
+        return {"USDT": {"available": 1000.0, "total": 1000.0}}
+
+    def get_positions(self) -> list[dict[str, Any]]:
+        # TODO: Plug into your positions endpoint
+        return []
+
+    def on_execution_report(self, er: PyExecutionReport) -> None:
+        if self.logger:
+            self.logger.info("[exe_rep.json]%s" % er)
+
+    def on_is_ready(self, is_ready: bool) -> None:
+        if self.logger:
+            self.logger.info("[is_ready.json]%s" % SLog(is_ready=is_ready))
+
+    def on_position_info(self, pi: PyPositionInfoList) -> None:
+        if self.logger:
+            self.logger.info("[pos_info.json]%s" % pi)
+
+    def on_books(self, books: PyBookList) -> None:
+        if self.logger:
+            self.logger.debug("[books.json]%s" % books)
+        for b in books:
+            self.books[b.exchange_security_id()] = b
+
+    def on_securities(self, securities: PySecurityList) -> None:
+        if self.logger:
+            self.logger.info("[securities.json]%s" % securities)
+        for s in securities:
+            self.securities[s.exchange_id()] = s
+
+    def on_trades(self, trades: PyTradeList):
+        if self.logger:
+            self.logger.info("[trades.json]%s" % trades)
+        for t in trades:
+            self.trades.insert(0, t)
+        if len(self.trades) > 100:
+            self.trades = self.trades[:100]
+
+
+backend: Optional[TradingBackend] = None
+
+# -------------- Validation models -----------------
+
+class PlaceOrderBody(BaseModel):
+    symbol: str
+    side: str = Field(pattern=r"^(Buy|Sell)$")
+    order_type: str = Field(pattern=r"^(Limit|Market)$")
+    qty: float = Field(gt=0)
+    price: Optional[float] = Field(default=None)
+    tif: str = Field(pattern=r"^(GTC|IOC|FOK|PostOnly)$")
+    reduce_only: bool = False
+    post_only: bool = False
+    client_order_id: Optional[str] = None
+    is_reconnecting: bool = False
+
+    @validator("price", always=True)
+    def price_required_for_limit(cls, v, values):
+        if values.get("order_type") == "Limit" and (v is None or v <= 0):
+            raise ValueError("price is required and > 0 for Limit orders")
+        return v
+
+
+class CancelOrderBody(BaseModel):
+    symbol: str
+    order_id: Optional[str] = None
+    client_order_id: Optional[str] = None
+
+    @validator("client_order_id")
+    def one_of_ids(cls, v, values):
+        if not v and not values.get("order_id"):
+            raise ValueError("Provide order_id or client_order_id")
+        return v
+
+
+# -------------- Flask App -----------------
+app = Flask(__name__)
+socketio = flask_socketio.SocketIO(app, async_mode="threading")
+gw: Optional[BybitExecutionGateway] = None
+
+
+# -- Pages --
+@app.get("/")
+def index():
+    html = ""
+    with open("/workspaces/spark/src/lib_python_interop/python_interop_plugins/bybit_order_ticket_app.xml") as f:
+        html = f.read()
+    return render_template_string(html, symbols=backend.list_symbols())
+
+
+# -- APIs --
+# @app.post("/api/reconnect")
+# def handle_reconnect():
+#     print("Simulating reconnect")
+#     if gw.stream_cli:
+#         gw.stream_cli.ws.close()
+#         return jsonify({"ok": True})
+#     return jsonify({"ok": False})
+#     # socketio.send(f"Echo: {msg}", broadcast=True)
+
+@socketio.on('message')
+def handle_reconnect(msg):
+    logger.info("[reconnect_req.json]%s" % msg)
+    req = json.loads(msg)
+    delay_secs = int(req.get("reconnect_after") or 1)
+    socketio.send(f"Reconnecting in {delay_secs}s")
+    if backend:
+        def delay_impl():
+            # time.sleep(delay_secs)
+            backend.disconnect_order_stream(req)
+        # threading.Thread(target=delay_impl)
+        delay_impl()
+
+@app.get("/api/symbols")
+def api_symbols():
+    return jsonify({"ok": True, "symbols": backend.list_symbols()})
+
+
+@app.get("/api/balances")
+def api_balances():
+    try:
+        return jsonify({"ok": True, "balances": backend.get_balances()})
+    except Exception as e:
+        return jsonify({"ok": False, "error": str(e)}), 500
+
+
+@app.get("/api/positions")
+def api_positions():
+    try:
+        return jsonify({"ok": True, "positions": backend.get_positions()})
+    except Exception as e:
+        return jsonify({"ok": False, "error": str(e)}), 500
+
+
+@app.post("/api/order")
+def api_place_order():
+    try:
+        data = PlaceOrderBody(**request.get_json(force=True))
+    except Exception as e:
+        return jsonify({"ok": False, "error": f"Invalid request: {e}"}), 400
+
+    if backend is None:
+        return jsonify({"ok": False, "result": "Backend not ready"})
+    try:
+        resp = backend.place_order(
+            symbol=data.symbol,
+            side=data.side,
+            order_type=data.order_type,
+            qty=data.qty,
+            price=data.price,
+            tif=data.tif,
+            reduce_only=data.reduce_only,
+            post_only=data.post_only,
+            client_order_id=data.client_order_id,
+        )
+        return jsonify({"ok": True, "result": resp})
+    except Exception as e:
+        return jsonify({"ok": False, "error": str(e)}), 500
+
+
+@app.post("/api/cancel")
+def api_cancel_order():
+    try:
+        data = CancelOrderBody(**request.get_json(force=True))
+    except Exception as e:
+        return jsonify({"ok": False, "error": f"Invalid request: {e}"}), 400
+
+    try:
+        resp = backend.cancel_order(
+            symbol=data.symbol,
+            order_id=data.order_id,
+            client_order_id=data.client_order_id,
+        )
+        return jsonify({"ok": True, "result": resp})
+    except Exception as e:
+        return jsonify({"ok": False, "error": str(e)}), 500
+
+
+class MyExecutionGatewayBridge:
+    def __init__(self, logger: Optional[logging.Logger] = logging.getLogger()):
+        self.logger = logger
+
+    def on_execution_report(self, er: PyExecutionReport) -> None:
+        if self.logger:
+            self.logger.info("[exe_rep.json]%s" % er)
+
+    def on_is_ready(self, is_ready: bool) -> None:
+        if self.logger:
+            self.logger.info("[is_ready.json]%s" % SLog(is_ready=is_ready))
+
+    def on_position_info(self, pi: PyPositionInfoList) -> None:
+        if self.logger:
+            self.logger.info("[pos_info.json]%s" % pi)
+
+
+def run_gw(gw: BybitExecutionGateway, bridge):
+    gw.start(bridge)
+    secs = PySecurityList()
+    sec = PySecurity()
+    sec.set_id(2)
+    sec.set_exchange_id("BTCUSDT")
+    sec.set_currency("USDT")
+    sec.set_group("BTC")
+    sec.set_name("BTCUSDT")
+    sec.set_asset("BTC")
+    fut = PyFuture()
+    fut.set_expiration("2100-01-01")
+    sec.set_future(fut)
+    spec = PyContractSpecification()
+    qty_infos = PyContractQuantityInfoList()
+    qi = PyContractQuantityInfo()
+    qi.set_tick_base(100)
+    qi.set_tick_unit(1)
+    qi.set_max_price(150000)
+    qi.set_min_value(0.003)
+    qi.set_max_value(1)
+    qi.set_multiplier(1)
+    qty_infos.append(qi)
+    spec.set_quantity_infos(qty_infos)
+    nqty_infos = PyContractQuantityInfoList()
+    spec.set_notional_quantity_infos(nqty_infos)
+    secs.append(sec)
+    gw.on_securities(secs)
+
+
+def run_md(md: BybitMdProvider, bridge):
+    md.start(bridge)
+
+
+if __name__ == "__main__":
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--port", type=int, help="Web server port")
+    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.add_argument("--show-config", action="store_true", dest="show_config")
+    ap.set_defaults(
+        currencies=None,
+        futures=None,
+        console_level=None,
+        file_level=None,
+        use_testnet=None,
+        show_config=False,
+    )
+    args = ap.parse_args()
+    gw = BybitExecutionGateway()
+    config = gw.config
+    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
+    if args.show_config:
+        print(json.dumps(config, indent=2))
+        exit()
+    logger = configure_logger(
+        logging.getLogger("bybit_execution_gateway_app"), "OrderMan", **(config.get("logging") or {})
+    )
+    md = BybitMdProvider()
+    backend = TradingBackend(gw, md, logger)
+    run_gw(gw, backend)
+    # run_md(md, backend)
+    app.run(host=os.getenv("HOST", "127.0.0.1"), port=int(os.getenv("PORT", "5000")), debug=os.getenv("FLASK_DEBUG", "0") == "1")

src/lib_python_interop/python_interop_plugins/bybit_order_ticket_app.xml

@@ -0,0 +1,234 @@
+<!doctype html>
+<html lang="en">
+<head>
+  <meta charset="utf-8" />
+  <meta name="viewport" content="width=device-width, initial-scale=1" />
+  <title>Bybit Order Ticket</title>
+  <link rel="preconnect" href="https://fonts.googleapis.com">
+  <link rel="preconnect" href="https://fonts.gstatic.com" crossorigin>
+  <link href="https://fonts.googleapis.com/css2?family=Inter:wght@400;600&display=swap" rel="stylesheet">
+  <script src="https://cdnjs.cloudflare.com/ajax/libs/socket.io/4.0.1/socket.io.js"></script>
+  <style>
+    :root { --bg:#0e1117; --card:#161a22; --muted:#8b949e; --fg:#e6edf3; --accent:#3b82f6; --danger:#ef4444; }
+    * { box-sizing: border-box; }
+    body { margin:0; font-family: Inter, system-ui, -apple-system, Segoe UI, Roboto, sans-serif; background:var(--bg); color:var(--fg); }
+    .wrap { max-width: 920px; margin: 24px auto; padding: 0 16px; }
+    .grid { display: grid; grid-template-columns: 1fr 1fr; gap: 16px; }
+    .card { background: var(--card); border-radius: 16px; padding: 16px; box-shadow: 0 8px 24px rgba(0,0,0,0.25); }
+    label { display:block; font-size: 12px; color: var(--muted); margin-bottom: 6px; }
+    input, select { width: 100%; padding: 10px 12px; background: #0b0f15; color: var(--fg); border: 1px solid #2a2f39; border-radius: 10px; }
+    input:disabled { opacity: .6; }
+    .row { display:grid; grid-template-columns: 1fr 1fr; gap: 12px; }
+    .btnbar { display:flex; gap: 10px; }
+    button { padding: 10px 14px; border:0; border-radius: 12px; cursor:pointer; color:#fff; }
+    .buy { background:#16a34a; }
+    .sell { background:#dc2626; }
+    .ghost { background: #2a2f39; color: var(--fg); }
+    .muted { color: var(--muted); font-size: 12px; }
+    .toast { position: fixed; right: 16px; bottom: 16px; background:#0b0f15; border:1px solid #2a2f39; padding: 12px 14px; border-radius: 12px; box-shadow: 0 6px 24px rgba(0,0,0,.4); display:none; }
+    table { width:100%; border-collapse: collapse; }
+    th, td { text-align:left; padding:8px 6px; border-bottom:1px solid #232833; font-size: 13px; }
+  </style>
+</head>
+<body>
+  <div class="wrap">
+    <h2 style="margin: 6px 0 18px 0;">Bybit Order Ticket</h2>
+
+    <div class="grid">
+      <div class="card">
+        <form id="ticket">
+          <div class="row">
+            <div>
+              <label>Symbol</label>
+              <select id="symbol" required>
+                {% for s in symbols %}
+                  <option value="{{s}}">{{s}}</option>
+                {% endfor %}
+              </select>
+            </div>
+            <div>
+              <label>Order Type</label>
+              <select id="otype">
+                <option value="Limit" selected>Limit</option>
+                <option value="Market">Market</option>
+              </select>
+            </div>
+          </div>
+
+          <div class="row" style="margin-top: 12px;">
+            <div>
+              <label>Quantity</label>
+              <input id="qty" type="number" min="0" step="any" placeholder="e.g. 0.01" required />
+            </div>
+            <div>
+              <label>Limit Price</label>
+              <input id="price" type="number" min="0" step="any" placeholder="e.g. 65000" />
+            </div>
+          </div>
+
+          <div class="row" style="margin-top: 12px;">
+            <div>
+              <label>Time In Force</label>
+              <select id="tif">
+                <option value="GTC">GTC</option>
+                <option value="IOC">IOC</option>
+                <option value="FOK">FOK</option>
+                <option value="PostOnly">PostOnly</option>
+              </select>
+            </div>
+            <div>
+              <label>Client Order ID (optional)</label>
+              <input id="cid" type="text" placeholder="my-oid-123" />
+            </div>
+          </div>
+
+          <div class="row" style="margin-top: 12px;">
+            <div>
+              <label><input type="checkbox" id="reduce_only"/> Reduce-Only</label>
+            </div>
+            <div>
+              <label><input type="checkbox" id="post_only"/> Post-Only</label>
+            </div>
+          </div>
+
+          <div class="btnbar" style="margin-top: 14px;">
+            <button type="button" class="buy" onclick="submitOrder('Buy')">Buy</button>
+            <button type="button" class="sell" onclick="submitOrder('Sell')">Sell</button>
+            <button type="button" class="ghost" onclick="cancelByClientId()">Cancel by ClientId</button>
+            <button type="button" class="Reconnect" onclick="reconnect()">Reconnect websocket</button>
+          </div>
+          <div class="muted" style="margin-top:8px;">Tip: Limit orders require price; Market ignores it. PostOnly is only meaningful for Limit.</div>
+        </form>
+      </div>
+
+      <div class="card">
+        <div style="display:flex; justify-content:space-between; align-items:center; margin-bottom:6px;">
+          <strong>Balances</strong>
+          <button class="ghost" onclick="refreshBalances()">Refresh</button>
+        </div>
+        <div id="balances">—</div>
+
+        <div style="display:flex; justify-content:space-between; align-items:center; margin:16px 0 6px 0;">
+          <strong>Positions</strong>
+          <button class="ghost" onclick="refreshPositions()">Refresh</button>
+        </div>
+        <div id="positions">—</div>
+      </div>
+    </div>
+
+    <div class="card" style="margin-top:16px;">
+      <div style="display:flex; justify-content:space-between; align-items:center;">
+        <strong>API Result</strong>
+        <button class="ghost" onclick="clearResult()">Clear</button>
+      </div>
+      <pre id="result" style="white-space:pre-wrap; overflow:auto; max-height: 280px; background:#0b0f15; padding:12px; border-radius:10px; margin-top:8px;"></pre>
+    </div>
+  </div>
+
+  <div class="toast" id="toast"></div>
+
+  <script>
+    const otype = document.getElementById('otype');
+    const price = document.getElementById('price');
+    var socket = io();
+    // const ws = new WebSocket('ws://127.0.0.1:5000/gateway/status');
+    // ws.addEventListener('message', (event) => {
+    //   console.log("Recv ws: ", event.data);
+    // });
+
+    socket.on("message", function(data) {
+      console.log("Socket recv - ", data);
+    });
+
+    function updatePriceState(){
+      const isLimit = otype.value === 'Limit';
+      price.disabled = !isLimit;
+      if (!isLimit) price.value = '';
+    }
+    otype.addEventListener('change', updatePriceState);
+    updatePriceState();
+
+    function toast(msg) {
+      const t = document.getElementById('toast');
+      t.textContent = msg;
+      t.style.display = 'block';
+      setTimeout(()=> t.style.display = 'none', 2500);
+    }
+
+    function getTicketPayload(side) {
+      const payload = {
+        symbol: document.getElementById('symbol').value,
+        side,
+        order_type: document.getElementById('otype').value,
+        qty: parseFloat(document.getElementById('qty').value),
+        price: document.getElementById('price').value ? parseFloat(document.getElementById('price').value) : null,
+        tif: document.getElementById('tif').value,
+        reduce_only: document.getElementById('reduce_only').checked,
+        post_only: document.getElementById('post_only').checked,
+        client_order_id: document.getElementById('cid').value || null,
+      };
+      return payload;
+    }
+
+    async function submitOrder(side) {
+      const body = getTicketPayload(side);
+      const res = await fetch('/api/order', { method:'POST', headers:{'Content-Type':'application/json'}, body: JSON.stringify(body) });
+      const json = await res.json();
+      document.getElementById('result').textContent = JSON.stringify(json, null, 2);
+      toast(json.ok ? 'Order sent' : `Error: ${json.error}`);
+    }
+
+    async function cancelByClientId(){
+      const symbol = document.getElementById('symbol').value;
+      const cid = document.getElementById('cid').value;
+      if (!cid) { toast('Enter a Client Order ID first'); return; }
+      const res = await fetch('/api/cancel', { method:'POST', headers:{'Content-Type':'application/json'}, body: JSON.stringify({ symbol, client_order_id: cid }) });
+      const json = await res.json();
+      document.getElementById('result').textContent = JSON.stringify(json, null, 2);
+      toast(json.ok ? 'Cancel requested' : `Error: ${json.error}`);
+    }
+
+    async function reconnect() {
+      // const body = { reconnect_after: 5 };
+      // const res = await fetch('/api/reconnect', { method: 'POST', headers: {'Content-Type': 'application/json'}, body: JSON.stringify(body)});
+      const body = { reconnect_after: 5 };
+      socket.send(JSON.stringify(body));
+      // const res = await fetch('/api/reconnect', { method:'POST', headers:{'Content-Type':'application/json'}, body: JSON.stringify(body) });
+      // const json = await res.json();
+    }
+
+    function renderTable(containerId, obj) {
+      const el = document.getElementById(containerId);
+      if (!obj || (Array.isArray(obj) && obj.length === 0)) { el.textContent = '—'; return; }
+      if (Array.isArray(obj)) {
+        const keys = Array.from(new Set(obj.flatMap(o => Object.keys(o))));
+        const rows = obj.map(o => `<tr>${keys.map(k=>`<td>${escapeHtml(String(o[k] ?? ''))}</td>`).join('')}</tr>`).join('');
+        el.innerHTML = `<div style="overflow:auto"><table><thead><tr>${keys.map(k=>`<th>${escapeHtml(k)}</th>`).join('')}</tr></thead><tbody>${rows}</tbody></table></div>`;
+      } else {
+        const rows = Object.entries(obj).map(([k,v])=>`<tr><th>${escapeHtml(k)}</th><td>${escapeHtml(typeof v === 'object' ? JSON.stringify(v) : String(v))}</td></tr>`).join('');
+        el.innerHTML = `<div style="overflow:auto"><table><tbody>${rows}</tbody></table></div>`;
+      }
+    }
+
+    function escapeHtml(s){return s.replace(/[&<>"']/g, c=>({'&':'&amp;','<':'&lt;','>':'&gt;','\"':'&quot;','\'':'&#039;'}[c]))}
+
+    async function refreshBalances(){
+      const res = await fetch('/api/balances');
+      const json = await res.json();
+      if (json.ok) renderTable('balances', json.balances); else toast('Error loading balances');
+    }
+
+    async function refreshPositions(){
+      const res = await fetch('/api/positions');
+      const json = await res.json();
+      if (json.ok) renderTable('positions', json.positions); else toast('Error loading positions');
+    }
+
+    // Initial load
+    refreshBalances();
+    refreshPositions();
+
+    function clearResult(){ document.getElementById('result').textContent = ''; }
+  </script>
+</body>
+</html>

src/lib_python_interop/python_interop_plugins/fake_execution_gateway_bridge.py

@@ -0,0 +1,22 @@
+import logging
+
+from typing import Optional
+from spark_core import StructuredLog as SLog
+from python_interop_module import PyExecutionReport, PyPositionInfo
+
+
+class FakeExecutionGatewayBridge:
+    def __init__(self, logger: Optional[logging.Logger] = logging.getLogger()):
+        self.logger = logger
+
+    def on_execution_report(self, er: PyExecutionReport) -> None:
+        if self.logger:
+            self.logger.info("[exe_rep.json]%s" % er)
+
+    def on_is_ready(self, is_ready: bool) -> None:
+        if self.logger:
+            self.logger.info("[is_ready.json]%s" % SLog(is_ready=is_ready))
+
+    def on_position_info(self, pi: PyPositionInfo) -> None:
+        if self.logger:
+            self.logger.info("[pos_info.json]%s" % pi)

src/lib_python_interop/python_interop_plugins/pybit_unified_trading.py

@@ -106,6 +106,12 @@ class WebSocket(_V5WebSocketManager):
 
     # Private topics
 
+    def reconnect(self):
+        if not self.WS_URL:
+            raise ValueError(f"WS_URL is empty")
+        self._reset()
+        self._connect(self.WS_URL)
+
     def position_stream(self, callback):
         """Subscribe to the position stream to see changes to your position data in real-time.
 

src/lib_python_interop/python_interop_plugins/pybit_websocket_stream.py

@@ -1,3 +1,4 @@
+from typing import Callable, Optional
 import websocket
 import threading
 import time
@@ -21,6 +22,8 @@ DOMAIN_ALT = "bytick"
 TLD_MAIN = "com"
 
 
+WebSocketError = websocket.WebSocketConnectionClosedException | ConnectionRefusedError | KeyboardInterrupt | SystemExit | Exception
+
 class _WebSocketManager:
     def __init__(
         self,
@@ -39,7 +42,26 @@ class _WebSocketManager:
         restart_on_error=True,
         trace_logging=False,
         private_auth_expire=1,
+        on_open: Optional[Callable[[websocket.WebSocketApp], None]] = None,
+        on_error : Optional[Callable[[websocket.WebSocketApp, WebSocketError], Optional[bool]]] = None,
+        on_close : Optional[Callable[[websocket.WebSocketApp, Optional[int], Optional[str]], bool]] = None,
     ):
+        """
+        Parameters
+        ----------
+        restart_on_error: bool
+            Reconnect when the websocket has an error.
+
+        on_open: Optional[Callable[[websocket.WebSocketApp], None]]
+            Optional websocket open handler.
+
+        on_error: Optional[Callable[[websocket.WebSocketApp, WebSocketError], Optional[bool]]]
+            Optional websocket error handler. Return a bool to override the 'restart_on_error'
+            param (True forces reconnect, False forces no reconnect).
+
+        on_close: Optional[Callable[[websocket.WebSocketApp, Optional[int], Optional[str]], bool]]
+            Optional websocket close handler. Return True to reconnect.
+        """
         self.testnet = testnet
         self.domain = domain
         self.tld = tld
@@ -50,6 +72,9 @@ class _WebSocketManager:
         self.api_secret = api_secret
 
         self.callback = _callback_function
+        self.open_callback = on_open
+        self.close_callback = on_close
+        self.error_callback = on_error
         self.ws_name = ws_name
         if api_key:
             self.ws_name += " (Auth)"
@@ -69,6 +94,7 @@ class _WebSocketManager:
         self.subscriptions = []
 
         # Set ping settings.
+        self._ping_timer: Optional[threading.Timer] = None
         self.ping_interval = ping_interval
         self.ping_timeout = ping_timeout
         self.custom_ping_message = json.dumps({"op": "ping"})
@@ -85,11 +111,13 @@ class _WebSocketManager:
         self._reset()
         self.attempting_connection = False
 
-    def _on_open(self):
+    def _on_open(self, ws: websocket.WebSocketApp):
         """
-        Log WS open.
+        Fires the open callback if it exists.
         """
-        logger.debug(f"WebSocket {self.ws_name} opened.")
+        logger.info(f"WebSocket {self.ws_name} opened. is_connected: {self.is_connected()}")
+        if self.open_callback:
+            self.open_callback(ws)
 
     def _on_message(self, message):
         """
@@ -153,9 +181,9 @@ class _WebSocketManager:
             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_close=lambda ws, *args: self._on_close(ws, *args), # type: ignore
+                on_open=lambda ws: self._on_open(ws), # type: ignore
+                on_error=lambda ws, err: self._on_error(ws, err), # type: ignore
                 on_pong=lambda ws, *args: self._on_pong(),
             )
 
@@ -216,10 +244,13 @@ class _WebSocketManager:
             )
         )
 
-    def _on_error(self, error):
+    def _on_error(self, ws: websocket.WebSocketApp, error: WebSocketError):
         """
         Exit on errors and raise exception, or attempt reconnect.
         """
+        reconnect = (self.error_callback(ws, error) if self.error_callback else None)
+        if reconnect is None:
+            reconnect = self.handle_error
         if type(error).__name__ not in [
             "WebSocketConnectionClosedException",
             "ConnectionResetError",
@@ -237,15 +268,16 @@ class _WebSocketManager:
             self.exit()
 
         # Reconnect.
-        if self.handle_error and not self.attempting_connection:
-            self._reset()
-            self._connect(self.endpoint)
+        if reconnect and not self.attempting_connection:
+            self._reconnect()
 
-    def _on_close(self):
+    def _on_close(self, ws: websocket.WebSocketApp, code: Optional[int], rsn: Optional[str]):
         """
-        Log WS close.
+        Fires the close callback if it exists.
         """
-        logger.debug(f"WebSocket {self.ws_name} closed.")
+        logger.warning(f"WebSocket {self.ws_name} closed.")
+        if self.close_callback and self.close_callback(ws, code, rsn):
+            self._reconnect()
 
     def _on_pong(self):
         """
@@ -259,15 +291,22 @@ class _WebSocketManager:
         """
         self._send_custom_ping()
 
+    def _reconnect(self):
+        if not self.attempting_connection:
+            self._reset()
+            self._connect(self.endpoint)
+
     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(
+        if self._ping_timer:
+            self._ping_timer.cancel()
+        self._ping_timer = threading.Timer(
             self.ping_interval, self._send_custom_ping
         )
-        timer.start()
+        self._ping_timer.start()
 
     @staticmethod
     def _is_custom_pong(message):
@@ -284,6 +323,8 @@ class _WebSocketManager:
         self.exited = False
         self.auth = False
         self.data = {}
+        if self._ping_timer:
+            self._ping_timer.cancel()
 
     def exit(self):
         """
@@ -303,6 +344,7 @@ class _V5WebSocketManager(_WebSocketManager):
             if kwargs.get("callback_function")
             else self._handle_incoming_message
         )
+        self.subscription_ack_callback: Optional[Callable[[list[str], bool], None]] = kwargs.pop("subscription_ack_callback") if "subscription_ack_callback" in kwargs else None
         super().__init__(callback_function, ws_name, **kwargs)
 
         self.subscriptions = {}
@@ -438,21 +480,25 @@ class _V5WebSocketManager(_WebSocketManager):
             )
 
     def _process_subscription_message(self, message):
-        if message.get("req_id"):
-            sub = self.subscriptions[message["req_id"]]
+        req_id = message.get("req_id")
+        if req_id:
+            sub = self.subscriptions[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]
+            raise ValueError("Subscription msg has no 'req_id' field")
 
+        success = message.get("success")
+        if success is None:
+            raise ValueError("Subscription msg has no 'success' field")
         # If we get successful futures subscription, notify user
-        if message.get("success") is True:
+        if success:
             logger.debug(f"Subscription to {sub} successful.")
-        # Futures subscription fail
-        elif message.get("success") is False:
+        else:
+            # Futures subscription fail
             response = message["ret_msg"]
             logger.error("Couldn't subscribe to topic." f"Error: {response}.")
             self._pop_callback(sub[0])
+        if self.subscription_ack_callback:
+            self.subscription_ack_callback(json.loads(sub).get("args"), success)
 
     def _process_normal_message(self, message):
         topic = message["topic"]

src/lib_python_interop/python_interop_plugins/pybit_websocket_trading.py

@@ -1,5 +1,6 @@
 from dataclasses import dataclass, field
 import json
+from typing import Optional
 import uuid
 import logging
 from pybit_websocket_stream import _WebSocketManager
@@ -20,6 +21,9 @@ class _V5TradeWebSocketManager(_WebSocketManager):
         self.referral_id = referral_id
         self._connect(TRADE_WSS)
 
+    def reconnect(self):
+        self._connect(TRADE_WSS)
+
     def _process_auth_message(self, message):
         # If we get successful auth, notify user
         if message.get("retCode") == 0:
@@ -75,9 +79,9 @@ class _V5TradeWebSocketManager(_WebSocketManager):
     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())
-
+    def _send_order_operation(self, operation, callback, request, error_callback=None, request_id:Optional[str] = None):
+        if not request_id:
+            request_id = str(uuid.uuid4())
         message = {
             "reqId": request_id,
             "header": {

src/lib_python_interop/python_interop_plugins/spark_core.py

@@ -7,6 +7,7 @@ import sys
 import threading
 import time
 from functools import reduce
+from pathlib import Path
 from typing import Any, Callable, Optional
 import websocket
 from python_interop_module import (
@@ -29,7 +30,12 @@ class JsonElement:
         self.value = value
 
     def __str__(self):
-        return json.dumps(self.value)
+        try:
+            to_json = getattr(self.value, "to_json", None)
+            return json.dumps(to_json(self.value) if callable(to_json) else self.value)
+        except Exception as e:
+            print(f"JsonElement formatting error: {e}")
+        return str(self.value)
 
 
 class StructuredLog:
@@ -39,7 +45,11 @@ class StructuredLog:
             kwargs["ts"] = time.time_ns()
 
     def __str__(self):
-        return json.dumps(self.kwargs)
+        try:
+            return json.dumps(self.kwargs)
+        except Exception as e:
+            print(f"StructuredLog formatting error: {e}")
+        return str(self.kwargs)
 
 
 def _log_level_name(level: int):
@@ -139,9 +149,10 @@ def configure_logger(
     stdout_handler.setLevel(kwargs.get("console_level", logging.WARNING))
     stdout_handler.formatter = ConsoleLogFormatter(subsystem=subsystem)
     # File
-    file_handler = logging.FileHandler(
-        f"/home/bts/sparkdata/data/logging.sparkd/{logger.name or 'py'}_log"
-    )
+    log_file=Path(f"/home/bts/sparkdata/data/logging.sparkd/{logger.name or 'py'}_log")
+    if not log_file.parent.exists():
+        os.makedirs(str(log_file.parent), exist_ok=True)
+    file_handler = logging.FileHandler(str(log_file))
     file_handler.setLevel(kwargs.get("file_level", logging.INFO))
     file_handler.formatter = FileLogFormatter(subsystem=subsystem)
     logger.handlers.clear()

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

@@ -0,0 +1,102 @@
+#include "PingpongAlgo.h"
+
+#include "spark/domain_model/Security.h"
+#include "spark/orderman/order_factory/IOrderFactory.h"
+#include "spark/orderman/orderman_utils/Book.h"
+#include "spark/orderman/orderman_utils/orderman_utils.h"
+#include "spark/orderman/orders/algo/IAlgoServices.h"
+#include "spark/orderman/orders/spreader_orders/spreader/SpreaderUtils.h"
+#include "spark/os/os.h"
+#include "spark/utils/PBUtils.h"
+#include "spark/utils/datetime.h"
+#include <stdint.h>
+
+namespace bts {
+namespace orderman {
+
+using namespace api::orderman;
+
+enum class PingpongAlgo::Status { WAITING_FOR_SIGNAL, WORKING, DONE };
+
+struct PingpongAlgo::PImpl {
+   Status state = Status::WAITING_FOR_SIGNAL;
+   IAlgoServices *algo_services = nullptr;
+   SecurityKey sec_key = SecurityKey(0);
+   //PingpongAlgo::PingpongAlgoParams algo_params;
+};
+
+PingpongAlgo::PingpongAlgo() : me(new PImpl()) {}
+PingpongAlgo::~PingpongAlgo() {}
+// initialization /////////////////////////////////////////////////////////////
+///////////////////////////////////////////////////////////////////////////////
+OrderOpResult PingpongAlgo::do_before_launch(
+    IBeforeLaunchAlgoServices & /*serv*/) {
+   return {OrderError::ORDER_ERROR_OK, std::string("OK")};
+}
+void PingpongAlgo::do_launch(IAlgoServices &algo_services) {
+   me->algo_services = &algo_services;
+   _algo().subscribe_status(me->sec_key);
+   //_algo().subscribe_timer();
+}
+void PingpongAlgo::do_launch_attached(
+    IAlgoServices &algo_services, OrderKey, ChildState) {
+   me->algo_services = &algo_services;
+   _cxl_and_end(true, "TAS Algo can not be attached to working orders");
+}
+// updates ////////////////////////////////////////////////////////////////////
+///////////////////////////////////////////////////////////////////////////////
+void PingpongAlgo::do_update(const api::orderman::Update &) {
+   LOG(OrderMan, Warn,
+       "PingpongAlgo::do_update order:{} state:{} update not allowed ",
+       _algo().order_key().value(), static_cast<int>(me->state));
+}
+// callbacks //////////////////////////////////////////////////////////////////
+///////////////////////////////////////////////////////////////////////////////
+void PingpongAlgo::on_book(SecurityKey, const Book &) {
+   // find the mean and std dev. for std dev, we'll most likely have to keep a sliding window of the
+   // bids and asks, perhaps a std::vector<std::pair<double, double>> and we recompute it everytime.
+
+   // we put buy and sell open pos orders at μ + ol.σ and μ + os.σ respectively
+   // (ol will be < 0, for example when ol = -2, our buy order will go at μ - 2σ).
+   // when the open long pos order gets filled, we send the close long position order at  μ + cl.σ.
+   // when the buy open pos order gets filled, we send the close long position order at  μ + cl.σ.
+   // similiarly for short.
+}
+void PingpongAlgo::on_child_change(OrderKey, ChildState const &) {}
+void PingpongAlgo::on_status(SecurityKey, co2::TradingStatus) {}
+void PingpongAlgo::on_timer() {}
+IAlgoServices &PingpongAlgo::_algo() const { return *me->algo_services; }
+///////////////////////////////////////////////////////////////////////////////
+
+ChildTemplate *PingpongAlgo::_create_order_template(
+    IBeforeLaunchAlgoServices &, SecurityKey, api::Side) {
+   return nullptr;
+}
+void PingpongAlgo::_switch_state(PingpongAlgo::Status new_state) {
+   switch (new_state) {
+   case (PingpongAlgo::Status::WAITING_FOR_SIGNAL):
+      bts_assert(false);
+      break;
+   case (PingpongAlgo::Status::WORKING):
+      bts_assert(me->state == PingpongAlgo::Status::WAITING_FOR_SIGNAL);
+      break;
+   case (PingpongAlgo::Status::DONE):
+      bts_assert(me->state != PingpongAlgo::Status::DONE);
+      break;
+   }
+   LOG(OrderMan, Info, "PingpongAlgo::_switch_state key:{} {}->{} ",
+       _algo().order_key(), static_cast<int>(me->state),
+       static_cast<int>(new_state));
+   me->state = new_state;
+}
+
+void PingpongAlgo::_cxl_and_end(bool is_error, const std::string &err_msg) {
+   if (is_error)
+      _algo().terminate_error(err_msg);
+   else
+      _algo().complete();
+   _switch_state(PingpongAlgo::Status::DONE);
+}
+
+} // namespace orderman
+} // namespace bts

src/lib_spark/spark/orderman/orders/algo/algo_orders/PingpongAlgo.h

@@ -0,0 +1,38 @@
+#pragma once
+
+#include "spark/orderman/orders/algo/IAlgoOrder.h"
+
+namespace bts::orderman {
+
+class ChildTemplate;
+
+class PingpongAlgo : public IAlgoOrder {
+ public:
+   PingpongAlgo();
+   ~PingpongAlgo();
+
+ public:
+   OrderOpResult do_before_launch(IBeforeLaunchAlgoServices &) override;
+   void do_launch(IAlgoServices &) override;
+   void do_launch_attached(IAlgoServices &, OrderKey, ChildState) override;
+   void do_update(api::orderman::Update const &) override;
+   void on_book(SecurityKey, Book const &) override;
+   void on_child_change(OrderKey, ChildState const &) override;
+   void on_status(SecurityKey, co2::TradingStatus) override;
+   void on_timer() override;
+
+ private:
+   enum class Status;
+   struct PImpl;
+   PImpl *me;
+   //std::unique_ptr<PImpl> me;
+
+ private:
+   IAlgoServices &_algo() const;
+   static ChildTemplate *_create_order_template(
+       IBeforeLaunchAlgoServices &, SecurityKey, api::Side);
+   // acts on md, cognizant of the state
+   void _switch_state(Status new_status);
+   void _cxl_and_end(bool is_error, const std::string &err_msg);
+};
+} // namespace bts::orderman

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

@@ -1,4 +1,3 @@
-
 #include "DormanAlgo.h"
 #include "OnSignalAlgo.h"
 #include "PyramidAlgo.h"
@@ -11,6 +10,7 @@
 #include "spark/config/protos/toggles.pb.h"
 #include "spark/orderman/orderman_utils/orderman_utils.h"
 #include "spark/orderman/orders/algo/algo_orders/LimitCoverAlgo.h"
+#include "spark/orderman/orders/algo/algo_orders/SprayAndSkewAlgo.h"
 #include "spark/orderman/orders/algo/algo_orders/TaskingEyeAlgo.h"
 #include "spark/orderman/orders/algo/algo_orders/VolAlgo.h"
 #include "spark/orderman/orders/algo/algo_orders/limit_cover/LimitCoverAdapter.h"
@@ -238,11 +238,11 @@ extern "C" void init_quoting_algo_plugin(bts::SparkServices &srv) {
    quoting_algo.set_dynamic_order_type("QuotingAlgo");
    *quoting_algo.add_param_desc() = param_int32("n_ticks");
    *quoting_algo.add_param_desc() = param_order("child_order");
-   srv.register_dynamic_algo_order(
-       quoting_algo, []() { return std::make_unique<orderman::QuotingAlgo>(); });
+   srv.register_dynamic_algo_order(quoting_algo, []() {
+      return std::make_unique<orderman::QuotingAlgo>();
+   });
 }
 
-
 extern "C" void init_on_signal_plugin(bts::SparkServices &srv) {
    DynamicOrderDescriptor on_signal;
    on_signal.set_dynamic_order_type("OnSignal");
@@ -307,6 +307,25 @@ extern "C" void init_execution_tester_plugin(bts::SparkServices &srv) {
       return std::make_unique<orderman::ExecutionTesterAlgo>();
    });
 }
+extern "C" void init_spray_and_skew_plugin(bts::SparkServices &srv) {
+   DynamicOrderDescriptor spray_and_skew;
+   spray_and_skew.set_dynamic_order_type("Spray and Skew");
+   // contracts
+   *spray_and_skew.add_param_desc() = param_double("Half Spread");
+   *spray_and_skew.add_param_desc() = param_double("Skew");
+   *spray_and_skew.add_param_desc() = param_double("C1");
+   *spray_and_skew.add_param_desc() = param_uint32("Looking Depth Levels");
+   *spray_and_skew.add_param_desc() = param_uint32("Window");
+   *spray_and_skew.add_param_desc() = param_double("Interval Sec");
+   *spray_and_skew.add_param_desc() = param_uint32("Grid Num");
+   *spray_and_skew.add_param_desc() = param_double("Grid Interval");
+   *spray_and_skew.add_param_desc() = param_double("Order Size");
+   *spray_and_skew.add_param_desc() = param_double("Max Position Dollar");
+
+   srv.register_dynamic_algo_order(spray_and_skew, []() {
+      return std::make_unique<orderman::SprayAndSkewAlgo>();
+   });
+}
 
 extern "C" void init_task_eye_plugin(bts::SparkServices &srv) {
    DynamicOrderDescriptor tasking_eye_algo;
@@ -350,6 +369,22 @@ extern "C" void init_vol_algo_plugin(bts::SparkServices &srv) {
        vol_algo, []() { return std::make_unique<orderman::VolAlgo>(); });
 }
 
+extern "C" void init_pingpong_plugin(bts::SparkServices &srv) {
+   DynamicOrderDescriptor pingpong_algo;
+   pingpong_algo.set_dynamic_order_type("Pingpong");
+   // μ + ol.σ
+   *pingpong_algo.add_param_desc() = param_double("Open Long Pos");
+   //μ + cl.σ
+   *pingpong_algo.add_param_desc() = param_double("Close Long Pos");
+   //  μ + os.σ
+   *pingpong_algo.add_param_desc() = param_double("Open Short Pos");
+   //  μ + cs.σ
+   *pingpong_algo.add_param_desc() = param_double("Close Short Pos");
+   *pingpong_algo.add_param_desc() = param_uint32("Sliding Window Seconds");
+   srv.register_dynamic_algo_order(
+       pingpong_algo, []() { return std::make_unique<orderman::VolAlgo>(); });
+}
+
 extern "C" void init_spark_plugin(bts::SparkServices &srv) {
    auto &dyn_algo_config = config::get<config::toggles::DynamicAlgos>();
    if (dyn_algo_config.enable_dorman())
@@ -369,9 +404,11 @@ extern "C" void init_spark_plugin(bts::SparkServices &srv) {
    if (dyn_algo_config.enable_execution_tester())
       init_execution_tester_plugin(srv);
    // init_sniper_plugin(srv);
+   init_spray_and_skew_plugin(srv);
    init_task_eye_plugin(srv);
    init_vol_algo_plugin(srv);
    init_quoting_algo_plugin(srv);
+   init_pingpong_plugin(srv);
 }
 
 } // namespace dorman

src/lib_spark/spark/orderman/orders/algo/algo_orders/SprayAndSkewAlgo.h

@@ -0,0 +1,57 @@
+#pragma once
+
+#include "spark/domain_model/SecurityKey.h"
+#include "spark/domain_model/common.h"
+#include "spark/orderman/orders/algo/IAlgoOrder.h"
+#include <openssl/ossl_typ.h>
+
+namespace bts::orderman {
+
+/*
+Spray tiny limit orders on both sides of the book every 500 ms (here via obi_mm), inside a very tight half-spread (≈1 bp).
+Lean (skew) the quotes away from your inventory—if buys fill and you're long, pull bids a tick lower and lift offers a tick higher—so the position mean-reverts instead of snowballing.
+Because fills come to you passively, each trade earns the maker rebate (-0.005 %), and the micro-edge from skewing usually caps adverse selection; over thousands of fills the rebates > tiny pnl bleed, so the net is positive even if price action is noise.
+
+The minimal loop looks like:
+while hbt.elapse(0.5e9) == 0:          # every 0.5 s
+    mid  = (best_bid + best_ask)/2
+    fair = mid + c1 * alpha            # alpha = std-score of OBI std score should be normalized between 1  and -1, c1 would be something lik 1 or 10
+    rp   = fair - skew * (pos/qty)     # inventory-adjusted reservation price, skew 1 and 10, dividing by quanity to make sure in ticks 
+    bid  = max(np.floor(rp - half_spread), best_bid) #half_spreads 50 or 100 ticks number of ticks
+    ask  = min(np.ceil (rp + half_spread), best_ask)
+    make_grid(bid, ask, qty)           # post / cancel as in obi_mm 
+    max_position_dollar # set as parameter, max position you can hold in terms of dollars and if this is met hold everything on that side keep algo running, algo max position in terms of contracts 
+    cancel and update: 
+    on update
+
+    0.005%
+
+That's all: churn volume, stay flat, clip rebates. Tweak half_spread, skew, and order_qty until the "rebate pnl" outweighs fees and slippage.*/
+struct custom_state;
+class SprayAndSkewAlgo : public IAlgoOrder {
+ public:
+   SprayAndSkewAlgo();
+   ~SprayAndSkewAlgo();
+   OrderOpResult do_before_launch(IBeforeLaunchAlgoServices &) override;
+   void on_book(SecurityKey, Book const &) override;
+   void do_launch(IAlgoServices &) override;
+   void do_launch_attached(IAlgoServices &, OrderKey, ChildState) override;
+   void do_update(api::orderman::Update const &) override;
+   void on_child_change(OrderKey, ChildState const &) override;
+   void on_status(SecurityKey, co2::TradingStatus) override;
+   void on_timer() override;
+   void on_position(PositionUpdate const &) override;
+   double get_rp(
+    double stddev, double imbalance, double mean, double mid);
+
+ private:
+   struct PImpl;
+   std::unique_ptr<PImpl> me;
+
+ private:
+   void _terminate_algo(bool is_error, const std::string &err_msg);
+   void update_custom_state(const custom_state &cs);
+   void timed_handler();
+};
+
+} // namespace bts::orderman

src/lib_spark/spark/plugin/plugin.cpp

@@ -22,6 +22,7 @@ void init_plugins(SparkServices &services) {
        boost::filesystem::path(), ""))>
        keep_alive;
 
+       init_spark_plugin(services);
    auto plugin_path = os::sparkd_wd() + "/plugins/";
    if (!boost::filesystem::is_directory(plugin_path)) {
       LOG(Plugin, Info,
@@ -31,7 +32,6 @@ void init_plugins(SparkServices &services) {
    }
 
    // initialize the plugins in libspark.so
-   init_spark_plugin(services);
 
    boost::filesystem::directory_iterator end;
    for (auto iter = boost::filesystem::directory_iterator(plugin_path);