Compare commits

...
5 Commits
Author SHA1 Message Date
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
13 changed files with 455 additions and 16 deletions

No files matched your search

@@ -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
+2
View File
@@ -0,0 +1,2 @@
dotnet pack
dotnet nuget push
+1 -1
View File
@@ -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
+1
View File
@@ -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',
+9 -8
View File
@@ -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
View File
@@ -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 ();
};
}}
+2
View File
@@ -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;