refactor: Add new glbx variable ticks to security (shm and proto)

This makes it consistent with other type names (we don't use "Sparkd"
anywhere else).

.env

@@ -29,9 +29,6 @@
 #     tools/deploy.sh --help
 #
 
-### Host/Runtime ###
-TZ=Etc/UTC
-
 ### Sparkd ###
 # The internal image for sparkd to use. For more info, view:
 #   src/server/sparkd/Dockerfile
@@ -41,9 +38,6 @@ SPARKD__CONFIG_DIR=src/app_sparkd/configurations/dev
 # Mount a specific volume or dir to "~/sparkdata/data" inside the docker.
 # Defaults to a volume named "sparkd_<config_port>" (e.g., sparkd_8000).
 # SPARKD__DATA_DIR=/home/bts/sparkdata/data
-# Mount volume or dir to "/md_recording" inside the docker. Defaults to volume
-# "sparkd_md_recording_<config_port>" (e.g., sparkd_md_recording_8000).
-# SPARKD__MD_RECORDING_DIR=/home/bts/md_recording
 SPARKD__SOURCE_DIR=install/sparkd-release-gcc
 SPARKD__RUNTIME_CONF_DIR=/home/ubuntu/8000/sparkdata/conf/
 SPARKD__CONFIG_PORT=8000
@@ -51,7 +45,6 @@ SPARKD__QUICKBUS_PORT=8001
 SPARKD__SIM_FIX_PORT=8002
 SPARKD__SHM_SIZE=4gb
 SPARKD__PYTHON_REQUIREMENTS=requirements.txt
-SPARKD__ENTRYPOINT=src/app_sparkd/sparkd-entrypoint.sh
 
 ### PubSubServer ###
 PUBSUB__CONFIG_DIR=src/DataStreamSuite/PubSubServer/Configurations/Development

CMakePresets.json

@@ -32,20 +32,14 @@
         "lhs": "${hostSystemName}",
         "rhs": "Windows"
       },
