Merge branch 'development.cboe-prod' of bitbucket.org:bluetradesys/spark into development.cboe-prod

.vscode/launch.json

@@ -2,9 +2,10 @@
   "version": "0.2.0",
   "configurations": [
     {
-      "name": ".NET Core Attach",
+      "name": "Attach Spark",
       "type": "coreclr",
-      "request": "attach"
+      "request": "attach",
+      "processId": "${command:pickProcess}"
     },
     {
       "name": "DataStreamPublisher",
@@ -326,4 +327,4 @@
       "description": "Test to run",
     },
   ]
-}
+}

global.json

@@ -1,6 +0,0 @@
-{
-  "sdk": {
-    "version": "6.0.200",
-    "rollForward": "latestMajor"
-  }
-}

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

@@ -45,7 +45,8 @@ namespace mbts.spark {
             Func<AnalyticsExpirationParameters, bool> is_unexpired = (p) => {
                var st = p.driver_security.security_tuple.to_api_sec_tuple();
                var exp = p.security.expiration;
-               return SecurityExpirationHelper.is_unexpired(st, exp);
+               return exp.Date >= DateTime.Now.Date;
+               //return SecurityExpirationHelper.is_unexpired(st, exp);
             };
             exp_params = exp_params.Where(t => is_unexpired(t));
             // listens to the commit tag and saves it

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

@@ -12,8 +12,8 @@ using bts.api;
 using mbts.ui.converters;
 
 namespace mbts.spark {
-   // This class gathers all the inputs needed to calculate InterpGreeks, pass it to 
-   // VolSurfacePrecomputedTheosProducer, and publishes the InterpGreeks to the bus
+   // This class gathers all the inputs needed to calculate InterpGreeks for a single vol surface,
+   // passes it to VolSurfacePrecomputedTheosProducer, and publishes the InterpGreeks to the bus
    // (called SingleVolSurfaceClientPrecomputedTheosPublisher in Edge)
    // It listens to
    // a. vol surface inputs

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

@@ -20,6 +20,7 @@ using Spark.Core.Protobuf;
 using mbts.ui;
 using bts.config.toggles;
 using System.Diagnostics.CodeAnalysis;
+using BCG;
 // using mbts.bb;
 
 namespace mbts.spark
@@ -92,11 +93,11 @@ namespace mbts.spark
       // in tags
       private readonly ITag<SystemContractData> _in_scdata_tag =
          SparkTags.SystemContractsTag;
+      private readonly ITag<NMSheetSymbol> _archive_ivs_tag = SparkTags.TC2Tags.ArchiveIVsTag;
       private readonly ITag<VolSurfaceSettings> _in_bb_create_tag = SparkTags.VolSurfaceCreateTag;
       private readonly ITag<int> _in_bb_delete_tag = SparkTags.VolSurfaceDeleteTag;
       private readonly ITag<AnalyticsExpirationParametersMap> _in_bb_exp_params_tag =
          SparkTags.ExpirationParametersInputTag;
-      private readonly ITag<ClientArchivedVolRequest> _in_bb_req_vol = SparkTags.RequestArchivedVols;
       // state
       private SystemContractData _contract_data = null;
       private IDictionary<int, SingleVolSurfaceManager> _vs_manager_map =
@@ -106,6 +107,43 @@ namespace mbts.spark
       private bts.api.IBus _bus;
       private List<IDisposable> _subs = new List<IDisposable>();
 
+      private static void add_temp_vsu_entries(IDictionary<string, int> name_id_map,
+                                                 VolSurfaceUsageList ulist) {
+         Action<string, Venue, string> add_entry = (option_name, option_venue, vs_name) => {
+
+            int vs_key;
+            if (name_id_map.TryGetValue(vs_name, out vs_key)) {
+               ulist.UsageList.Add(new VolSurfaceUsageEntry() {
+                  VsId = vs_key,
+                  SymbolName = option_name,
+                  SecTup = new bts.api.SecurityTuple {
+                     Type = SecurityType.Option,
+                     Venue = option_venue,
+                     Symbol = option_name
+                  }
+               });
+            }
+
+         };
+         add_entry("SPX", Venue.Cboe, "CBOE_skews");
+         add_entry("SPXW", Venue.Cboe, "CBOE_skews");
+         add_entry("E1B", Venue.Glbx, "CME_Skews");
+         add_entry("E1C", Venue.Glbx, "CME_Skews");
+         add_entry("E3B", Venue.Glbx, "CME_Skews");
+         add_entry("E3C", Venue.Glbx, "CME_Skews");
+         add_entry("E3D", Venue.Glbx, "CME_Skews");
+         add_entry("E4A", Venue.Glbx, "CME_Skews");
+         add_entry("E4B", Venue.Glbx, "CME_Skews");
+         add_entry("E4C", Venue.Glbx, "CME_Skews");
+         add_entry("E4C", Venue.Glbx, "CME_Skews");
+         add_entry("E4D", Venue.Glbx, "CME_Skews");
+         add_entry("ES", Venue.Glbx, "CME_Skews");
+         add_entry("EW1", Venue.Glbx, "CME_Skews");
+         add_entry("EW2", Venue.Glbx, "CME_Skews");
+         add_entry("EW3", Venue.Glbx, "CME_Skews");
+         add_entry("EW4", Venue.Glbx, "CME_Skews");
+      }
+
       public VolSurfaceBusManager(bts.api.IBus bus,
                                          bts.api.analytics.AnalyticsService analytics_service,
                                          bts.api.orderman.OrderService order_service)
@@ -119,7 +157,7 @@ namespace mbts.spark
          _bb.subscribe(_in_bb_create_tag);
          _bb.subscribe(_in_bb_delete_tag);
          _bb.subscribe(_in_bb_exp_params_tag);
-         _bb.subscribe(_in_bb_req_vol);
+         _bb.subscribe(_archive_ivs_tag);
          _bb.data_callback += _on_bb_update;
          _bb.timer_callback += _on_timer;
          var contracts_throttle = TimeSpan.FromMilliseconds(200);
@@ -127,11 +165,25 @@ namespace mbts.spark
          var vs_sub = _analytics_service.subscribei_volsurfaces()
             .ObserveOn(SynchronizationContext.Current)
             .Subscribe(_on_vss);
+         //var hist_theos_sub_obs = _analytics_service.subscribei_archived_theos_table();
+         //var obs = hist_theos_sub_obs.ObserveOn(SynchronizationContext.Current);//.Subcribe(_on_hist_theos);
+         //obs.Subscribe(tu => {
+         //   Log.Gen.info("Received historical theos");
+         //   _on_hist_theos(tu);
+         //});
+
+         var archived_theos_obs = _analytics_service.subscribei_archived_theos().ObserveOn(SynchronizationContext.Current).Subscribe(tu => {
+            Log.Gen.info("Received historical theos");
+            _on_hist_theos(tu);
+         });
+         
+
          var vs_usage_sub = _analytics_service.subscribei_volsurface_usage()
             .ObserveOn(SynchronizationContext.Current)
             .Subscribe(_on_vs_usage_update);
          _subs.Add(vs_sub);
          _subs.Add(vs_usage_sub);
+         _subs.Add(archived_theos_obs);
          _subs.Add(new AnalyticsParametersManager(bus, _analytics_service));
          _subs.Add(new ResourceAllocator(bus, _bb, contracts_throttle));
          _subs.Add(new SymbolAggregationPublisher<OptionDriverPriceMap>(
@@ -143,7 +195,7 @@ namespace mbts.spark
          _subs.Add(new LiveTheoPublisher(_bb));
          _subs.Add(new InterpTheosServerPublisher(_bb, bus, order_service, contracts_throttle));
          _subs.Add(new VolSurfaceInterpGreeksMultiplexer(_bb));
-         // usage         
+         // list of all volsurface. in this callback, we read the usage config, and detemine the id of the named vol surface         
          var list_sub = _bus.subscribe(SparkTags.VolSurfaceListTag).OnDispatcher().Subscribe(t =>
          {
             VolSurfaceUsageList ulist = new VolSurfaceUsageList();
@@ -154,9 +206,12 @@ namespace mbts.spark
             }
             Log.Ui.info($"VolSurfaceUsageList name_to_ids count: {name_to_ids.Count}");
             var vs_usage = mbts.config.Configuration.get<bts.config.toggles.VolSurfaceUsage>();
-            Log.Ui.info($"VolSurfaceUsageList vs_usage acount: {vs_usage.Items.Count}");
+            Log.Ui.info($"VolSurfaceUsageList vs_usage acount: {vs_usage.Items.Count}");            
             foreach (var u in vs_usage.Items)
             {
+               if (string.Compare(u.OptionName, "SPX", 0) == 0 || string.Compare(u.OptionName, "SPXW", 0) == 0){
+                  continue;
+               }
                if (string.Compare(u.OptionName, "SPX", 0) == 0 || string.Compare(u.OptionName, "SPXW", 0) == 0 && u.OptionVenue == Venue.Unset)
                {
                   u.OptionVenue = Venue.Cboe;
@@ -186,10 +241,11 @@ namespace mbts.spark
                else
                {
                   Log.Gen.error($"Could not find vol surface with name: {vs_name}");
-               }
+               }               
             }
+            add_temp_vsu_entries(name_to_ids, ulist);
             var list_imm = new VolSurfaceUsageListImmutable(ulist);
-            Log.Ui.info(list_imm.log_string());
+            Log.Ui.info("VolSurfaceUsage: " + list_imm.log_string());
             _bb.post(SparkTags.VolSurfaceUsageTag, list_imm);
             _spin_vss(list_imm);
          });
@@ -292,6 +348,8 @@ namespace mbts.spark
 
       void _on_vs_usage_update(Update<VolSurfaceUsageListImmutable, int> update)
       {
+         var state = update.all;
+         Log.Gen.info($"VSBBMan._on_vs_usage_update size: {state.Count()}");
          // TODO
          return;
       }
@@ -310,6 +368,17 @@ namespace mbts.spark
          _init_vss(to_publish.Keys);
       }
 
+      void _on_hist_theos(Update<bts.api.analytics.ArchivedTheoListImmutable, ulong> update) {
+         //var total_state = update.state;
+         Log.Gen.info($"VSBBMan._on_hist_theos update: {update}");
+         foreach (var kv in update.all) {
+            bts.Log.Gen.info($"kv: {kv}");
+            _bb.post(SparkTags.ArchivedTheos, kv);
+            break;
+         }
+         //Log.assert(false);
+      }
+
       private void _init_vss(IEnumerable<int> vs_ids)
       {
          // we consider every vs in the db as "activated" which might prove to be expensive
@@ -362,29 +431,56 @@ namespace mbts.spark
             _bb.post(SparkTags.VolSurfaceNonVolInputTag,
                                           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");
-               ArchivedTheoList empty = new ArchivedTheoList();
-               _bb.post(SparkTags.ResponseArchivedVols, Tuple.Create(client_req, empty));
-               return;
+         // this is publishe dby the sheets 
+         else if (_archive_ivs_tag.Equals(tag)){
+            //Log.assert(false);
+            NMSheetSymbol sym = data as NMSheetSymbol;
+            SystemContractData sc_data = _bb.read_state(SparkTags.SystemContractsTag);            
+            var associated_syms = _get_associated_symbols(sym);
+            List<ArchivedSecurityTheo> theo_list = new List<ArchivedSecurityTheo>();
+            foreach (NMSheetSymbol s in associated_syms){
+               ExpStrikeKeyedData<TheoPoint> symbol_theos = _bb.read_state(SparkTags.TC2Tags.LiveTheosTag((s.option_name, s.option_exchange)));
+               foreach (var t in symbol_theos.get_data_linearly()){
+                  Venue v = s.option_exchange;
+                  string symbol = s.option_name;
+                  DateTime exp = t.first;
+                  double strike = t.second;
+                  TheoPoint tp = t.third;
+                  Tuple<uint?, uint?> cp_id = sc_data.get_option_sec_id(v, symbol, exp, strike);                  
+                  if (cp_id != null && cp_id.Item1.HasValue && cp_id.Item2.HasValue){
+                     // call
+                     ArchivedSecurityTheo c_theo = new ArchivedSecurityTheo();
+                     c_theo.Id = cp_id.Item1.Value;
+                     c_theo.Driver = tp.driver;
+                     c_theo.Vol = tp.vol;
+                     c_theo.Theo = tp.call;
+                     c_theo.Delta = tp.cdelta;
+                     theo_list.Add(c_theo);
+                     // put
+                     ArchivedSecurityTheo p_theo = new ArchivedSecurityTheo();
+                     p_theo.Id = cp_id.Item1.Value;
+                     p_theo.Driver = tp.driver;
+                     p_theo.Vol = tp.vol;
+                     p_theo.Theo = tp.put;
+                     p_theo.Delta = tp.pdelta;
+                     theo_list.Add(p_theo);
+                  }
+               }
             }
-            bts.Log.Ui.info($"VolSurfaceBusManager ArchivedTheoRetrieval id: {client_req.id} save_time: {ret.SaveTime} " +
-                           $"sec_tup: {ret.SecTup.Venue}_{ret.SecTup.Type}_{ret.SecTup.Symbol}");
-            var timer = Stopwatch.StartNew();
-            ArchivedTheoList arch_theos = _analytics_service.get_archived_theos(ret);
-            timer.Stop();
-            bts.Log.Ui.info($"VolSurfaceBusManager ArchivedTheoRetrieval time: {timer.Elapsed.TotalMilliseconds}ms " +
-                            $"n_entries: {arch_theos.SecTheo.Count}");
-            _bb.post(SparkTags.ResponseArchivedVols, Tuple.Create(client_req, arch_theos));
+            ArchivedTheoList rv_list = new ArchivedTheoList();
+            // we dont care abouyt the save time any more, and dont want it to te the ky of the table anymore
+            rv_list.SaveTime = 0;
+            rv_list.SecTheo.AddRange(theo_list);
+            bts.Log.Gen.info($"VSBBMan archiving {rv_list.SecTheo.Count} entries");
+            _analytics_service.clean_and_archive_theos(rv_list);
+            // send it to the server
          }
       }
 
+      private List<NMSheetSymbol> _get_associated_symbols(NMSheetSymbol symbol){
+         return new List<NMSheetSymbol>() {new NMSheetSymbol(Venue.Cboe, "SPX"),new NMSheetSymbol(Venue.Cboe, "SPXW")};
+      }
+
       private int _timer_count = 0;
       private void _on_timer()
       {

src/client/spark/AW2/LiveTheoPublisher.cs

@@ -14,6 +14,11 @@ namespace mbts.spark {
    // Publishes theos for all options (on the side, also publishes the theos for TC2).
    // Listens to driver and interp theos, on a change, calls _interp_and_publish
    public class LiveTheoPublisher : IDisposable {
+      public enum Impetus { 
+         DRIVER,
+         SKEW,
+      }
+
       private ThreadedBlackboard _bb;
       private readonly ITag<IReadOnlyDictionary<(string, Venue), OptionTheoInterpolationData>>
          _greeks_tag = SparkTags.AllInterpGreeks;
@@ -30,13 +35,14 @@ namespace mbts.spark {
          _bb.subscribe(_drivers_tag);
          _bb.data_callback += _on_data;
          _bb.timer_callback += _on_timer;
-         _interp_and_publish();
+         _interp_and_publish(Impetus.SKEW);
       }
 
       private void _on_data(object tag, object data) {
          if (tag.Equals(_drivers_tag) ||
             tag.Equals(_greeks_tag)) {
-            _interp_and_publish();
+            Impetus imp = tag.Equals(_drivers_tag) ? Impetus.DRIVER : Impetus.SKEW;
+            _interp_and_publish(imp);
          }
       }
 
@@ -103,7 +109,7 @@ namespace mbts.spark {
          return TheoPoint.getNulledObject();
       }
 
-      private void _publish_tc2_theos(IDictionary<SecurityStrike, TheoPoint> theos) {
+      private void _publish_tc2_theos(IDictionary<SecurityStrike, TheoPoint> theos, Impetus imp) {
          var rv = new Dictionary<(string, Venue), ExpStrikeKeyedData<TheoPoint>>();
          foreach (var kv in theos) {
             SecurityStrike k = kv.Key;
@@ -118,6 +124,9 @@ namespace mbts.spark {
          _override_theos(ref rv);
          foreach (var kv in rv) {
             _bb.post(SparkTags.TC2Tags.LiveTheosTag(kv.Key), kv.Value);
+            if (imp == Impetus.SKEW) {
+               _bb.post(SparkTags.TC2Tags.VolSaveEventTheosTag(kv.Key), kv.Value);
+            }
          }
       }
 
@@ -132,6 +141,8 @@ namespace mbts.spark {
                   double strike = t.second;
                   TheoPoint tp = t.third;
                   var call_sec_md = sc_data.get_option(exchange, sym, exp, strike, PutOrCall.Call);
+                  if (call_sec_md == null)
+                     continue;
                   {
                      NativePriceSize call_bid, call_ask;
                      ulong time = 0;
@@ -141,6 +152,8 @@ namespace mbts.spark {
                   }
                   {
                      var put_sec_md = sc_data.get_option(exchange, sym, exp, strike, PutOrCall.Put);
+                     if (put_sec_md == null)
+                        continue;
                      NativePriceSize put_bid, put_ask;
                      ulong time = 0;
                      put_sec_md.Value.md.get_bbo(ref time, out put_bid, out put_ask);
@@ -163,7 +176,7 @@ namespace mbts.spark {
       }
 
 
-      private void _interp_and_publish() {
+      private void _interp_and_publish(Impetus imp) {
          var timer = Stopwatch.StartNew();
          var res = new Dictionary<uint, GreeksTuple>();
          var driver_map = _bb.read_state(_drivers_tag);
@@ -205,13 +218,13 @@ namespace mbts.spark {
          }
          // copy and publish
          IReadOnlyDictionary<uint, GreeksTuple> res_ro = res;
-         bts.Log.Gen.info($"LiveTheoPublisher._interp_and_publish posting " +
+         bts.Log.Gen.info($"LiveTheoPublisher._interp_and_publish Impetus: {imp} posting " +
                              $"count: {res_ro.Count}");
          timer.Stop();
          _logger.log(_log_id, _log_id, timer.Elapsed.TotalMilliseconds);
          _bb.post(SparkTags.Greeks, res_ro);
          // oh and those TC2 theos for comptability
-         _publish_tc2_theos(k_theos);
+         _publish_tc2_theos(k_theos, imp);
       }
 
       // If we do any rolls like SP for futures, it will be here

src/client/spark/Analytics/Core/VolSurfaceConstants.cs

@@ -8,7 +8,7 @@ using System.Threading.Tasks;
 
 namespace mbts.analytics {
    public static class VolSurfaceConstants {
-      public static int vol_time_sample_mins = 1;
+      public static int vol_time_sample_mins = 5;
    }
    // TODO: remove this
    public static class VolSurfaceTemp {

src/client/spark/Analytics/Misc/VolSurfaceTimeCalculatorFactory.cs

@@ -21,13 +21,23 @@ namespace mbts.analytics {
       private class TimeCalculator : IVolSurfaceTimeCalculator {
          // TODO, this should be passed in the exp itself
          private Dictionary<string, TimeSpan> utc_time_of_exp_map = new Dictionary<string, TimeSpan>() {
+            // TODO: get it from config
             {"BTC", TimeSpan.FromHours(8)},
-            {"ETH", TimeSpan.FromHours(8)}
+            {"ETH", TimeSpan.FromHours(8)},
+            {"SPXW", TimeSpan.FromHours(20)},
+            {"SPX", new TimeSpan(13, 30, 0)}
          };
          public void compute_vol_time(VolSurfaceInput vs_input, VolSurfaceExpiration exp,
                                       DateTime voltime, double days_shift,
                                       out double yrs, out double days_per_yr) {
+
             days_per_yr = 365;
+            if (vs_input != null && vs_input.surface_settings != null && vs_input.surface_settings.core_settings != null
+               && vs_input.surface_settings.core_settings.time_settings != null) {
+               if (vs_input.surface_settings.core_settings.time_settings.vol_time_measure.account_for_holidays_and_weekends) {
+                  days_per_yr = vs_input.surface_settings.core_settings.time_settings.vol_time_measure.days_per_year;
+               }
+            }
             TimeSpan time_of_exp = TimeSpan.Zero;
             utc_time_of_exp_map.TryGetValue(exp.symbol, out time_of_exp);
             // assumes exp is the utc date of expiration

src/client/spark/Analytics/Pricing/VolSurfaceInterpGreeksProducer.cs

@@ -9,8 +9,8 @@ using bts.api;
 using mbts.spark;
 
 namespace mbts.analytics {
-   // This class computes the theos/greeks for a number of driver
-   // prices (specified in the vs settings).
+   // This class computes interp theos/greeks (ie for a number of driver
+   // prices (specified in the vs settings)) for a volsurface
    // It has public setters that get
    // 1. contracts
    // 2. vs input
@@ -41,7 +41,8 @@ namespace mbts.analytics {
       private IDictionary<VolSurfaceExpiration, Tuple<double, double>>
       _computed_driver_range_map = new Dictionary<VolSurfaceExpiration, Tuple<double, double>>();
       private readonly IEnumerable<double> _default_abs_moves =
-         new double[] { 300, -200, -125, -100, -75, -50, -25, 0, 25, 50, 75, 100, 125, 200, 300 };
+         //new double[] { 300, -200, -125, -100, -75, -50, -25, 0, 25, 50, 75, 100, 125, 200, 300 };
+         new double[] { -100, -50, 0, 50, 100 };
       private readonly double _cushion_percentage = 25.0;
 
       public event
@@ -135,8 +136,8 @@ namespace mbts.analytics {
                      false, true, false, false, false, _filter_map);
                timer.stop();
                vol_compute_time = timer.last_interval_time_ms;
-               Log.Gen.info($"VolSurfacePrecomputedTheosProducer._compute_and_fire volmat time " +
-                           $"{timer.last_interval_time_ms}");
+               Log.Gen.info($"VolSurfacePrecomputedTheosProducer._compute_and_fire volmat input_type: {input_type} vs_id: {_vs_id} time " +
+                           $"{timer.last_interval_time_ms} tid: {System.Threading.Thread.CurrentThread.ManagedThreadId}");
                timer.start();
                IDictionary<VolSurfaceExpiration, IEnumerable<Theos>> exp_theo_map =
                  VolSurfaceTheoCalculator.compute_standard_greeks(synced_vs_input.surface_settings,

src/client/spark/Daemons/MergedMDDaemon.cs

@@ -111,13 +111,43 @@ namespace bts
          _merge_and_post();
       }
 
+
+
+      private static bool is_active(SecurityMD sec_md) {
+         SecurityImmutable sec = sec_md.security;
+         // temp: to see if spreads work
+         bool fut_or_option = (sec.security_type() == bts.api.SecurityType.Future || sec.security_type() == bts.api.SecurityType.Option);
+         if (!fut_or_option)
+            return true;
+         Func<bool> is_daily_option = () => {
+            bool is_daily = false;
+            if (sec.Exchange == Venue.Cboe) {
+               is_daily = sec.Symbol.ToLower() == "spxw";
+            }
+            if (sec.Exchange == Venue.Glbx) {
+               string symbol = sec.Symbol.ToLower();
+               int unused;
+               is_daily = symbol.Length >= 3 && symbol[0] == 'e' && Int32.TryParse(symbol[1].ToString(), out unused);
+            }
+            return is_daily;
+         };
+         bool is_expired = !SecurityExpirationHelper.is_unexpired(sec);
+         if (is_expired) {
+            return false;
+         }
+         if (sec.security_type() == bts.api.SecurityType.Option) {
+            int days_to_exp = Convert.ToInt32((sec.earliest_expiration() - DateTime.Now).TotalDays);
+            if (is_daily_option()) {
+               return days_to_exp <= 14;
+            } else {
+               return days_to_exp <= 90;
+            }
+         }
+         return true;            
+      }   
+
       private void _merge_and_post()
       {
-         Func<SecurityMD, bool> is_active = (sec_md) =>
-         {
-            var sec = sec_md.security;
-            return SecurityExpirationHelper.is_unexpired(sec);
-         };
          var to_pub = _server_sec_map.Where(kv => is_active(kv.Value)).ToDictionary(kv => kv.Key, kv => kv.Value);
          foreach (var kv in _fake_sec_map)
          {

src/client/spark/NMSheets/BlackboardStructure/NMSheetSingleOptionData.cs

@@ -237,7 +237,7 @@ namespace BCG {
          PNLRollOverDataUnit u;
          if (_roll_over_data != null && (u = _roll_over_data.getValue(exp, strike, true)) != null) {
             cp_row[(int)NMSheetColumnName.vol_change] = tp.vol - u.vol;
-            cp_row[(int)NMSheetColumnName.straddle_change] = ((tp.vol - u.vol) * tp.cvega * 2.0 * 100.0) / tp.driver;
+            cp_row[(int)NMSheetColumnName.straddle_change] = (tp.vol - u.vol) * tp.cvega * 2.0 * 100.0;
          } else {
             cp_row[(int)NMSheetColumnName.vol_change] = Double.NaN;
             cp_row[(int)NMSheetColumnName.straddle_change] = Double.NaN;

src/client/spark/NMSheets/DataProviders/NMSheetArchivedVolProvider.cs

@@ -14,103 +14,23 @@ using mbts.analytics;
 using bts.api.analytics;
 
 namespace mbts.spark {
-   public class ClientArchivedVolRequest {
-      public readonly int id;
-      public readonly ArchivedTheoRetrieval ret;
-      public ClientArchivedVolRequest(Int32 id, ArchivedTheoRetrieval ret) {
-         this.id = id;
-         this.ret = ret;
-      }
-      private static int _id = 0;
-      public static int get_new_id() {
-         int rv = 0;
-         lock (typeof(ClientArchivedVolRequest)) {
-            rv = ++_id;
-         }
-         return rv;
-      }
-      public override string ToString() {
-         return $"id: {id} sym: {ret.SecTup.Symbol} vol_time: {ret.SaveTime}";
-      }
-   }
    // This is created by the NMSheetSingleOptionMainDataSource - it downloads the
    // archived theos and provides access to it
    public class NMSheetArchivedVolProvider : IDisposable {
       private readonly ThreadedBlackboard _bb;
-      private readonly NMSheetSymbol _sym;
-      private readonly bts.api.SecurityTuple _sec_tup;
       private Action _cb = () => { };
       private readonly ITag<SystemContractData> _sc_tag = SparkTags.SystemContractsTag;
-      private readonly ITag<NMSheetSettings> _settings_tag;
-      private readonly ITag<Tuple<ClientArchivedVolRequest, bts.api.analytics.ArchivedTheoList>>
-         _response_tag = SparkTags.ResponseArchivedVols;
-      private bool _waiting_for_reponse = false;
-      private ClientArchivedVolRequest _last_req;
-      bts.api.analytics.ArchivedTheoList _cached_data = null;
+      private Nullable<ArchivedTheoListImmutable> _cached_data;
 
       public NMSheetArchivedVolProvider(string req_id, ThreadedBlackboard bb,
                                        NMSheetSymbol symbol,
                                        Action cb) {
          bts.Log.Ui.info($"NMSheetArchivedVolProvider req_id: {req_id} symbol: {symbol}");
          _bb = bb;
-         _sym = symbol;
-         _sec_tup = new bts.api.SecurityTuple() {
-            Venue = _sym.option_exchange,
-            Type = SecurityType.Option,
-            Symbol = _sym.option_name
-         };
-         _last_req = new ClientArchivedVolRequest(
-            0,
-            new ArchivedTheoRetrieval() {
-               SaveTime = 0,
-               SecTup = _sec_tup
-            });
          _cb = cb;
-         _settings_tag = SparkTags.TC2Tags.NMSheetSettingsTag(req_id);
          _bb.subscribe(_sc_tag);
-         _bb.subscribe(_settings_tag);
-         _bb.subscribe(_response_tag);
-         _bb.data_callback += _on_data;
-         _check_and_send_request(true);
-      }
-
-      private static UInt64 _get_vol_time(NMSheetSettings settings) {
-         NMSheetSettingLayout.VolDay vol_day = settings.setting_layout.vol_day;
-         TimeSpan vol_time = settings.setting_layout.vol_time;
-         DateTime date = DateTime.Today.Date;
-         switch (vol_day) {
-            case NMSheetSettingLayout.VolDay.DayBeforeYesterday:
-               date = date.Subtract(TimeSpan.FromDays(2));
-               break;
-            case NMSheetSettingLayout.VolDay.Yesterday:
-               date = date.Subtract(TimeSpan.FromDays(1));
-               break;
-            case NMSheetSettingLayout.VolDay.Today:
-               break;
-         }
-         DateTime dt = date.Add(vol_time);
-         var rv = VolSurfaceUtilityFunctions.get_unix_epoch_msecs(dt);
-         bts.Log.Ui.info($"NMSheetArchivedVolProvider._get_vol_time vol_day: {vol_day} " +
-            $"vol_time: {vol_time} dt: {dt} epoch_md: {rv}");
-         return rv;
-      }
-
-      private void _check_and_send_request(bool force = false) {
-         NMSheetSettings settings = _bb.read_state(_settings_tag);
-         if (settings != null && settings.setting_layout != null) {
-            UInt64 vol_time = _get_vol_time(settings);
-            if (force || _last_req.ret.SaveTime != vol_time) {
-               _last_req = new ClientArchivedVolRequest(
-                     ClientArchivedVolRequest.get_new_id(),
-                     new ArchivedTheoRetrieval() {
-                        SaveTime = vol_time,
-                        SecTup = _sec_tup
-                     });
-               _waiting_for_reponse = true;
-               bts.Log.Gen.info($"NMSheetArchivedVolProvider._check_and_send_request req: {_last_req}");
-               _bb.post(SparkTags.RequestArchivedVols, _last_req);
-            }
-         }
+         _bb.subscribe(SparkTags.ArchivedTheos);
+         _bb.data_callback += _on_data;         
       }
 
       private void _compute_and_post() {
@@ -118,34 +38,22 @@ namespace mbts.spark {
       }
 
       private void _on_data(object tag, object data) {
-         if (_settings_tag.Equals(tag)) {
-            _check_and_send_request();
-         } else if (_response_tag.Equals(tag)) {
-            if (_waiting_for_reponse) {
-               Tuple<ClientArchivedVolRequest, bts.api.analytics.ArchivedTheoList> resp =
-                  _bb.read_state(_response_tag);
-               if (resp != null) {
-                  ClientArchivedVolRequest responsed_req = resp.Item1;
-                  if (responsed_req.id == _last_req.id) {
-                     // got we were looking for
-                     _cached_data = resp.Item2;
-                     _waiting_for_reponse = false;
-                     bts.Log.Gen.info($"NMSheetArchivedVolProvider._on_data recvd data, req: {_last_req}");
-                     _compute_and_post();
-                  }
-               }
-            }
-         } else if (_sc_tag.Equals(tag)) {
+         // just listen to archihved theos and fire callback
+         if (tag.Equals(SparkTags.ArchivedTheos)) {
+            _cached_data = new Nullable<ArchivedTheoListImmutable>((ArchivedTheoListImmutable)data);
+            _compute_and_post();
+          }
+         else if (_sc_tag.Equals(tag)) {
             _compute_and_post();
          }
       }
 
       public OptionContractKeyedData<PNLRollOverDataUnit> get_all_data() {
          SystemContractData sc_data = _bb.read_state(_sc_tag);
-         if (_cached_data != null && sc_data != null) {
+         if (sc_data != null && _cached_data != null && _cached_data.HasValue) {
             OptionContractKeyedData<PNLRollOverDataUnit> rv =
                new OptionContractKeyedData<PNLRollOverDataUnit>();
-            foreach (var v in _cached_data.SecTheo) {
+            foreach (var v in _cached_data.Value.SecTheo) {
                var sec_md = sc_data.get_sec_md(v.Id);
                try {
                   if (sec_md != null) {

src/client/spark/NMSheets/UIComponents/MainSheet/NMSheetControl.Designer.cs

@@ -1,5 +1,7 @@
-namespace BCG {
-   partial class NMSheetControl {
+namespace BCG
+{
+   partial class NMSheetControl
+   {
       /// <summary> 
       /// Required designer variable.
       /// </summary>
@@ -9,9 +11,11 @@
       /// Clean up any resources being used.
       /// </summary>
       /// <param name="disposing">true if managed resources should be disposed; otherwise, false.</param>
-      protected override void Dispose(bool disposing) {
+      protected override void Dispose(bool disposing)
+      {
          _logical_dispose(disposing);
-         if (disposing && (components != null)) {
+         if (disposing && (components != null))
+         {
             components.Dispose();
 
          }
@@ -24,7 +28,8 @@
       /// Required method for Designer support - do not modify 
       /// the contents of this method with the code editor.
       /// </summary>
-      private void InitializeComponent() {
+      private void InitializeComponent()
+      {
          this.components = new System.ComponentModel.Container();
          this._menu_panel = new System.Windows.Forms.Panel();
          this._button_minimize_spreads = new System.Windows.Forms.Button();
@@ -67,6 +72,7 @@
          this._toolstripmenuitem_task_manage_templates = new System.Windows.Forms.ToolStripMenuItem();
          this._toolstripseperator_task_eye = new System.Windows.Forms.ToolStripSeparator();
          this._toolstrip_menuitem_save_data_to_file = new System.Windows.Forms.ToolStripMenuItem();
+         this._toolstrip_menuitem_save_ivs = new System.Windows.Forms.ToolStripMenuItem();
          this._toolstripseperator_export_data = new System.Windows.Forms.ToolStripSeparator();
          this._toolstripmenuitem_card = new System.Windows.Forms.ToolStripMenuItem();
          this._toolstripseperator_expirations = new System.Windows.Forms.ToolStripSeparator();
@@ -293,7 +299,8 @@
             this._toolstripmenuitem_atm,
             this._toolseperator_atm,
          };
-         if (_api_tasking_enabled) {
+         if (_api_tasking_enabled)
+         {
             right_click_menu.AddRange(new System.Collections.Generic.List<System.Windows.Forms.ToolStripItem>() {
                this._toolstripmenuitem_task_eye,
                // no quoter
@@ -307,7 +314,8 @@
          //       this._toolstripmenuitem_task_override_size,
          //    });
          // }
-         if (_api_tasking_enabled) {
+         if (_api_tasking_enabled)
+         {
             right_click_menu.AddRange(new System.Collections.Generic.List<System.Windows.Forms.ToolStripItem>() {
                this._toolstripmenuitem_task_eye_switch_all_off,
                // We can do this from the algo studio now
@@ -317,6 +325,7 @@
          right_click_menu.AddRange(new System.Collections.Generic.List<System.Windows.Forms.ToolStripItem>() {
             // this._toolstripseperator_task_eye,
             this._toolstrip_menuitem_save_data_to_file,
+            this._toolstrip_menuitem_save_ivs,
             this._toolstripseperator_export_data,
             this._toolstripmenuitem_card,
             this._toolstripseperator_expirations,
@@ -324,7 +333,8 @@
             this._toolstripmenuitem_minimize_spreads,
             this._toolstripseperator_filter,
          });
-         if (_api_tasking_enabled) {
+         if (_api_tasking_enabled)
+         {
             right_click_menu.AddRange(new System.Collections.Generic.List<System.Windows.Forms.ToolStripItem>() {
                this._toolstripmenuitem_filter,
                this._toolseperator_quotesets
@@ -497,6 +507,13 @@
          this._toolstripseperator_task_eye.Name = "_toolstripseperator_task_eye";
          this._toolstripseperator_task_eye.Size = new System.Drawing.Size(213, 6);
          // 
+         // _toolstrip_menuitem_save_ivs
+         // 
+         this._toolstrip_menuitem_save_ivs.Name = "_toolstrip_menuitem_save_ivs";
+         this._toolstrip_menuitem_save_ivs.Size = new System.Drawing.Size(216, 22);
+         this._toolstrip_menuitem_save_ivs.Text = "Save IVs";
+         //
+         // 
          // _toolstrip_menuitem_save_data_to_file
          // 
          this._toolstrip_menuitem_save_data_to_file.Name = "_toolstrip_menuitem_save_data_to_file";
@@ -581,7 +598,8 @@
             this._toolstripmenuitem_settings_volchange,
             this._toolstripmenuitem_settings_straddle
          };
-         if (_api_tasking_enabled) {
+         if (_api_tasking_enabled)
+         {
             sheet_settings_items.AddRange(new System.Collections.Generic.List<System.Windows.Forms.ToolStripItem>() {
                this._toolseperator_task_ee,
                this._toolstripmenuitem_settings_show_task_ee_data,
@@ -900,6 +918,7 @@
       private System.Windows.Forms.ToolStripSeparator _toolstripseparator_task_size;
       private System.Windows.Forms.ToolStripMenuItem _toolstripmenuitem_task_undo_size_override;
       private System.Windows.Forms.ToolStripMenuItem _toolstrip_menuitem_save_data_to_file;
+      private System.Windows.Forms.ToolStripMenuItem _toolstrip_menuitem_save_ivs;
       private System.Windows.Forms.ToolStripSeparator _toolstripseperator_export_data;
       private System.Windows.Forms.Button _button_theo_acceptance_confirmation;
       private System.Windows.Forms.ToolStripMenuItem _toolstripmenuitem_filter_strikes_with_pos_edge;

src/client/spark/NMSheets/UIComponents/MainSheet/NMSheetControl.cs

@@ -20,6 +20,8 @@ using System.Collections;
 using System.Reactive.Linq;
 using System.Threading;
 using bts.api.orderman;
+using bts;
+using mbts.analytics;
 
 namespace BCG {
    public partial class NMSheetControl : UserControl {
@@ -252,7 +254,7 @@ namespace BCG {
          /// windows timer
          _windows_timer = new System.Windows.Forms.Timer(components);
          _windows_timer.Enabled = true;
-         _windows_timer.Interval = 20;
+         _windows_timer.Interval = 100;
          _windows_timer.Tick += _windows_timer_tick;
          _toolstripmenuitem_card.Visible = false;
          /// status bar
@@ -317,6 +319,9 @@ namespace BCG {
             _set_new_settings(_settings);
             _theo_grid.theo_grid.scroll_to_point(Point.Empty);
          };
+         _toolstrip_menuitem_save_ivs.Click += (s, e) => {
+            _threaded_bb.post(SparkTags.TC2Tags.ArchiveIVsTag, _nmsheet_symbol_list[0]);
+         };
          _toolstripmenuitem_filter_strikes_in_tasks.Click += (s, e) => {
             _settings.setting_filter.filter_ee_task_strikes = !_settings.setting_filter.filter_ee_task_strikes;
             _set_new_settings(_settings);
@@ -612,7 +617,7 @@ namespace BCG {
                cell.col == NMSheetColumnName.put_ask_price || cell.col == NMSheetColumnName.put_ask_size) {
                _launch_order_entry_window(cell, _is_control_down() ? OrderEntryWindow.Ladder : OrderEntryWindow.Ticket);
             } else if (cell.col == NMSheetColumnName.call_theo || cell.col == NMSheetColumnName.put_theo ||
-                       cell.col == NMSheetColumnName.straddle || cell.col == NMSheetColumnName.strike) {
+                         cell.col == NMSheetColumnName.straddle || cell.col == NMSheetColumnName.strike) {
                if (cell.call_selected)
                   _add_spread_leg(cell.expiration, cell.strike, true, false, leg_size, state);
                if (cell.put_selected)
@@ -717,9 +722,9 @@ namespace BCG {
 
       private void _windows_timer_tick(object sender, EventArgs e) {
          _publish_activation_state();
-         //_threaded_bb.sec_thread_dowork();
+         // this computes the data and then paints it
          _check_and_propagate_data();
-         _poll_theo_acceptance_button();
+         //_poll_theo_acceptance_button();
          _refresh_statusbar_data(false);
       }
 
@@ -780,21 +785,31 @@ namespace BCG {
       #region Coalescer NMSheetData interaction
 
       private DateTime _last_data_draw_time = DateTime.MinValue;
-      private const double _min_draw_period_ms = 333.3;
+      private const double _min_draw_period_ms = 500;
       private NMSheetGeometricLayout _last_geo_lo_info = null;
+      private VolSurfaceTimer _vs_timer = new VolSurfaceTimer();
+      // this is where we assemble the data (in _coalescer.get_data())
+      // and then painted in _respond_to_data
       private void _check_and_propagate_data() {
          if (is_visible) {
             if ((DateTime.Now - _last_data_draw_time).Milliseconds >= Math.Max(_settings.setting_grid_internal.grid_paint_throttle_ms, _min_draw_period_ms)) {
+               _vs_timer.start();
                NMSheetData new_sheet_data = _coalescer.get_data();
+               _vs_timer.stop();
+               Log.Gen.info($"NMSheetControl._check_and_propagate_data get_data time: {_vs_timer.last_interval_time_ms} msecs symbols: {_title_prefix} tid: {System.Threading.Thread.CurrentThread.ManagedThreadId}");
                if (new_sheet_data != null) {
                   _last_data_draw_time = DateTime.Now;
                   _sheet_data = new_sheet_data;
+                  _vs_timer.start();
                   _respond_to_data();
+                  _vs_timer.stop();
+                  Log.Gen.info($"NMSheetControl._check_and_propagate_data _respond_to_data time: {_vs_timer.last_interval_time_ms} msecs symbols: {_title_prefix} tid: {System.Threading.Thread.CurrentThread.ManagedThreadId}");
                }
             }
          }
       }
 
+      // this is the painting
       private void _respond_to_data() {
          if (_sheet_data != null) {
             _set_driver_control_display();

src/client/spark/Security/SecurityExpirationHelper.cs

@@ -36,9 +36,8 @@ namespace mbts {
       }
 
       public static bool is_unexpired(bts.api.SecurityTuple st, DateTime exp_date) {
-         if (st.Venue == Venue.Cboe || st.Venue == Venue.Glbx) {
-            // server is already getting teh righ contracts
-            return true;
+         if (st.Venue == Venue.Cboe) {
+            return exp_date.Date >= DateTime.Now.Date;
          }
          TimeSpan exp_time = get_exp_time(st);
          DateTime full_exp_utc = exp_date.ToUniversalTime().Date.Add(exp_time);
@@ -47,8 +46,12 @@ namespace mbts {
       }
 
       public static bool is_unexpired(bts.api.SecurityImmutable sec) {
-         if (sec.security_tuple().Venue == Venue.Cboe || sec.security_tuple().Venue == Venue.Glbx)
-            return true;
+         if (sec.security_tuple().Venue == Venue.Cboe || sec.security_tuple().Venue == Venue.Glbx) {
+            var exp = sec.to_sec_exp().expiration;
+            bool is_unexp = exp.Date >= DateTime.Now.Date;
+//            Log.Gen.info($"SecurityExpirationMapper.is_unexpired exp: {exp} is_unexp: {is_unexp}");
+            return is_unexp;
+         }
          var sec_type = sec.security_type();
          switch (sec_type) {
             case bts.api.SecurityType.Future:

src/client/spark/core/Bus/SparkTags.cs

@@ -353,11 +353,8 @@ namespace mbts.spark {
 
       public static ITag<Nothing> ShowGlobalAlertsTag { get; } = _mk_tag<Nothing>();
 
-      public static ITag<ClientArchivedVolRequest> RequestArchivedVols { get; } = _mk_tag<ClientArchivedVolRequest>();
-
-      public static ITag<Tuple<ClientArchivedVolRequest, bts.api.analytics.ArchivedTheoList>>
-      ResponseArchivedVols { get; } =
-         _mk_tag<Tuple<ClientArchivedVolRequest, bts.api.analytics.ArchivedTheoList>>();
+      public static ITag<ArchivedTheoListImmutable>
+      ArchivedTheos { get; } = _mk_tag<ArchivedTheoListImmutable>();
 
       static SparkTag<T> _mk_tag<T>([CallerMemberName] string name = null) =>
          new(name ?? throw new ArgumentNullException(nameof(name)));
@@ -373,6 +370,7 @@ namespace mbts.spark {
       public class Nothing { Nothing() { } }
 
       public class TC2Tags {
+         public static ITag<NMSheetSymbol> ArchiveIVsTag {get;} = _mk_tag<NMSheetSymbol>();
          public static ITag<NMSheetAllTaskingData> TaskData { get; } =
             _mk_tag<NMSheetAllTaskingData>();
          // Data request for a sheet 
@@ -386,6 +384,11 @@ namespace mbts.spark {
          public static ITag<ExpStrikeKeyedData<TheoPoint>> LiveTheosTag((string, Venue) symbol) =>
             _mk_tag<ExpStrikeKeyedData<TheoPoint>, (string, Venue)>(symbol);
 
+         // these are the theos at time of the vol save (we use it to extrapolate the theos prices for the spreads)
+         public static ITag<ExpStrikeKeyedData<TheoPoint>> VolSaveEventTheosTag((string, Venue) symbol) =>
+            _mk_tag<ExpStrikeKeyedData<TheoPoint>, (string, Venue)>(symbol);
+
+
          // Aggregated version of the above
          public static ITag<IReadOnlyDictionary<(string, Venue), ExpStrikeKeyedData<TheoPoint>>>
          AllLiveTheosTag { get; } =

src/lib/api/services/AnalyticsService.proto

@@ -41,7 +41,7 @@ message VolSurfaceUsageList {
 }
 
 message ArchivedSecurityTheo {
-  uint32 id = 10;
+  uint32 id = 10[ (bts.api.key) = true ];
   double driver = 20;
   double vol = 30;
   double theo = 40;
@@ -49,7 +49,7 @@ message ArchivedSecurityTheo {
 }
 
 message ArchivedTheoList {
-  uint64 save_time = 10;
+  uint64 save_time = 10[ (bts.api.key) = true ];
   // these are sec_tups that are part of this save
   repeated bts.api.SecurityTuple sec_tups = 20;
   repeated ArchivedSecurityTheo sec_theo = 30;
@@ -195,8 +195,10 @@ service AnalyticsService {
 
   /// theos
   rpc archive_theos(ArchivedTheoList) returns (bts.api.ResultMessage);
+  rpc clean_and_archive_theos(ArchivedTheoList) returns (bts.api.ResultMessage);
   // finds the archived_theo closest to this point
   rpc get_archived_theos(ArchivedTheoRetrieval) returns (ArchivedTheoList);
+  rpc archived_theos(bts.api.VoidMessage) returns (stream ArchivedTheoList);
 
   /// reports
   rpc get_new_report_id(bts.api.VoidMessage)

src/lib/config/profiles/default/toggles.conf

@@ -114,13 +114,139 @@ VolSurfaceUsage {
      items: {
       option_name: "SPX"
       option_venue: VENUE_CBOE
-      vs_name: "Main"
+      vs_name: "CBOE_skews"
     }
     items: {
       option_name: "SPXW"
       option_venue: VENUE_CBOE
-      vs_name: "Main"
+      vs_name: "CBOE_skews"
     }
+    items: {
+      option_name: "ES"
+      option_venue: VENUE_GLBX
+      vs_name: "CME_Skews"
+    }     
+     items: {
+      option_name: "E1A"
+      option_venue: VENUE_GLBX
+      vs_name: "CME_Skews"
+    }
+     items: {
+      option_name: "E2A"
+      option_venue: VENUE_GLBX
+      vs_name: "CME_Skews"
+    }
+     items: {
+      option_name: "E3A"
+      option_venue: VENUE_GLBX
+      vs_name: "CME_Skews"
+    }
+    items: {
+      option_name: "E4A"
+      option_venue: VENUE_GLBX
+      vs_name: "CME_Skews"
+    }
+    items: {
+      option_name: "E5A"
+      option_venue: VENUE_GLBX
+      vs_name: "CME_Skews"
+    }
+    items: {
+      option_name: "E1B"
+      option_venue: VENUE_GLBX
+      vs_name: "CME_Skews"
+    }
+     items: {
+      option_name: "E2B"
+      option_venue: VENUE_GLBX
+      vs_name: "CME_Skews"
+    }
+     items: {
+      option_name: "E3B"
+      option_venue: VENUE_GLBX
+      vs_name: "CME_Skews"
+    }
+    items: {
+      option_name: "E4B"
+      option_venue: VENUE_GLBX
+      vs_name: "CME_Skews"
+    }
+    items: {
+      option_name: "E5B"
+      option_venue: VENUE_GLBX
+      vs_name: "CME_Skews"
+    }
+    items: {
+      option_name: "E1C"
+      option_venue: VENUE_GLBX
+      vs_name: "CME_Skews"
+    }
+     items: {
+      option_name: "E2C"
+      option_venue: VENUE_GLBX
+      vs_name: "CME_Skews"
+    }
+     items: {
+      option_name: "E3C"
+      option_venue: VENUE_GLBX
+      vs_name: "CME_Skews"
+    }
+    items: {
+      option_name: "E4C"
+      option_venue: VENUE_GLBX
+      vs_name: "CME_Skews"
+    }
+    items: {
+      option_name: "E5C"
+      option_venue: VENUE_GLBX
+      vs_name: "CME_Skews"
+    }
+    items: {
+      option_name: "E1D"
+      option_venue: VENUE_GLBX
+      vs_name: "CME_Skews"
+    }
+     items: {
+      option_name: "E2D"
+      option_venue: VENUE_GLBX
+      vs_name: "CME_Skews"
+    }
+     items: {
+      option_name: "E3D"
+      option_venue: VENUE_GLBX
+      vs_name: "CME_Skews"
+    }
+    items: {
+      option_name: "E4D"
+      option_venue: VENUE_GLBX
+      vs_name: "CME_Skews"
+    }
+    items: {
+      option_name: "E5D"
+      option_venue: VENUE_GLBX
+      vs_name: "CME_Skews"
+    }
+    items: {
+      option_name: "EW1"
+      option_venue: VENUE_GLBX
+      vs_name: "CME_Skews"
+    }
+     items: {
+      option_name: "EW2"
+      option_venue: VENUE_GLBX
+      vs_name: "CME_Skews"
+    }
+     items: {
+      option_name: "EW3"
+      option_venue: VENUE_GLBX
+      vs_name: "CME_Skews"
+    }
+    items: {
+      option_name: "EW4"
+      option_venue: VENUE_GLBX
+      vs_name: "CME_Skews"
+    }
+
 
   }
 }

src/lib/domain/SecurityCommon.h

@@ -10,6 +10,7 @@
 #include <limits>
 #include <nlohmann/json.hpp>
 #include <string>
+#include <vector>
 
 namespace bts::domain {
 
@@ -31,6 +32,7 @@ class SecurityCommon {
    double min_quantity = 0;
    double max_quantity = std::numeric_limits<double>::max();
    double cash_per_point = 1;
+   std::vector<double> price_ticks;
 
    SecurityCommon() {}
    SecurityCommon(const bts::api::Security &s)

src/lib/md/GeneralMarketDataListener.cpp

@@ -746,6 +746,11 @@ void GeneralMarketDataListener::_copy_security_common(
    to.display_factor = 0.0;
    to.price_break = 0.0;
    to.low_tick = to.high_tick = from.security_common.min_price_increment;
+   if(from.security_common.price_ticks.size() == 3){
+      to.price_break = from.security_common.price_ticks[0];
+      to.low_tick =  from.security_common.price_ticks[1];
+      to.high_tick=  from.security_common.price_ticks[2];
+   }
    auto &sc = from.security_common;
    double min_qty = sc.min_quantity;
    double qty_incr = sc.min_quantity_increment;

src/server/sparkd/analytics/AnalyticsServiceServer.cpp

@@ -106,6 +106,8 @@ struct AnalyticsServiceServer::PImpl {
    AnalyticsServiceBase::Exp_parametersStatefulPublisher exp_params_publisher;
    AnalyticsServiceBase::Volsurface_usageStatefulPublisher vs_usage_publisher;
    AnalyticsServiceBase::ReportsStatefulPublisher reports_publisher;
+   AnalyticsServiceBase::Archived_theosStatefulPublisher
+       archived_theos_publisher;
    PImpl(
        bts::api::ServiceContext &context,
        const AnalyticsServiceBase::VolsurfacesStatefulPublisher &vs_publisher,
@@ -113,7 +115,9 @@ struct AnalyticsServiceServer::PImpl {
            &exp_params_publisher,
        const AnalyticsServiceBase::Volsurface_usageStatefulPublisher
            &vs_usage_publisher,
-       const AnalyticsServiceBase::ReportsStatefulPublisher &reports_publisher)
+       const AnalyticsServiceBase::ReportsStatefulPublisher &reports_publisher,
+       const AnalyticsServiceBase::Archived_theosStatefulPublisher
+           &archived_theos_publisher_)
        : context(context),
          vs_store(std::make_unique<persistence::SqliteStore>("volsurfaces")),
          exp_params_store(std::make_unique<persistence::SqliteStore>(
@@ -125,9 +129,17 @@ struct AnalyticsServiceServer::PImpl {
              std::make_unique<persistence::SqliteStore>("archived_theos")),
          vs_publisher(vs_publisher), exp_params_publisher(exp_params_publisher),
          vs_usage_publisher(vs_usage_publisher),
-         reports_publisher(reports_publisher) {}
+         reports_publisher(reports_publisher),
+         archived_theos_publisher(archived_theos_publisher_) {}
 };
 
+void AnalyticsServiceServer::_publish_archived_theos(
+    const api::analytics::ArchivedTheoList &l) {
+   // remove all rows
+   me->archived_theos_publisher.add_or_update(l);
+   me->archived_theos_publisher.flush();
+}
+
 AnalyticsServiceServer::AnalyticsServiceServer(
     bts::api::ServiceContext &context)
     : AnalyticsServiceBase(context),
@@ -140,7 +152,8 @@ AnalyticsServiceServer::AnalyticsServiceServer(
           // make_volsurfaces_stateful_publisher(),
           make_exp_parameters_stateful_publisher(),
           make_volsurface_usage_stateful_publisher(),
-          make_reports_stateful_publisher())) {
+          make_reports_stateful_publisher(),
+          make_archived_theos_stateful_publisher())) {
    // publishing all vol surfaces
    LOG(Gen, Info, "Starting AnalyticsServiceServer");
    _clean_up_vss();
@@ -206,6 +219,28 @@ AnalyticsServiceServer::AnalyticsServiceServer(
       }
       me->reports_publisher.flush();
    }
+   // archived theos
+   {
+      size_t n = 0;
+      // ns_table is ns vs {key, val}[]
+      auto ns_table = me->theos_store->retrieve_all();
+      LOG(OrderMan, Info,
+          "AnalyticsServiceServer::cons loaded archived_theos n: {}",
+          ns_table.size());
+      bts_assert(ns_table.size() == 0 || ns_table.size() == 1);
+      for (const auto &ns_vals : ns_table) {
+         for (const auto &kvp : ns_vals.second) {
+            ArchivedTheoList theo_list;
+            utils::pb::from_string(kvp.second, &theo_list);
+            _publish_archived_theos(theo_list);
+            n += theo_list.sec_theo().size();
+         }
+      }
+      LOG(Gen, Info,
+          "AnalyticsServiceServer::AnalyticsServiceServer publishing theos n: "
+          "{}",
+          n);
+   }
 }
 
 void AnalyticsServiceServer::get_new_volsurface_id(
@@ -452,6 +487,31 @@ void AnalyticsServiceServer::archive_theos(
    fire_cb(resp, true);
 }
 
+void AnalyticsServiceServer::clean_and_archive_theos(
+    const bts::api::analytics::ArchivedTheoList &theo_list,
+    const api::Response<::bts::api::ResultMessage> &resp) {
+   LOG(Gen, Info,
+       "AnalyticsServiceServer::clean_and_archive_theos theo_list count: {}",
+       theo_list.sec_theo_size());
+   if (theo_list.sec_theo().empty()) {
+      std::string err_msg = "Empty archived theo list, call failed";
+      LOG(Gen, Error, "AnalyticsServiceServer::clean_and_archive_theos {}",
+          err_msg);
+      fire_cb(resp, false, &err_msg);
+      return;
+   }
+   // for (const auto &t : theo_list.sec_theo()) {
+   //    me->archived_theos_publisher.add_or_update(t);
+   // }
+   me->theos_store->remove_all(me->archived_theos_ns);
+   const std::string &ns = me->archived_theos_ns;
+   const std::string key = "dont care"; //std::to_string(theo_list.save_time());
+   std::string value = utils::pb::to_string(theo_list);
+   me->theos_store->insert(ns, key, value);
+   _publish_archived_theos(theo_list);
+   fire_cb(resp, true);
+}
+
 void AnalyticsServiceServer::get_archived_theos(
     const bts::api::analytics::ArchivedTheoRetrieval &ret,
     const bts::api::Response<bts::api::analytics::ArchivedTheoList> &resp) {

src/server/sparkd/analytics/AnalyticsServiceServer.h

@@ -1,5 +1,6 @@
 #pragma once
 
+#include "lib/api/services/AnalyticsService.pb.h"
 #include "lib/api/services/AnalyticsService.server.pb.h"
 
 namespace bts {
@@ -17,6 +18,10 @@ class AnalyticsServiceServer
    struct PImpl;
    std::unique_ptr<PImpl> me;
 
+ private:
+   void _publish_archived_theos(
+       const api::analytics::ArchivedTheoList &total_state);
+
  public:
    AnalyticsServiceServer(bts::api::ServiceContext &context);
 
@@ -62,6 +67,10 @@ class AnalyticsServiceServer
        const bts::api::analytics::ArchivedTheoList &theos,
        const api::Response<::bts::api::ResultMessage> &reply) override;
 
+   void clean_and_archive_theos(
+       const bts::api::analytics::ArchivedTheoList &theos,
+       const api::Response<::bts::api::ResultMessage> &reply) override;
+
    void get_archived_theos(
        const bts::api::analytics::ArchivedTheoRetrieval &theo_ret,
        const bts::api::Response<bts::api::analytics::ArchivedTheoList> &reply)

src/server/sparkd/cmds/Run.cpp

@@ -85,7 +85,7 @@ static OrderServiceServer create_order_service_server(
 static void execute() {
    using namespace bts;
    LOG(Gen, Warn,
-       "------------------------------------ Sparkd v2.2.5.4 launched "
+       "------------------------------------ Sparkd v2.3 launched "
        "--------------------------------");
 
    // Initializes curl for the application, and eventual cleanup in destructor

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

@@ -30,9 +30,11 @@
 #include <cassert>
 #include <chrono>
 #include <cstddef>
+#include <cstdint>
 #include <databento/constants.hpp>
 #include <databento/historical.hpp>
 #include <databento/live.hpp>
+#include <filesystem>
 #include <optional>
 #include <ratio>
 #include <string>
@@ -77,6 +79,11 @@ void update_file(
     const uint time_to_run,
     std::string file_path) {
 
+   // Delete the existing file if it exists
+   LOG(md, Info, "Removing old file");
+   std::filesystem::remove(file_path);
+   LOG(md, Info, "Starting securities sub");
+
    auto client =
        databento::LiveBuilder{}
            .SetKey(key)
@@ -172,7 +179,10 @@ void DatabentoDatahandler::run(
                          .debug_sec();
    auto sleep_duration_updated = sleep_duration;
    std::vector<std::vector<std::tuple<uint64_t, std::string>>> all_times;
-   std::unordered_map<uint32_t, bts::domain::SecurityID> sec_id_map;
+   std::unordered_map<uint32_t, bts::domain::SecurityID>
+       exchnangeid_to_systemid;
+   std::unordered_map<std::string, bts::domain::SecurityID> symbol_to_systemid;
+
    if (!use_file) {
       sleep_duration_updated = std::chrono::duration<
           long double>::zero(); // Set to zero for live data
@@ -199,8 +209,32 @@ void DatabentoDatahandler::run(
       }
 
       auto duration = months * 32 * 24 * 60 * 60;
-      return expiration_time.seconds() <
-             bts::core::TimeStamp::now().seconds() + duration;
+      return expiration_time.seconds() >
+                 (bts::core::TimeStamp::now().seconds() - 60 * 60 * 24) &&
+             expiration_time.seconds() <
+                 bts::core::TimeStamp::now().seconds() + duration;
+   };
+   auto update_maps = [&](const databento::Record &record) {
+      const auto &msg = record.Get<databento::InstrumentDefMsg>();
+      LOG(md, Info, "got key for {} ", std::string(msg.RawSymbol()));
+      const auto exp_nanos = msg.expiration;
+      std::string exp_string = databento::ToIso8601(exp_nanos);
+      const auto expiration_time =
+          bts::core::TimeStamp::from_iso_string(exp_string);
+      if (!sec_filter(
+              expiration_time, msg.Underlying(), venue,
+              msg.strike_price / 1000000000)) {
+         return databento::KeepGoing::Continue;
+      }
+
+      if (!symbol_to_systemid.contains(msg.RawSymbol())) {
+         return databento::KeepGoing::Continue;
+      }
+      auto systemid = symbol_to_systemid[msg.RawSymbol()];
+      LOG(md, Info, "Updating map for {} to {} ", std::string(msg.RawSymbol()),
+          msg.hd.instrument_id);
+      exchnangeid_to_systemid.insert({msg.hd.instrument_id, systemid});
+      return databento::KeepGoing::Continue;
    };
    auto process_security = [&](const databento::Record &record) -> bool {
       const auto &msg = record.Get<databento::InstrumentDefMsg>();
@@ -242,6 +276,17 @@ void DatabentoDatahandler::run(
          sec.security.emplace<bts::domain::OptionSecurity>(
              expiration_time, bts::domain::Price(msg.strike_price, -9),
              call_put);
+         if (sec.venue() == api::domain::VENUE_CBOE) {
+            if (sec.security_common.symbol == "SPX" ||
+                sec.security_common.symbol == "SPXW") {
+               sec.security_common.min_quantity_increment = 1;
+               sec.security_common.min_quantity = 1;
+               sec.security_common.price_ticks.push_back(3.0);
+               sec.security_common.price_ticks.push_back(0.05);
+               sec.security_common.price_ticks.push_back(0.1);
+               sec.security_common.cash_per_point = 100;
+            }
+         }
          break;
       }
       default:
@@ -249,27 +294,28 @@ void DatabentoDatahandler::run(
              false, "Only futures and options supported on databento");
       }
       *sec.security_common.id = sec_id_store->get_or_add(
-          venue, sec_type, std::to_string(msg.hd.instrument_id));
+          venue, sec_type,sec.security_common.desc );
       listener->on_securities({&sec});
-      sec_id_map.insert({msg.hd.instrument_id, sec.security_common.id});
+      // exchnangeid_to_systemid.insert({msg.hd.instrument_id, sec.security_common.id});
+      symbol_to_systemid.insert({msg.RawSymbol(), sec.security_common.id});
       return true;
    };
-   auto debug_log = [&](const databento::Mbp1Msg &msg) {
+   auto debug_log = [&](const databento::Cmbp1Msg &msg) {
       if (DEBUG_SEC_ID != msg.hd.instrument_id)
          return;
       const double denom = 1000000000;
       // auto delta = std::to_string(std::chrono::duration<double, std::micro>(msg.ts_in_delta).count());
-      auto seq_num = msg.sequence;
+      // auto seq_num = msg.sequence;
       auto ap = msg.levels[0].ask_px / denom;
       auto bp = msg.levels[0].bid_px / denom;
       auto as = msg.levels[0].ask_sz;
       auto bs = msg.levels[0].bid_sz;
-      LOG(md, Sys, "ID: {}, seq_num: {}, B_P: {}, B_S: {}, A_P: {}, A_S: {}",
-          msg.hd.instrument_id, seq_num, bp, bs, ap, as);
+      LOG(md, Sys, "ID: {},  B_P: {}, B_S: {}, A_P: {}, A_S: {}",
+          msg.hd.instrument_id, bp, bs, ap, as);
    };
 
    auto process_bbo = [&](const databento::Record &record) {
-      const auto &msg = record.Get<databento::Mbp1Msg>();
+      const auto &msg = record.Get<databento::Cmbp1Msg>();
       debug_log(msg);
 
       // std::vector<std::tuple<uint64_t,std::string>>& times = all_times.emplace_back();
@@ -285,9 +331,10 @@ void DatabentoDatahandler::run(
       // prev = current;
       // current = bts::datetime::unix_nano_now();
       // times.push_back({current-prev,"MISC to check log time"});
-      if (!sec_id_map.contains(msg.hd.instrument_id))
+      if (!exchnangeid_to_systemid.contains(msg.hd.instrument_id))
          return databento::KeepGoing::Continue;
-      bts::domain::SecurityID sec_id = sec_id_map.at(msg.hd.instrument_id);
+      bts::domain::SecurityID sec_id =
+          exchnangeid_to_systemid.at(msg.hd.instrument_id);
 
       // prev = current;
       // current = bts::datetime::unix_nano_now();
@@ -297,7 +344,34 @@ void DatabentoDatahandler::run(
       // prev = current;
       // current = bts::datetime::unix_nano_now();
       // times.push_back({current-prev,"Sleep done"});
+      if (msg.action == 'T') {
+         domain::TradeRecord trade;
+         trade.trade_price = bts::domain::Price(msg.price, -9);
+         trade.trade_quantity = bts::domain::Quantity(msg.size);
+         trade.sec_id = sec_id;
+         switch (msg.side) {
+         case databento::Side::Ask:
+            trade.agg_side = domain::AggressorSide::Sell;
+            break;
+         case databento::Side::Bid:
+            trade.agg_side = domain::AggressorSide::Buy;
+            break;
+         case databento::Side::None:
+            trade.agg_side = domain::AggressorSide::None;
+            break;
+         }
+         trade.time_stamp = bts::core::TimeStamp::from_iso_string(
+             databento::ToIso8601(msg.ts_recv));
+         std::vector<const domain::TradeRecord *> trade_ptrs;
+         trade_ptrs.push_back(&trade);
+         listener->on_trades(trade_ptrs);
 
+         return databento::KeepGoing::Continue;
+      }
+      if (msg.action != 'A') {
+         LOG(md, Error, "Got with action neither A or T action:{}", msg.action);
+         return databento::KeepGoing::Continue;
+      }
       domain::Book book;
       book.set_security_id(sec_id);
       // prev = current;
@@ -333,40 +407,41 @@ void DatabentoDatahandler::run(
           Continue; //only added so the same functions work with the threaded builder, not really need in the blocking version
    };
 
-   auto process_trades = [&](const databento::Record &record) {
-      const auto &msg = record.Get<databento::TradeMsg>();
-      // The security type to is needed to get the id. Since the sec type isn't known,
-      // check for an id using both types (OPTION and FUTURE). This is safe to do since
-      // options and futures never have common names (instrument ids).
-      if (!sec_id_map.contains(msg.hd.instrument_id))
-         return databento::KeepGoing::Continue;
-      bts::domain::SecurityID sec_id = sec_id_map.at(msg.hd.instrument_id);
+   // auto process_trades = [&](const databento::Record &record) {
+   //    const auto &msg = record.Get<databento::TradeMsg>();
+   //    // The security type to is needed to get the id. Since the sec type isn't known,
+   //    // check for an id using both types (OPTION and FUTURE). This is safe to do since
+   //    // options and futures never have common names (instrument ids).
+   //    if (!exchnangeid_to_systemid.contains(msg.hd.instrument_id))
+   //       return databento::KeepGoing::Continue;
+   //    bts::domain::SecurityID sec_id =
+   //        exchnangeid_to_systemid.at(msg.hd.instrument_id);
 
-      // sleep_for(sleep_duration_updated);
-      domain::TradeRecord trade;
-      trade.trade_price = bts::domain::Price(msg.price, -9);
-      trade.trade_quantity = bts::domain::Quantity(msg.size);
-      trade.sec_id = sec_id;
-      switch (msg.side) {
-      case databento::Side::Ask:
-         trade.agg_side = domain::AggressorSide::Sell;
-         break;
-      case databento::Side::Bid:
-         trade.agg_side = domain::AggressorSide::Buy;
-         break;
-      case databento::Side::None:
-         trade.agg_side = domain::AggressorSide::None;
-         break;
-      }
-      trade.time_stamp = bts::core::TimeStamp::from_iso_string(
-          databento::ToIso8601(msg.ts_recv));
-      std::vector<const domain::TradeRecord *> trade_ptrs;
-      trade_ptrs.push_back(&trade);
-      listener->on_trades(trade_ptrs);
+   //    // sleep_for(sleep_duration_updated);
+   //    domain::TradeRecord trade;
+   //    trade.trade_price = bts::domain::Price(msg.price, -9);
+   //    trade.trade_quantity = bts::domain::Quantity(msg.size);
+   //    trade.sec_id = sec_id;
+   //    switch (msg.side) {
+   //    case databento::Side::Ask:
+   //       trade.agg_side = domain::AggressorSide::Sell;
+   //       break;
+   //    case databento::Side::Bid:
+   //       trade.agg_side = domain::AggressorSide::Buy;
+   //       break;
+   //    case databento::Side::None:
+   //       trade.agg_side = domain::AggressorSide::None;
+   //       break;
+   //    }
+   //    trade.time_stamp = bts::core::TimeStamp::from_iso_string(
+   //        databento::ToIso8601(msg.ts_recv));
+   //    std::vector<const domain::TradeRecord *> trade_ptrs;
+   //    trade_ptrs.push_back(&trade);
+   //    listener->on_trades(trade_ptrs);
 
-      return databento::KeepGoing::
-          Continue; //only added so the same functions work with the threaded builder, not really need in the blocking version
-   };
+   //    return databento::KeepGoing::
+   //        Continue; //only added so the same functions work with the threaded builder, not really need in the blocking version
+   // };
    if (use_file) {
       auto client = databento::HistoricalBuilder{}.SetKey(key).Build();
       databento::DbnFileStore dbn_file_store(securities_path);
@@ -390,8 +465,9 @@ void DatabentoDatahandler::run(
       }
       //ocassionaly run security writer to update security dbn
       if (!boost::filesystem::exists(securities_path) || load_reference_data) {
-         update_file(key, dataset, symbols, 15 * 60, securities_path);
-      }
+         LOG(md, Info, "Reloading securities file");
+         update_file(key, dataset, symbols, 3 * 60, securities_path);
+      } 
 
       //regular flow
       //read from security dbn, filter contracts that we need and subscribe to those in batches of two thousand
@@ -434,36 +510,44 @@ void DatabentoDatahandler::run(
       // size_t size = all_sec.size();
       LOG(md, Info, "total {} secs loaded", total_count);
 
-
       bool first = true;
       auto client = databento::LiveBlocking{
           databento::ILogReceiver::Default(),           key, dataset, true,
           databento::VersionUpgradePolicy::UpgradeToV2, {}};
       std::string start = config::get<bts::api::md::MarketDataConfig>()
-          .cboe()
-          .at(0)
-          .databento_settings()
-          .bbo_start_time();
-      if(start != ""){
-         for (std::vector vec : all_sec) {
-            client.Subscribe(
-                vec, databento::Schema::Mbp1, databento::SType::RawSymbol, start);
-            client.Subscribe(
-                vec, databento::Schema::Trades, databento::SType::RawSymbol,start);
-         }   
-      }else{
-         for (std::vector vec : all_sec) {
-            client.Subscribe(
-                vec, databento::Schema::Mbp1, databento::SType::RawSymbol);
-            client.Subscribe(
-                vec, databento::Schema::Trades, databento::SType::RawSymbol);
-         } 
+                              .cboe()
+                              .at(0)
+                              .databento_settings()
+                              .bbo_start_time();
+      if (total_count == 0) {
+         LOG(md, Error, "NO SECURITIES FOR DATABENTO SUBSCRIPTION");
+         return;
       }
+      if (start != "") {
+         for (std::vector vec : all_sec) {
+            client.Subscribe(
+                vec, databento::Schema::Cmbp1, databento::SType::RawSymbol,
+                start);
+            // client.Subscribe(
+            //     vec, databento::Schema::Trades, databento::SType::RawSymbol,
+            //     start);
+         }
+      } else {
+         for (std::vector vec : all_sec) {
+            client.Subscribe(
+                vec, databento::Schema::Cmbp1, databento::SType::RawSymbol);
+            // client.Subscribe(
+            //     vec, databento::Schema::Trades, databento::SType::RawSymbol);
+         }
+      }
+      client.Subscribe(
+          symbols, databento::Schema::Definition, databento::SType::Parent,
+          databento::UnixNanos{});
       bool only_securities = config::get<bts::api::md::MarketDataConfig>()
-                         .cboe()
-                         .at(0)
-                         .databento_settings()
-                         .only_securities();
+                                 .cboe()
+                                 .at(0)
+                                 .databento_settings()
+                                 .only_securities();
       client.Start();
       while (true) {
          try {
@@ -476,21 +560,21 @@ void DatabentoDatahandler::run(
                LOG(md, Error, "Timed out waiting for data");
                break;
             }
-            if(only_securities)
+            if (only_securities)
                continue;
-            if (record->Holds<databento::Mbp1Msg>()) {
+            if (record->Holds<databento::Cmbp1Msg>()) {
                process_bbo(*record);
             }
             if (all_times.size() == 1001) {
                log_time_statistics(all_times);
                client.Stop();
             }
-            //  if (record->Holds<databento::InstrumentDefMsg>()) {
-            //      process_security(*record);
-            //  }
-            if (record->Holds<databento::TradeMsg>()) {
-               process_trades(*record);
+            if (record->Holds<databento::InstrumentDefMsg>()) {
+               update_maps(*record);
             }
+            // if (record->Holds<databento::TradeMsg>()) {
+            //    process_trades(*record);
+            // }
          } catch (const databento::TcpError &error) {
             std::string err = error.what();
             LOG(md, Error, "Caught TCP error: {} \nAttempting reconnect\n",
@@ -503,7 +587,11 @@ void DatabentoDatahandler::run(
                 "Caught DbnResponseError what: {} ,msg:{} \nShutting down\n",
                 msg, what);
             client.Stop();
-            // bts_assert(false);
+            bts_assert_msgf(
+                false,
+                "Caught DbnResponseError what: {} ,msg:{} \nShutting down\n",
+                msg, what);
+            bts_assert(false);
          } catch (const databento::Exception &excp) {
             std::string err = excp.Message();
             std::string err2 = excp.what();
@@ -512,13 +600,18 @@ void DatabentoDatahandler::run(
                 "\nShutting down\n",
                 err, err2);
             client.Stop();
-            // bts_assert(false);
+            bts_assert_msgf(
+                false,
+                "Caught unknown dadtabento::exception what: {}, msg: {} "
+                "\nShutting down\n",
+                err, err2);
          } catch (const std::exception &exc) {
             std::string err = exc.what();
             LOG(md, Error, "Caught unknown exception: {} \nShutting down\n",
                 err);
             client.Stop();
-            // bts_assert(false);
+            bts_assert_msgf(
+                false, "Caught unknown exception: {} \nShutting down\n", err);
          }
       }
    }