fix buid: Don't goto ERROR just bc venv already exists

This seems like a typo

src/client/spark/AW2/Bus/SingleVolSurfaceGateway.cs

@@ -55,8 +55,10 @@
       private void _on_vss(bts.api.core.Update<VolSurfaceEntryImmutable, UInt64> update) {
          IEnumerable<VolSurfaceEntryImmutable> all_vss = update.all;
          var this_vs_vss = all_vss.Where(vs => vs.VsId == _volsurface_id);
-         bts.Log.Gen.info(String.Format("SingleVolSurfaceGateway._on_vss count:{0} vs_count:{1}",
-            all_vss.Count(), this_vs_vss.Count()));
+         bts.Log.Gen.info(
+            $"SingleVolSurfaceGateway._on_vss vs_count: {this_vs_vss.Count()} " +
+            $"all: {update.all.Count()} added: {update.added.Count()} " +
+            $"updated: {update.changed.Count()} removed: {update.removed.Count()}");
          if (this_vs_vss.Count() == 0) {
             return;
          }
@@ -110,8 +112,13 @@
                   "SingleVolSurfaceGateway._on_bb_data Time to commit volsurface {0} {1} msecs",
                   _volsurface_id, _timer.lastIntervalTime));
                var archived_theos = _get_theos();
