build: Clean up client CMake

webapp check pt?
cleanup: Remove empty header FTXSecurity.h
2024-09-03 12:39:02 -04:00 · 2024-09-01 20:07:31 +00:00 · 2024-09-01 20:07:31 +00:00 · 2024-09-01 20:07:31 +00:00 · 2024-09-01 20:07:31 +00:00 · 2024-09-01 20:07:31 +00:00
773 changed files with 26361 additions and 17864 deletions
@@ -87,10 +87,6 @@ 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 \
@@ -234,7 +234,7 @@
        "program": "${workspaceFolder}/build/sparkd-debug-clang/src/app_sparkd/sparkd",
      },
      "args": [
-        "run"
+        "test-py-md-provider"
      ],
      "cwd": "${workspaceFolder}/build/sparkd-debug-clang/src/app_sparkd",
      "stopAtEntry": false,
@@ -1,5 +1,5 @@
 cmake_minimum_required(VERSION 3.12)
-project(spark)
+project(spark_root)
 set(CMAKE_CXX_STANDARD 20)

 if(WIN32)
@@ -105,7 +105,6 @@ 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)
@@ -53,7 +53,6 @@
        "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",
@@ -77,7 +76,7 @@
          "value": "ON"
        },
        "CCAPI_LOG_LEVEL": "Info",
-        "CONFIG_DEFAULTS_PATH": "spark/config/profiles/md_dev/*.conf;spark/config/profiles/default/*.conf"
+        "CONFIG_DEFAULTS_PATH": "config/profiles/md_dev/*.conf;config/profiles/default/*.conf"
      }
    },

@@ -88,7 +87,7 @@
      "cacheVariables": {
        "CMAKE_BUILD_TYPE": "Release",
        "CCAPI_LOG_LEVEL": "Warn",
-        "CONFIG_DEFAULTS_PATH": "spark/config/profiles/md_prod/*.conf;spark/config/profiles/default/*.conf"
+        "CONFIG_DEFAULTS_PATH": "config/profiles/md_prod/*.conf;config/profiles/default/*.conf"
      }
    },

@@ -1,5 +1,3 @@
-include_guard(GLOBAL)
-
 macro(export_to_parent)
  foreach(var IN ITEMS ${ARGN})
    set(${var} ${${var}} PARENT_SCOPE)
@@ -1,133 +1,66 @@
 include_guard(DIRECTORY)
 include(${CMAKE_SOURCE_DIR}/cmake/ExportToParent.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)
+include(${CMAKE_SOURCE_DIR}/cmake/TernaryOperator.cmake)

 function(init_library)
  set(_options SHARED STATIC)
-  set(_one_value_args
-    NAME
-    SOURCE_DIR
-    INCLUDE_DIR
-  )
+  set(_one_value_args SOURCE_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)

-  # Cpp sources
+  set_ternary(_source_dir _ARGS_SOURCE_DIR _ARGS_SOURCE_DIR ${PROJECT_NAME})
+  # Glob cpp sources
  file(
-    GLOB_RECURSE ${_name}_SOURCES
+    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 ${${_name}_SOURCES})
-    if(${item} MATCHES "tests/.*\.cpp$")
-      list(APPEND ${_name}_TEST_SOURCES ${item})
+  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()
-  foreach(item ${${_name}_TEST_SOURCES})
-    list(REMOVE_ITEM ${_name}_SOURCES ${item})
-  endforeach()
-
-  # 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}
-  )

  export_to_parent(
-    ${_name}_SOURCES
-    ${_name}_TEST_SOURCES
-    ${_name}_PROTO_SOURCES
-    ${_name}_CXX_PROTO_SOURCES
-    ${_name}_CXX_PROTO_HEADERS
+    ${PROJECT_NAME}_SOURCES
+    ${PROJECT_NAME}_TEST_SOURCES
  )

-  # Add main lib
+  # Add main library
  if (_ARGS_STATIC)
-    add_library(${_name} STATIC ${${_name}_SOURCES} ${${_name}_CXX_PROTO_SOURCES})
+    add_library(${PROJECT_NAME} STATIC ${${PROJECT_NAME}_SOURCES})
  elseif(_ARGS_SHARED)
-    add_library(${_name} SHARED ${${_name}_SOURCES} ${${_name}_CXX_PROTO_SOURCES})
+    add_library(${PROJECT_NAME} SHARED ${${PROJECT_NAME}_SOURCES})
  else()
-    add_library(${_name} ${${_name}_SOURCES} ${${_name}_CXX_PROTO_SOURCES})
+    add_library(${PROJECT_NAME} ${${PROJECT_NAME}_SOURCES})
  endif()
+
  target_include_directories(
-    ${_name}
+    ${PROJECT_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(
-    ${_name}
+    ${PROJECT_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 (${_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})
-
+  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(${_name}-tests PRIVATE Catch2::Catch2WithMain)
+    target_link_libraries(${PROJECT_NAME}-tests PRIVATE Catch2::Catch2WithMain)
  endif()
 endfunction()
@@ -0,0 +1,55 @@
+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)
@@ -1,39 +0,0 @@
-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()
@@ -0,0 +1,30 @@
+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()
@@ -1,14 +1,6 @@
-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)
@@ -25,20 +17,20 @@ function(pack_basic)
  endif()

  write_basic_package_version_file(
-    "${CMAKE_CURRENT_BINARY_DIR}/${_name}ConfigVersion.cmake"
+    "${CMAKE_CURRENT_BINARY_DIR}/${PROJECT_NAME}ConfigVersion.cmake"
    VERSION ${PROJECT_VERSION}
    COMPATIBILITY AnyNewerVersion
  )

  configure_package_config_file(
    "${CMAKE_CURRENT_SOURCE_DIR}/Config.cmake.in"
-    "${CMAKE_CURRENT_BINARY_DIR}/${_name}Config.cmake"
-    INSTALL_DESTINATION ${CMAKE_INSTALL_LIBDIR}/cmake/${_name}
+    "${CMAKE_CURRENT_BINARY_DIR}/${PROJECT_NAME}Config.cmake"
+    INSTALL_DESTINATION ${CMAKE_INSTALL_LIBDIR}/cmake/${PROJECT_NAME}
  )

  install(
    FILES
-      "${CMAKE_CURRENT_BINARY_DIR}/${_name}Config.cmake"
-      "${CMAKE_CURRENT_BINARY_DIR}/${_name}ConfigVersion.cmake"
-    DESTINATION ${CMAKE_INSTALL_LIBDIR}/cmake/${_name})
+      "${CMAKE_CURRENT_BINARY_DIR}/${PROJECT_NAME}Config.cmake"
+      "${CMAKE_CURRENT_BINARY_DIR}/${PROJECT_NAME}ConfigVersion.cmake"
+    DESTINATION ${CMAKE_INSTALL_LIBDIR}/cmake/${PROJECT_NAME})
 endfunction()
@@ -2,16 +2,8 @@ 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)
@@ -34,7 +26,6 @@ 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"
@@ -42,7 +33,7 @@ function(protobuf_generate_cpp)
      COMMAND
        ${_ARGS_PROTOC}
      ARGS
