SPARK-1548 Enable merging live market data with simulator

release_notes.txt

@@ -1,3 +1,7 @@
+1.1.20
+- Update CME market data handler for price precision extension.
+  https://www.cmegroup.com/confluence/display/EPICSANDBOX/Price+Precision+Extension
+
 1.1.8
 - Enabled wild cards in risk management UI
 - Fixed spreader order updates

src/client/spark_core/Co2/MDConsumerHandle.cs

@@ -124,8 +124,8 @@ namespace mbts.co2
       public double session_high { get { return _cold->session_high;} }
       public double session_low { get {  return _cold->session_low; } }
       public double open_interest { get {  return _cold->open_interest; } }
-      public double session_high_bid { get {  return _cold->session_high_bid; } }
-      public double session_low_offer { get {  return _cold->session_low_offer; } }
+      // public double session_high_bid { get {  return _cold->session_high_bid; } }
+      // public double session_low_offer { get {  return _cold->session_low_offer; } }
       public double fixing_price { get {  return _cold->fixing_price; } }
 
       //

src/lib/CMakeDeps.cmake

@@ -1 +1 @@
-set(sources api/apicore.proto;api/apigateway.proto;api/apioptions.proto;api/core/ApiClientInterop.cpp;api/core/ApiServerInterop.cpp;api/public/ApiServer.cpp;api/public/ClientChannel.cpp;api/public/QuickbusClientChannel.cpp;api/public/ServerChannel.cpp;api/public/ServiceContext.cpp;api/service_helpers/AdminServiceUtils.cpp;api/services/AdminService.proto;api/services/ApiUnitTestService.proto;api/services/ErrorReportService.proto;api/services/HelloService.proto;api/services/MarketDataRelay.proto;api/services/OrderService.proto;api/services/OutOfProcessURLListener.proto;api/services/PersistenceService.proto;api/services/ShellDiagnosticService.proto;api/services/ShellLatencyService.proto;api/services/SyntheticSecurityService.proto;api/services/UserService.proto;api/test/ApiTests.cpp;api/test/HelloServiceServer.cpp;config/ConfigCheck.cpp;config/ConfigDownloader.cpp;config/ConfigErrorCollector.cpp;config/ConfigLog.cpp;config/ConfigPropertyEvaluator.cpp;config/ConfigValueQueue.cpp;config/ConfigurationException.cpp;config/ConfigurationState.cpp;config/PreprocessInputStream.cpp;config/config.cpp;config/defaults/Executor.conf;config/defaults/MarketData.conf;config/defaults/Ports.conf;config/defaults/Symbols.conf;config/defaults/cpp_config.conf;config/defaults/edge.conf;config/defaults/expiration_display_offset.conf;config/defaults/jetfix.conf;config/defaults/logging.conf;config/defaults/testkeys.conf;config/defaults/toggles.conf;config/extensions/extensions.proto;config/invalid/invalid.proto;config/protos/ExchangeSimCfg.proto;config/protos/MarketData.proto;config/protos/Ports.proto;config/protos/Symbols.proto;config/protos/UI.proto;config/protos/cpp_config.proto;config/protos/environment.proto;config/protos/expiration_display_offset.proto;config/protos/jetfix.proto;config/protos/logging.proto;config/protos/quickbus_cfg.proto;config/protos/system.proto;config/protos/testkeys.proto;config/protos/toggles.proto;config/tests/HTTPServer.cpp;config/tests/config_ref_tests.cpp;config/tests/config_tests.cpp;config/text_format/text_format.cpp;domain_model/Security.cpp;domain_model/Security.proto;domain_model/SecurityTests.cpp;expression/BooleanGrammar.cpp;expression/ExpressionEvaluator.cpp;expression/ParserExceptions.cpp;expression/Value.cpp;expression/tests/expression_parser_tests.cpp;logging/LoggerHub.cpp;logging/Sink.cpp;logging/logging.cpp;logging/tests/logging_tests.cpp;md/Diag/MDPriceLantencyTester.cpp;md/Diag/ShmDumper.cpp;md/MDConsumer.cpp;md/MDConsumerHandle.cpp;md/MDConsumerInterop.cpp;md/MDProducer.cpp;md/MDProducerHandle.cpp;md/MDRecoverableConsumer.cpp;md/MemorySegmentUtils.cpp;md/PlatformLoader.cpp;md/PriceSize.cpp;md/SecurityDB.cpp;md/SecurityIDRecorder.cpp;md/SettlementRecorder.cpp;md/Shm/FixedSecurity.cpp;md/SyntheticSecurityService.cpp;md/common.cpp;md/protos/co2.proto;md/relay/QueryServer.cpp;md/relay/RelayClient.cpp;md/relay/RelayRecorder.cpp;md/test/Co2TestInstrument.cpp;md/test/md2_tests.cpp;md/test/synthetic_book_merge_tests.cpp;metrics/metrics.cpp;metrics/system_metrics.cpp;os/os.cpp;os/tests/os_tests.cpp;quickbus/core/LocalNode.cpp;quickbus/core/Quickbus.cpp;quickbus/core/SubscriptionID.cpp;quickbus/core/TableID.cpp;quickbus/message_model/Update.cpp;quickbus/message_model/merge_updates.cpp;quickbus/network/QBConnection.cpp;quickbus/network/QBConnectionManager.cpp;quickbus/network/endpoint_description.cpp;quickbus/network/quickbus_msgs.proto;quickbus/pubsub/TablePublisher.cpp;quickbus/pubsub/TableSubscriber.cpp;quickbus/pubsub/tests/qb_pubsub_tests.cpp;quickbus/routing/INode.cpp;quickbus/routing/NodeID.cpp;quickbus/routing/Router.cpp;quickbus/rpc/RPCClient.cpp;quickbus/rpc/RPCServer.cpp;quickbus/rpc/RPCSessionID.cpp;quickbus/rpc/rpc_utils.cpp;quickbus/rpc/tests/qb_rpc_tests.cpp;quickbus/tests/quickbus_tests.cpp;sparkproto/SparkProtos.proto;testing/catch_test_runner.cpp;utils/Dispatcher.cpp;utils/HTTPStream.cpp;utils/PBUtils.cpp;utils/Puid.cpp;utils/SecureLogon.cpp;utils/SharedMemory.cpp;utils/Sqlite_utils.cpp;utils/Uuid.cpp;utils/datetime.cpp;utils/download.cpp;utils/file_utils.cpp;utils/init_spark_lib.cpp;utils/is_unittest_environment.cpp)
+set(sources api/apicore.proto;api/apigateway.proto;api/apioptions.proto;api/core/ApiClientInterop.cpp;api/core/ApiServerInterop.cpp;api/public/ApiServer.cpp;api/public/ClientChannel.cpp;api/public/QuickbusClientChannel.cpp;api/public/ServerChannel.cpp;api/public/ServiceContext.cpp;api/service_helpers/AdminServiceUtils.cpp;api/services/AdminService.proto;api/services/ApiUnitTestService.proto;api/services/ErrorReportService.proto;api/services/HelloService.proto;api/services/MarketDataRelay.proto;api/services/OrderService.proto;api/services/OutOfProcessURLListener.proto;api/services/PersistenceService.proto;api/services/ShellDiagnosticService.proto;api/services/ShellLatencyService.proto;api/services/SyntheticSecurityService.proto;api/services/UserService.proto;api/test/ApiTests.cpp;api/test/HelloServiceServer.cpp;config/ConfigCheck.cpp;config/ConfigDownloader.cpp;config/ConfigErrorCollector.cpp;config/ConfigLog.cpp;config/ConfigPropertyEvaluator.cpp;config/ConfigValueQueue.cpp;config/ConfigurationException.cpp;config/ConfigurationState.cpp;config/PreprocessInputStream.cpp;config/config.cpp;config/defaults/Executor.conf;config/defaults/MarketData.conf;config/defaults/Ports.conf;config/defaults/Symbols.conf;config/defaults/cpp_config.conf;config/defaults/edge.conf;config/defaults/expiration_display_offset.conf;config/defaults/jetfix.conf;config/defaults/logging.conf;config/defaults/testkeys.conf;config/defaults/toggles.conf;config/extensions/extensions.proto;config/invalid/invalid.proto;config/protos/ExchangeSimCfg.proto;config/protos/MarketData.proto;config/protos/Ports.proto;config/protos/Symbols.proto;config/protos/UI.proto;config/protos/cpp_config.proto;config/protos/environment.proto;config/protos/expiration_display_offset.proto;config/protos/jetfix.proto;config/protos/logging.proto;config/protos/quickbus_cfg.proto;config/protos/system.proto;config/protos/testkeys.proto;config/protos/toggles.proto;config/tests/HTTPServer.cpp;config/tests/config_ref_tests.cpp;config/tests/config_tests.cpp;config/text_format/text_format.cpp;domain_model/Security.cpp;domain_model/Security.proto;domain_model/SecurityTests.cpp;expression/BooleanGrammar.cpp;expression/ExpressionEvaluator.cpp;expression/ParserExceptions.cpp;expression/Value.cpp;expression/tests/expression_parser_tests.cpp;logging/LoggerHub.cpp;logging/Sink.cpp;logging/logging.cpp;logging/tests/logging_tests.cpp;md/Diag/MDPriceLantencyTester.cpp;md/Diag/ShmDumper.cpp;md/MDConsumer.cpp;md/MDConsumerHandle.cpp;md/MDConsumerInterop.cpp;md/MDProducer.cpp;md/MDProducerHandle.cpp;md/MDRecoverableConsumer.cpp;md/MemorySegmentUtils.cpp;md/PlatformLoader.cpp;md/PriceSize.cpp;md/SecurityDB.cpp;md/SecurityIDRecorder.cpp;md/SettlementRecorder.cpp;md/Shm/FixedSecurity.cpp;md/SyntheticSecurityService.cpp;md/common.cpp;md/protos/co2.proto;md/relay/QueryServer.cpp;md/relay/RelayClient.cpp;md/relay/RelayRecorder.cpp;md/test/Co2TestInstrument.cpp;md/test/md2_tests.cpp;md/test/synthetic_book_merge_tests.cpp;os/os.cpp;os/tests/os_tests.cpp;quickbus/core/LocalNode.cpp;quickbus/core/Quickbus.cpp;quickbus/core/SubscriptionID.cpp;quickbus/core/TableID.cpp;quickbus/message_model/Update.cpp;quickbus/message_model/merge_updates.cpp;quickbus/network/QBConnection.cpp;quickbus/network/QBConnectionManager.cpp;quickbus/network/endpoint_description.cpp;quickbus/network/quickbus_msgs.proto;quickbus/pubsub/TablePublisher.cpp;quickbus/pubsub/TableSubscriber.cpp;quickbus/pubsub/tests/qb_pubsub_tests.cpp;quickbus/routing/INode.cpp;quickbus/routing/NodeID.cpp;quickbus/routing/Router.cpp;quickbus/rpc/RPCClient.cpp;quickbus/rpc/RPCServer.cpp;quickbus/rpc/RPCSessionID.cpp;quickbus/rpc/rpc_utils.cpp;quickbus/rpc/tests/qb_rpc_tests.cpp;quickbus/tests/quickbus_tests.cpp;sparkproto/SparkProtos.proto;testing/catch_test_runner.cpp;utils/Dispatcher.cpp;utils/HTTPStream.cpp;utils/PBUtils.cpp;utils/Puid.cpp;utils/SecureLogon.cpp;utils/SharedMemory.cpp;utils/Sqlite_utils.cpp;utils/Uuid.cpp;utils/datetime.cpp;utils/download.cpp;utils/file_utils.cpp;utils/init_spark_lib.cpp;utils/is_unittest_environment.cpp)