-               if (archived_theos != null)
+               if (archived_theos != null) {
+                  _timer.start();
                   _analytics_service.archive_theos(archived_theos);
+                  _timer.stop();
+                  Log.Gen.info($"SingleVolSurfaceGateway._on_bb_data archiving theos vs_id: {_volsurface_id} size: " +
+                               $"{archived_theos.SecTheo.Count} time: {_timer.lastIntervalTime} msecs");
+               }
             } catch (Exception ex) {
                Log.Gen.warn(ex.ToString());
                _bb.post(SparkTags.MessageDisplayTag,

src/client/spark/AW2/Bus/VolSurfaceBlackboardManager.cs

@@ -270,6 +270,7 @@ namespace mbts.spark {
                                           new mbts.analytics.VolSurfaceNonVolInput(exp_param_map.exp_map.ToDictionary()));
          } else if (_in_bb_req_vol.Equals(tag)) {
             ClientArchivedVolRequest client_req = _bb.read_state(_in_bb_req_vol);
+            bts.Log.Ui.info($"VolSurfaceBusManager._on_bb_update ClientArchivedVolRequest: {client_req.ToString()}");
             ArchivedTheoRetrieval ret = client_req.ret;
             if (ret == null || ret.SecTup == null) {
                bts.Log.Ui.warn($"VolSurfaceBusManager malformed ArchivedTheoRetrieval");

src/client/spark/AW2/InterpTheosServerPublisher.cs

@@ -5,6 +5,7 @@ using bts.api;
 using bts.collections;
 using System.Linq;
 using bts.api.orderman;
+using bts.api.analytics;
 using bts;
 
 namespace mbts.spark {
@@ -23,6 +24,8 @@ namespace mbts.spark {
          _order_service = order_service;
          _bb.subscribe(SparkTags.AllInterpGreeks);
          _bb.subscribe(SparkTags.SystemContractsTag);
+         _bb.subscribe(SparkTags.CurrentUser);
+         _bb.subscribe(SparkTags.VolSurfaceUsageTag);
          _bb.data_callback += _on_data;
       }
 
@@ -55,6 +58,27 @@ namespace mbts.spark {
          return rv;
       }
 
+      private IReadOnlyDictionary<string, OptionTheoInterpolationData>
+         _filter_for_publish(IReadOnlyDictionary<string, OptionTheoInterpolationData> all_theos) {
+         // find the symbol vs vs_id, get the committed vs, see if it is
+         // owned by the current user, and only then publish it
+         HashSet<string> syms = new HashSet<string>();
+         VolSurfaceUsageListImmutable usage = _bb.read_state(SparkTags.VolSurfaceUsageTag);
+         string current_user = _bb.read_state(SparkTags.CurrentUser).name;
+         foreach (var u in usage.UsageList) {
+            VolSurfaceInput vs_input =
+               _bb.read_state(SparkTags.VolSurfaceCommittedInputTag(u.VsId));
+            if (vs_input != null) {
+               if (vs_input.surface_settings.core_settings.vs_manager ==
+                  current_user) {
+                  syms.Add(u.SecTup.Symbol);
+               }
+            }
+         }
+         return
+            all_theos.Where(kv => syms.Contains(kv.Key)).ToDictionary(kv => kv.Key, kv => kv.Value);
+      }
+
       private void _publish_data() {
          SystemContractData sc_data = _bb.read_state(SparkTags.SystemContractsTag, null);
          IReadOnlyDictionary<string, OptionTheoInterpolationData> all_theos =
@@ -64,6 +88,11 @@ namespace mbts.spark {
          if (sc_data == null || all_theos == null || all_drivers == null)
             return;
          _timer.start();
+         all_theos = _filter_for_publish(all_theos);
+         if (all_theos.Count == 0) {
+            bts.Log.Gen.info($"InterpTheosServerPublisher._publish_data all_theos.Count == 0");
+            return;
+         }
          IDictionary<UInt32, DriverMD> driver_map = _get_drivers(all_drivers, sc_data);
          IEnumerable<Tuple<SecurityImmutable, TheoPoint>> theo_map = _get_theos(all_theos, sc_data);
          // assemble it together and send it to the server
@@ -86,6 +115,7 @@ namespace mbts.spark {
             ptp.Vega = is_call ? tp.cvega : tp.pvega;
             ptp.Theta = is_call ? tp.c_theta : tp.p_theta;
             ptp.Rho = Double.NaN;
+            ptp.Vol = tp.vol;
             ptp_list.Add(ptp);
          }
          bts.api.orderman.PrecomputedTheos server_theos = new PrecomputedTheos();
@@ -96,7 +126,7 @@ namespace mbts.spark {
          _order_service.set_precomputed_theos(server_theos);
          _timer.stop();
          var pub_time = _timer.lastIntervalTime;
-         bts.Log.Gen.info($"InterpTheosServerPublisher._publish_data " +
+         bts.Log.Gen.info($"InterpTheosServerPublisher._publish_data symbols: {all_theos.Count} " +
                           $"size: {server_theos.Theos.Count} comp_time: {comp_time} msecs " +
                           $"pub_time: {pub_time}");
       }

src/lib/CMakeDeps_win.cmake

@@ -1 +1 @@
-set(sources api/analytics/ExpirationDriver.proto;api/analytics/TimeWeightedInterpolatedDriver.proto;api/apicore.proto;api/apigateway.proto;api/apioptions.proto;api/core/ApiClientInterop.cpp;api/core/ApiServerInterop.cpp;api/domain/Currency.proto;api/domain/SecurityCriteria.proto;api/domain/SecurityCriteriaClause.proto;api/domain/SecurityExpiration.proto;api/domain/SecurityMetadata.proto;api/domain/SecurityMetadataCollection.proto;api/domain/SecurityMetadataItem.proto;api/domain/SecurityType.proto;api/domain/TickDisplaySettings.proto;api/domain/TickFactors.proto;api/domain/Venue.proto;api/io/URL.proto;api/md/MarketDataSubscriptionParams.proto;api/md/binance/BinanceDepthStreamParams.proto;api/md/binance/BinanceMarketDataSubscriptionParams.proto;api/md/deribit/DeribitMarketDataSubscriptionParams.proto;api/md/ftx/FTXDepthStreamParams.proto;api/md/ftx/FTXMarketDataSubscriptionParams.proto;api/md/glbx/GlbxMarketDataSubscriptionParams.proto;api/md/huobi/HuobiMarketDataSubscriptionParams.proto;api/public/ApiServer.cpp;api/public/ClientChannel.cpp;api/public/QuickbusClientChannel.cpp;api/public/ServerChannel.cpp;api/public/ServiceContext.cpp;api/service_helpers/AdminServiceUtils.cpp;api/service_helpers/AlgoServiceUtils.cpp;api/service_helpers/PositionUtilities.cpp;api/services/AdminService.proto;api/services/AlgoDefinitionService.proto;api/services/AlgoService.proto;api/services/AnalyticsService.proto;api/services/ApiUnitTestService.proto;api/services/AuthService.proto;api/services/BasePositionService.proto;api/services/ErrorReportService.proto;api/services/HelloService.proto;api/services/MBOService.proto;api/services/MarginService.proto;api/services/MarketDataAuxiliaryService.proto;api/services/MarketDataRelay.proto;api/services/MarketDataService.proto;api/services/MoneyFileService.proto;api/services/OrderService.proto;api/services/OutOfProcessURLListener.proto;api/services/PersistenceService.proto;api/services/PositionService.proto;api/services/ShellDiagnosticService.proto;api/services/ShellLatencyService.proto;api/services/SyntheticSecurityService.proto;api/test/ApiTests.cpp;api/test/HelloServiceServer.cpp;config/ConfigCheck.cpp;config/ConfigDownloader.cpp;config/ConfigErrorCollector.cpp;config/ConfigLog.cpp;config/ConfigPropertyEvaluator.cpp;config/ConfigValueQueue.cpp;config/ConfigurationException.cpp;config/ConfigurationState.cpp;config/PreprocessInputStream.cpp;config/config.cpp;config/defaults/MarketData.conf;config/defaults/Ports.conf;config/defaults/SecurityMetadataCollection.conf;config/defaults/Symbols.conf;config/defaults/cme_reference_data.conf;config/defaults/edge.conf;config/defaults/jetfix.conf;config/defaults/keycode.conf;config/defaults/logging.conf;config/defaults/orderman_cfg.conf;config/defaults/orders_fills_cleanup.conf;config/defaults/position_cfg.conf;config/defaults/span.conf;config/defaults/spreader.conf;config/defaults/testkeys.conf;config/defaults/threads.conf;config/defaults/toggles.conf;config/extensions/extensions.proto;config/invalid/invalid.proto;config/protos/algo_types.proto;config/protos/cme_reference_data.proto;config/protos/exchange_sim_cfg.proto;config/protos/jetfix.proto;config/protos/keycode.proto;config/protos/logging.proto;config/protos/market_data.proto;config/protos/mercury_cfg.proto;config/protos/orderman_cfg.proto;config/protos/orders_fills_cleanup.proto;config/protos/perflog_cfg.proto;config/protos/ports.proto;config/protos/position_cfg.proto;config/protos/quickbus_cfg.proto;config/protos/span.proto;config/protos/spartan.proto;config/protos/spreader.proto;config/protos/subprocess.proto;config/protos/symbols.proto;config/protos/system.proto;config/protos/testkeys.proto;config/protos/threads.proto;config/protos/toggles.proto;config/protos/ui.proto;config/tests/HTTPServer.cpp;config/tests/config_ref_tests.cpp;config/tests/config_tests.cpp;config/text_format/text_format.cpp;domain_model/Security.cpp;domain_model/Security.proto;domain_model/SecurityTests.cpp;domain_model/sod.cpp;domain_model/tests/domain_tests.cpp;expression/BooleanGrammar.cpp;expression/ExpressionEvaluator.cpp;expression/ParserExceptions.cpp;expression/Value.cpp;expression/tests/expression_parser_tests.cpp;io/QueryParams.cpp;io/URL.cpp;io/http/CurlHandle.cpp;io/http/HTTPClient.cpp;io/http/HTTPHeaders.cpp;io/http/HTTPRequest.cpp;io/http/HTTPResponse.cpp;io/websocket/WebsocketClient.cpp;io/websocket/WebsocketClientConfiguration.cpp;io/websocket/WebsocketError.cpp;io/websocket/WebsocketMessage.cpp;io/websocket/WebsocketSSLClientInternal.cpp;io/websocket/tests/WebsocketClientTests.cpp;logging/LoggerHub.cpp;logging/Sink.cpp;logging/logging.cpp;logging/tests/logging_tests.cpp;md/CCApiDepthWalker.cpp;md/CCApiListenerAdaptor.cpp;md/CCApiRequestor.cpp;md/Diag/MDPeriodicDumper.cpp;md/Diag/ShmDumper.cpp;md/MBOProcessor.cpp;md/MBOServiceServer.cpp;md/MDClient.cpp;md/MDConsumer.cpp;md/MDConsumerHandle.cpp;md/MDConsumerInterop.cpp;md/MDProducer.cpp;md/MDProducerHandle.cpp;md/MarketDataError.cpp;md/MarketDataGateway.cpLine truncated
+set(sources api/analytics/ExpirationDriver.proto;api/analytics/TimeWeightedInterpolatedDriver.proto;api/apicore.proto;api/apigateway.proto;api/apioptions.proto;api/core/ApiClientInterop.cpp;api/core/ApiServerInterop.cpp;api/domain/Currency.proto;api/domain/SecurityCriteria.proto;api/domain/SecurityCriteriaClause.proto;api/domain/SecurityExpiration.proto;api/domain/SecurityMetadata.proto;api/domain/SecurityMetadataCollection.proto;api/domain/SecurityMetadataItem.proto;api/domain/SecurityType.proto;api/domain/TickDisplayRule.proto;api/domain/TickDisplaySettings.proto;api/domain/TickFactors.proto;api/domain/Venue.proto;api/io/URL.proto;api/md/MarketDataSubscriptionParams.proto;api/md/binance/BinanceDepthStreamParams.proto;api/md/binance/BinanceMarketDataSubscriptionParams.proto;api/md/deribit/DeribitMarketDataSubscriptionParams.proto;api/md/ftx/FTXDepthStreamParams.proto;api/md/ftx/FTXMarketDataSubscriptionParams.proto;api/md/glbx/GlbxMarketDataSubscriptionParams.proto;api/md/huobi/HuobiMarketDataSubscriptionParams.proto;api/public/ApiServer.cpp;api/public/ClientChannel.cpp;api/public/QuickbusClientChannel.cpp;api/public/ServerChannel.cpp;api/public/ServiceContext.cpp;api/service_helpers/AdminServiceUtils.cpp;api/service_helpers/AlgoServiceUtils.cpp;api/service_helpers/PositionUtilities.cpp;api/services/AdminService.proto;api/services/AlgoDefinitionService.proto;api/services/AlgoService.proto;api/services/AnalyticsService.proto;api/services/ApiUnitTestService.proto;api/services/AuthService.proto;api/services/BasePositionService.proto;api/services/ErrorReportService.proto;api/services/HelloService.proto;api/services/MBOService.proto;api/services/MarginService.proto;api/services/MarketDataAuxiliaryService.proto;api/services/MarketDataRelay.proto;api/services/MarketDataService.proto;api/services/MoneyFileService.proto;api/services/OrderService.proto;api/services/OutOfProcessURLListener.proto;api/services/PersistenceService.proto;api/services/PositionService.proto;api/services/ShellDiagnosticService.proto;api/services/ShellLatencyService.proto;api/services/SyntheticSecurityService.proto;api/test/ApiTests.cpp;api/test/HelloServiceServer.cpp;config/ConfigCheck.cpp;config/ConfigDownloader.cpp;config/ConfigErrorCollector.cpp;config/ConfigLog.cpp;config/ConfigPropertyEvaluator.cpp;config/ConfigValueQueue.cpp;config/ConfigurationException.cpp;config/ConfigurationState.cpp;config/PreprocessInputStream.cpp;config/config.cpp;config/defaults/MarketData.conf;config/defaults/Ports.conf;config/defaults/SecurityMetadataCollection.conf;config/defaults/Symbols.conf;config/defaults/cme_reference_data.conf;config/defaults/edge.conf;config/defaults/jetfix.conf;config/defaults/keycode.conf;config/defaults/logging.conf;config/defaults/orderman_cfg.conf;config/defaults/orders_fills_cleanup.conf;config/defaults/position_cfg.conf;config/defaults/span.conf;config/defaults/spreader.conf;config/defaults/testkeys.conf;config/defaults/threads.conf;config/defaults/toggles.conf;config/extensions/extensions.proto;config/invalid/invalid.proto;config/protos/algo_types.proto;config/protos/cme_reference_data.proto;config/protos/exchange_sim_cfg.proto;config/protos/jetfix.proto;config/protos/keycode.proto;config/protos/logging.proto;config/protos/market_data.proto;config/protos/mercury_cfg.proto;config/protos/orderman_cfg.proto;config/protos/orders_fills_cleanup.proto;config/protos/perflog_cfg.proto;config/protos/ports.proto;config/protos/position_cfg.proto;config/protos/quickbus_cfg.proto;config/protos/span.proto;config/protos/spartan.proto;config/protos/spreader.proto;config/protos/subprocess.proto;config/protos/symbols.proto;config/protos/system.proto;config/protos/testkeys.proto;config/protos/threads.proto;config/protos/toggles.proto;config/protos/ui.proto;config/tests/HTTPServer.cpp;config/tests/config_ref_tests.cpp;config/tests/config_tests.cpp;config/text_format/text_format.cpp;domain_model/Security.cpp;domain_model/Security.proto;domain_model/SecurityTests.cpp;domain_model/sod.cpp;domain_model/tests/domain_tests.cpp;expression/BooleanGrammar.cpp;expression/ExpressionEvaluator.cpp;expression/ParserExceptions.cpp;expression/Value.cpp;expression/tests/expression_parser_tests.cpp;io/QueryParams.cpp;io/URL.cpp;io/http/CurlHandle.cpp;io/http/HTTPClient.cpp;io/http/HTTPHeaders.cpp;io/http/HTTPRequest.cpp;io/http/HTTPResponse.cpp;io/websocket/WebsocketClient.cpp;io/websocket/WebsocketClientConfiguration.cpp;io/websocket/WebsocketError.cpp;io/websocket/WebsocketMessage.cpp;io/websocket/WebsocketSSLClientInternal.cpp;io/websocket/tests/WebsocketClientTests.cpp;logging/LoggerHub.cpp;logging/Sink.cpp;logging/logging.cpp;logging/tests/logging_tests.cpp;md/CCApiDepthWalker.cpp;md/CCApiListenerAdaptor.cpp;md/CCApiRequestor.cpp;md/Diag/MDPeriodicDumper.cpp;md/Diag/ShmDumper.cpp;md/MBOProcessor.cpp;md/MBOServiceServer.cpp;md/MDClient.cpp;md/MDConsumer.cpp;md/MDConsumerHandle.cpp;md/MDConsumerInterop.cpp;md/MDProducer.cpp;md/MDProducerHandle.cpp;md/MarketDataLine truncated

src/lib/api/public/StatefulPublisher.h

@@ -4,6 +4,7 @@
 #include <set>
 #include <unordered_set>
 #include <vector>
+#include <map>
 
 namespace bts {
 namespace api {
@@ -55,8 +56,14 @@ class StatefulPublisher {
       }
    }
    void erase(TKey m) {
+      // there is an assumption that we are not dealing with that within a 
+      // transaction, add_or_update and erase is called just once a key
       if (_has_key(m)) {
          _erased.insert(m);
+         if (_keep_state)
+            _state_data.erase(m);
+         else
+            _state.erase(m);
       }
    }
    void flush() {

src/lib/api/services/OrderService.proto

@@ -44,6 +44,7 @@ message PrecomputedTheoPoint {
   double vega = 60;
   double theta = 70;
   double rho = 80;
+  double vol = 90;
 }
 
 message PrecomputedTheos {

src/lib/domain_model/Security.h

@@ -114,6 +114,10 @@ inline bool is_same(Price p1, Price p2) {
    return std::abs(p1.value() - p2.value()) < market_epsilon();
 }
 
+inline bool is_same(double d1, double d2) {
+   return std::abs(d1 - d2) < market_epsilon();
+}
+
 inline bool is_equal_or_better(api::Side side, Price p1, Price p2) {
    return is_same(p1, p2) || is_better(side, p1, p2);
 }

src/lib/orderman/calculations/ITheoCalculator.h

@@ -25,5 +25,9 @@ class ITheoCalculator {
        bts::api::orderman::PrecomputedTheoPoint &result) const = 0;
    // gets the driver security for a security with id sec_id
    virtual std::optional<uint32_t> get_driver_sec(uint32_t sec_id) const = 0;
+   // given a vol for an option and the current driver, returns the theo that
+   // corresponds to that vol.
+   virtual double get_theo_correspoding_to_vol(
+       uint32_t sec_id, double driver, double vol) const = 0;
 };
 } // namespace bts::orderman::analytics

src/lib/orderman/calculations/TheoCalculator.cpp

@@ -50,12 +50,15 @@ void interpolate(
        lower.driver(), lower.theta(), higher.driver(), higher.theta(), driver);
    double rho = lin_interp_exterp(
        lower.driver(), lower.rho(), higher.driver(), higher.rho(), driver);
+   double vol = lin_interp_exterp(
+       lower.driver(), lower.vol(), higher.driver(), higher.vol(), driver);
    result.set_theo(theo);
    result.set_delta(delta);
    result.set_gamma(gamma);
    result.set_vega(vega);
    result.set_theta(theta);
    result.set_rho(rho);
+   result.set_vol(vol);
 }
 
 struct DriverNeighbors {
@@ -213,6 +216,24 @@ std::optional<uint32_t> TheoCalculator::get_driver_sec(uint32_t sec_id) const {
    return me->total_theo_data.get_driver_sec(sec_id);
 }
 
+double TheoCalculator::get_theo_correspoding_to_vol(
+    uint32_t sec_id, double driver, double vol) const {
+   PrecomputedTheoPoint rv;
+   if (!get_theo(sec_id, driver, rv)) {
+      return bts::os::nan();
+   }
+   // use the vega to compute the theo
+   double d_vol = vol - rv.vol();
+   double d_theo = rv.vega() * d_vol;
+   double new_theo = rv.theo() + d_theo;
+   LOG(Gen, Info,
+       "TheoCalculator::get_theo_correspoding_to_vol sec_id: {} driver: {} "
+       "vol: {} rv.theo: {} rv.vol: {} rv.vega: {} "
+       "new_theo: {}",
+       sec_id, driver, vol, rv.theo(), rv.vol(), rv.vega(), new_theo);
+   return new_theo;
+}
+
 void TheoCalculator::set_data(const PrecomputedTheos &theos) {
    LOG(Gen, Info, "TheoCalculator::set_data size: {}", theos.theos_size());
    WriteLock write_lock(me->lock);

src/lib/orderman/calculations/TheoCalculator.h

@@ -23,6 +23,8 @@ class TheoCalculator : public ITheoCalculator {
        double driver,
        bts::api::orderman::PrecomputedTheoPoint &result) const override;
    std::optional<uint32_t> get_driver_sec(uint32_t sec_id) const override;
+   double get_theo_correspoding_to_vol(
+       uint32_t sec_id, double driver, double vol) const override;
    void set_data(const bts::api::orderman::PrecomputedTheos &theos);
 
  private:

src/lib/orderman/orderman_utils/orderman_common.h

@@ -279,5 +279,13 @@ inline Quantity get_update_quantity(const bts::api::orderman::Update &u) {
    return rv;
 }
 
+inline void set_update_quantity(
+    bts::api::orderman::Update &state, Quantity cqty) {
+   uint32_t int_qty = cqty.int_part();
+   double fractional_qty = cqty.fractional_part();
+   state.set_quantity(int_qty);
+   state.mutable_fractional_quantity()->set_fraction(fractional_qty);
+}
+
 } // namespace orderman
 } // namespace bts

src/lib/orderman/orders/algo/algo_orders/RegisterDorman.cpp

@@ -12,6 +12,7 @@
 #include "lib/orderman/orderman_utils/orderman_utils.h"
 #include "lib/orderman/orders/algo/algo_orders/LimitCoverAlgo.h"
 #include "lib/orderman/orders/algo/algo_orders/TaskingEyeAlgo.h"
+#include "lib/orderman/orders/algo/algo_orders/VolAlgo.h"
 #include "lib/orderman/orders/algo/algo_orders/limit_cover/LimitCoverAdapter.h"
 #include "lib/orderman/orders/algo/algo_orders/tests/ExecutionTesterAlgo.h"
 #include "lib/plugin/SparkServices.h"
@@ -325,6 +326,19 @@ extern "C" void init_task_eye_plugin(bts::SparkServices &srv) {
    });
 }
 
+extern "C" void init_vol_algo_plugin(bts::SparkServices &srv) {
+   DynamicOrderDescriptor vol_algo;
+   vol_algo.set_dynamic_order_type("Vol");
+   //*vol_algo.add_param_desc() = param_string("Account Override");
+   *vol_algo.add_param_desc() = param_double("Vol");
+   *vol_algo.add_param_desc() = param_bool("Cross At Entry?");
+   *vol_algo.add_param_desc() = param_bool("Requote?");
+   *vol_algo.add_param_desc() = param_bool("Improve Price?");
+   *vol_algo.add_param_desc() = param_int32("Distance From Top");
+   srv.register_dynamic_algo_order(
+       vol_algo, []() { return std::make_unique<orderman::VolAlgo>(); });
+}
+
 extern "C" void init_spark_plugin(bts::SparkServices &srv) {
    auto &dyn_algo_config = config::get<config::toggles::DynamicAlgos>();
    if (dyn_algo_config.enable_dorman())
@@ -345,6 +359,7 @@ extern "C" void init_spark_plugin(bts::SparkServices &srv) {
       init_execution_tester_plugin(srv);
    // init_sniper_plugin(srv);
    init_task_eye_plugin(srv);
+   init_vol_algo_plugin(srv);
 }
 
 } // namespace dorman

src/lib/orderman/orders/algo/algo_orders/VolAlgo.cpp

@@ -0,0 +1,354 @@
+#include "VolAlgo.h"
+#include "lib/api/services/OrderService.pb.h"
+#include "lib/domain_model/Security.h"
+#include "lib/domain_model/Security.pb.h"
+#include "lib/domain_model/SecurityKey.h"
+#include "lib/domain_model/common.h"
+#include "lib/logging/logging.h"
+#include "lib/md/PriceSize.h"
+#include "lib/orderman/ChildOrder.h"
+#include "lib/orderman/OrderBuilder.h"
+#include "lib/orderman/limiter/SecurityLimitNode.h"
+#include "lib/orderman/order_factory/IOrderFactory.h"
+#include "lib/orderman/orderman_utils/Book.h"
+#include "lib/orderman/orderman_utils/orderman_common.h"
+#include "lib/orderman/orderman_utils/orderman_utils.h"
+#include "lib/orderman/orders/OrderOpResult.h"
+#include "lib/orderman/orders/algo/ChildTemplate.h"
+#include "lib/orderman/orders/algo/IAlgoOrder.h"
+#include "lib/orderman/orders/algo/IAlgoServices.h"
+#include "lib/orderman/orders/algo/IBeforeLaunchAlgoServices.h"
+#include "lib/os/os.h"
+#include "lib/utils/PBUtils.h"
+#include "lib/utils/bts_assert.h"
+#include <bits/stdint-uintn.h>
+#include <google/protobuf/util/time_util.h>
+#include <optional>
+#include <string>
+
+using namespace bts::api::orderman;
+
+namespace bts::orderman {
+
+ChildTemplate *get_vol_algo_child_template(
+    IBeforeLaunchAlgoServices &serv,
+    SecurityKey key,
+    api::Side side,
+    std::optional<std::string> account_override) {
+   OrderDefinition order_def = serv.definition();
+   order_def.set_security_key(key.value());
+   order_def.set_side(side);
+   order_def.set_time_in_force(order_def.time_in_force());
+   if (account_override.has_value()) {
+      // no need for it now
+   }
+   order_def.mutable_custom_def()->mutable_limit()->MergeFrom(
+       api::orderman::LimitOrderDefinition());
+   return &(serv.make_child_template(order_def));
+}
+
+struct VolAlgoParams {
+   std::optional<std::string> account_override;
+   double vol_perc;
+   bool cross_at_entry = false;
+   bool requote = false;
+   bool improve_price = false;
+   int distance_from_top = 2;
+};
+
+std::optional<VolAlgoParams> parse_parameters(
+    const DynamicOrderDefinition &def) {
+   const auto &d_params = def.order_params();
+   VolAlgoParams rv = {};
+   // vol
+   auto vol_perc = get_if<double>(d_params, "Vol");
+   if (!vol_perc.has_value())
+      return std::nullopt;
+   rv.vol_perc = vol_perc.value();
+   // cross at entry
+   auto cross_at_entry = get_if<bool>(d_params, "Cross At Entry?");
+   rv.cross_at_entry =
+       cross_at_entry.has_value() ? cross_at_entry.value() : false;
+   // requote
+   auto requote = get_if<bool>(d_params, "Requote?");
+   rv.requote = requote.has_value() ? requote.value() : false;
+   // improve price
+   auto improve_price = get_if<bool>(d_params, "Improve Price?");
+   rv.improve_price = improve_price.has_value() ? improve_price.value() : false;
+   // Distance from Top
+   auto distance_from_top = get_if<int>(d_params, "Distance From Top");
+   if (!distance_from_top.has_value())
+      return std::nullopt;
+   rv.distance_from_top = distance_from_top.value();
+   return rv;
+}
+
+struct VolAlgo ::PImpl {
+   IAlgoServices *service_ptr = nullptr;
+   analytics::ITheoCalculator const *theo_calc = nullptr;
+   OrderKey order_key;
+   uint32_t sec_key;
+   api::Side side;
+   std::optional<std::string> account_override;
+   double qty = 0;
+   VolAlgoParams params = {};
+   uint32_t driver_sec_key = 0;
+   bool subbed_to_driver = false;
+   std::optional<Book> driver_book = std::nullopt;
+   std::optional<Book> sec_book = std::nullopt;
+   // out
+   ChildTemplate *child_order = nullptr;
+   std::optional<OrderKey> child_order_key = std::nullopt;
+   Price current_price = Price{os::nan()};
+   bool inited() const {
+      // once we have receved both books we are ready to go
+      return driver_book.has_value() && sec_book.has_value();
+   }
+   bool order_fired() const { return child_order_key.has_value(); }
+};
+
+VolAlgo::VolAlgo() : me(new PImpl()) { LOG(OrderMan, Info, "VolAlgo created"); }
+
+VolAlgo::~VolAlgo() {}
+
+OrderOpResult VolAlgo::do_before_launch(IBeforeLaunchAlgoServices &serv) {
+   LOG(OrderMan, Info,
+       "VolAlgo::do_before_launch order_key: {} security: {} "
+       "definition: {}",
+       serv.order_key().value(), utils::pb::to_string(serv.security()),
+       utils::pb::to_string(serv.definition()));
+   auto &def = serv.definition();
+   me->order_key = serv.order_key();
+   me->sec_key = def.security_key();
+   me->side = def.side();
+   if (def.has_custom_def() && def.custom_def().has_dynamic()) {
+      auto va_params = parse_parameters(def.custom_def().dynamic());
+      if (!va_params.has_value()) {
+         return {
+             bts::api::orderman::OrderError::ORDER_ERROR_INVALID_PARAMETER,
+             std::string("Parameters are not well formed")};
+      }
+      me->params = va_params.value();
+      me->child_order = get_vol_algo_child_template(
+          serv, SecurityKey(me->sec_key), me->side,
+          me->params.account_override);
+      LOG(OrderMan, Info,
+          "VolAlgo::do_before_launch order_key: {} sec_key: {} side: {} vol: "
+          "{} cross_at_entry: {} re_quote: {} improve_price: {} "
+          "distance_from_top: {}",
+          me->order_key, me->sec_key, bts::api::Side_Name(me->side),
+          me->params.vol_perc, me->params.cross_at_entry, me->params.requote,
+          me->params.improve_price, me->params.distance_from_top);
+      return {};
+   }
+   return {
+       bts::api::orderman::OrderError::ORDER_ERROR_INVALID_PARAMETER,
+       std::string("Parameters are not well formed")};
+}
+
+void VolAlgo::do_launch(IAlgoServices &serv) {
+   LOG(OrderMan, Info, "VolAlgo::do_launch order_key: {} qty: {}",
+       serv.order_key().value(), serv.quantity());
+   me->service_ptr = &serv;
+   me->qty = serv.quantity().full_value();
+   me->service_ptr->subscribe_theos();
+   me->service_ptr->subscribe_timer();
+   me->service_ptr->subscribe_book(SecurityKey(me->sec_key));
+}
+
+void VolAlgo::do_launch_attached(IAlgoServices &serv, OrderKey, ChildState) {
+   me->service_ptr = &serv;
+   _terminate_algo(true, "Vol Algo can not be attached to working orders");
+}
+
+void VolAlgo::on_book(SecurityKey sec_key, Book const &book) {
+   LOG(OrderMan, Diag, "VolAlgo::on_book order_key: {} sec_key: {}",
+       me->order_key, sec_key);
+   if (sec_key.value() == me->sec_key) {
+      me->sec_book = book;
+      _analyze_and_act();
+   } else if (sec_key.value() == me->driver_sec_key) {
+      me->driver_book = book;
+      _analyze_and_act();
+   }
+}
+void VolAlgo::do_update(api::orderman::Update const &u) {
+   LOG(OrderMan, Info,
+       "VolAlgo::do_update order_key: {} sec_key: {} "
+       "update: {}",
+       me->order_key, me->sec_key, utils::pb::to_string(u));
+   const auto &p = u.custom().dynamic();
+   auto params = parse_parameters(p);
+   if (!params.has_value()) {
+      LOG(Gen, Error,
+          "VolAlgo::do_update order_key: {} parameters are not well formed",
+          me->order_key);
+      return;
+   }
+   me->qty = u.quantity();
+   me->params = params.value();
+   _analyze_and_act(true);
+   // price changes are ignored the price is the vol
+}
+
+void VolAlgo::on_child_change(OrderKey, ChildState const &state) {
+   // price / qty is reflected from the child
+   me->service_ptr->update_fill_state(state.total_fills, state.last_price);
+   switch (state.status) {
+   case ChildStatus::WORKING:
+      break;
+   case ChildStatus::COMPLETED:
+      me->service_ptr->complete();
+      break;
+   case ChildStatus::CANCELED:
+      me->service_ptr->cancel("Child Order Cancelled");
+      break;
+   }
+}
+
+void VolAlgo::on_status(SecurityKey, co2::TradingStatus) {}
+void VolAlgo::on_timer() {}
+
+void VolAlgo::on_theos(analytics::ITheoCalculator const *theo_calculator) {
+   LOG(OrderMan, Info, "VolAlgo::on_theos order_key: {}", me->order_key);
+   me->theo_calc = theo_calculator;
+   if (me->subbed_to_driver) {
+      _analyze_and_act();
+      return;
+   }
+   auto driver_sec = me->theo_calc->get_driver_sec(me->sec_key);
+   if (driver_sec.has_value()) {
+      LOG(OrderMan, Info,
+          "VolAlgo::on_theos order_key: {} subscribing to driver sec_key: {}",
+          me->order_key, driver_sec.value());
+      me->subbed_to_driver = true;
+      me->driver_sec_key = driver_sec.value();
+      me->service_ptr->subscribe_book(SecurityKey(me->driver_sec_key));
+   } else {
+      LOG(OrderMan, Warn,
+          "VolAlgo::on_theos order_key: {} could not find driver",
+          me->order_key);
+   }
+}
+void VolAlgo::on_position(PositionUpdate const &) {}
+
+// this is where all the action happens, both new order (just once) as well as
+// order updates
+void VolAlgo::_analyze_and_act(bool force_reprice) {
+   // check for period (to stop frequent updates)
+   auto get_driver = [&]() {
+      const md::PriceSize &driver_bid = me->driver_book->bid().at(0);
+      const md::PriceSize &driver_ask = me->driver_book->ask().at(0);
+      if (bts::securities::is_same(driver_bid.size(), 0) ||
+          bts::securities::is_same(driver_ask.size(), 0))
+         return os::nan();
+      // finding weighted mean (TODO change it to mark price later)
+      double driver = (driver_bid.price() * driver_bid.size() +
+                       driver_ask.price() * driver_ask.size()) /
+                      (driver_bid.size() + driver_ask.size());
+      return driver;
+   };
+   auto crosses_opp_market = [&](Price order_price) {
+      if (me->side == api::SIDE_BUY) {
+         double top = me->sec_book->ask().at(0).price();
+         return securities::gr_than_eq(order_price.value(), top);
+      } else {
+         double top = me->sec_book->bid().at(0).price();
+         return securities::gr_than_eq(top, order_price.value());
+      }
+   };
+   if (me->order_fired()) {
+      // repricing if parameters (and arg) allows
+      if (!me->params.requote && !force_reprice)
+         return;
+      // working order
+      double driver = get_driver();
+      if (os::is_nan(driver)) {
+         // keep working it
+         return;
+      }
+      double theo_unnorm = me->theo_calc->get_theo_correspoding_to_vol(
+          me->sec_key, driver, me->params.vol_perc / 100.0);
+      // TODO: normalize
+      double theo = theo_unnorm / driver;
+      Price new_price = _get_order_price(theo);
+      LOG(OrderMan, Info,
+          "VolAlgo::_analyze_and_act (calc updating) order_key: {} "
+          "theo_unnorm: {} driver: {} theo: {} order_price: {}",
+          me->order_key, theo_unnorm, driver, theo, new_price);
+      if (securities::is_same(me->current_price, new_price)) {
+         return;
+      }
+      bool improving =
+          securities::is_better(me->side, new_price, me->current_price);
+      if (!improving || me->params.improve_price || force_reprice) {
+         bts::api::orderman::Update update;
+         update.set_price(new_price.value());
+         set_update_quantity(update, Quantity::create_new(me->qty));
+         me->current_price = new_price;
+         LOG(Gen, Info,
+             "VolAlgo::_analyze_and_act updating order order_key: {} price: {} "
+             "qty: {}",
+             me->order_key, me->current_price, me->qty);
+         me->service_ptr->update_child(me->child_order_key.value(), update);
+         return;
+      }
+   } else {
+      // new order path
+      if (me->inited()) {
+         // time to fire
+         double driver = get_driver();
+         if (os::is_nan(driver)) {
+            return;
+         }
+         // entering order
+         double theo_unnorm = me->theo_calc->get_theo_correspoding_to_vol(
+             me->sec_key, driver, me->params.vol_perc / 100.0);
+         double theo = theo_unnorm / driver;
+         Price order_price = _get_order_price(theo);
+         LOG(OrderMan, Info,
+             "VolAlgo::_analyze_and_act (calc new) order_key: {} theo_unnorm: "
+             "{} driver: {} theo: {} order_price: {}",
+             me->order_key, theo_unnorm, driver, theo, order_price);
+         if (!me->params.cross_at_entry && crosses_opp_market(order_price))
+            return;
+         me->current_price = order_price;
+         LOG(Gen, Info,
+             "VolAlgo::_analyze_and_act placing order order_key: {} price: {} "
+             "qty: {}",
+             me->order_key, me->current_price, me->qty);
+         me->child_order_key = me->service_ptr->launch_child(
+             *me->child_order, me->current_price,
+             Quantity::create_new(me->qty));
+      }
+   }
+}
+
+void VolAlgo::_terminate_algo(bool is_error, const std::string &err_msg) {
+   // TODO unsub ?
+   if (is_error) {
+      me->service_ptr->terminate_error(err_msg);
+   } else {
+      me->service_ptr->complete();
+   }
+}
+
+Price VolAlgo::_get_order_price(double theo) const {
+   // BTC Opts (p <=  0.005) : price = 0.0001 qty = 0.1
+   // BTC Opts (p > 0.005)   : price = 0.0005 qty = 0.1
+   // -------------------------------------------------
+   // ETH Opts (p <=  0.005) : price = 0.0001 qty = 1
+   // ETH Opts (p > 0.005)   : price = 0.0005 qty = 1
+   // TODO: get the above from the security itself
+   double rv = 0;
+   if (theo > 0.005) {
+      rv = securities::raw_round_to_quote_price(theo, 0.0005, me->side);
+   } else {
+      rv = securities::raw_round_to_quote_price(theo, 0.0001, me->side);
+   }
+   LOG(Gen, Info,
+       "VolAlgo::_get_order_price order_key: {} theo: {} order_price: {}",
+       me->order_key, theo, rv);
+   return Price{rv};
+}
+} // namespace bts::orderman

src/lib/orderman/orders/algo/algo_orders/VolAlgo.h

@@ -0,0 +1,38 @@
+#pragma once
+
+#include "lib/domain_model/SecurityKey.h"
+#include "lib/domain_model/common.h"
+#include "lib/orderman/calculations/ITheoCalculator.h"
+#include "lib/orderman/orders/algo/IAlgoOrder.h"
+#include <openssl/ossl_typ.h>
+
+namespace bts::orderman {
+
+// This is a test algo, not to be used in prod, that sends an order and modifies
+// it by moving the price up and down at a given period
+class VolAlgo : public IAlgoOrder {
+ public:
+   VolAlgo();
+   ~VolAlgo();
+   OrderOpResult do_before_launch(IBeforeLaunchAlgoServices &) override;
+   void on_book(SecurityKey, Book const &) override;
+   void do_launch(IAlgoServices &) override;
+   void do_launch_attached(IAlgoServices &, OrderKey, ChildState) override;
+   void do_update(api::orderman::Update const &) override;
+   void on_child_change(OrderKey, ChildState const &) override;
+   void on_status(SecurityKey, co2::TradingStatus) override;
+   void on_timer() override;
+   void on_theos(analytics::ITheoCalculator const *theo_calculator) override;
+   void on_position(PositionUpdate const &) override;
+
+ private:
+   struct PImpl;
+   std::unique_ptr<PImpl> me;
+
+ private:
+   void _analyze_and_act(bool force_reprice = false);
+   void _terminate_algo(bool is_error, const std::string &err_msg);
+   Price _get_order_price(double theo) const;
+};
+
+} // namespace bts::orderman

src/server/sparkd/analytics/AnalyticsServiceServer.cpp

@@ -8,6 +8,7 @@
 #include "lib/persistence/SqliteStore.h"
 #include "lib/utils/PBUtils.h"
 #include "lib/utils/Puid.h"
+#include "lib/utils/TimeStamp.h"
 #include "lib/utils/bts_assert.h"
 #include "server/sparkd/algo/AlgoDefinitionServiceServer.h"
 #include <boost/algorithm/string.hpp>
@@ -90,6 +91,14 @@ struct AnalyticsServiceServer::PImpl {
    const std::string vs_usage_ns = "volsurface_usage";
    const std::string reports_ns = "reports";
    const std::string archived_theos_ns = "archived_theos";
+   // this is the total number of vs data that will be stored in the db,
+   // beyond that, old values will be recycled
+   const size_t vs_history_sz = 500;
+   // anologous to above for archived theos (however unlike the above this is
+   // recycled periodically, and is not published out)
+   const size_t archived_theos_history_sz = 5000;
+   const size_t n_theos_archives_before_cleanup = 100;
+   size_t n_theos_archives = 0;
    std::unique_ptr<persistence::IStore> vs_store;
    std::unique_ptr<persistence::IStore> exp_params_store;
    std::unique_ptr<persistence::IStore> vs_usage_store;
@@ -126,24 +135,36 @@ AnalyticsServiceServer::AnalyticsServiceServer(
     : AnalyticsServiceBase(context),
       me(new PImpl(
           context,
-          make_volsurfaces_stateful_publisher(),
+          bts::api::StatefulPublisher<VolSurfaceEntry, uint64_t>(
+              make_volsurfaces_publisher(),
+              [](const VolSurfaceEntry &v) { return v.save_time(); },
+              true),
+          // make_volsurfaces_stateful_publisher(),
           make_exp_parameters_stateful_publisher(),
           make_volsurface_usage_stateful_publisher(),
           make_reports_stateful_publisher())) {
    // publishing all vol surfaces
    LOG(Gen, Info, "Starting AnalyticsServiceServer");
    _clean_up_vss();
+   _clean_up_archived_theos();
    auto vs_table = me->vs_store->retrieve_all();
    for (auto const &ns_vals : vs_table) {
       const std::string &vs_name = ns_vals.first;
       const std::map<std::string, std::string> &key_vals = ns_vals.second;
-      LOG(OrderMan, Info, "vs:{} entries:{}", vs_name, key_vals.size());
+      LOG(OrderMan, Info, "vs: {} entries: {}", vs_name, key_vals.size());
       for (const auto &kvp : key_vals) {
          const std::string &key = kvp.first;
          const std::string &val = kvp.second;
-         LOG(OrderMan, Diag, "vs:{} key:{} val:{}", vs_name, key, val);
+         LOG(OrderMan, Info,
+             "AnalyticsServiceServer::cons loaded from db vs: {} key: {}",
+             vs_name, key);
          VolSurfaceEntry vs_entry;
          utils::pb::from_string(val, &vs_entry);
+         LOG(Gen, Diag,
+             "AnalyticsServiceServer::cons loading to vs_publisher ns: {} "
+             "key: {} vs_name: {} vs_id: {} vs.save_time: {}",
+             vs_name, key, vs_entry.vs_name(), vs_entry.vs_id(),
+             vs_entry.save_time());
          me->vs_publisher.add_or_update(vs_entry);
       }
    }
@@ -213,6 +234,36 @@ void AnalyticsServiceServer::get_new_volsurface_id(
 void AnalyticsServiceServer::save_volsurface(
     const bts::api::analytics::VolSurfaceEntry &vs_entry_arg,
     const api::Response<::bts::api::ResultMessage> &response) {
+   auto t0 = utils::TimeStamp::now().to_uint64();
+   auto remove_oldest = [&](const std::string &vs_ns) {
+      // return;
+      int vs_id = vs_entry_arg.vs_id();
+      // find all the keys belonging to this vs
+      std::vector<uint64_t> vs_keys = {};
+      for (const auto &kv : me->vs_publisher.state_data()) {
+         if (kv.second.vs_id() == vs_id) {
+            vs_keys.push_back(kv.first);
+         }
+      }
+      LOG(Gen, Info,
+          "AnalyticsServiceServer::save_volsurface vs_id: {} n_entries: {}",
+          vs_entry_arg.vs_id(), vs_keys.size());
+      if (vs_keys.size() <= me->vs_history_sz)
+         return;
+      // since we are already truncating it at the the start, we'll just
+      // find the earliest key and delete it         
+      uint64_t key_to_delete = std::numeric_limits<uint64_t>::max();
+      for (uint64_t k : vs_keys) {
+         if (k < key_to_delete) {
+            key_to_delete = k;
+         }
+      }
+      LOG(Gen, Info,
+          "AnalyticsServiceServer::save_volsurface removing row ns: {} key: {}",
+          vs_ns, key_to_delete);
+      me->vs_store->remove(vs_ns, {std::to_string(key_to_delete)});
+      me->vs_publisher.erase(key_to_delete);
+   };
    uint64_t ms_u = get_epoch();
    VolSurfaceEntry vs_entry = vs_entry_arg;
    vs_entry.set_save_time(ms_u);
@@ -226,9 +277,15 @@ void AnalyticsServiceServer::save_volsurface(
    std::string value = utils::pb::to_string(vs_entry);
    me->vs_store->insert(ns, key, value);
    me->vs_publisher.add_or_update(vs_entry);
+   remove_oldest(ns);
    me->vs_publisher.flush();
    api::ResultMessage ret;
    ret.set_success(true);
+   auto t1 = utils::TimeStamp::now().to_uint64();
+   LOG(Gen, Info,
+       "AnalyticsServiceServer::save_volsurface ns: {} key: {} n_entries: {} "
+       "time: {} ns",
+       ns, key, me->vs_publisher.state_data().size(), t1 - t0);
    response(ret);
 }
 
@@ -370,9 +427,8 @@ void AnalyticsServiceServer::delete_report(
 void AnalyticsServiceServer::archive_theos(
     const bts::api::analytics::ArchivedTheoList &theo_list,
     const api::Response<::bts::api::ResultMessage> &resp) {
-   LOG(Gen, Info,
-       "AnalyticsServiceServer::archive_theos sec_tups: {} sec_theo: {}",
-       theo_list.sec_tups().size(), theo_list.sec_theo().size());
+   me->n_theos_archives++;
+   uint64_t t0 = utils::TimeStamp::now().to_uint64();
    if (theo_list.sec_theo().empty()) {
       std::string err_msg = "Empty archived theo list, call failed";
       LOG(Gen, Error, "AnalyticsServiceServer::archive_theos {}", err_msg);
@@ -383,12 +439,23 @@ void AnalyticsServiceServer::archive_theos(
    const std::string key = std::to_string(theo_list.save_time());
    std::string value = utils::pb::to_string(theo_list);
    me->theos_store->insert(ns, key, value);
+   uint64_t t1 = utils::TimeStamp::now().to_uint64();
+   uint64_t t_diff = t1 - t0;
+   LOG(Gen, Info,
+       "AnalyticsServiceServer::archive_theos sec_tups: {} sec_theo: {} time: "
+       "{} ns",
+       theo_list.sec_tups().size(), theo_list.sec_theo().size(), t_diff);
+   if (me->n_theos_archives >= me->n_theos_archives_before_cleanup) {
+      me->n_theos_archives = 0;
+      _clean_up_archived_theos();
+   }
    fire_cb(resp, true);
 }
 
 void AnalyticsServiceServer::get_archived_theos(
     const bts::api::analytics::ArchivedTheoRetrieval &ret,
     const bts::api::Response<bts::api::analytics::ArchivedTheoList> &resp) {
+   uint64_t t0 = utils::TimeStamp::now().to_uint64();
    LOG(Gen, Info, "AnalyticsServiceServer::get_archived_theos ret: {}",
        utils::pb::to_string(ret));
    auto contains_st = [&](const ArchivedTheoList &tl) {
@@ -422,54 +489,80 @@ void AnalyticsServiceServer::get_archived_theos(
          }
       }
    }
-   LOG(Gen, Info, "AnalyticsServiceServer::get_archived_theos ret: {} key: {}",
-       utils::pb::to_string(ret), key);
+   uint64_t t1 = utils::TimeStamp::now().to_uint64();
+   uint64_t t_diff = t1 - t0;
+   LOG(Gen, Info,
+       "AnalyticsServiceServer::get_archived_theos ret: {} key: {} time: {} ns",
+       utils::pb::to_string(ret), key, t_diff);
    resp(rv);
 }
 
 AnalyticsServiceServer::~AnalyticsServiceServer() {}
 
+// for a map where the key is the epoch timestamp, this returns the keys
+// to delete such that final size is final_sz
+std::set<std::string> get_keys_to_delete(
+    const std::map<std::string, std::string> &time_keyed_map, size_t final_sz) {
+   if (time_keyed_map.size() <= final_sz) {
+      return {};
+   }
+   std::vector<uint64_t> keys_uint;
+   for (const auto &kv : time_keyed_map) {
+      try {
+         keys_uint.push_back(std::stoull(kv.first));
+      } catch (int) {
+         LOG(Gen, Error,
+             "AnalyticsServiceServer::get_keys_to_delete could not "
+             "convert "
+             "key: {} to uint",
+             kv.first);
+      }
+   }
+   // this is sorted lowest to highest
+   std::sort(keys_uint.begin(), keys_uint.end());
+   size_t n_del_rows = keys_uint.size() - final_sz;
+   std::vector<uint64_t> keys_uint_to_del =
+       std::vector<uint64_t>(keys_uint.begin(), keys_uint.begin() + n_del_rows);
+   std::set<std::string> keys_to_del;
+   for (auto k : keys_uint_to_del) {
+      keys_to_del.insert(std::to_string(k));
+   }
+   return keys_to_del;
+}
+
 void AnalyticsServiceServer::_clean_up_vss() {
-   const size_t final_sz = 500;
+   const size_t final_sz = me->vs_history_sz;
    auto all_rows = me->vs_store->retrieve_all();
    for (const auto &ns_k : all_rows) {
+      auto t0 = utils::TimeStamp::now().to_uint64();
       const auto &vs = ns_k.first;
       const auto &vs_row_map = ns_k.second;
-      if (vs_row_map.size() <= final_sz) {
-         LOG(Gen, Info,
-             "AnalyticsServiceServer::_clean_up_vss ignoring vs: {} because it "
-             "has only {} rows",
-             vs, vs_row_map.size());
-         continue;
-      }
-      std::vector<uint64_t> keys_uint;
-      for (const auto &r : vs_row_map) {
-         const auto &k = r.first;
-         try {
-            keys_uint.push_back(std::stoull(k));
-         } catch (int) {
-            LOG(Gen, Error,
-                "AnalyticsServiceServer::_clean_up_vss could not convert "
-                "key: "
-                "{} to uint",
-                k);
-         }
-      }
-      // sort and remove
-      std::sort(keys_uint.begin(), keys_uint.end());
-      size_t n_del_rows = keys_uint.size() - final_sz;
-      std::vector<uint64_t> keys_uint_to_del = std::vector<uint64_t>(
-          keys_uint.begin(), keys_uint.begin() + n_del_rows);
-      std::set<std::string> keys_to_del;
-      for (auto k : keys_uint_to_del) {
-         keys_to_del.insert(std::to_string(k));
-      }
-      LOG(Gen, Info,
-          "AnalyticsServiceServer::_clean_up_vss vs: {} n_rows: {} removing {} "
-          "rows",
-          vs, keys_uint.size(), keys_to_del.size());
+      auto keys_to_del = get_keys_to_delete(vs_row_map, final_sz);
       me->vs_store->remove(vs, keys_to_del);
+      auto t1 = utils::TimeStamp::now().to_uint64();
+      LOG(Gen, Info,
+          "AnalyticsServiceServer::_clean_up_vss vs: {} n_rows: {} "
+          "n_dels {} time: {} ns",
+          vs, vs_row_map.size(), keys_to_del.size(), t1 - t0);
    }
 }
 
+void AnalyticsServiceServer::_clean_up_archived_theos() {
+   uint64_t t0 = utils::TimeStamp::now().to_uint64();
+   std::map<std::string, std::map<std::string, std::string>> all_data =
+       me->theos_store->retrieve_all();
+   auto it = all_data.find(me->archived_theos_ns);
+   if (it == all_data.end())
+      return;
+   const std::map<std::string, std::string> &theo_data = it->second;
+   std::set<std::string> keys_to_del =
+       get_keys_to_delete(theo_data, me->archived_theos_history_sz);
+   me->theos_store->remove(me->archived_theos_ns, keys_to_del);
+   uint64_t t1 = utils::TimeStamp::now().to_uint64();
+   uint64_t t_diff = t1 - t0;
+   LOG(Gen, Info,
+       "AnalyticsServiceServer::_clean_up_archived_theos n_dels: {} "
+       "time: {} ns",
+       keys_to_del.size(), t_diff);
+}
 } // namespace bts::analytics

src/server/sparkd/analytics/AnalyticsServiceServer.h

@@ -72,5 +72,6 @@ class AnalyticsServiceServer
 
  private:
    void _clean_up_vss();
+   void _clean_up_archived_theos();
 };
 } // namespace bts::analytics

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

@@ -115,6 +115,7 @@ void Run::run(
    auto impl = [&mbo_writer, &tss_dispatcher, &type, &cleanup]() {
       LOG(md, Info, "Run::impl lambda t_id:{} t_name", os::get_thread_id(),
           get_thread_name());
+      return;
       if (cleanup)
          md::cmds::Cleanup::execute();
       if (type == Run::MDType::Configured) {
@@ -148,6 +149,7 @@ void Run::run(
          threading::make_thread("Emumba", []() { run_emumba(); }).detach();
       }
       if (bts::config::get<bts::md::MarketDataConfig>().use_sim()) {
+         return;
          threading::make_thread("ExchangeSimGLBX", [&mbo_writer]() {
             sim::start_exchange_sim(api::Venue::VENUE_SIM, mbo_writer);
          }).detach();

tools/ensure_venv.cmd

@@ -8,17 +8,18 @@ if exist %venv_dir% (
       echo Removing python virtual environment 
       rmdir /S /Q %venv_dir%
    ) else (
-      GOTO ERROR
+      GOTO END
    )
 )
 
 echo Creating python virtual environment
 @call python -m venv %venv_dir%
 @call %venv_dir%\Scripts\activate.bat
-@call pip install --upgrade pip
+@call %venv_dir%\Scripts\python -m pip install --upgrade pip
 @call %venv_dir%\Scripts\pip install -r %~dp0..\requirements.txt
 @call deactivate
 
+:END
 EXIT /B %ERRORLEVEL%
 
 :ERROR

tools/jenkins_build_spark.cmd

@@ -0,0 +1,8 @@
+@echo off
+
+call "C:\\Program Files\\Microsoft Visual Studio\\2022\\Community\\Common7\\Tools\\VsDevCmd.bat"
+call .venv\\Scripts\\activate
+call set PYTHONPATH=%~dp0\..\.venv\Scripts\python
+call .venv\Scripts\python.exe tools\py\build_spark.py %*
+
+EXIT /B %ERRORLEVEL%

tools/py/build_spark.py

@@ -173,8 +173,7 @@ def _get_arg_parser():
         metavar='debug|release',
         type=str,
         nargs='?',
-        help='The build configuration',
-        default='Debug')
+        help='The build configuration')
 
     parser.add_argument(
         '-o',
@@ -336,9 +335,10 @@ def _main():
     link_build_dir = arg_to_path(get_arg(args_dict, 'link_output_dir'))
     targets = flatten_arglist(get_arg(args_dict, 'target'))
     verbosity = get_arg(args_dict, 'verbosity') or None
-
+    configuration = args.configuration or 'Release' if targets is not None and any(
+        filter(lambda t: t.lower() == 'publish', targets)) else 'Debug'
     build(
-        configuration=args.configuration,
+        configuration=configuration,
         clean=args.clean,
         restore=args.restore,
         build_dir=build_dir,