-        --cpp_out=${_ARGS_CXX_OUT_DIR} ${_proto_include_cli_args} ${_fil}
+        --cpp_out=${_ARGS_CXX_OUT_DIR} ${_proto_include_cli_args} ${_abs_fil}
      DEPENDS ${_abs_fil}
      COMMENT "Running C++ protocol buffer compiler on ${_fil}"
      VERBATIM
@@ -1,58 +0,0 @@
-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()
@@ -1,10 +0,0 @@
-include_guard(GLOBAL)
-
-macro(set_first_defined RESULT)
-  foreach(_arg ${ARGN})
-    if (DEFINED ${_arg})
-      set(${RESULT} ${${_arg}})
-      break()
-    endif()
-  endforeach()
-endmacro()
@@ -1,11 +0,0 @@
-#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
@@ -24,6 +24,7 @@ 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" \
@@ -1,11 +1,7 @@
 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 "${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"
+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"
 WORKDIR "/home/bts/bin"
@@ -0,0 +1,6 @@
+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"
@@ -0,0 +1,49 @@
+// 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);
+//            }
+
+//       }
+//    }
+// }
@@ -1,7 +1,6 @@
 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 {

@@ -5,11 +5,10 @@ using System.Text;
 using System.IO;
 using Google.Protobuf;
 using System.Globalization;
-using BCG;
+using bts;

-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)
+namespace BCG {
+   // Writes all the udpates to a stream in a file.
   // File naming:
   // token_epoch_yyyyMMdd_HHmmss
   // File location: directory(passed)\token\
@@ -95,10 +94,10 @@ namespace bts {
      void _on_change(object new_value) {
         DataPacket dp = new_value as DataPacket;
         byte[] serialized_update = null;
-         bool first_write = _serializer == null || _n_updates_in_current_file == 0;
+         bool _write_create = _serializer == null || _n_updates_in_current_file == 0;
         _serializer = _serializer ?? new UniversalStreamIOSerializer();
-         if (first_write) {
-            // first file always starts with create
+         if (_write_create) {
+            // new files 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)));
@@ -106,8 +105,8 @@ namespace bts {
         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);
@@ -0,0 +1,190 @@
+// 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
+//    }
+// }
@@ -0,0 +1,68 @@
+// 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);
+//       }
+//    }
+// }
@@ -5,8 +5,7 @@ using System.IO;
 using BCG;

 namespace bts {
-   // 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
+   // this saves the total state in the filr provided in the cons
   [DataStreamUserRespresentation("saver")]
   public class SaverStream : ViewStream {
      private ViewStream _stream;
@@ -32,6 +31,7 @@ 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);
+
      }
   }
 }
@@ -3,7 +3,6 @@ using System.Collections.Generic;
 using System.Linq;
 using System.Globalization;
 using System.IO;
-using bts;


 namespace BCG {
@@ -1,8 +1,5 @@
 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 \
@@ -11,6 +8,6 @@ RUN apt-get update \
    && apt-get autoremove -yqq --purge wget && rm -rf /var/lib/apt/lists/*

 USER bts
-COPY --chown=bts:bts "${SOURCE_DIR}" "bin"
-COPY --chown=bts:bts "${CONF_FILE}" ".local/share/bts/DataStreamPublisher/conf/spark.conf"
+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"
 WORKDIR "/home/bts/bin"
@@ -37,17 +37,14 @@ 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;
@@ -59,45 +56,38 @@ 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;
@@ -115,31 +105,25 @@ 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;
@@ -148,19 +132,16 @@ 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_;
@@ -168,8 +149,7 @@ 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}";
@@ -195,15 +175,13 @@ 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;
@@ -218,13 +196,10 @@ 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;
@@ -234,30 +209,26 @@ 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;
@@ -266,18 +237,15 @@ 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; }
      }

@@ -290,14 +258,12 @@ 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}";
      }

@@ -305,13 +271,10 @@ 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++;
@@ -319,21 +282,16 @@ 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);
      }

@@ -346,16 +304,13 @@ 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"));
@@ -363,31 +318,23 @@ 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"));
            }
         }
@@ -398,10 +345,8 @@ 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;
@@ -413,8 +358,7 @@ 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);
@@ -433,34 +377,28 @@ 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,
@@ -468,35 +406,28 @@ 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)
-      {
-         // AsyncReceiveMessage rcv_message = res.AsyncState as AsyncReceiveMessage;
+      private void _log_incoming_msg(IAsyncResult res) {
+         // var rcv_message = ExceptionAssert.Cast(res.AsyncState).To<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);
      }

@@ -512,15 +443,13 @@ 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);
@@ -528,32 +457,23 @@ 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;
@@ -561,28 +481,21 @@ 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);
@@ -590,28 +503,23 @@ 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}");
         }
@@ -620,8 +528,7 @@ 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);
@@ -630,8 +537,7 @@ 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}");
         }
@@ -642,19 +548,15 @@ 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) {
@@ -672,10 +574,8 @@ 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);
@@ -286,4 +286,4 @@ namespace BCG {



-}
+}
@@ -1,8 +1,5 @@
 FROM bluetradingsystems/linux_dotnet7

-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"
+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"
 WORKDIR "/home/bts/bin"
@@ -0,0 +1,33 @@
+#!/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"
+
@@ -17,7 +17,7 @@ type Tab =
  | PositionManagementTab
  | MarginTab

-let venues = [| Venue.BINANCE; Venue.BinanceUs; Venue.BinanceUsdsFutures; Venue.CBOE; Venue.COINBASE; Venue.DERIBIT; Venue.GLBX; Venue.ICE; Venue.TAIFEX; Venue.SIM; |]
+let venues = [| Venue.BINANCE; Venue.BinanceUs; Venue.BinanceUsdsFutures; Venue.COINBASE; Venue.DERIBIT; Venue.GLBX; Venue.ICE; Venue.SIM; |]

 let securityTypes = [| SecurityType.SPOT; SecurityType.FUTURE; SecurityType.OPTION; SecurityType.SPREAD; |]

@@ -69,11 +69,6 @@ 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
@@ -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
@@ -1,10 +1,8 @@
-cmake_minimum_required(VERSION 3.14)
+cmake_minimum_required(VERSION 3.5)
 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)
@@ -59,14 +57,6 @@ 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)
@@ -140,8 +130,9 @@ 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
@@ -157,11 +148,8 @@ target_link_libraries(
  -static-libstdc++
  -pthread)

-# Catch2
-find_package(Catch2 CONFIG REQUIRED)
-target_link_libraries(${PROJECT_NAME} Catch2::Catch2)
-
-install_basic()
+include(${CMAKE_SOURCE_DIR}/cmake/InstallBasicConfig.cmake)
+install_basic_config()
 install(
  FILES
    "${EXTERNALS_LIB}/libOnixS.IceImpactMulticastPriceFeedHandlerCpp.so.3.16.0.0"
@@ -1,19 +0,0 @@
-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"
@@ -0,0 +1,10 @@
+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"
@@ -1,5 +1,4 @@
 #include "AlgoDefinitionServiceServer.h"
-#include "spark/spark_data_directory.h"

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

 AlgoDefinitionServiceServer::AlgoDefinitionServiceServer(
@@ -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 "system/logging/logging.h"
+#include <memory>

 #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
@@ -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, rowid, existing_val] = store->retrieve_one(ns, key);
+   auto [exists, 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,8 +384,7 @@ 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, rowid, val] =
-       me->reports_store->retrieve_one(me->reports_ns, key);
+   auto [present, 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);
@@ -407,8 +406,7 @@ 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, rowid, val] =
-       me->reports_store->retrieve_one(me->reports_ns, key);
+   auto [present, val] = me->reports_store->retrieve_one(me->reports_ns, key);
   api::ResultMessage res;
   res.set_success(false);
   if (!present) {
@@ -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<logging::protobufs::Logging>();
+   auto cfg = config::get<config::Logging>();
   cfg.mutable_console()->set_default_level("Fatal");
-   utils::SparkLibInit spark_lib{cfg, false, true};
+   log::configure_logging(cfg);

   std::map<api::CMEMics, std::string> mic_map;
   mic_map[api::CMEMics::CME] = "XCME";
@@ -1,14 +1,13 @@
 #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>
@@ -2,6 +2,8 @@

 #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"
@@ -11,8 +13,6 @@
 #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>
@@ -1,5 +1,5 @@
 #include "QuickTest.h"
-#include "system/logging/logging.h"
+#include "spark/logging/logging.h"

 #include <thread>

@@ -1,32 +1,31 @@
 #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/query_server.h"
+#include "spark/md/relay/QueryServer.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"
@@ -37,7 +36,7 @@
 #include "sparkd/support/ShellLatencyServer.h"
 #include "sparkd/user/AdminServiceServer.h"
 #include "sparkd/user/AuthServiceServer.h"
-#include "system/logging/logging.h"
+

 namespace bts::cmds {

@@ -50,7 +49,8 @@ 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,10 +63,7 @@ static void execute() {

   bts::md::MBOProcessor mbo_processor;
   bts::md::TradeSummarySignalDispatcher tss_dispatcher;
-   bts::md::MarketDataGateway md_gateway{
-       config::get<api::md::MarketDataConfig>(),
-       persistence::IDGenerator::instance_ptr()};
-   md_gateway.start();
+   md::cmds::Run::run();

   api::synthetic::SyntheticSecurityService synthetic_service(sc);
   bts::md::MDClient mdclient(
@@ -1,7 +1,7 @@
 #include "TestPythonMdProviderCmd.h"
 #include "python_md/md_sink_impl.h"
 #include "python_md/py_md_executor.h"
-#include "system/logging/logging.h"
+#include "spark/logging/logging.h"

 #include <pybind11/embed.h>
 #include <pybind11/pybind11.h>
@@ -38,14 +38,10 @@ static void execute(
      py_md_exec.start();
      LOG(md, Info, "Stopping py md executor.");
      py_md_exec.stop();
-      // 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);
+      int i = 0;
+      while (true) {
+         std::cout << "Sleep loop count " << ++i << std::endl;
+         std::this_thread::sleep_for(3s);
      }
   } catch (const std::exception &ex) {
      LOG(md, Error, "Py md executor threw: {}", string{ex.what()});
@@ -62,7 +58,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 = "dummy_md_provider.DummyMdProvider";
+         args_->py_md_provider = "md_provider_impl.MdProviderImpl";
      }
      if (args_->module_paths.empty()) {
         args_->module_paths.push_back(".");
@@ -1,22 +1,30 @@
 #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/query_server.h"
+#include "spark/md/relay/QueryServer.h"
 #include "spark/metrics/metrics.h"
 #include "spark/metrics/system_metrics.h"
-#include "spark/quickbus/core/Quickbus.h"
+#include "spark/quickbus/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 {

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

@@ -9,16 +7,13 @@

 namespace bts::cmds {

-using bts::api::md::MarketDataConfig;
-using bts::persistence::IDGenerator;
-
-LiveLadderDemo::LiveLadderDemo(const MarketDataConfig &config) {
+LiveLadderDemo::LiveLadderDemo(const api::md::MarketDataConfig &) {
   char ssl_cert_file[] = "SSL_CERT_FILE=/etc/ssl/cert/ca-certificates.crt";
   putenv(ssl_cert_file);

-   md::MarketDataGateway gw{config, IDGenerator::instance_ptr()};
+   md::MarketDataGateway gw;
   md::listeners::LiveLadderPrinter ladder_printer;
-   gw.start();
+   gw.run();

   const std::size_t n_loops = 16;
   for (std::size_t n = 0; n < n_loops; n++) {
@@ -3,10 +3,9 @@
 #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>

@@ -1,30 +1,11 @@
 #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([] {
-      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, {});
+      auto channels = bts::md::get_configured_cme_channels();
      for (const auto &ch : channels) {
         std::cout << ch.id << " - ";
         std::set<std::string> symbols;
@@ -1,10 +1,9 @@
 #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>

@@ -1,6 +1,5 @@
 #include "Record.h"
-#include "spark/md/relay/relay_recorder.h"
-
+#include "spark/md/relay/RelayRecorder.h"
 #include <boost/filesystem.hpp>

 namespace bts::cmds::md::cmds {
@@ -9,14 +8,13 @@ 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());
   });
 }

@@ -1,5 +1,5 @@
 #include "Relay.h"
-#include "spark/md/relay/relay_control.h"
+#include "spark/md/relay/RelayClient.h"

 namespace bts::cmds::md::cmds {

@@ -1,11 +1,10 @@
 #include "Replay.h"
-#include "spark/config/config.h"
-#include "spark/config/protos/market_data.pb.h"
-#include "spark/md/relay/relay_recorder.h"
+#include "spark/md/relay/RelayRecorder.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>();
@@ -1,19 +1,25 @@
 #include "Run.h"
 #include "Cleanup.h"
-#include "spark/api/md/MarketDataConfig.pb.h"
+#include "domain/security_criteria.h"
 #include "spark/config/protos/market_data.pb.h"
-#include "spark/persistence/IDGenerator.h"
-#include "spark/utils/datetime.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/utils/thread_manager.h"
 #include "sparkd/md/MarketDataGateway.h"
-#include "system/logging/logging.h"
-#include "system/os.h"
-
+#include "sparkd/md/platforms/IMDPlatform.h"
+#include "sparkd/md/platforms/ice_onixs/ICEMain.h"
+#include "sparkd/md/platforms/mdp3_onixs/MDP3Main.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();
@@ -37,17 +43,57 @@ 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{
-       config::get<api::md::MarketDataConfig>(),
-       persistence::IDGenerator::instance_ptr()};
-   gwy.start();
-   while (true) {
-      using namespace std::chrono_literals;
-      std::this_thread::sleep_for(2s);
-   }
+   bts::md::MarketDataGateway gwy = {};
+   // this keeps going in a loop
+   gwy.run();
 }

 void Run::run() {
@@ -62,8 +108,13 @@ void Run::run() {
 Run::Run()
    : CLI::App(
          "Start market data feed, and store up-to-date snapshots of mbo, and "
-          "mbp in shared memory.",
+          "mbp in shared memory.  If no subcommand specified, uses the "
+          "configured default feed (replay, or onixs)",
          "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(); });
 }

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

   /**
@@ -1,5 +1,5 @@
 #include "Print.h"
-#include "spark/md/relay/relay_recorder.h"
+#include "spark/md/relay/RelayRecorder.h"

 namespace bts::cmds::md::cmds::rec_tool::cmds {

@@ -1,5 +1,5 @@
 #include "Trace.h"
-#include "spark/md/relay/relay_recorder.h"
+#include "spark/md/relay/RelayRecorder.h"

 namespace bts::cmds::md::cmds::rec_tool::cmds {

@@ -1,5 +1,5 @@
 #include "Unprint.h"
-#include "spark/md/relay/relay_recorder.h"
+#include "spark/md/relay/RelayRecorder.h"

 namespace bts::cmds::md::cmds::rec_tool::cmds {

@@ -18,9 +18,7 @@ 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_CBOE),
-       NAME_PAIR(api::VENUE_TAIFEX)};
+       NAME_PAIR(api::VENUE_BINANCE_USDS_FUTURES)};
 }

 std::string VenueOption::name(api::Venue venue) {
@@ -47,10 +45,6 @@ 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));
   }
@@ -6,29 +6,31 @@
 #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 "system/logging/logging.h"
-#include "system/os.h"
-#include "system/protobufs/config/system_config.pb.h"
+#include "spark/os/os.h"

 #include "spark/domain_model/Security.pb.h"
 #include "spark/md/IntPrice.h"
 #include "spark/md/MBOProcessor.h"
-#include "spark/md/relay/relay_recorder.h"
+#include "spark/md/relay/RelayClient.h"
+#include "spark/md/relay/RelayRecorder.h"
+#include "spark/utils/bts_chrono.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"
@@ -55,7 +57,7 @@ md::PriceSize coerce_precision(const co2::PriceSizeMessage &p) {
 struct MDPlayer : md::IRelayWriter {
   MDPlayer(
       IMDPlayerObserver &obs,
-       std::map<uint32_t, api::Security> &known_securities)
+       std::map<uint32_t, md::Security> &known_securities)
       : obs(obs), known_securities(known_securities),
         forward_implied(config::get<config::ExchangeSim>().forward_implied()) {
   }
@@ -128,7 +130,7 @@ struct MDPlayer : md::IRelayWriter {
         }
   }
   IMDPlayerObserver &obs;
-   std::map<uint32_t, api::Security> &known_securities;
+   std::map<uint32_t, md::Security> &known_securities;
   bool forward_implied;
 };

@@ -142,7 +144,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<SystemConfig>().data_path());
+          config::get<config::SystemConfig>().data_path());
      boost::replace_all(str_settings, "$SIM_VENUE", venue_string);
      std::string port_string;
      switch (venue) {
@@ -170,7 +172,7 @@ int start_exchange_sim(bts::api::Venue venue, md::MBOWriter *mbo) {

      acceptor.start();

-      std::map<uint32_t, api::Security> known_securities;
+      std::map<uint32_t, md::Security> known_securities;
      const auto &conf = config::get<config::ExchangeSim>();
      while (true) {
         MDPlayer player(application, known_securities);
@@ -5,9 +5,11 @@
 #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/fixed_security.h"
+#include "spark/md/Shm/FixedSecurity.h"
+#include "spark/md/Shm/MDStructure.h"
 #include "spark/orderman/orderman_utils/fix_order_codec.h"
 #include "spark/persistence/SqliteStore.h"
 #include "spark/utils/Dispatcher.h"
@@ -15,7 +17,6 @@
 #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>
@@ -582,11 +583,10 @@ void negate_id(api::Security *sec) {
 }

 void ExchangeSimApplication::persist_book(OrderBook const &b) {
-   auto [exists, rowid, oldval] =
-       me->store->retrieve_one("books", b.sec_desc());
+   auto existing = me->store->retrieve_one("books", b.sec_desc());
   auto val = utils::pb::to_string(b);

-   if (exists)
+   if (existing.first)
      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(
-    api::Security const &sec, md::IntPrice price) {
+    md::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,8 +629,7 @@ void ExchangeSimApplication::on_settlement(
   }
 }

-void ExchangeSimApplication::on_high(
-    api::Security const &sec, md::Price price) {
+void ExchangeSimApplication::on_high(md::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()) {
@@ -639,7 +638,7 @@ void ExchangeSimApplication::on_high(
   }
 }

-void ExchangeSimApplication::on_low(api::Security const &sec, md::Price price) {
+void ExchangeSimApplication::on_low(md::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()) {
@@ -648,8 +647,7 @@ void ExchangeSimApplication::on_low(api::Security const &sec, md::Price price) {
   }
 }

-void ExchangeSimApplication::on_open(
-    api::Security const &sec, md::Price price) {
+void ExchangeSimApplication::on_open(md::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()) {
@@ -659,7 +657,7 @@ void ExchangeSimApplication::on_open(
 }

 void ExchangeSimApplication::on_close(
-    api::Security const &sec, md::Price price) {
+    md::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()) {
@@ -669,7 +667,7 @@ void ExchangeSimApplication::on_close(
 }

 void ExchangeSimApplication::on_depth(
-    api::Security const &sec,
+    md::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,
@@ -69,8 +69,7 @@ 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,
@@ -118,14 +117,14 @@ class ExchangeSimApplication : public sim::IMDPlayerObserver,
   void update_md(std::string const &sec_desc);
   void persist_book(OrderBook const &b);

-   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_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_depth(
-       api::Security const &s,
+       md::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,
@@ -1,7 +1,7 @@
 #pragma once

 #include "spark/md/PriceSize.h"
-#include "spark/md/shm/hot_depth_entry.h"
+#include "spark/md/Shm/MDStructure.h"

 namespace bts {
 namespace api {
@@ -10,14 +10,14 @@ class Security;
 namespace sim {

 struct IMDPlayerObserver {
-   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_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_depth(
-       api::Security const &s,
+       md::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,
@@ -1,6 +1,5 @@
 #include "sparkd/exchange_sim/book/book_transform.h"
-
-#include <catch2/catch_all.hpp>
+#include <catch/catch.hpp>
 #include <stdlib.h>

 namespace bts {
@@ -5,6 +5,7 @@
 #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"
@@ -19,14 +20,13 @@
 #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"
@@ -4,9 +4,8 @@
 #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>
@@ -1,6 +1,5 @@
 #include "MultipartFormDecoder.h"
-#include "system/logging/logging.h"
-
+#include "spark/logging/logging.h"
 #include <boost/algorithm/string.hpp>
 #include <boost/algorithm/string/iter_find.hpp>

@@ -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>
@@ -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
@@ -1,11 +1,9 @@
 #pragma once

-#include <memory>
+#include "spark/md/IMarketDataListener.h"

 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.
@@ -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 {
@@ -2,11 +2,14 @@
 #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"
@@ -15,14 +18,12 @@
 #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>

@@ -36,6 +37,8 @@ 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()) {
@@ -47,51 +50,40 @@ static bool validate(const MarketDataConfig &cfg, vector<string> &errs) {

 struct MarketDataGateway::PImpl {
   struct MdSink {
-      shared_ptr<GeneralSecurityIndex> sec_index;
-      shared_ptr<MDProducer> md_producer;
+      GeneralSecurityIndex sec_index;
+      MDProducer md_producer;
      shared_ptr<IMarketDataListener> listener;

-      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)) {}
+      MdSink(api::Venue venue, IDGenerator *sec_id_store)
+          : sec_index(), listener(std::make_shared<GeneralMarketDataListener>(
+                             sec_index, md_producer, venue, sec_id_store)) {}
   };

-   MarketDataConfig config;
-   shared_ptr<IDGenerator> sec_id_store;
+   MarketDataConfig config = config::get<MarketDataConfig>();
+   IDGenerator &sec_id_store = IDGenerator::instance();
   vector<unique_ptr<IMarketDataProvider>> providers;
   vector<unique_ptr<MdSink>> sinks;
   vector<unique_ptr<IMarketDataSource>> sources;

-   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));
+   PImpl() {
+      providers.emplace_back(std::make_unique<CCApiMarketDataProvider>(config));
      providers.emplace_back(
-          std::make_unique<EmumbaMarketDataProvider>(this->config));
+          std::make_unique<EmumbaMarketDataProvider>(config));
+      providers.emplace_back(std::make_unique<OnixsMarketDataProvider>(config));
      providers.emplace_back(
-          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));
+          std::make_unique<PyMarketDataProvider>(config, &sec_id_store));
+      providers.emplace_back(
+          std::make_unique<ReplayMarketDataProvider>(config, &sec_id_store));
   }
 };

-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()
+    : me_(std::make_unique<MarketDataGateway::PImpl>()) {}

 MarketDataGateway::~MarketDataGateway() = default;

 // starts the market data sources and then keeps the function going
-void MarketDataGateway::start() {
+void MarketDataGateway::run() {
   vector<string> cfg_errs;
   if (!validate(me_->config, cfg_errs)) {
      BTS_FATAL_MSGF("Invalid MarketDataConfig:", FMT_JOIN(cfg_errs, "\n"));
@@ -103,10 +95,15 @@ void MarketDataGateway::start() {
      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
@@ -2,14 +2,6 @@

 #include <memory>

-namespace bts::api::md {
-class MarketDataConfig;
-}
-
-namespace bts::persistence {
-class IDGenerator;
-}
-
 namespace bts::md {

 /**
@@ -18,12 +10,10 @@ namespace bts::md {
 */
 class MarketDataGateway {
 public:
-   explicit MarketDataGateway(
-       api::md::MarketDataConfig config,
-       std::shared_ptr<persistence::IDGenerator> sec_id_store);
+   MarketDataGateway();
   ~MarketDataGateway();

-   void start();
+   void run();

 private:
   struct PImpl;
@@ -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, std::optional<utils::TimeStamp>) override;
+   void on_book(const domain::Book &book) 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*/, std::optional<utils::TimeStamp>) {}
+inline void MarketDataListener::on_book(const domain::Book & /*book*/) {}

 inline void MarketDataListener::on_trades(
    const std::vector<const domain::TradeRecord *> & /*trades*/) {}
@@ -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 =
-    utils::ValueWrapper<struct MarketDataListenedIDTag, std::uint32_t>;
+    core::ValueWrapper<struct MarketDataListenedIDTag, std::uint32_t>;

 } // namespace bts::md

@@ -11,20 +11,23 @@
 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;
-   std::shared_ptr<IDGenerator> sec_id_store;
-   vector<shared_ptr<OnixsGlbxSubscription>> sources;
+   IDGenerator &sec_id_store = IDGenerator::instance();
+   shared_ptr<OnixsGlbxSubscription> source;
 };

+OnixsMarketDataProvider::OnixsMarketDataProvider()
+    : bts::md::OnixsMarketDataProvider(bts::config::get<MarketDataConfig>()) {}
+
 OnixsMarketDataProvider::OnixsMarketDataProvider(MarketDataConfig config)
    : me_(std::make_unique<PImpl>()) {
   me_->config = std::move(config);
@@ -32,14 +35,12 @@ OnixsMarketDataProvider::OnixsMarketDataProvider(MarketDataConfig config)

 OnixsMarketDataProvider::~OnixsMarketDataProvider() = default;

-vector<pair<Venue, shared_ptr<IMarketDataSource>>>
+vector<pair<domain::protobufs::Venue, shared_ptr<IMarketDataSource>>>
 OnixsMarketDataProvider::get_sources() {
-   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()});
-   }
+   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});
   return rv;
 }

