Merged in ice_excel (pull request #388 )

fix RTD crash for ice Approved-by: Jake Norbut <jake.norbut@bluetradesys.com>
fix RTD crash for ice
2020-01-10 22:39:27 +00:00 · 2020-01-10 14:43:06 -05:00 · 2020-01-10 08:35:33 -05:00 · 2020-01-10 08:30:17 -05:00 · 2020-01-09 10:40:47 -05:00 · 2020-01-02 13:57:54 -05:00
15 changed files with 471 additions and 18 deletions
@@ -71,12 +71,12 @@ The first step is to start the spark_sdk Docker container with the examples dire

 Windows Command Prompt:
 {% clipable() %}
-docker run --rm -it -v %CD%:/examples -u bts -w /examples --shm-size 1024m --name spark_sdk -h spark_sdk --cap-add=SYS_PTRACE --security-opt seccomp=unconfined -p 9000:9000 -p 9001:9001 spark_sdk bash
+docker run --rm -it -v %CD%:/examples -u bts -w /examples --shm-size 2048m --name spark_sdk -h spark_sdk --cap-add=SYS_PTRACE --security-opt seccomp=unconfined -p 9000:9000 -p 9001:9001 spark_sdk bash
 {% end %}

 **Powershell** / Linux / Mac:
 {% clipable() %}
-docker run --rm -it -v ${PWD}:/examples -u bts -w /examples --shm-size 1024m --name spark_sdk -h spark_sdk --cap-add=SYS_PTRACE --security-opt seccomp=unconfined -p 9000:9000 -p 9001:9001 spark_sdk bash
+docker run --rm -it -v ${PWD}:/examples -u bts -w /examples --shm-size 2048m --name spark_sdk -h spark_sdk --cap-add=SYS_PTRACE --security-opt seccomp=unconfined -p 9000:9000 -p 9001:9001 spark_sdk bash
 {% end %}

 ### Attach
@@ -0,0 +1,2 @@
+dotnet pack
+dotnet nuget push
@@ -66,9 +66,23 @@ namespace bts.excel.rtd
         if (string.IsNullOrEmpty(contract))
            return Observable.Return(new []{"Can't Parse Contract: " + contract ?? string.Empty});

-         Func<SecurityImmutable, bool> where = (x => x.Exchange == ven && x.Symbol == symbol && x.security_type() == sectype &&
-                                                x.Glbx.SecurityDesc == contract);
+         Func<SecurityImmutable, bool> desc = null;
+         switch(ven)
+         {
+            case Venue.Sim:
+            case Venue.Glbx:
+               desc = (x=> x.Glbx.SecurityDesc == contract);
+            break;
+            case Venue.Ice:
+               desc = (x=> x.Ice.SecurityDesc == contract);
+               break;
+            default:
+               desc = (x => true);
+            break;
+         }

+         Func<SecurityImmutable, bool> where = (x => x.Exchange == ven && x.Symbol == symbol && x.security_type() == sectype && desc(x) );
+                                                
         var stream = market_data(ven).subscribe_combined(where);
         return stream;
      }      
