Put cme config url in GlbxMarketDataConfig proto

Instead of using magic strings

.devcontainer/Dockerfile

@@ -87,6 +87,10 @@ RUN dpkg -i /emumba/emumba-deribit-connector-3.3.1-Linux.LICENSE_DISABLED.deb \
 # Pcap
 RUN apt-get update && apt-get install -y libpcap0.8-dev
 
+# Databento (deps)
+RUN apt-get update && apt-get -y install libzstd-dev
+
+# Docker in Docker
 RUN apt-get update && apt-get install -y \
     docker.io \
     docker-compose \

.vscode/launch.json

@@ -234,7 +234,7 @@
         "program": "${workspaceFolder}/build/sparkd-debug-clang/src/app_sparkd/sparkd",
       },
       "args": [
-        "test-py-md-provider"
+        "run"
       ],
       "cwd": "${workspaceFolder}/build/sparkd-debug-clang/src/app_sparkd",
       "stopAtEntry": false,

CMakeLists.txt

@@ -1,5 +1,5 @@
 cmake_minimum_required(VERSION 3.12)
-project(spark_root)
+project(spark)
 set(CMAKE_CXX_STANDARD 20)
 
 if(WIN32)
@@ -105,6 +105,7 @@ else()
   # subdirs
   if (BUILD_SPARKD)
     add_subdirectory(src/lib_system)
+    add_subdirectory(src/lib_http)
     add_subdirectory(src/lib_domain)
     add_subdirectory(src/lib_python_md)
     add_subdirectory(src/app_sparkd)

CMakePresets.json

@@ -53,6 +53,7 @@
         "CMAKE_INSTALL_PREFIX": "${sourceDir}/install/${presetName}",
         "CMAKE_PREFIX_PATH": "${sourceDir}/cmake/;${sourceDir}/.venv/lib/python3.10/site-packages/pybind11/share/cmake/pybind11",
         "CMAKE_EXE_LINKER_FLAGS": "-Wno-undef",
+        "CMAKE_TOOLCHAIN_FILE": "${sourceDir}/vcpkg/scripts/buildsystems/vcpkg.cmake",
         "EXTERNALS_ROOT": "${sourceDir}/externals/linux",
         "BUILD_SDK": {
           "type": "BOOL",
@@ -76,7 +77,7 @@
           "value": "ON"
         },
         "CCAPI_LOG_LEVEL": "Info",
-        "CONFIG_DEFAULTS_PATH": "config/profiles/md_dev/*.conf;config/profiles/default/*.conf"
+        "CONFIG_DEFAULTS_PATH": "spark/config/profiles/md_dev/*.conf;spark/config/profiles/default/*.conf"
       }
     },
 
@@ -87,7 +88,7 @@
       "cacheVariables": {
         "CMAKE_BUILD_TYPE": "Release",
         "CCAPI_LOG_LEVEL": "Warn",
-        "CONFIG_DEFAULTS_PATH": "config/profiles/md_prod/*.conf;config/profiles/default/*.conf"
+        "CONFIG_DEFAULTS_PATH": "spark/config/profiles/md_prod/*.conf;spark/config/profiles/default/*.conf"
       }
     },
 

cmake/ExportToParent.cmake

@@ -1,3 +1,5 @@
+include_guard(GLOBAL)
+
 macro(export_to_parent)
   foreach(var IN ITEMS ${ARGN})
     set(${var} ${${var}} PARENT_SCOPE)

cmake/InitLibrary.cmake

@@ -1,66 +1,133 @@
 include_guard(DIRECTORY)
 include(${CMAKE_SOURCE_DIR}/cmake/ExportToParent.cmake)
-include(${CMAKE_SOURCE_DIR}/cmake/TernaryOperator.cmake)
+include(${CMAKE_SOURCE_DIR}/cmake/InstallBasic.cmake)
+include(${CMAKE_SOURCE_DIR}/cmake/ProtobufGenerateCpp.cmake)
+include(${CMAKE_SOURCE_DIR}/cmake/SetFirstDefined.cmake)
+include(${CMAKE_SOURCE_DIR}/cmake/safe_dep_glob.cmake)
 
 function(init_library)
   set(_options SHARED STATIC)
-  set(_one_value_args SOURCE_DIR)
+  set(_one_value_args
+    NAME
+    SOURCE_DIR
+    INCLUDE_DIR
+  )
   set(_multi_value_args)
   cmake_parse_arguments(_ARGS "${_options}" "${_one_value_args}" "${_multi_value_args}" ${ARGN})
-
   if (_ARGS_SHARED AND _ARGS_STATIC)
     message(FATAL_ERROR "Invalid args passed to init_library: cannot specify STATIC and SHARED")
   endif()
+  set_first_defined(_name _ARGS_NAME PROJECT_NAME)
+  set_first_defined(_source_dir _ARGS_SOURCE_DIR _name)
+  set_first_defined(_include_dir _ARGS_INCLUDE_DIR _source_dir)
 