@@ -1,15 +1,13 @@
 #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;
@@ -20,17 +18,16 @@ using std::vector;
 using Path = boost::filesystem::path;

 struct PyMarketDataProvider::PImpl {
-   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)) {}
+   api::md::MarketDataConfig config;
+   IDGenerator *sec_id_store;
+   vector<shared_ptr<PyMarketDataSubscription>> py_md_subs = {};
 };

 PyMarketDataProvider::PyMarketDataProvider(
-    MarketDataConfig config, shared_ptr<IDGenerator> sec_id_store)
-    : me_(std::make_unique<PImpl>(std::move(config), std::move(sec_id_store))) {
+    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;
 }

 PyMarketDataProvider::~PyMarketDataProvider() = default;
@@ -41,7 +38,7 @@ PyMarketDataProvider::get_sources() {
   const Path wd{sparkd_wd()};
   vector<string> module_paths{
       wd.string(),
-       (wd / "python_md_providers").string(),
+       (wd / "py_market_data_providers").string(),
   };
   for (const auto &py_plugin : me_->config.python_md_providers()) {
      vector<string> module_path;
@@ -12,7 +12,7 @@ class PyMarketDataProvider : public IMarketDataProvider {
 public:
   explicit PyMarketDataProvider(
       api::md::MarketDataConfig config,
-       std::shared_ptr<persistence::IDGenerator> sec_id_store);
+       persistence::IDGenerator *sec_id_store);
   ~PyMarketDataProvider();

   std::vector<
@@ -2,8 +2,6 @@
 #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"
@@ -15,11 +13,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>
@@ -55,13 +53,11 @@ struct TradeTuple {

 struct PyMarketDataSubscription::PImpl : private py_interop::MdSink {
   PImpl(
-       string py_plugin,
-       vector<string> module_paths,
-       std::shared_ptr<IDGenerator> sec_id_store)
+       string py_plugin, vector<string> module_paths, 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_(std::move(sec_id_store)) {}
+         sec_id_store_(sec_id_store) {}

   void start(shared_ptr<IMarketDataListener> listener) {
      bts_assert(listener != nullptr);
@@ -79,7 +75,7 @@ struct PyMarketDataSubscription::PImpl : private py_interop::MdSink {
   PyMdExecutor py_md_exec_;
   shared_ptr<IMarketDataListener> listener_;

-   std::shared_ptr<IDGenerator> sec_id_store_;
+   IDGenerator *sec_id_store_;
   // The main store for securities.
   vector<domain::Security> securities_;
   vector<const domain::Security *> security_ptrs_;
@@ -147,19 +143,13 @@ struct PyMarketDataSubscription::PImpl : private py_interop::MdSink {
      trade_ptrs_.clear();
   }

-   SecurityID get_sec_id_(
-       const string &venue_name,
-       const string &type_name,
-       const string &sec_name) {
+   SecurityID get_sec_id_(const string &venue_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));
-      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);
+      auto sec_id = sec_id_store_->get_id(venue, sec_name, true);
      bts_assert_msgf(
          sec_id, "Failed to get security id for py_book. venue: {}; name: {}",
          venue, sec_name);
@@ -178,8 +168,7 @@ 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.type(), py_sec.exchange_id());
+      sec_comm.id = get_sec_id_(py_sec.venue(), 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();
@@ -214,9 +203,7 @@ 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.security_type(),
-          source.exchange_security_id());
+      dest.sec_id = get_sec_id_(source.venue(), 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')
@@ -242,20 +229,15 @@ 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.security_type(),
-          source.exchange_security_id()));
+      dest.set_security_id(
+          get_sec_id_(source.venue(), source.exchange_security_id()));
   }
 };

 PyMarketDataSubscription::PyMarketDataSubscription(
-    string py_plugin,
-    vector<string> module_paths,
-    shared_ptr<IDGenerator> sec_id_store)
+    string py_plugin, vector<string> module_paths, IDGenerator *sec_id_store)
    : me_(std::make_unique<PyMarketDataSubscription::PImpl>(
-          std::move(py_plugin),
-          std::move(module_paths),
-          std::move(sec_id_store))) {}
+          std::move(py_plugin), std::move(module_paths), sec_id_store)) {}

 PyMarketDataSubscription::~PyMarketDataSubscription() { me_->stop(); }

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

   void start(std::shared_ptr<IMarketDataListener> listener) override;