@@ -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/service_helpers/AlgoServiceUtils.cpp;api/service_helpers/PositionUtilities.cpp;api/services/AdminService.proto;api/services/AlgoDefinitionService.proto;api/services/AlgoService.proto;api/services/ApiUnitTestService.proto;api/services/AuthService.proto;api/services/BasePositionService.proto;api/services/ErrorReportService.proto;api/services/HelloService.proto;api/services/MBOService.proto;api/services/MarginService.proto;api/services/MarketDataAuxiliaryService.proto;api/services/MarketDataRelay.proto;api/services/MoneyFileService.proto;api/services/OrderService.proto;api/services/OutOfProcessURLListener.proto;api/services/PersistenceService.proto;api/services/PositionService.proto;api/services/ShellDiagnosticService.proto;api/services/ShellLatencyService.proto;api/services/SyntheticSecurityService.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/MarketData.conf;config/defaults/Ports.conf;config/defaults/Symbols.conf;config/defaults/cme_reference_data.conf;config/defaults/edge.conf;config/defaults/jetfix.conf;config/defaults/keycode.conf;config/defaults/logging.conf;config/defaults/orderman_cfg.conf;config/defaults/orders_fills_cleanup.conf;config/defaults/position_cfg.conf;config/defaults/span.conf;config/defaults/spreader.conf;config/defaults/testkeys.conf;config/defaults/threads.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/cme_reference_data.proto;config/protos/jetfix.proto;config/protos/keycode.proto;config/protos/logging.proto;config/protos/mercury_cfg.proto;config/protos/orderman_cfg.proto;config/protos/orders_fills_cleanup.proto;config/protos/perflog_cfg.proto;config/protos/position_cfg.proto;config/protos/quickbus_cfg.proto;config/protos/span.proto;config/protos/spreader.proto;config/protos/system.proto;config/protos/testkeys.proto;config/protos/threads.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;domain_model/sod.cpp;domain_model/tests/domain_tests.cpp;expression/BooleanGrammar.cpp;expression/ExpressionEvaluator.cpp;expression/ParserExceptions.cpp;expression/Value.cpp;expression/tests/expression_parser_tests.cpp;jetfix/codec/FIXArray.cpp;jetfix/common/OutFIX.cpp;jetfix/connection/AsyncReader.cpp;jetfix/connection/IConnection.cpp;jetfix/connection/SocketConnection.cpp;jetfix/gateway/CoreGateway.cpp;jetfix/gateway/JetFIXGateway.cpp;jetfix/gateway/Session.cpp;jetfix/store/FIXFileStore.cpp;jetfix/store/session_name.cpp;jetfix/tests/async_reader_tests.cpp;jetfix/tests/decoding_tests.cpp;jetfix/tests/fix_encoding_stream_tests.cpp;jetfix/tests/fix_fragment_tests.cpp;jetfix/tests/jetfix_performance_tests.cpp;jetfix/tests/sample_fix.cpp;jetfix/tests/store_tests.cpp;logging/LoggerHub.cpp;logging/Sink.cpp;logging/logging.cpp;logging/tests/logging_tests.cpp;md/Diag/MDPeriodicDumper.cpp;md/Diag/ShmDumper.cpp;md/MBOProcessor.cpp;md/MBOServiceServer.cpp;md/MDClient.cpp;md/MDConsumer.cpp;md/MDConsumerHandle.cpp;md/MDConsumerInterop.cpp;md/MDProducer.cpp;md/MDProducerHandle.cpp;md/MemorySegmentUtils.cpp;md/PlatformLoader.cpp;md/PriceSize.cpp;md/SecurityCache.cpp;md/SecurityDB.cpp;md/SecurityIDRecorder.cpp;md/SecurityMapper.cpp;md/SecurityOracle.cpp;md/SettlementRecorder.cpp;md/Shm/FixedSecurity.cpp;md/SyntheticSecurityService.cpp;md/cme/MDP3Utils.cpp;md/common.cpp;md/ice/ICEUtils.cpp;md/protos/co2.proto;md/relay/QueryServer.cpp;md/relay/RelayClient.cpp;md/relay/RelayRecorder.cpp;md/test/Co2ManagedTestHelper.cpp;md/test/Co2TestInstrument.cpp;md/test/mbo_tests.cpp;md/test/md2_tests.cpp;md/test/md_test_utils.cpp;md/test/synthetic_book_merge_tests.cpp;md/vendor/OnixsUtils.cpp;mercury/actors/IPollableActor.cpp;mercury/broker/Broker.cpp;mercury/broker/BrokerProxy.cpp;mercury/broker/ILocalBroker.cpp;mercury/broker/IRemoteBroker.cpp;mercury/common/mercury_common.cpp;mercury/events/TimerEvent.cpp;mercury/executor/Executor.cpp;mercury/topics/Topic.cpp;metrics/metrics.cpp;metrics/system_metrics.cpp;orderman/actors/AsyncFactory.cpp;orderman/actors/ExecutionGateway.cpp;orderman/actors/FillStore.cpp;orLine truncated
+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/service_helpers/AlgoServiceUtils.cpp;api/service_helpers/PositionUtilities.cpp;api/services/AdminService.proto;api/services/AlgoDefinitionService.proto;api/services/AlgoService.proto;api/services/ApiUnitTestService.proto;api/services/AuthService.proto;api/services/BasePositionService.proto;api/services/ErrorReportService.proto;api/services/HelloService.proto;api/services/MBOService.proto;api/services/MarginService.proto;api/services/MarketDataAuxiliaryService.proto;api/services/MarketDataRelay.proto;api/services/MoneyFileService.proto;api/services/OrderService.proto;api/services/OutOfProcessURLListener.proto;api/services/PersistenceService.proto;api/services/PositionService.proto;api/services/ShellDiagnosticService.proto;api/services/ShellLatencyService.proto;api/services/SyntheticSecurityService.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/MarketData.conf;config/defaults/Ports.conf;config/defaults/Symbols.conf;config/defaults/cme_reference_data.conf;config/defaults/edge.conf;config/defaults/jetfix.conf;config/defaults/keycode.conf;config/defaults/logging.conf;config/defaults/orderman_cfg.conf;config/defaults/orders_fills_cleanup.conf;config/defaults/position_cfg.conf;config/defaults/span.conf;config/defaults/spreader.conf;config/defaults/testkeys.conf;config/defaults/threads.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/cme_reference_data.proto;config/protos/jetfix.proto;config/protos/keycode.proto;config/protos/logging.proto;config/protos/mercury_cfg.proto;config/protos/orderman_cfg.proto;config/protos/orders_fills_cleanup.proto;config/protos/perflog_cfg.proto;config/protos/position_cfg.proto;config/protos/quickbus_cfg.proto;config/protos/span.proto;config/protos/spreader.proto;config/protos/subprocess.proto;config/protos/system.proto;config/protos/testkeys.proto;config/protos/threads.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;domain_model/sod.cpp;domain_model/tests/domain_tests.cpp;expression/BooleanGrammar.cpp;expression/ExpressionEvaluator.cpp;expression/ParserExceptions.cpp;expression/Value.cpp;expression/tests/expression_parser_tests.cpp;jetfix/codec/FIXArray.cpp;jetfix/common/OutFIX.cpp;jetfix/connection/AsyncReader.cpp;jetfix/connection/IConnection.cpp;jetfix/connection/SocketConnection.cpp;jetfix/gateway/CoreGateway.cpp;jetfix/gateway/JetFIXGateway.cpp;jetfix/gateway/Session.cpp;jetfix/store/FIXFileStore.cpp;jetfix/store/session_name.cpp;jetfix/tests/async_reader_tests.cpp;jetfix/tests/decoding_tests.cpp;jetfix/tests/fix_encoding_stream_tests.cpp;jetfix/tests/fix_fragment_tests.cpp;jetfix/tests/jetfix_performance_tests.cpp;jetfix/tests/sample_fix.cpp;jetfix/tests/store_tests.cpp;logging/LoggerHub.cpp;logging/Sink.cpp;logging/logging.cpp;logging/tests/logging_tests.cpp;md/Diag/MDPeriodicDumper.cpp;md/Diag/ShmDumper.cpp;md/MBOProcessor.cpp;md/MBOServiceServer.cpp;md/MDClient.cpp;md/MDConsumer.cpp;md/MDConsumerHandle.cpp;md/MDConsumerInterop.cpp;md/MDProducer.cpp;md/MDProducerHandle.cpp;md/MemorySegmentUtils.cpp;md/PlatformLoader.cpp;md/PriceSize.cpp;md/SecurityCache.cpp;md/SecurityDB.cpp;md/SecurityIDRecorder.cpp;md/SecurityMapper.cpp;md/SecurityOracle.cpp;md/SettlementRecorder.cpp;md/Shm/FixedSecurity.cpp;md/SyntheticSecurityService.cpp;md/cme/MDP3Utils.cpp;md/common.cpp;md/ice/ICEUtils.cpp;md/protos/co2.proto;md/relay/QueryServer.cpp;md/relay/RelayClient.cpp;md/relay/RelayRecorder.cpp;md/test/Co2ManagedTestHelper.cpp;md/test/Co2TestInstrument.cpp;md/test/mbo_tests.cpp;md/test/md2_tests.cpp;md/test/md_test_utils.cpp;md/test/synthetic_book_merge_tests.cpp;md/vendor/OnixsUtils.cpp;mercury/actors/IPollableActor.cpp;mercury/broker/Broker.cpp;mercury/broker/BrokerProxy.cpp;mercury/broker/ILocalBroker.cpp;mercury/broker/IRemoteBroker.cpp;mercury/common/mercury_common.cpp;mercury/events/TimerEvent.cpp;mercury/executor/Executor.cpp;mercury/topics/Topic.cpp;metrics/metrics.cpp;metrics/system_metrics.cpp;orderman/actors/AsyncFactory.cpp;orderman/actors/FillStore.cpp;ordermanLine truncated
@@ -49,6 +49,7 @@ enum class MessageTypes : uint32_t
   TestRequest                 = '1',
   TradingSessionStatus        = 'h',
   TradingSessionStatusRequest = 'g',