-      "generator": "Ninja Multi-Config",
+      "generator": "Visual Studio 17 2022",
       "environment": {
-        "SPARK_CMAKE_CXX_FLAGS": "/WX /wd4250 /wd4251 /wd4244 /wd4267 /wd4355 /wd4275 /wd4345 /bigobj /EHsc /DCURL_STATICLIB /D_WINSOCK_DEPRECATED_NO_WARNINGS /D__TBB_NO_IMPLICIT_LINKAGE /DBOOST_ALL_NO_LIB /std:c++20 /D_SCL_SECURE_NO_WARNINGS /D_CRT_SECURE_NO_WARNINGS /D_WIN32_WINNT=0x0600"
+        "SPARK_CMAKE_CXX_FLAGS": "/MP /WX /wd4250 /wd4251 /wd4244 /wd4267 /wd4355 /wd4275 /wd4345 /bigobj /EHsc /DCURL_STATICLIB /D_WINSOCK_DEPRECATED_NO_WARNINGS /D__TBB_NO_IMPLICIT_LINKAGE /DBOOST_ALL_NO_LIB /std:c++20 /D_SCL_SECURE_NO_WARNINGS /D_CRT_SECURE_NO_WARNINGS /D_WIN32_WINNT=0x0600"
       },
       "cacheVariables": {
-        "CMAKE_C_COMPILER": "clang-cl",
-        "CMAKE_CXX_COMPILER": "clang-cl",
         "CMAKE_CXX_FLAGS": "$env{SPARK_CMAKE_CXX_FLAGS}",
         "CMAKE_CXX_FLAGS_DEBUG": "$env{SPARK_CMAKE_CXX_FLAGS} /MTd",
         "CMAKE_CXX_FLAGS_RELEASE": "$env{SPARK_CMAKE_CXX_FLAGS} /MT",
-        "CMAKE_EXPORT_COMPILE_COMMANDS": {
-          "type": "BOOL",
-          "value": "ON"
-        },
         "CMAKE_SHARED_LINKER_FLAGS_DEBUG": "/DEBUG",
         "CMAKE_SHARED_LINKER_FLAGS_RELEASE": "/OPT:REF /OPT:ICF /LTCG",
         "EXTERNALS_ROOT": "${sourceDir}/externals/windows"

compose.yaml

@@ -2,7 +2,6 @@ services:
   sparkd:
     volumes:
       - ${SPARKD__DATA_DIR:-sparkd_data}:/home/bts/.local/share/bts/sparkd/data
-      - ${SPARKD__MD_RECORDING_DIR:-md_recording}:/md_recording
     build:
       dockerfile: src/app_sparkd/Dockerfile
       args:
@@ -10,34 +9,24 @@ services:
         - SOURCE_DIR=${SPARKD__SOURCE_DIR}
         - CONFIG_DIR=${SPARKD__CONFIG_DIR}
         - PYTHON_REQUIREMENTS=${SPARKD__PYTHON_REQUIREMENTS}
-        - ENTRYPOINT_SCRIPT=${SPARKD__ENTRYPOINT}
     shm_size: "${SPARKD__SHM_SIZE:-512mb}"
-    env_file:
-      - .env
-      - ${SPARKD__CONFIG_DIR}/.env
-    ports:
-      - "${SPARKD__CONFIG_PORT}:${SPARKD__CONFIG_PORT}"
-      - "${SPARKD__QUICKBUS_PORT}:${SPARKD__QUICKBUS_PORT}"
-      - "${SPARKD__SIM_FIX_PORT}:${SPARKD__SIM_FIX_PORT}"
-    deploy:
-      resources:
-        limits:
-          memory: "${SPARKD__MEMORY:-4gb}"
-
-  sparkd-shell:
-    volumes:
-      - ${SPARKD__DATA_DIR:-sparkd_data}:/home/bts/.local/share/bts/sparkd/data
-      - ${SPARKD__MD_RECORDING_DIR:-md_recording}:/md_recording
-    build:
-      dockerfile: docker/images/bluetradingsystems/linux_shell/Dockerfile
-      args:
-        - IMAGE=${SPARKD__IMAGE}
-        - SOURCE_DIR=${SPARKD__SOURCE_DIR}
-        - CONFIG_DIR=${SPARKD__CONFIG_DIR}
-        - PYTHON_REQUIREMENTS=${SPARKD__PYTHON_REQUIREMENTS}
-        - ENTRYPOINT_SCRIPT=${SPARKD__ENTRYPOINT}
-    shm_size: "${SPARKD__SHM_SIZE:-512mb}"
-    command: "do sleep 1; done"
+    command: >
+      sh -c "
+        echo bts | sudo -S chown -R bts:bts /home/bts/.local/share/bts/sparkd/data
+        if [ -e "/home/bts/.local/share/bts/sparkd/data/logging.sparkd/log" ]; then
+          mv "/home/bts/.local/share/bts/sparkd/data/logging.sparkd/log" "/home/bts/.local/share/bts/sparkd/data/logging.sparkd/log_$(date '+%Y%m%d-%H%M')"
+        fi
+        if [ -e "/home/bts/.local/share/bts/conf/.env.local" ]; then
+          echo 'Sourcing /home/bts/.local/share/bts/conf/.env.local'
+          set -a
+          . /home/bts/.local/share/bts/sparkd/conf/.env.local
+          set +a
+        fi
+        if [ -d /home/bts/.local/share/bts/sparkd/conf/plugins ]; then
+          echo 'Copying /home/bts/.local/share/bts/sparkd/conf/plugins to runtime dir (/home/bts/bin)'
+          cp -r /home/bts/.local/share/bts/sparkd/conf/plugins /home/bts/bin/
+        fi
+        ./sparkd run"
     env_file:
       - .env
       - ${SPARKD__CONFIG_DIR}/.env
@@ -61,8 +50,26 @@ services:
         - SOURCE_DIR=${SPARKD__SOURCE_DIR}
         - CONFIG_DIR=${SPARKD__CONFIG_DIR}
         - PYTHON_REQUIREMENTS=${SPARKD__PYTHON_REQUIREMENTS}
-        - ENTRYPOINT_SCRIPT=${SPARKD__ENTRYPOINT}
     shm_size: "${SPARKD__SHM_SIZE:-512mb}"
+    command: >
+      sh -c "
+        echo bts | sudo -S chown -R bts:bts /home/bts/.local/share/bts/sparkd
+        if [ -e "/home/bts/.local/share/bts/sparkd/data/logging.sparkd/log" ]; then
+          mv "/home/bts/.local/share/bts/sparkd/data/logging.sparkd/log" "/home/bts/.local/share/bts/sparkd/data/logging.sparkd/log_$(date '+%Y%m%d-%H%M')"
+        fi
+        ls -alh /home/bts/.local/share/bts/sparkd/conf/
+        cat /home/bts/.local/share/bts/sparkd/conf/.env.local
+        if [ -e "/home/bts/.local/share/bts/sparkd/conf/.env.local" ]; then
+          echo 'Sourcing /home/bts/.local/share/bts/sparkd/conf/.env.local'
+          set -a
+          . /home/bts/.local/share/bts/sparkd/conf/.env.local
+          set +a
+        fi
+        if [ -d /home/bts/.local/share/bts/sparkd/conf/plugins ]; then
+          echo 'Copying /home/bts/.local/share/bts/sparkd/conf/plugins to runtime dir (/home/bts/bin)'
+          cp -r /home/bts/.local/share/bts/sparkd/conf/plugins /home/bts/bin/
+        fi
+        ./sparkd run"
     env_file:
       - .env
       - ${SPARKD__CONFIG_DIR}/.env
@@ -188,6 +195,3 @@ volumes:
     name: "md_publisher_22019"
   historic_md:
     name: "historic_md"
-  md_recording:
-    name: "sparkd_md_recording_${SPARKD__CONFIG_PORT:-8000}"
-    external: true

docker/images/bluetradingsystems/linux_dotnet8/Dockerfile

@@ -8,7 +8,6 @@ RUN apt-get update && apt-get install -y ca-certificates
 RUN apt-get update && apt-get install -y libssh2-1-dev
 RUN apt-get update && apt-get install -y software-properties-common
 RUN apt-get update && apt-get install -y sudo
-RUN apt-get update && apt-get install -y tzdata
 
 # GCC
 ARG gcc_major=13

docker/images/bluetradingsystems/linux_shell/Dockerfile

@@ -1,11 +0,0 @@
-FROM ubuntu:24.04
-
-ARG DEBIAN_FRONTEND=noninteractive
-
-RUN apt-get update && apt-get install -y \
-  fish \
-  less \
-  vim
-
-USER bts
-WORKDIR /home/bts

src/DataStreamSuite/DataStream/Primitives/Crushing/AverageCollectorStream.cs

@@ -20,7 +20,7 @@ namespace BCG
         {
             IDictionary<string, double> columnSums = new Dictionary<string, double>();
 
-            IBlueRow? resultRow = null;
+            IBlueRow resultRow = null;
             foreach (IBlueRow row in rows)
             {
                 resultRow = row;
@@ -29,7 +29,7 @@ namespace BCG
                 {
                     double sum = 0.0f;
                     columnSums.TryGetValue(colName, out sum);
-                    sum += (double)row[colName]!;
+                    sum += (double) row[colName];
                     columnSums[colName] = sum;
                 }
             }

src/DataStreamSuite/DataStream/Primitives/Distinct/DistinctRowEvaluator.cs

@@ -1,3 +1,8 @@
+using System;
+using System.Collections;
+
+using BCG.Utilities;
+
 namespace BCG
 {
 	public class DistinctRowEvaluator : GroupByEvaluator
@@ -9,8 +14,10 @@ namespace BCG
 		public DistinctRowEvaluator(string colMapExpression, string[] columnNames)
          : base (colMapExpression, null, columnNames) {}
 
-      public override IBlueRow? evaluate(IDataPacket packet) {
-         myOutputTable ??= new GroupByOutputTable(packet.Schema);
+
+      public override IBlueRow evaluate (DataPacket packet)
+      {
+         myOutputTable = myOutputTable == null ? new GroupByOutputTable (packet.schema) : myOutputTable;
          string hashString = GroupByStream.getRepString (packet, myGroupByCols);
          if (packet.newRows.Count == 0)
             //if there are no rows for this group we remove the group from the output table

src/DataStreamSuite/DataStream/Primitives/GroupBy/GroupByBundle.cs

@@ -43,7 +43,7 @@ namespace bts.SilverPine {
          operations = operations_;
       }
 
-      public static IBlueRow? group_by(List<IBlueRow> rows, MultiRowOpHelper parameters) {
+      public static IBlueRow group_by(List<IBlueRow> rows, MultiRowOpHelper parameters) {
          if (rows.Count() == 0) {
             return null;
          }
@@ -73,7 +73,7 @@ namespace bts.SilverPine {
       // rolling_rows are the rolling window of rows. the 0th row is the one thats gettin
       // calculated. rows 1 - n-1 are its window rows. the schema is based on the operations.
 
-      public IBlueRow? rolling_calc(List<IBlueRow> rolling_rows) {
+      public IBlueRow rolling_calc(List<IBlueRow> rolling_rows) {
          if (rolling_rows.Count() == 0) {
             return null;
          }
@@ -123,4 +123,4 @@ namespace bts.SilverPine {
       }
 
    }
-}
+}

src/DataStreamSuite/DataStream/Primitives/GroupBy/GroupByEvaluator.cs

@@ -9,18 +9,18 @@ namespace BCG
 	public abstract class GroupByEvaluator : IGroupByEvaluator
 	{
       protected string[] myGroupByCols;
-      protected string? myOutputColumn;
+      protected string myOutputColumn;
       protected GroupByOutputTable? myOutputTable;
       protected string? myColMapExpression;
 
       #region constructors
-		public GroupByEvaluator (string? outputCol, string[] groupByCols)
+		public GroupByEvaluator (string outputCol, string[] groupByCols)
 		{
          myOutputColumn = outputCol;
          myGroupByCols = groupByCols;
 		}
 
-      public GroupByEvaluator (string colMapExpression, string? outputCol, string[] groupByCols)
+      public GroupByEvaluator (string colMapExpression, string outputCol, string[] groupByCols)
          : this (outputCol, groupByCols)  {myColMapExpression = colMapExpression;}
       #endregion
 
@@ -42,7 +42,7 @@ namespace BCG
          return retVal;
       }
 
-      public abstract IBlueRow? evaluate(IDataPacket dp);
+      public abstract IBlueRow evaluate (DataPacket dp) ;
       #endregion
 
       #region Helper methods
@@ -52,7 +52,7 @@ namespace BCG
       /// <param name="r"></param>
       /// <param name="id"></param>
       /// <returns></returns>
-      protected IBlueRow? myProject(IBlueRow? r, int id) =>
+      protected IBlueRow? myProject(IBlueRow r, int id) =>
          r is null
             ? null
             : myColMapExpression is null || myColMapExpression.Length < 1
@@ -64,7 +64,7 @@ namespace BCG
       /// </summary>
       /// <param name="r"></param>
       /// <returns></returns>
-      protected IBlueRow? myProject(IBlueRow? r) =>
+      protected IBlueRow? myProject(IBlueRow r) =>
          myProject(r, r is null ? int.MinValue : r.Id());
       #endregion
 

src/DataStreamSuite/DataStream/Primitives/GroupBy/GroupByStream.cs

@@ -10,13 +10,12 @@ namespace BCG {
    internal class GroupByStream : MethodDrivenStream {
       private class MyFuncGroupByEvaluator : IGroupByEvaluator {
          private string[] _groupby_cols;
-         readonly Func<IDataPacket, IBlueRow> _groupby_func;
-         public MyFuncGroupByEvaluator(IEnumerable<string> groupby_cols, Func<IDataPacket, IBlueRow> groupby_func) {
+         private Func<DataPacket, IBlueRow> _groupby_func;
+         public MyFuncGroupByEvaluator(IEnumerable<string> groupby_cols, Func<DataPacket, IBlueRow> groupby_func) {
             _groupby_cols = groupby_cols.ToArray();
             _groupby_func = groupby_func;
          }
-         public IBlueRow evaluate(IDataPacket packet) =>
-            _groupby_func(packet);
+         public IBlueRow evaluate(DataPacket packet) { return _groupby_func(packet); }
 
          public string[] groupByColumns { get { return _groupby_cols; } }
 
@@ -36,7 +35,7 @@ namespace BCG {
       IBlueTableSchema? myOutputSchema;
       private int myIdGenerator = 0;
 
-      public GroupByStream(ViewStream stream, IEnumerable<string> groupby_cols, Func<IDataPacket, IBlueRow> groupby_func)
+      public GroupByStream(ViewStream stream, IEnumerable<string> groupby_cols, Func<DataPacket, IBlueRow> groupby_func)
          : this(stream, new MyFuncGroupByEvaluator(groupby_cols, groupby_func)) { }
 
       public GroupByStream(ViewStream stream, IGroupByEvaluator groupByEvaluator)
@@ -197,11 +196,11 @@ namespace BCG {
          }
       }
 
-      public static string getRepString(IDataPacket dp, string[] colNames) {
+      public static string getRepString(DataPacket dp, string[] colNames) {
          return BlueDataUtility.rowToString(getRepRow(dp), colNames);
       }
 
-      public static IBlueRow? getRepRow(IDataPacket dp) {
+      public static IBlueRow? getRepRow(DataPacket dp) {
          if (dp.newRows.Count > 0)
             return BlueDataUtility.extractRow(dp.newRows, 0);
          else if (dp.addedRows.Count > 0)

src/DataStreamSuite/DataStream/Primitives/GroupBy/IGroupByEvaluator.cs

@@ -1,3 +1,5 @@
+using System;
+
 namespace BCG
 {
    /// <summary>
@@ -12,7 +14,7 @@ namespace BCG
       /// deletetion) and the added and deletedRows being the increments. If a group has
       /// to be totally removed then a null should be returned
       /// </summary>
-      IBlueRow? evaluate(IDataPacket packet);
+      IBlueRow evaluate (DataPacket packet);
 
       /// <summary>
       /// Returns the columns (belonging to the original schema) on which this IGroupByEvaluator

src/DataStreamSuite/DataStream/Primitives/GroupBy/MaxGroupByEvaluator.cs

@@ -1,3 +1,8 @@
+using System;
+using System.Collections;
+
+using BCG.Utilities;
+
 namespace BCG {
     /// <summary>
     /// Computes the max for a given column
@@ -9,13 +14,14 @@ namespace BCG {
         public MaxGroupByEvaluator(string colMapEpression, string outputCol, string[] groupByCols)
            : base(colMapEpression, outputCol, groupByCols) { }
 
-        public override IBlueRow? evaluate(IDataPacket dp) {
-            myOutputTable ??= new GroupByOutputTable(dp.Schema);
+        public override IBlueRow evaluate(DataPacket dp) {
+            myOutputTable = myOutputTable == null ? new GroupByOutputTable(dp.schema) :
+               myOutputTable;
             string hashString = BlueDataUtility.rowToString(GroupByStream.getRepRow(dp), myGroupByCols);
-            var currMaxRow = myOutputTable.getRow(hashString);
+            IBlueRow currMaxRow = myOutputTable.getRow(hashString);
             IBlueRow maxRow = BlueDataUtility.computeMaxRow(dp.deletedRows, myOutputColumn);
             if (maxRow != null) {
-                bts.Log.notnull(currMaxRow);
+                bts.Log.assert(currMaxRow != null);
                 bts.Log.assert(BlueDataUtility.compare_rows(maxRow, currMaxRow, myOutputColumn) <= 0);
                 if (BlueDataUtility.compare_rows(maxRow, currMaxRow, myOutputColumn) == 0) {
                     maxRow = BlueDataUtility.computeMaxRow(dp.newRows, myOutputColumn);

src/DataStreamSuite/DataStream/Primitives/GroupBy/MinGroupByEvaluator.cs

@@ -1,3 +1,7 @@
+using System;
+
+using BCG.Utilities;
+
 namespace BCG {
     public class MinGroupByEvaluator : MaxGroupByEvaluator {
         public MinGroupByEvaluator(string outputCol, string[] groupByCols)
@@ -6,10 +10,10 @@ namespace BCG {
         public MinGroupByEvaluator(string colMapEpression, string outputCol, string[] groupByCols) :
            base(colMapEpression, outputCol, groupByCols) { }
 
-        public override IBlueRow? evaluate(IDataPacket dp) {
-            myOutputTable ??= new GroupByOutputTable(dp.Schema);
+        public override IBlueRow evaluate(DataPacket dp) {
+            myOutputTable = myOutputTable == null ? new GroupByOutputTable(dp.schema) : myOutputTable;
             string stringRep = BlueDataUtility.rowToString(GroupByStream.getRepRow(dp), myGroupByCols);
-            var currMinRow = myOutputTable.getRow(stringRep);
+            IBlueRow currMinRow = myOutputTable.getRow(stringRep);
             IBlueRow minRow = BlueDataUtility.computeMinRow(dp.deletedRows, myOutputColumn);
             if (minRow != null) {
                 // BlueEventLog.logAssert (currMinRow != null);

src/DataStreamSuite/DataStream/Primitives/GroupBy/SumGroupByEvaluator.cs

@@ -1,3 +1,5 @@
+using System;
+
 namespace BCG {
     /// <summary>
     /// Computes the sum for a given column
@@ -8,8 +10,8 @@ namespace BCG {
         public SumGroupByEvaluator(string colMapEpression, string outputCol, string[] groupByCols)
            : base(colMapEpression, outputCol, groupByCols) { }
 
-        public override IBlueRow evaluate(IDataPacket dp) {
-            myOutputTable ??= new GroupByOutputTable(dp.Schema);
+        public override IBlueRow evaluate(DataPacket dp) {
+            myOutputTable = myOutputTable == null ? new GroupByOutputTable(dp.schema) : myOutputTable;
 
             IBlueRow sumRow = BlueDataUtility.computeSum(dp.newRows, myOutputColumn);
             return sumRow;
@@ -31,7 +33,7 @@ namespace BCG {
 
         #region IGroupByEvaluator Members
 
-        public IBlueRow evaluate(IDataPacket packet) => BlueDataUtility.computeSum(packet.newRows, myOutputCols);
+        public IBlueRow evaluate(DataPacket packet) { return BlueDataUtility.computeSum(packet.newRows, myOutputCols); }
 
         public string[] groupByColumns { get { return myGroupByCols; } }
 
@@ -49,8 +51,9 @@ namespace BCG {
         public SumGroupByEvaluator(string colMapEpression, string outputCol, string[] groupByCols)
            : base(colMapEpression, outputCol, groupByCols) { }
 
-        public override IBlueRow evaluate(IDataPacket dp) {
-            myOutputTable ??= new GroupByOutputTable(dp.Schema);
+        public override IBlueRow evaluate(DataPacket dp) {
+            myOutputTable = myOutputTable == null ? new GroupByOutputTable(dp.schema) : myOutputTable;
+
             IBlueRow sumRow = BlueDataUtility.computeSum<T>(dp.newRows, myOutputColumn);
             return sumRow;
         }
@@ -65,7 +68,7 @@ namespace BCG {
 
         #region IGroupByEvaluator Members
 
-        public IBlueRow evaluate(IDataPacket packet) => BlueDataUtility.computeSum<T>(packet.newRows, myOutputCols);
+        public IBlueRow evaluate(DataPacket packet) { return BlueDataUtility.computeSum<T>(packet.newRows, myOutputCols); }
 
         public string[] groupByColumns { get { return myGroupByCols; } }
 

src/DataStreamSuite/DataStream/Primitives/GroupBy/SumStream.cs

@@ -45,13 +45,12 @@ namespace BCG {
         }
 
         public SumStream(ViewStream stream, string[] output_columns, string[] groupby_columns, bool force_null_into_other_columns) {
-            ViewStream? rs = null;
+            ViewStream rs = null;
             groupby_columns = groupby_columns ?? new string[] { };
             if (groupby_columns.Length == 0) {
                 // rs = new AddColumnsStream(stream, "fake_col=0");
                 // rs = new GroupByStream(rs, new MultiColumnSumGroupByEvaluator(output_columns, new string[] { "fake_col" }));
                 // rs = new ProjectStream(rs, "!fake_col");
-                throw new ArgumentException("No groupby columns specified", nameof(groupby_columns));
             } else {
                 rs = new GroupByStream(stream, new MultiColumnSumGroupByEvaluator(output_columns, groupby_columns));
             }

src/DataStreamSuite/DataStream/Primitives/Map/GlobalMapStream.cs

@@ -2,8 +2,6 @@ using BCG;
 
 using MongoDB.Driver.Core.WireProtocol.Messages;
 
-using Spark.Core.Diagnostics;
-
 using System;
 using System.Collections.Generic;
 using System.Linq;
@@ -39,12 +37,12 @@ namespace bts {
       }
 
       void _handle_stream_update(object dp_obj) {
-         var dp = ExceptionAssert.Argument.Is<IDataPacket>(dp_obj);
+         DataPacket dp = dp_obj as DataPacket;
          IDictionary<int, List<IBlueRow>> groups = !String.IsNullOrEmpty(_group_by_col) ?
          BlueDataUtility.group_together(dp.newRows, new string[] { _group_by_col })
          :
          new Dictionary<int, List<IBlueRow>>() { { 0, dp.newRows.to_list() } };
-         BlueTable? out_table = null;
+         BlueTable out_table = null;
          foreach (var kv in groups) {
             List<IBlueRow> group_rows = kv.Value;
             group_rows.Sort((r1, r2) => BlueDataUtility.compare_rows(r2, r1, _sort_col_name));
@@ -73,4 +71,4 @@ namespace bts {
    }
 
 
-}
+}

src/DataStreamSuite/SharedStreams/Business/Indicators/ChartsUtils.cs

@@ -3,7 +3,6 @@ using bts.config;
 using System;
 using System.Collections;
 using System.Collections.Generic;
-using System.Diagnostics.CodeAnalysis;
 
 namespace BCG {
 
@@ -22,9 +21,7 @@ namespace BCG {
     }
     public class RenkoCalculator {
         public readonly double brick_size;
-        public RenkoBrick? current_brick;
-
-        [MemberNotNull(nameof(current_brick))]
+        public RenkoBrick current_brick = null;
         private bool _add_price(double price) {
             if (current_brick == null) {
                 current_brick = new RenkoBrick {
@@ -312,4 +309,4 @@ namespace BCG {
             return wma_values;
         }
     }
-}
+}

src/Directory.Build.targets

@@ -187,32 +187,4 @@
       Name="Test"
       DependsOnTargets="Build"></Target>
 
-  <ItemGroup Condition="Exists('@(NativeWindowsAsset)')">
-    <Content Include="@(NativeWindowsAsset)">
-      <CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
-      <TargetPath>%(FileName)%(Extension)</TargetPath>
-      <Pack>false</Pack>
-    </Content>
-    <Content Include="@(NativeWindowsAsset)">
-      <CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
-      <TargetPath>runtimes/win-x64/native/%(FileName)%(Extension)</TargetPath>
-      <Pack>true</Pack>
-      <PackagePath>runtimes/win-x64/native</PackagePath>
-    </Content>
-  </ItemGroup>
-
-  <ItemGroup Condition="Exists('@(NativeLinuxAsset)')">
-    <Content Include="@(NativeLinuxAsset)">
-      <CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
-      <TargetPath>%(FileName)%(Extension)</TargetPath>
-      <Pack>false</Pack>
-    </Content>
-    <Content Include="@(NativeLinuxAsset)">
-      <CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
-      <TargetPath>runtimes/linux-x64/native/%(FileName)%(Extension)</TargetPath>
-      <Pack>true</Pack>
-      <PackagePath>runtimes/linux-x64/native</PackagePath>
-    </Content>
-  </ItemGroup>
-
 </Project>

src/app_sparkd/Dockerfile

@@ -2,14 +2,12 @@ ARG IMAGE="sparkd"
 ARG CONFIG_DIR="src/app_sparkd/configurations/dev"
 ARG SOURCE_DIR="install/sparkd-release-gcc"
 ARG PYTHON_REQUIREMENTS="requirements.txt"
-ARG ENTRYPOINT_SCRIPT="src/app_sparkd/sparkd-entrypoint.sh"
 
 # sparkd_base
 FROM bluetradingsystems/linux_dotnet8 AS sparkd_base
 ARG CONFIG_DIR
 ARG SOURCE_DIR
 ARG PYTHON_REQUIREMENTS
-ARG ENTRYPOINT_SCRIPT
 
 USER "root"
 RUN apt-get update && apt-get install -y liblttng-ust-dev
@@ -23,7 +21,6 @@ RUN pip3 install -r "/home/bts/bin/requirements.txt"
 USER "bts"
 COPY --chown=bts:bts "${SOURCE_DIR}" "."
 COPY --chown=bts:bts "${CONFIG_DIR}" "/home/bts/.local/share/bts/sparkd/conf"
-COPY --chown=bts:bts "${ENTRYPOINT_SCRIPT}" "/home/bts/bin/sparkd-entrypoint.sh"
 WORKDIR "/home/bts/bin"
 
 # sparkd
@@ -38,7 +35,6 @@ COPY --chown=bts:bts "${MD_RECORDING_DIR}" "/home/bts/.local/share/bts/sparkd/da
 FROM bluetradingsystems/linux_dotnet8 AS sparkd_symbols
 ARG SOURCE_DIR
 ARG PYTHON_REQUIREMENTS="requirements.txt"
-ARG ENTRYPOINT_SCRIPT
 
 USER "root"
 RUN apt-get update && apt-get install -y liblttng-ust-dev
@@ -51,9 +47,7 @@ RUN pip3 install -r "/home/bts/bin/requirements.txt"
 
 USER "bts"
 COPY --chown=bts:bts "${SOURCE_DIR}" "."
-COPY --chown=bts:bts "${ENTRYPOINT_SCRIPT}" "/home/bts/bin/sparkd-entrypoint.sh"
 WORKDIR "/home/bts/bin"
 
 # selected image
 FROM ${IMAGE}
-ENTRYPOINT "./sparkd-entrypoint.sh"

src/app_sparkd/configurations/cme_replay/.env

@@ -1,4 +0,0 @@
-SPARKD__CONFIG_DIR=src/app_sparkd/configurations/cme_replay
-SPARKD__PYTHON_REQUIREMENTS=src/app_sparkd/configurations/cme_replay/requirements.txt
-SPARKD__ENTRYPOINT=src/app_sparkd/sparkd-entrypoint.sh
-TZ=America/Chicago

src/app_sparkd/configurations/cme_replay/requirements.txt

@@ -1 +0,0 @@
-requests==2.31.0

src/app_sparkd/configurations/cme_replay/spark.conf

@@ -1,38 +0,0 @@
-PublisherOptions {
-  value {
-    network_interface: "0.0.0.0"
-    heartbeat_ms: 5000
-    heartbeat_boot_seconds: 60
-    skip_options: true
-    hot_initial: 1024
-    hot_growth: 1024
-    cold_initial: 8192
-    cold_growth: 2048
-    channel_prefix: ""
-    disable_heartbeat: true
-  }
-}
-
-MarketDataConfig {
-  value {
-   # glbx{
-   #   use_databento: true
-   #  databento_settings {
-   #     read_from_file: true
-   #    key: "db-DgxpuTXneNgVvd7sJCAEKYDhyCApW"
-   #     bbo_file_name: "glbx-mdp3-20251103.mbp-1.dbn.zst"
-   #     contracts_file_name: "glbx-mdp3-20251103.definition.dbn.zst"
-   #   }
-   # }
-   # recording{
-   #    venue: VENUE_GLBX
-   #     num_of_seconds_to_run_file_writer: 240
-   #     num_of_seconds_to_run_db_writer : 0
-   #     n_book_levels: 10 
-   #     batch_size: 100 #for the db writer
-   # }
-   replay {
-      venue: VENUE_GLBX
-    }
-  }
-}

src/app_sparkd/configurations/cme_replay/symbols.conf

@@ -1,48 +0,0 @@
-Symbols {
-  value {
-    venues {
-        name: "VENUE_GLBX"
-        future: "ES"
-        # option: "E1A"
-        # option: "E1B"
-        # option: "E1C"
-        # option: "E1D"
-        # option: "E2A"
-        # option: "E2B"
-        # option: "E2C"
-        # option: "E2D"
-        # option: "E3A"
-        # option: "E3B"
-        # option: "E3C"
-        # option: "E3D"
-        # option: "E4A"
-        # option: "E4B"
-        # option: "E4C"
-        # option: "E4D"
-        # option: "E5A"
-        # option: "E5B"
-        # option: "E5C"
-        # option: "E5D"
-        # option: "EW1"
-        # option: "EW2"
-        # option: "EW3"
-        # option: "EW4"
-        # option: "ESH9"
-        # option: "ESZ6"
-        # option: "ESH0"
-        # option: "ESM6"
-        # option: "ESZ7"
-        # option: "ESH6"
-        # option: "ESM7"
-        # option: "ESZ8"
-        # option: "ESH7"
-        # option: "ESH8"
-        # option: "ESU6"
-        # option: "ESZ5"
-        # option: "ESZ9"
-        # option: "E1D"
-        # option: "ESZ5"
-        # option: "ES"
-    }
-  }
-}

src/app_sparkd/sparkd-entrypoint.sh

@@ -1,45 +0,0 @@
-#!/usr/bin/env bash
-
-init_timezone() {
-  tz="${1:-$TZ}"
-  zoneinfo_dir=/usr/share/zoneinfo
-  if [ -f "$zoneinfo_dir/$tz" -a "$(readlink -f /etc/localtime)" != "$zoneinfo_dir/$tz" ]; then
-    echo "Initializing locales and timezone"
-    ln -snf "$zoneinfo_dir/$tz" /etc/localtime
-    sudo echo $tz > /etc/timezone
-  fi
-}
-
-load_env() {
-  f="$1"
-  if [ -e "$f" ]; then
-    echo "Sourcing $f"
-    set -a
-    . "$f"
-    set +a
-  fi
-}
-
-symlink() {
-  src="$1"
-  dst="$2"
-  if [ -L "$dst" ]; then
-    rm "$dst"
-  fi
-  ln -s "$src" "$dst"
-}
-
-roll_log() {
-  f="$1"
-  if [ -e "$f" ]; then
-    mv "$f" "${f}_$(date '+%Y%m%d-%H%M')"
-  fi
-}
-
-data_root="${HOME:-/home/bts}/.local/share/bts/sparkd"
-echo bts | sudo -S chown -R bts:bts "$data_root"
-load_env "$data_root/conf/.env.local"
-init_timezone "$TZ"
-roll_log "$data_root/data/logging.sparkd/log"
-symlink "$data_root/conf/plugins" /home/bts/bin/plugins
-bash -c "${@:-./sparkd run}"

src/app_sparkd/sparkd/cmds/Run.cpp

@@ -10,13 +10,14 @@
 #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/os/os.h"
 #include "spark/os/spark_data_directory.h"
 #include "spark/persistence/PersistenceServiceServer.h"
 #include "spark/plugin/SparkServices.h"
@@ -31,7 +32,6 @@
 #include "sparkd/execution/execution_gateway_provider.h"
 #include "sparkd/http/WebServer.h"
 #include "sparkd/md/MarketDataAuxiliaryServiceServer.h"
-#include "sparkd/md/market_data_session.h"
 #include "sparkd/md/recording/MDRecorder.h"
 #include "sparkd/position/MarginServiceServer.h"
 #include "sparkd/position/PositionServiceServer.h"
@@ -46,7 +46,6 @@
 namespace bts::cmds {
 
 using bts::api::ServiceContext;
-using bts::md::MarketDataSession;
 using bts::md::MDClient;
 using bts::md::TradeSummarySignalDispatcher;
 using bts::metrics::MetricsContainer;
@@ -108,7 +107,7 @@ static void execute() {
    auto py_plugin_man = std::make_shared<python_interop::PluginManager>();
    // start work loop in background
    py_plugin_man->start_work_thread();
-   MarketDataSession md_session{py_plugin_man};
+   md::cmds::Run::run(py_plugin_man);
 
    api::synthetic::SyntheticSecurityService synthetic_service(sc);
    bts::md::MDClient mdclient(

src/app_sparkd/sparkd/cmds/Web.cpp

@@ -6,7 +6,7 @@
 #include "spark/config/protos/quickbus_cfg.pb.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"

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

@@ -1,11 +1,11 @@
 #include "DumpSecurity.h"
 #include "spark/api/domain/SecurityType.h"
 #include "spark/config/protos/market_data.pb.h"
+#include "spark/core/TableFormatter.h"
 #include "spark/domain_model/Security.h"
 #include "spark/domain_model/Security.pb.h"
 #include "spark/domain_model/SecurityType.h"
 #include "spark/domain_model/Venue.h"
-#include "spark/format/table.h"
 #include "spark/md/Diag/ShmDumper.h"
 #include "spark/md/SecurityDB.h"
 #include "spark/md/SymbolLoader.h"
@@ -23,8 +23,7 @@ using bts::api::SecurityType;
 using bts::api::SecurityType_Name;
 using bts::api::Venue;
 using bts::api::venue_parse_smart;
-using bts::format::format_table;
-using bts::format::detail::TableSchema;
+using bts::core::TableFormatter;
 using bts::md::securitytype_of;
 using bts::md::SymbolLoader;
 using bts::md::db::load_metadata;
@@ -70,7 +69,7 @@ void print_securities(function<bool(const Security &)> predicate) {
 
 void print_db_securities(function<bool(const Security &)> predicate) {
    vector<Security> secs;
-   const TableSchema<Security> schema{
+   const TableFormatter::Schema<Security> schema{
        make_tuple(
            "id", 12, [](const Security &s) { return std::to_string(s.id()); }),
        make_tuple(
@@ -94,7 +93,7 @@ void print_db_securities(function<bool(const Security &)> predicate) {
       secs.push_back(sec);
    };
    load_securities([](auto) { return true; }, on_sec);
-   format_table(std::cout, secs, schema);
+   TableFormatter::format(std::cout, secs, schema);
 }
 
 template <class T, class ParseT = function<bool(const string &, T *)>>

src/app_sparkd/sparkd/execution/py_execution_gateway_adapter.cpp

@@ -23,7 +23,7 @@
 #include "spark/api/services/native_order_type.h"
 #include "spark/api/services/time_in_force.h"
 #include "spark/config/protos/jetfix.pb.h"
-#include "spark/core/time_stamp.h"
+#include "spark/core/TimeStamp.h"
 #include "spark/core/start_stop_state_machine.h"
 #include "spark/domain/tick_utils.h"
 #include "spark/domain_model/Security.h"

src/app_sparkd/sparkd/md/MarketDataGateway.cpp

@@ -4,7 +4,7 @@
 #include "spark/api/md/MarketDataConfig.pb.h"
 #include "spark/api/md/MarketDataConfigUtils.h"
 #include "spark/core/FMT.h"
-#include "spark/core/time_span.h"
+#include "spark/core/TimeSpan.h"
 #include "spark/domain/DomainSecurityUtils.h"
 #include "spark/domain/SecurityID.h"
 #include "spark/domain_model/Venue.h"
@@ -141,6 +141,5 @@ MarketDataGateway::MarketDataGateway(
 MarketDataGateway::~MarketDataGateway() = default;
 
 void MarketDataGateway::start() { me_->start(); }
-void MarketDataGateway::stop() { me_->stop(); }
 
 } // namespace bts::md

src/app_sparkd/sparkd/md/MarketDataGateway.h

@@ -29,7 +29,6 @@ class MarketDataGateway {
    ~MarketDataGateway();
 
    void start();
-   void stop();
 
  private:
    struct PImpl;

src/app_sparkd/sparkd/md/PyMarketDataSubscription.cpp

@@ -15,8 +15,8 @@
 #include "spark/api/domain/SecurityType.h"
 #include "spark/api/domain/Venue.h"
 #include "spark/api/md/PyMarketDataProviderConfig.pb.h"
-#include "spark/core/time_span.h"
-#include "spark/core/time_stamp.h"
+#include "spark/core/TimeSpan.h"
+#include "spark/core/TimeStamp.h"
 #include "spark/domain/AggressorSide.h"
 #include "spark/domain/FutureSecurity.h"
 #include "spark/domain/Security.h"

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

@@ -1,6 +1,6 @@
 #pragma once
 
-#include "spark/core/enum_macros.h"
+#include "spark/core/EnumMacros.h"
 #include "spark/core/FMT_IMPL.h"
 
 #include <fmt/format.h>

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

@@ -1,6 +1,6 @@
 #pragma once
 
-#include "spark/core/enum_macros.h"
+#include "spark/core/EnumMacros.h"
 #include "spark/core/FMT_IMPL.h"
 
 #include <fmt/format.h>

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

@@ -1,6 +1,6 @@
 #pragma once
 
-#include "spark/core/enum_macros.h"
+#include "spark/core/EnumMacros.h"
 #include "spark/core/FMT_IMPL.h"
 
 #include <fmt/format.h>

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

@@ -1,6 +1,6 @@
 #pragma once
 
-#include "spark/core/enum_macros.h"
+#include "spark/core/EnumMacros.h"
 #include "spark/core/FMT_IMPL.h"
 
 #include <fmt/format.h>

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

@@ -15,7 +15,7 @@ namespace bts::md::binance {
  */
 class BinanceMiniTickerEventType {
    // Unlike most binance-related enums, this one isn't defined via macros (in
-   // enum_macros.h). The macros assume the enum value names are identical to
+   // EnumMacros.h). The macros assume the enum value names are identical to
    // their json field counter-parts. Unfortunately, one of the json fields for
    // this enum, '24hrMiniTicker', isn't a legal name in cpp.
  public:

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

@@ -1,6 +1,6 @@
 #ifdef SPARK_ENABLE_CCAPI
 #include "BinanceMiniTickerUpdate.h"
-#include "spark/core/time_stamp.h"
+#include "spark/core/TimeStamp.h"
 #include "spark/domain/Price.h"
 #include "spark/domain/Quantity.h"
 #include "spark/domain/SecurityMarketDataStatistics.h"
@@ -21,11 +21,10 @@ using bts::domain::SecurityMarketDataStatistics;
 using bts::io::JsonReader;
 using std::map;
 using std::string;
-using std::string_view;
 using std::vector;
 
 static bool parse(
-    const map<string_view, string> &name_values,
+    const map<string, string> &name_values,
     vector<BinanceMiniTickerUpdate> &ticker_upds) {
    auto it = name_values.find(CCAPI_HTTP_BODY);
    if (it == name_values.end()) {

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

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

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

@@ -1,6 +1,6 @@
 #pragma once
 
-#include "spark/core/enum_macros.h"
+#include "spark/core/EnumMacros.h"
 #include "spark/core/FMT_IMPL.h"
 
 #include <fmt/format.h>

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

@@ -1,6 +1,6 @@
 #pragma once
 
-#include "spark/core/enum_macros.h"
+#include "spark/core/EnumMacros.h"
 #include "spark/core/FMT_IMPL.h"
 
 #include <fmt/format.h>

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

@@ -1,5 +1,5 @@
 #include "BinancePriceStatistics.h"
-#include "spark/core/time_stamp.h"
+#include "spark/core/TimeStamp.h"
 #include "spark/domain/Price.h"
 #include "spark/domain/Quantity.h"
 #include "spark/domain/SecurityMarketDataStatistics.h"
@@ -21,11 +21,10 @@ using bts::domain::SecurityMarketDataStatistics;
 using bts::io::JsonReader;
 using std::map;
 using std::string;
-using std::string_view;
 using std::vector;
 
 static bool parse(
-    const map<string_view, string> &name_values,
+    const map<string, string> &name_values,
     vector<BinancePriceStatistics> &result) {
    CCApiReader ccapi_reader{&name_values};
    nlohmann::json js;

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

@@ -1,6 +1,6 @@
 #pragma once
 
-#include "spark/core/time_stamp.h"
+#include "spark/core/TimeStamp.h"
 
 #include <cstdint>
 #include <nlohmann/json.hpp>

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

@@ -1,7 +1,7 @@
 #pragma once
 
 #include "spark/api/domain/Currency.pb.h"
-#include "spark/core/enum_macros.h"
+#include "spark/core/EnumMacros.h"
 #include "spark/core/FMT_IMPL.h"
 
 #include <boost/algorithm/string/case_conv.hpp>

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

@@ -180,7 +180,7 @@ struct BinanceSecurityRequestor::PImpl {
       for (const auto *msg : msgs) {
          bts_assert_msgf(
              msg->getType() == ccapi::Message::Type::GENERIC_PUBLIC_REQUEST,
-             "Received msg w/ unexpected type:\n{}", msg->toPrettyString());
+             "Received msg w/ unexpected type:\n{}", msg->toStringPretty());
          parse_secs(*msg, types, added_secs);
       }
       auto added_sec_descs = get_sec_descs(added_secs);
@@ -214,7 +214,7 @@ struct BinanceSecurityRequestor::PImpl {
          _FATAL_MSGF(
              "on_exchange_info. {} event was unexpected:\n{}",
              ccapi::Event::typeToString(event.getType()),
-             event.toPrettyString());
+             event.toStringPretty());
          break;
       }
       return true;

src/app_sparkd/sparkd/md/binance_coin_futures/binance_coin_futures_contract_type.h

@@ -1,6 +1,6 @@
 #pragma once
 
-#include "spark/core/enum_macros.h"
+#include "spark/core/EnumMacros.h"
 #include "spark/core/FMT_IMPL.h"
 
 #include <fmt/format.h>

src/app_sparkd/sparkd/md/binance_coin_futures/binance_coin_futures_message_type.h

@@ -1,6 +1,6 @@
 #pragma once
 
-#include "spark/core/enum_macros.h"
+#include "spark/core/EnumMacros.h"
 #include "spark/core/FMT_IMPL.h"
 
 #include <fmt/format.h>

src/app_sparkd/sparkd/md/binance_coin_futures/binance_coin_futures_price_statistics.cpp

@@ -21,11 +21,10 @@ using std::map;
 using std::pair;
 using std::size_t;
 using std::string;
-using std::string_view;
 using std::vector;
 
 static bool parse(
-    const map<string_view, string> &name_values,
+    const map<string, string> &name_values,
     vector<BinanceCoinFuturesPriceStatistics> &result) {
    CCApiReader ccapi_reader{&name_values};
    nlohmann::json js;

src/app_sparkd/sparkd/md/binance_coin_futures/binance_coin_futures_price_statistics_adaptor.cpp

@@ -60,7 +60,7 @@ void BinanceCoinFuturesPriceStatisticsAdaptor::on_messages(
          const auto n_total = n_hits + n_misses;
          _ERROR(
              "on_messages. Parse failed on {}/{} elements:\n{}", n_misses,
-             n_total, event.toPrettyString());
+             n_total, event.toStringPretty());
          continue;
       }
       for (const auto &bsr : binance_stats_recs) {

src/app_sparkd/sparkd/md/binance_coin_futures/binance_coin_futures_security_requestor.cpp

@@ -177,7 +177,7 @@ struct BinanceCoinFuturesSecurityRequestor::PImpl {
       for (const auto *msg : msgs) {
          bts_assert_msgf(
              msg->getType() == ccapi::Message::Type::GENERIC_PUBLIC_REQUEST,
-             "Received msg w/ unexpected type:\n{}", msg->toPrettyString());
+             "Received msg w/ unexpected type:\n{}", msg->toStringPretty());
          parse_secs(*msg, added_secs);
       }
       auto added_sec_descs = get_sec_descs(added_secs);
@@ -211,7 +211,7 @@ struct BinanceCoinFuturesSecurityRequestor::PImpl {
          _FATAL_MSGF(
              "on_exchange_info. {} event was unexpected:\n{}",
              ccapi::Event::typeToString(event.getType()),
-             event.toPrettyString());
+             event.toStringPretty());
          break;
       }
       return true;

src/app_sparkd/sparkd/md/binance_us/BinanceUsMessageType.h

@@ -1,6 +1,6 @@
 #pragma once
 
-#include "spark/core/enum_macros.h"
+#include "spark/core/EnumMacros.h"
 #include "spark/core/FMT_IMPL.h"
 
 #include <fmt/format.h>

src/app_sparkd/sparkd/md/binance_us/BinanceUsPriceStatistics.cpp

@@ -1,5 +1,5 @@
 #include "BinanceUsPriceStatistics.h"
-#include "spark/core/time_stamp.h"
+#include "spark/core/TimeStamp.h"
 #include "spark/domain/Price.h"
 #include "spark/domain/Quantity.h"
 #include "spark/domain/SecurityMarketDataStatistics.h"
@@ -23,11 +23,10 @@ using std::int64_t;
 using std::map;
 using std::ostream;
 using std::string;
-using std::string_view;
 using std::vector;
 
 static bool parse(
-    const map<string_view, string> &name_values,
+    const map<string, string> &name_values,
     vector<BinanceUsPriceStatistics> &result) {
    CCApiReader ccapi_reader{&name_values};
    nlohmann::json js;

src/app_sparkd/sparkd/md/binance_us/BinanceUsPriceStatistics.h

@@ -1,6 +1,6 @@
 #pragma once
 
-#include "spark/core/time_stamp.h"
+#include "spark/core/TimeStamp.h"
 
 #include <cstdint>
 #include <nlohmann/json.hpp>

src/app_sparkd/sparkd/md/binance_us/BinanceUsSecurityRequestor.cpp

@@ -69,10 +69,10 @@ static vector<string> get_sec_descs(const vector<Security> &sec_ptrs) {
 static void validate_response(const ccapi::Event &event, bool is_expected) {
    bts_assert_msgf(
        event.getType() == ccapi::Event::Type::RESPONSE,
-       "Received event w/ unexpected type:\n{}", event.toPrettyString());
+       "Received event w/ unexpected type:\n{}", event.toStringPretty());
    bts_assert_msgf(
        is_expected, "Received unexpected response from Binance US:\n{}",
-       event.toPrettyString());
+       event.toStringPretty());
 }
 
 struct BinanceUsSecurityRequestor::PImpl {
@@ -181,7 +181,7 @@ struct BinanceUsSecurityRequestor::PImpl {
       for (const auto *msg : msgs) {
          bts_assert_msgf(
              msg->getType() == ccapi::Message::Type::GENERIC_PUBLIC_REQUEST,
-             "Received msg w/ unexpected type:\n{}", msg->toPrettyString());
+             "Received msg w/ unexpected type:\n{}", msg->toStringPretty());
          parse_secs<T>(*msg, types, added_secs);
       }
       auto added_sec_descs = get_sec_descs(added_secs);
@@ -200,7 +200,7 @@ struct BinanceUsSecurityRequestor::PImpl {
        const CCApiCorrelationID & /*corr_id*/) {
       validate_response(event, exch_info_req_pending);
       LOG(md, Diag, "Received Binance US exchange info (spots):\n{}",
-          event.toPrettyString());
+          event.toStringPretty());
       update_securities<BinanceUsSpot>(msgs, {SEC_TYPE(SPOT)});
       exch_info_req_pending = false;
       if (external_req_pending && !is_busy()) {

src/app_sparkd/sparkd/md/binance_usds_futures/BinanceUsdsFuturesContractType.h

@@ -1,6 +1,6 @@
 #pragma once
 
-#include "spark/core/enum_macros.h"
+#include "spark/core/EnumMacros.h"
 #include "spark/core/FMT_IMPL.h"
 
 #include <fmt/format.h>

src/app_sparkd/sparkd/md/binance_usds_futures/BinanceUsdsFuturesMessageType.h

@@ -1,6 +1,6 @@
 #pragma once
 
-#include "spark/core/enum_macros.h"
+#include "spark/core/EnumMacros.h"
 #include "spark/core/FMT_IMPL.h"
 
 #include <fmt/format.h>

src/app_sparkd/sparkd/md/binance_usds_futures/BinanceUsdsFuturesPriceStatistics.cpp

@@ -21,11 +21,10 @@ using std::map;
 using std::pair;
 using std::size_t;
 using std::string;
-using std::string_view;
 using std::vector;
 
 static bool parse(
-    const map<string_view, string> &name_values,
+    const map<string, string> &name_values,
     vector<BinanceUsdsFuturesPriceStatistics> &result) {
    CCApiReader ccapi_reader{&name_values};
    nlohmann::json js;

src/app_sparkd/sparkd/md/binance_usds_futures/BinanceUsdsFuturesPriceStatisticsAdaptor.cpp

@@ -60,7 +60,7 @@ void BinanceUsdsFuturesPriceStatisticsAdaptor::on_messages(
          const auto n_total = n_hits + n_misses;
          _ERROR(
              "on_messages. Parse failed on {}/{} elements:\n{}", n_misses,
-             n_total, event.toPrettyString());
+             n_total, event.toStringPretty());
          continue;
       }
       for (const auto &bsr : binance_stats_recs) {

src/app_sparkd/sparkd/md/binance_usds_futures/BinanceUsdsFuturesSecurityRequestor.cpp

@@ -178,7 +178,7 @@ struct BinanceUsdsFuturesSecurityRequestor::PImpl {
       for (const auto *msg : msgs) {
          bts_assert_msgf(
              msg->getType() == ccapi::Message::Type::GENERIC_PUBLIC_REQUEST,
-             "Received msg w/ unexpected type:\n{}", msg->toPrettyString());
+             "Received msg w/ unexpected type:\n{}", msg->toStringPretty());
          parse_secs(*msg, added_secs);
       }
       auto added_sec_descs = get_sec_descs(added_secs);
@@ -212,7 +212,7 @@ struct BinanceUsdsFuturesSecurityRequestor::PImpl {
          _FATAL_MSGF(
              "on_exchange_info. {} event was unexpected:\n{}",
              ccapi::Event::typeToString(event.getType()),
-             event.toPrettyString());
+             event.toStringPretty());
          break;
       }
       return true;

src/app_sparkd/sparkd/md/ccapi/CCApiCorrelationID.h

@@ -1,8 +1,8 @@
 #pragma once
 
-#include "spark/algorithm/string/find.h"
-#include "spark/algorithm/string/join.h"
 #include "spark/core/FMT_IMPL.h"
+#include "spark/core/StringFinder.h"
+#include "spark/core/StringJoiner.h"
 #include "spark/utils/bts_assert.h"
 
 #include <boost/algorithm/string/split.hpp>
@@ -119,7 +119,7 @@ inline CCApiCorrelationID CCApiCorrelationID::from_string(
 template <class... Parts>
 inline CCApiCorrelationID CCApiCorrelationID::from_parts(Parts... parts) {
    // TODO: create the parts vector/string simultaneously (more efficient)
-   auto parts_str = bts::algorithm::join(delimiter(), parts...);
+   auto parts_str = core::StringJoiner::join(delimiter(), parts...);
    return from_string(parts_str);
 }
 
@@ -231,7 +231,7 @@ inline void CCApiCorrelationID::advance_(std::size_t len) {
       return;
    }
    cursor_ += len;
-   auto idx = bts::algorithm::find_nth(str_, '.', len);
+   auto idx = core::StringFinder::find(str_, '.', len);
    str_ = str_.substr(idx);
 }
 

src/app_sparkd/sparkd/md/ccapi/CCApiDepthWalker.cpp

@@ -8,7 +8,7 @@ namespace bts::md {
 using bts::domain::Quantity;
 
 bool CCApiDepthWalker::parse_field(
-    const std::string_view &name, const std::string &value) {
+    const std::string &name, const std::string &value) {
    bts_assert(depth_ptr);
    auto &depth = *depth_ptr;
    if (name == price_key) {

src/app_sparkd/sparkd/md/ccapi/CCApiDepthWalker.h

@@ -15,7 +15,7 @@ class CCApiDepthWalker {
    std::size_t price_idx = 0;
    std::size_t qty_idx = 0;
 
-   bool parse_field(const std::string_view &name, const std::string &value);
+   bool parse_field(const std::string &name, const std::string &value);
 };
 
 } // namespace bts::md

src/app_sparkd/sparkd/md/ccapi/CCApiEventDemuxer.cpp

@@ -57,7 +57,7 @@ CCApiEventDemuxer::CCApiEventDemuxer(std::size_t corr_id_len)
 
 CCApiEventDemuxer::~CCApiEventDemuxer() = default;
 
-void CCApiEventDemuxer::processEvent(
+bool CCApiEventDemuxer::processEvent(
     const ccapi::Event &event, ccapi::Session *session) {
    unordered_map<CCApiCorrelationID, vector<const ccapi::Message *>>
        demuxed_msgs;
@@ -91,6 +91,7 @@ void CCApiEventDemuxer::processEvent(
          hnd.on_messages(event, session, msgs, hnd_corr_id);
       }
    }
+   return true;
 }
 
 void CCApiEventDemuxer::on_messages(

src/app_sparkd/sparkd/md/ccapi/CCApiEventDemuxer.h

@@ -19,7 +19,7 @@ class CCApiEventDemuxer : public ccapi::EventHandler,
    explicit CCApiEventDemuxer(std::size_t corr_id_len = 0);
    ~CCApiEventDemuxer() override;
 
-   void processEvent(
+   bool processEvent(
        const ccapi::Event &event, ccapi::Session *session) override;
 
    void on_messages(

src/app_sparkd/sparkd/md/ccapi/CCApiEventLogger.h

@@ -10,7 +10,7 @@ class CCApiRequestor_Handler : public ccapi::EventHandler {
  public:
    bool processEvent(
        const ccapi::Event &event, ccapi::Session * /*session*/) override {
-      LOG(md, Info, "on ccapi event: {}", event.toPrettyString(2, 2));
+      LOG(md, Info, "on ccapi event: {}", event.toStringPretty(2, 2));
       return true;
    }
 };

src/app_sparkd/sparkd/md/ccapi/CCApiJsonRpcRequest.h

@@ -40,7 +40,7 @@ inline ccapi::Subscription CCApiJsonRpcRequest::rpc_sub(
     const io::websocket::JsonRpcRequest &req,
     std::string venue_name,
     const CCApiCorrelationID &corr_id) {
-   return ccapi::Subscription{
+   return {
        std::move(venue_name), "", CCAPI_GENERIC_PUBLIC_SUBSCRIPTION,
        req.to_string(), corr_id.to_string()};
 }

src/app_sparkd/sparkd/md/ccapi/CCApiJsonRpcRouter.cpp

@@ -38,7 +38,7 @@ void CCApiJsonRpcRouter::on_messages(
       BTS_FATAL_MSGF(
           "Exception occurred handling rpc "
           "message.\n***ex***:{}\n***event***:\n{}",
-          msg, event.toPrettyString());
+          msg, event.toStringPretty());
    });
 }
 

src/app_sparkd/sparkd/md/ccapi/CCApiMarketDepthAdaptor.cpp

@@ -45,7 +45,7 @@ static bool parse(const ccapi::Message &msg, domain::Book &book) {
                 ask_walker.parse_field(name, value);
          } catch (const std::exception &ex) {
             LOG(md, Fatal, "Error parsing depth update: {}\nmsg:\n{}",
-                std::string{ex.what()}, msg.toPrettyString());
+                std::string{ex.what()}, msg.toStringPretty());
             bts_assert_msg(false, "Fatal error parsing depth update.");
          }
       }

src/app_sparkd/sparkd/md/ccapi/CCApiMessageLogger.cpp

@@ -1,6 +1,6 @@
 #include "CCApiMessageLogger.h"
-#include "spark/algorithm/string/join.h"
-#include "spark/core/time_stamp.h"
+#include "spark/core/StringJoiner.h"
+#include "spark/core/TimeStamp.h"
 
 #include <fmt/format.h>
 #include <range/v3/view.hpp>
@@ -9,7 +9,7 @@
 
 namespace bts::md {
 
-using bts::algorithm::join;
+using bts::core::StringJoiner;
 using std::string;
 
 CCApiMessageLogger::CCApiMessageLogger(std::string name)
@@ -22,9 +22,9 @@ void CCApiMessageLogger::on_messages(
     const CCApiCorrelationID &corr_id) {
 
    auto msg_strs = msgs | ranges::views::transform([](const ccapi::Message *m) {
-                      return m->toPrettyString();
+                      return m->toStringPretty();
                    });
-   auto msgs_str = join(msg_strs.begin(), msg_strs.end(), "\n");
+   auto msgs_str = StringJoiner::join(msg_strs.begin(), msg_strs.end(), "\n");
    LOG(md, Info, "Received CCApi messages. name: {}; corr_id: {}; msgs:\n{}",
        name_, corr_id, msgs_str);
 }

src/app_sparkd/sparkd/md/ccapi/CCApiReader.h

@@ -15,7 +15,7 @@ namespace bts::md {
 
 class CCApiReader {
  public:
-   using NameValues = std::map<std::string_view, std::string>;
+   using NameValues = std::map<std::string, std::string>;
 
    explicit CCApiReader(const NameValues *name_values);
 

src/app_sparkd/sparkd/md/ccapi/CCApiTradeAdaptor.cpp

@@ -1,5 +1,5 @@
 #include "CCApiTradeAdaptor.h"
-#include "spark/core/time_stamp.h"
+#include "spark/core/TimeStamp.h"
 #include "spark/domain/SecurityID.h"
 #include "spark/domain/TradeRecord.h"
 #include "spark/utils/bts_fatal.h"
@@ -39,7 +39,7 @@ void CCApiTradeAdaptor::on_messages(
              ccapi::Message::typeToString(msg->getType()));
          return;
       }
-      LOG(md, Diag, "Received CCApi trade msg:\n{}", msg->toPrettyString());
+      LOG(md, Diag, "Received CCApi trade msg:\n{}", msg->toStringPretty());
       auto ts = TimeStamp::from_int64(
           msg->getTimeReceived().time_since_epoch().count());
       auto sec_desc = corr_id.peek_string();
@@ -54,7 +54,7 @@ void CCApiTradeAdaptor::on_messages(
    }
    if (trades.empty()) {
       LOG(md, Warn, "Received ccapi trade msg with no trades:\n{}",
-          event.toPrettyString());
+          event.toStringPretty());
       return;
    }
    std::vector<const domain::TradeRecord *> trade_ptrs;

src/app_sparkd/sparkd/md/coinbase/CoinbaseMessageType.h

@@ -1,6 +1,6 @@
 #pragma once
 
-#include "spark/core/enum_macros.h"
+#include "spark/core/EnumMacros.h"
 #include "spark/core/FMT_IMPL.h"
 
 #include <fmt/format.h>

src/app_sparkd/sparkd/md/coinbase/CoinbasePriceStatistics.cpp

@@ -1,5 +1,5 @@
 #include "CoinbasePriceStatistics.h"
-#include "spark/core/time_stamp.h"
+#include "spark/core/TimeStamp.h"
 #include "spark/domain/Price.h"
 #include "spark/domain/Quantity.h"
 #include "spark/domain/SecurityMarketDataStatistics.h"
@@ -21,11 +21,10 @@ using std::int64_t;
 using std::map;
 using std::ostream;
 using std::string;
-using std::string_view;
 using std::vector;
 
 static bool parse(
-    const map<string_view, string> &name_values,
+    const map<string, string> &name_values,
     vector<CoinbasePriceStatistics> &result) {
    auto it = name_values.find(CCAPI_HTTP_BODY);
    const auto &body = it->second;

src/app_sparkd/sparkd/md/coinbase/CoinbasePriceStatistics.h

@@ -1,6 +1,6 @@
 #pragma once
 
-#include "spark/core/time_stamp.h"
+#include "spark/core/TimeStamp.h"
 
 #include <cstdint>
 #include <nlohmann/json.hpp>

src/app_sparkd/sparkd/md/coinbase/CoinbaseSecurityRequestor.cpp

@@ -62,10 +62,10 @@ static vector<string> get_sec_descs(const vector<Security> &sec_ptrs) {
 static void validate_response(const ccapi::Event &event, bool is_expected) {
    bts_assert_msgf(
        event.getType() == ccapi::Event::Type::RESPONSE,
-       "Received event w/ unexpected type:\n{}", event.toPrettyString());
+       "Received event w/ unexpected type:\n{}", event.toStringPretty());
    bts_assert_msgf(
        is_expected, "Received unexpected response from Coinbase:\n{}",
-       event.toPrettyString());
+       event.toStringPretty());
 }
 
 struct CoinbaseSecurityRequestor::PImpl {
@@ -172,7 +172,7 @@ struct CoinbaseSecurityRequestor::PImpl {
       for (const auto *msg : msgs) {
          bts_assert_msgf(
              msg->getType() == ccapi::Message::Type::GET_INSTRUMENTS,
-             "Received msg w/ unexpected type:\n{}", msg->toPrettyString());
+             "Received msg w/ unexpected type:\n{}", msg->toStringPretty());
          parse_secs<T>(*msg, types, added_secs);
       }
       auto added_sec_descs = get_sec_descs(added_secs);
@@ -191,7 +191,7 @@ struct CoinbaseSecurityRequestor::PImpl {
        const CCApiCorrelationID & /*corr_id*/) {
       validate_response(event, exch_info_req_pending);
       LOG(md, Diag, "Received Coinbase exchange info (spots):\n{}",
-          event.toPrettyString());
+          event.toStringPretty());
       std::cout << msgs.size();
       update_securities<CoinbaseSpot>(msgs, {SEC_TYPE(SPOT)});
       exch_info_req_pending = false;

src/app_sparkd/sparkd/md/databento/DatabentoDataHandler.cpp

@@ -23,7 +23,7 @@
 #include "spark/logging/logging.h"
 #include "spark/md/IMarketDataListener.h"
 #include "spark/persistence/IDGenerator.h"
-#include "spark/core/time_stamp.h"
+#include "spark/core/TimeStamp.h"
 #include "spark/utils/TimeStamp.h"
 #include "spark/utils/bts_assert.h"
 #include "spark/utils/bts_chrono.h"

src/app_sparkd/sparkd/md/deribit/DeribitCombo.h

@@ -1,6 +1,6 @@
 #pragma once
 
-#include "spark/core/time_stamp.h"
+#include "spark/core/TimeStamp.h"
 #include "sparkd/md/deribit/DeribitComboLeg.h"
 
 #include <cstdint>

src/app_sparkd/sparkd/md/deribit/DeribitInstrument.h

@@ -1,6 +1,6 @@
 #pragma once
 
-#include "spark/core/time_stamp.h"
+#include "spark/core/TimeStamp.h"
 #include "sparkd/md/deribit/DeribitInstrumentKind.h"
 
 #include <nlohmann/json_fwd.hpp>

src/app_sparkd/sparkd/md/deribit/DeribitInstrumentKind.h

@@ -3,7 +3,7 @@
 #include "spark/algorithm/StringCaseConverter.h"
 #include "spark/api/domain/SecurityType.pb.h"
 #include "spark/collections/HASH_IMPL.h"
-#include "spark/core/enum_macros.h"
+#include "spark/core/EnumMacros.h"
 #include "spark/core/FMT_IMPL.h"
 #include "spark/domain_model/Security.pb.h"
 

src/app_sparkd/sparkd/md/deribit/DeribitInstrumentState.h

@@ -1,7 +1,7 @@
 #pragma once
 
 #include "spark/core/FMT_IMPL.h"
-#include "spark/core/time_stamp.h"
+#include "spark/core/TimeStamp.h"
 #include "sparkd/md/deribit/DeribitInstrumentStateType.h"
 
 #include <fmt/format.h>

src/app_sparkd/sparkd/md/deribit/DeribitInstrumentStateType.h

@@ -1,6 +1,6 @@
 #pragma once
 
-#include "spark/core/enum_macros.h"
+#include "spark/core/EnumMacros.h"
 #include "spark/core/FMT_IMPL.h"
 
 #include <fmt/format.h>

src/app_sparkd/sparkd/md/deribit/DeribitMessageType.h

@@ -1,6 +1,6 @@
 #pragma once
 
-#include "spark/core/enum_macros.h"
+#include "spark/core/EnumMacros.h"
 #include "spark/core/FMT_IMPL.h"
 
 #include <fmt/format.h>

src/app_sparkd/sparkd/md/deribit/DeribitSubscription.cpp

@@ -6,7 +6,7 @@
 #include "spark/collections/Remove.h"
 #include "spark/collections/Transform.h"
 #include "spark/core/FMT.h"
-#include "spark/core/time_stamp.h"
+#include "spark/core/TimeStamp.h"
 #include "spark/domain/Security.h"
 #include "spark/domain/SecurityCriteria.h"
 #include "spark/domain_model/Security.pb.h"
@@ -99,7 +99,7 @@ static ccapi::Subscription mk_deribit_sub(Channels channels, size_t sub_id) {
        "public/subscribe",
        {{"channels", vector<string>{channels.begin(), channels.end()}}},
        JsonRpcId::create()};
-   return ccapi::Subscription{
+   return {
        CCAPI_VENUE_NAME, "", CCAPI_GENERIC_PUBLIC_SUBSCRIPTION, req.to_string(),
        FMT("{}.SUBSCRIPTION_{}", VENUE_NAME, sub_id)};
 }

src/app_sparkd/sparkd/md/deribit/DeribitTradeAdaptor.cpp

@@ -1,5 +1,5 @@
 #include "DeribitTradeAdaptor.h"
-#include "spark/core/time_stamp.h"
+#include "spark/core/TimeStamp.h"
 #include "spark/domain/AggressorSide.h"
 #include "spark/domain/SecurityID.h"
 #include "spark/domain/TradeRecord.h"

src/app_sparkd/sparkd/md/ftx/FTXFuture.cpp

@@ -1,8 +1,8 @@
 #include "FTXFuture.h"
 #include "spark/api/domain/Venue.pb.h"
+#include "spark/core/TableFormatter.h"
 #include "spark/domain/FutureSecurity.h"
 #include "spark/domain/Security.h"
-#include "spark/format/table.h"
 #include "spark/io/JsonReader.h"
 #include "spark/logging/logging.h"
 
@@ -14,10 +14,9 @@
 
 namespace bts::md::ftx {
 
+using bts::core::TableFormatter;
 using bts::domain::FutureSecurity;
 using bts::domain::Security;
-using bts::format::format_table_kv_rows;
-using bts::format::detail::TableSchemaBuilder;
 using bts::io::JsonReader;
 using std::string;
 using std::stringstream;
@@ -134,7 +133,7 @@ void from_json(const nlohmann::json &j, FTXFuture &fut) {
 
 string FTXFuture::to_string_pretty() const {
    stringstream out;
-   TableSchemaBuilder<FTXFuture> schema;
+   TableFormatter::SchemaBuilder<FTXFuture> schema;
    schema.add("ask", [](const auto &f) { return f.ask; });
    schema.add("bid", [](const auto &f) { return f.bid; });
    schema.add("change_1h", [](const auto &f) { return f.change_1h; });
@@ -166,7 +165,7 @@ string FTXFuture::to_string_pretty() const {
    schema.add("underlying", [](const auto &f) { return f.underlying; });
    schema.add("upper_bound", [](const auto &f) { return f.upper_bound; });
    schema.add("type", [](const auto &f) { return f.type; });
-   format_table_kv_rows(out, *this, schema.result);
+   TableFormatter::format_record(out, *this, schema.result);
    return out.str();
 }
 

src/app_sparkd/sparkd/md/ftx/FTXFuture.h

@@ -1,6 +1,6 @@
 #pragma once
 
-#include "spark/core/time_stamp.h"
+#include "spark/core/TimeStamp.h"
 #include "sparkd/md/ftx/FTXSecurityType.h"
 
 #include <nlohmann/json.hpp>

src/app_sparkd/sparkd/md/ftx/FTXHistoricalPriceRecord.cpp

@@ -19,11 +19,10 @@ using bts::domain::SecurityMarketDataStatistics;
 using bts::io::JsonReader;
 using std::map;
 using std::string;
-using std::string_view;
 using std::vector;
 
 static bool parse(
-    const map<string_view, string> &name_values,
+    const map<string, string> &name_values,
     std::vector<FTXHistoricalPriceRecord> &price_recs) {
    CCApiReader ccapi_reader{&name_values};
    nlohmann::json js;

src/app_sparkd/sparkd/md/ftx/FTXHistoricalPriceRecord.h

@@ -1,6 +1,6 @@
 #pragma once
 
-#include "spark/core/time_stamp.h"
+#include "spark/core/TimeStamp.h"
 #include "spark/domain/SecurityID.h"
 
 #include <nlohmann/json.hpp>

src/app_sparkd/sparkd/md/ftx/FTXMarket.cpp

@@ -1,7 +1,7 @@
 #include "FTXMarket.h"
 #include "spark/api/domain/Venue.pb.h"
+#include "spark/core/TableFormatter.h"
 #include "spark/domain/Security.h"
-#include "spark/format/table.h"
 #include "spark/io/JsonReader.h"
 #include "spark/logging/logging.h"
 #include "spark/utils/bts_fatal.h"
@@ -19,9 +19,8 @@ namespace bts::md::ftx {
 #define SEC_TYPE(x) bts::api::domain::SECURITY_TYPE_##x
 
 using bts::api::domain::Venue;
+using bts::core::TableFormatter;
 using bts::domain::Security;
-using bts::format::format_table_kv_rows;
-using bts::format::detail::TableSchemaBuilder;
 using bts::io::JsonReader;
 using std::ostream;
 using std::string;
@@ -164,7 +163,7 @@ void from_json(const nlohmann::json &j, FTXMarket &mkt) {
 }
 
 ostream &operator<<(ostream &out, const FTXMarket &mkt) noexcept {
-   TableSchemaBuilder<FTXMarket> schema;
+   core::TableFormatter::SchemaBuilder<FTXMarket> schema;
    schema.add("name", [](const auto &s) { return s.name; });
    schema.add("base_currency", [](const auto &s) { return s.base_currency; });
    schema.add("quote_currency", [](const auto &s) { return s.quote_currency; });
@@ -195,7 +194,7 @@ ostream &operator<<(ostream &out, const FTXMarket &mkt) noexcept {
       return s.large_order_threshold;
    });
    schema.add("is_etf_market", [](const auto &s) { return s.is_etf_market; });
-   format_table_kv_rows(out, mkt, schema.result);
+   TableFormatter::format_record(out, mkt, schema.result);
    return out;
 }
 

src/app_sparkd/sparkd/md/ftx/FTXMessageType.h

@@ -1,6 +1,6 @@
 #pragma once
 
-#include "spark/core/enum_macros.h"
+#include "spark/core/EnumMacros.h"
 #include "spark/core/FMT_IMPL.h"
 
 #include <fmt/format.h>

src/app_sparkd/sparkd/md/ftx/FTXSecurityRequestor.cpp

@@ -84,10 +84,10 @@ static vector<string> get_sec_descs(const vector<Security> &sec_ptrs) {
 static void validate_response(const ccapi::Event &event, bool is_expected) {
    bts_assert_msgf(
        event.getType() == ccapi::Event::Type::RESPONSE,
-       "Received event w/ unexpected type:\n{}", event.toPrettyString());
+       "Received event w/ unexpected type:\n{}", event.toStringPretty());
    bts_assert_msgf(
        is_expected, "Received unexpected response from FTX:\n{}",
-       event.toPrettyString());
+       event.toStringPretty());
 }
 
 struct FTXSecurityRequestor::PImpl {
@@ -200,7 +200,7 @@ struct FTXSecurityRequestor::PImpl {
       for (const auto *msg : msgs) {
          bts_assert_msgf(
              msg->getType() == ccapi::Message::Type::GENERIC_PUBLIC_REQUEST,
-             "Received msg w/ unexpected type:\n{}", msg->toPrettyString());
+             "Received msg w/ unexpected type:\n{}", msg->toStringPretty());
          parse_secs<T>(*msg, types, added_secs);
       }
       auto added_sec_descs = get_sec_descs(added_secs);
@@ -218,7 +218,7 @@ struct FTXSecurityRequestor::PImpl {
        const vector<const ccapi::Message *> &msgs,
        const CCApiCorrelationID & /*corr_id*/) {
       validate_response(event, mkts_req_pending);
-      LOG(md, Diag, "Received FTX markets:\n{}", event.toPrettyString());
+      LOG(md, Diag, "Received FTX markets:\n{}", event.toStringPretty());
       update_securities<FTXMarket>(msgs, {SEC_TYPE(SPOT)});
       mkts_req_pending = false;
       if (external_req_pending && !is_busy()) {
@@ -236,7 +236,7 @@ struct FTXSecurityRequestor::PImpl {
        const vector<const ccapi::Message *> &msgs,
        const CCApiCorrelationID & /*corr_id*/) {
       validate_response(event, futs_req_pending);
-      LOG(md, Diag, "Received FTX futures:\n{}", event.toPrettyString());
+      LOG(md, Diag, "Received FTX futures:\n{}", event.toStringPretty());
       update_securities<FTXFuture>(msgs, {SEC_TYPE(FUTURE)});
       futs_req_pending = false;
       if (external_req_pending && !is_busy()) {

src/app_sparkd/sparkd/md/ftx/FTXSecurityType.h

@@ -1,7 +1,7 @@
 #pragma once
 
 #include "spark/api/domain/SecurityType.pb.h"
-#include "spark/core/enum_macros.h"
+#include "spark/core/EnumMacros.h"
 #include "spark/core/FMT_IMPL.h"
 
 #include <nlohmann/json.hpp>

src/app_sparkd/sparkd/md/market_data_session.cpp

@@ -1,84 +0,0 @@
-#include "market_data_session.h"
-#include "python_interop/plugin_manager.h"
-#include "spark/api/md/MarketDataConfig.pb.h"
-#include "spark/config/config.h"
-#include "spark/config/protos/market_data.pb.h"
-#include "spark/core/time_stamp.h"
-#include "spark/persistence/IDGenerator.h"
-#include "spark/utils/datetime.h"
-#include "spark/utils/thread_manager.h"
-#include "sparkd/cmds/md/cmds/Cleanup.h"
-#include "sparkd/md/MarketDataGateway.h"
-
-namespace bts::md {
-
-using bts::api::md::MarketDataConfig;
-using bts::cmds::md::cmds::Cleanup;
-using bts::core::TimeStamp;
-using bts::persistence::IDGenerator;
-using bts::threading::configure_current_thread;
-using std::shared_ptr;
-using std::thread;
-using std::this_thread::sleep_for;
-
-static void boot_sleep() {
-   auto conf = config::get<bts::md::MDBootDelayWindow>();
-   const auto *start = &conf.standard_start();
-   const auto *end = &conf.standard_end();
-   if (datetime::is_local_time_sunday()) {
-      start = &conf.sunday_start();
-      end = &conf.sunday_end();
-   }
-   int hh_start = 0, mm_start = 0, hh_end = 0, mm_end = 0;
-   sscanf(start->c_str(), "%i:%i", &hh_start, &mm_start);
-   sscanf(end->c_str(), "%i:%i", &hh_end, &mm_end);
-   int64_t sleep_secs =
-       datetime::is_local_time_between(hh_start, mm_start, hh_end, mm_end)
-           ? conf.delay_seconds()
-           : 2;
-   LOG(md, Info, R"([md_session_boot_sleep.json]{{ "dur": "{}s", "ts": {} }})",
-       sleep_secs, TimeStamp::now_ns());
-   sleep_for(std::chrono::seconds(sleep_secs));
-}
-
-struct MarketDataSession::PImpl {
-   volatile bool is_running = true;
-   thread run_thread;
-
-   PImpl(shared_ptr<python_interop::PluginManager> py_worker) {
-      LOG(md, Info, R"([md_session_starting.json]{{ "ts": {} }})",
-          TimeStamp::now_ns());
-      Cleanup::execute();
-      run_thread = thread([&]() {
-         configure_current_thread("MarketDataGateway");
-         LOG(md, Info, R"([md_session_started.json]{{ "ts": {} }})",
-             TimeStamp::now_ns());
-         MarketDataGateway gw{
-             config::get<MarketDataConfig>(), IDGenerator::instance_ptr(),
-             py_worker};
-         gw.start();
-         while (is_running) {
-            std::this_thread::sleep_for(std::chrono::seconds(1));
-         }
-         gw.stop();
-         LOG(md, Info, R"([md_session_stopped.json]{{ "ts": {} }})",
-             TimeStamp::now_ns());
-      });
-      boot_sleep();
-   }
-
-   ~PImpl() {
-      LOG(md, Info, R"([md_session_stopping.json]{{ "ts": {} }})",
-          TimeStamp::now_ns());
-      is_running = false;
-      run_thread.join();
-   }
-};
-
-MarketDataSession::MarketDataSession(
-    shared_ptr<python_interop::PluginManager> py_worker)
-    : me(std::make_unique<PImpl>(std::move(py_worker))) {}
-
-MarketDataSession::~MarketDataSession() = default;
-
-} // namespace bts::md

src/app_sparkd/sparkd/md/market_data_session.h

@@ -1,19 +0,0 @@
-#pragma once
-
-#include "python_interop/plugin_manager.h"
-
-#include <memory>
-
-namespace bts::md {
-
-class MarketDataSession {
- public:
-   MarketDataSession(std::shared_ptr<python_interop::PluginManager> py_worker);
-   ~MarketDataSession();
-
- private:
-   struct PImpl;
-   std::unique_ptr<PImpl> me;
-};
-
-} // namespace bts::md

src/app_sparkd/sparkd/md/okx/OkxChannelDemuxer.cpp

@@ -131,7 +131,7 @@ void OkxChannelDemuxer::on_messages(
       if (me_->process_subscription_error(payload_j)) {
          continue;
       }
-      _ERROR("on_messages. Unknown payload type:\n{}", event.toPrettyString());
+      _ERROR("on_messages. Unknown payload type:\n{}", event.toStringPretty());
    }
    me_->flush();
 }

src/app_sparkd/sparkd/md/okx/OkxInstrument.h

@@ -1,6 +1,6 @@
 #pragma once
 
-#include "spark/core/time_stamp.h"
+#include "spark/core/TimeStamp.h"
 #include "sparkd/md/okx/OkxInstrumentStatus.h"
 #include "sparkd/md/okx/OkxInstrumentType.h"
 

src/app_sparkd/sparkd/md/okx/OkxInstrumentStatus.h

@@ -1,7 +1,7 @@
 #pragma once
 
 #include "spark/algorithm/StringCaseConverter.h"
-#include "spark/core/enum_macros.h"
+#include "spark/core/EnumMacros.h"
 #include "spark/core/FMT_IMPL.h"
 
 #include <fmt/format.h>

src/app_sparkd/sparkd/md/okx/OkxInstrumentStore.cpp

@@ -1,6 +1,6 @@
 #include "OkxInstrumentStore.h"
 #include "spark/api/domain/SecurityType.h"
-#include "spark/core/time_stamp.h"
+#include "spark/core/TimeStamp.h"
 #include "spark/domain/Security.h"
 #include "spark/domain_model/Security.pb.h"
 #include "spark/logging/Common.h"

src/app_sparkd/sparkd/md/okx/OkxInstrumentType.h

@@ -1,6 +1,6 @@
 #pragma once
 
-#include "spark/core/enum_macros.h"
+#include "spark/core/EnumMacros.h"
 #include "spark/core/FMT_IMPL.h"
 
 #include <fmt/format.h>

src/app_sparkd/sparkd/md/okx/OkxMessageType.h

@@ -1,6 +1,6 @@
 #pragma once
 
-#include "spark/core/enum_macros.h"
+#include "spark/core/EnumMacros.h"
 
 #include <fmt/format.h>
 #include <nlohmann/json.hpp>