@@ -10,7 +10,6 @@
 namespace bts::md {

 using bts::api::md::MarketDataConfig;
-using bts::domain::DomainSecurityUtils;
 using bts::md::recording::MDReplayer;
 using bts::persistence::IDGenerator;
 using std::pair;
@@ -20,21 +19,20 @@ using std::vector;

 struct ReplayMarketDataProvider::PImpl {
   MarketDataConfig config;
-   shared_ptr<IDGenerator> sec_id_store;
+   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(
-          config::get<MarketDataConfig>(), IDGenerator::instance_ptr()) {}
+          bts::config::get<MarketDataConfig>(), &IDGenerator::instance()) {}

 ReplayMarketDataProvider::ReplayMarketDataProvider(
-    MarketDataConfig config, shared_ptr<IDGenerator> sec_id_store)
-    : me_(std::make_unique<PImpl>(std::move(config), std::move(sec_id_store))) {
+    MarketDataConfig config, IDGenerator *sec_id_store)
+    : me_(std::make_unique<PImpl>()) {
+   me_->config = std::move(config);
+   me_->sec_id_store = sec_id_store;
 }

 ReplayMarketDataProvider::~ReplayMarketDataProvider() = default;
@@ -44,7 +42,7 @@ ReplayMarketDataProvider::get_sources() {
   const auto replay_cfgs =
       MarketDataConfigUtils::get_effective_replay_config(me_->config);
   for (const auto &r : replay_cfgs) {
-      const auto v = DomainSecurityUtils::convert(r.venue());
+      const auto v = domain::DomainSecurityUtils::convert(r.venue());
      me_->sources.push_back({v, std::make_shared<MDReplayer>(r.venue())});
   }
   return me_->sources;
@@ -13,7 +13,7 @@ class ReplayMarketDataProvider : public IMarketDataProvider {
   ReplayMarketDataProvider();
   explicit ReplayMarketDataProvider(
       api::md::MarketDataConfig config,
-       std::shared_ptr<persistence::IDGenerator> sec_id_store);
+       persistence::IDGenerator *sec_id_store);
   ~ReplayMarketDataProvider();

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

   if (!jreader.read("result", ticker_upds)) {
      LOG(md, Error,
-          "BinanceMiniTickerUpdate json field 'result' is invalid:\n{}",
-          js.dump());
+          "BinanceMiniTickerUpdate json field 'result' is invalid:\n{}", js);
   }
   return !ticker_upds.empty();
 }