+   MDIncrementalUpdate         = 'X',
   ICETraderLogin              = ('U' << 16) + ('C' << 8) + 'G',
   ICETraderLoginResponse      = ('U' << 16) + ('C' << 8) + 'I',   
   ICETraderLogout             = ('U' << 16) + ('C' << 8) + 'H',   
@@ -18,8 +18,8 @@

 #include <fmt/format.h>
 #include <fstream>
-#include <thread>
 #include <regex>
+#include <thread>

 namespace bts {
 namespace jetfix {
@@ -114,16 +114,17 @@ std::string store_path(std::string base_path,
   return result.string();
 }

-bool is_valid_store_line(std::string const&line) {
-   std::regex rx("\\d\\d\\d\\d\\d\\d\\d\\d-"
-                 "\\d\\d:\\d\\d:\\d\\d\\."
-                 "\\d\\d\\d . 8=FIX\\.4\\.2.*");
-   return std::regex_match(line, rx);
+static std::regex store_line_rx("\\d\\d\\d\\d\\d\\d\\d\\d-"
+                                "\\d\\d:\\d\\d:\\d\\d\\."
+                                "\\d\\d\\d . 8=FIX\\.4\\.2.*");
+
+bool is_valid_store_line(std::string const &line) {
+   return std::regex_match(line, store_line_rx);
 }

 std::map<uint32_t, InMsgBuffer>
 load_replay(config::JetFIX::Connection const &cfg, std::string const &path,
-            uint32_t ) {
+            uint32_t) {
   std::map<uint32_t, InMsgBuffer> result;
   if (boost::filesystem::exists(path)) {

@@ -137,7 +138,7 @@ load_replay(config::JetFIX::Connection const &cfg, std::string const &path,
         if (!f.eof()) {
            if (!is_valid_store_line(line_buffer))
               continue;
-            
+
            if (line_buffer[22] == '<') {
               auto ts =
                   datetime::parse_FIX_timestamp(line_buffer.substr(0, 21));
@@ -6,7 +6,7 @@ namespace bts {
 namespace orderman {

 class ChildTemplate;
-class LegData;
+struct LegData;

 class PyramidAlgo : public IAlgoOrder {
 public:
@@ -6,7 +6,7 @@ namespace bts {
 namespace orderman {

 class ChildTemplate;
-class LegData;
+struct LegData;

 class TASAlgo : public IAlgoOrder {
 public:
@@ -2,7 +2,7 @@
   "name": "Spark sdk container",
   "image": "spark_sdk",
   "runArgs": [
-      "--shm-size", "1024m",
+      "--shm-size", "2048m",
      "--cap-add=SYS_PTRACE", "--security-opt", "seccomp=unconfined",
      "-u", "bts"
   ],
@@ -2,7 +2,7 @@
   "name": "Spark sdk container",
   "image": "spark_sdk",
   "runArgs": [
-      "--shm-size", "1024m",
+      "--shm-size", "2048m",
      "--cap-add=SYS_PTRACE", "--security-opt", "seccomp=unconfined",
      "-u", "bts"
   ],
@@ -1 +1 @@
-set(sources algo/AlgoDefinitionServiceServer.cpp;cmds/AdminReport.cpp;cmds/AutoCert.cpp;cmds/Benchmark.cpp;cmds/CmdRegistry.cpp;cmds/ConfigCheck.cpp;cmds/FIXSim.cpp;cmds/Help.cpp;cmds/KeyCode.cpp;cmds/Md.cpp;cmds/PrintHedgeOrders.cpp;cmds/QuickTest.cpp;cmds/Run.cpp;cmds/RunSim.cpp;cmds/Span.cpp;cmds/Test.cpp;cmds/TestCeloxica.cpp;cmds/Web.cpp;dspark/mirage/quick_queue/quick_queue.cpp;exchange_sim/ExchangeSim.cpp;exchange_sim/ExchangeSimApplication.cpp;exchange_sim/book/Book.cpp;exchange_sim/book/ExchangeSim.proto;exchange_sim/book/Order.cpp;exchange_sim/book/OrderMatcher.cpp;exchange_sim/book/book_transform.cpp;exchange_sim/tests/test_book_transform.cpp;http/ApiGateway.cpp;http/MultipartFormDecoder.cpp;http/WebServer.cpp;http/tests/MultiPartFormTests.cpp;main.cpp;md/ChannelVolume.cpp;md/MarketDataAuxiliaryServiceServer.cpp;md/VolumeAtPriceState.cpp;md/platforms/IMDPlatform.cpp;md/platforms/celoxica/CXListener.cpp;md/platforms/celoxica/CXMain.cpp;md/platforms/celoxica/tests/cx_tests.cpp;md/platforms/ice_onixs/ICEListener.cpp;md/platforms/ice_onixs/ICEMain.cpp;md/platforms/ice_onixs/tests/ice_tests.cpp;md/platforms/mdp3_onixs/MDP3Listener.cpp;md/platforms/mdp3_onixs/MDP3Main.cpp;md/platforms/mdp3_onixs/tests/mdp3_tests.cpp;md/run_md.cpp;position/IPositionObserver.cpp;position/MarginServiceCore.cpp;position/MarginServiceServer.cpp;position/PositionAggregator.cpp;position/PositionMap.cpp;position/PositionServicePNLHelper.cpp;position/PositionServiceServer.cpp;position/import/BasePositionServiceServer.cpp;position/import/DownloadCommonStateMachine.cpp;position/import/DownloadWindow.cpp;position/import/MoneyFileServiceServer.cpp;position/import/parsing/gmi_parser.cpp;position/import/parsing/money_parser.cpp;position/import/parsing/parser_common.cpp;position/tests/PositionTests.cpp;position/tests/position_tests_internal.cpp;support/ErrorReportService.cpp;support/ShellLatencyServer.cpp;user/AdminServiceServer.cpp;user/AuthServiceServer.cpp;user/tests/admin_service_tests.cpp)
+set(sources algo/AlgoDefinitionServiceServer.cpp;cmds/AdminReport.cpp;cmds/AutoCert.cpp;cmds/Benchmark.cpp;cmds/CmdRegistry.cpp;cmds/ConfigCheck.cpp;cmds/FIXSim.cpp;cmds/Help.cpp;cmds/KeyCode.cpp;cmds/Md.cpp;cmds/ParseHistoricalFIX.cpp;cmds/PrintHedgeOrders.cpp;cmds/QuickTest.cpp;cmds/Run.cpp;cmds/RunSim.cpp;cmds/Span.cpp;cmds/Test.cpp;cmds/TestCeloxica.cpp;cmds/Web.cpp;dspark/mirage/quick_queue/quick_queue.cpp;exchange_sim/ExchangeSim.cpp;exchange_sim/ExchangeSimApplication.cpp;exchange_sim/book/Book.cpp;exchange_sim/book/ExchangeSim.proto;exchange_sim/book/Order.cpp;exchange_sim/book/OrderMatcher.cpp;exchange_sim/book/book_transform.cpp;exchange_sim/tests/test_book_transform.cpp;http/ApiGateway.cpp;http/MultipartFormDecoder.cpp;http/WebServer.cpp;http/tests/MultiPartFormTests.cpp;main.cpp;md/ChannelVolume.cpp;md/MarketDataAuxiliaryServiceServer.cpp;md/VolumeAtPriceState.cpp;md/platforms/IMDPlatform.cpp;md/platforms/celoxica/CXListener.cpp;md/platforms/celoxica/CXMain.cpp;md/platforms/celoxica/tests/cx_tests.cpp;md/platforms/ice_onixs/ICEListener.cpp;md/platforms/ice_onixs/ICEMain.cpp;md/platforms/ice_onixs/tests/ice_tests.cpp;md/platforms/mdp3_onixs/MDP3Listener.cpp;md/platforms/mdp3_onixs/MDP3Main.cpp;md/platforms/mdp3_onixs/tests/mdp3_tests.cpp;md/run_md.cpp;position/IPositionObserver.cpp;position/MarginServiceCore.cpp;position/MarginServiceServer.cpp;position/PositionAggregator.cpp;position/PositionMap.cpp;position/PositionServicePNLHelper.cpp;position/PositionServiceServer.cpp;position/import/BasePositionServiceServer.cpp;position/import/DownloadCommonStateMachine.cpp;position/import/DownloadWindow.cpp;position/import/MoneyFileServiceServer.cpp;position/import/parsing/gmi_parser.cpp;position/import/parsing/money_parser.cpp;position/import/parsing/parser_common.cpp;position/tests/PositionTests.cpp;position/tests/position_tests_internal.cpp;subprocess/SubProcessRunner.cpp;support/ErrorReportService.cpp;support/ShellLatencyServer.cpp;user/AdminServiceServer.cpp;user/AuthServiceServer.cpp;user/tests/admin_service_tests.cpp)
@@ -0,0 +1,416 @@
+#include "ParseHistoricalFIX.h"
+#include "lib/jetfix/codec/FIXMsg.h"
+#include "lib/jetfix/codec/FIXParser.h"
+#include "lib/jetfix/codec/Fields.h"
+#include "lib/jetfix/codec/Group.h"
+#include "lib/jetfix/codec/MessageTypes.h"
+#include "lib/utils/bts_chrono.h"
+
+#include <fstream>
+#include <unordered_map>
+
+namespace bts {
+namespace cmd {
+
+using namespace jetfix;
+
+typedef StringField<75> TradeDate;
+typedef IntField<34> SeqNum;
+typedef StringField<52> SendingTime;
+typedef StringField<60> TransactTime;
+typedef StringField<5799> MatchEventIndicator;
+
+typedef IntField<268> NoMDEntries;
+typedef StringField<279> MDUpateAction;
+typedef StringField<269> MDEntryType;
+typedef StringField<48> SecurityID;
+typedef StringField<55> Symbol;
+typedef StringField<83> ReptSeq;
+typedef FloatField<270> MDEntryPx;
+typedef IntField<271> MDEntrySize;
+typedef IntField<346> NumberOfOrders;
+typedef IntField<1023> MDPriceLevel;
+typedef StringField<286> OpenCloseSettlFlag;
+typedef StringField<731> SettlPriceType;
+typedef StringField<5797> AggressorSide;
+typedef StringField<37711> MDTradeEntryID;
+typedef StringField<5796> TradingReferenceDate;
+typedef StringField<1149> HighLimitPrice;
+typedef StringField<1148> LowLimitPrice;
+typedef StringField<1143> MaxPriceVariation;
+typedef StringField<1180> ApplID;
+typedef IntField<9633> ReferenceID;
+typedef IntField<32> LastQty;
+
+typedef IntField<37705> NoOrderIDEntries;
+typedef StringField<37> OrderID;
+typedef IntField<37707> MDOrderPriority;
+typedef IntField<37706> MDDisplayQty;
+typedef StringField<37708> OrderUpdateAction;
+
+typedef Group<NoOrderIDEntries, OrderID, MDDisplayQty, MDOrderPriority,
+              OrderUpdateAction, ReferenceID, LastQty>
+    OrderIDEntries;
+
+typedef Group<NoMDEntries, MDUpateAction, MDEntryType, SecurityID, Symbol,
+              ReptSeq, MDEntryPx, MDEntrySize, NumberOfOrders, MDPriceLevel,
+              OpenCloseSettlFlag, SettlPriceType, AggressorSide, MDTradeEntryID,
+              TradingReferenceDate, HighLimitPrice, LowLimitPrice,
+              MaxPriceVariation, ApplID, ReferenceID, LastQty, OrderID,
+              MDOrderPriority, MDDisplayQty>
+    MDEntries;
+
+typedef FIXMsg<MessageTypes::MDIncrementalUpdate, TradeDate, SeqNum,
+               SendingTime, TransactTime, MatchEventIndicator, MDEntries,
+               OrderIDEntries>
+    MDIncrementalUpdate;
+
+std::string to_str(std::string const &s) { return s; }
+
+template <class V> std::string to_str(V const &v) { return std::to_string(v); }
+
+template <class F, class G> std::string get_if_valid(G const &update) {
+   if (update.template has<F>())
+      return to_str(update.template get<F>());
+   else
+      return "";
+}
+
+void print_msg_headers(std::ostream &out) {
+   out << "MsgID,"
+       << "TradeDate,"
+       << "SeqNum,"
+       << "SendingTime,"
+       << "TransactTime,"
+       << "MatchEventIndicator,"
+       << "NoMDEntries,"
+       << "NoOrderIDEntries\n";
+}
+
+void print_msgs(size_t msg_idx, MDIncrementalUpdate const &line,
+                std::ostream &out) {
+   line.get<SeqNum>();
+   out << msg_idx << "," << get_if_valid<TradeDate>(line) << ","
+       << get_if_valid<SeqNum>(line) << "," << get_if_valid<SendingTime>(line)
+       << "," << get_if_valid<TransactTime>(line) << ","
+       << get_if_valid<MatchEventIndicator>(line) << ","
+       << (line.has<MDEntries>() ? line.get<MDEntries>().size() : 0) << ","
+       << (line.has<OrderIDEntries>() ? line.get<OrderIDEntries>().size() : 0)
+       << "\n";
+}
+
+struct PriceSize {
+   double price = 0;
+   int32_t size = 0;
+};
+
+struct Book {
+   Book() {
+      book[0].resize(10);
+      book[1].resize(10);
+   }
+   std::string symbol;
+   std::string transact_time;
+   std::string send_time;
+   uint32_t seqnum;
+   std::array<std::vector<PriceSize>, 2> book;
+};
+
+size_t n_printed_books = 2;
+
+void print_book_side(std::vector<PriceSize> const &side,
+                     std::ostream &book_stream) {
+   for (size_t i = 0; i < n_printed_books; i++)
+      if (i < side.size())
+         book_stream << "," << side[i].price << "," << side[i].size;
+      else
+         book_stream << "," << 0 << "," << 0;
+}
+
+void print_book_header(std::ostream &book_stream) {
+   book_stream << "Symbol,SendingTime,TransactTime, SeqnNum";
+   for (size_t i = 0; i < n_printed_books; i++)
+      book_stream << ",bid_p_" << i << ",bid_sz_" << i;
+   for (size_t i = 0; i < n_printed_books; i++)
+      book_stream << ",ask_p_" << i << ",ask_sz_" << i;
+   book_stream << ",\n";
+}
+
+void print_book(Book &b, std::ostream &book_stream) {
+   book_stream << b.symbol << "," << b.send_time << "," << b.transact_time
+               << "," << b.seqnum;
+   print_book_side(b.book[0], book_stream);
+   print_book_side(b.book[1], book_stream);
+   book_stream << "\n";
+}
+
+void update_and_maybe_print_book(MDIncrementalUpdate const &update,
+                                 std::unordered_map<std::string, Book> &books,
+                                 std::ostream &book_stream) {
+   std::set<Book *> to_update;
+   if (update.has<MDEntries>()) {
+      auto &md_entries = update.get<MDEntries>();
+      for (size_t i = 0; i < md_entries.size(); i++) {
+         auto &entry = md_entries[i];
+         if (entry.has<MDPriceLevel>()) {
+            if (entry.get<MDEntryType>() == "0" ||
+                entry.get<MDEntryType>() == "1") {
+               auto side = entry.get<MDEntryType>()[0] - '0';
+               auto &sym = entry.get<Symbol>();
+               auto &book = books[sym];
+               to_update.insert(&book);
+               auto &book_side = book.book[side];
+               book.symbol = sym;
+               book.transact_time = update.get<TransactTime>();
+               book.send_time = update.get<SendingTime>();
+               book.transact_time = update.get<TransactTime>();
+               book.seqnum = update.get<SeqNum>();
+               auto level = entry.get<MDPriceLevel>() - 1;
+
+               if (entry.get<MDUpateAction>() == "0") // new
+                  book_side.insert(
+                      book_side.begin() + level,
+                      {entry.get<MDEntryPx>(), entry.get<MDEntrySize>()});
+               else if (entry.get<MDUpateAction>() == "1")
+                  book_side[level] = {entry.get<MDEntryPx>(),
+                                      entry.get<MDEntrySize>()};
+               else
+                  book_side.erase(book_side.begin() + level);
+            }
+         }
+      }
+      auto indicator = update.get<MatchEventIndicator>();
+      if (indicator[5] == '1')
+         for (auto b : to_update)
+            print_book(*b, book_stream);
+   }
+}
+void print_trades_header(std::ostream &trade_stream) {
+   trade_stream << "Symbol,SendingTime,TransactTime,MDEntryPx,MDEntrySize,"
+                   "NumberOfOrders,AggressorSide,MDTradeEntryID,AgressorOID,"
+                   "RestingOID,PartialQty";
+   trade_stream << "\n";
+}
+
+struct TradeSummaryItem {
+   std::string symbol;
+   std::string sending_time;
+   std::string transact_time;
+   double md_entry_px;
+   uint32_t md_entry_size;
+   uint32_t n_orders;
+   std::string agressor_side;
+   std::string md_trade_entry_id;
+};
+
+struct TradeOrder {
+   std::string order_id;
+   uint32_t last_qty;
+};
+
+struct TradeSummaryEvent {
+   std::vector<TradeSummaryItem> md_entries;
+   std::vector<TradeOrder> orderid_entries;
+};
+
+void print_trades(TradeSummaryEvent &event, std::ostream &trade_stream) {
+   size_t current_order = 0;
+   for (auto &trade : event.md_entries) {
+      for (auto i = current_order; i < current_order + trade.n_orders; i++) {
+         trade_stream << trade.symbol << "," << trade.sending_time << ","
+                      << trade.transact_time << "," << trade.md_entry_px << ","
+                      << trade.md_entry_size << "," << trade.n_orders << ","
+                      << trade.agressor_side << "," << trade.md_trade_entry_id
+                      << "," << event.orderid_entries[current_order].order_id
+                      << "," << event.orderid_entries[i].order_id << ","
+                      << event.orderid_entries[i].last_qty << "\n";
+      }
+      current_order += trade.n_orders;
+   }
+}
+
+void update_and_maybe_print_trades(MDIncrementalUpdate const &update,
+                                   TradeSummaryEvent &event,
+                                   std::ostream &trade_stream) {
+   if (update.has<MDEntries>()) {
+      auto &md_entries = update.get<MDEntries>();
+      bool is_trade_summary = false;
+      for (size_t i = 0; i < md_entries.size(); i++) {
+
+         auto &entry = md_entries[i];
+         if (entry.get<MDEntryType>() == "2") {
+            is_trade_summary = true;
+            TradeSummaryItem item;
+            item.symbol = entry.get<Symbol>();
+            item.sending_time = update.get<SendingTime>();
+            item.transact_time = update.get<TransactTime>();
+            item.md_entry_px = entry.get<MDEntryPx>();
+            item.md_entry_size = entry.get<MDEntrySize>();
+            item.n_orders = entry.get<NumberOfOrders>();
+            item.agressor_side = entry.get<AggressorSide>();
+            item.md_trade_entry_id = entry.get<MDTradeEntryID>();
+            event.md_entries.emplace_back(std::move(item));
+         }
+
+         if (is_trade_summary) {
+            if (update.has<OrderIDEntries>()) {
+               auto &oid_entries = update.get<OrderIDEntries>();
+               for (size_t i = 0; i < oid_entries.size(); i++) {
+                  auto &oid = oid_entries[i];
+                  TradeOrder order_id_entry;
+                  order_id_entry.order_id = oid.get<OrderID>();
+                  order_id_entry.last_qty = oid.get<LastQty>();
+                  event.orderid_entries.emplace_back(std::move(order_id_entry));
+               }
+            }
+            auto indicator = update.get<MatchEventIndicator>();
+            if (indicator[7] == '1') {
+               print_trades(event, trade_stream);
+               event.md_entries.clear();
+               event.orderid_entries.clear();
+            }
+         }
+      }
+   }
+}
+
+void print_orders_header(std::ostream &orders_stream) {
+   orders_stream << "Symbol,"
+                 << "SendingTime,"
+                 << "TransactTime,"
+                 << "MDEntryPx,"
+                 << "OrderID,"
+                 << "MDDisplayQty,"
+                 << "MDOrderPriority,"
+                 << "OrderUpdateAction,"
+                 << "Side"
+                 << "\n";
+}
+
+struct Order {
+   std::string sending_time;
+   std::string transact_time;
+   std::string side;
+   
+   std::string order_id;
+   std::string order_update_action;
+   uint32_t md_order_priority;
+   uint32_t display_qty;
+
+   std::string symbol;
+   double md_entry_px;
+};
+
+void print_order(Order const& order, std::ostream &orders_stream) {
+   orders_stream << order.symbol << ","
+                 << order.sending_time << ","
+                 << order.transact_time << ","
+                 << order.md_entry_px << ","
+                 << order.order_id << ","
+                 << order.display_qty << ","
+                 << order.md_order_priority << ","
+                 << order.order_update_action << ","
+                 << order.side << "\n";
+}
+
+
+void print_orders(MDIncrementalUpdate const &update,std::ostream &orders_stream) {
+   if (update.has<MDEntries>()) {
+      auto &md_entries = update.get<MDEntries>();
+      auto & first_entry = md_entries[0];
+
+      if (first_entry.get<MDEntryType> () == "0" ||
+          first_entry.get<MDEntryType> () == "1") {
+
+         Order order;
+         order.sending_time = update.get<SendingTime>();
+         order.transact_time = update.get<TransactTime>();
+         order.side = first_entry.get<MDEntryType> ();
+         
+         if (first_entry.has<MDPriceLevel>()) {
+            // combined template https://www.cmegroup.com/confluence/display/EPICSANDBOX/Incremental+Refresh+SBE+Template+Book+Processing
+            if (update.has<OrderIDEntries>()) {
+               auto &oid_entries = update.get<OrderIDEntries>();
+               for (size_t i = 0; i < oid_entries.size(); i++) {
+
+                  auto &oid = oid_entries[i];
+                  order.order_id = oid.get<OrderID>();
+                  order.order_update_action = oid.get<OrderUpdateAction>();
+                  order.md_order_priority = oid.get<MDOrderPriority>();
+                  order.display_qty = oid.get<MDDisplayQty>();
+                  
+                  auto &md_entry = update.get<MDEntries>()[oid.get<ReferenceID>()-1];
+                  order.symbol = md_entry.get<Symbol>();
+                  order.md_entry_px = md_entry.get<MDEntryPx>();
+                  print_order(order, orders_stream);
+               }
+            }
+         }
+         else {
+            // MBO only template
+            for (size_t i = 0; i < md_entries.size(); i++) {
+               auto &entry = md_entries[i]; 
+               order.order_id = entry.get<OrderID>();
+               order.order_update_action = entry.get<MDUpateAction>();
+               order.md_order_priority = entry.get<MDOrderPriority>();
+               order.display_qty = entry.get<MDDisplayQty>();
+               order.symbol = entry.get<Symbol>();
+               order.md_entry_px = entry.get<MDEntryPx>();
+               print_order(order,orders_stream);
+            }
+         }
+      }
+   }
+}
+
+int ParseHistoricalFIX::execute(std::vector<std::string> const &args) {
+   std::ifstream input(args[0]);
+   std::ofstream books_stream(args[0] + "_books.csv");
+   std::ofstream msgs_stream(args[0] + "_msgs.csv");
+   std::ofstream trades_stream(args[0] + "_trades.csv");
+   std::ofstream orders_stream(args[0] + "_orders.csv");
+
+   std::unordered_map<std::string, Book> books;
+   TradeSummaryEvent current_trade_summary;
+
+   MDIncrementalUpdate update;
+   FIXParser<MDIncrementalUpdate> parser;
+
+   std::string line;
+   size_t c = 0;
+   auto ok = 0;
+   auto start = Timer::now();
+
+   print_book_header(books_stream);
+   print_msg_headers(msgs_stream);
+   print_trades_header(trades_stream);
+   print_orders_header(orders_stream);
+      
+   while (std::getline(input, line)) {
+      c++;
+      auto parsed_ptr = parser.parse(InMsgBuffer(
+          line.data(), line.size(), std::shared_ptr<std::string>{}));
+      if (parsed_ptr) {
+         auto msg =
+             std::dynamic_pointer_cast<const MDIncrementalUpdate>(parsed_ptr);
+         update_and_maybe_print_book(*msg, books, books_stream);
+         print_msgs(c, *msg, msgs_stream);
+         update_and_maybe_print_trades(*msg, current_trade_summary,
+                                       trades_stream);
+         print_orders(*msg, orders_stream);
+      }
+      ok += (bool)(parsed_ptr);
+   }
+   auto sec_duration = sec_diff(Timer::now(), start);
+   std::cout << (int)(c / 1000 / sec_duration) << " Km/sec parsed: " << ok
+             << " out of " << c << " time: " << sec_duration << " secs\n";
+   return 0;
+}
+
+std::string ParseHistoricalFIX::help() {
+   return "ParseHistoricalFIX <input> will produce 2 files, "
+          "<input>_mdentries.csv and <input>order_entries.csv";
+}
+
+} // namespace cmd
+} // namespace bts
@@ -0,0 +1,17 @@
+#pragma once
+
+#include "ICmd.h"
+
+namespace bts {
+namespace cmd {
+
+class ParseHistoricalFIX : public ICmd 
+{
+public:
+   int execute(std::vector<std::string> const &args);
+   std::string help ();
+};
+
+}}
+
+
@@ -12,6 +12,7 @@
 #include "server/sparkd/cmds/Help.h"
 #include "server/sparkd/cmds/KeyCode.h"
 #include "server/sparkd/cmds/Md.h"
+#include "server/sparkd/cmds/ParseHistoricalFIX.h"
 #include "server/sparkd/cmds/PrintHedgeOrders.h"
 #include "server/sparkd/cmds/QuickTest.h"
 #include "server/sparkd/cmds/Run.h"
@@ -41,6 +42,7 @@ int main(int argc, char **argv) {
   REGISTER(FIXSim);
   REGISTER(Span);
   REGISTER(Web);
+   REGISTER(ParseHistoricalFIX);
   REGISTER(PrintHedgeOrders);
   REGISTER(AdminReport);
   END_CMD;
Author	SHA1	Message	Date
Kevin Darby	b80f96fa5e	Merged in ice_excel (pull request #388 ) fix RTD crash for ice Approved-by: Jake Norbut <jake.norbut@bluetradesys.com>	2020-01-10 22:39:27 +00:00
Kevin Darby	b7dea6817c	fix RTD crash for ice	2020-01-10 14:43:06 -05:00
Pedro Pinto	f4dc550f1a	Fixed slowness in FIX log parsing. This slowness was actually causing us to take too long to complete the FIX connection logon process, resulting in a timeout and failure to establish the session. It took almost 1ms per log line, with logs commonly having 20K lines. k	2020-01-10 08:35:33 -05:00
Pedro Pinto	bc978ed065	Merge branch 'master' of bitbucket.org:bluetradesys/spark	2020-01-10 08:30:17 -05:00
Eric Harding	202c5faffd	update devcontainer to allow 2g shm	2020-01-09 10:40:47 -05:00
Pedro Pinto	65d55a7618	Added a parser of historical MD to facilitate analysis.	2020-01-02 13:57:54 -05:00
Pedro Pinto	57c3ec34c2	Fixed clang build.	2020-01-02 13:55:49 -05:00