src/lib/config/protos/MarketData.proto

@@ -145,4 +145,12 @@ message CMETradingSchedule {
    // we can elaborate later, but for now it's very simple:
    // this is "hours after midnight, in Central Time"
    optional double gen_open_hour = 1 [default = 8.5]; // 8:30 am CT
+}
+
+message SimSource {
+    oneof source {
+        string replay_url = 1 [default = "http://mdreplay.bluetradesys.local/spark/futures.coconut"];
+        string replay_file = 2 [default = "futures.coconut"];
+        bool local_relay = 3;
+    }
 }

src/lib/md/Diag/ShmDumper.cpp

@@ -28,8 +28,10 @@ double dump(TimeStamp time) {
 string dump(PriceSize ps) {
    stringstream res;
    res << "{";
-   res << "'price': " << ps.price() << ", ";
-   res << "'size': " << ps.size();
+   res << "'price': " << std::setprecision(11) << ps.price() << ", ";
+   res << "'size': " << ps.size() << " ";
+   // res << "'mant': " << ps.mantissa() << ", ";
+   // res << "'exp': " << (int)ps.exponent() << " ";
    res << "}";
    return res.str();
 }

src/lib/md/relay/RelayClient.cpp

@@ -283,15 +283,38 @@ struct RelayScheduler {
 
 void relay_everything() {
    api::ServiceContext ctx;
-   auto relay = ctx.locate<MarketDataRelay>();
    std::set<UserSymbol> system_symbols;
    get_system_md_symbols(system_symbols);
    RelayProducer producer(true);
-   RelayClient client(relay, nullptr, system_symbols, producer);
-   RelayScheduler scheduler(ctx, config::get<RelayOptions>().naggle_ms(), client);
-   ctx.post([&]() { scheduler.run(); });
+   relay_selected(ctx, system_symbols, producer);
 }
 
+void relay_selected(api::ServiceContext& ctx, const std::set<UserSymbol>& symbols, IRelayWriter& writer) {
+   auto relay = ctx.locate<MarketDataRelay>();
+   RelayClient client(relay, nullptr, symbols, writer);
+   RelayScheduler scheduler(ctx, config::get<RelayOptions>().naggle_ms(), client);
+   ctx.post([&]() { scheduler.run(); });
+
+   // block forever, I guess
+   utils::Dispatcher::current().run();
+}
+void local_relay(IRelayWriter& writer) {
+   api::ServiceContext ctx;
+   auto relay = ctx.locate<MarketDataRelay>();
+   std::set<UserSymbol> symbols;
+   {
+      std::set<UserSymbol> system_symbols;
+      get_system_md_symbols(system_symbols);
+
+      // Filter out VENUE_SIM
+      for (const auto& s : system_symbols) {
+         if (s.exchange() != api::VENUE_SIM) {
+            symbols.insert(s);
+         }
+      }
+   }
+   relay_selected(ctx, symbols, writer);
+}
 
 void record_everything(std::string path, std::string filename) {
    auto options = config::get<RecordOptions>();

src/lib/md/relay/RelayClient.h

@@ -14,6 +14,9 @@ namespace bts {
 namespace pubsub {
    class PubsubHub;
 }
+namespace api {
+   class ServiceContext;
+}
 namespace md {
 
 struct RelayControl;
@@ -26,6 +29,9 @@ struct IRelayWriter {
 
 utils::Dispatcher& scan_securities_thread();
 
+void relay_selected(api::ServiceContext& ctx, const std::set<UserSymbol>& symbols, IRelayWriter& writer);
+void local_relay(IRelayWriter& writer);
+
 extern "C" {
 BTS_EXPORT void relay_everything();
 

src/lib/md/relay/RelayRecorder.cpp

@@ -157,7 +157,7 @@ void replay_recording(istream& in, IRelayWriter& writer, int32_t naggle_ms) {
    }
 }
 
-void replay_file(const std::string& filename, IRelayWriter& writer, int32_t naggle_ms, bool loop = false) {
+void replay_file(const std::string& filename, IRelayWriter& writer, int32_t naggle_ms, bool loop) {
    do {
       ifstream in;
       in.open(filename, ios::binary | ios::in);

src/lib/md/relay/RelayRecorder.h

@@ -30,6 +30,7 @@ void unprint_recording(const std::string& src, const std::string& dest);
 void trace_recording(const std::string& file, security_key key);
 void replay_recording(const std::string&, int32_t delay_ms);
 void replay_url(const std::string& url, IRelayWriter& writer, int32_t naggle_ms);
+void replay_file(const std::string& filename, IRelayWriter& writer, int32_t naggle_ms, bool loop=false);
 
 }}//namespace
 

src/server/sparkd/CMakeLists.txt

@@ -24,7 +24,7 @@ target_link_libraries( sparkd
   tbb
   double-conversion
   pugixml
-  onixs-cmemdh-all-4.5.2
+  onixs-cmemdh-all-4.8.0
   quickfix
   benchmark
   lttng-ust

src/server/sparkd/exchange_sim/ExchangeSim.cpp

@@ -104,11 +104,27 @@ int start_exchange_sim()
       acceptor.start();
       
       std::map<uint32_t,md::Security> known_securities;
+      const auto& conf = config::get<md::SimSource>(); 
       while ( true ) {
          MDPlayer player(application,known_securities);
-         md::replay_url("http://mdreplay.bluetradesys.local/spark/futures.coconut",
-                        player,
-                        config::get<md::RelayOptions>().naggle_ms());
+         switch(conf.source_case()) {
+            case md::SimSource::SOURCE_NOT_SET:
+            case md::SimSource::kReplayUrl:
+               md::replay_url(conf.replay_url(),
+                  player,
+                  config::get<md::RelayOptions>().naggle_ms());
+            break;
+            case md::SimSource::kReplayFile:
+               md::replay_file(conf.replay_file(),
+                  player,
+                  config::get<md::RelayOptions>().naggle_ms(), true);
+            break;
+            case md::SimSource::kLocalRelay: {
+               md::local_relay(player);
+            }
+            break;
+
+         }
       }
       acceptor.stop();
       return 0;

src/server/sparkd/md/platforms/mdp3_onixs/MDP3Listener.cpp

@@ -11,74 +11,6 @@ using namespace bts::co2;
 namespace bts {
 namespace md {
 
-void decode_ticks(int vtt, double mintick, FixedSecurity& dest) {
-   switch(vtt) {
-      default:
-      case 0:
-         dest.low_tick = mintick;
-         dest.high_tick = mintick;
-         dest.price_break = 0;
-         break;
-      case 1:
-         dest.price_break = 500;
-         dest.low_tick = 5;
-         dest.high_tick = 10;
-         break;
-      case 2:
-         dest.price_break = 5;
-         dest.low_tick = 0.5;
-         dest.high_tick = 1;
-         break;
-      case 3:
-         dest.price_break = 10;
-         dest.low_tick = 1;
-         dest.high_tick = 2;
-         break;
-      case 4:
-         dest.price_break = 500;
-         dest.low_tick = 5;
-         dest.high_tick = 25;
-         break;
-      case 10:
-         dest.price_break = 300;
-         dest.low_tick = 5;
-         dest.high_tick = 25;
-         break;
-      case 11:
-         dest.price_break = 300;
-         dest.low_tick = 5;
-         dest.high_tick = 10;
-         break;
-      case 12:
-         dest.price_break = 5;
-         dest.low_tick = 0.25;
-         dest.high_tick = 0.5;
-         break;
-   }
-}
-
-int floorlog10(int64_t m) {
-   bts_assert(m >= 0);
-   if (m < 10) return 0;
-   if (m < 100) return 1;
-   if (m < 1000) return 2;
-   if (m < 10000) return 3;
-   if (m < 100000) return 4;
-   if (m < 1000000) return 5;
-   if (m < 10000000) return 6;
-   if (m < 100000000) return 7;
-   if (m < 1000000000) return 8;
-   if (m < 10000000000) return 9;
-   if (m < 100000000000) return 10;
-   if (m < 1000000000000) return 11;
-   if (m < 10000000000000) return 12;
-   if (m < 100000000000000) return 13;
-   if (m < 1000000000000000) return 14;
-   if (m < 10000000000000000) return 15;
-   if (m < 100000000000000000) return 16;
-   if (m < 1000000000000000000) return 17;
-   return 19;
-}
 
 MDP3Listener::MDP3Listener(
       CMESecurityIndex& index,
@@ -182,159 +114,6 @@ void MDP3Listener::onEndOfPacket(onixs::Handler&, const onixs::PacketArgs&) {
 //
 // Security discovery
 
-template <class Price>
-double double_of_price(const Price& price, int8_t display_factor) {
-   // order of multiplication is important because of floating point error
-   return (price.mantissa() * pow10(price.exponent())) * pow10(display_factor);
-}
-
-inline datetime::LLDateTime ll_of_timestamp(const onixs::Timestamp& ts) {
-   return datetime::ll_from_parts(ts.year(), (int)ts.month(), ts.day(), ts.hour(), ts.minute(), ts.second(), ts.millisecond());
-}
-
-datetime::Date date_of_timestamp(const onixs::Timestamp& ts) {
-   return datetime::Date(ts.year(), (int)ts.month(), ts.day());
-}
-
-template <class Definition>
-datetime::Date date_of_maturity(const Definition& def) {
-   for (const auto& event : def.events()) {
-      if (event.eventType() == onixs::EventType::LastEligibleTradeDate) {
-         return date_of_timestamp(event.eventTime());
-      }
-   }
-   return datetime::Date();
-}
-
-
-template <class Definition>
-void set_security_common(const Definition& def, FixedSecurity& fsec) {
-   fsec.numeric_id = def.securityId();
-   fsec.symbol(def.asset().items(), def.asset().size());
-   fsec.cfi_code(def.cfiCode().items(), def.cfiCode().size());
-   fsec.alt_symbol(def.securityGroup().items(), def.securityGroup().size());
-   fsec.security_desc(def.symbol().items(), def.symbol().size());
-   fsec.exchange = api::Venue::VENUE_GLBX;
-   // tickRule not present in futures
-   fsec.cme_market_segment_id = def.marketSegmentId();
-   fsec.display_factor = def.displayFactor().exponent() + floorlog10(def.displayFactor().mantissa());
-
-   def.mainFraction(fsec.cme_main_fraction);
-   def.subFraction(fsec.cme_sub_fraction);
-   def.priceDisplayFormat(fsec.cme_price_display_format);
-}
-
-void MDP3Listener::onDefinition(onixs::Handler&, const onixs::Security& osec, const onixs::InstrumentDefinitionFutureArgs& def) {
-   tracepoint (bts_trace,trace_order,(char*) "onDefinition.future", _latest_sequence, (uint16_t) _latest_sequence, 0);
-   const auto& sec = def.message();
-   auto exchange_id = sec.securityId();
-   // LOG(md, Info, "onDefinition future {} {}",  
-   //    std::string(sec.asset().items(), sec.asset().size()),
-   //    std::string(sec.symbol().items(), sec.symbol().size()));
-
-   if (filter_security(exchange_id, sec) != ADD) return;
-   bts_assert(exchange_id == osec.id());
-
-   FixedFutureSecurity fsec;
-
-   set_security_common(sec, fsec);
-   fsec.price_break = 0;
-   // low/high tick are stored non-premultiplied and multiplied in conversion to api::Secutity
-   double min_tick = double_of_price(sec.minPriceIncrement(), 0);
-   fsec.low_tick = fsec.high_tick = min_tick;
-   onixs::PRICE min_tick_amount;
-   sec.minPriceIncrementAmount(min_tick_amount);
-   double min_tick_dbl = double_of_price(min_tick_amount, 0);
-   fsec.cash_per_point = min_tick_dbl / min_tick / pow10(fsec.display_factor);
-   auto expiration = date_of_maturity(sec);
-   fsec.expiration_year = expiration.year;
-   fsec.expiration_month = expiration.month;
-   fsec.expiration_day = expiration.day;
-   add_security(fsec);
-}
-void MDP3Listener::onDefinition(onixs::Handler&, const onixs::Security&, const onixs::InstrumentDefinitionOptionArgs& def) {
-   tracepoint (bts_trace,trace_order,(char*) "onDefinition.option", _latest_sequence, (uint16_t) _latest_sequence, 0);
-   const auto& sec = def.message();
-   auto exchange_id = sec.securityId();
-   // LOG(Info, MarketData) << "onDefinition option " << exchange_id;
-   if (filter_security(exchange_id, sec) != ADD) return;
-
-   FixedOptionSecurity fsec;
-   set_security_common(sec, fsec);
-   auto expiration = date_of_maturity(sec);
-   fsec.expiration_year = expiration.year;
-   fsec.expiration_month = expiration.month;
-   fsec.expiration_day = expiration.day;
-   
-   onixs::PRICE strike_price;
-   bts_assert(sec.strikePrice(strike_price));
-   fsec.strike = double_of_price(strike_price, fsec.display_factor);
-
-   onixs::PRICE min_tick, min_tick_amount;
-   sec.minPriceIncrement(min_tick);
-   sec.minPriceIncrementAmount(min_tick_amount);
-
-   int8_t tick_rule;
-   if (sec.tickRule(tick_rule)) {
-      decode_ticks(tick_rule, double_of_price(min_tick, 0), fsec);
-   }
-   fsec.cash_per_point = double_of_price(min_tick_amount, 0) / fsec.low_tick / pow10(fsec.display_factor);
-
-   fsec.call_put = sec.putOrCall() == onixs::PutOrCall::Call ? api::PUTORCALL_CALL : api::PUTORCALL_PUT;
-
-   add_security(fsec);
-}
-
-void MDP3Listener::onDefinition(onixs::Handler&, const onixs::Security&, const onixs::InstrumentDefinitionSpreadArgs& spread_args) {
-   tracepoint (bts_trace,trace_order,(char*) "onDefinition.spread", _latest_sequence, (uint16_t) _latest_sequence, 0);
-   const auto& sec = spread_args.message();
-   auto exchange_id = sec.securityId();
-   // LOG(md, Info, "onDefinition spread {} {}",  
-   //    std::string(sec.asset().items(), sec.asset().size()),
-   //    std::string(sec.symbol().items(), sec.symbol().size()));
-   if (filter_security(exchange_id, sec) != ADD) return;
-   
-   // Filter out any already expired spreads
-   auto expiration = date_of_maturity(sec);
-   if (expiration < datetime::today()) return;
-
-   FixedSpreadSecurityL fsec;
-   set_security_common(sec, fsec);
-   vector<uint32_t> leg_exchange_ids;
-   leg_exchange_ids.reserve(sec.legs().size());
-   
-   onixs::PRICE min_tick = sec.minPriceIncrement();
-   
-   int8_t tick_rule;
-   if (sec.tickRule(tick_rule)) {
-      decode_ticks(tick_rule, double_of_price(min_tick, 0), fsec);
-   }
-   else {
-      double mintick = double_of_price(min_tick, 0);
-      fsec.low_tick = mintick;
-      fsec.high_tick = mintick;
-      fsec.price_break = 0;
-   }
-
-   fsec.set_leg_count(sec.legs().size());
-   for(size_t x = 0; x < fsec.leg_count(); ++x) {
-      const auto& leg = sec.legs()[x];
-      onixs::PRICE price;
-      if (leg.legPrice(price)) {
-         fsec.legs[x].leg_price = double_of_price(price, fsec.display_factor);
-      }
-      onixs::DecimalQty qty;
-      if (leg.legOptionDelta(qty)) {
-         fsec.legs[x].leg_delta = double_of_price(qty, 0);
-      }
-      fsec.legs[x].leg_quantity = leg.legRatioQty();
-      fsec.legs[x].leg_side = leg.legSide() == onixs::LegSide::BuySide ? api::SIDE_BUY : api::SIDE_SELL;
-      leg_exchange_ids.push_back(leg.legSecurityId());
-   }
-   add_security(fsec);
-   _index.add_spread_legs(fsec.numeric_id, leg_exchange_ids);
-}
-
 void MDP3Listener::onBookUpdate(onixs::Handler&, const onixs::Security& sec, const onixs::DirectBook& book) {
    if (_booktype == CMEOptions::BOOKTYPE_DIRECT) {
       onBookUpdate(sec, book);
@@ -362,52 +141,16 @@ AggressorSide map_aggressor(onixs::AggressorSide::Enum agside) {
    }
 }
 
-void on_trade(SecurityData* data, const onixs::PRICE& price, int size, AggressorSide agside, const onixs::Timestamp& onixs_date) {
+void on_trade(SecurityData* data, const onixs::Decimal& price, int size, onixs::AggressorSide::Enum agside, const onixs::Timestamp& onixs_date) {
    auto ts = utils::TimeStamp::now();
    PriceSize ps = pricesize_of_price_size(price, data->handle.security().display_factor, size);
    auto date = ll_of_timestamp(onixs_date);
-   data->handle.add_trade(ts, ps, date, agside, data->electronic_volume + size);
-}
-
-void MDP3Listener::onTrade(onixs::Handler&, const onixs::Security& sec, const onixs::RecoveryEntryArgs& arg) {
-   tracepoint (bts_trace,trace_order,(char*) "onTrade.recovery", _latest_sequence, (uint16_t) _latest_sequence, 0);
-   auto* data = _index.get_security(sec.id());
-   if (data != NULL) {
-      int size = 0;
-      arg.entry().entrySize(size);
-      onixs::PRICE price;
-      arg.entry().entryPx(price);
-      on_trade(data, price, size, AGGRESSOR_SIDE_NONE, arg.message().transactTime());
-   }
-}
-void MDP3Listener::onTrade(onixs::Handler&, const onixs::Security& sec, const onixs::TradeSummaryArgs& arg) {
-   tracepoint (bts_trace,trace_order,(char*) "onTrade.summary", _latest_sequence, (uint16_t) _latest_sequence, 0);
-   auto* data = _index.get_security(sec.id());
-   if (data != NULL) {
-      auto size = arg.entry().entrySize();
-      onixs::PRICE price = arg.entry().entryPx();
-      onixs::AggressorSide::Enum agside = onixs::AggressorSide::NoAggressor;
-      arg.entry().aggressorSide(agside);
-      on_trade(data, price, size, map_aggressor(agside), arg.message().transactTime());
-   }
-}
-void MDP3Listener::onTrade(onixs::Handler&, const onixs::Security& sec, const onixs::TradeArgs& arg) {
-   tracepoint (bts_trace,trace_order,(char*) "onTrade.trade_args", _latest_sequence, (uint16_t) _latest_sequence, 0);
-   if (arg.entry().updateAction() == onixs::UpdateAction::New) {
-      auto* data = _index.get_security(sec.id());
-      if (data != nullptr) {
-         int size = arg.entry().entrySize();
-         onixs::PRICE price = arg.entry().entryPx();
-         onixs::AggressorSide::Enum agside = onixs::AggressorSide::NoAggressor;
-         arg.entry().aggressorSide(agside);
-         on_trade(data, price, size, map_aggressor(agside), arg.message().transactTime());
-      }
-   }
+   data->handle.add_trade(ts, ps, date, map_aggressor(agside), data->electronic_volume + size);
 }
 
 //
 // RFQ
-void MDP3Listener::onQuoteRequest(onixs::Handler&, const onixs::Security& sec, const onixs::QuoteRequestRelatedSymArgs& arg) {
+void MDP3Listener::onQuoteRequest(onixs::Handler&, const onixs::Security& sec, const onixs::QuoteRequest39RelatedSymArgs& arg) {
    tracepoint (bts_trace,trace_order,(char*) "onQuoteRequest", _latest_sequence, (uint16_t) _latest_sequence, 0);
    auto* data = _index.get_security(sec.id());
    if (data != nullptr) {
@@ -456,7 +199,7 @@ void MDP3Listener::onQuoteRequest(onixs::Handler&, const onixs::Security& sec, c
 
 //
 // Volume
-void MDP3Listener::onElectronicVolume(onixs::Handler&, const onixs::Security& sec, const onixs::VolumeArgs& args) {
+void MDP3Listener::onElectronicVolume(onixs::Handler&, const onixs::Security& sec, const onixs::Volume37Args& args) {
    tracepoint (bts_trace,trace_order,(char*) "onElectronicVolume.volume", _latest_sequence, (uint16_t) _latest_sequence, 0);
    auto* data = _index.get_security(sec.id());
    if (data != nullptr) {
@@ -466,162 +209,6 @@ void MDP3Listener::onElectronicVolume(onixs::Handler&, const onixs::Security& se
       data->electronic_volume = args.entry().entrySize();
    }
 }
-void MDP3Listener::onElectronicVolume(onixs::Handler&, const onixs::Security& sec, const onixs::RecoveryEntryArgs& args) {
-   tracepoint (bts_trace,trace_order,(char*) "onElectronicVolume.recovery", _latest_sequence, (uint16_t) _latest_sequence, 0);
-   auto* data = _index.get_security(sec.id());
-   if (data) {
-      int32_t size;
-      if (args.entry().entrySize(size)) {
-         data->electronic_volume = size;
-      }
-   }
-}
-
-//
-// Misc
-void MDP3Listener::onSettlementPrice(onixs::Handler&, const onixs::Security& sec, const onixs::DailyStatisticsArgs& args) {
-   tracepoint (bts_trace,trace_order,(char*) "onSettlementPrice.daily", _latest_sequence, (uint16_t) _latest_sequence, 0);
-   auto* data = _index.get_security(sec.id());
-   if (data != nullptr) {
-      onixs::PRICE price;
-      onixs::Timestamp time;
-      args.entry().entryPx(price);
-      args.entry().tradingReferenceDate(time);
-      auto display_factor = data->handle.security().display_factor;
-      data->handle.update_settlement(
-         utils::TimeStamp::now(), 
-         double_of_price(price, display_factor),
-         date_of_timestamp(time));
-   }
-}
-void MDP3Listener::onSettlementPrice(onixs::Handler&, const onixs::Security& sec, const onixs::RecoveryEntryArgs& args) {
-   tracepoint (bts_trace,trace_order,(char*) "onSettlementPrice.recovery", _latest_sequence, (uint16_t) _latest_sequence, 0);
-   auto* data = _index.get_security(sec.id());
-   if (data != nullptr) {
-      onixs::PRICE price;
-      onixs::Timestamp time;
-      args.entry().entryPx(price);
-      args.entry().tradingReferenceDate(time);
-      auto display_factor = data->handle.security().display_factor;
-      data->handle.update_settlement(
-         utils::TimeStamp::now(), 
-         double_of_price(price, display_factor),
-         date_of_timestamp(time));
-   }
-}
-void MDP3Listener::onOpenInterest(onixs::Handler&, const onixs::Security& sec, const onixs::DailyStatisticsArgs& args) {
-   tracepoint (bts_trace,trace_order,(char*) "onOpenInterest.daily", _latest_sequence, (uint16_t) _latest_sequence, 0);
-   auto* data = _index.get_security(sec.id());
-   if (data != nullptr) {
-      onixs::PRICE price;
-      args.entry().entryPx(price);
-      auto display_factor = data->handle.security().display_factor;
-      data->handle.update_open_interest(utils::TimeStamp::now(), double_of_price(price, display_factor));
-   }
-}
-void MDP3Listener::onOpenInterest(onixs::Handler&, const onixs::Security& sec, const onixs::RecoveryEntryArgs& args) {
-   tracepoint (bts_trace,trace_order,(char*) "onOpenInterest.recovery", _latest_sequence, (uint16_t) _latest_sequence, 0);
-   auto* data = _index.get_security(sec.id());
-   if (data != nullptr) {
-      onixs::PRICE price;
-      args.entry().entryPx(price);
-      auto display_factor = data->handle.security().display_factor;
-      data->handle.update_open_interest(utils::TimeStamp::now(), double_of_price(price, display_factor));
-   }
-}
-void MDP3Listener::onOpeningPrice(onixs::Handler&, const onixs::Security& sec, const onixs::SessionStatisticsArgs& args) {
-   tracepoint (bts_trace,trace_order,(char*) "onOpeningPrice.session", _latest_sequence, (uint16_t) _latest_sequence, 0);
-   auto* data = _index.get_security(sec.id());
-   if (data != nullptr) {
-      onixs::PRICE price = args.entry().entryPx();
-      onixs::Timestamp time = args.message().transactTime();
-      auto display_factor = data->handle.security().display_factor;
-      data->handle.update_opening(
-         utils::TimeStamp::now(), 
-         double_of_price(price, display_factor),
-         date_of_timestamp(time));
-   }
-}
-void MDP3Listener::onOpeningPrice(onixs::Handler&, const onixs::Security& sec, const onixs::RecoveryEntryArgs& args) {
-   tracepoint (bts_trace,trace_order,(char*) "onOpeningPrice.recovery", _latest_sequence, (uint16_t) _latest_sequence, 0);
-   auto* data = _index.get_security(sec.id());
-   if (data != nullptr) {
-      onixs::PRICE price; 
-      args.entry().entryPx(price);
-      onixs::Timestamp time;
-      args.entry().tradingReferenceDate(time);
-      auto display_factor = data->handle.security().display_factor;
-      data->handle.update_opening(
-         utils::TimeStamp::now(), 
-         double_of_price(price, display_factor),
-         date_of_timestamp(time));
-   }
-}
-void MDP3Listener::onHighTradePrice(onixs::Handler&, const onixs::Security& sec, const onixs::SessionStatisticsArgs& args) {
-   tracepoint (bts_trace,trace_order,(char*) "onHighTradePrice.session", _latest_sequence, (uint16_t) _latest_sequence, 0);
-   auto* data = _index.get_security(sec.id());
-   if (data != nullptr) {
-      auto display_factor = data->handle.security().display_factor;
-      data->handle.update_session_high(
-         utils::TimeStamp::now(),
-         double_of_price(args.entry().entryPx(), display_factor));
-   }
-}
-void MDP3Listener::onHighTradePrice(onixs::Handler&, const onixs::Security& sec, const onixs::RecoveryEntryArgs& args) {
-   tracepoint (bts_trace,trace_order,(char*) "onHighTradePrice.recovery", _latest_sequence, (uint16_t) _latest_sequence, 0);
-   auto* data = _index.get_security(sec.id());
-   if (data != nullptr) {
-      auto display_factor = data->handle.security().display_factor;
-      onixs::PRICE price; 
-      args.entry().entryPx(price);
-      data->handle.update_session_high(
-         utils::TimeStamp::now(),
-         double_of_price(price, display_factor));
-   }
-}
-void MDP3Listener::onLowTradePrice(onixs::Handler&, const onixs::Security& sec, const onixs::SessionStatisticsArgs& args) {
-   tracepoint (bts_trace,trace_order,(char*) "onLowTradePrice.session", _latest_sequence, (uint16_t) _latest_sequence, 0);
-   auto* data = _index.get_security(sec.id());
-   if (data != nullptr) {
-      auto display_factor = data->handle.security().display_factor;
-      data->handle.update_session_low(
-         utils::TimeStamp::now(),
-         double_of_price(args.entry().entryPx(), display_factor));
-   }
-}
-void MDP3Listener::onLowTradePrice(onixs::Handler&, const onixs::Security& sec, const onixs::RecoveryEntryArgs& args) {
-   tracepoint (bts_trace,trace_order,(char*) "onLowTradePrice.recovery", _latest_sequence, (uint16_t) _latest_sequence, 0);
-   auto* data = _index.get_security(sec.id());
-   if (data != nullptr) {
-      auto display_factor = data->handle.security().display_factor;
-      onixs::PRICE price; 
-      args.entry().entryPx(price);
-      data->handle.update_session_low(
-         utils::TimeStamp::now(),
-         double_of_price(price, display_factor));
-   }
-}
-void MDP3Listener::onHighestBid(onixs::Handler&, const onixs::Security& sec, const onixs::SessionStatisticsArgs& args) {
-   tracepoint (bts_trace,trace_order,(char*) "onHighestBid.session", _latest_sequence, (uint16_t) _latest_sequence, 0);
-   auto* data = _index.get_security(sec.id());
-   if (data != nullptr) {
-      auto display_factor = data->handle.security().display_factor;
-      data->handle.update_session_high_bid(
-         utils::TimeStamp::now(),
-         double_of_price(args.entry().entryPx(), display_factor));
-   }
-}
-void MDP3Listener::onLowestOffer(onixs::Handler&, const onixs::Security& sec, const onixs::SessionStatisticsArgs& args) {
-   tracepoint (bts_trace,trace_order,(char*) "onLowestOffer.session", _latest_sequence, (uint16_t) _latest_sequence, 0);
-   auto* data = _index.get_security(sec.id());
-   if (data != nullptr) {
-      auto display_factor = data->handle.security().display_factor;
-      data->handle.update_session_low_offer(
-         utils::TimeStamp::now(),
-         double_of_price(args.entry().entryPx(), display_factor));
-   }
-}
-
 
 //
 // Non-onixs calls

src/server/sparkd/md/platforms/mdp3_onixs/MDP3Listener.h

@@ -5,7 +5,9 @@
 #include "CMESecurityIndex.h"
 #include "server/sparkd/tracing/bts_trace.h"
 
+#include "MDP3ListenerHelpers.h"
 #include "onixs.h"
+
 #include <tbb/concurrent_queue.h>
 #include <vector>
 #include <atomic>
@@ -14,11 +16,6 @@
 namespace bts {
 namespace md {
 
-inline PriceSize pricesize_of_price_size(const onixs::PRICE price, int8_t display_factor, Size size) {
-   PriceSize result(price.mantissa(), price.exponent() + display_factor, size);
-   return result;
-}
-
 class MDP3Listener :
    public onixs::HandlerListener,
    public onixs::MarketDataListener,
@@ -54,9 +51,21 @@ public:
 
    //
    // Security discovery
-   void onDefinition(onixs::Handler&, const onixs::Security&, const onixs::InstrumentDefinitionFutureArgs&) override;
-   void onDefinition(onixs::Handler&, const onixs::Security&, const onixs::InstrumentDefinitionOptionArgs&) override;
-   void onDefinition(onixs::Handler&, const onixs::Security&, const onixs::InstrumentDefinitionSpreadArgs&) override;
+   template <typename InstrumentDefinitionFuture>
+   void onFuture(onixs::Handler&, const onixs::Security&, const InstrumentDefinitionFuture&);
+   template <typename InstrumentDefinition>
+   void onOption(onixs::Handler&, const onixs::Security&, const InstrumentDefinition&);
+   template <typename InstrumentDefinition>
+   void onSpread(onixs::Handler&, const onixs::Security&, const InstrumentDefinition&);
+
+   void onDefinition(onixs::Handler& h, const onixs::Security& s, const onixs::InstrumentDefinitionFuture27Args& def) override { onFuture(h, s, def); }
+   void onDefinition(onixs::Handler& h, const onixs::Security& s, const onixs::InstrumentDefinitionFuture54Args& def) override { onFuture(h, s, def); }
+
+   void onDefinition(onixs::Handler& h, const onixs::Security& s, const onixs::InstrumentDefinitionOption41Args& def) override { onOption(h, s, def); }
+   void onDefinition(onixs::Handler& h, const onixs::Security& s, const onixs::InstrumentDefinitionOption55Args& def) override { onOption(h, s, def); }
+
+   void onDefinition(onixs::Handler& h, const onixs::Security& s, const onixs::InstrumentDefinitionSpread29Args& def) override { onSpread(h, s, def); }
+   void onDefinition(onixs::Handler& h, const onixs::Security& s, const onixs::InstrumentDefinitionSpread56Args& def) override { onSpread(h, s, def); }
 
    //
    // Book
@@ -66,79 +75,74 @@ public:
    void onBookUpdate(onixs::Handler&, const onixs::Security&, const onixs::ConsolidatedBook&) override;
 
    template <class TBook>
-   void onBookUpdate(const onixs::Security& sec, const TBook& book) {
-      tracepoint (bts_trace,trace_order,(char*) "onBookUpdate", _latest_sequence, (uint16_t) _latest_sequence, 0);
-
-      auto data = _index.get_security(sec.id());
-      if (data == nullptr) return;
-      bts_assert(data->id == (uint32_t)sec.id());
-      auto display_factor = data->handle.security().display_factor;
-
-      // Put the data into the cache. Flush at the end of the packet.
-      auto key = data->handle.key();
-      auto find_iter = std::find_if(_book_updates.begin(), _book_updates.end(), [key](const BookData& entry) {
-         return key == entry.handle.key();
-      });
-      bool found = find_iter != _book_updates.end();
-      if (!found) {
-         _book_updates.emplace_back(data->handle);
-      }
-      BookData& cache = found ? *find_iter : _book_updates.back();
-
-      // Overwrite all entries every time
-      size_t bids_size = book.bids().size();
-      size_t asks_size = book.offers().size();
-      if (!data->handle.has_depth()) {
-         bids_size = std::min(bids_size, size_t { 1 });
-         asks_size = std::min(asks_size, size_t { 1 });
-      }
-      for (size_t x = 0; x < bids_size; ++x) {
-         cache.bids[x] = pricesize_of_price_size(book.bids()[x].price(), display_factor, book.bids()[x].quantity());
-      }
-      for (size_t x = 0; x < asks_size; ++x) {
-         cache.asks[x] = pricesize_of_price_size(book.offers()[x].price(), display_factor, book.offers()[x].quantity());
-      }
-      // NaN out the remaining depth
-      for (size_t x = bids_size; x < COCONUT_MAX_DEPTH; ++x) {
-         cache.bids[x] = PriceSize::NO_MARKET;
-      }
-      for (size_t x = asks_size; x < COCONUT_MAX_DEPTH; ++x) {
-         cache.asks[x] = PriceSize::NO_MARKET;
-      }
-   }
+   void onBookUpdate(const onixs::Security& sec, const TBook& book);
 
    //
    // Trades
-   void onTrade(onixs::Handler&, const onixs::Security&, const onixs::RecoveryEntryArgs&) override;
-   void onTrade(onixs::Handler&, const onixs::Security&, const onixs::TradeArgs&) override;
-   void onTrade(onixs::Handler&, const onixs::Security&, const onixs::TradeSummaryArgs&) override;
+   template <typename TradeRecovery>
+   void onTradeRecovery(const onixs::Security&, const TradeRecovery&);
+   void onTrade(onixs::Handler&, const onixs::Security& s, const onixs::Recovery52EntryArgs& arg) override { onTradeRecovery(s, arg); }
+   void onTrade(onixs::Handler&, const onixs::Security& s, const onixs::Recovery38EntryArgs& arg) override { onTradeRecovery(s, arg); }
+
+   template <typename Trade>
+   void onTradeSummary(const onixs::Security&, const Trade&);
+   void onTrade(onixs::Handler&, const onixs::Security& s, const onixs::TradeSummary48Args& arg) override { onTradeSummary(s, arg); }
+   void onTrade(onixs::Handler&, const onixs::Security& s, const onixs::TradeSummary42Args& arg) override { onTradeSummary(s, arg); }
 
    //
    // RFQ
-   void onQuoteRequest(onixs::Handler&, const onixs::Security&, const onixs::QuoteRequestRelatedSymArgs&) override;
+   void onQuoteRequest(onixs::Handler&, const onixs::Security&, const onixs::QuoteRequest39RelatedSymArgs&) override;
 
    //
    // Volume
-   void onElectronicVolume(onixs::Handler&, const onixs::Security&, const onixs::VolumeArgs&) override;
-   void onElectronicVolume(onixs::Handler&, const onixs::Security&, const onixs::RecoveryEntryArgs&) override;
+   void onElectronicVolume(onixs::Handler&, const onixs::Security&, const onixs::Volume37Args&) override;
+   template <typename RecoveryEntry>
+   void onElectronicVolume(const onixs::Security&, const RecoveryEntry&);
+   void onElectronicVolume(onixs::Handler&, const onixs::Security& s, const onixs::Recovery52EntryArgs& arg) override { onElectronicVolume(s, arg); }
+   void onElectronicVolume(onixs::Handler&, const onixs::Security& s, const onixs::Recovery38EntryArgs& arg) override { onElectronicVolume(s, arg); }
 
    //
    // Misc
-   void onSettlementPrice(onixs::Handler&, const onixs::Security&, const onixs::DailyStatisticsArgs&) override;
-   void onSettlementPrice(onixs::Handler&, const onixs::Security&, const onixs::RecoveryEntryArgs&) override;
-   void onOpenInterest(onixs::Handler&, const onixs::Security&, const onixs::DailyStatisticsArgs&) override;
-   void onOpenInterest(onixs::Handler&, const onixs::Security&, const onixs::RecoveryEntryArgs&) override;
-   void onOpeningPrice(onixs::Handler&, const onixs::Security&, const onixs::SessionStatisticsArgs&) override;
-   void onOpeningPrice(onixs::Handler&, const onixs::Security&, const onixs::RecoveryEntryArgs&) override;
-   void onHighTradePrice(onixs::Handler&, const onixs::Security&, const onixs::SessionStatisticsArgs&) override;
-   void onHighTradePrice(onixs::Handler&, const onixs::Security&, const onixs::RecoveryEntryArgs&) override;
-   void onLowTradePrice(onixs::Handler&, const onixs::Security&, const onixs::SessionStatisticsArgs&) override;
-   void onLowTradePrice(onixs::Handler&, const onixs::Security&, const onixs::RecoveryEntryArgs&) override;
+   template <typename StatisticsArgs>
+   void onSettlementPrice(const onixs::Security& sec, const StatisticsArgs& args);
+   void onSettlementPrice(onixs::Handler&, const onixs::Security& s, const onixs::DailyStatistics49Args& arg) override { onSettlementPrice(s, arg); }
+   void onSettlementPrice(onixs::Handler&, const onixs::Security& s, const onixs::DailyStatistics33Args& arg) override { onSettlementPrice(s, arg); }
+   void onSettlementPrice(onixs::Handler&, const onixs::Security& s, const onixs::Recovery52EntryArgs& arg) override { onSettlementPrice(s, arg); }
+   void onSettlementPrice(onixs::Handler&, const onixs::Security& s, const onixs::Recovery38EntryArgs& arg) override { onSettlementPrice(s, arg); }
 
-   void onHighestBid(onixs::Handler&, const onixs::Security&, const onixs::SessionStatisticsArgs&) override;
-   void onLowestOffer(onixs::Handler&, const onixs::Security&, const onixs::SessionStatisticsArgs&) override;
-   
+   template <typename InterestArgs>
+   void onOpenInterest(const onixs::Security& sec, const InterestArgs& args);
+   void onOpenInterest(onixs::Handler&, const onixs::Security& s, const onixs::DailyStatistics49Args& arg) override { onOpenInterest(s, arg); }
+   void onOpenInterest(onixs::Handler&, const onixs::Security& s, const onixs::DailyStatistics33Args& arg) override { onOpenInterest(s, arg); }
+   void onOpenInterest(onixs::Handler&, const onixs::Security& s, const onixs::Recovery52EntryArgs& arg) override { onOpenInterest(s, arg); }
+   void onOpenInterest(onixs::Handler&, const onixs::Security& s, const onixs::Recovery38EntryArgs& arg) override { onOpenInterest(s, arg); }
 
+   template <typename SessionArgs>
+   void onSessionOpeningPrice(const onixs::Security&, const SessionArgs&);
+   template <typename RecoveryArgs>
+   void onRecoveryOpeningPrice(const onixs::Security& sec, const RecoveryArgs& args);
+   void onOpeningPrice(onixs::Handler&, const onixs::Security& s, const onixs::SessionStatistics51Args& arg) override { onSessionOpeningPrice(s, arg); }
+   void onOpeningPrice(onixs::Handler&, const onixs::Security& s, const onixs::SessionStatistics35Args& arg) override { onSessionOpeningPrice(s, arg); }
+   void onOpeningPrice(onixs::Handler&, const onixs::Security& s, const onixs::Recovery52EntryArgs& arg) override { onRecoveryOpeningPrice(s, arg); }
+   void onOpeningPrice(onixs::Handler&, const onixs::Security& s, const onixs::Recovery38EntryArgs& arg) override { onRecoveryOpeningPrice(s, arg); }
+
+   template <typename SessionArgs>
+   void onSessionHighTradePrice(const onixs::Security&, const SessionArgs&);
+   template <typename RecoveryArgs>
+   void onRecoveryHighTradePrice(const onixs::Security&, const RecoveryArgs&);
+   void onHighTradePrice(onixs::Handler&, const onixs::Security& s, const onixs::SessionStatistics51Args& arg) override { onSessionHighTradePrice(s, arg); }
+   void onHighTradePrice(onixs::Handler&, const onixs::Security& s, const onixs::SessionStatistics35Args& arg) override { onSessionHighTradePrice(s, arg); }
+   void onHighTradePrice(onixs::Handler&, const onixs::Security& s, const onixs::Recovery52EntryArgs& arg) override { onRecoveryHighTradePrice(s, arg); }
+   void onHighTradePrice(onixs::Handler&, const onixs::Security& s, const onixs::Recovery38EntryArgs& arg) override { onRecoveryHighTradePrice(s, arg); }
+
+   template <typename SessionArgs>
+   void onSessionLowTradePrice(const onixs::Security&, const SessionArgs&);
+   template <typename RecoveryArgs>
+   void onRecoveryLowTradePrice(const onixs::Security&, const RecoveryArgs&);
+   void onLowTradePrice(onixs::Handler&, const onixs::Security& s, const onixs::SessionStatistics51Args& arg) override { onSessionLowTradePrice(s, arg); }
+   void onLowTradePrice(onixs::Handler&, const onixs::Security& s, const onixs::SessionStatistics35Args& arg) override { onSessionLowTradePrice(s, arg); }
+   void onLowTradePrice(onixs::Handler&, const onixs::Security& s, const onixs::Recovery52EntryArgs& arg) override { onRecoveryLowTradePrice(s, arg); }
+   void onLowTradePrice(onixs::Handler&, const onixs::Security& s, const onixs::Recovery38EntryArgs& arg) override { onRecoveryLowTradePrice(s, arg); }
 
    //
    // Discover securities
@@ -190,4 +194,345 @@ private:
 
 };
 
+//////////////////////////////////////////
+// Template Member Function Definitions
+//////////////////////////////////////////
+
+template <class Definition>
+void set_security_common(const Definition& def, FixedSecurity& fsec) {
+   fsec.numeric_id = def.securityId();
+   fsec.symbol(def.asset().items(), def.asset().size());
+   fsec.cfi_code(def.cfiCode().items(), def.cfiCode().size());
+   fsec.alt_symbol(def.securityGroup().items(), def.securityGroup().size());
+   fsec.security_desc(def.symbol().items(), def.symbol().size());
+   fsec.exchange = api::Venue::VENUE_GLBX;
+   // tickRule not present in futures
+   fsec.cme_market_segment_id = def.marketSegmentId();
+   fsec.display_factor = def.displayFactor().exponent() + floorlog10(def.displayFactor().mantissa());
+
+   def.mainFraction(fsec.cme_main_fraction);
+   def.subFraction(fsec.cme_sub_fraction);
+   def.priceDisplayFormat(fsec.cme_price_display_format);
+}
+
+
+template <typename InstrumentDefinition>
+void MDP3Listener::onFuture(onixs::Handler&, const onixs::Security& osec, const InstrumentDefinition& def) {
+   tracepoint (bts_trace,trace_order,(char*) "onDefinition.future", _latest_sequence, (uint16_t) _latest_sequence, 0);
+   const auto& sec = def.message();
+   auto exchange_id = sec.securityId();
+   // LOG(md, Info, "onDefinition future {} {}",  
+   //    std::string(sec.asset().items(), sec.asset().size()),
+   //    std::string(sec.symbol().items(), sec.symbol().size()));
+
+   if (filter_security(exchange_id, sec) != ADD) return;
+   bts_assert(exchange_id == osec.id());
+
+   FixedFutureSecurity fsec;
+
+   set_security_common(sec, fsec);
+   fsec.price_break = 0;
+   // low/high tick are stored non-premultiplied and multiplied in conversion to api::Secutity
+   double min_tick = double_of_price(sec.minPriceIncrement(), 0);
+   fsec.low_tick = fsec.high_tick = min_tick;
+   onixs::Decimal min_tick_amount;
+   sec.minPriceIncrementAmount(min_tick_amount);
+   double min_tick_dbl = double_of_price(min_tick_amount, 0);
+   fsec.cash_per_point = min_tick_dbl / min_tick / pow10(fsec.display_factor);
+   auto expiration = date_of_maturity(sec);
+   fsec.expiration_year = expiration.year;
+   fsec.expiration_month = expiration.month;
+   fsec.expiration_day = expiration.day;
+   add_security(fsec);
+}
+
+template <typename InstrumentDefinition>
+void MDP3Listener::onOption(onixs::Handler&, const onixs::Security&, const InstrumentDefinition& def) {
+   tracepoint (bts_trace,trace_order,(char*) "onDefinition.option", _latest_sequence, (uint16_t) _latest_sequence, 0);
+   const auto& sec = def.message();
+   auto exchange_id = sec.securityId();
+   // LOG(Info, MarketData) << "onDefinition option " << exchange_id;
+   if (filter_security(exchange_id, sec) != ADD) return;
+
+   FixedOptionSecurity fsec;
+   set_security_common(sec, fsec);
+   auto expiration = date_of_maturity(sec);
+   fsec.expiration_year = expiration.year;
+   fsec.expiration_month = expiration.month;
+   fsec.expiration_day = expiration.day;
+   
+   onixs::Decimal strike_price;
+   bts_assert(sec.strikePrice(strike_price));
+   fsec.strike = double_of_price(strike_price, fsec.display_factor);
+
+   onixs::Decimal min_tick, min_tick_amount;
+   sec.minPriceIncrement(min_tick);
+   sec.minPriceIncrementAmount(min_tick_amount);
+
+   int8_t tick_rule;
+   if (sec.tickRule(tick_rule)) {
+      decode_ticks(tick_rule, double_of_price(min_tick, 0), fsec);
+   }
+   fsec.cash_per_point = double_of_price(min_tick_amount, 0) / fsec.low_tick / pow10(fsec.display_factor);
+
+   fsec.call_put = sec.putOrCall() == onixs::PutOrCall::Call ? api::PUTORCALL_CALL : api::PUTORCALL_PUT;
+
+   add_security(fsec);
+}
+
+// min_tick specializations
+inline onixs::Decimal tick_rule(const onixs::InstrumentDefinitionSpread29& sec) {
+   return sec.minPriceIncrement();
+}
+inline onixs::Decimal tick_rule(const onixs::InstrumentDefinitionSpread56& sec) {
+   onixs::Decimal min_tick;
+   sec.minPriceIncrement(min_tick);
+   return min_tick;
+}
+
+template <typename InstrumentDefinition>
+void MDP3Listener::onSpread(onixs::Handler&, const onixs::Security&, const InstrumentDefinition& spread_args) {
+   tracepoint (bts_trace,trace_order,(char*) "onDefinition.spread", _latest_sequence, (uint16_t) _latest_sequence, 0);
+   const auto& sec = spread_args.message();
+   auto exchange_id = sec.securityId();
+   // LOG(md, Info, "onDefinition spread {} {}",  
+   //    std::string(sec.asset().items(), sec.asset().size()),
+   //    std::string(sec.symbol().items(), sec.symbol().size()));
+   if (filter_security(exchange_id, sec) != ADD) return;
+   
+   // Filter out any already expired spreads
+   auto expiration = date_of_maturity(sec);
+   if (expiration < datetime::today()) return;
+
+   FixedSpreadSecurityL fsec;
+   set_security_common(sec, fsec);
+   std::vector<uint32_t> leg_exchange_ids;
+   leg_exchange_ids.reserve(sec.legs().size());
+
+   onixs::Decimal min_tick = tick_rule(sec);
+   
+   int8_t tick_rule;
+   if (sec.tickRule(tick_rule)) {
+      decode_ticks(tick_rule, double_of_price(min_tick, 0), fsec);
+   }
+   else {
+      double mintick = double_of_price(min_tick, 0);
+      fsec.low_tick = mintick;
+      fsec.high_tick = mintick;
+      fsec.price_break = 0;
+   }
+
+   fsec.set_leg_count(sec.legs().size());
+   for(size_t x = 0; x < fsec.leg_count(); ++x) {
+      const auto& leg = sec.legs()[x];
+      onixs::Decimal price;
+      if (leg.legPrice(price)) {
+         fsec.legs[x].leg_price = double_of_price(price, fsec.display_factor);
+      }
+      onixs::Decimal qty;
+      if (leg.legOptionDelta(qty)) {
+         fsec.legs[x].leg_delta = double_of_price(qty, 0);
+      }
+      fsec.legs[x].leg_quantity = leg.legRatioQty();
+      fsec.legs[x].leg_side = leg.legSide() == onixs::LegSide::BuySide ? api::SIDE_BUY : api::SIDE_SELL;
+      leg_exchange_ids.push_back(leg.legSecurityId());
+   }
+   add_security(fsec);
+   _index.add_spread_legs(fsec.numeric_id, leg_exchange_ids);
+}
+
+
+template <class TBook>
+void MDP3Listener::onBookUpdate(const onixs::Security& sec, const TBook& book) {
+   tracepoint (bts_trace,trace_order,(char*) "onBookUpdate", _latest_sequence, (uint16_t) _latest_sequence, 0);
+
+   auto data = _index.get_security(sec.id());
+   if (data == nullptr) return;
+   bts_assert(data->id == (uint32_t)sec.id());
+   auto display_factor = data->handle.security().display_factor;
+
+   // Put the data into the cache. Flush at the end of the packet.
+   auto key = data->handle.key();
+   auto find_iter = std::find_if(_book_updates.begin(), _book_updates.end(), [key](const BookData& entry) {
+      return key == entry.handle.key();
+   });
+   bool found = find_iter != _book_updates.end();
+   if (!found) {
+      _book_updates.emplace_back(data->handle);
+   }
+   BookData& cache = found ? *find_iter : _book_updates.back();
+
+   // Overwrite all entries every time
+   size_t bids_size = book.bids().size();
+   size_t asks_size = book.offers().size();
+   if (!data->handle.has_depth()) {
+      bids_size = std::min(bids_size, size_t { 1 });
+      asks_size = std::min(asks_size, size_t { 1 });
+   }
+   for (size_t x = 0; x < bids_size; ++x) {
+      cache.bids[x] = pricesize_of_price_size(book.bids()[x].price(), display_factor, book.bids()[x].quantity());
+   }
+   for (size_t x = 0; x < asks_size; ++x) {
+      cache.asks[x] = pricesize_of_price_size(book.offers()[x].price(), display_factor, book.offers()[x].quantity());
+   }
+   // NaN out the remaining depth
+   for (size_t x = bids_size; x < COCONUT_MAX_DEPTH; ++x) {
+      cache.bids[x] = PriceSize::NO_MARKET;
+   }
+   for (size_t x = asks_size; x < COCONUT_MAX_DEPTH; ++x) {
+      cache.asks[x] = PriceSize::NO_MARKET;
+   }
+}
+
+void on_trade(SecurityData* data, const onixs::Decimal& price, int size, onixs::AggressorSide::Enum agside, const onixs::Timestamp& onixs_date);
+
+template <typename TradeRecovery>
+void MDP3Listener::onTradeRecovery(const onixs::Security& sec, const TradeRecovery& arg) {
+   tracepoint (bts_trace,trace_order,(char*) "onTrade.recovery", _latest_sequence, (uint16_t) _latest_sequence, 0);
+   auto* data = _index.get_security(sec.id());
+   if (data != NULL) {
+      int size = 0;
+      arg.entry().entrySize(size);
+      onixs::Decimal price;
+      arg.entry().entryPx(price);
+      on_trade(data, price, size, onixs::AggressorSide::NoAggressor, arg.message().transactTime());
+   }
+}
+
+template <typename TradeSummary>
+void MDP3Listener::onTradeSummary(const onixs::Security& sec, const TradeSummary& arg) {
+   tracepoint (bts_trace,trace_order,(char*) "onTrade.summary", _latest_sequence, (uint16_t) _latest_sequence, 0);
+   auto* data = _index.get_security(sec.id());
+   if (data != NULL) {
+      auto size = arg.entry().entrySize();
+      onixs::Decimal price = arg.entry().entryPx();
+      onixs::AggressorSide::Enum agside = onixs::AggressorSide::NoAggressor;
+      arg.entry().aggressorSide(agside);
+      on_trade(data, price, size, agside, arg.message().transactTime());
+   }
+}
+
+
+template <typename RecoveryEntryArgs>
+void MDP3Listener::onElectronicVolume(const onixs::Security& sec, const RecoveryEntryArgs& args) {
+   tracepoint (bts_trace,trace_order,(char*) "onElectronicVolume.recovery", _latest_sequence, (uint16_t) _latest_sequence, 0);
+   auto* data = _index.get_security(sec.id());
+   if (data) {
+      int32_t size;
+      if (args.entry().entrySize(size)) {
+         data->electronic_volume = size;
+      }
+   }
+}
+
+template <typename StatisticsArgs>
+void MDP3Listener::onSettlementPrice(const onixs::Security& sec, const StatisticsArgs& args) {
+   tracepoint (bts_trace,trace_order,(char*) "onSettlementPrice.daily", _latest_sequence, (uint16_t) _latest_sequence, 0);
+   auto* data = _index.get_security(sec.id());
+   if (data != nullptr) {
+      onixs::Decimal price;
+      onixs::Timestamp time;
+      args.entry().entryPx(price);
+      args.entry().tradingReferenceDate(time);
+      auto display_factor = data->handle.security().display_factor;
+      data->handle.update_settlement(
+         utils::TimeStamp::now(), 
+         double_of_price(price, display_factor),
+         date_of_timestamp(time));
+   }
+}
+
+template <typename InterestArgs>
+void MDP3Listener::onOpenInterest(const onixs::Security& sec, const InterestArgs& args) {
+   tracepoint (bts_trace,trace_order,(char*) "onOpenInterest.daily", _latest_sequence, (uint16_t) _latest_sequence, 0);
+   auto* data = _index.get_security(sec.id());
+   if (data != nullptr) {
+      onixs::Decimal price;
+      args.entry().entryPx(price);
+      auto display_factor = data->handle.security().display_factor;
+      data->handle.update_open_interest(utils::TimeStamp::now(), double_of_price(price, display_factor));
+   }
+}
+
+template <typename SessionArgs>
+void MDP3Listener::onSessionOpeningPrice(const onixs::Security& sec, const SessionArgs& args) {
+   tracepoint (bts_trace,trace_order,(char*) "onOpeningPrice.session", _latest_sequence, (uint16_t) _latest_sequence, 0);
+   auto* data = _index.get_security(sec.id());
+   if (data != nullptr) {
+      onixs::Decimal price = args.entry().entryPx();
+      onixs::Timestamp time = args.message().transactTime();
+      auto display_factor = data->handle.security().display_factor;
+      data->handle.update_opening(
+         utils::TimeStamp::now(), 
+         double_of_price(price, display_factor),
+         date_of_timestamp(time));
+   }
+}
+template <typename RecoveryArgs>
+void MDP3Listener::onRecoveryOpeningPrice(const onixs::Security& sec, const RecoveryArgs& args) {
+   tracepoint (bts_trace,trace_order,(char*) "onOpeningPrice.recovery", _latest_sequence, (uint16_t) _latest_sequence, 0);
+   auto* data = _index.get_security(sec.id());
+   if (data != nullptr) {
+      onixs::Decimal price; 
+      args.entry().entryPx(price);
+      onixs::Timestamp time;
+      args.entry().tradingReferenceDate(time);
+      auto display_factor = data->handle.security().display_factor;
+      data->handle.update_opening(
+         utils::TimeStamp::now(), 
+         double_of_price(price, display_factor),
+         date_of_timestamp(time));
+   }
+}
+
+template <typename SessionArgs>
+void MDP3Listener::onSessionHighTradePrice(const onixs::Security& sec, const SessionArgs& args) {
+   tracepoint (bts_trace,trace_order,(char*) "onHighTradePrice.session", _latest_sequence, (uint16_t) _latest_sequence, 0);
+   auto* data = _index.get_security(sec.id());
+   if (data != nullptr) {
+      auto display_factor = data->handle.security().display_factor;
+      data->handle.update_session_high(
+         utils::TimeStamp::now(),
+         double_of_price(args.entry().entryPx(), display_factor));
+   }
+}
+template <typename RecoveryArgs>
+void MDP3Listener::onRecoveryHighTradePrice(const onixs::Security& sec, const RecoveryArgs& args) {
+   tracepoint (bts_trace,trace_order,(char*) "onHighTradePrice.recovery", _latest_sequence, (uint16_t) _latest_sequence, 0);
+   auto* data = _index.get_security(sec.id());
+   if (data != nullptr) {
+      auto display_factor = data->handle.security().display_factor;
+      onixs::Decimal price; 
+      args.entry().entryPx(price);
+      data->handle.update_session_high(
+         utils::TimeStamp::now(),
+         double_of_price(price, display_factor));
+   }
+}
+template <typename SessionArgs>
+void MDP3Listener::onSessionLowTradePrice(const onixs::Security& sec, const SessionArgs& args) {
+   tracepoint (bts_trace,trace_order,(char*) "onLowTradePrice.session", _latest_sequence, (uint16_t) _latest_sequence, 0);
+   auto* data = _index.get_security(sec.id());
+   if (data != nullptr) {
+      auto display_factor = data->handle.security().display_factor;
+      data->handle.update_session_low(
+         utils::TimeStamp::now(),
+         double_of_price(args.entry().entryPx(), display_factor));
+   }
+}
+template <typename RecoveryArgs>
+void MDP3Listener::onRecoveryLowTradePrice(const onixs::Security& sec, const RecoveryArgs& args) {
+   tracepoint (bts_trace,trace_order,(char*) "onLowTradePrice.recovery", _latest_sequence, (uint16_t) _latest_sequence, 0);
+   auto* data = _index.get_security(sec.id());
+   if (data != nullptr) {
+      auto display_factor = data->handle.security().display_factor;
+      onixs::Decimal price; 
+      args.entry().entryPx(price);
+      data->handle.update_session_low(
+         utils::TimeStamp::now(),
+         double_of_price(price, display_factor));
+   }
+}
+
+
 }}//namespace

src/server/sparkd/md/platforms/mdp3_onixs/MDP3ListenerHelpers.h

@@ -0,0 +1,112 @@
+#include "lib/utils/bts_assert.h"
+#include "lib/md/coconut2.h"
+
+#include "onixs.h"
+
+#include <vector>
+#include <atomic>
+#include <array>  
+
+namespace bts {
+namespace md {
+
+inline PriceSize pricesize_of_price_size(const onixs::Decimal price, int8_t display_factor, Size size) {
+   PriceSize result(price.mantissa(), price.exponent() + display_factor, size);
+   return result;
+}
+
+template <class Price>
+double double_of_price(const Price& price, int8_t display_factor) {
+   // order of multiplication is important because of floating point error
+   return (price.mantissa() * pow10(price.exponent())) * pow10(display_factor);
+}
+
+inline void decode_ticks(int vtt, double mintick, FixedSecurity& dest) {
+   switch(vtt) {
+      default:
+      case 0:
+         dest.low_tick = mintick;
+         dest.high_tick = mintick;
+         dest.price_break = 0;
+         break;
+      case 1:
+         dest.price_break = 500;
+         dest.low_tick = 5;
+         dest.high_tick = 10;
+         break;
+      case 2:
+         dest.price_break = 5;
+         dest.low_tick = 0.5;
+         dest.high_tick = 1;
+         break;
+      case 3:
+         dest.price_break = 10;
+         dest.low_tick = 1;
+         dest.high_tick = 2;
+         break;
+      case 4:
+         dest.price_break = 500;
+         dest.low_tick = 5;
+         dest.high_tick = 25;
+         break;
+      case 10:
+         dest.price_break = 300;
+         dest.low_tick = 5;
+         dest.high_tick = 25;
+         break;
+      case 11:
+         dest.price_break = 300;
+         dest.low_tick = 5;
+         dest.high_tick = 10;
+         break;
+      case 12:
+         dest.price_break = 5;
+         dest.low_tick = 0.25;
+         dest.high_tick = 0.5;
+         break;
+   }
+}
+
+inline int floorlog10(int64_t m) {
+   bts_assert(m >= 0);
+   if (m < 10) return 0;
+   if (m < 100) return 1;
+   if (m < 1000) return 2;
+   if (m < 10000) return 3;
+   if (m < 100000) return 4;
+   if (m < 1000000) return 5;
+   if (m < 10000000) return 6;
+   if (m < 100000000) return 7;
+   if (m < 1000000000) return 8;
+   if (m < 10000000000) return 9;
+   if (m < 100000000000) return 10;
+   if (m < 1000000000000) return 11;
+   if (m < 10000000000000) return 12;
+   if (m < 100000000000000) return 13;
+   if (m < 1000000000000000) return 14;
+   if (m < 10000000000000000) return 15;
+   if (m < 100000000000000000) return 16;
+   if (m < 1000000000000000000) return 17;
+   return 19;
+}
+
+inline datetime::LLDateTime ll_of_timestamp(const onixs::Timestamp& ts) {
+   return datetime::ll_from_parts(ts.year(), (int)ts.month(), ts.day(), ts.hour(), ts.minute(), ts.second(), ts.millisecond());
+}
+
+inline datetime::Date date_of_timestamp(const onixs::Timestamp& ts) {
+   return datetime::Date(ts.year(), (int)ts.month(), ts.day());
+}
+
+template <class Definition>
+datetime::Date date_of_maturity(const Definition& def) {
+   for (const auto& event : def.events()) {
+      if (event.eventType() == onixs::EventType::LastEligibleTradeDate) {
+         return date_of_timestamp(event.eventTime());
+      }
+   }
+   return datetime::Date();
+}
+
+
+}}//namespace

src/server/sparkd/md/platforms/mdp3_onixs/MDP3Main.cpp

@@ -299,6 +299,12 @@ unique_ptr<onixs::Handler> MDP3Main::prepare_handler(const string& channel_id) {
       loggerSettings.filename(log_path() + "/log_" + channel_id);
       loggerSettings.severityLevel((onixs::LogSeverity::Enum)options.onixs_log_mode());
       logger = new onixs::FileLogger(loggerSettings);
+      if (options.onixs_log_mode() == CMEOptions::Debug) {
+         // trace all the things
+         settings.debugLogging().traceBooks(true);
+         settings.debugLogging().traceFeeds(true);
+         settings.debugLogging().traceMessages(onixs::MessageTracing::FixFormat);
+      }
    }