@@ -1,6 +1,6 @@
 #pragma once

-#include "domain/security_market_data_statistics.h"
+#include "spark/domain/SecurityMarketDataStatistics.h"
 #include "sparkd/md/binance/BinanceMiniTickerEventType.h"
 #include "system/time_stamp.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>
@@ -1,7 +1,7 @@
 #include "BinancePriceStatistics.h"
 #include "domain/price.h"
 #include "domain/quantity.h"
-#include "domain/security_market_data_statistics.h"
+#include "spark/domain/SecurityMarketDataStatistics.h"
 #include "sparkd/md/ccapi/CCApiReader.h"
 #include "system/json/json_dumper.h"
 #include "system/json/json_reader.h"
@@ -33,8 +33,7 @@ static bool parse(
   }
   JsonReader jr{&js};
   if (!jr.read(result)) {
-      LOG(md, Error, "Failed to read BinancePriceStatistics json:\n{}",
-          js.dump());
+      LOG(md, Error, "Failed to read BinancePriceStatistics json:\n{}", js);
      return false;
   }
   return true;
@@ -1,6 +1,6 @@
 #include "BinancePriceStatisticsAdaptor.h"
 #include "domain/security_id.h"
-#include "domain/security_market_data_statistics.h"
+#include "spark/domain/SecurityMarketDataStatistics.h"
 #include "spark/md/IMarketDataListener.h"
 #include "spark/utils/bts_fatal.h"
 #include "sparkd/md/binance/BinancePriceStatistics.h"
