Merged in PIQachu (pull request #223)

RTm: spark-2316 mbo support for simulator

Approved-by: Pedro Pinto <pedro.pinto+atlassian@bluetradesys.com>

doc/user/build.sh

@@ -7,7 +7,7 @@ zola_path=$(readlink -f "../../externals/linux/bin/zola/zola")
 
 mkdir -p $temp_path
 cp -r ./* $temp_path
-cp -r ../../src/sdk/doc/* $temp_path/content/bts_spark_sdk/Server/
+
 
 cd $temp_path
 $zola_path build -u '/docs/' -o $1 >$tmpfile 2>&1

src/client/spark/view/Spark/Ladder/LadderGridHost.cs

@@ -97,7 +97,7 @@ namespace mbts.spark {
 
       public event Action visible_columns_changed;
 
-      public IEnumerable<ColumnType> visible_columns =>
+      public IEnumerable<ColumnType> visible_column_types =>
          columns.Where(c => c.settings.show)
          .Select(c => c.column_type);
 
@@ -253,6 +253,7 @@ namespace mbts.spark {
       #endregion
 
       Column[] columns { get; set; } = new Column[0];
+      IEnumerable<Column> _visible_columns => columns.Where(c => c.settings.show);
 
       public void drag_out_of_bounds() =>
          _tactile_adorner?.drag_out_of_bounds();
@@ -330,7 +331,7 @@ namespace mbts.spark {
 
       #region Column Utility Methods
       Column _column_for_point(Point point) =>
-         columns.FirstOrDefault(c => c.x_right > point.X, columns.Length > 0 ? columns[columns.Length - 1] : Column.Empty);
+         _visible_columns.FirstOrDefault(c => c.x_right > point.X, columns.Length > 0 ? columns[columns.Length - 1] : Column.Empty);
 
       double _column_start(ColumnType col_type) =>
          columns.TakeWhile(c => c.column_type != col_type).Sum(c => c.settings.width);
@@ -913,4 +914,4 @@ namespace mbts.spark {
          }
       }
    }
-}
+}

src/client/spark/view/Spark/Ladder/LadderSettings.cs

@@ -38,6 +38,14 @@ namespace mbts.spark {
          font_weight = FontWeights.SemiBold.ToOpenTypeWeight()
       };
 
+      public static LadderColumnSettings DefaultPiqColumnSettings => new LadderColumnSettings {
+         show = false,
+         width = 50,
+         font_family = "Segoe UI",
+         font_size = 10,
+         font_weight = FontWeights.SemiBold.ToOpenTypeWeight()
+      };
+
       [ProtoMember(1)]
       public bool show { get; set; }
       [ProtoMember(2)]
@@ -250,7 +258,7 @@ namespace mbts.spark {
                ask_column = LadderColumnSettings.Default,
                price_column = LadderColumnSettings.Default,
                last_column = LadderColumnSettings.Default,
-               piq_column = LadderColumnSettings.Default;
+               piq_column = LadderColumnSettings.DefaultPiqColumnSettings;
             orders_column.background = BTSColors.LadderSpecial.default_column_background;
             bid_column.foreground = 0xFFFFFF;
             bid_column.background = BTSColors.MarketData.bid;

src/client/spark/view/Spark/Ladder/LadderView.xaml.cs

@@ -15,7 +15,7 @@ namespace mbts.spark {
          InitializeComponent();
          _column_headers_presenter.Columns = new GridViewColumnCollection();
          _ladder.visible_columns_changed += () => _column_headers_presenter.Columns.SetItems(
-            _ladder.visible_columns.Select(t => ColumnFactory.create(_ladder, t)));
+            _ladder.visible_column_types.Select(t => ColumnFactory.create(_ladder, t)));
          this.vm = vm;
          DataContext = vm;
          DataContextChanged += datacontext_changed;

src/client/spark_api/Co2/Co2NativeMethods.cs

@@ -35,6 +35,7 @@ namespace mbts.co2
          public readonly IntPtr hot;
          public readonly IntPtr cold_data;
          public readonly IntPtr cold_header;
+         private readonly IntPtr unused_mbo;
       }
 
       [DllImport("libspark")]

src/client/spark_test/System/LimitsTests.cs

@@ -78,7 +78,7 @@ namespace bts.tests
          return securities;
       }
 
-   [Test(Timeout = 30)]
+      [Test(Timeout = 30)]
       public async Task test_position_limits() {
          string test_symbol = "CL";
 

src/client/sparkcmd/Cmd/MBOCommand.cs

@@ -0,0 +1,56 @@
+using System;
+using System.Collections.Generic;
+using System.Reflection;
+using bts.api.core;
+using bts.md;
+using CommandLine;
+using System.Linq;
+
+namespace bts {
+
+   [Verb("mbo", HelpText="MBO diagnostic command")]
+   public class MBOCommandOptions {
+      [Option('k', "key", Required = true)]
+      public UInt32 key { get; set; }
+
+      [Option('s', "side", Default = api.Side.Undefined)]
+      public api.Side side { get; set; }
+   }
+
+   public class MBOCommand {
+      public static int Run(MBOCommandOptions o) {
+         dump_mbo(o.key, o.side);
+         return 0;
+      }
+      
+      static void dump_mbo(UInt32 key, api.Side side) {
+         using (var ctx = new ServiceContext()) {
+            var mbo = ctx.locate<MBOService>();
+            var data = mbo.get_orders_for_security(new MBOSecurityQuery { SecurityId = key });
+
+            Console.WriteLine($"Order data for {data.SecurityId}");
+            IEnumerable<OrderPosition> orders = data.Bid;
+            switch(side) {
+               case api.Side.Undefined:
+                  orders = data.Ask.Concat(data.Bid);
+               break;
+               case api.Side.Buy:
+                  orders = data.Bid;
+               break;
+               case api.Side.Sell:
+                  orders = data.Ask;
+               break;
+            }
+            var price = 0.0;
+            foreach(var e in orders) {
+               if (e.Price != price) {
+                  price = e.Price;
+                  Console.WriteLine("================");
+               }
+               Console.WriteLine(e);
+            }
+         }
+      }
+   }
+
+}

src/client/sparkcmd/Cmd/OrderStress.cs

@@ -0,0 +1,157 @@
+using bts.api;
+using bts.api.core;
+using bts.api.orderman;
+using bts.co2;
+using bts.test;
+using CommandLine;
+using mbts.co2;
+using mbts.config;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading.Tasks;
+
+namespace bts {
+
+   [Verb("orderstress", HelpText = "Create a bunch of orders")]
+   public class OrderStressOptions {
+      // Maybe options to create fewer orders or sleep
+
+      [Option('s', "sleep", HelpText = "Sleep ms between order submission", Default = 100)]
+      public int sleep { get; set; }
+
+      [Option('d', "depth", HelpText = "How many orders per side", Default = 10u)]
+      public uint depth { get; set; }
+
+      [Option('q', "quanityt", HelpText = "Order quantity", Default = 1u)]
+      public uint quantity { get; set; }
+   };
+
+   public struct Market {
+      public SecurityImmutable security;
+      public IntPrice? bid;
+      public IntPrice? ask;
+   }
+
+   public static class OrderStress {
+      static IEnumerable<Market> all_markets(ServiceContext ctx) {
+         var channels =
+            UserSymbol.get_all()
+            .Select(u =>
+               new SecurityChannel() {
+                  Time = 0,
+                  Flags = (int)(MDRequest.Types.Flags.Direct | MDRequest.Types.Flags.Implied),
+                  Type = u.type,
+                  Symbol = u.symbol,
+                  Exchange = u.exchange
+               }
+            );
+         var request = new MDRequest() { Host = "system_test" };
+         request.Channels.AddRange(channels);
+
+         var relay = ctx.locate<MarketDataRelay>();
+         System.Threading.Thread.Sleep(100);
+         var result = relay.requestmd(request);
+
+         foreach (var ch in result.Channels) {
+            foreach (var update in ch.Updates) {
+               if (update.Security != null) {
+                  var m = new Market() {
+                     security = new SecurityImmutable(update.Security),
+                     bid = best(
+                        (a,b) => a > b ? a : b,
+                        update.Direct?.Bids?.FirstOrDefault(), update.Implied?.Bids?.FirstOrDefault()),
+                     ask = best(
+                        (a,b) => a < b ? a : b,
+                        update.Direct?.Asks?.FirstOrDefault(), update.Implied?.Asks?.FirstOrDefault())
+                  };
+                  if (m.bid.HasValue || m.ask.HasValue) {
+                     yield return m;
+                  }
+               }
+            }
+         }
+      }
+
+      private static IntPrice? best(Func<IntPrice, IntPrice,IntPrice> comp, params PriceSizeMessage[] ps) {
+         IntPrice? result = null;
+         foreach (var p in ps) {
+            if (p != null) {
+               var ip = new IntPrice(p.Mantissa, (sbyte)p.Exponent);
+               if (result == null) {
+                  result = ip;
+               }
+               else {
+                  result = comp(result.Value, ip);
+               }
+            }
+         }
+         return result;
+      }
+
+      static Create mkorder(uint security, double price, uint size, Side side) {
+         return new Create {
+            Price = price,
+            Quantity = size,
+            NewOrder = new OrderDefinition {
+               SecurityKey = security,
+               Side = side,
+               TimeInForce = TimeInForce.Day,
+               ManualOrderIndicator = ManualOrderIndicator.Yes,
+               CtiCode = CTICode.CtiCode1,
+               CustomerOrFirm = CustomerOrFirm.Firm,
+               AccountId = 1,
+               UserId = 1,
+               CustomDef = new CustomOrderDefinition {
+                  Limit = new LimitOrderDefinition()
+               }
+            }
+         };
+      }
+
+      public static int Run(OrderStressOptions o) {
+
+         var ctx = new ServiceContext();
+         ConsoleEx.WriteLine("Please log in to Spark", ConsoleColor.Yellow);
+         ConsoleEx.Write("Username: ", ConsoleColor.Yellow);
+         var name = Console.ReadLine();
+         ConsoleEx.Write("Password: ", ConsoleColor.Yellow);
+         var pw = Console.ReadLine();
+
+         var login_result = ctx.login(name, pw);
+         if (!login_result.Result.success) {
+            ConsoleEx.WriteLine("Invalid Credentials", ConsoleColor.Red);
+         }
+
+         if (Configuration.get<config.DeploymentType>().type.ToUpper() == "PROD") {
+            ConsoleEx.WriteLine("Refusing to run on production system", ConsoleColor.Red);
+            return -1;
+         }
+
+         var orders = ctx.locate<OrderService>();
+         foreach (var m in all_markets(ctx)) {
+            ConsoleEx.WriteLine(m.security.security_desc(), ConsoleColor.White);
+            var tick = m.security.ContractSpec.LowTick;
+            if (m.ask.HasValue) {
+               for (var x = 0; x < o.depth; ++x) {
+                  var price = m.ask.Value.price() + (tick * x);
+                  ConsoleEx.WriteLine($"Sell {o.quantity} at {price}", ConsoleColor.Red);
+                  orders.create_order_async(mkorder(m.security.Id, price, o.quantity, Side.Sell));
+                  System.Threading.Thread.Sleep(o.sleep);
+               }
+            }
+            if (m.bid.HasValue) {
+               // start at bid and place 1 order at 10 levels
+               for (var x=0; x < o.depth; ++x) {
+                  var price = m.bid.Value.price() - (tick * x);
+                  ConsoleEx.WriteLine($"Buy {o.quantity} at {price}", ConsoleColor.Blue);
+                  orders.create_order_async(mkorder(m.security.Id, price, o.quantity, Side.Buy));
+                  System.Threading.Thread.Sleep(o.sleep);
+               }
+            }
+         }
+         return 3;
+      }
+   }
+
+}

src/client/sparkcmd/Cmd/ServiceTestCommand.cs

@@ -17,16 +17,9 @@ namespace bts {
       [Option(nameof(admin))]
       public bool admin { get; set; }
 
-      [Option(nameof(mbo))]
-      public bool mbo { get; set; }
-
-      [Option("key")]
-      public UInt32 key { get; set; }
-
-      [Option("side")]
-      public api.Side side { get; set; }
    }
 
+   [Verb("servicetest", HelpText="Scratch space for testing services")]
    public class ServiceTest {
       public static int Run(ServiceTestOptions o) {
          if (o.synthetic) {
@@ -38,9 +31,6 @@ namespace bts {
          if (o.admin) {
             admin_test();
          }
-         if (o.mbo) {
-            mbo_test(o);
-         }
          return 0;
       }
 
@@ -51,25 +41,6 @@ namespace bts {
          return "";
       }
 
-      private static void mbo_test(ServiceTestOptions o) {
-         var key = o.key;
-         var side = o.side;
-         using (var ctx = new ServiceContext()) {
-            var mbo = ctx.locate<MBOService>();
-            var data = mbo.get_orders_for_security(new MBOSecurityQuery { SecurityId = key });
-
-            Console.WriteLine($"Order data for {data.SecurityId}");
-            var vector = side == api.Side.Buy ? data.Bid : data.Ask;
-            var price = 0.0;
-            foreach(var e in vector) {
-               if (e.Price != price) {
-                  price = e.Price;
-                  Console.WriteLine("================");
-               }
-               Console.WriteLine(e);
-            }
-         }
-      }
 
       private static void synthetic_test() {
          using (var ctx = new ServiceContext()) {

src/client/sparkcmd/Program.cs

@@ -26,14 +26,34 @@ namespace sparkcmd {
          mbts.utils.BTSNativeMethods.provide_command_line(args, args.Length);
          // note: this can be made to infer the options type list from the list of delegates
          return Parser.Default
-         .ParseArguments<ConfigcheckOptions,ScriptClientOptions,TailServiceOptions,ServiceTestOptions,SettingsManagerOptions,TestCommandOptions >(args)
-         .MapResult<ConfigcheckOptions,ScriptClientOptions,TailServiceOptions,ServiceTestOptions,SettingsManagerOptions,TestCommandOptions,int>(
+         .ParseArguments<
+            ConfigcheckOptions,
+            ScriptClientOptions,
+            TailServiceOptions,
+            ServiceTestOptions,
+            SettingsManagerOptions,
+            TestCommandOptions,
+            OrderStressOptions,
+            MBOCommandOptions
+             >(args)
+         .MapResult<
+               ConfigcheckOptions,
+               ScriptClientOptions,
+               TailServiceOptions,
+               ServiceTestOptions,
+               SettingsManagerOptions,
+               TestCommandOptions,
+               OrderStressOptions,
+               MBOCommandOptions,
+               int>(
             Configcheck.Run,
             ScriptClientCommand.Run,
             TailService.Run,
             ServiceTest.Run,
             SettingsManagerCommand.Run,
             TestCommand.Run,
+            OrderStress.Run,
+            MBOCommand.Run,
             errors => 1);
       }
    }

src/data_files/files/Onixs.BTS.lic

@@ -1,21 +1,16 @@
 <license version="1.0">
   <type>production</type>
   <customer>
-    <name>Blue Trading Systems</name>
+    <name>Blue Trading Systems OF 9420-2</name>
   </customer>
   <period>
-    <since>2018-08-30</since>
-    <till>2019-10-16</till>
+    <since>2019-05-02</since>
+    <till>2020-05-13</till>
   </period>
-  <dotNetProducts>
-    <product id="41" name="FIX Analyser" />
-  </dotNetProducts>
-  <dotNetCheckSum>6O68+aAtrLqRaXxY9hLigajhC2ZunVeIev3NbyEXFjHk61WFrNmRTjy6tEduDtFUvGXvS6VUW3VXASaMTLrRgYteZBzfp7+dFTfVIVbzv6cn7c7dnSxWCGO5zPrjXw9cxMXlFz9cshyue6U/DRR5J36ThOhgt2kpY/OP9cgXWJo=</dotNetCheckSum>
   <products>
-    <product id="1" name="OnixS FIX Engine for C++" />
     <product id="4" name="CME Market Data Handler for C++" />
     <product id="7" name="ICE iMpact Multicast Price Feed Handler for C++" />
-    <product id="13" name="NYSE ArcaBook Multicast for Equities Handler for C++" />
+    <product id="67" name="Eurex Market and Reference Data Interfaces (EMDI, RDI) Handler for C++" />
   </products>
-  <checksum>18F98ED1E0A87B4B11DA322403D7BDDD</checksum>
+  <checksum>5E50A1DACD94D7C1BECD316EEAE5AB02</checksum>
 </license>

src/lib/api/public/Update.h

@@ -81,11 +81,6 @@ namespace api {
                for (size_t x = 0; x < raw.added_keys.size(); ++x) {
                   (*state)[raw.added_keys[x]] = raw.added_items[x];
                }
-               std::vector<T> all;
-               all.reserve(state->size());
-               for (const auto& item : *state) {
-                  all.push_back(item.second);
-               }
                //
                // Change
                std::vector<Change<T>> changed;
@@ -94,6 +89,13 @@ namespace api {
                   changed.push_back(Change<T>{(*state)[raw.changed_keys[x]], raw.changed_items[x]});
                   (*state)[raw.changed_keys[x]] = raw.changed_items[x];
                }
+               //
+               // All
+               std::vector<T> all;
+               all.reserve(state->size());
+               for (const auto& item : *state) {
+                  all.push_back(item.second);
+               }
                
                return Update<T, TKey, ServiceType>(all, raw.added_items, changed, removed);
             }

src/lib/md/MBOServiceServer.cpp

@@ -148,14 +148,14 @@ std::optional<uint32_t> search_orders(const ExchangeOrderKey oid, const std::vec
       const auto& oo = orders[x];
       if (oo.id == oid) {
          if (oo.price == price) return { position };
-         else return 0 ;
+         else return 0;
       }
       else if (price == oo.price) {
-         position += 1;
+         position += oo.size;
       }
       else {
          price = oo.price;
-         position = 1;
+         position = oo.size;
       }
    }
    return std::nullopt;

src/lib/md/MBOServiceServer.h

@@ -2,6 +2,7 @@
 #include "lib/api/public/ServiceContext.h"
 #include "lib/api/services/MBOService.server.pb.h"
 #include <memory>
+#include <optional>
 
 namespace bts { 
 namespace utils { class Dispatcher; }
@@ -24,4 +25,5 @@ class MBOServiceServer : public MBOServiceBase {
 
 };
 
+
 }} //namespace 

src/lib/md/test/mbo_tests.cpp

@@ -9,6 +9,10 @@
 
 namespace bts {
 namespace md {
+
+// fwd declare from MBOServiceServer.cpp
+std::optional<uint32_t> search_orders(const ExchangeOrderKey oid, const std::vector<SnapshotOrder>& orders);
+
 namespace test {
 int mbo_tests = 0;
 
@@ -202,6 +206,7 @@ TEST_CASE("MBO sort order", "[md][mbo]") {
    }
 }
 
+
 TEST_CASE("MBO change price and size", "[md][mbo]") {
    // When an order's size goes up it gets a new priority
    Co2TestInstrument instr { "MBO", api::SECURITYTYPE_FUTURE };
@@ -288,7 +293,27 @@ TEST_CASE("MBO change price and size", "[md][mbo]") {
       }
    }
    }
+
+   // Do some tests on the snapshot
+   MBOSnapshot result;
+   copy_to_snapshot(state, &result);
+
+   auto ask30 = search_orders(ExchangeOrderKey(701267928346), result.asks); 
+   REQUIRE(ask30);
+   REQUIRE(*ask30 == 0);
+
+   auto bid2875 = search_orders(ExchangeOrderKey(701267993135), result.bids);
+   REQUIRE(bid2875);
+   REQUIRE(*bid2875 == 20);
    
+   bid2875 = search_orders(ExchangeOrderKey(701267299079), result.bids);
+   REQUIRE(bid2875);
+   REQUIRE(*bid2875 == 6);
+
+   auto not_bid = search_orders(ExchangeOrderKey(701967299879), result.bids);
+   auto not_ask = search_orders(ExchangeOrderKey(701967299879), result.asks);
+   REQUIRE(!not_bid);
+   REQUIRE(!not_ask);
 }
 
 }}}

src/server/sparkd/cmds/RunSim.cpp

@@ -6,7 +6,7 @@ namespace cmd {
 
 int RunSim::execute(std::vector<std::string> const &)
 {
-   sim::start_exchange_sim();
+   sim::start_exchange_sim(nullptr);
    return 0;
 }
 

src/server/sparkd/exchange_sim/ExchangeSim.cpp

@@ -20,6 +20,7 @@
 #include "lib/domain_model/Security.pb.h"
 #include "lib/md/relay/RelayRecorder.h"
 #include "lib/md/IntPrice.h"
+#include "lib/md/MBOProcessor.h"
 
 #include <set>
 
@@ -124,7 +125,7 @@ struct MDPlayer : md::IRelayWriter
 };
 
 
-int start_exchange_sim()
+int start_exchange_sim(md::MBOWriter* mbo)
 {
    try
    {
@@ -137,7 +138,7 @@ int start_exchange_sim()
       std::istringstream str{str_settings};
       FIX::SessionSettings settings( str );
 
-      ex_sim::ExchangeSimApplication application;
+      ex_sim::ExchangeSimApplication application(mbo);
       FIX::FileStoreFactory storeFactory( settings );
       FIX::FileLogFactory logFactory( settings );
       FIX::SocketAcceptor acceptor( application, storeFactory, settings, logFactory );

src/server/sparkd/exchange_sim/ExchangeSim.h

@@ -3,8 +3,9 @@
 #include <memory>
 
 namespace bts {
+namespace md { class MBOWriter; }
 namespace sim {
 
-int start_exchange_sim();
+int start_exchange_sim(md::MBOWriter*);
 
 }}

src/server/sparkd/exchange_sim/ExchangeSimApplication.cpp

@@ -10,12 +10,14 @@
 #include "lib/md/Shm/FixedSecurity.h"
 #include "lib/md/Shm/MDStructure.h"
 #include "lib/md/get_symbols.h"
+#include "lib/md/MBOProcessor.h"
 #include "lib/utils/Dispatcher.h"
 #include "lib/utils/datetime.h"
 #include "server/sparkd/exchange_sim/book/Order.h"
 #include "server/sparkd/exchange_sim/book/OrderMatcher.h"
 #include "server/sparkd/exchange_sim/book/book_transform.h"
 #include "lib/persistence/SqliteStore.h"
+#include "lib/orderman/orderman_utils/fix_order_codec.h"
 
 #include <boost/date_time/posix_time/posix_time.hpp>
 #include <fmt/format.h>
@@ -97,14 +99,20 @@ struct ExchangeSimApplication::MDConsumerRec {
    utils::TimeStamp last_read;
 };
 
+inline FIX::OrderID exchange_key(const std::string& clOrdID) {
+   return FIX::OrderID(clOrdID.substr(3));
+}
+
 struct ExchangeSimApplication::PImpl {
-   PImpl()
+   PImpl(md::MBOWriter* mbo)
        : producer(co2::PLATFORM_SIM),
          md_sn(1),
          timer(utils::Dispatcher::background(), 200,
                [=]() { producer.heartbeat(); }),
          do_merge_with_md(config::get<config::ExchangeSim>().merge_with_md()),
-         store(new persistence::SqliteStore("ex_sim_orders")) {}
+         store(new persistence::SqliteStore("ex_sim_orders")),
+         lock(),
+         mbo(mbo) {}
 
    IDGenerator m_generator;
    md::MDProducer producer;
@@ -116,9 +124,10 @@ struct ExchangeSimApplication::PImpl {
    bool do_merge_with_md;
    std::unique_ptr<persistence::IStore> store;
    std::mutex lock;
+   md::MBOWriter* mbo;
 };
 
-ExchangeSimApplication::ExchangeSimApplication() : me(new PImpl()) {
+ExchangeSimApplication::ExchangeSimApplication(md::MBOWriter* mbo) : me(new PImpl(mbo)) {
    auto previous_book = me->store->retrieve_all();
 
    auto all = me->store->retrieve_all()["books"];
@@ -355,7 +364,7 @@ void ExchangeSimApplication::updateOrder(const Order &order, char status) {
    FIX::TargetSubID(order.sender_sub_id());
 
    FIX42::ExecutionReport fixOrder;
-   fixOrder.set(FIX::OrderID("12345678901234567"));
+   fixOrder.set(exchange_key(order.clordid()));
    fixOrder.set(FIX::ExecID(me->m_generator.genExecutionID()));
    fixOrder.set(FIX::ExecTransType(FIX::ExecTransType_NEW));
    fixOrder.set(FIX::ExecType(status));
@@ -405,7 +414,7 @@ void ExchangeSimApplication::rejectOrder(
    FIX::SenderCompID senderCompID(target.getValue());
 
    FIX42::ExecutionReport fixOrder;
-   fixOrder.set(FIX::OrderID("12345678901234567"));
+   fixOrder.set(exchange_key(clOrdID.getString()));
    fixOrder.set(FIX::ExecID(me->m_generator.genExecutionID()));
    fixOrder.set(FIX::ExecTransType(FIX::ExecTransType_NEW));
    fixOrder.set(FIX::ExecType(FIX::ExecType_REJECTED));
@@ -480,7 +489,7 @@ void ExchangeSimApplication::update_md(std::string const &sec_desc) {
       handle.update_direct(utils::TimeStamp::now(), bids_temp, asks_temp, me->md_sn++);
 
       auto new_trades = matcher.get_new_trades();
-      for (auto &t : new_trades)
+      for (auto &t : new_trades) {
          handle.add_trade(
              utils::TimeStamp::now(),
              md::PriceSize_from_double<ex_sim::precision>(t.price.value() * price_factor,t.quantity.value()),
@@ -488,6 +497,29 @@ void ExchangeSimApplication::update_md(std::string const &sec_desc) {
              t.bid_aggressor ? co2::AGGRESSOR_SIDE_BUY 
                : ( t.ask_aggressor ? co2::AGGRESSOR_SIDE_SELL 
                   : co2::AGGRESSOR_SIDE_NONE) );
+      }
+
+      // publish mbo
+      auto mbo = me->mbo;
+      if (mbo) {
+         // get the list of trades
+         auto ob = matcher.get_order_book(nullptr);
+         auto snap = mbo->allocate_snapshot();
+         for (const auto& order : ob.order()) {
+            auto ek = exchange_key(order.clordid());
+            md::SnapshotOrder sno { 
+               md::ExchangeOrderKey(std::atoll(ek.getString().c_str())), 
+               md::IntPrice_from_double<precision>(order.price() * price_factor),
+               (int)order.open_quantity() };
+            if (order.side() == api::SIDE_BUY) {
+               snap->bids.push_back(sno);
+            }
+            else if (order.side() == api::SIDE_SELL) {
+               snap->asks.push_back(sno);
+            }
+         }
+         mbo->publish_snapshot(SecurityKey(it->second.key()), &snap);
+      }
    }
 }
 

src/server/sparkd/exchange_sim/ExchangeSimApplication.h

@@ -43,7 +43,7 @@ GCC_DIAG_OFF("-Wignored-qualifiers")
 
 namespace bts {
 namespace api {class Security;}
-namespace md {class MDConsumerHandle;}
+namespace md {class MDConsumerHandle; class MBOWriter;}
 namespace ex_sim {
 static const int precision = 9;
 class Order;
@@ -55,7 +55,7 @@ class ExchangeSimApplication
      public FIX::MessageCracker
 {
 public:
-   ExchangeSimApplication();
+   ExchangeSimApplication(md::MBOWriter*);
    ~ExchangeSimApplication();
    void merge_with_md_book();
 

src/server/sparkd/exchange_sim/book/OrderMatcher.cpp

@@ -83,13 +83,17 @@ void OrderMatcher::add_or_merge(Order const &o, std::vector<Order> *out) {
 
    auto range = find_level(*side,o.price());
 
-   for (auto it = range.first; it != range.second; it++) 
-      if (it->owner() == o.owner()) {
-         auto no_const = const_cast<Order *>(&(*it));
-         no_const->set_open_quantity(it->open_quantity() + o.quantity());
+   // If the last order at this level is the same owner we can update it
+   if (range.first != range.second) {
+      auto last = range.second;
+      last--;
+      if (last->owner() == o.owner()) {
+         auto no_const = const_cast<Order *>(&(*last));
+         no_const->set_open_quantity(last->open_quantity() + o.quantity());
          match(out);
          return;
       }
+   }
    add_new(o,out);
 }
 
@@ -232,14 +236,14 @@ Book OrderMatcher::get_book() const {
    return result;
 }
 
-OrderBook OrderMatcher::get_order_book(std::string const&exclude_owner) const
+OrderBook OrderMatcher::get_order_book(const char* exclude_owner) const
 {
    OrderBook result;
    for (auto &b : me->bids)
-      if (b.owner() != exclude_owner)
+      if (exclude_owner == nullptr || b.owner() != exclude_owner)
          *result.add_order() = b;
    for (auto &a : me->asks)
-      if (a.owner() != exclude_owner)
+      if (exclude_owner == nullptr || a.owner() != exclude_owner)
          *result.add_order() = a;
    return result;
 }

src/server/sparkd/exchange_sim/book/OrderMatcher.h

@@ -30,7 +30,7 @@ class OrderMatcher {
                                      Quantity by_quantity,
                                      std::string const& order_owner);
    Book get_book() const;
-   OrderBook get_order_book(std::string const&exclude_owner) const;
+   OrderBook get_order_book(const char*) const;
    
    Order const *get_order(std::string const &clorid);
   

src/server/sparkd/md/run_md.cpp

@@ -152,7 +152,7 @@ void run_market_datapublishers(MBOWriter* mbo) {
    }));
 
   if (config::get<MarketDataConfig>().use_sim()) { 
-   threading::make_thread("ExchangeSim", []() { sim::start_exchange_sim(); })
+   threading::make_thread("ExchangeSim", [mbo]() { sim::start_exchange_sim(mbo); })
        .detach();
   }
   boot_sleep();