-  set_ternary(_source_dir _ARGS_SOURCE_DIR _ARGS_SOURCE_DIR ${PROJECT_NAME})
-  # Glob cpp sources
+  # Cpp sources
   file(
-    GLOB_RECURSE ${PROJECT_NAME}_SOURCES
+    GLOB_RECURSE ${_name}_SOURCES
     RELATIVE ${CMAKE_CURRENT_SOURCE_DIR}
     ${_source_dir}/*.cpp
   )
   # Note: cpp files inside 'tests' dir(s) are considered tests.
   # Remove tests from main sources, and add them to test sources
-  foreach(item ${${PROJECT_NAME}_SOURCES})
-    if(${item} MATCHES "tests/*.cpp")
-      list(REMOVE_ITEM ${PROJECT_NAME}_SOURCES ${item})
-      list(APPEND ${PROJECT_NAME}_TEST_SOURCES ${item})
+  foreach(item ${${_name}_SOURCES})
+    if(${item} MATCHES "tests/.*\.cpp$")
+      list(APPEND ${_name}_TEST_SOURCES ${item})
     endif()
   endforeach()
+  foreach(item ${${_name}_TEST_SOURCES})
+    list(REMOVE_ITEM ${_name}_SOURCES ${item})
+  endforeach()
 
-  export_to_parent(
-    ${PROJECT_NAME}_SOURCES
-    ${PROJECT_NAME}_TEST_SOURCES
+  # Proto sources 
+  file(
+    GLOB_RECURSE ${_name}_PROTO_SOURCES
+    RELATIVE ${CMAKE_CURRENT_SOURCE_DIR}
+    ${_source_dir}/protobufs/*.proto
+  )
+  update_deps_file("${${_name}_PROTO_SOURCES};")
+  protobuf_generate_cpp(
+    CXX_SOURCES ${_name}_CXX_PROTO_SOURCES
+    CXX_HEADERS ${_name}_CXX_PROTO_HEADERS
+    CXX_OUT_DIR ${CMAKE_CURRENT_BINARY_DIR}
+    PROTO_INCLUDE_DIRS
+      ${EXTERNALS_INCLUDE}
+      ${CMAKE_CURRENT_SOURCE_DIR}
+    PROTO_SOURCES
+      ${${_name}_PROTO_SOURCES}
   )
 
-  # Add main library
-  if (_ARGS_STATIC)
-    add_library(${PROJECT_NAME} STATIC ${${PROJECT_NAME}_SOURCES})
-  elseif(_ARGS_SHARED)
-    add_library(${PROJECT_NAME} SHARED ${${PROJECT_NAME}_SOURCES})
-  else()
-    add_library(${PROJECT_NAME} ${${PROJECT_NAME}_SOURCES})
-  endif()
+  export_to_parent(
+    ${_name}_SOURCES
+    ${_name}_TEST_SOURCES
+    ${_name}_PROTO_SOURCES
+    ${_name}_CXX_PROTO_SOURCES
+    ${_name}_CXX_PROTO_HEADERS
+  )
 
+  # Add main lib
+  if (_ARGS_STATIC)
+    add_library(${_name} STATIC ${${_name}_SOURCES} ${${_name}_CXX_PROTO_SOURCES})
+  elseif(_ARGS_SHARED)
+    add_library(${_name} SHARED ${${_name}_SOURCES} ${${_name}_CXX_PROTO_SOURCES})
+  else()
+    add_library(${_name} ${${_name}_SOURCES} ${${_name}_CXX_PROTO_SOURCES})
+  endif()
   target_include_directories(
-    ${PROJECT_NAME}
+    ${_name}
     PUBLIC
       $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}>
       $<INSTALL_INTERFACE:${CMAKE_INSTALL_INCLUDEDIR}>
   )
+  # Include protos (if any)
+  if (${_name}_PROTO_SOURCES)
+    target_include_directories(
+      ${_name}
+      PUBLIC
+        $<BUILD_INTERFACE:${CMAKE_CURRENT_BINARY_DIR}>
+    )
+  endif()
   target_link_directories(
-    ${PROJECT_NAME}
+    ${_name}
     PUBLIC
       $<BUILD_INTERFACE:${CMAKE_CURRENT_BINARY_DIR}>
       $<INSTALL_INTERFACE:${CMAKE_INSTALL_LIBDIR}>
   )
+  # Install main lib
+  foreach(_hdr ${${_name}_CXX_PROTO_HEADERS})
+    file(RELATIVE_PATH _relative_path ${CMAKE_CURRENT_BINARY_DIR} ${_hdr})
+    install(FILES ${_hdr} DESTINATION ${CMAKE_INSTALL_INCLUDEDIR}/${_relative_path})
+  endforeach()
+  install_basic(
+    NAME ${_name}
+    INCLUDE_DIR ${_include_dir}
+  )
 
   # Add tests executable (if there are any)
-  if (${PROJECT_NAME}_TEST_SOURCES)
-    add_executable(${PROJECT_NAME}-tests ${${PROJECT_NAME}_TEST_SOURCES})
-    target_link_libraries(${PROJECT_NAME}-tests PRIVATE ${PROJECT_NAME})
+  if (${_name}_TEST_SOURCES)
+    add_executable(${_name}-tests ${${_name}_TEST_SOURCES})
+
+    get_target_property(_lib_type ${_name} TYPE)
+    if (_lib_type STREQUAL "SHARED_LIBRARY")
+      # Set RPATH so the tests exe knows where to find the shared lib.
+      set_target_properties(
+        ${_name}-tests
+        PROPERTIES
+          BUILD_RPATH "$ORIGIN"
+          INSTALL_RPATH "$ORIGIN/../${CMAKE_INSTALL_LIBDIR}"
+      )
+    endif()
+
+    # Link main lib
+    target_link_directories(
+      ${_name}-tests
+      PRIVATE
+        $<BUILD_INTERFACE:${CMAKE_CURRENT_BINARY_DIR}>
+        $<INSTALL_INTERFACE:${CMAKE_INSTALL_LIBDIR}>
+    )
+    target_link_libraries(${_name}-tests PRIVATE ${_name})
+
     # Catch2
     find_package(Catch2 CONFIG REQUIRED COMPONENTS Catch2WithMain)
-    target_link_libraries(${PROJECT_NAME}-tests PRIVATE Catch2::Catch2WithMain)
+    target_link_libraries(${_name}-tests PRIVATE Catch2::Catch2WithMain)
   endif()
 endfunction()

cmake/InitSharedLibrary.cmake

@@ -1,55 +0,0 @@
-function(init_shared_library)
-  include(${CMAKE_SOURCE_DIR}/cmake/ExportToParent.cmake)
-  set(source_dir)
-  if (${ARGC} EQUAL 0)
-    set(source_dir ${PROJECT_NAME})
-  elseif(${ARGC} GREATER 1)
-    message(FATAL_ERROR "init_shared_library received invalid arguments. Usage: init_shared_library(<OPTIONAL:source_dir>)")
-  else()
-    set(source_dir ${ARGV})
-  endif()
-
-  # Glob cpp sources
-  file(
-    GLOB_RECURSE ${PROJECT_NAME}_SOURCES
-    RELATIVE ${CMAKE_CURRENT_SOURCE_DIR}
-    ${source_dir}/*.cpp
-  )
-  # Note: cpp files inside 'tests' dir(s) are considered tests.
-  # Remove tests from main sources, and add them to test sources
-  foreach(item ${${PROJECT_NAME}_SOURCES})
-    if(${item} MATCHES "tests/*.cpp")
-      list(REMOVE_ITEM ${PROJECT_NAME}_SOURCES ${item})
-      list(APPEND ${PROJECT_NAME}_TEST_SOURCES ${item})
-    endif()
-  endforeach()
-
-  export_to_parent(
-    ${PROJECT_NAME}_SOURCES
-    ${PROJECT_NAME}_TEST_SOURCES
-  )
-
-  # Add main library
-  add_library(${PROJECT_NAME} SHARED ${${PROJECT_NAME}_SOURCES})
-  target_include_directories(
-    ${PROJECT_NAME}
-    PUBLIC
-      $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}>
-      $<INSTALL_INTERFACE:${CMAKE_INSTALL_INCLUDEDIR}>
-  )
-  target_link_directories(
-    ${PROJECT_NAME}
-    PUBLIC
-      $<BUILD_INTERFACE:${CMAKE_CURRENT_BINARY_DIR}>
-      $<INSTALL_INTERFACE:${CMAKE_INSTALL_LIBDIR}>
-  )
-
-  # Add tests executable (if there are any)
-  if (${PROJECT_NAME}_TEST_SOURCES)
-    add_executable(${PROJECT_NAME}-tests ${${PROJECT_NAME}_TEST_SOURCES})
-    target_link_libraries(${PROJECT_NAME}-tests PRIVATE ${PROJECT_NAME})
-    # Catch2
-    find_package(Catch2 CONFIG REQUIRED COMPONENTS Catch2WithMain)
-    target_link_libraries(${PROJECT_NAME}-tests PRIVATE Catch2::Catch2WithMain)
-  endif()
-endfunction(init_shared_library)

cmake/InstallBasic.cmake

@@ -0,0 +1,39 @@
+include_guard(DIRECTORY)
+include(GNUInstallDirs)
+include(${CMAKE_SOURCE_DIR}/cmake/PackBasic.cmake)
+include(${CMAKE_SOURCE_DIR}/cmake/SetFirstDefined.cmake)
+
+function(install_basic)
+  set(_options)
+  set(_one_value_args NAME INCLUDE_DIR)
+  set(_multi_value_args)
+  cmake_parse_arguments(_ARGS "${_options}" "${_one_value_args}" "${_multi_value_args}" ${ARGN})
+  set_first_defined(_name _ARGS_NAME PROJECT_NAME)
+  set_first_defined(_include_dir _ARGS_INCLUDE_DIR _name)
+
+  file(
+    GLOB_RECURSE ${_name}_HEADERS
+    RELATIVE ${CMAKE_CURRENT_SOURCE_DIR}
+    ${_include_dir}/*.h)
+
+  target_sources(${_name}
+    PUBLIC
+      FILE_SET HEADERS FILES ${${_name}_HEADERS})
+
+  install(
+    TARGETS ${_name}
+    EXPORT ${_name}Targets
+    RUNTIME DESTINATION ${CMAKE_INSTALL_BINDIR}
+    LIBRARY DESTINATION ${CMAKE_INSTALL_LIBDIR}
+    ARCHIVE DESTINATION ${CMAKE_INSTALL_LIBDIR}
+    FILE_SET HEADERS DESTINATION ${CMAKE_INSTALL_INCLUDEDIR}
+  )
+
+  install(
+    EXPORT ${_name}Targets
+    FILE ${_name}Targets.cmake
+    NAMESPACE ${_name}::
+    DESTINATION ${CMAKE_INSTALL_LIBDIR}/cmake/${_name})
+
+  pack_basic()
+endfunction()

cmake/InstallBasicConfig.cmake

@@ -1,30 +0,0 @@
-include(${CMAKE_SOURCE_DIR}/cmake/PackBasic.cmake)
-include(GNUInstallDirs)
-
-function(install_basic_config)
-  file(
-    GLOB_RECURSE ${PROJECT_NAME}_HEADERS
-    RELATIVE ${CMAKE_CURRENT_SOURCE_DIR}
-    *.h)
-
-  target_sources(${PROJECT_NAME}
-    PUBLIC
-      FILE_SET HEADERS FILES ${${PROJECT_NAME}_HEADERS})
-
-  install(
-    TARGETS ${PROJECT_NAME}
-    EXPORT ${PROJECT_NAME}Targets
-    RUNTIME DESTINATION ${CMAKE_INSTALL_BINDIR}
-    LIBRARY DESTINATION ${CMAKE_INSTALL_LIBDIR}
-    ARCHIVE DESTINATION ${CMAKE_INSTALL_LIBDIR}
-    FILE_SET HEADERS DESTINATION ${CMAKE_INSTALL_INCLUDEDIR}
-  )
-
-  install(
-    EXPORT ${PROJECT_NAME}Targets
-    FILE ${PROJECT_NAME}Targets.cmake
-    NAMESPACE ${PROJECT_NAME}::
-    DESTINATION ${CMAKE_INSTALL_LIBDIR}/cmake/${PROJECT_NAME})
-
-  pack_basic()
-endfunction()

cmake/PackBasic.cmake

@@ -1,6 +1,14 @@
+include_guard(DIRECTORY)
 include(CMakePackageConfigHelpers)
+include(${CMAKE_SOURCE_DIR}/cmake/SetFirstDefined.cmake)
 
 function(pack_basic)
+  set(_options)
+  set(_one_value_args NAME)
+  set(_multi_value_args)
+  cmake_parse_arguments(_ARGS "${_options}" "${_one_value_args}" "${_multi_value_args}" ${ARGN})
+  set_first_defined(_name _ARGS_NAME PROJECT_NAME)
+
   find_package(Git QUIET)
 
   if(GIT_FOUND)
@@ -17,20 +25,20 @@ function(pack_basic)
   endif()
 
   write_basic_package_version_file(
-    "${CMAKE_CURRENT_BINARY_DIR}/${PROJECT_NAME}ConfigVersion.cmake"
+    "${CMAKE_CURRENT_BINARY_DIR}/${_name}ConfigVersion.cmake"
     VERSION ${PROJECT_VERSION}
     COMPATIBILITY AnyNewerVersion
   )
 
   configure_package_config_file(
     "${CMAKE_CURRENT_SOURCE_DIR}/Config.cmake.in"
-    "${CMAKE_CURRENT_BINARY_DIR}/${PROJECT_NAME}Config.cmake"
-    INSTALL_DESTINATION ${CMAKE_INSTALL_LIBDIR}/cmake/${PROJECT_NAME}
+    "${CMAKE_CURRENT_BINARY_DIR}/${_name}Config.cmake"
+    INSTALL_DESTINATION ${CMAKE_INSTALL_LIBDIR}/cmake/${_name}
   )
 
   install(
     FILES
-      "${CMAKE_CURRENT_BINARY_DIR}/${PROJECT_NAME}Config.cmake"
-      "${CMAKE_CURRENT_BINARY_DIR}/${PROJECT_NAME}ConfigVersion.cmake"
-    DESTINATION ${CMAKE_INSTALL_LIBDIR}/cmake/${PROJECT_NAME})
+      "${CMAKE_CURRENT_BINARY_DIR}/${_name}Config.cmake"
+      "${CMAKE_CURRENT_BINARY_DIR}/${_name}ConfigVersion.cmake"
+    DESTINATION ${CMAKE_INSTALL_LIBDIR}/cmake/${_name})
 endfunction()

cmake/ProtobufGenerateCpp.cmake

@@ -2,8 +2,16 @@ include_guard(GLOBAL)
 
 function(protobuf_generate_cpp)
   set(_options)
-  set(_one_value_args CXX_SOURCES CXX_HEADERS CXX_OUT_DIR PROTOC)
-  set(_multi_value_args PROTO_INCLUDE_DIRS PROTO_SOURCES)
+  set(_one_value_args
+    CXX_SOURCES
+    CXX_HEADERS
+    CXX_OUT_DIR
+    PROTOC
+  )
+  set(_multi_value_args
+    PROTO_INCLUDE_DIRS
+    PROTO_SOURCES
+  )
   cmake_parse_arguments(_ARGS "${_options}" "${_one_value_args}" "${_multi_value_args}" ${ARGN})
 
   if (NOT DEFINED _ARGS_PROTOC)
@@ -26,6 +34,7 @@ function(protobuf_generate_cpp)
     endif()
     list(APPEND ${_ARGS_CXX_SOURCES} "${_ARGS_CXX_OUT_DIR}/${_fil_we}.pb.cc")
     list(APPEND ${_ARGS_CXX_HEADERS} "${_ARGS_CXX_OUT_DIR}/${_fil_we}.pb.h")
+    message(STATUS "add_custom_command for ${_fil} -> ${_ARGS_CXX_OUT_DIR}/${_fil_we}.pb.h")
     add_custom_command(
       OUTPUT
         "${_ARGS_CXX_OUT_DIR}/${_fil_we}.pb.cc"
@@ -33,7 +42,7 @@ function(protobuf_generate_cpp)
       COMMAND
         ${_ARGS_PROTOC}
       ARGS
-        --cpp_out=${_ARGS_CXX_OUT_DIR} ${_proto_include_cli_args} ${_abs_fil}
+        --cpp_out=${_ARGS_CXX_OUT_DIR} ${_proto_include_cli_args} ${_fil}
       DEPENDS ${_abs_fil}
       COMMENT "Running C++ protocol buffer compiler on ${_fil}"
       VERBATIM

cmake/ProtobufGenerateCppConfig.cmake

@@ -0,0 +1,58 @@
+function(protobuf_generate_cpp_config)
+  set(_options)
+  set(_one_value_args
+    PROTO_SOURCES_HEADER
+    PROTO_DEFAULTS_HEADER
+  )
+  set(_multi_value_args
+    PROTO_SOURCES
+    PROTO_DEFAULTS
+  )
+  cmake_parse_arguments(_ARGS "${_options}" "${_one_value_args}" "${_multi_value_args}" ${ARGN})
+
+  # Configure the 'config_proto_files' header. It provides an array of strings populated with the
+  # files in PROTO_SOURCES. At runtime, we use this array to generate a special 'config' proto
+  # message with fields for each message defined in said files. A global instance of this 'config'
+  # proto is then constructed, and populated with [compile-time] defaults, and [runtime] overrides
+  # (i.e., in spark.conf). Calls to `bts::config::get<T>()` retrieve the value from the field
+  # associated with type `T` in this global instance.
+  message(STATUS "PROTO_SOURCES: ${_ARGS_PROTO_SOURCES}")
+  message(STATUS "PROTO_SOURCES_HEADER: ${_ARGS_PROTO_SOURCES_HEADER}")
+  message(STATUS "PROTO_DEFAULTS_HEADER: ${_ARGS_PROTO_DEFAULTS_HEADER}")
+  if (_ARGS_PROTO_SOURCES_HEADER)
+    set(PROTO_SOURCES)
+    set(PROTO_SOURCES_CXX_ARRAY_ITEMS "")
+    foreach(_proto_path ${_ARGS_PROTO_SOURCES})
+      list(APPEND PROTO_SOURCES ${_proto_path})
+      string(APPEND PROTO_SOURCES_CXX_ARRAY_ITEMS "\"${_proto_path}\",\n")
+    endforeach()
+    list(LENGTH PROTO_SOURCES PROTO_SOURCES_LENGTH)
+    message(STATUS "CONFIGURING PROTO_SOURCES_HEADER): ${_ARGS_PROTO_SOURCES_HEADER}")
+    configure_file(${CMAKE_SOURCE_DIR}/cmake/config_proto_files.h.in ${_ARGS_PROTO_SOURCES_HEADER})
+  endif()
+
+  # Generate static defaults. Ideally, this would be a single string, but we use an array because
+  # the compiler complains if the string is longer than 65k.
+  set(_cfg_defaults_str "const char * config_static_defaults[] = { ")
+
+  foreach(_cfg_file_path ${_ARGS_PROTO_DEFAULTS})
+    file(READ ${_cfg_file_path} _cfg_file_content)
+    string(REPLACE "\"" "\\\"" _cfg_file_content_escaped ${_cfg_file_content})
+    string(REPLACE "\n" ";" _cfg_lines ${_cfg_file_content_escaped})
+
+    foreach(_line ${_cfg_lines})
+      string(APPEND _cfg_defaults_str "\n  \"${_line}\",")
+    endforeach(_line)
+  endforeach(_cfg_file_path)
+
+  string(APPEND _cfg_defaults_str "};")
+  set(_old_cfg_defaults_str "")
+
+  if(EXISTS ${_ARGS_PROTO_DEFAULTS_HEADER})
+    file(READ ${_ARGS_PROTO_DEFAULTS_HEADER} _old_cfg_defaults_str)
+  endif()
+
+  if(NOT _cfg_defaults_str STREQUAL _old_cfg_defaults_str)
+    file(WRITE "${_ARGS_PROTO_DEFAULTS_HEADER}" "${_cfg_defaults_str}")
+  endif()
+endfunction()

cmake/SetFirstDefined.cmake

@@ -0,0 +1,10 @@
+include_guard(GLOBAL)
+
+macro(set_first_defined RESULT)
+  foreach(_arg ${ARGN})
+    if (DEFINED ${_arg})
+      set(${RESULT} ${${_arg}})
+      break()
+    endif()
+  endforeach()
+endmacro()

cmake/config_proto_files.h.in

@@ -0,0 +1,11 @@
+#pragma once
+
+#include <array>
+
+namespace bts::@PROJECT_NAME@::protobufs {
+
+inline std::array<const char *, @PROTO_SOURCES_LENGTH@> config_proto_files = {
+@PROTO_SOURCES_CXX_ARRAY_ITEMS@
+};
+
+} // namespace bts::@PROJECT_NAME@::protobufs

docker/ubuntu-22.04/dotnet-7/Dockerfile

@@ -24,7 +24,6 @@ RUN locale-gen en_US.UTF-8
 ARG docker_user="bts"
 ARG docker_uid="1000"
 ARG docker_pass="bts"
-# ARG entrypoint
 
 RUN \
   groupadd -g "$docker_uid" "$docker_user" \

src/DataStreamSuite/Bts.DataStreamWebApp.Server/Dockerfile

@@ -1,7 +1,11 @@
 FROM bluetradingsystems/linux_dotnet7
 
+ARG SOURCE_DIR="build/Release/bin/net7.0/linux-x64/publish"
+ARG WEB_ROOT="build/Bts.DataStreamWebApp.Client/production/dist"
+ARG CONF_FILE="src/DataStreamSuite/Bts.DataStreamWebApp.Server/spark.conf"
+
 RUN dotnet dev-certs https --trust
-COPY --chown=bts:bts "build/Release/bin/net7.0/linux-x64/publish" "bin"
-COPY --chown=bts:bts "build/Bts.DataStreamWebApp.Client/production/dist" "bin/wwwroot"
-COPY --chown=bts:bts "src/DataStreamSuite/Bts.DataStreamWebApp.Server/spark.conf" ".local/share/bts/Bts.DataStreamWebApp.Server/conf/spark.conf"
+COPY --chown=bts:bts "${SOURCE_DIR}" "bin"
+COPY --chown=bts:bts "${WEB_ROOT}" "bin/wwwroot"
+COPY --chown=bts:bts "${CONF_FILE}" ".local/share/bts/Bts.DataStreamWebApp.Server/conf/spark.conf"
 WORKDIR "/home/bts/bin"

src/DataStreamSuite/Bts.DataStreamWebApp.Server/production.Dockerfile

@@ -1,6 +0,0 @@
-FROM bluetradingsystems/linux_dotnet7
-
-RUN dotnet dev-certs https --trust
-COPY --chown=bts:bts "bin" "bin"
-COPY --chown=bts:bts "webapp.spark.conf" ".local/share/DataStream/conf/spark.conf"
-WORKDIR "/home/bts/bin"

src/DataStreamSuite/DataStream/Primitives/FileStreams/DumpLoaderStream.cs

@@ -1,49 +0,0 @@
-// using System;
-// using System.Collections.Generic;
-// using System.Text;
-// using System.IO;
-
-// using System.Runtime.Serialization.Formatters.Binary;
-
-// namespace BCG
-// {
-//    public class DumpLoaderStream : ViewStream, ISideEffectStream
-//    {
-//       //wtf
-//       //public class Exception : System.Exception
-//       //{public Exception(string msg) : base(msg) { }}
-
-//       public DumpLoaderStream(string fileName, int secondsBetweenPolls)
-//       {
-//          mySetRealSource(
-//             new PollStream(() => new LoaderStream(fileName).rows, secondsBetweenPolls, 
-//                ViewStreamDatabase.instance));
-//       }
-
-//       public DumpLoaderStream(string fileName)
-//       {
-//         if (fileName == null)
-//            throw new LoaderStream.Exception("File name cannot be null");
-
-//         for (int i = 0; i < 5; i++)
-//            try
-//            {
-//               using (FileStream fs = new FileStream(fileName, FileMode.Open, FileAccess.Read))
-//               {
-//                  IClientDeserializer deser = new ClientUniversalDeserializer();
-//                  byte[] the_ops;
-//                  Array the_rows;
-//                  deser.deserialize(new Stream[] { fs }, out the_ops, out the_rows);
-
-//                  mySetRealSource(new BlueRowCollectionStaticStream((IBlueRowCollection)the_rows.GetValue(0)));
-//               }
-//               break;
-//            }
-//            catch (IOException)
-//            {
-//               System.Threading.Thread.Sleep(50);
-//            }
-
-//       }
-//    }
-// }

src/DataStreamSuite/DataStream/Primitives/FileStreams/LoaderStream.cs

@@ -1,6 +1,7 @@
 using BCG;
 
 namespace bts {
+   // This is the other side of the SaverStream, its "loads" the stream and fires once
    [DataStreamUserRespresentation("loader")]
    public class LoaderStream : ViewStream {
 

src/DataStreamSuite/DataStream/Primitives/FileStreams/PersistorStream.cs

@@ -5,10 +5,11 @@ using System.Text;
 using System.IO;
 using Google.Protobuf;
 using System.Globalization;
-using bts;
+using BCG;
 
-namespace BCG {
-   // Writes all the udpates to a stream in a file.
+namespace bts {
+   // Writes all the udpates to a stream in a file or multiples files dependong on the rollover size (as opposed to Saver
+   // which just writes one update, as a create)
    // File naming:
    // token_epoch_yyyyMMdd_HHmmss
    // File location: directory(passed)\token\
@@ -94,10 +95,10 @@ namespace BCG {
       void _on_change(object new_value) {
          DataPacket dp = new_value as DataPacket;
          byte[] serialized_update = null;
-         bool _write_create = _serializer == null || _n_updates_in_current_file == 0;
+         bool first_write = _serializer == null || _n_updates_in_current_file == 0;
          _serializer = _serializer ?? new UniversalStreamIOSerializer();
-         if (_write_create) {
-            // new files always starts with create
+         if (first_write) {
+            // first file always starts with create
             serialized_update = _serializer.serialize(new bts.StreamIOMessage(new bts.StreamIOMessage.CreateMessage(dp.newRows)));
          } else
             serialized_update = _serializer.serialize(new bts.StreamIOMessage(new bts.StreamIOMessage.MultipleRemoveAddMessage(dp.deletedRows, dp.addedRows)));
@@ -105,8 +106,8 @@ namespace BCG {
          long utc_epoch = epoch_now();
 
          //time, length of data, data
-         bw.Write(utc_epoch);
-         bw.Write(serialized_update.Length);
+         //bw.Write(utc_epoch);
+         //bw.Write(serialized_update.Length);
          bw.Write(serialized_update);
          _out_fs.Flush();
          myFireEvent(dp);

src/DataStreamSuite/DataStream/Primitives/FileStreams/PlaybackStream.cs

@@ -1,190 +0,0 @@
-// using System;
-// using System.Collections.Generic;
-// using System.Linq;
-// using System.Text;
-// using System.IO;
-// using System.Runtime.Serialization.Formatters.Binary;
-// using BCG.Utilities.General;
-// using bts.collections;
-// using System.Reflection.Metadata;
-
-
-// namespace BCG {
-
-//    public class PlaybackStream : ViewStream, IWorker {
-//       private readonly PlaybackType _type;
-//       private BlueTable _current_table;
-//       private IClientDeserializer _deserilializer = new ClientUniversalDeserializer();
-//       // these are the packets already downloaded from the current file
-//       private List<PlaybackPacket> _packets = new List<PlaybackPacket>();
-//       //private IEnumerable<FileStream> _file_streams = new List<FileStream>();
-//       private IEnumerable<string> _file_names;
-//       private Tuple<int, FileStream> _current_fs = new Tuple<int, FileStream>(-1, null);
-//       private DateTime _playback_t0, _playback_t1;
-//       private bool _all_files_read = false;
-//       SnapshotData _snapshot_data = null; // only initialized when in snapshot mode
-
-//       public PlaybackStream(string datetime_start, string datetime_end, string directory_name, string token) {
-//          _type = PlaybackType.Streaming;
-//          _playback_t0 = PersistorStream.parse_datetime(datetime_start);
-//          _playback_t1 = PersistorStream.parse_datetime(datetime_end);
-//          string token_directory_name = $"{directory_name}\\{token}";
-//          bts.Log.Gen.info($"PlaybackStream t0: {_playback_t0} t1: {_playback_t1} dir: {token_directory_name}");
-//          List<string> files_in_range = new List<string>();
-//          if (Directory.Exists(token_directory_name)) {
-//             _file_names =
-//          Directory.GetFiles(token_directory_name).Where(f => {
-//             DateTime file_time = PersistorStream.parse_datetime_from_filename(f);
-//             return file_time > _playback_t0 && file_time < _playback_t1;
-//          });
-//          }
-//       }
-
-//       protected override void myOnFirstSubscriber() {
-//          FileStream fs = new FileStream(_snapshot_data.snapshot_file, FileMode.Open, FileAccess.Read);
-//          _snapshot_data.packets = StreamPersistorUtils.load_packets_from_file(fs);
-//          PlaybackPacket snapshot_packet = StreamPersistorUtils.find_packet_at_time(_snapshot_data.packets, _snapshot_data.snapshot_time);
-//          if (snapshot_packet != null) {
-//             foreach (PlaybackPacket p in _snapshot_data.packets) {
-//                byte[] ops;
-//                Array row_array;
-//                _deserilializer.deserialize(new Stream[] { new MemoryStream(p.data) }, out ops, out row_array);
-//                for (int i = 0; i < ops.Length; i++) {
-//                   IBlueRowCollection[] rows_args = row_array.GetValue(i) as IBlueRowCollection[];
-//                   byte op = ops[i];
-//                   if (op == StreamOpCode.create) {
-//                      IBlueRowCollection full_table = rows_args[0];
-//                      _current_table = new BlueTable(full_table);
-//                   } else if (op == StreamOpCode.multipleRemoveAdd) {
-//                      //IBlueRowCollection[] del_added = rows_args.GetValue(0) as IBlueRowCollection[];
-//                      IBlueRowCollection dels = rows_args.GetValue(0) as IBlueRowCollection;
-//                      IBlueRowCollection adds = rows_args.GetValue(1) as IBlueRowCollection;
-//                      BlueDataUtility.absorbChanges(_current_table, dels, adds);
-//                   } else {
-//                      bts.Log.assert(false, $"Oprator: {op} is not expected");
-//                   }
-//                   if (p.idx == snapshot_packet.idx) {
-//                      myFireEvent(new DataPacket(_current_table, _current_table, new BlueTable(_current_table.schema)));
-//                      break;
-//                   }
-//                }
-//             }
-//          }
-//       }
-
-
-//       private void _load_packets_from_current_file() {
-//          int curr_idx = _current_fs.Item1;
-//          FileStream curr_fs = _current_fs.Item2;
-//          _packets = new List<PlaybackPacket>();
-//          _packets.AddRange(StreamPersistorUtils.load_packets_from_file(curr_fs));
-//       }
-
-//       private bool current_file_all_replayed {
-//          get {
-//             bool rv = true;
-//             _packets.ForEach(p => rv = rv && p.replayed);
-//             return rv;
-//          }
-//       }
-
-//       protected override void myOnLastSubscriber() {
-//          var fs = _current_fs.Item2;
-//          if (fs != null) {
-//             fs.Close();
-//          }
-//          ViewStreamDatabase.instance.removeWorker(this);
-//       }
-
-//       // public event PlaybackStreamProgress onProgress;
-//       // public event Action onFileExhausted;
-//       private PlaybackPacket _locate_packet_to_replay() {
-//          foreach (var p in _packets) {
-//             // returns the first non replayed packet
-//             if (!p.replayed)
-//                return p;
-//          }
-//          //else
-//          {
-//             double secondsElapsed = new BinaryReader(myFileReader).ReadDouble();
-//             myNextPacketTime = myPlaybackStart + new TimeSpan(0, 0, 0, 0, (int)(1000 * secondsElapsed / myPlaybackSpeed));
-//             if (onProgress != null)
-//                onProgress(myFileReader.Position, myFileReader.Length);
-//          }
-//       }
-
-//       #region IWorker Members
-
-//       public void doWork() {
-//          if (_all_files_read)
-//             return;
-//          if (current_file_all_replayed) {
-//             _load_next_file();
-//          }
-//          PlaybackPacket curr_packet = _locate_packet_to_replay();
-//          if (curr_packet.idx == 0) { // first packet in the file, fire it rightaway 
-//             byte[] ops;
-//             Array row_array;
-//             // what a stupid interface!
-//             _deserilializer.deserialize(new Stream[] { new MemoryStream(curr_packet.data) }, out ops, out row_array);
-//             _fire_packet(ops, row_array);
-//             curr_packet.replay_timestamp = PersistorStream.epoch_now();
-//          } else {
-//             PlaybackPacket prev_packet = _get_previous_packet(curr_packet);
-//             // this is the time we need to fire (replay) this packet
-//             long time_to_fire = prev_packet.replay_timestamp + curr_packet.offset_from_previous;
-//             long now_epoch = PersistorStream.epoch_now();
-//             // fire when we are past the 
-//             if (now_epoch >= time_to_fire) {
-//                byte[] ops;
-//                Array row_array;
-//                _deserilializer.deserialize(new Stream[] { new MemoryStream(curr_packet.data) }, out ops, out row_array);
-//                _fire_packet(ops, row_array);
-//                curr_packet.replay_timestamp = PersistorStream.epoch_now();
-//             }
-//          }
-//       }
-
-//       private void _fire_packet(byte[] ops, Array row_array) {
-//          for (int i = 0; i < ops.Length; i++) {
-//             IBlueRowCollection[] rows_args = row_array.GetValue(i) as IBlueRowCollection[];
-//             byte op = ops[i];
-//             if (op == StreamOpCode.create) {
-//                IBlueRowCollection full_table = rows_args[0];
-//                _current_table = new BlueTable(full_table);
-//                //_current_table.addRows(full_table);
-//                myFireEvent(new DataPacket(full_table, full_table, myLastValue == null ? new BlueTable(full_table.schema) :
-//                                           (myLastValue as DataPacket).newRows));
-//             } else if (op == StreamOpCode.multipleRemoveAdd) {
-//                //IBlueRowCollection[] del_added = rows_args.GetValue(0) as IBlueRowCollection[];
-//                IBlueRowCollection dels = rows_args.GetValue(0) as IBlueRowCollection;
-//                IBlueRowCollection adds = rows_args.GetValue(1) as IBlueRowCollection;
-//                BlueDataUtility.absorbChanges(_current_table, dels, adds);
-//                myFireEvent(new DataPacket(_current_table, adds, dels));
-//             } else {
-//                bts.Log.assert(false, $"Oprator: {op} is not expected");
-//             }
-
-//          }
-//       }
-
-//       private void _load_next_file() {
-//          var new_idx = _current_fs.Item1 + 1;
-//          if (new_idx >= _file_names.Count()) {
-//             _all_files_read = true;
-//             bts.Log.Gen.info("All files read");
-//             return;
-//          }
-//          var prev_fs = _current_fs.Item2;
-//          if (prev_fs != null)
-//             prev_fs.Close();
-//          var new_filename = _file_names.ElementAt(new_idx);
-//          bts.Log.Gen.info("PlaybackStream_load_next_file new_filename=: {new_filename}");
-//          var new_fs = new FileStream(new_filename, FileMode.Open, FileAccess.Read);
-//          _current_fs = new Tuple<int, FileStream>(new_idx, new_fs);
-//          _load_packets_from_current_file();
-//       }
-
-//       #endregion
-//    }
-// }

src/DataStreamSuite/DataStream/Primitives/FileStreams/RecorderStream.cs

@@ -1,68 +0,0 @@
-// using System;
-// using System.Collections.Generic;
-// using System.Linq;
-// using System.Text;
-// using System.IO;
-
-// namespace BCG
-// {
-//    public class RecorderStream : ViewStream
-//    {
-//       private ViewStream mySource;
-//       private DateTime myStartTime;
-//       private FileStream myOutFile;
-//       private string myFileName;
-//       private IBlueDataSerializer mySerializer;
-
-//       public RecorderStream(ViewStream src, string recordFile)
-//       {
-//          mySource = src;
-//          myFileName = recordFile;
-
-//          myFileName = myFileName.Replace("%DATE%", DateTime.Now.ToString("yyyyMMdd_HHmmss")); 
-//       }
-
-//       protected override void myOnFirstSubscriber()
-//       {
-//          myStartTime = DateTime.Now;         
-//          mySource.onChange += new ChangeDelegate(mySource_onChange);
-//          myOutFile = new FileStream(myFileName, FileMode.OpenOrCreate);
-//       }
-
-//       protected override void myOnLastSubscriber()
-//       {
-//          mySource.onChange -= mySource_onChange;
-//          myOutFile.Close();
-//       }
-
-//       void mySource_onChange(object newValue)
-//       {
-//          DataPacket dp = (DataPacket)newValue;         
-
-//          BlueTable toWrite = new BlueTable(dp.schema);
-
-//          if (mySerializer == null)
-//          {
-//             mySerializer = new UniversalSerializer(dp.schema);
-//          }
-//          foreach (IBlueRow r in dp.newRows)
-//             toWrite.addRow(r);
-
-//          byte[] result;
-
-//          if (myLastValue == null)
-//             mySerializer.serialize(StreamOpCode.create, new IBlueRowCollection[] { dp.newRows }, out result);
-//          else
-//             mySerializer.serialize(StreamOpCode.multipleRemoveAdd, new IBlueRowCollection[] { dp.deletedRows, dp.addedRows }, out result);
-
-//          BinaryWriter bw = new BinaryWriter(myOutFile);
-//          //write down what time this happened
-//          bw.Write((DateTime.Now - myStartTime).TotalSeconds);
-//          bw.Write(result.Length);
-//          bw.Write(result);         
-//          myOutFile.Flush();
-
-//          myFireEvent(dp);
-//       }
-//    }
-// }

src/DataStreamSuite/DataStream/Primitives/FileStreams/SaverStream.cs

@@ -5,7 +5,8 @@ using System.IO;
 using BCG;
 
 namespace bts {
-   // this saves the total state in the filr provided in the cons
+   // this saves the updates as a Create on the same file, each update calls -save, which
+   // reqrites the whole data to the packet. The other side of this stream is the Loader
    [DataStreamUserRespresentation("saver")]
    public class SaverStream : ViewStream {
       private ViewStream _stream;
@@ -31,7 +32,6 @@ namespace bts {
          StreamIOMessage.CreateMessage cr_msg = new StreamIOMessage.CreateMessage(dp.newRows);
          byte[] str_bytes = _serializer.serialize(new StreamIOMessage(cr_msg));
          StreamSerializationUtils.write_to_file(str_bytes, _filename);
-
       }
    }
 }

src/DataStreamSuite/DataStream/Primitives/FileStreams/StreamPersistorUtils.cs

@@ -3,6 +3,7 @@ using System.Collections.Generic;
 using System.Linq;
 using System.Globalization;
 using System.IO;
+using bts;
 
 
 namespace BCG {

src/DataStreamSuite/DataStreamPublisher/Dockerfile

@@ -1,5 +1,8 @@
 FROM bluetradingsystems/linux_dotnet7
 
+ARG SOURCE_DIR="build/Release/bin/net7.0/linux-x64/publish"
+ARG CONF_FILE="src/DataStreamSuite/DataStreamPublisher/spark.conf"
+
 ENV DOCKERIZE_VERSION v0.7.0
 USER root
 RUN apt-get update \
@@ -8,6 +11,6 @@ RUN apt-get update \
     && apt-get autoremove -yqq --purge wget && rm -rf /var/lib/apt/lists/*
 
 USER bts
-COPY --chown=bts:bts "build/Release/bin/net7.0/linux-x64/publish" "bin"
-COPY --chown=bts:bts "src/DataStreamSuite/DataStreamPublisher/spark.conf" ".local/share/bts/DataStreamPublisher/conf/spark.conf"
+COPY --chown=bts:bts "${SOURCE_DIR}" "bin"
+COPY --chown=bts:bts "${CONF_FILE}" ".local/share/bts/DataStreamPublisher/conf/spark.conf"
 WORKDIR "/home/bts/bin"

src/DataStreamSuite/PubSub/ManagedTCP/AsyncTCPConnection_APM.cs

@@ -37,14 +37,17 @@ using Bts.PubSub;
 using Spark.Core;
 using mbts.utils;
 
-namespace bts {
-   public class AsyncTCPConnection_APM : ITCPConnection, IWorker, IDisposable {
+namespace bts
+{
+   public class AsyncTCPConnection_APM : ITCPConnection, IWorker, IDisposable
+   {
 
       // this class is used as the argument of the completion function for BeginSend/Receive   
       // serves as the base class for both Send and ReeceiveMessage classes.
       // since both TCP sends and receives happen in multiple transfers, this
       // provides a wrapper around copying the partial transfers into the complete buffer
-      public class AsyncMessageBase {
+      public class AsyncMessageBase
+      {
          readonly string msg_id = Guid.NewGuid().ToString();
          // every BeginSend or BeginReceive has a unique key associated with it
          public int key;
@@ -56,38 +59,45 @@ namespace bts {
          public int transfer_ptr;
          protected int _n_transfers = 0;
          public bool completed_sync = false;
-         public AsyncMessageBase(byte[] the_buffer, int the_offset) {
+         public AsyncMessageBase(byte[] the_buffer, int the_offset)
+         {
             buffer = the_buffer;
             transfer_ptr = the_offset;
          }
-         public int remaining_size {
-            get {
+         public int remaining_size
+         {
+            get
+            {
                return buffer.Length - transfer_ptr;
             }
          }
 
          public int original_size => buffer.Length;
 
-         public void advance_transfer_pointer(int advance_by) {
+         public void advance_transfer_pointer(int advance_by)
+         {
             transfer_ptr += advance_by;
             _n_transfers++;
-            if (transfer_ptr > original_size) {
+            if (transfer_ptr > original_size)
+            {
                bts.Log.Gen.error($"transfer went beyond initial size {ToString()}");
             }
             // TODO: log if it goes past the buffer
          }
 
-         public bool fully_transfered() {
+         public bool fully_transfered()
+         {
             return transfer_ptr >= original_size;
          }
 
-         public override string ToString() {
+         public override string ToString()
+         {
             return $"{GetType().ToString()} msg_id: {msg_id} key: {key} original_length: {original_size} transfer_ptr: {transfer_ptr} n_transfers: {_n_transfers} " +
                    $"fully_transfered: {fully_transfered()} completed_sync: {completed_sync}";
          }
       }
-
-      public class AsyncSendMessage : AsyncMessageBase {
+      public class AsyncSendMessage : AsyncMessageBase
+      {
          public string message_id;
          public SendConfirmationDelegate? on_send_del;
          //private int _n_transfers = 0;
@@ -105,25 +115,31 @@ namespace bts {
          }
       }
 
-      public class AsyncReceiveMessage : AsyncMessageBase {
-         public enum ReadState {
+      public class AsyncReceiveMessage : AsyncMessageBase
+      {
+         public enum ReadState
+         {
             reading_size,
             reading_data
          };
          readonly ReadState read_state;
 
          public AsyncReceiveMessage(int size, ReadState the_read_state) :
-         this(new byte[size], 0, the_read_state) { }
+         this(new byte[size], 0, the_read_state)
+         { }
          public AsyncReceiveMessage(byte[] the_buffer, int the_transfer_ptr, ReadState the_read_state)
-         : base(the_buffer, the_transfer_ptr) {
+         : base(the_buffer, the_transfer_ptr)
+         {
             read_state = the_read_state;
          }
 
-         public int read_int32(int start_idx = 0) {
+         public int read_int32(int start_idx = 0)
+         {
             return BitConverter.ToInt32(buffer, start_idx);
          }
 
-         public byte[] read_bytes(int start_idx) {
+         public byte[] read_bytes(int start_idx)
+         {
             byte[] rv = new byte[buffer.Length - start_idx];
             Array.Copy(buffer, start_idx, rv, 0, rv.Length);
             return rv;
@@ -132,16 +148,19 @@ namespace bts {
          public bool is_reading_size() => read_state == ReadState.reading_size;
       }
 
-      public class ErrorMessage {
+      public class ErrorMessage
+      {
          public readonly SocketError? socket_error = null;
          public readonly Exception? exception;
          public readonly bool is_exception = false;
          public readonly string context = String.Empty;
-         public ErrorMessage(SocketError socket_error_, string context_) {
+         public ErrorMessage(SocketError socket_error_, string context_)
+         {
             socket_error = socket_error_;
             context = context_;
          }
-         public ErrorMessage(Exception exception_, string context_) {
+         public ErrorMessage(Exception exception_, string context_)
+         {
             is_exception = true;
             exception = exception_;
             context = context_;
@@ -149,7 +168,8 @@ namespace bts {
 
          public bool oom => exception != null && exception is OutOfMemoryException;
 
-         public override string ToString() {
+         public override string ToString()
+         {
             if (is_exception)
                return $"ErrorMessage exception: {exception} context: {context}";
             return $"socket_error: {socket_error} context: {context}";
@@ -175,13 +195,15 @@ namespace bts {
       private bool _hb_received_yet = false, _data_received_yet = false, _hb_sent_once = false;
       private DateTime _last_hb_received_time = new DateTime(1900, 1, 1);
 
-      public AsyncTCPConnection_APM(Socket sock, TimeSpan hb_period, ITCPConnection.ConnectionSide connection_side) {
+      public AsyncTCPConnection_APM(Socket sock, TimeSpan hb_period, ITCPConnection.ConnectionSide connection_side)
+      {
          _sock = sock;
          _hb_period = hb_period;
          _connection_side = connection_side;
          _sock.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.DontRoute, true);
          #region testing options
-         switch (connection_side) {
+         switch (connection_side)
+         {
             case ITCPConnection.ConnectionSide.Server:
                TCPSocketUtils.set_server_socket_options(ref _sock);
                break;
@@ -196,10 +218,13 @@ namespace bts {
          _detect_connection_status_change();
       }
 
-      private void _detect_connection_status_change() {
-         Action testing_func = () => {
+      private void _detect_connection_status_change()
+      {
+         Action testing_func = () =>
+         {
             bool connected = _sock.Connected;
-            while (true) {
+            while (true)
+            {
                if (connected != _sock.Connected)
                   Log.Gen.error($"Connection status changed this: {ToString()} {connected} -> {_sock.Connected}");
                connected = _sock.Connected;
@@ -209,26 +234,30 @@ namespace bts {
          ThreadPool.QueueUserWorkItem(_ => testing_func());
       }
 
-      public override string ToString() {
+      public override string ToString()
+      {
          return $"AsyncTCPConnection_APM side: {_connection_side} peer: {_peer_conn_info}, self: {_self_conn_info}";
       }
 
       #region ITCPConnection Members
 
-      public void sendWithConfirmation(byte[] data, ref string conf_id, SendConfirmationDelegate del) {
+      public void sendWithConfirmation(byte[] data, ref string conf_id, SendConfirmationDelegate del)
+      {
          conf_id = Guid.NewGuid().ToString();
          byte[] msg = _build_data_message(data);
          _begin_or_resume_sending(new AsyncSendMessage(msg, 0, conf_id, del));
       }
 
-      public void send(byte[] data) {
+      public void send(byte[] data)
+      {
          //Log.Gen.info($"AsyncTCPConnection_APM.send data: {AsyncTCPConnection_AsyncWrapper.to_hex(data, 0, data.Length)}");
          byte[] msg = _build_data_message(data);
          //Log.Gen.info($"AsyncTCPConnection_APM.send msg: {AsyncTCPConnection_AsyncWrapper.to_hex(msg, 0, msg.Length)}");
          _begin_or_resume_sending(new AsyncSendMessage(msg, 0));
       }
 
-      public void register(ITCPMessageHandler messageHandler) {
+      public void register(ITCPMessageHandler messageHandler)
+      {
          bts.Log.assert_d(_message_handler == null, "Message Handler is already assigned");
          //bts.Log.Gen.info($"AsyncTCPConnection_APM for peer = {_peer_conn_info} being registered");
          _message_handler = messageHandler;
@@ -237,15 +266,18 @@ namespace bts {
          _read_next_msg();
       }
 
-      public RemoteConnectionInfo get_peer_address() {
+      public RemoteConnectionInfo get_peer_address()
+      {
          return TCPSocketUtils.get_peer_address(_sock);
       }
 
-      public RemoteConnectionInfo get_self_address() {
+      public RemoteConnectionInfo get_self_address()
+      {
          return TCPSocketUtils.get_peer_address(_sock);
       }
 
-      public bool isDisposed {
+      public bool isDisposed
+      {
          get { return !_is_alive; }
       }
 
@@ -258,12 +290,14 @@ namespace bts {
          // TODO: check heartbeats
       }
 
-      private static int _tid() {
+      private static int _tid()
+      {
          var tid = System.Threading.Thread.CurrentThread.ManagedThreadId;
          return tid;
       }
 
-      private static string _cb_txt(IAsyncResult res) {
+      private static string _cb_txt(IAsyncResult res)
+      {
          return $"completed?: {res.IsCompleted} sync?: {res.CompletedSynchronously}";
       }
 
@@ -271,10 +305,13 @@ namespace bts {
 
       #region sending data
 
-      private void _begin_or_resume_sending(AsyncSendMessage message) {
+      private void _begin_or_resume_sending(AsyncSendMessage message)
+      {
          //bts.Log.Gen.info($"AsyncTCPConnection_APM._begin_or_resume_sending this: {ToString()} tid: {_tid()} message: {message} _is_alive: {_is_alive}");
-         if (_is_alive) {
-            try {
+         if (_is_alive)
+         {
+            try
+            {
                SocketError begin_send_err;
                // since we get message on on_send we dont need to keep the state around
                message.key = _send_key++;
@@ -282,16 +319,21 @@ namespace bts {
                                  out begin_send_err, new AsyncCallback(BeginSend_callback_function), message);
                if (begin_send_err != SocketError.Success)
                   _handle_error(new ErrorMessage(begin_send_err, "_begin_or_resume_sending"));
-            } catch (Exception ex) {
+            }
+            catch (Exception ex)
+            {
                _handle_error(new ErrorMessage(ex, "_begin_or_resume_sending"));
             }
          }
       }
 
-      private void BeginSend_callback_function(IAsyncResult res) {
-         if (res.CompletedSynchronously) {
+      private void BeginSend_callback_function(IAsyncResult res)
+      {
+         if (res.CompletedSynchronously)
+         {
             ThreadPool.QueueUserWorkItem(_ => _on_send_async(res));
-         } else
+         }
+         else
             _on_send_async(res);
       }
 
@@ -304,13 +346,16 @@ namespace bts {
          _send_key_map.Add(message.key);
          // bts.Log.Gen.info($"AsyncTCPConnection_APM._on_send: pre endSend this: {ToString()} tid: {_tid()} " +
          //                         $"res: {_cb_txt(res)} message: {res.AsyncState} _is_alive: {_is_alive}");
-         if (_is_alive) {
-            try {
+         if (_is_alive)
+         {
+            try
+            {
                SocketError send_err;
                int bytes_sent = _sock.EndSend(res, out send_err);
                //message.offset += bytesSent;
                message.advance_transfer_pointer(bytes_sent);
-               if (bytes_sent == 0) {
+               if (bytes_sent == 0)
+               {
                   bts.Log.Gen.error($"AsyncTCPConnection_APM._on_send this: {this} sent 0 bytes, throwing exception");
                   _handle_error(new ErrorMessage(new ApplicationException("Sent 0 bytes, peer socket is likely shutdown"),
                         "_on_send"));
@@ -318,23 +363,31 @@ namespace bts {
                }
                bts.Log.Gen.info($"AsyncTCPConnection_APM._on_send: post endSend this: {ToString()} tid: {_tid()} res: {_cb_txt(res)} " +
                $"message: {res.AsyncState} bytes_sent: {bytes_sent} _is_alive: {_is_alive} send_err: {send_err}");
-               if (send_err == SocketError.Success) {
+               if (send_err == SocketError.Success)
+               {
                   // successful send of some bytes
                   //if (message.offset == message.buffer.Length) {
-                  if (message.fully_transfered()) {
+                  if (message.fully_transfered())
+                  {
                      // fully sent, if confirmation needs to be sent out                     
-                     if (!String.IsNullOrEmpty(message.message_id)) {
+                     if (!string.IsNullOrEmpty(message.message_id)) {
                         ArgumentNullException.ThrowIfNull(message.on_send_del);
                         message.on_send_del(message.message_id, true);
                      }
-                  } else {
+                  }
+                  else
+                  {
                      // there is more to send
                      _begin_or_resume_sending(message);
                   }
-               } else {
+               }
+               else
+               {
                   _handle_error(new ErrorMessage(send_err, "_on_send"));
                }
-            } catch (Exception ex) {
+            }
+            catch (Exception ex)
+            {
                _handle_error(new ErrorMessage(ex, "_on_send"));
             }
          }
@@ -345,8 +398,10 @@ namespace bts {
       /// future references
       /// </summary>
       /// <returns></returns>
-      private byte[] _build_hb_message() {
-         if (_cached_hb_message == null) {
+      private byte[] _build_hb_message()
+      {
+         if (_cached_hb_message == null)
+         {
             BinaryWriter bw = new BinaryWriter(new MemoryStream());
             // the total msg =[msg_sz, msg_type] 
             int msg_sz = 4;
@@ -358,7 +413,8 @@ namespace bts {
       }
 
       // writes size prefixed msg
-      private byte[] _build_data_message(byte[] data) {
+      private byte[] _build_data_message(byte[] data)
+      {
          // TODO, avoid unecessary memory creation (write directly into byte[] using BitConverter)
          MemoryStream str = new MemoryStream();
          BinaryWriter bw = new BinaryWriter(str);
@@ -377,28 +433,34 @@ namespace bts {
 
       #region receiving data
 
-      private void _read_next_msg() {
+      private void _read_next_msg()
+      {
          _read_size();
       }
 
-      private void _read_size() {
+      private void _read_size()
+      {
          _begin_or_resume_reading(new AsyncReceiveMessage(4, AsyncReceiveMessage.ReadState.reading_size));
       }
 
-      private void _read_data(int data_size) {
+      private void _read_data(int data_size)
+      {
          // TODO, this could be a problem is the size is too big - maybe read it in chunks (later)
          _begin_or_resume_reading(new AsyncReceiveMessage(data_size, AsyncReceiveMessage.ReadState.reading_data));
       }
 
-      private static string IAsyncResult_txt(IAsyncResult res) {
+      private static string IAsyncResult_txt(IAsyncResult res)
+      {
          return $"IsCompleted: {res.IsCompleted} CompletedSynchronously: {res.CompletedSynchronously}";
       }
 
-      private void _begin_or_resume_reading(AsyncReceiveMessage rcv_msg) {
+      private void _begin_or_resume_reading(AsyncReceiveMessage rcv_msg)
+      {
          //bts.Log.Gen.info($"AsyncTCPConnection_APM._begin_or_resume_reading: this: {ToString()} tid: {_tid()} message: {rcv_msg} _is_alive: {_is_alive}");
          if (!_is_alive)
             return;
-         try {
+         try
+         {
             SocketError read_sock_err;
             rcv_msg.key = _receive_key++;
             var res = _sock.BeginReceive(rcv_msg.buffer, rcv_msg.transfer_ptr,
@@ -406,28 +468,35 @@ namespace bts {
                                  SocketFlags.None, out read_sock_err,
                                  new AsyncCallback(BeginReceive_callback_function),
                                  rcv_msg);
-            if (read_sock_err != SocketError.Success) {
+            if (read_sock_err != SocketError.Success)
+            {
                bts.Log.Gen.error($"_begin_or_resume_reading this: {this} error: {read_sock_err} rcv_msg: {rcv_msg}");
                _handle_error(new ErrorMessage(read_sock_err, "_beging_or_resume_reading"));
             }
-         } catch (Exception ex) {
+         }
+         catch (Exception ex)
+         {
             _handle_error(new ErrorMessage(ex, "_beging_or_resume_reading"));
          }
       }
 
-      private void _log_incoming_msg(IAsyncResult res) {
-         // var rcv_message = ExceptionAssert.Cast(res.AsyncState).To<AsyncReceiveMessage>();
+      private void _log_incoming_msg(IAsyncResult res)
+      {
+         // AsyncReceiveMessage rcv_message = res.AsyncState as AsyncReceiveMessage;
          // var buff_txt = AsyncTCPConnection_AsyncWrapper.to_hex(rcv_message.buffer, 0, rcv_message.buffer.Length);
          // Log.Gen.info($"BeginReceive_callback_function _log_incoming_msg buff: {buff_txt}");
 
       }
 
       // this is the function passed to BeginReceive, we 
-      private void BeginReceive_callback_function(IAsyncResult res) {
+      private void BeginReceive_callback_function(IAsyncResult res)
+      {
          _log_incoming_msg(res);
-         if (res.CompletedSynchronously) {
+         if (res.CompletedSynchronously)
+         {
             ThreadPool.QueueUserWorkItem(_ => _on_read_async(res));
-         } else
+         }
+         else
             _on_read_async(res);
       }
 
@@ -443,13 +512,15 @@ namespace bts {
          _read_key_map.Add(rcv_message.key);
          if (!_is_alive)
             return;
-         try {
+         try
+         {
             SocketError read_err;
             Log.Gen.info($"AsyncTCPConnection_APM._on_read pre-EndReceive rcv_message: {rcv_message} res: {IAsyncResult_txt(res)}");
             int bytes_read = _sock.EndReceive(res, out read_err);
             rcv_message.advance_transfer_pointer(bytes_read);
             Log.Gen.info($"AsyncTCPConnection_APM._on_read post-EndReceive rcv_message: {rcv_message} res: {IAsyncResult_txt(res)} bytes_read: {bytes_read}");
-            if (bytes_read == 0) {
+            if (bytes_read == 0)
+            {
                bts.Log.Gen.error($"AsyncTCPConnection_APM._on_read this: {this} read 0 bytes, throwing exception");
                //bts.Log.Gen.error($"AsyncTCPConnection_APM._on_read this: {this} read 0 bytes, sleeping and ignoring");
                //System.Threading.Thread.Sleep(50);
@@ -457,23 +528,32 @@ namespace bts {
                return;
             }
             //return;
-            if (read_err == SocketError.Success) {
-               if (rcv_message.fully_transfered()) {
-                  if (rcv_message.is_reading_size()) {
+            if (read_err == SocketError.Success)
+            {
+               if (rcv_message.fully_transfered())
+               {
+                  if (rcv_message.is_reading_size())
+                  {
                      var msg_size = _on_read_size_completely(rcv_message);
-                     if (msg_size != null && msg_size.HasValue) {
+                     if (msg_size != null && msg_size.HasValue)
+                     {
                         _read_data(msg_size.Value);
                      }
                      return;
-                  } else {
+                  }
+                  else
+                  {
                      // data is read
                      int msg_type = rcv_message.read_int32();
                      bts.Log.assert(msg_type == HB_MSG_SIGNAL || msg_type == DATA_MSG_SIGNAL);
-                     if (msg_type == DATA_MSG_SIGNAL) {
+                     if (msg_type == DATA_MSG_SIGNAL)
+                     {
                         _on_read_data_completely(rcv_message);
                         _read_next_msg();
                         return;
-                     } else if (msg_type == HB_MSG_SIGNAL) {
+                     }
+                     else if (msg_type == HB_MSG_SIGNAL)
+                     {
                         _on_read_hb_completely(rcv_message);
                         _read_next_msg();
                         return;
@@ -481,21 +561,28 @@ namespace bts {
                      // go back to reading size for a new message
                      //_read_size();
                   }
-               } else {
+               }
+               else
+               {
                   // msg has not been fully rd, we have already advanced the transfer ptr
                   _begin_or_resume_reading(rcv_msg: rcv_message);
                }
-            } else {
+            }
+            else
+            {
                _handle_error(new ErrorMessage(read_err, "_on_read_async"));
             }
-         } catch (Exception ex) {
+         }
+         catch (Exception ex)
+         {
             _handle_error(new ErrorMessage(ex, "_on_read_async"));
          }
       }
 
       #endregion
 
-      private int? _on_read_size_completely(AsyncReceiveMessage msg) {
+      private int? _on_read_size_completely(AsyncReceiveMessage msg)
+      {
          bts.Log.assert(msg.is_reading_size());
          bts.Log.assert(msg.fully_transfered());
          bts.Log.assert(msg.original_size == 4);
@@ -503,23 +590,28 @@ namespace bts {
          //var buff_txt = AsyncTCPConnection_AsyncWrapper.to_hex(msg.buffer, 0, msg.buffer.Length);
          //bts.Log.Gen.info($"AsyncTCPConnection_APM._on_read_size_completely buff: {buff_txt}");
          bts.Log.Gen.info($"AsyncTCPConnection_APM._on_read_size_completely msg: {msg} msg_sz: {msg_sz}");
-         if (msg_sz < 0 || msg_sz > _max_msg_size) {
+         if (msg_sz < 0 || msg_sz > _max_msg_size)
+         {
             string err_msg = $"Encountered malformed message, prefixed_size: {msg_sz}";
             _handle_error(new ErrorMessage(new ApplicationException(err_msg), "_on_read_size_completely"));
             return null;
-         } else {
+         }
+         else
+         {
             // sz is determined, now we read that many bytes
             //_read_data(msg_sz);
             return msg_sz;
          }
       }
 
-      private void _on_read_hb_completely(AsyncReceiveMessage msg) {
+      private void _on_read_hb_completely(AsyncReceiveMessage msg)
+      {
          bts.Log.assert(!msg.is_reading_size());
          bts.Log.assert(msg.fully_transfered());
          bts.Log.assert(msg.original_size == 4);
          bts.Log.assert(msg.read_int32(0) == HB_MSG_SIGNAL);
-         if (!_hb_received_yet) {
+         if (!_hb_received_yet)
+         {
             _hb_received_yet = true;
             //bts.Log.Gen.info($"AsyncTCPConnection_APM HB received for the first time from peer {_peer_conn_info}");
          }
@@ -528,7 +620,8 @@ namespace bts {
          _message_handler.onData(new RemoteMessage(null, RemoteMessageType.Heartbeat, _peer_conn_info));
       }
 
-      private void _on_read_data_completely(AsyncReceiveMessage msg) {
+      private void _on_read_data_completely(AsyncReceiveMessage msg)
+      {
          bts.Log.assert(!msg.is_reading_size());
          bts.Log.assert(msg.fully_transfered());
          var type = msg.read_int32(0);
@@ -537,7 +630,8 @@ namespace bts {
          byte[] data = msg.read_bytes(4);
          ///callbacks can be made through thread pool to avoid this
          ///thread from being blocked, by this callback
-         if (!_data_received_yet) {
+         if (!_data_received_yet)
+         {
             _data_received_yet = true;
             //bts.Log.Gen.info($"AsyncTCPConnection_APM data received for the first time from peer {_peer_conn_info}");
          }
@@ -548,15 +642,19 @@ namespace bts {
 
       #region IDisposable Members
 
-      public void Dispose() {
+      public void Dispose()
+      {
          // socket is closed and disposed here
-         lock (this) {
+         lock (this)
+         {
             bts.Log.Gen.info($"AsyncTCPConnection_APM.Dispose this: {this} Dispose called _dipose_impetus: {_dipose_impetus}");
-            if (_is_alive) {
+            if (_is_alive)
+            {
                _master_worker.removeWorker(this);
                _is_alive = false;
                bts.Log.Gen.info($"AsyncTCPConnection_APM.Dispose this: {this} being disposed, last HB was received at {_last_hb_received_time}");
-               try {
+               try
+               {
                   bts.Log.Gen.info($"AsyncTCPConnection_APM.Dispose this: {this} shutting down socket now");
                   _sock.Shutdown(SocketShutdown.Both);
                } catch (OutOfMemoryException) {
@@ -574,8 +672,10 @@ namespace bts {
       #endregion
 
       private string _dipose_impetus = String.Empty;
-      private void _handle_error(ErrorMessage err_msg) {
-         lock (this) {
+      private void _handle_error(ErrorMessage err_msg)
+      {
+         lock (this)
+         {
             Log.Gen.error($"AsyncTCPConnection_APM._handle_error this: {this} error_msg: {err_msg}");
             if (err_msg.oom) {
                throw Log.notnull(err_msg.exception);

src/DataStreamSuite/PubSub/Server/Communication/ManagedTCP/ManagedTCPMessageHub.cs

@@ -286,4 +286,4 @@ namespace BCG {
 
 
 
-}
+}

src/DataStreamSuite/PubSubServer/Dockerfile

@@ -1,5 +1,8 @@
 FROM bluetradingsystems/linux_dotnet7
 
-COPY --chown=bts:bts "build/Release/bin/net7.0/linux-x64/publish" "bin"
-COPY --chown=bts:bts "src/DataStreamSuite/PubSubServer/spark.conf" ".local/share/bts/PubSubServer/conf/spark.conf"
+ARG SOURCE_DIR="build/Release/bin/net7.0/linux-x64/publish"
+ARG CONF_FILE="src/DataStreamSuite/PubSubServer/spark.conf"
+
+COPY --chown=bts:bts "${SOURCE_DIR}" "bin"
+COPY --chown=bts:bts "${CONF_FILE}" ".local/share/bts/PubSubServer/conf/spark.conf"
 WORKDIR "/home/bts/bin"

src/DataStreamSuite/temp3.sh

@@ -1,33 +0,0 @@
-#!/bin/bash
-
-# Define the bash array
-# values=("value1" "value2" "value3")
-values=$(dotnet sln DataStreamSuite.sln list | tail +3)
-
-# Function to get the key for a value
-get_key() {
-  local value=$1
-  # Replace this with your actual logic to get the key
-  echo "key_for_$value"
-}
-
-# Function to get the JSON array of child values for a value
-get_children() {
-  local value=$1
-  # Replace this with your actual logic to get the JSON array of child values
-  echo '["child1", "child2", "child3"]'
-}
-
-# Create an empty JSON object
-json_object="{}"
-
-# Iterate through the array and construct the JSON object
-for value in ${values[@]}; do
-  key=$(get_key "$value")
-  children=$(get_children "$value")
-  json_object=$(echo "$json_object" | jq --arg key "$key" --argjson children "$children" '. + {($key): $children}')
-done
-
-# Print the final JSON object
-echo "$json_object"
-

src/Spark.Admin.WebApp/src/Accounts/Types.fs

@@ -17,7 +17,7 @@ type Tab =
   | PositionManagementTab
   | MarginTab
 
-let venues = [| Venue.BINANCE; Venue.BinanceUs; Venue.BinanceUsdsFutures; Venue.COINBASE; Venue.DERIBIT; Venue.GLBX; Venue.ICE; Venue.SIM; |]
+let venues = [| Venue.BINANCE; Venue.BinanceUs; Venue.BinanceUsdsFutures; Venue.CBOE; Venue.COINBASE; Venue.DERIBIT; Venue.GLBX; Venue.ICE; Venue.TAIFEX; Venue.SIM; |]
 
 let securityTypes = [| SecurityType.SPOT; SecurityType.FUTURE; SecurityType.OPTION; SecurityType.SPREAD; |]
 
@@ -69,6 +69,11 @@ type ICEExchangeField =
   | AccountCode of string
   | MiscTagMapping of JSMap<string>
 
+type TaifexExchangeField =
+  | CTI of CTICode
+  | Origin of CustomerOrFirm
+  | ClearingAccount of string
+
 type SIMExchangeField =
   | CTI of CTICode
   | Origin of CustomerOrFirm

src/Spark.Admin.WebApp/src/Accounts/View.fs

@@ -797,9 +797,9 @@ let venueToString (venue:bts.api.Venue) =
     | bts.api.Venue.BINANCE -> "BINANCE"
     | bts.api.Venue.BinanceUs -> "BinanceUs"
     | bts.api.Venue.BinanceUsdsFutures -> "BinanceUsdsFutures"
-    | bts.api.Venue.COINBASE -> "COINBASE"
     | bts.api.Venue.FTX -> "FTX"
     | bts.api.Venue.OKX -> "OKX"
+    | bts.api.Venue.TAIFEX -> "TAIFEX"
 
 let exchange_display expirationNameList =
   let s = sprintf "%O" expirationNameList.product.exchange

src/app_sparkd/CMakeLists.txt

@@ -1,8 +1,10 @@
-cmake_minimum_required(VERSION 3.5)
+cmake_minimum_required(VERSION 3.14)
 project(sparkd)
 
+include(FetchContent)
 include(${CMAKE_SOURCE_DIR}/cmake/safe_dep_glob.cmake)
 include(${CMAKE_SOURCE_DIR}/cmake/ProtobufGenerateCpp.cmake)
+include(${CMAKE_SOURCE_DIR}/cmake/InstallBasic.cmake)
 include(GNUInstallDirs)
 
 if (NOT DEFINED PROJECT_VERSION)
@@ -57,6 +59,14 @@ target_include_directories(
 )
 add_dependencies(${PROJECT_NAME} rapidjson_project)
 
+# databento
+FetchContent_Declare(
+  databento
+  GIT_REPOSITORY https://github.com/databento/databento-cpp
+  GIT_TAG HEAD
+)
+FetchContent_MakeAvailable(databento)
+
 # deribit-demo log level
 if(NOT DEFINED DERIBIT_DEMO_LOG_LEVEL)
   add_definitions(-DSPDLOG_ACTIVE_LEVEL=0)
@@ -130,9 +140,8 @@ target_link_libraries(
   python_md
   python_md_module
   spark
+  databento::databento
   ${emumba-deribit-connector-library_LIBRARIES}
-  ssl
-  cryptopp
   tbb
   onixs-cmemdh-all-4.8.0
   OnixS.IceImpactMulticastPriceFeedHandlerCpp
@@ -148,8 +157,11 @@ target_link_libraries(
   -static-libstdc++
   -pthread)
 
-include(${CMAKE_SOURCE_DIR}/cmake/InstallBasicConfig.cmake)
-install_basic_config()
+# Catch2
+find_package(Catch2 CONFIG REQUIRED)
+target_link_libraries(${PROJECT_NAME} Catch2::Catch2)
+
+install_basic()
 install(
   FILES
     "${EXTERNALS_LIB}/libOnixS.IceImpactMulticastPriceFeedHandlerCpp.so.3.16.0.0"

src/app_sparkd/Dockerfile

@@ -0,0 +1,19 @@
+FROM bluetradingsystems/linux_dotnet7
+
+ARG SOURCE_DIR="install/sparkd-release-gcc"
+ARG CONF_FILE="src/app_sparkd/spark.current.conf"
+
+USER "root"
+RUN apt-get update \
+  && apt-get install -y liblttng-ust-dev
+
+RUN apt-get update \
+  && apt-get install -y libpcap0.8-dev
+
+RUN apt-get update \
+  && apt-get install -y python3.10-dev
+
+USER "bts"
+COPY --chown=bts:bts "${SOURCE_DIR}" "."
+COPY --chown=bts:bts "${CONF_FILE}" ".local/share/bts/sparkd/conf/spark.conf"
+WORKDIR "/home/bts/bin"

src/app_sparkd/sparkd/Dockerfile

@@ -1,10 +0,0 @@
-FROM bluetradingsystems/linux_dotnet7
-
-USER "root"
-RUN apt-get update \
-  && apt-get install -y liblttng-ust-dev
-
-USER "bts"
-COPY --chown=bts:bts "install/Release/bin" "bin"
-COPY --chown=bts:bts "src/server/sparkd/spark.current.conf" ".local/share/bts/sparkd/conf/spark.conf"
-WORKDIR "/home/bts/bin"

src/app_sparkd/sparkd/algo/AlgoDefinitionServiceServer.cpp

@@ -1,4 +1,5 @@
 #include "AlgoDefinitionServiceServer.h"
+#include "spark/spark_data_directory.h"
 
 namespace bts {
 namespace api {
@@ -19,8 +20,11 @@ struct AlgoDefinitionServiceServer::PImpl {
        const AlgoDefinitionServiceBase::Algo_definitionsStatefulPublisher
            &publisher)
        : table_name("algo_definitions"),
-         store(new SqliteStore("algo_definition")), context(context),
-         publisher(publisher) {}
+         store(std::make_unique<SqliteStore>(
+             spark::SparkDataDirectory{}
+                 .database_file("algo_definitions")
+                 .string())),
+         context(context), publisher(publisher) {}
 };
 
 AlgoDefinitionServiceServer::AlgoDefinitionServiceServer(

src/app_sparkd/sparkd/algo/AlgoDefinitionServiceServer.h

@@ -4,19 +4,19 @@
 #include "spark/api/services/AlgoDefinitionService.server.pb.h"
 #include "spark/api/services/OrderService.client.pb.h"
 #include "spark/api/services/OrderService.pb.h"
-#include "spark/logging/logging.h"
 #include "spark/persistence/IStore.h"
 #include "spark/persistence/SqliteStore.h"
 #include "spark/utils/PBUtils.h"
 #include "spark/utils/TimeStamp.h"
 #include "spark/utils/bts_assert.h"
 #include "spark/utils/bts_no_copy.h"
-#include <memory>
+#include "system/logging/logging.h"
 
 #include <boost/tokenizer.hpp>
 #include <fmt/format.h>
 #include <google/protobuf/util/time_util.h>
 #include <map>
+#include <memory>
 #include <set>
 #include <string>
 #include <utility>
@@ -69,4 +69,4 @@ class AlgoDefinitionServiceServer : public AlgoDefinitionServiceBase {
 } // namespace algo
 } // namespace orderman
 } // namespace api
-} // namespace bts
+} // namespace bts

src/app_sparkd/sparkd/analytics/AnalyticsServiceServer.cpp

@@ -2,13 +2,13 @@
 #include "spark/api/apicore.pb.h"
 #include "spark/api/services/AnalyticsService.pb.h"
 #include "spark/api/services/AnalyticsService.server.pb.h"
-#include "spark/logging/logging.h"
 #include "spark/persistence/IStore.h"
 #include "spark/persistence/SqliteStore.h"
 #include "spark/utils/PBUtils.h"
 #include "spark/utils/TimeStamp.h"
 #include "spark/utils/bts_assert.h"
 #include "sparkd/algo/AlgoDefinitionServiceServer.h"
+#include "system/logging/logging.h"
 
 #include <boost/algorithm/string.hpp>
 #include <boost/tokenizer.hpp>
@@ -50,7 +50,7 @@ void add_or_update_to_table(
     const std::string &ns,
     const std::string &key,
     const std::string &val) {
-   auto [exists, existing_val] = store->retrieve_one(ns, key);
+   auto [exists, rowid, existing_val] = store->retrieve_one(ns, key);
    LOG(Gen, Info,
        "add_or_update ns: {} key: {} val: {} exists: {} existing_val: {}", ns,
        key, val, exists, existing_val);
@@ -384,7 +384,8 @@ void AnalyticsServiceServer::save_report(
     const api::Response<::bts::api::ResultMessage> &resp) {
    Report report = report_arg;
    std::string key = std::to_string(report.id());
-   auto [present, val] = me->reports_store->retrieve_one(me->reports_ns, key);
+   auto [present, rowid, val] =
+       me->reports_store->retrieve_one(me->reports_ns, key);
    if (!present) {
       int new_id = get_max_id(me->reports_store.get(), me->reports_ns) + 1;
       report.set_id(new_id);
@@ -406,7 +407,8 @@ void AnalyticsServiceServer::delete_report(
     const bts::api::analytics::IntWrapper &id,
     const api::Response<::bts::api::ResultMessage> &resp) {
    std::string key = std::to_string(id.data());
-   auto [present, val] = me->reports_store->retrieve_one(me->reports_ns, key);
+   auto [present, rowid, val] =
+       me->reports_store->retrieve_one(me->reports_ns, key);
    api::ResultMessage res;
    res.set_success(false);
    if (!present) {

src/app_sparkd/sparkd/cmds/AdminReport.cpp

@@ -2,12 +2,12 @@
 
 #include "spark/api/public/ServiceContext.h"
 #include "spark/config/config.h"
-#include "spark/config/protos/logging.pb.h"
-#include "spark/logging/logging.h"
 #include "spark/quickbus/quickbus.h"
 
 #include "spark/api/service_helpers/AdminServiceUtils.h"
 #include "spark/api/services/AdminService.client.pb.h"
+#include "spark/utils/init_spark_lib.h"
+#include "system/protobufs/logging.pb.h"
 
 #include <algorithm>
 #include <boost/algorithm/string.hpp>
@@ -17,9 +17,9 @@
 namespace bts::cmds {
 
 void md_report() {
-   auto cfg = config::get<config::Logging>();
+   auto cfg = config::get<logging::protobufs::Logging>();
    cfg.mutable_console()->set_default_level("Fatal");
-   log::configure_logging(cfg);
+   utils::SparkLibInit spark_lib{cfg, false, true};
 
    std::map<api::CMEMics, std::string> mic_map;
    mic_map[api::CMEMics::CME] = "XCME";

src/app_sparkd/sparkd/cmds/CMECert.cpp

@@ -1,13 +1,14 @@
 #include "CMECert.h"
-#include "spark/logging/logging.h"
 #include "spark/orderman/ilink3/IL3Dispatcher.h"
 #include "spark/orderman/ilink3/IL3GW.h"
 #include "spark/orderman/ilink3/IL3Reactor.h"
 #include "spark/orderman/ilink3/ILink3Utils.h"
-#include "spark/os/os.h"
 #include "spark/utils/Dispatcher.h"
 #include "spark/utils/SecureLogon.h"
 #include "spark/utils/TimeStamp.h"
+#include "system/logging/logging.h"
+#include "system/os.h"
+
 #include <chrono>
 #include <fstream>
 #include <future>

src/app_sparkd/sparkd/cmds/KeyCmd.cpp

@@ -2,8 +2,6 @@
 
 #include "spark/config/config.h"
 #include "spark/config/protos/keycode.pb.h"
-#include "spark/logging/logging.h"
-#include "spark/os/os.h"
 #include "spark/utils/AuthToken.h"
 #include "spark/utils/FileUtils.h"
 #include "spark/utils/bts_assert.h"
@@ -13,6 +11,8 @@
 #include "spark/utils/crypto/PublicKey.h"
 #include "spark/utils/datetime.h"
 #include "spark/utils/helper_macros.h"
+#include "system/logging/logging.h"
+#include "system/os.h"
 
 #include <boost/algorithm/string/replace.hpp>
 #include <boost/date_time/gregorian/gregorian.hpp>

src/app_sparkd/sparkd/cmds/QuickTest.cpp

@@ -1,5 +1,5 @@
 #include "QuickTest.h"
-#include "spark/logging/logging.h"
+#include "system/logging/logging.h"
 
 #include <thread>
 

src/app_sparkd/sparkd/cmds/Run.cpp

@@ -1,31 +1,32 @@
 #include "Run.h"
+#include "spark/api/md/MarketDataConfig.pb.h"
 #include "spark/api/public/ServiceContext.h"
 #include "spark/config/config.h"
 #include "spark/config/protos/ports.pb.h"
 #include "spark/config/protos/quickbus_cfg.pb.h"
-#include "spark/logging/logging.h"
 #include "spark/md/MBOProcessor.h"
 #include "spark/md/MBOServiceServer.h"
 #include "spark/md/MDClient.h"
 #include "spark/md/SyntheticSecurityService.h"
-#include "spark/md/relay/QueryServer.h"
+#include "spark/md/relay/query_server.h"
 #include "spark/md/service/MarketDataServiceServer.h"
 #include "spark/metrics/metrics.h"
 #include "spark/metrics/system_metrics.h"
 #include "spark/orderman/OrderServiceServer.h"
 #include "spark/orderman/events/TradeSummarySignalDispatcher.h"
 #include "spark/orderman/order_factory/DynamicOrderRegistry.h"
+#include "spark/persistence/IDGenerator.h"
 #include "spark/persistence/PersistenceServiceServer.h"
 #include "spark/plugin/SparkServices.h"
 #include "spark/plugin/plugin.h"
 #include "spark/quickbus/core/Quickbus.h"
 #include "spark/tracing/PerfLog.h"
 #include "spark/utils/Dispatcher.h"
-#include "md/cmds/Run.h"
 #include "sparkd/algo/AlgoDefinitionServiceServer.h"
 #include "sparkd/analytics/AnalyticsServiceServer.h"
 #include "sparkd/http/WebServer.h"
 #include "sparkd/md/MarketDataAuxiliaryServiceServer.h"
+#include "sparkd/md/MarketDataGateway.h"
 #include "sparkd/md/recording/MDRecorder.h"
 #include "sparkd/position/MarginServiceServer.h"
 #include "sparkd/position/PositionServiceServer.h"
@@ -36,7 +37,7 @@
 #include "sparkd/support/ShellLatencyServer.h"
 #include "sparkd/user/AdminServiceServer.h"
 #include "sparkd/user/AuthServiceServer.h"
-
+#include "system/logging/logging.h"
 
 namespace bts::cmds {
 
@@ -49,8 +50,7 @@ static void execute() {
    auto qb_cfg = config::get<config::QuickbusCfg>().node();
    auto &ports = config::get<config::Ports>();
    qb_cfg.set_my_port(ports.qbus_port());
-   qbus::Quickbus quickbus(
-       qb_cfg, qbus::QuickbusThreadOptions::DEFERRED_START);
+   qbus::Quickbus quickbus(qb_cfg, qbus::QuickbusThreadOptions::DEFERRED_START);
    api::ServiceContext sc(&quickbus);
 
    trace::PerfLog perf_log;
@@ -63,7 +63,10 @@ static void execute() {
 
    bts::md::MBOProcessor mbo_processor;
    bts::md::TradeSummarySignalDispatcher tss_dispatcher;
-   md::cmds::Run::run();
+   bts::md::MarketDataGateway md_gateway{
+       config::get<api::md::MarketDataConfig>(),
+       persistence::IDGenerator::instance_ptr()};
+   md_gateway.start();
 
    api::synthetic::SyntheticSecurityService synthetic_service(sc);
    bts::md::MDClient mdclient(

src/app_sparkd/sparkd/cmds/TestPythonMdProviderCmd.cpp

@@ -1,7 +1,7 @@
 #include "TestPythonMdProviderCmd.h"
 #include "python_md/md_sink_impl.h"
 #include "python_md/py_md_executor.h"
-#include "spark/logging/logging.h"
+#include "system/logging/logging.h"
 
 #include <pybind11/embed.h>
 #include <pybind11/pybind11.h>
@@ -38,10 +38,14 @@ static void execute(
       py_md_exec.start();
       LOG(md, Info, "Stopping py md executor.");
       py_md_exec.stop();
-      int i = 0;
-      while (true) {
-         std::cout << "Sleep loop count " << ++i << std::endl;
-         std::this_thread::sleep_for(3s);
+      // The following brief sleep loop provides a means to test that 'ctrl+c' interrupts exit
+      // quickly (there was a bug during early development of the 'python market data' feature that
+      // caused such interrupts to hang for a few seconds).
+      std::cout << "Press ctrl+c to exit..." << std::endl;
+      static const int SLEEP_CYCLES = 3;
+      static const auto SLEEP_INTERVAL = 3s;
+      for (int i = 0; i < SLEEP_CYCLES; i++) {
+         std::this_thread::sleep_for(SLEEP_INTERVAL);
       }
    } catch (const std::exception &ex) {
       LOG(md, Error, "Py md executor threw: {}", string{ex.what()});
@@ -58,7 +62,7 @@ TestPythonMdProviderCmd::TestPythonMdProviderCmd()
        "--module-path", args_->module_paths, "Module path for python plugin");
    callback([this]() {
       if (args_->py_md_provider.empty()) {
-         args_->py_md_provider = "md_provider_impl.MdProviderImpl";
+         args_->py_md_provider = "dummy_md_provider.DummyMdProvider";
       }
       if (args_->module_paths.empty()) {
          args_->module_paths.push_back(".");

src/app_sparkd/sparkd/cmds/Web.cpp

@@ -1,30 +1,22 @@
 #include "Web.h"
+#include "md/cmds/Run.h"
 #include "spark/api/public/ServiceContext.h"
-#include "spark/api/services/MarketDataService.pb.h"
-#include "spark/api/services/SyntheticSecurityService.pb.h"
 #include "spark/config/config.h"
-#include "spark/config/protos/market_data.pb.h"
 #include "spark/config/protos/ports.pb.h"
 #include "spark/config/protos/quickbus_cfg.pb.h"
-#include "spark/md/MBOProcessor.h"
 #include "spark/md/MDClient.h"
 #include "spark/md/SyntheticSecurityService.h"
-#include "spark/md/relay/QueryServer.h"
+#include "spark/md/relay/query_server.h"
 #include "spark/metrics/metrics.h"
 #include "spark/metrics/system_metrics.h"
-#include "spark/quickbus/quickbus.h"
+#include "spark/quickbus/core/Quickbus.h"
 #include "spark/tracing/PerfLog.h"
 #include "spark/utils/Dispatcher.h"
-#include "spark/utils/thread_manager.h"
-#include "md/cmds/Cleanup.h"
-#include "md/cmds/Run.h"
-#include "sparkd/exchange_sim/ExchangeSim.h"
 #include "sparkd/http/WebServer.h"
 #include "sparkd/user/AdminServiceServer.h"
 #include "sparkd/user/AuthServiceServer.h"
-#include <optional>
+
 #include <string>
-#include <vector>
 
 namespace bts::cmds {
 

src/app_sparkd/sparkd/cmds/crypto/DeribitDemo.cpp

@@ -1,6 +1,6 @@
 #include "DeribitDemo.h"
-#include "spark/logging/logging.h"
 #include "sparkd/cmds/crypto/DeribitDemoListener.h"
+#include "system/logging/logging.h"
 
 #include <csignal>
 #include <emumba-deribit-connector/deribit_exception.h>

src/app_sparkd/sparkd/cmds/crypto/DeribitDemoCmd.cpp

@@ -1,7 +1,8 @@
 #include "DeribitDemoCmd.h"
 #include "DeribitDemo.h"
 #include "spark/config/config.h"
-#include "spark/os/os.h"
+#include "system/os.h"
+
 #include <CLI11/CLI11.hpp>
 #include <functional>
 #include <optional>

src/app_sparkd/sparkd/cmds/crypto/DeribitDemoConfiguration.cpp

@@ -1,7 +1,8 @@
 #include "DeribitDemoConfiguration.h"
 #include "RapidJsonReader.h"
-#include "spark/logging/logging.h"
 #include "rapidjson/istreamwrapper.h"
+#include "system/logging/logging.h"
+
 #include <emumba-deribit-connector/deribit_utils.h>
 #include <fstream>
 #include <list>

src/app_sparkd/sparkd/cmds/crypto/LiveLadderDemo.cpp

@@ -1,5 +1,7 @@
 #include "LiveLadderDemo.h"
 #include "spark/api/md/MarketDataConfig.pb.h"
+#include "spark/config/config.h"
+#include "spark/persistence/IDGenerator.h"
 #include "sparkd/md/MarketDataGateway.h"
 #include "sparkd/md/listeners/LiveLadderPrinter.h"
 
@@ -7,13 +9,16 @@
 
 namespace bts::cmds {
 
-LiveLadderDemo::LiveLadderDemo(const api::md::MarketDataConfig &) {
+using bts::api::md::MarketDataConfig;
+using bts::persistence::IDGenerator;
+
+LiveLadderDemo::LiveLadderDemo(const MarketDataConfig &config) {
    char ssl_cert_file[] = "SSL_CERT_FILE=/etc/ssl/cert/ca-certificates.crt";
    putenv(ssl_cert_file);
 
-   md::MarketDataGateway gw;
+   md::MarketDataGateway gw{config, IDGenerator::instance_ptr()};
    md::listeners::LiveLadderPrinter ladder_printer;
-   gw.run();
+   gw.start();
 
    const std::size_t n_loops = 16;
    for (std::size_t n = 0; n < n_loops; n++) {

src/app_sparkd/sparkd/cmds/crypto/LiveLadderDemoCmd.cpp

@@ -3,9 +3,10 @@
 #include "LiveLadderDemo.h"
 #include "spark/api/md/MarketDataConfig.pb.h"
 #include "spark/io/ProtoReader.h"
-#include "spark/logging/logging.h"
 #include "sparkd/cmds/common/Validators.h"
 #include "sparkd/cmds/crypto/LiveLadderDemo.h"
+#include "system/logging/logging.h"
+
 #include <CLI11/CLI11.hpp>
 #include <optional>
 

src/app_sparkd/sparkd/cmds/md/cmds/Channels.cpp

@@ -1,11 +1,30 @@
 #include "Channels.h"
+#include "spark/api/md/MarketDataConfig.pb.h"
+#include "spark/config/config.h"
 #include "spark/md/cme/MDP3Utils.h"
+#include "spark/md/cme/mdp3_data_directory.h"
 
 namespace bts::cmds::md::cmds {
 
+using bts::md::await_cme_config;
+using bts::md::get_cme_channels;
+using bts::md::Mdp3DataDirectory;
+
 Channels::Channels() : CLI::App("List configured cme channels", "channels") {
    callback([] {
-      auto channels = bts::md::get_configured_cme_channels();
+      Mdp3DataDirectory data_dir;
+      auto md_cfg = bts::config::get<api::md::MarketDataConfig>();
+      if (md_cfg.glbx().empty()) {
+         std::cerr << "'MarketDataConfig.glbx' not defined (check spark.conf)."
+                   << std::endl;
+      }
+      const auto cfg_file = data_dir.config_file().string();
+      auto glbx_cfg = md_cfg.glbx()[0];
+      const auto templates_file = data_dir.templates_file().string();
+      await_cme_config(
+          glbx_cfg.cme_config_url(), cfg_file, glbx_cfg.cme_templates_url(),
+          templates_file);
+      auto channels = get_cme_channels(cfg_file, {});
       for (const auto &ch : channels) {
          std::cout << ch.id << " - ";
          std::set<std::string> symbols;

src/app_sparkd/sparkd/cmds/md/cmds/Cleanup.cpp

@@ -1,9 +1,10 @@
 #include "Cleanup.h"
 #include "spark/config/config.h"
 #include "spark/config/protos/market_data.pb.h"
-#include "spark/logging/logging.h"
 #include "spark/md/ShmSegmentName.h"
 #include "spark/md/protos/co2.pb.h"
+#include "system/logging/logging.h"
+
 #include <boost/filesystem.hpp>
 #include <string>
 

src/app_sparkd/sparkd/cmds/md/cmds/Record.cpp

@@ -1,5 +1,6 @@
 #include "Record.h"
-#include "spark/md/relay/RelayRecorder.h"
+#include "spark/md/relay/relay_recorder.h"
+
 #include <boost/filesystem.hpp>
 
 namespace bts::cmds::md::cmds {
@@ -8,13 +9,14 @@ struct Record::Args {
    std::string path;
 };
 
-Record::Record() :
-    CLI::App("Record market data to a file", "record"),
-    _args(new Record::Args()) {
+Record::Record()
+    : CLI::App("Record market data to a file", "record"),
+      _args(new Record::Args()) {
    add_option("out,--out", _args->path, "Output file");
    callback([&] {
       boost::filesystem::path p(_args->path);
-      bts::md::record_everything(p.parent_path().string(), p.filename().string());
+      bts::md::record_everything(
+          p.parent_path().string(), p.filename().string());
    });
 }
 

src/app_sparkd/sparkd/cmds/md/cmds/Relay.cpp

@@ -1,5 +1,5 @@
 #include "Relay.h"
-#include "spark/md/relay/RelayClient.h"
+#include "spark/md/relay/relay_control.h"
 
 namespace bts::cmds::md::cmds {
 

src/app_sparkd/sparkd/cmds/md/cmds/Replay.cpp

@@ -1,10 +1,11 @@
 #include "Replay.h"
-#include "spark/md/relay/RelayRecorder.h"
+#include "spark/config/config.h"
+#include "spark/config/protos/market_data.pb.h"
+#include "spark/md/relay/relay_recorder.h"
 
 namespace bts::cmds::md::cmds {
 
-Replay::Replay() :
-    CLI::App("Replay recorded market data", "replay") {
+Replay::Replay() : CLI::App("Replay recorded market data", "replay") {
    callback([] {
       auto naggle_ms = config::get<bts::md::RelayOptions>().naggle_ms();
       const auto &options = config::get<bts::md::ReplayOptions>();

src/app_sparkd/sparkd/cmds/md/cmds/Run.cpp

@@ -1,25 +1,19 @@
 #include "Run.h"
 #include "Cleanup.h"
-#include "domain/security_criteria.h"
+#include "spark/api/md/MarketDataConfig.pb.h"
 #include "spark/config/protos/market_data.pb.h"
-#include "spark/logging/logging.h"
-#include "spark/md/SymbolLoader.h"
-#include "spark/md/UserSymbol.h"
-#include "spark/md/relay/RelayRecorder.h"
-#include "spark/os/os.h"
-#include "spark/utils/bts_assert.h"
+#include "spark/persistence/IDGenerator.h"
+#include "spark/utils/datetime.h"
 #include "spark/utils/thread_manager.h"
 #include "sparkd/md/MarketDataGateway.h"
-#include "sparkd/md/platforms/IMDPlatform.h"
-#include "sparkd/md/platforms/ice_onixs/ICEMain.h"
-#include "sparkd/md/platforms/mdp3_onixs/MDP3Main.h"
+#include "system/logging/logging.h"
+#include "system/os.h"
+
 #include <string>
+#include <thread>
 
 namespace bts::cmds::md::cmds {
 
-using bts::domain::SecurityCriteria;
-using bts::md::SymbolLoader;
-
 static void boot_sleep() {
    auto conf = config::get<bts::md::MDBootDelayWindow>();
    const auto *start = &conf.standard_start();
@@ -43,57 +37,17 @@ static void boot_sleep() {
    }
 }
 
-static void run_replay() {
-   LOG(md, Info, "Run::run_replay t_id:{} t_name", os::get_thread_id(),
-       os::get_thread_name());
-   const auto &replay_opts = config::get<bts::md::ReplayOptions>();
-   bts::md::replay_recording(
-       replay_opts.url(), config::get<bts::md::RelayOptions>().naggle_ms(),
-       replay_opts.fallback());
-}
-
-static void run_onixs(
-    bts::md::MBOWriter *mbo_writer,
-    bts::md::TradeSummarySignalDispatcher *tss_dispatcher) {
-   LOG(md, Info, "Run::run_onixs t_id:{} t_name", os::get_thread_id(),
-       os::get_thread_name());
-   std::vector<std::shared_ptr<bts::md::IMDPlatform>> platforms;
-   // get all glbx symbols
-   auto cme_symbols =
-       SymbolLoader::load(SecurityCriteria::create({domain::protobufs::VENUE_GLBX}));
-   if (!cme_symbols.empty()) {
-      auto *main = new bts::md::MDP3Main();
-      main->start(
-          {cme_symbols.begin(), cme_symbols.end()}, mbo_writer, tss_dispatcher);
-      platforms.emplace_back(main);
-   }
-   // ice
-   const auto ice_symbols =
-       SymbolLoader::load(SecurityCriteria::create({domain::protobufs::VENUE_ICE}));
-   if (!ice_symbols.empty()) {
-      auto *imain = new bts::md::ice::ICEMain();
-      imain->start({ice_symbols.begin(), ice_symbols.end()});
-      platforms.emplace_back(imain);
-   }
-   // deribit
-   // auto *d_platform = new bts::md::EmMDListener()
-   // Timers
-   auto &disp = utils::Dispatcher::current();
-   utils::DispatcherTimer heartbeat_timer(disp, 250, [&]() {
-      for (auto &p : platforms)
-         p->send_heartbeats();
-   });
-   heartbeat_timer.start();
-
-   disp.run();
-}
-
 static void run_md_gwy() {
    LOG(md, Info, "Run::run_md_gwy t_id:{} t_name:{}", os::get_thread_id(),
        os::get_thread_name());
-   bts::md::MarketDataGateway gwy = {};
-   // this keeps going in a loop
-   gwy.run();
+   bts::md::MarketDataGateway gwy{
+       config::get<api::md::MarketDataConfig>(),
+       persistence::IDGenerator::instance_ptr()};
+   gwy.start();
+   while (true) {
+      using namespace std::chrono_literals;
+      std::this_thread::sleep_for(2s);
+   }
 }
 
 void Run::run() {
@@ -108,13 +62,8 @@ void Run::run() {
 Run::Run()
     : CLI::App(
           "Start market data feed, and store up-to-date snapshots of mbo, and "
-          "mbp in shared memory.  If no subcommand specified, uses the "
-          "configured default feed (replay, or onixs)",
+          "mbp in shared memory.",
           "run") {
-   add_subcommand("replay", "Run replay market data")->callback(run_replay);
-   add_subcommand("onixs", "Run onixs market data")->callback([] {
-      run_onixs(nullptr, nullptr);
-   });
    callback([] { run(); });
 }
 

src/app_sparkd/sparkd/cmds/md/cmds/Run.h

@@ -9,12 +9,6 @@ namespace bts::cmds::md::cmds {
  * shared memory.
  */
 struct Run : public CLI::App {
-   enum class MDType {
-      Configured,
-      Replay,
-      Onixs
-   };
-
    Run();
 
    /**

src/app_sparkd/sparkd/cmds/md/cmds/rec_tool/cmds/Print.cpp

@@ -1,5 +1,5 @@
 #include "Print.h"
-#include "spark/md/relay/RelayRecorder.h"
+#include "spark/md/relay/relay_recorder.h"
 
 namespace bts::cmds::md::cmds::rec_tool::cmds {
 

src/app_sparkd/sparkd/cmds/md/cmds/rec_tool/cmds/Trace.cpp

@@ -1,5 +1,5 @@
 #include "Trace.h"
-#include "spark/md/relay/RelayRecorder.h"
+#include "spark/md/relay/relay_recorder.h"
 
 namespace bts::cmds::md::cmds::rec_tool::cmds {
 

src/app_sparkd/sparkd/cmds/md/cmds/rec_tool/cmds/Unprint.cpp

@@ -1,5 +1,5 @@
 #include "Unprint.h"
-#include "spark/md/relay/RelayRecorder.h"
+#include "spark/md/relay/relay_recorder.h"
 
 namespace bts::cmds::md::cmds::rec_tool::cmds {
 

src/app_sparkd/sparkd/cmds/md/common/VenueOption.cpp

@@ -18,7 +18,9 @@ EnumOptionNames<api::Venue> VenueOption::names() {
        NAME_PAIR(api::VENUE_BINANCE),
        NAME_PAIR(api::VENUE_BINANCE_US),
        NAME_PAIR(api::VENUE_COINBASE),
-       NAME_PAIR(api::VENUE_BINANCE_USDS_FUTURES)};
+       NAME_PAIR(api::VENUE_BINANCE_USDS_FUTURES),
+       NAME_PAIR(api::VENUE_CBOE),
+       NAME_PAIR(api::VENUE_TAIFEX)};
 }
 
 std::string VenueOption::name(api::Venue venue) {
@@ -45,6 +47,10 @@ std::string VenueOption::name(api::Venue venue) {
       return "coinbase";
    case api::VENUE_BINANCE_USDS_FUTURES:
       return "binance_usds_futures";
+   case api::VENUE_CBOE:
+      return "cboe";
+   case api::VENUE_TAIFEX:
+      return "taifex";
    default:
       return std::to_string(static_cast<int32_t>(venue));
    }

src/app_sparkd/sparkd/exchange_sim/ExchangeSim.cpp

@@ -6,31 +6,29 @@
 #include <quickfix/SessionSettings.h>
 #include <quickfix/SocketAcceptor.h>
 
-#include <fstream>
 #include <iostream>
 #include <string>
 
 #include "spark/config/config.h"
 #include "spark/config/protos/exchange_sim_cfg.pb.h"
 #include "spark/config/protos/ports.pb.h"
-#include "spark/config/protos/system_config.pb.h"
 #include "spark/domain_model/Security.pb.h"
-#include "spark/logging/logging.h"
 #include "spark/md/PriceSize.h"
-#include "spark/os/os.h"
+#include "system/logging/logging.h"
+#include "system/os.h"
+#include "system/protobufs/config/system_config.pb.h"
 
 #include "spark/domain_model/Security.pb.h"
 #include "spark/md/IntPrice.h"
 #include "spark/md/MBOProcessor.h"
-#include "spark/md/relay/RelayClient.h"
-#include "spark/md/relay/RelayRecorder.h"
-#include "spark/utils/bts_chrono.h"
+#include "spark/md/relay/relay_recorder.h"
 #include <quickfix/FileLog.h>
 
-#include <set>
-
 namespace bts {
 namespace sim {
+
+using bts::system::protobufs::SystemConfig;
+
 auto cfg = "[DEFAULT]\n"
            "ConnectionType=acceptor\n"
            "SocketAcceptPort=$EXCHANGE_SIM_FIX_PORT\n"
@@ -57,7 +55,7 @@ md::PriceSize coerce_precision(const co2::PriceSizeMessage &p) {
 struct MDPlayer : md::IRelayWriter {
    MDPlayer(
        IMDPlayerObserver &obs,
-       std::map<uint32_t, md::Security> &known_securities)
+       std::map<uint32_t, api::Security> &known_securities)
        : obs(obs), known_securities(known_securities),
          forward_implied(config::get<config::ExchangeSim>().forward_implied()) {
    }
@@ -130,7 +128,7 @@ struct MDPlayer : md::IRelayWriter {
          }
    }
    IMDPlayerObserver &obs;
-   std::map<uint32_t, md::Security> &known_securities;
+   std::map<uint32_t, api::Security> &known_securities;
    bool forward_implied;
 };
 
@@ -144,7 +142,7 @@ int start_exchange_sim(bts::api::Venue venue, md::MBOWriter *mbo) {
       auto venue_string = boost::to_lower_copy(api::Venue_Name(venue));
       boost::replace_all(
           str_settings, "$BTS_DATA_PATH",
-          config::get<config::SystemConfig>().data_path());
+          config::get<SystemConfig>().data_path());
       boost::replace_all(str_settings, "$SIM_VENUE", venue_string);
       std::string port_string;
       switch (venue) {
@@ -172,7 +170,7 @@ int start_exchange_sim(bts::api::Venue venue, md::MBOWriter *mbo) {
 
       acceptor.start();
 
-      std::map<uint32_t, md::Security> known_securities;
+      std::map<uint32_t, api::Security> known_securities;
       const auto &conf = config::get<config::ExchangeSim>();
       while (true) {
          MDPlayer player(application, known_securities);

src/app_sparkd/sparkd/exchange_sim/ExchangeSimApplication.cpp

@@ -5,11 +5,9 @@
 #include "spark/config/config.h"
 #include "spark/config/protos/exchange_sim_cfg.pb.h"
 #include "spark/domain_model/Security.h"
-#include "spark/logging/logging.h"
 #include "spark/md/MBOProcessor.h"
 #include "spark/md/MDProducer.h"
-#include "spark/md/Shm/FixedSecurity.h"
-#include "spark/md/Shm/MDStructure.h"
+#include "spark/md/shm/fixed_security.h"
 #include "spark/orderman/orderman_utils/fix_order_codec.h"
 #include "spark/persistence/SqliteStore.h"
 #include "spark/utils/Dispatcher.h"
@@ -17,6 +15,7 @@
 #include "sparkd/exchange_sim/book/Order.h"
 #include "sparkd/exchange_sim/book/OrderMatcher.h"
 #include "sparkd/exchange_sim/book/book_transform.h"
+#include "system/logging/logging.h"
 
 #include <boost/algorithm/string.hpp>
 #include <boost/date_time/posix_time/posix_time.hpp>
@@ -583,10 +582,11 @@ void negate_id(api::Security *sec) {
 }
 
 void ExchangeSimApplication::persist_book(OrderBook const &b) {
-   auto existing = me->store->retrieve_one("books", b.sec_desc());
+   auto [exists, rowid, oldval] =
+       me->store->retrieve_one("books", b.sec_desc());
    auto val = utils::pb::to_string(b);
 
-   if (existing.first)
+   if (exists)
       me->store->update("books", b.sec_desc(), val);
    else
       me->store->insert("books", b.sec_desc(), val);
@@ -619,7 +619,7 @@ void ExchangeSimApplication::on_new_security(api::Security const &s) {
 }
 
 void ExchangeSimApplication::on_settlement(
-    md::Security const &sec, md::IntPrice price) {
+    api::Security const &sec, md::IntPrice price) {
    auto sec_desc = get_sim_sec_desc(sec);
    auto iter = me->md_producer_handles.find(sec_desc);
    if (iter != me->md_producer_handles.end()) {
@@ -629,7 +629,8 @@ void ExchangeSimApplication::on_settlement(
    }
 }
 
-void ExchangeSimApplication::on_high(md::Security const &sec, md::Price price) {
+void ExchangeSimApplication::on_high(
+    api::Security const &sec, md::Price price) {
    auto sec_desc = get_sim_sec_desc(sec);
    auto iter = me->md_producer_handles.find(sec_desc);
    if (iter != me->md_producer_handles.end()) {
@@ -638,7 +639,7 @@ void ExchangeSimApplication::on_high(md::Security const &sec, md::Price price) {
    }
 }
 
-void ExchangeSimApplication::on_low(md::Security const &sec, md::Price price) {
+void ExchangeSimApplication::on_low(api::Security const &sec, md::Price price) {
    auto sec_desc = get_sim_sec_desc(sec);
    auto iter = me->md_producer_handles.find(sec_desc);
    if (iter != me->md_producer_handles.end()) {
@@ -647,7 +648,8 @@ void ExchangeSimApplication::on_low(md::Security const &sec, md::Price price) {
    }
 }
 
-void ExchangeSimApplication::on_open(md::Security const &sec, md::Price price) {
+void ExchangeSimApplication::on_open(
+    api::Security const &sec, md::Price price) {
    auto sec_desc = get_sim_sec_desc(sec);
    auto iter = me->md_producer_handles.find(sec_desc);
    if (iter != me->md_producer_handles.end()) {
@@ -657,7 +659,7 @@ void ExchangeSimApplication::on_open(md::Security const &sec, md::Price price) {
 }
 
 void ExchangeSimApplication::on_close(
-    md::Security const &sec, md::Price price) {
+    api::Security const &sec, md::Price price) {
    auto sec_desc = get_sim_sec_desc(sec);
    auto iter = me->md_producer_handles.find(sec_desc);
    if (iter != me->md_producer_handles.end()) {
@@ -667,7 +669,7 @@ void ExchangeSimApplication::on_close(
 }
 
 void ExchangeSimApplication::on_depth(
-    md::Security const &sec,
+    api::Security const &sec,
     md::PriceSize bids[md::COCONUT_MAX_DEPTH],
     md::PriceSize asks[md::COCONUT_MAX_DEPTH],
     std::array<md::PriceSize, md::COCONUT_MAX_IMPLIED_DEPTH> *implied_bids,

src/app_sparkd/sparkd/exchange_sim/ExchangeSimApplication.h

@@ -69,7 +69,8 @@ class ExchangeSimApplication : public sim::IMDPlayerObserver,
    void onLogon(const FIX::SessionID &sessionID) override;
    void onLogout(const FIX::SessionID &sessionID) override;
    void toAdmin(FIX::Message &, const FIX::SessionID &) override;
-   void toApp(FIX::Message &, const FIX::SessionID &) EXCEPT(FIX::DoNotSend) override;
+   void toApp(FIX::Message &, const FIX::SessionID &)
+       EXCEPT(FIX::DoNotSend) override;
    void fromAdmin(const FIX::Message &, const FIX::SessionID &) EXCEPT(
        FIX::FieldNotFound,
        FIX::IncorrectDataFormat,
@@ -117,14 +118,14 @@ class ExchangeSimApplication : public sim::IMDPlayerObserver,
    void update_md(std::string const &sec_desc);
    void persist_book(OrderBook const &b);
 
-   void on_new_security(md::Security const &) override;
-   void on_high(md::Security const &sec, md::Price price) override;
-   void on_low(md::Security const &sec, md::Price price) override;
-   void on_open(md::Security const &sec, md::Price price) override;
-   void on_close(md::Security const &sec, md::Price price) override;
-   void on_settlement(md::Security const &, md::IntPrice) override;
+   void on_new_security(api::Security const &) override;
+   void on_high(api::Security const &sec, md::Price price) override;
+   void on_low(api::Security const &sec, md::Price price) override;
+   void on_open(api::Security const &sec, md::Price price) override;
+   void on_close(api::Security const &sec, md::Price price) override;
+   void on_settlement(api::Security const &, md::IntPrice) override;
    void on_depth(
-       md::Security const &s,
+       api::Security const &s,
        md::PriceSize bids[md::COCONUT_MAX_DEPTH],
        md::PriceSize asks[md::COCONUT_MAX_DEPTH],
        std::array<md::PriceSize, md::COCONUT_MAX_IMPLIED_DEPTH> *implied_bids,

src/app_sparkd/sparkd/exchange_sim/IMDPlayerObserver.h

@@ -1,7 +1,7 @@
 #pragma once
 
 #include "spark/md/PriceSize.h"
-#include "spark/md/Shm/MDStructure.h"
+#include "spark/md/shm/hot_depth_entry.h"
 
 namespace bts {
 namespace api {
@@ -10,14 +10,14 @@ class Security;
 namespace sim {
 
 struct IMDPlayerObserver {
-   virtual void on_new_security(md::Security const &) = 0;
-   virtual void on_high(const md::Security &, md::Price) = 0;
-   virtual void on_low(const md::Security &, md::Price) = 0;
-   virtual void on_open(const md::Security &, md::Price) = 0;
-   virtual void on_close(const md::Security &, md::Price) = 0;
-   virtual void on_settlement(const md::Security &, md::IntPrice) = 0;
+   virtual void on_new_security(api::Security const &) = 0;
+   virtual void on_high(const api::Security &, md::Price) = 0;
+   virtual void on_low(const api::Security &, md::Price) = 0;
+   virtual void on_open(const api::Security &, md::Price) = 0;
+   virtual void on_close(const api::Security &, md::Price) = 0;
+   virtual void on_settlement(const api::Security &, md::IntPrice) = 0;
    virtual void on_depth(
-       md::Security const &s,
+       api::Security const &s,
        md::PriceSize bids[md::COCONUT_MAX_DEPTH],
        md::PriceSize asks[md::COCONUT_MAX_DEPTH],
        std::array<md::PriceSize, md::COCONUT_MAX_IMPLIED_DEPTH> *implied_bids,

src/app_sparkd/sparkd/exchange_sim/tests/test_book_transform.cpp

@@ -1,5 +1,6 @@
 #include "sparkd/exchange_sim/book/book_transform.h"
-#include <catch/catch.hpp>
+
+#include <catch2/catch_all.hpp>
 #include <stdlib.h>
 
 namespace bts {

src/app_sparkd/sparkd/execution/EMExecutionGateway.cpp

@@ -5,7 +5,6 @@
 #include "spark/domain_model/Security.pb.h"
 #include "spark/domain_model/SecurityKey.h"
 #include "spark/domain_model/common.h"
-#include "spark/logging/logging.h"
 #include "spark/md/SecurityOracle.h"
 #include "spark/orderman/CancelOrderMsg.h"
 #include "spark/orderman/CancelReplaceOrderMsg.h"
@@ -20,13 +19,14 @@
 #include "spark/orderman/orderman_utils/ExchangePosition.h"
 #include "spark/orderman/orderman_utils/ExecutionReport.h"
 #include "spark/orderman/orderman_utils/orderman_common.h"
-#include "spark/os/os.h"
 #include "spark/tracing/bts_trace.h"
 #include "spark/utils/bts_assert.h"
 #include "spark/utils/datetime.h"
 //#include "spark/utils/helper_macros.h"
 #include "sparkd/md/platforms/emumba/EmMDListener.h"
 #include "sparkd/md/platforms/emumba/IEMOrderMuxer.h"
+#include "system/logging/logging.h"
+#include "system/os.h"
 
 #include "spark/api/service_helpers/AdminServiceUtils.h"
 #include "spark/config/config.h"

src/app_sparkd/sparkd/http/ApiGateway.cpp

@@ -4,8 +4,9 @@
 #include "spark/api/public/ApiUtils.h"
 #include "spark/api/services/AdminService.client.pb.h"
 #include "spark/api/services/AuthService.client.pb.h"
-#include "spark/logging/logging.h"
 #include "spark/utils/bts_assert.h"
+#include "system/logging/logging.h"
+
 #include <mutex>
 #include <unordered_map>
 #include <vector>

src/app_sparkd/sparkd/http/MultipartFormDecoder.cpp

@@ -1,5 +1,6 @@
 #include "MultipartFormDecoder.h"
-#include "spark/logging/logging.h"
+#include "system/logging/logging.h"
+
 #include <boost/algorithm/string.hpp>
 #include <boost/algorithm/string/iter_find.hpp>
 

src/app_sparkd/sparkd/http/WebServer.cpp

@@ -5,12 +5,12 @@
 #include "MultipartFormDecoder.h"
 #include "spark/api/apigateway.pb.h"
 #include "spark/api/core/ApiClientInterop.h"
-#include "spark/logging/logging.h"
-#include "spark/os/os.h"
 #include "spark/tracing/PerfLog.h"
 #include "spark/utils/Dispatcher.h"
 #include "spark/utils/TimeStamp.h"
 #include "spark/utils/bts_chrono.h"
+#include "system/logging/logging.h"
+#include "system/os.h"
 
 #include <websocketpp/config/asio_no_tls.hpp>
 #include <websocketpp/server.hpp>

src/app_sparkd/sparkd/http/tests/MultiPartFormTests.cpp

@@ -1,6 +1,6 @@
 #include "../MultipartFormDecoder.h"
-#include "catch/catch.hpp"
 
+#include <catch2/catch_all.hpp>
 #include <string>
 
 namespace bts {
@@ -49,4 +49,4 @@ TEST_CASE("MultiPart form decoder", "[http]") {
 
 } // namespace test
 } // namespace http
-} // namespace bts
+} // namespace bts

src/app_sparkd/sparkd/md/IMarketDataSource.h

@@ -1,9 +1,11 @@
 #pragma once
 
-#include "spark/md/IMarketDataListener.h"
+#include <memory>
 
 namespace bts::md {
 
+class IMarketDataListener;
+
 /**
  * @brief This interface represents a source of market data. Call `IMarketDataSource::start` to
  * connect an `IMarketDataListener` to the source and begin receiving market data.

src/app_sparkd/sparkd/md/MarketDataAuxiliaryServiceServer.cpp

@@ -5,10 +5,10 @@
 #include "spark/api/services/MarketDataRelay.pb.h"
 #include "spark/config/config.h"
 #include "spark/config/protos/toggles.pb.h"
-#include "spark/logging/logging.h"
 #include "spark/utils/Dispatcher.h"
 #include "spark/utils/PBUtils.h"
 #include "spark/utils/bts_chrono.h"
+#include "system/logging/logging.h"
 
 namespace bts {
 namespace md {

src/app_sparkd/sparkd/md/MarketDataGateway.cpp

@@ -2,14 +2,11 @@
 #include "domain/security_id.h"
 #include "spark/api/md/MarketDataConfig.pb.h"
 #include "spark/api/md/MarketDataConfigUtils.h"
-#include "spark/config/config.h"
 #include "spark/domain/DomainSecurityUtils.h"
-#include "spark/logging/logging.h"
 #include "spark/md/GeneralMarketDataListener.h"
 #include "spark/md/GeneralSecurityIndex.h"
 #include "spark/md/IMarketDataListener.h"
 #include "spark/md/MDProducer.h"
-#include "spark/os/os.h"
 #include "spark/persistence/IDGenerator.h"
 #include "spark/utils/bts_fatal.h"
 #include "sparkd/md/IMarketDataProvider.h"
@@ -18,12 +15,14 @@
 #include "sparkd/md/PyMarketDataSubscription.h"
 #include "sparkd/md/ReplayMarketDataProvider.h"
 #include "sparkd/md/ccapi/CCApiMarketDataProvider.h"
+#include "sparkd/md/databento/DatabentoMarketDataProvider.h"
 #include "sparkd/md/emumba/EmumbaMarketDataProvider.h"
+#include "sparkd/md/taifex/TaifexMarketDataProvider.h"
 #include "system/FMT.h"
+#include "system/logging/logging.h"
+#include "system/os.h"
 
 #include <bits/stdint-uintn.h>
-#include <boost/filesystem.hpp>
-#include <boost/filesystem/path.hpp>
 #include <memory>
 #include <vector>
 
@@ -37,8 +36,6 @@ using std::string;
 using std::unique_ptr;
 using std::vector;
 
-using Path = boost::filesystem::path;
-
 static bool validate(const MarketDataConfig &cfg, vector<string> &errs) {
    auto orig_n_errs = errs.size();
    if (!cfg.deribit().empty() && !cfg.emumba_deribit().empty()) {
@@ -50,40 +47,51 @@ static bool validate(const MarketDataConfig &cfg, vector<string> &errs) {
 
 struct MarketDataGateway::PImpl {
    struct MdSink {
-      GeneralSecurityIndex sec_index;
-      MDProducer md_producer;
+      shared_ptr<GeneralSecurityIndex> sec_index;
+      shared_ptr<MDProducer> md_producer;
       shared_ptr<IMarketDataListener> listener;
 
-      MdSink(api::Venue venue, IDGenerator *sec_id_store)
-          : sec_index(), listener(std::make_shared<GeneralMarketDataListener>(
-                             sec_index, md_producer, venue, sec_id_store)) {}
+      MdSink(api::Venue venue, shared_ptr<IDGenerator> sec_id_store)
+          : sec_index(std::make_shared<GeneralSecurityIndex>()),
+            md_producer(std::make_shared<MDProducer>()),
+            listener(std::make_shared<GeneralMarketDataListener>(
+                sec_index, md_producer, venue, sec_id_store)) {}
    };
 
-   MarketDataConfig config = config::get<MarketDataConfig>();
-   IDGenerator &sec_id_store = IDGenerator::instance();
+   MarketDataConfig config;
+   shared_ptr<IDGenerator> sec_id_store;
    vector<unique_ptr<IMarketDataProvider>> providers;
    vector<unique_ptr<MdSink>> sinks;
    vector<unique_ptr<IMarketDataSource>> sources;
 
-   PImpl() {
-      providers.emplace_back(std::make_unique<CCApiMarketDataProvider>(config));
+   PImpl(MarketDataConfig config, shared_ptr<IDGenerator> sec_id_store)
+       : config(std::move(config)), sec_id_store(std::move(sec_id_store)) {
+      providers.emplace_back(std::make_unique<CCApiMarketDataProvider>(
+          this->config, this->sec_id_store));
       providers.emplace_back(
-          std::make_unique<EmumbaMarketDataProvider>(config));
-      providers.emplace_back(std::make_unique<OnixsMarketDataProvider>(config));
+          std::make_unique<EmumbaMarketDataProvider>(this->config));
       providers.emplace_back(
-          std::make_unique<PyMarketDataProvider>(config, &sec_id_store));
-      providers.emplace_back(
-          std::make_unique<ReplayMarketDataProvider>(config, &sec_id_store));
+          std::make_unique<OnixsMarketDataProvider>(this->config));
+      providers.emplace_back(std::make_unique<DatabentoMarketDataProvider>(
+          this->config, this->sec_id_store));
+      providers.emplace_back(std::make_unique<TaifexMarketDataProvider>(
+          this->config, this->sec_id_store));
+      providers.emplace_back(std::make_unique<PyMarketDataProvider>(
+          this->config, this->sec_id_store));
+      providers.emplace_back(std::make_unique<ReplayMarketDataProvider>(
+          this->config, this->sec_id_store));
    }
 };
 
-MarketDataGateway::MarketDataGateway()
-    : me_(std::make_unique<MarketDataGateway::PImpl>()) {}
+MarketDataGateway::MarketDataGateway(
+    MarketDataConfig config, shared_ptr<IDGenerator> sec_id_store)
+    : me_(std::make_unique<MarketDataGateway::PImpl>(
+          std::move(config), std::move(sec_id_store))) {}
 
 MarketDataGateway::~MarketDataGateway() = default;
 
 // starts the market data sources and then keeps the function going
-void MarketDataGateway::run() {
+void MarketDataGateway::start() {
    vector<string> cfg_errs;
    if (!validate(me_->config, cfg_errs)) {
       BTS_FATAL_MSGF("Invalid MarketDataConfig:", FMT_JOIN(cfg_errs, "\n"));
@@ -95,15 +103,10 @@ void MarketDataGateway::run() {
       for (auto &[venue, source] : provider->get_sources()) {
          const auto v = domain::DomainSecurityUtils::convert(venue);
          auto &sink = me_->sinks.emplace_back(
-             std::make_unique<PImpl::MdSink>(v, &me_->sec_id_store));
+             std::make_unique<PImpl::MdSink>(v, me_->sec_id_store));
          source->start(sink->listener);
       }
    }
-   while (true) {
-      std::this_thread::sleep_for(std::chrono::seconds(1));
-   }
-   LOG(md, Info, "MarketDataGateway::run ending t_id:{} t_name:{}",
-       os::get_thread_id(), os::get_thread_name());
 }
 
 } // namespace bts::md

src/app_sparkd/sparkd/md/MarketDataGateway.h

@@ -2,6 +2,14 @@
 
 #include <memory>
 
+namespace bts::api::md {
+class MarketDataConfig;
+}
+
+namespace bts::persistence {
+class IDGenerator;
+}
+
 namespace bts::md {
 
 /**
@@ -10,10 +18,12 @@ namespace bts::md {
  */
 class MarketDataGateway {
  public:
-   MarketDataGateway();
+   explicit MarketDataGateway(
+       api::md::MarketDataConfig config,
+       std::shared_ptr<persistence::IDGenerator> sec_id_store);
    ~MarketDataGateway();
 
-   void run();
+   void start();
 
  private:
    struct PImpl;

src/app_sparkd/sparkd/md/MarketDataListener.h

@@ -3,9 +3,9 @@
 #include "domain/book.h"
 #include "domain/rfq.h"
 #include "domain/security.h"
+#include "domain/security_market_data_statistics.h"
 #include "domain/security_state.h"
 #include "domain/trade_record.h"
-#include "spark/domain/SecurityMarketDataStatistics.h"
 #include "spark/md/IMarketDataListener.h"
 #include "sparkd/md/MarketDataError.h"
 
@@ -19,7 +19,7 @@ class MarketDataListener : public IMarketDataListener {
    void on_security_statistics(
        const domain::SecurityMarketDataStatistics &security_stats) override;
 
-   void on_book(const domain::Book &book) override;
+   void on_book(const domain::Book &book, std::optional<utils::TimeStamp>) override;
    void on_trades(
        const std::vector<const domain::TradeRecord *> &trades) override;
    void on_rfqs(const std::vector<const domain::RFQ *> &rfqs) override;
@@ -39,7 +39,7 @@ inline void MarketDataListener::on_security_state(
 inline void MarketDataListener::on_security_statistics(
     const domain::SecurityMarketDataStatistics & /*security_stats*/) {}
 
-inline void MarketDataListener::on_book(const domain::Book & /*book*/) {}
+inline void MarketDataListener::on_book(const domain::Book & /*book*/, std::optional<utils::TimeStamp>) {}
 
 inline void MarketDataListener::on_trades(
     const std::vector<const domain::TradeRecord *> & /*trades*/) {}

src/app_sparkd/sparkd/md/MarketDataListenerID.h

@@ -1,13 +1,13 @@
 #pragma once
 #ifndef _WIN32
 
-#include "spark/core/ValueWrapper.h"
 #include "system/FMT_IMPL.h"
+#include "system/value_wrapper.h"
 
 namespace bts::md {
 
 using MarketDataListenerID =
-    core::ValueWrapper<struct MarketDataListenedIDTag, std::uint32_t>;
+    utils::ValueWrapper<struct MarketDataListenedIDTag, std::uint32_t>;
 
 } // namespace bts::md
 

src/app_sparkd/sparkd/md/OnixsMarketDataProvider.cpp

@@ -11,23 +11,20 @@
 namespace bts::md {
 
 using bts::api::md::MarketDataConfig;
+using bts::domain::protobufs::Venue;
+using bts::md::OnixsMarketDataProvider;
 using bts::persistence::IDGenerator;
 using std::pair;
 using std::shared_ptr;
 using std::string;
 using std::vector;
 
-using Path = boost::filesystem::path;
-
 struct OnixsMarketDataProvider::PImpl {
    MarketDataConfig config;
-   IDGenerator &sec_id_store = IDGenerator::instance();
-   shared_ptr<OnixsGlbxSubscription> source;
+   std::shared_ptr<IDGenerator> sec_id_store;
+   vector<shared_ptr<OnixsGlbxSubscription>> sources;
 };
 
-OnixsMarketDataProvider::OnixsMarketDataProvider()
-    : bts::md::OnixsMarketDataProvider(bts::config::get<MarketDataConfig>()) {}
-
 OnixsMarketDataProvider::OnixsMarketDataProvider(MarketDataConfig config)
     : me_(std::make_unique<PImpl>()) {
    me_->config = std::move(config);
@@ -35,12 +32,14 @@ OnixsMarketDataProvider::OnixsMarketDataProvider(MarketDataConfig config)
 
 OnixsMarketDataProvider::~OnixsMarketDataProvider() = default;
 
-vector<pair<domain::protobufs::Venue, shared_ptr<IMarketDataSource>>>
+vector<pair<Venue, shared_ptr<IMarketDataSource>>>
 OnixsMarketDataProvider::get_sources() {
-   vector<pair<domain::protobufs::Venue, shared_ptr<IMarketDataSource>>> rv;
-   me_->source =
-       std::make_shared<OnixsGlbxSubscription>(&me_->sec_id_store, me_->config);
-   rv.push_back({domain::protobufs::VENUE_GLBX, me_->source});
+   vector<pair<Venue, shared_ptr<IMarketDataSource>>> rv;
+   for (const auto &glbx : me_->config.glbx()) {
+      me_->sources.push_back(
+          std::make_shared<OnixsGlbxSubscription>(me_->sec_id_store, glbx));
+      rv.push_back({Venue::VENUE_GLBX, me_->sources.back()});
+   }
    return rv;
 }
 

src/app_sparkd/sparkd/md/PyMarketDataProvider.cpp

@@ -1,13 +1,15 @@
 #include "PyMarketDataProvider.h"
 #include "spark/api/md/MarketDataConfig.pb.h"
-#include "spark/os/os.h"
 #include "spark/persistence/IDGenerator.h"
+#include "spark/utils/bts_assert.h"
 #include "sparkd/md/IMarketDataSource.h"
 #include "sparkd/md/MarketDataSubscription.h"
 #include "sparkd/md/PyMarketDataSubscription.h"
+#include "system/os.h"
 
 namespace bts::md {
 
+using bts::api::md::MarketDataConfig;
 using bts::os::sparkd_wd;
 using bts::persistence::IDGenerator;
 using std::pair;
@@ -18,16 +20,17 @@ using std::vector;
 using Path = boost::filesystem::path;
 
 struct PyMarketDataProvider::PImpl {
-   api::md::MarketDataConfig config;
-   IDGenerator *sec_id_store;
-   vector<shared_ptr<PyMarketDataSubscription>> py_md_subs = {};
+   MarketDataConfig config;
+   shared_ptr<IDGenerator> sec_id_store;
+   vector<shared_ptr<PyMarketDataSubscription>> py_md_subs;
+
+   PImpl(MarketDataConfig config, shared_ptr<IDGenerator> sec_id_store)
+       : config(std::move(config)), sec_id_store(std::move(sec_id_store)) {}
 };
 
 PyMarketDataProvider::PyMarketDataProvider(
-    api::md::MarketDataConfig config, IDGenerator *sec_id_store)
-    : me_(std::make_unique<PImpl>()) {
-   me_->config = std::move(config);
-   me_->sec_id_store = sec_id_store;
+    MarketDataConfig config, shared_ptr<IDGenerator> sec_id_store)
+    : me_(std::make_unique<PImpl>(std::move(config), std::move(sec_id_store))) {
 }
 
 PyMarketDataProvider::~PyMarketDataProvider() = default;
@@ -38,7 +41,7 @@ PyMarketDataProvider::get_sources() {
    const Path wd{sparkd_wd()};
    vector<string> module_paths{
        wd.string(),
-       (wd / "py_market_data_providers").string(),
+       (wd / "python_md_providers").string(),
    };
    for (const auto &py_plugin : me_->config.python_md_providers()) {
       vector<string> module_path;

src/app_sparkd/sparkd/md/PyMarketDataProvider.h

@@ -12,7 +12,7 @@ class PyMarketDataProvider : public IMarketDataProvider {
  public:
    explicit PyMarketDataProvider(
        api::md::MarketDataConfig config,
-       persistence::IDGenerator *sec_id_store);
+       std::shared_ptr<persistence::IDGenerator> sec_id_store);
    ~PyMarketDataProvider();
 
    std::vector<

src/app_sparkd/sparkd/md/PyMarketDataSubscription.cpp

@@ -2,6 +2,8 @@
 #include "domain/future_security.h"
 #include "domain/protobufs/aggressor_side.pb.h"
 #include "domain/protobufs/call_put.pb.h"
+#include "domain/protobufs/security_type.h"
+#include "domain/protobufs/security_type.pb.h"
 #include "domain/protobufs/venue.h"
 #include "domain/security.h"
 #include "domain/security_id.h"
@@ -13,11 +15,11 @@
 #include "python_md/py_security.h"
 #include "python_md/py_security_list.h"
 #include "python_md/py_trade.h"
-#include "spark/logging/logging.h"
 #include "spark/md/IMarketDataListener.h"
-#include "spark/os/os.h"
 #include "spark/persistence/IDGenerator.h"
 #include "spark/utils/bts_assert.h"
+#include "system/logging/logging.h"
+#include "system/os.h"
 #include "system/time_stamp.h"
 
 #include <memory>
@@ -53,11 +55,13 @@ struct TradeTuple {
 
 struct PyMarketDataSubscription::PImpl : private py_interop::MdSink {
    PImpl(
-       string py_plugin, vector<string> module_paths, IDGenerator *sec_id_store)
+       string py_plugin,
+       vector<string> module_paths,
+       std::shared_ptr<IDGenerator> sec_id_store)
        : py_plugin_(std::move(py_plugin)),
          module_paths_(std::move(module_paths)),
          py_md_exec_(py_plugin_, module_paths_, this),
-         sec_id_store_(sec_id_store) {}
+         sec_id_store_(std::move(sec_id_store)) {}
 
    void start(shared_ptr<IMarketDataListener> listener) {
       bts_assert(listener != nullptr);
@@ -75,7 +79,7 @@ struct PyMarketDataSubscription::PImpl : private py_interop::MdSink {
    PyMdExecutor py_md_exec_;
    shared_ptr<IMarketDataListener> listener_;
 
-   IDGenerator *sec_id_store_;
+   std::shared_ptr<IDGenerator> sec_id_store_;
    // The main store for securities.
    vector<domain::Security> securities_;
    vector<const domain::Security *> security_ptrs_;
@@ -143,13 +147,19 @@ struct PyMarketDataSubscription::PImpl : private py_interop::MdSink {
       trade_ptrs_.clear();
    }
 
-   SecurityID get_sec_id_(const string &venue_name, const string &sec_name) {
+   SecurityID get_sec_id_(
+       const string &venue_name,
+       const string &type_name,
+       const string &sec_name) {
       domain::protobufs::Venue venue;
       // Although an overload of 'get_id' accepts a string, we parse `venue_name` to the enum to
       // normalize different spellings of the same venue. For example, "deribt", "DERIBIT", and
       // "VENUE_deribit" all represent the VENUE_DERIBIT enum (see `venue_parse_smart` for details).
       bts_assert(domain::protobufs::venue_parse_smart(venue_name, &venue));
-      auto sec_id = sec_id_store_->get_id(venue, sec_name, true);
+      domain::protobufs::SecurityType type;
+      bts_assert(
+          domain::protobufs::security_type_parse_smart(type_name, &type));
+      auto sec_id = sec_id_store_->get_id(venue, type, sec_name, true);
       bts_assert_msgf(
           sec_id, "Failed to get security id for py_book. venue: {}; name: {}",
           venue, sec_name);
@@ -168,7 +178,8 @@ struct PyMarketDataSubscription::PImpl : private py_interop::MdSink {
          LOG(md, Error, "Failed to parse venue from '{}'", py_sec.venue());
          return std::nullopt;
       }
-      sec_comm.id = get_sec_id_(py_sec.venue(), py_sec.exchange_id());
+      sec_comm.id =
+          get_sec_id_(py_sec.venue(), py_sec.type(), py_sec.exchange_id());
       sec_comm.exchange_sec_id = py_sec.exchange_id();
       sec_comm.currency = py_sec.currency();
       sec_comm.asset = py_sec.asset();
@@ -203,7 +214,9 @@ struct PyMarketDataSubscription::PImpl : private py_interop::MdSink {
    }
 
    void to_bts_trade_(const PyTrade &source, domain::TradeRecord &dest) {
-      dest.sec_id = get_sec_id_(source.venue(), source.exchange_security_id());
+      dest.sec_id = get_sec_id_(
+          source.venue(), source.security_type(),
+          source.exchange_security_id());
       dest.time_stamp = core::TimeStamp::from_nanos(source.timestamp());
       dest.agg_side = source.aggressor_side().starts_with('B') ||
                               source.aggressor_side().starts_with('b')
@@ -229,15 +242,20 @@ struct PyMarketDataSubscription::PImpl : private py_interop::MdSink {
    void to_book_(const PyBook &source, domain::Book &dest) {
       to_book_side_(source.bids(), dest.bids());
       to_book_side_(source.asks(), dest.asks());
-      dest.set_security_id(
-          get_sec_id_(source.venue(), source.exchange_security_id()));
+      dest.set_security_id(get_sec_id_(
+          source.venue(), source.security_type(),
+          source.exchange_security_id()));
    }
 };
 
 PyMarketDataSubscription::PyMarketDataSubscription(
-    string py_plugin, vector<string> module_paths, IDGenerator *sec_id_store)
+    string py_plugin,
+    vector<string> module_paths,
+    shared_ptr<IDGenerator> sec_id_store)
     : me_(std::make_unique<PyMarketDataSubscription::PImpl>(
-          std::move(py_plugin), std::move(module_paths), sec_id_store)) {}
+          std::move(py_plugin),
+          std::move(module_paths),
+          std::move(sec_id_store))) {}
 
 PyMarketDataSubscription::~PyMarketDataSubscription() { me_->stop(); }
 

src/app_sparkd/sparkd/md/PyMarketDataSubscription.h

@@ -17,7 +17,7 @@ class PyMarketDataSubscription : public IMarketDataSource {
    explicit PyMarketDataSubscription(
        std::string py_plugin,
        std::vector<std::string> module_paths,
-       persistence::IDGenerator *sec_id_store);
+       std::shared_ptr<persistence::IDGenerator> sec_id_store);
    ~PyMarketDataSubscription();
 
    void start(std::shared_ptr<IMarketDataListener> listener) override;

src/app_sparkd/sparkd/md/ReplayMarketDataProvider.cpp

@@ -10,6 +10,7 @@
 namespace bts::md {
 
 using bts::api::md::MarketDataConfig;
+using bts::domain::DomainSecurityUtils;
 using bts::md::recording::MDReplayer;
 using bts::persistence::IDGenerator;
 using std::pair;
@@ -19,20 +20,21 @@ using std::vector;
 
 struct ReplayMarketDataProvider::PImpl {
    MarketDataConfig config;
-   IDGenerator *sec_id_store;
+   shared_ptr<IDGenerator> sec_id_store;
    vector<pair<domain::protobufs::Venue, shared_ptr<IMarketDataSource>>>
        sources;
+
+   PImpl(MarketDataConfig config, shared_ptr<IDGenerator> sec_id_store)
+       : config(std::move(config)), sec_id_store(std::move(sec_id_store)) {}
 };
 
 ReplayMarketDataProvider::ReplayMarketDataProvider()
     : ReplayMarketDataProvider(
-          bts::config::get<MarketDataConfig>(), &IDGenerator::instance()) {}
+          config::get<MarketDataConfig>(), IDGenerator::instance_ptr()) {}
 
 ReplayMarketDataProvider::ReplayMarketDataProvider(
-    MarketDataConfig config, IDGenerator *sec_id_store)
-    : me_(std::make_unique<PImpl>()) {
-   me_->config = std::move(config);
-   me_->sec_id_store = sec_id_store;
+    MarketDataConfig config, shared_ptr<IDGenerator> sec_id_store)
+    : me_(std::make_unique<PImpl>(std::move(config), std::move(sec_id_store))) {
 }
 
 ReplayMarketDataProvider::~ReplayMarketDataProvider() = default;
@@ -42,7 +44,7 @@ ReplayMarketDataProvider::get_sources() {
    const auto replay_cfgs =
        MarketDataConfigUtils::get_effective_replay_config(me_->config);
    for (const auto &r : replay_cfgs) {
-      const auto v = domain::DomainSecurityUtils::convert(r.venue());
+      const auto v = DomainSecurityUtils::convert(r.venue());
       me_->sources.push_back({v, std::make_shared<MDReplayer>(r.venue())});
    }
    return me_->sources;

src/app_sparkd/sparkd/md/ReplayMarketDataProvider.h

@@ -13,7 +13,7 @@ class ReplayMarketDataProvider : public IMarketDataProvider {
    ReplayMarketDataProvider();
    explicit ReplayMarketDataProvider(
        api::md::MarketDataConfig config,
-       persistence::IDGenerator *sec_id_store);
+       std::shared_ptr<persistence::IDGenerator> sec_id_store);
    ~ReplayMarketDataProvider();
 
    std::vector<

src/app_sparkd/sparkd/md/ReplayMarketDataSubscription.cpp

@@ -1,11 +1,11 @@
 #include "ReplayMarketDataSubscription.h"
 #include "spark/api/md/MarketDataReplayConfig.pb.h"
-#include "spark/logging/logging.h"
 #include "spark/md/GeneralMarketDataListener.h"
 #include "spark/md/IMarketDataListener.h"
 #include "spark/md/MDProducer.h"
 #include "spark/utils/thread_manager.h"
 #include "sparkd/md/recording/MDReplayer.h"
+#include "system/logging/logging.h"
 
 #include <ccapi_cpp/ccapi_event_dispatcher.h>
 #include <optional>

src/app_sparkd/sparkd/md/binance/BinanceIcebergPartsFilter.h

@@ -1,7 +1,7 @@
 #pragma once
 
-#include "spark/io/JsonMacros.h"
 #include "sparkd/md/binance/BinanceFilterType.h"
+#include "system/json/json_macros.h"
 #include "system/json/json_reader.h"
 
 #include <iostream>

src/app_sparkd/sparkd/md/binance/BinanceLotSizeFilter.h

@@ -1,8 +1,8 @@
 #pragma once
 
-#include "spark/logging/logging.h"
 #include "sparkd/md/binance/BinanceFilterType.h"
 #include "system/json/json_reader.h"
+#include "system/logging/logging.h"
 
 #include <nlohmann/json.hpp>
 #include <string>

src/app_sparkd/sparkd/md/binance/BinanceMinNotionalFilter.h

@@ -1,7 +1,7 @@
 #pragma once
 
-#include "spark/io/JsonMacros.h"
 #include "sparkd/md/binance/BinanceFilterType.h"
+#include "system/json/json_macros.h"
 #include "system/json/json_reader.h"
 
 #include <iostream>

src/app_sparkd/sparkd/md/binance/BinanceMiniTickerUpdate.cpp

@@ -2,7 +2,7 @@
 #include "BinanceMiniTickerUpdate.h"
 #include "domain/price.h"
 #include "domain/quantity.h"
-#include "spark/domain/SecurityMarketDataStatistics.h"
+#include "domain/security_market_data_statistics.h"
 #include "sparkd/md/ccapi/CCApiReader.h"
 #include "system/json/json_dumper.h"
 #include "system/json/json_reader.h"
@@ -38,13 +38,15 @@ static bool parse(
    JsonReader jreader{&js};
    if (!jreader.read<bool>("success")) {
       LOG(md, Error,
-          "BinanceMiniTickerUpdate json field 'success' is false:\n{}", js);
+          "BinanceMiniTickerUpdate json field 'success' is false:\n{}",
+          js.dump());
       return false;
    }
 
    if (!jreader.read("result", ticker_upds)) {
       LOG(md, Error,
-          "BinanceMiniTickerUpdate json field 'result' is invalid:\n{}", js);
+          "BinanceMiniTickerUpdate json field 'result' is invalid:\n{}",
+          js.dump());
    }
    return !ticker_upds.empty();
 }

src/app_sparkd/sparkd/md/binance/BinanceMiniTickerUpdate.h

@@ -1,6 +1,6 @@
 #pragma once
 
-#include "spark/domain/SecurityMarketDataStatistics.h"
+#include "domain/security_market_data_statistics.h"
 #include "sparkd/md/binance/BinanceMiniTickerEventType.h"
 #include "system/time_stamp.h"
 

src/app_sparkd/sparkd/md/binance/BinanceNotionalFilter.h

@@ -1,7 +1,7 @@
 #pragma once
 
-#include "spark/io/JsonMacros.h"
 #include "sparkd/md/binance/BinanceFilterType.h"
+#include "system/json/json_macros.h"
 #include "system/json/json_reader.h"
 
 #include <nlohmann/json.hpp>

src/app_sparkd/sparkd/md/binance/BinancePriceStatistics.cpp

@@ -1,7 +1,7 @@
 #include "BinancePriceStatistics.h"
 #include "domain/price.h"
 #include "domain/quantity.h"
-#include "spark/domain/SecurityMarketDataStatistics.h"
+#include "domain/security_market_data_statistics.h"
 #include "sparkd/md/ccapi/CCApiReader.h"
 #include "system/json/json_dumper.h"
 #include "system/json/json_reader.h"
@@ -33,7 +33,8 @@ static bool parse(
    }
    JsonReader jr{&js};
    if (!jr.read(result)) {
-      LOG(md, Error, "Failed to read BinancePriceStatistics json:\n{}", js);
+      LOG(md, Error, "Failed to read BinancePriceStatistics json:\n{}",
+          js.dump());
       return false;
    }
    return true;

src/app_sparkd/sparkd/md/binance/BinancePriceStatisticsAdaptor.cpp

@@ -1,6 +1,6 @@
 #include "BinancePriceStatisticsAdaptor.h"
 #include "domain/security_id.h"
-#include "spark/domain/SecurityMarketDataStatistics.h"
+#include "domain/security_market_data_statistics.h"
 #include "spark/md/IMarketDataListener.h"
 #include "spark/utils/bts_fatal.h"
 #include "sparkd/md/binance/BinancePriceStatistics.h"
@@ -12,18 +12,19 @@ namespace bts::md::binance {
 
 using bts::domain::SecurityID;
 using bts::domain::SecurityMarketDataStatistics;
-using std::pair;
+using bts::domain::protobufs::SecurityType;
 using std::string;
+using std::tuple;
 using std::vector;
 
 #define VENUE_NAME (Venue_Name(Venue::BINANCE))
 
 struct BinancePriceStatisticsAdaptor::PImpl {
-   vector<pair<string, SecurityID>> securities;
+   vector<tuple<SecurityType, string, SecurityID>> securities;
    IMarketDataListener *listener;
 
    SecurityID get_sec_id(const string &sec_desc) const {
-      for (const auto &[desc, id] : securities) {
+      for (const auto &[type, desc, id] : securities) {
          if (desc == sec_desc) {
             return id;
          }
@@ -36,7 +37,8 @@ struct BinancePriceStatisticsAdaptor::PImpl {
 };
 
 BinancePriceStatisticsAdaptor::BinancePriceStatisticsAdaptor(
-    vector<pair<string, SecurityID>> securities, IMarketDataListener *listener)
+    vector<tuple<SecurityType, string, SecurityID>> securities,
+    IMarketDataListener *listener)
     : me_(std::make_unique<PImpl>()) {
    bts_assert(listener != nullptr);
    me_->securities = std::move(securities);

src/app_sparkd/sparkd/md/binance/BinancePriceStatisticsAdaptor.h

@@ -1,9 +1,11 @@
 #pragma once
 
+#include "domain/protobufs/security_type.pb.h"
 #include "domain/security_id.h"
 #include "sparkd/md/ccapi/ICCApiMessageHandler.h"
 
 #include <string>
+#include <tuple>
 #include <utility>
 #include <vector>
 
@@ -20,7 +22,10 @@ namespace bts::md::binance {
 class BinancePriceStatisticsAdaptor : public ICCApiMessageHandler {
  public:
    BinancePriceStatisticsAdaptor(
-       std::vector<std::pair<std::string, domain::SecurityID>> securities,
+       std::vector<std::tuple<
+           domain::protobufs::SecurityType,
+           std::string,
+           domain::SecurityID>> securities,
        IMarketDataListener *listener);
 
    ~BinancePriceStatisticsAdaptor();

src/app_sparkd/sparkd/md/binance/BinanceSecurityRequestor.cpp

@@ -3,7 +3,6 @@
 #include "domain/protobufs/venue.pb.h"
 #include "domain/security.h"
 #include "domain/security_criteria.h"
-#include "spark/logging/logging.h"
 #include "spark/persistence/IDGenerator.h"
 #include "spark/utils/bts_fatal.h"
 #include "sparkd/md/binance/BinanceMessageType.h"
@@ -14,6 +13,7 @@
 #include "sparkd/md/ccapi/ICCApiMessageHandler.h"
 #include "system/collections/contains.h"
 #include "system/collections/remove.h"
+#include "system/logging/logging.h"
 
 #include <algorithm>
 #include <ccapi_cpp/ccapi_event_dispatcher.h>
@@ -49,6 +49,7 @@ using std::vector;
 #define _WARN(...) _LOG(Warn, __VA_ARGS__)
 #define _ERROR(...) _LOG(Error, __VA_ARGS__)
 
+#define VENUE Venue::VENUE_BINANCE
 #define VENUE_NAME (Venue_Name(Venue::VENUE_BINANCE))
 #define CCAPI_VENUE_NAME CCAPI_EXCHANGE_NAME_BINANCE
 #define BINANCE_MSG_TYPE_NAME(x)                                               \
@@ -156,7 +157,8 @@ struct BinanceSecurityRequestor::PImpl {
       for (auto &sec : added_secs) {
          // Since the ids haven't been set yet, we should expect them to be 0
          bts_assert(sec.id().value() == 0);
-         SecurityID sec_id{sec_id_store->get_or_add(VENUE_NAME, sec.desc())};
+         SecurityID sec_id{
+             sec_id_store->get_or_add(VENUE, sec.type(), sec.desc())};
          sec.security_common.id = sec_id;
       }
    }

src/app_sparkd/sparkd/md/binance/BinanceSpot.cpp

@@ -2,8 +2,8 @@
 #include "domain/protobufs/venue.pb.h"
 #include "domain/quantity.h"
 #include "domain/security.h"
-#include "spark/logging/logging.h"
 #include "system/json/json_reader.h"
+#include "system/logging/logging.h"
 
 #include <ccapi_cpp/ccapi_message.h>
 #include <iostream>

src/app_sparkd/sparkd/md/binance/BinanceSpot.h

@@ -1,13 +1,13 @@
 #pragma once
 
 #include "domain/protobufs/venue.h"
-#include "spark/io/JsonMacros.h"
 #include "sparkd/md/binance/BinanceFilterTuple.h"
 #include "sparkd/md/binance/BinanceInstrumentStatus.h"
 #include "sparkd/md/binance/BinanceOrderType.h"
 #include "sparkd/md/binance/BinancePermission.h"
 #include "sparkd/md/binance/BinanceQuoteAsset.h"
 #include "system/FMT_IMPL.h"
+#include "system/json/json_macros.h"
 
 #include <nlohmann/json.hpp>
 #include <ostream>