@@ -12,19 +12,18 @@ namespace bts::md::binance {

 using bts::domain::SecurityID;
 using bts::domain::SecurityMarketDataStatistics;
-using bts::domain::protobufs::SecurityType;
+using std::pair;
 using std::string;
-using std::tuple;
 using std::vector;

 #define VENUE_NAME (Venue_Name(Venue::BINANCE))

 struct BinancePriceStatisticsAdaptor::PImpl {
-   vector<tuple<SecurityType, string, SecurityID>> securities;
+   vector<pair<string, SecurityID>> securities;
   IMarketDataListener *listener;

   SecurityID get_sec_id(const string &sec_desc) const {
-      for (const auto &[type, desc, id] : securities) {
+      for (const auto &[desc, id] : securities) {
         if (desc == sec_desc) {
            return id;
         }
@@ -37,8 +36,7 @@ struct BinancePriceStatisticsAdaptor::PImpl {
 };

 BinancePriceStatisticsAdaptor::BinancePriceStatisticsAdaptor(
-    vector<tuple<SecurityType, string, SecurityID>> securities,
-    IMarketDataListener *listener)
+    vector<pair<string, SecurityID>> securities, IMarketDataListener *listener)
    : me_(std::make_unique<PImpl>()) {
   bts_assert(listener != nullptr);
   me_->securities = std::move(securities);
@@ -1,11 +1,9 @@
 #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>

@@ -22,10 +20,7 @@ namespace bts::md::binance {
 class BinancePriceStatisticsAdaptor : public ICCApiMessageHandler {
 public:
   BinancePriceStatisticsAdaptor(
-       std::vector<std::tuple<
-           domain::protobufs::SecurityType,
-           std::string,
-           domain::SecurityID>> securities,
+       std::vector<std::pair<std::string, domain::SecurityID>> securities,
       IMarketDataListener *listener);

   ~BinancePriceStatisticsAdaptor();
@@ -3,6 +3,7 @@
 #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"
@@ -13,7 +14,6 @@
 #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,7 +49,6 @@ 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)                                               \
@@ -157,8 +156,7 @@ 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, sec.type(), sec.desc())};
+         SecurityID sec_id{sec_id_store->get_or_add(VENUE_NAME, sec.desc())};
         sec.security_common.id = sec_id;
      }
   }
@@ -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>
@@ -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>
@@ -2,9 +2,10 @@
 #include "domain/protobufs/venue.pb.h"
 #include "domain/security.h"
 #include "domain/security_criteria.h"
-#include "http/query_params.h"
 #include "spark/api/md/BinanceMarketDataConfig.pb.h"
 #include "spark/io/ProtoPrinter.h"
+#include "spark/io/QueryParams.h"
+#include "spark/logging/logging.h"
 #include "spark/md/IMarketDataListener.h"
 #include "spark/persistence/IDGenerator.h"
 #include "sparkd/md/binance/BinanceMessageType.h"
@@ -18,7 +19,6 @@
 #include "sparkd/md/ccapi/CCApiTradeAdaptor.h"
 #include "system/FMT.h"
 #include "system/collections/unordered_set_factory.h"
-#include "system/logging/logging.h"

 #include <ccapi_cpp/ccapi_event_dispatcher.h>
 #include <ccapi_cpp/ccapi_macro.h>
@@ -38,11 +38,10 @@ using bts::api::md::BinanceMarketDataConfig;
 using bts::domain::Security;
 using bts::domain::SecurityCriteria;
 using bts::domain::SecurityID;
-using bts::domain::protobufs::SecurityType;
 using bts::domain::protobufs::Venue;
 using bts::domain::protobufs::Venue_Name;
-using bts::http::QueryParams;
 using bts::io::ProtoPrinter;
+using bts::io::QueryParams;
 using bts::persistence::IDGenerator;
 using std::pair;
 using std::string;
@@ -51,7 +50,6 @@ using std::unique_ptr;
 using std::unordered_map;
 using std::vector;

-#define VENUE Venue::VENUE_BINANCE
 #define VENUE_NAME (Venue_Name(Venue::VENUE_BINANCE))
 #define CCAPI_VENUE_NAME CCAPI_EXCHANGE_NAME_BINANCE

@@ -125,15 +123,14 @@ static CCApiCorrelationID mk_stats_corr_id(const string &desc) {
 }

 static ccapi::Request mk_stats_req(
-    const vector<std::pair<SecurityType, string>> &sec_descs,
-    uint32_t batch_id) {
+    const vector<string> &sec_descs, uint32_t batch_id) {
   const auto corr_id = mk_stats_corr_id(FMT("BATCH_{}", batch_id));
   ccapi::Request req{
       ccapi::Request::Operation::GENERIC_PUBLIC_REQUEST, CCAPI_VENUE_NAME};
   req.setCorrelationId(corr_id.to_string());
   auto quoted_sec_descs =
-       sec_descs | ranges::views::transform(
-                       [](const auto &s) { return "\"" + s.second + "\""; });
+       sec_descs |
+       ranges::views::transform([](const auto &s) { return "\"" + s + "\""; });
   const string symbols_query_arg = FMT_JOIN(quoted_sec_descs, ",");
   req.appendParam(
       {{CCAPI_HTTP_PATH, BINANCE_PRICE_STATS_PATH},
@@ -216,18 +213,16 @@ struct BinanceSubscription::PImpl {
      sec_requestor.request();
   }

-   DepthHandler &mk_depth_handler(
-       SecurityType sec_type, const string &sec_desc) {
-      SecurityID sec_id{sec_id_store->get_or_add(VENUE, sec_type, sec_desc)};
+   DepthHandler &mk_depth_handler(const string &sec_desc) {
+      SecurityID sec_id{sec_id_store->get_or_add(VENUE_NAME, sec_desc)};
      auto [it, is_new] =
          depth_handlers.try_emplace(sec_desc, sec_id, listener.get());
      bts_assert(is_new);
      return it->second;
   }

-   TickerHandler &mk_ticker_handler(
-       SecurityType sec_type, const string &sec_desc) {
-      SecurityID sec_id{sec_id_store->get_or_add(VENUE, sec_type, sec_desc)};
+   TickerHandler &mk_ticker_handler(const string &sec_desc) {
+      SecurityID sec_id{sec_id_store->get_or_add(VENUE_NAME, sec_desc)};
      auto [it, is_new] =
          ticker_handlers.try_emplace(sec_desc, sec_id, listener.get());
      bts_assert_msgf(
@@ -236,25 +231,24 @@ struct BinanceSubscription::PImpl {
   }

   StatsHandler &mk_stats_handler(
-       const vector<std::pair<SecurityType, string>> &sec_descs,
-       uint32_t hnd_id) {
-      vector<std::tuple<SecurityType, string, SecurityID>> sec_desc_id_tuples;
-      sec_desc_id_tuples.reserve(sec_descs.size());
-      for (auto [type, name] : sec_descs) {
-         SecurityID id{sec_id_store->get_or_add(VENUE, type, name)};
-         sec_desc_id_tuples.emplace_back(type, std::move(name), id);
+       const vector<string> &sec_descs, uint32_t hnd_id) {
+      vector<pair<string, SecurityID>> sec_desc_id_pairs;
+      sec_desc_id_pairs.reserve(sec_descs.size());
+      for (const auto &sec_desc : sec_descs) {
+         sec_desc_id_pairs.emplace_back(
+             sec_desc,
+             SecurityID{sec_id_store->get_or_add(VENUE_NAME, sec_desc)});
      }
      auto [it, is_new] = stats_handlers.try_emplace(
          hnd_id,
-          std::make_unique<StatsHandler>(sec_desc_id_tuples, listener.get()));
+          std::make_unique<StatsHandler>(sec_desc_id_pairs, listener.get()));
      bts_assert_msgf(
          is_new, "StatsHandler already created with id '{}'", hnd_id);
      return *it->second;
   }

   void add_stats_handler(
-       const vector<std::pair<SecurityType, string>> &sec_descs,
-       CCApiSession::BatchRequest &reqs) {
+       const vector<string> &sec_descs, CCApiSession::BatchRequest &reqs) {
      if (sec_descs.empty()) {
         return;
      }
@@ -279,14 +273,14 @@ struct BinanceSubscription::PImpl {
      CCApiSession::BatchRequest stats_reqs;

      const int N_SECS_PER_STATS_REQ = 16;
-      vector<std::pair<SecurityType, string>> curr_stats_sec_descs;
-      for (const auto &[type, desc] : securities) {
+      vector<string> curr_stats_sec_descs;
+      for (const auto &sec_desc : securities) {
         depth_subs.push_back(
-             {mk_depth_sub(desc, depth_opts), &mk_depth_handler(type, desc)});
+             {mk_depth_sub(sec_desc, depth_opts), &mk_depth_handler(sec_desc)});
         ticker_subs.push_back(
-             {mk_ticker_sub(desc, ticker_opts),
-              &mk_ticker_handler(type, desc)});
-         curr_stats_sec_descs.push_back(std::make_pair(type, desc));
+             {mk_ticker_sub(sec_desc, ticker_opts),
+              &mk_ticker_handler(sec_desc)});
+         curr_stats_sec_descs.push_back(sec_desc);
         if (curr_stats_sec_descs.size() == N_SECS_PER_STATS_REQ) {
            add_stats_handler(curr_stats_sec_descs, stats_reqs);
            curr_stats_sec_descs.clear();
@@ -310,7 +304,7 @@ struct BinanceSubscription::PImpl {
    */
   void on_securities(vector<Security> secs) {
      // 'sec_descs' is used to make depth subscriptions
-      vector<pair<SecurityType, string>> sec_descs;
+      vector<string> sec_descs;
      sec_descs.reserve(secs.size());
      // 'sec_ptrs' is passed to IMarketDataListeners
      vector<const Security *> sec_ptrs;
@@ -323,8 +317,7 @@ struct BinanceSubscription::PImpl {
             id, std::make_unique<Security>(std::move(s)));
         bts_assert_msgf(
             is_new, "Received duplicate Binance security '{}'", desc);
-         sec_descs.push_back(
-             std::make_pair(it->second->type(), std::move(desc)));
+         sec_descs.push_back(std::move(desc));
         // add ptr to entry in 'securities' (not the loop var, we moved that)
         sec_ptrs.push_back(it->second.get());
      }
@@ -1,7 +1,7 @@
 #include "BinanceUsPriceStatistics.h"
 #include "domain/price.h"
 #include "domain/quantity.h"
-#include "domain/security_market_data_statistics.h"
+#include "spark/domain/SecurityMarketDataStatistics.h"
 #include "sparkd/md/ccapi/CCApiReader.h"
 #include "system/json/json_dumper.h"
 #include "system/json/json_reader.h"
@@ -36,7 +36,7 @@ static bool parse(
   JsonReader jr{&js};
   if (!jr.read(result)) {
      LOG(md, Error, "Failed to read BinanceUsPriceChangeStatistics json:\n{}",
-          js.dump());
+          js);
      return false;
   }
   return true;