Merging

src/client/spark/AW2/LiveTheoPublisher.cs

@@ -133,18 +133,18 @@ namespace mbts.spark {
                   TheoPoint tp = t.third;
                   var call_sec_md = sc_data.get_option(exchange, sym, exp, strike, PutOrCall.Call);
                   {
-                     NativePriceSize call_bid, call_ask;
-                     ulong time = 0;
-                     call_sec_md.Value.md.get_bbo(ref time, out call_bid, out call_ask);
-                     double call_mid = (call_bid.price + call_ask.price) / 2;
+                     double bid_price, ask_price;
+                     float bid_sz, ask_sz;
+                     call_sec_md.Value.md.get_bbo(out bid_price, out bid_sz, out ask_price, out ask_sz);
+                     double call_mid = (bid_price + ask_price) / 2;
                      tp.call = call_mid;
                   }
                   {
                      var put_sec_md = sc_data.get_option(exchange, sym, exp, strike, PutOrCall.Put);
-                     NativePriceSize put_bid, put_ask;
-                     ulong time = 0;
-                     put_sec_md.Value.md.get_bbo(ref time, out put_bid, out put_ask);
-                     double put_mid = (put_bid.price + put_ask.price) / 2;
+                     double bid_price, ask_price;
+                     float bid_sz, ask_sz;
+                     put_sec_md.Value.md.get_bbo(out bid_price, out bid_sz, out ask_price, out ask_sz);
+                     double put_mid = (bid_price + ask_price) / 2;
                      tp.put = put_mid;
                   }
                }
@@ -164,6 +164,10 @@ namespace mbts.spark {
 
 
       private void _interp_and_publish() {
+         // int i = 0;
+         // if (i < Int32.MaxValue) {
+         //    return;
+         // }
          var timer = Stopwatch.StartNew();
          var res = new Dictionary<uint, GreeksTuple>();
          var driver_map = _bb.read_state(_drivers_tag);

src/client/spark/AW2/Misc/SystemContractData.cs

@@ -3,6 +3,8 @@ using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Linq;
 
+using Apache.Arrow;
+
 using bts;
 using bts.api;
 using bts.collections;
@@ -161,11 +163,11 @@ namespace mbts.spark {
          if (!sec_md_map.TryGetValue(id.Value, out sec_md)) {
             return null;
          }
-         NativePriceSize bid, ask;
-         ulong time = 0;
-         sec_md.md.get_bbo(ref time, out bid, out ask);
+         double bid_price, ask_price;
+         float bid_sz, ask_sz;
+         sec_md.md.get_bbo(out bid_price, out bid_sz, out ask_price, out ask_sz);
          return Tuple.Create(sec_md.security,
-                             new VolSurfaceMDUnit(bid, ask, sec_md.md.fixing_price));
+                             new VolSurfaceMDUnit(bid_price, bid_sz, ask_price, ask_sz, sec_md.md.fixing_price));
       }
 
       public Tuple<SecurityImmutable, VolSurfaceMDUnit> get_exp_md(SecurityExpiration sec_exp) {

src/client/spark/Daemons/LocalMDDaemon.cs

@@ -0,0 +1,243 @@
+using bts.api;
+
+using mbts.co2;
+
+using System;
+using System.Diagnostics;
+using System.IO;
+using System.IO.Pipes;
+using System.Text;
+using System.Threading;
+
+namespace bts {
+   public class LocalMDConsumerHandle : mbts.co2.IMDConsumerHandle {
+      private readonly SecurityImmutable _sec;
+      private readonly LocalMDDaemon.BBO _bbo;
+      // private static NativePriceSize bid_nps = new NativePriceSize(new IntPrice(123.45), Convert.ToSingle(100));
+      // private static NativePriceSize ask_nps = new NativePriceSize(new IntPrice(123.46), Convert.ToSingle(100));
+      public LocalMDConsumerHandle(SecurityImmutable sec, LocalMDDaemon.BBO bbo) {
+         _sec = sec;
+         _bbo = bbo;
+      }
+
+      public UInt32 key => _sec.Id;
+      // The exchange key can change at CBOE. This numeric_id will always be correct
+      public ulong numeric_id => Convert.ToUInt64(_sec.Cboe.ExchangeId);
+
+      public void set_bbo(double bid_price, UInt32 bid_qty, double ask_price, UInt32 ask_qty) {
+         _bbo.bid_price = bid_price;
+         _bbo.bid_qty = bid_qty;
+         _bbo.ask_price = ask_price;
+         _bbo.ask_qty = ask_qty;
+      }
+
+      public bool is_heart_beating => true;
+      public bool has_depth => true;
+
+      // This property exists for specific performance optimization purposes
+      // Most callers should just use get_bbo/get_depth
+      public ulong market_timestamp => 0;
+      public ulong misc_timestamp => 0;
+
+      public ulong total_volume => 0;
+
+      public double settlement_price => Double.NaN;
+      public DateTime settlement_date => DateTime.Now;
+      public double previous_settlement_price => Double.NaN;
+      public DateTime previous_settlement_date => DateTime.Now;
+
+      public double open_price => Double.NaN;
+      public DateTime open_date => DateTime.Now;
+
+      public double close_price => Double.NaN;
+      public DateTime close_date => DateTime.Now;
+      public double session_high => Double.NaN;
+      public double session_low => Double.NaN;
+      public double fixing_price => Double.NaN;
+
+
+      public bool get_bbo(out double bid_price, out float bid_sz, out double ask_price, out float ask_sz) {
+         bid_price = _bbo.bid_price;
+         bid_sz = _bbo.bid_qty;
+         ask_price = _bbo.ask_price;
+         ask_sz = _bbo.ask_qty;
+         return true;
+      }
+
+      public bool get_bbo(ref ulong time, out NativePriceSize bid, out NativePriceSize ask) {
+         bid = new NativePriceSize(new IntPrice(_bbo.bid_price), Convert.ToSingle(_bbo.bid_qty));
+         ask = new NativePriceSize(new IntPrice(_bbo.ask_price), Convert.ToSingle(_bbo.ask_qty));
+         // bid = bid_nps;
+         // ask = ask_nps;
+         return true;
+      }
+
+      public bool get_depth(
+         ref ulong time, ref NativePriceSize[] bids, ref NativePriceSize[] asks, uint depth) {
+         return true;
+      }
+
+      public bool get_last_trade(
+         ref ulong time, out NativePriceSize ps, out DateTime timestamp, out ulong total_volume) {
+         ps = new NativePriceSize();
+         timestamp = DateTime.MinValue;
+         total_volume = 0;
+         return false;
+      }
+
+      public bool get_last_trade(
+         ref ulong time,
+         out NativePriceSize ps,
+         out DateTime timestamp,
+         out ulong total_volume,
+         out bool from_spread
+      ) {
+         ps = new NativePriceSize();
+         timestamp = DateTime.MinValue;
+         total_volume = 0;
+         from_spread = false;
+         return false;
+      }
+
+      public class PriceLevel {
+         public double price;
+         public uint qty;
+
+         public PriceLevel(double price_arg, uint qty_arg) {
+            price = price_arg;
+            qty = qty_arg;
+         }
+
+         public static PriceLevel empty => new(double.NaN, 0);
+
+         public static IntPrice intprice_from_double(double d) =>
+            double.IsNaN(d) ? IntPrice.NaN : new IntPrice((long)d, 0);
+
+         public NativePriceSize get_native_ps() =>
+            new(intprice_from_double(price), qty);
+      }
+   }
+
+   public class LocalMDDaemon {
+
+      public class BBO {
+         public double bid_price = Double.NaN;
+         public UInt32 bid_qty = 0;
+         public double ask_price = Double.NaN;
+         public UInt32 ask_qty = 0;
+         public UInt32 exchange_id = 0;
+         public BBO(uint exchange_id) :
+         this(Double.NaN, 0, Double.NaN, 0, exchange_id) { }
+         public BBO(double bid_price_, uint bid_qty_, double ask_price_, uint ask_qty_, uint exchange_id_) {
+            bid_price = bid_qty_ == 0 ? Double.NaN : bid_price_;
+            bid_qty = bid_qty_;
+            ask_price = ask_qty_ == 0 ? Double.NaN : ask_price_;
+            ask_qty = ask_qty_;
+            exchange_id = exchange_id_;
+         }
+
+         public override string ToString() {
+            return $"bid_price: {bid_price} bid_qty: {bid_qty} ask_price: {ask_price} ask_qty: {ask_qty} exchange_id: {exchange_id}";
+         }
+      }
+      static Process producer_process = null;
+      static void start_or_restart_producer_process(string exePath) {
+         foreach (var process in Process.GetProcessesByName(Path.GetFileNameWithoutExtension(exePath))) {
+            try {
+               bts.Log.Gen.info($"Stopping existing producer process (PID: {process.Id})...");
+               process.Kill();
+               process.WaitForExit();
+            } catch (Exception ex) {
+               bts.Log.Gen.error($"Stopping existing producer process (PID: {process.Id})...ex: {ex.Message}");
+            }
+         }
+         if (producer_process != null && !producer_process.HasExited) {
+            bts.Log.Gen.info("Stopping existing producer process...");
+            producer_process.Kill();
+            producer_process.WaitForExit();
+            producer_process.Dispose();
+         }
+
+         bts.Log.Gen.info("Starting new producer process...");
+
+         producer_process = new Process();
+         producer_process.StartInfo.FileName = exePath;  // Path to the EXE
+         producer_process.StartInfo.UseShellExecute = false;
+         // producer_process.StartInfo.RedirectStandardOutput = true;
+         // producer_process.StartInfo.RedirectStandardError = true;
+         producer_process.Start();
+
+         bts.Log.Gen.info("Producer process started.");
+      }
+      public Action<LocalMDDaemon.BBO> on_bbo = (bbo) => { };
+
+      public LocalMDDaemon() {
+         Log.Gen.info("LocalMDDaemon.cons called, starting the reading thread now");
+         System.Threading.Thread pipe_read_thr = new Thread(_start_reading);
+         pipe_read_thr.IsBackground = true;
+         pipe_read_thr.Start();
+
+      }
+
+      private void _start_reading() {
+         // TODO, fix this
+         string exe_path = @"dist\producer_delayed\producer_delayed.exe";
+         //@"dist\producer\producer.exe";
+
+         // delete pipe if already exists
+         using NamedPipeClientStream old_pipe_client = new NamedPipeClientStream(".", "MyPipe", PipeDirection.In);
+         try {
+            old_pipe_client.Close();
+            old_pipe_client.Dispose();
+         } catch (Exception ex) {
+            Log.Gen.error($"LocalMDDaemon._start_reading ex: {ex}");
+         }
+         start_or_restart_producer_process(exe_path);
+
+         // Connect to the named pipe
+         using NamedPipeClientStream pipe_client = new NamedPipeClientStream(".", "MyPipe", PipeDirection.In);
+
+         System.Threading.Thread.Sleep(100);
+         pipe_client.Connect();
+
+         bts.Log.Gen.info("Connected to pipe. Waiting for messages...");
+
+         byte[] buffer = new byte[1024];
+
+         while (true) {
+            try {
+               int bytes_read = pipe_client.Read(buffer, 0, buffer.Length);
+
+               if (bytes_read == 0)// this means there was an error
+               {
+                  System.Threading.Thread.Sleep(2);
+                  continue;
+               }
+               Func<double, bool> qty_ok = (d) => {
+                  if (d < 0 || d > Convert.ToDouble(UInt32.MaxValue))
+                     return false;
+                  return true;
+               };
+               double double1 = BitConverter.ToDouble(buffer, 0);
+               double double2 = BitConverter.ToDouble(buffer, 8);
+               double double3 = BitConverter.ToDouble(buffer, 16);
+               double double4 = BitConverter.ToDouble(buffer, 24);
+               uint int5 = BitConverter.ToUInt32(buffer, 32);
+               //Log.Gen.info($"Read from pipe {double1} ,  {double2} , {double3} , {double4} , {int5}");
+               if (!qty_ok(double2) || !qty_ok(double4))
+                  continue;
+               LocalMDDaemon.BBO bbo = new BBO(double1, Convert.ToUInt32(double2),
+                                          double3, Convert.ToUInt32(double4), int5);
+               on_bbo(bbo);
+            } catch (IOException) {
+               bts.Log.Gen.error("Pipe disconnected. Attempting to reconnect...");
+               pipe_client.Dispose();
+               System.Threading.Thread.Sleep(1000);
+               pipe_client.Connect(); // Try reconnecting
+            } catch (Exception ex) {
+               bts.Log.Gen.error($"Exception: {ex.Message}");
+            }
+         }
+      }
+   }
+}

src/client/spark/Daemons/LocalMDDaemon2.cs

@@ -0,0 +1,104 @@
+using bts;
+
+using System;
+using System.Diagnostics;
+using System.IO;
+using System.IO.Pipes;
+using System.Text;
+
+class LocalMDDaemon2 {
+   static Process producerProcess = null;
+   static void StartOrRestartProducerProcess(string exePath) {
+      // foreach (var process in Process.GetProcessesByName(Path.GetFileNameWithoutExtension(exePath)))
+      // {
+      //     try
+      //     {
+      //         Console.WriteLine($"Stopping existing producer process (PID: {process.Id})...");
+      //         process.Kill();
+      //         process.WaitForExit();
+      //     }
+      //     catch (Exception ex)
+      //     {
+      //         Console.WriteLine($"Error stopping process: {ex.Message}");
+      //     }
+      // }
+      // if (producerProcess != null && !producerProcess.HasExited)
+      // {
+      //     Console.WriteLine("Stopping existing producer process...");
+      //     producerProcess.Kill();
+      //     producerProcess.WaitForExit();
+      //     producerProcess.Dispose();
+      // }
+
+      Console.WriteLine("Starting new producer process...");
+
+      producerProcess = new Process();
+      producerProcess.StartInfo.FileName = exePath;  // Path to the EXE
+      producerProcess.StartInfo.UseShellExecute = false;
+      // producerProcess.StartInfo.RedirectStandardOutput = true;
+      // producerProcess.StartInfo.RedirectStandardError = true;
+      producerProcess.Start();
+      Console.WriteLine("Producer process started.");
+   }
+
+   public static void start() {
+
+      // string exePath = @".\dist\producer\producer.exe"; //uncomment for live
+      // string exePath = @".\dist\producer_delayed\producer_delayed.exe"; //UNCOMMENT FOR DELAYED
+      string exePath = //@".\dist\producer_replay\producer_replay.exe"; //UNCOMMENT FOR replay
+      @"C:\Users\m_tah\Downloads\HelloWorld\HelloWorld\dist\producer_replay\producer_replay.exe";
+
+
+      // delete pipe if already exists
+      // using NamedPipeClientStream oldpipeClient = new NamedPipeClientStream(".", "MyPipe", PipeDirection.In);
+      // try{
+      //     oldpipeClient.Close();
+      //     oldpipeClient.Dispose();
+      // }catch{
+
+      // }
+      StartOrRestartProducerProcess(exePath);
+
+      // Connect to the named pipe
+      using NamedPipeClientStream pipeClient = new NamedPipeClientStream(".", "MyPipe", PipeDirection.In);
+
+      // Thread.Sleep(2000);
+      pipeClient.Connect();
+
+      Console.WriteLine("Connected to pipe. Waiting for messages...");
+
+      byte[] buffer = new byte[1024];
+      var start = true;
+      while (true) {
+         try {
+            int bytesRead = pipeClient.Read(buffer, 0, buffer.Length);
+
+            if (bytesRead == 0)// this means there was an error
+            {
+               Console.WriteLine("0 bytes recieved ending...");
+               if (start) {
+                  start = false;
+                  continue;
+               }
+               break;
+            }
+
+            double double1 = BitConverter.ToDouble(buffer, 0);
+            double double2 = BitConverter.ToDouble(buffer, 8);
+            double double3 = BitConverter.ToDouble(buffer, 16);
+            double double4 = BitConverter.ToDouble(buffer, 24);
+            uint int5 = BitConverter.ToUInt32(buffer, 32);
+            Log.Gen.info($"Received: {double1}, {double2}, {double3}, {double4}, {int5}");
+
+         } catch (IOException) {
+            Console.WriteLine("Pipe disconnected. Attempting to reconnect...");
+            pipeClient.Dispose();
+            System.Threading.Thread.Sleep(1000);
+            pipeClient.Connect(); // Try reconnecting
+         }
+      }
+
+      // Console.WriteLine("Pipe closed. Exiting...");
+      // producerProcess?.WaitForExit();
+   }
+}

src/client/spark/Daemons/MDDaemon.cs

@@ -136,6 +136,7 @@ namespace bts {
                var (priceDisplay, qtyDisplay) = GetDisplaySettings(smd.security);
                exch_secs[smd.id] = new(smd.security, smd.md, priceDisplay, qtyDisplay);
             }
+            Log.Gen.info($"MDDaemon.initialize posting sec, n: {exch_secs.Count}");
             _bus.post(SparkTags.ServerSecuritiesTag, exch_secs);
          }
 

src/client/spark/Daemons/MergedMDDaemon.cs

@@ -1,10 +1,7 @@
 using BCG;
 
 using bts.api;
-using bts.api.synthetic;
-using bts.collections;
-using bts.md;
-using bts.utils;
+
 
 using mbts;
 using mbts.co2;
@@ -14,78 +11,103 @@ using mbts.ui;
 using Spark.Core.Disposables;
 
 using System;
+using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Collections.ObjectModel;
 using System.Composition;
 using System.Linq;
 using System.Reactive.Linq;
 using System.Threading.Tasks;
+using System.Threading;
 
 using SecurityKey = System.UInt32;
 
-namespace bts
-{
+namespace bts {
+
    [Export(typeof(IShellDaemon))]
-   internal class MergedMDDaemon : IShellDaemon, IDisposable
-   {
+   internal class MergedMDDaemon : IShellDaemon, IDisposable {
       private readonly IBus _bus;
+      private LocalMDDaemon _local_md_daemon;
       IDisposable _dispose;
-      Dictionary<uint, SecurityMD> _server_sec_map = new Dictionary<uint, SecurityMD>();
+      ConcurrentDictionary<uint, SecurityMD> _server_sec_map = new ConcurrentDictionary<uint, SecurityMD>();
       Dictionary<uint, SecurityMD> _fake_sec_map = new Dictionary<uint, SecurityMD>();
+      private ConcurrentDictionary<uint, SecurityKey> _exchange_id_sec_id_map = new ConcurrentDictionary<uint, SecurityKey>();
       private readonly IList<IDisposable> _subs = new List<IDisposable>();
+      private SyncQueue<LocalMDDaemon.BBO> _waiting_bbo = new SyncQueue<LocalMDDaemon.BBO>();
+      private readonly Venue _local_md_exchange = Venue.Cboe;
+      private bool _localmd_daemon_inited = false;
+
+      private void _init_local_md_daemon() {
+         if (_localmd_daemon_inited)
+            return;
+         //LocalMDDaemon2.start();
+         _localmd_daemon_inited = true;
+         _local_md_daemon = new LocalMDDaemon();
+         _local_md_daemon.on_bbo += (bbo_pack) => {
+            //bts.Log.Gen.info($"MergedMDDaemon on_bbo, local bbo callback, bbo: {bbo_pack}");
+            _waiting_bbo.enqueue(bbo_pack);
+         };
+         System.Threading.Thread dequeue_bbo_thr = new Thread(_dequeue_bbo_thread_func);
+         dequeue_bbo_thr.IsBackground = true;
+         dequeue_bbo_thr.Priority = ThreadPriority.Normal;
+         dequeue_bbo_thr.Start();
+      }
 
       [ImportingConstructor]
-      public MergedMDDaemon(IBus bus)
-      {
-         //TempTester.test_analytics();
+      public MergedMDDaemon(IBus bus) {
          bts.Log.Gen.info("MergedMDDaemon started");
          _bus = bus;
+
          _subs.Add(_bus.subscribe(SparkTags.ServerSecuritiesTag).OnDispatcher().Subscribe(
             d => _on_server_sec_map(d)));
          _subs.Add(_bus.subscribe(SparkTags.FakeSecuritiesTag).OnDispatcher().Subscribe(
             d => _on_fake_sec_map(d)));
-         // This is done periodically to remove expired contracts
-         // (disabling it right now because we are not sure if deleting securities 
-         // works with the various UI windows)
-         // _subs.Add(_bus.subscribe(SparkTags.SparkTimer).BufferLatest(TimeSpan.FromMinutes(1))
-         //             .Subscribe(t => _merge_and_post()));
-         //_test_md_stream();
-      }
-      // BCG.MDStream _md_stream = null;
-      // private void _test_md_stream()
-      // {
-      //    _bus.subscribe(SparkTags.SparkTimer).Subscribe(t =>
-      //       {
-      //          if (_md_stream == null)
-      //          {
-      //             _md_stream = new BCG.MDStream();
-      //             _md_stream.onChange += _on_md;
-      //          }
-      //          else
-      //          {
-      //             BCG.ViewStreamDatabase.instance.doWork();
-      //          }
-      //       });
-      // }
-
-      private void _on_md(object obj)
-      {
-         BCG.DataPacket dp = obj as DataPacket;
-         bts.Log.Gen.info($"MDStream callback dp: {dp}");
       }
 
-      private void _log_options(IReadOnlyDictionary<uint, SecurityMD> sec_map)
-      {
+      private void _dequeue_bbo_thread_func() {
+         while (true) {
+            System.Threading.Thread.Sleep(TimeSpan.FromMilliseconds(0.1));
+            _process_local_bbo_updates();
+         }
+      }
+
+      private void _process_local_bbo_updates() {
+         if (_waiting_bbo.getCount() == 0)
+            return;
+         var bbo_list = _waiting_bbo.dequeue_all();
+         // if (bbo_list.Count > -1)
+         //    return;
+         if (bbo_list.Count >= 10) {
+            Log.Gen.warn($"MergedMDDaemon._process_local_bbo_updates dequeued item count: {bbo_list.Count}");
+         }
+         List<LocalMDDaemon.BBO> unprocessed_bbo_list = new List<LocalMDDaemon.BBO>();
+         foreach (var bbo in bbo_list) {
+            bool processed = false;
+            //bts.Log.Gen.info($"MergedMDDaemon._process_local_bbo_updates read from queue bbo: {bbo}");
+            UInt32 exchange_id = bbo.exchange_id;
+            SecurityKey sec_key;
+            if (_exchange_id_sec_id_map.TryGetValue(exchange_id, out sec_key)) {
+               SecurityMD sec_md;
+               if (_server_sec_map.TryGetValue(sec_key, out sec_md)) {
+                  sec_md.md.set_bbo(bbo.bid_price, bbo.bid_qty, bbo.ask_price, bbo.ask_qty);
+                  processed = true;
+               }
+            }
+            if (!processed) {
+               // TODO: put it back in the queue
+            }
+         }
+      }
+
+      private void _log_options(IReadOnlyDictionary<uint, SecurityMD> sec_map) {
          string rv = "MergedMDDaemon Options:\n";
-         foreach (var kv in sec_map)
-         {
+         foreach (var kv in sec_map) {
             var sec = kv.Value.security;
             var md = kv.Value.md;
             ulong time = 0;
             NativePriceSize bid, ask;
             var has_md = md.get_bbo(ref time, out bid, out ask);
-            if (sec.HasOption)
-            {
+            if (sec.HasOption) {
                rv += $"id: {kv.Key} desc: {sec.security_desc()} sym: {sec.Symbol} " +
                      $"exp: {sec.security_expiration()} " +
                      $"k: {sec.Option.Strike} cp: {sec.Option.CallPut} " +
@@ -96,31 +118,40 @@ namespace bts
          Log.Gen.info(rv);
       }
 
-      private void _on_server_sec_map(IReadOnlyDictionary<uint, SecurityMD> sec_map)
-      {
+      private void _on_server_sec_map(IReadOnlyDictionary<uint, SecurityMD> sec_map) {
          //_log_options(sec_map);
          bts.Log.Gen.info(String.Format("MergedMDDaemon._on_server_sec_map sz: {0}", sec_map.Count()));
-         _server_sec_map = new Dictionary<uint, SecurityMD>(sec_map);
+         foreach (var kv in sec_map) {
+            SecurityMD md = kv.Value;
+            uint sec_id = kv.Key;
+            SecurityMD sec_md = kv.Value;
+            if (kv.Value.security.Exchange == _local_md_exchange) {
+               UInt32 exchange_sec_id = Convert.ToUInt32(kv.Value.security.Cboe.ExchangeId);
+               _exchange_id_sec_id_map[exchange_sec_id] = sec_id;
+               _server_sec_map[sec_id] = new SecurityMD(sec_md.security, new LocalMDConsumerHandle(sec_md.security, new LocalMDDaemon.BBO(exchange_sec_id)));
+            } else {
+               _server_sec_map[sec_id] = sec_md;
+            }
+         }
          _merge_and_post();
+         _init_local_md_daemon();
       }
 
-      private void _on_fake_sec_map(IReadOnlyDictionary<uint, SecurityMD> sec_map)
-      {
+      private void _on_fake_sec_map(IReadOnlyDictionary<uint, SecurityMD> sec_map) {
          bts.Log.Gen.info(String.Format("MergedMDDaemon._on_fake_sec_map sz: {0}", sec_map.Count()));
          _fake_sec_map = new Dictionary<uint, SecurityMD>(sec_map);
          _merge_and_post();
       }
 
-      private void _merge_and_post()
-      {
-         Func<SecurityMD, bool> is_active = (sec_md) =>
-         {
+      private void _merge_and_post() {
+         Func<SecurityMD, bool> is_active = (sec_md) => {
             var sec = sec_md.security;
             return SecurityExpirationHelper.is_unexpired(sec);
          };
+         // server secs
          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)
-         {
+         // add fake secs to the map
+         foreach (var kv in _fake_sec_map) {
             to_pub[kv.Key] = kv.Value;
          }
          bts.Log.Gen.info(String.Format("MergedMDDaemon._merge_and_post sz: {0}", to_pub.Count()));
@@ -132,7 +163,9 @@ namespace bts
          _bus.post(SparkTags.TC2Tags.ExchangeSecurityMapTag, exs_map);
       }
 
-      public void Dispose() =>
+      public void Dispose() {
          Disposable.dispose(ref _dispose);
+         Log.Gen.info("LocalMDDaemon.DIspose called collecting resources now");
+      }
    }
 }

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

@@ -16,7 +16,7 @@ namespace mbts.spark {
       private readonly ThreadedBlackboard _bb;
       HashSet<BCG.NMSheetSymbol> _sym_map;
       private Action _cb = () => { };
-      private readonly TimeSpan fire_period = TimeSpan.FromMilliseconds(300);
+      private readonly TimeSpan fire_period = TimeSpan.FromMilliseconds(333);
 
       public NMSheetBBODataSource(ThreadedBlackboard bb, List<BCG.NMSheetSymbol> syms, Action cb) {
          _bb = bb;

src/client/spark_api/MarketData/FakeMDConsumerHandle.cs

@@ -12,6 +12,10 @@ namespace mbts.co2 {
 
       readonly ulong _numeric_key;
 
+      public void set_bbo(double bid_price, UInt32 bid_qty, double ask_price, UInt32 ask_qty) {
+
+      }
+
       public FakeMDConsumerHandle(
          security_key key,
          ulong numeric_key,
@@ -52,6 +56,17 @@ namespace mbts.co2 {
       public double session_low => 500;
       public double fixing_price => 20;
 
+      public bool get_bbo(out double bid_price, out float bid_sz, out double ask_price, out float ask_sz) {
+         NativePriceSize bid;
+         NativePriceSize ask;
+         ulong time = 0;
+         bool rv = get_bbo(ref time, out bid, out ask);
+         bid_price = bid.price;
+         bid_sz = bid.size;
+         ask_price = ask.price;
+         ask_sz = ask.size;
+         return rv;
+      }
 
       public bool get_bbo(ref ulong time, out NativePriceSize bid, out NativePriceSize ask) {
          bool rv = false;

src/client/spark_api/MarketData/IMDConsumerHandle.cs

@@ -7,6 +7,7 @@ namespace mbts.co2 {
       bool is_heart_beating { get; }
       bool has_depth { get; }
       security_key key { get; }
+      public void set_bbo(double bid_price, UInt32 bid_qty, double ask_price, UInt32 ask_qty);
 
       // The exchange key can change at CBOE. This numeric_id will always be correct
       ulong numeric_id { get; }
@@ -32,6 +33,7 @@ namespace mbts.co2 {
       double session_low { get; }
       double fixing_price { get; }
 
+      bool get_bbo(out double bid_price, out float bid_sz, out double ask_price, out float ask_sz);
       bool get_bbo(ref ulong time, out NativePriceSize bid, out NativePriceSize ask);
 
       bool get_depth(

src/client/spark_api/MarketData/IntPrice.cs

@@ -4,13 +4,38 @@ using System.Runtime.InteropServices;
 namespace mbts.co2 {
    [StructLayout(LayoutKind.Explicit)]
    public readonly struct IntPrice : IEquatable<IntPrice> {
-      public static readonly IntPrice NaN = new(0,127);
+      public static readonly IntPrice NaN = new(0, 127);
 
       [FieldOffset(0)]
       readonly long _mantissa;
       [FieldOffset(7)]
       readonly sbyte _exponent;
 
+      private static bool _is_invalid_double(double value) {
+         return (Double.IsNaN(value) || double.IsInfinity(value) ||
+            value >= long.MaxValue || value <= long.MinValue);
+      }
+
+      private static Tuple<long, sbyte> double_to_mant_exp(double v) {
+         if (_is_invalid_double(v)) {
+            return Tuple.Create<long, sbyte>(0, 127);
+         }
+         sbyte exp = -7;
+         double v_m = v * Math.Pow(10, Math.Abs(exp));
+         if (_is_invalid_double(v_m)) {
+            return Tuple.Create<long, sbyte>(0, 127);
+         }
+         long m = Convert.ToInt64(v_m);
+         return Tuple.Create<long, sbyte>(m, exp);
+      }
+
+      public IntPrice(double v) :
+            this(double_to_mant_exp(v)) {
+      }
+
+      public IntPrice(Tuple<long, sbyte> t)
+      : this(t.Item1, t.Item2) { }
+
       public IntPrice(long mantissa, sbyte exponent) {
          _mantissa = mantissa;
          _exponent = exponent;
@@ -38,10 +63,10 @@ namespace mbts.co2 {
       public static bool operator ==(IntPrice a, IntPrice b) => a.Equals(b);
       public static bool operator !=(IntPrice a, IntPrice b) => !a.Equals(b);
 
-      public static bool operator < (IntPrice a, IntPrice b) =>
+      public static bool operator <(IntPrice a, IntPrice b) =>
          a.exponent == b.exponent ? a.mantissa < b.mantissa : a.price() < b.price();
 
-      public static bool operator > (IntPrice a, IntPrice b) =>
+      public static bool operator >(IntPrice a, IntPrice b) =>
          a.exponent == b.exponent ? a.mantissa > b.mantissa : a.price() > b.price();
    }
 }

src/client/spark_api/MarketData/MDConsumerHandle.cs

@@ -11,6 +11,9 @@
    using security_key = System.UInt32;
 
    public unsafe class MDConsumerHandle : IMDConsumerHandle {
+      public void set_bbo(double bid_price, UInt32 bid_qty, double ask_price, UInt32 ask_qty) {
+         throw new NotImplementedException();
+      }
       readonly HotEntry* _hot;
       readonly HotDepthEntry* _hot_depth;
       readonly ColdEntry* _cold;
@@ -74,6 +77,19 @@
       public double session_low => _cold->session_low;
       public double fixing_price => _cold->fixing_price;
 
+      public bool get_bbo(out double bid_price, out float bid_sz, out double ask_price, out float ask_sz) {
+         NativePriceSize bid;
+         NativePriceSize ask;
+         ulong time = 0;
+         bool rv = get_bbo(ref time, out bid, out ask);
+         bid_price = bid.price;
+         bid_sz = bid.size;
+         ask_price = ask.price;
+         ask_sz = ask.size;
+         return rv;
+      }
+
+
       public bool get_bbo(ref ulong time, out NativePriceSize bid, out NativePriceSize ask) {
          var updated = market_timestamp;
          if (updated <= time || updated == TimeStamp.invalid_value) {

src/client/spark_api/MarketData/SecurityMD.cs

@@ -13,9 +13,9 @@ namespace mbts {
    using System.Linq;
    using System.Threading;
 
-   public readonly struct SecurityMD : IEquatable<SecurityMD> {
+   public /*readonly*/ struct SecurityMD : IEquatable<SecurityMD> {
       public static readonly Func<SecurityMD, uint> key_fn = smd => smd.security.Id;
-
+      private IMDConsumerHandle _md;
       public SecurityMD(
          SecurityImmutable security,
          IMDConsumerHandle md,
@@ -23,7 +23,7 @@ namespace mbts {
          TickDisplaySettingsImmutable? quantity_display = null,
          IEnumerable<TickDisplayRuleImmutable>? price_display_rules = null) {
          this.security = security;
-         this.md = md;
+         _md = md;
          this.price_display = price_display ?? Security.Defaults.price_display_settingsi;
          this.quantity_display = quantity_display ??
             Security.Defaults.quantity_display_settingsi;
@@ -38,7 +38,14 @@ namespace mbts {
 
       public SecurityImmutable security { get; }
       public uint id => security.Id;
-      public IMDConsumerHandle md { get; }
+      public IMDConsumerHandle md {
+         get {
+            return _md;
+         }
+         set {
+            _md = value;
+         }
+      }
       public TickDisplaySettingsImmutable price_display { get; }
       public TickDisplaySettingsImmutable quantity_display { get; }
       public ImmutableList<TickDisplayRuleImmutable> price_display_rules { get; }

src/lib/api/md/DatabentoMarketDataConfig.proto

@@ -15,4 +15,5 @@ message DatabentoMarketDataConfig {
   string dataset = 70;
   uint32 filter_securities_upto_n_months= 80;
   bool load_reference_data = 90;
+  uint32 debug_sec = 100;
 }

src/lib/logging/Sink.cpp

@@ -9,6 +9,7 @@
 #include <boost/date_time/posix_time/posix_time.hpp>
 #include <boost/filesystem.hpp>
 #include <boost/filesystem/path.hpp>
+#include <cstdint>
 #include <fmt/ostream.h>
 #include <fstream>
 #include <iostream>
@@ -94,7 +95,7 @@ std::ifstream::pos_type filesize(std::string const &filename) {
 Sink file_sink(LoggingConfig const &cfg) {
    static std::ofstream stream;
    static uint64_t bytes_written = 0;
-   const auto max_sz = 120 * 1024 * 1024;
+   const uintmax_t max_sz =  1024 * 1024 * 1024 ;
 
    auto f = formatter(cfg);
    auto path = log_path();

src/lib/md/GeneralMarketDataListener.cpp

@@ -36,6 +36,7 @@
 #include "lib/utils/bts_assert.h"
 
 #include <bits/stdint-uintn.h>
+#include <cstddef>
 #include <google/protobuf/timestamp.pb.h>
 #include <google/protobuf/util/time_util.h>
 #include <memory>
@@ -732,11 +733,13 @@ static bool get_qty_info(
    }
    return true;
 }
-
+bool is_number(const std::string &s) {
+   return !s.empty() && std::all_of(s.begin(), s.end(), ::isdigit);
+ }
 void GeneralMarketDataListener::_copy_security_common(
     const bts::domain::Security &from, uint64_t key, FixedSecurity &to) const {
    to.key = key;
-   to.numeric_id = key;
+   to.numeric_id =is_number(from.security_common.exchange_sec_id)?std::stoul(from.security_common.exchange_sec_id):key;
    auto symbol = from.security_common.symbol;
    to.symbol(symbol);
    to.security_desc(from.security_common.desc);

src/lib/persistence/IDGenerator.cpp

@@ -73,7 +73,7 @@ std::optional<uint64_t> IDGenerator::get_id(
       return std::optional<uint64_t>(rowid);
    } else {
       if (add_if_new) {
-         auto all = me->kv_store->retrieve_all();
+         // auto all = me->kv_store->retrieve_all();
          auto new_id = me->kv_store->insert(ns, key, "");
          LOG(Gen, Diag, "IDGenerator::get_id ns: {} key: id: {}", ns, key,
              new_id);

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

@@ -1,5 +1,6 @@
 #include "DatabentoDataHandler.h"
 #include "databento/datetime.hpp"
+#include "databento/dbn_encoder.hpp"
 #include "databento/enums.hpp"
 #include "databento/exceptions.hpp"
 #include "databento/live_blocking.hpp"
@@ -7,8 +8,8 @@
 #include "databento/record.hpp"
 #include "databento/timeseries.hpp"
 #include "lib/api/domain/Venue.pb.h"
+#include "lib/api/md/MarketDataConfig.pb.h"
 #include "lib/config/config.h"
-#include "lib/config/protos/system_config.pb.h"
 #include "lib/core/TimeStamp.h"
 #include "lib/domain/AggressorSide.h"
 #include "lib/domain/Book.h"
@@ -22,14 +23,21 @@
 #include "lib/utils/TimeStamp.h"
 #include "lib/utils/bts_assert.h"
 #include "lib/utils/bts_chrono.h"
+#include "lib/utils/datetime.h"
 #include "lib/utils/helper_macros.h"
+#include <boost/filesystem/operations.hpp>
 #include <cassert>
 #include <chrono>
+#include <cstddef>
 #include <databento/constants.hpp>
 #include <databento/historical.hpp>
 #include <databento/live.hpp>
 #include <optional>
+#include <ratio>
 #include <string>
+#include <string_view>
+#include <tuple>
+#include <unordered_map>
 #include <vector>
 
 namespace bts::md::DatabentoHandler {
@@ -61,7 +69,87 @@ static SecurityType parse_sec_type(std::string_view str) {
    }
    return SEC_TYPE(UNSPECIFIED);
 }
+void update_file(
+    const std::string &key,
+    const std::string &dataset,
+    const std::vector<std::string> &parent_symbols,
+    const uint time_to_run,
+    std::string file_path) {
 
+   auto client =
+       databento::LiveBuilder{}
+           .SetKey(key)
+           .SetDataset(dataset)
+           .SetUpgradePolicy(databento::VersionUpgradePolicy::UpgradeToV2)
+           .BuildBlocking();
+
+   client.Subscribe(
+       parent_symbols, databento::Schema::Definition, databento::SType::Parent,
+       databento::UnixNanos{});
+   // We start streaming
+   databento::Metadata metadata = client.Start();
+   {
+      // We open a file for writing and encode the metadata
+      databento::OutFileStream out_file{file_path};
+      databento::DbnEncoder encoder{metadata, &out_file};
+      // Wait for at least one record and 5 seconds before closing the connection
+      const std::chrono::seconds timeout{time_to_run};
+      const auto start = std::chrono::steady_clock::now();
+      while (auto *record = client.NextRecord(timeout)) {
+         encoder.EncodeRecord(*record);
+         if (std::chrono::steady_clock::now() - start >= timeout) {
+            break;
+         }
+      }
+   }
+   return;
+}
+void log_time_statistics(
+    const std::vector<std::vector<std::tuple<uint64_t, std::string>>>
+        &all_times) {
+   std::unordered_map<std::string, std::vector<uint64_t>> subpart_times;
+
+   // Collect all timing values per subpart
+   for (const auto &entry : all_times) {
+      for (const auto &[time, label] : entry) {
+         subpart_times[label].push_back(time);
+      }
+   }
+
+   // Function to compute statistics
+   auto compute_statistics = [](std::vector<uint64_t> &values)
+       -> std::tuple<double, double, double, double, double, double, double> {
+      if (values.empty())
+         return std::make_tuple(0.0, 0.0, 0, 0, 0, 0, 0);
+
+      std::sort(values.begin(), values.end());
+
+      double sum = std::accumulate(values.begin(), values.end(), 0.0);
+      double avg = sum / values.size();
+
+      double sq_sum =
+          std::inner_product(values.begin(), values.end(), values.begin(), 0.0);
+      double stddev = std::sqrt(sq_sum / values.size() - avg * avg);
+
+      auto percentile = [&](double p) -> uint64_t {
+         size_t idx = static_cast<size_t>(p * (values.size() - 1));
+         return values[idx];
+      };
+
+      return std::make_tuple(
+          avg, stddev, percentile(0.00), percentile(0.25), percentile(0.50),
+          percentile(0.75), percentile(1.00));
+   };
+
+   // Compute and log statistics
+   for (auto &[label, times] : subpart_times) {
+      auto [avg, stddev, p0, p25, p50, p75, p100] = compute_statistics(times);
+      LOG(md, Sys,
+          "Subpart: {}, Avg: {:.2f} ns, Std: {:.2f} ns, P0: {} ns, P25: {} ns, "
+          "P50: {} ns, P75: {} ns, P100: {} ns",
+          label, avg, stddev, p0, p25, p50, p75, p100);
+   }
+}
 void DatabentoDatahandler::run(
     bool use_file,
     std::shared_ptr<IMarketDataListener> listener,
@@ -74,59 +162,71 @@ void DatabentoDatahandler::run(
     const uint filter_securities_upto_n_months,
     const bool load_reference_data,
     const std::chrono::duration<long double> sleep_duration) {
-
+   uint DEBUG_SEC_ID = 0;
+   if (venue == api::domain::VENUE_CBOE)
+      DEBUG_SEC_ID = config::get<bts::api::md::MarketDataConfig>()
+                         .cboe()
+                         .at(0)
+                         .databento_settings()
+                         .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;
    if (!use_file) {
       sleep_duration_updated = std::chrono::duration<
           long double>::zero(); // Set to zero for live data
    }
-   auto process_security = [&](const databento::Record &record) {
-      const auto &msg = record.Get<databento::InstrumentDefMsg>();
-      std::string id = remove_spaces(msg.RawSymbol());
-      std::string symbol;
-
+   auto sec_filter = [&](const core::TimeStamp expiration_time,
+                         const std::string &symbol,
+                         const api::domain::Venue &venue, uint strike) {
+      uint months = filter_securities_upto_n_months;
       switch (venue) {
       case api::domain::Venue::VENUE_CBOE: {
-         symbol = remove_spaces(
-             std::string(msg.underlying.data(), msg.underlying.size()));
+         if (strike < 3800 || strike > 8000) {
+            return false;
+         }
+         if (symbol == "SPX")
+            months = 6;
+         else if (symbol == "SPXW")
+            months = 1;
          break;
       }
       case api::domain::Venue::VENUE_GLBX:
-         if (id.length() > 5)
-            return databento::KeepGoing::Continue;
-         symbol = id.substr(0, 2);
-         //  sec.security_common.cpp = msg.unit_of_measure_qty;
-
-         break;
+         return true;
       default:
          bts_assert_msg(false, "Exchange not supported on databento");
-         return databento::KeepGoing::Stop;
+      }
+
+      auto duration = months * 32 * 24 * 60 * 60;
+      return expiration_time.seconds() <
+             bts::core::TimeStamp::now().seconds() + duration;
+   };
+   auto process_security = [&](const databento::Record &record) -> bool {
+      const auto &msg = record.Get<databento::InstrumentDefMsg>();
+      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 false;
       }
       bts::domain::Security sec;
       sec.security_common = domain::SecurityCommon();
       sec.security_common.venue = venue;
-      sec.security_common.exchange_sec_id = msg.hd.instrument_id;
+      sec.security_common.exchange_sec_id = std::to_string(msg.hd.instrument_id);
       sec.security_common.desc = msg.RawSymbol();
-      sec.security_common.symbol = msg.Asset();
+      sec.security_common.symbol = msg.Underlying();
       sec.security_common.asset = msg.Asset();
       sec.security_common.group = msg.Group();
       sec.security_common.min_price_increment = 0.01;
       sec.security_common.min_quantity = 0.01;
       sec.security_common.min_quantity_increment = 0.01;
       sec.security_common.currency = {msg.currency.data(), msg.currency.size()};
-      // sec.security_common.cpp = msg.hd.
 
       const SecurityType sec_type =
           parse_sec_type({msg.security_type.data(), msg.security_type.size()});
-      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);
-      auto duration = filter_securities_upto_n_months*30*24*60*60;
-      if( expiration_time.seconds() >= bts::core::TimeStamp::now().seconds() + duration){
-         return databento::KeepGoing::Continue;
-      }
 
       switch (sec_type) {
       case SEC_TYPE(FUTURE):
@@ -134,11 +234,10 @@ void DatabentoDatahandler::run(
          break;
       case SEC_TYPE(OPTION): {
          auto call_put = CallPut::PUT;
-         if (msg.instrument_class == 'C') {
+         if (msg.instrument_class == 'C')
             call_put = CallPut::CALL;
-         } else {
+         else
             bts_assert(msg.instrument_class == 'P');
-         }
          sec.security.emplace<bts::domain::OptionSecurity>(
              expiration_time, bts::domain::Price(msg.strike_price, -9),
              call_put);
@@ -147,36 +246,62 @@ void DatabentoDatahandler::run(
       default:
          bts_assert_msg(
              false, "Only futures and options supported on databento");
-         return databento::KeepGoing::Stop;
       }
-
       *sec.security_common.id = sec_id_store->get_or_add(
           venue, sec_type, std::to_string(msg.hd.instrument_id));
       listener->on_securities({&sec});
-      return databento::KeepGoing::Continue;
+      sec_id_map.insert({msg.hd.instrument_id, sec.security_common.id});
+      return true;
+   };
+   auto debug_log = [&](const databento::Mbp1Msg &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 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);
    };
 
    auto process_bbo = [&](const databento::Record &record) {
       const auto &msg = record.Get<databento::Mbp1Msg>();
+      debug_log(msg);
+
+      // std::vector<std::tuple<uint64_t,std::string>>& times = all_times.emplace_back();
+      // times.reserve(8);
+      // return databento::KeepGoing::Continue; //only added so the same functions work with the threaded builder, not really need in the blocking version
       // 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).
-      const auto name = std::to_string(msg.hd.instrument_id);
-      auto sec_id_opt =
-          sec_id_store->get_id(venue, SEC_TYPE(OPTION), name, false);
-      if (!sec_id_opt.has_value()) {
-         sec_id_opt =
-             sec_id_store->get_id(venue, SEC_TYPE(FUTURE), name, false);
-         if (!sec_id_opt.has_value()) {
-            return databento::KeepGoing::Continue;
-         }
-      }
+      // auto current = bts::datetime::unix_nano_now();
+      // auto prev = current;
+      // times.push_back({current-prev,"Start Process bbo"});
+
+      // 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))
+         return databento::KeepGoing::Continue;
+      bts::domain::SecurityID sec_id = sec_id_map.at(msg.hd.instrument_id);
+
+      // prev = current;
+      // current = bts::datetime::unix_nano_now();
+      // times.push_back({current-prev,"Sec_ID done"});
+
+      // sleep_for(sleep_duration_updated);
+      // prev = current;
+      // current = bts::datetime::unix_nano_now();
+      // times.push_back({current-prev,"Sleep done"});
 
-      sleep_for(sleep_duration_updated);
-      bts::domain::SecurityID sec_id =
-          bts::domain::SecurityID(sec_id_opt.value());
       domain::Book book;
       book.set_security_id(sec_id);
+      // prev = current;
+      // current = bts::datetime::unix_nano_now();
+      // times.push_back({current-prev,"Empty Book Made"});
 
       auto &bids = book.bids();
       auto &asks = book.asks();
@@ -186,130 +311,287 @@ void DatabentoDatahandler::run(
       asks.begin()->price = bts::domain::Price(msg.levels[0].ask_px, -9);
       asks.begin()->total_quantity =
           bts::domain::Quantity(msg.levels[0].ask_sz);
+      //     prev = current;
+      //     current = bts::datetime::unix_nano_now();
+      // times.push_back({current-prev,"Complete book done"});
 
       auto recv_time = bts::core::TimeStamp::from_iso_string(
           databento::ToIso8601(msg.ts_recv));
+      // prev = current;
+      // current = bts::datetime::unix_nano_now();
+      // times.push_back({current-prev,"Time converted"});
+
       listener->on_book(
           book,
           utils::TimeStamp{datetime::ptime_to_timestamp(recv_time.to_ptime())});
-      return databento::KeepGoing::Continue;
+      // prev = current;
+      // current = bts::datetime::unix_nano_now();
+      // times.push_back({current-prev,"On book done"});
+
+      return databento::KeepGoing::
+          Continue; //only added so the same functions work with the threaded builder, not really need in the blocking version
    };
+   auto process_tbbo = [&](const databento::Record &record) {
+      const auto &msg = record.Get<databento::TbboMsg>();
+      debug_log(msg);
+
+      // std::vector<std::tuple<uint64_t,std::string>>& times = all_times.emplace_back();
+      // times.reserve(8);
+      // return databento::KeepGoing::Continue; //only added so the same functions work with the threaded builder, not really need in the blocking version
+      // 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).
+      // auto current = bts::datetime::unix_nano_now();
+      // auto prev = current;
+      // times.push_back({current-prev,"Start Process bbo"});
+
+      // 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))
+         return databento::KeepGoing::Continue;
+      bts::domain::SecurityID sec_id = sec_id_map.at(msg.hd.instrument_id);
+
+      // prev = current;
+      // current = bts::datetime::unix_nano_now();
+      // times.push_back({current-prev,"Sec_ID done"});
+
+      // sleep_for(sleep_duration_updated);
+      // prev = current;
+      // current = bts::datetime::unix_nano_now();
+      // times.push_back({current-prev,"Sleep done"});
+
+      domain::Book book;
+      book.set_security_id(sec_id);
+      // prev = current;
+      // current = bts::datetime::unix_nano_now();
+      // times.push_back({current-prev,"Empty Book Made"});
+
+      auto &bids = book.bids();
+      auto &asks = book.asks();
+      bids.begin()->price = bts::domain::Price(msg.levels[0].bid_px, -9);
+      bids.begin()->total_quantity =
+          bts::domain::Quantity(msg.levels[0].bid_sz);
+      asks.begin()->price = bts::domain::Price(msg.levels[0].ask_px, -9);
+      asks.begin()->total_quantity =
+          bts::domain::Quantity(msg.levels[0].ask_sz);
+      //     prev = current;
+      //     current = bts::datetime::unix_nano_now();
+      // times.push_back({current-prev,"Complete book done"});
+
+      auto recv_time = bts::core::TimeStamp::from_iso_string(
+          databento::ToIso8601(msg.ts_recv));
+      // prev = current;
+      // current = bts::datetime::unix_nano_now();
+      // times.push_back({current-prev,"Time converted"});
+
+      listener->on_book(
+          book,
+          utils::TimeStamp{datetime::ptime_to_timestamp(recv_time.to_ptime())});
+
+
+          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);
+      
+      // prev = current;
+      // current = bts::datetime::unix_nano_now();
+      // times.push_back({current-prev,"On book done"});
+
+      return databento::KeepGoing::
+          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).
-      const auto name = std::to_string(msg.hd.instrument_id);
-      auto sec_id_opt =
-          sec_id_store->get_id(venue, SEC_TYPE(OPTION), name, false);
-      if (!sec_id_opt.has_value()) {
-         sec_id_opt =
-             sec_id_store->get_id(venue, SEC_TYPE(FUTURE), name, false);
-         if (!sec_id_opt.has_value()) {
-            return databento::KeepGoing::Continue;
-         }
-      }
+      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);
 
-      sleep_for(sleep_duration_updated);
-      bts::domain::SecurityID sec_id =
-          bts::domain::SecurityID(sec_id_opt.value());
+      // 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.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;
+
+      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);
-      dbn_file_store.Replay(process_security);
+      // dbn_file_store.Replay(process_security); // fix this
       databento::DbnFileStore dbn_file_store2(books_path);
       dbn_file_store2.Replay(process_bbo);
-   }else {
-    bts_assert(dataset!="");
-    // auto client = databento::LiveBuilder{}
-    //               .SetKey(key)
-    //               .SetDataset(dataset)
-    //               .BuildThreaded();
-    auto client = databento::LiveBlocking{
-        databento::ILogReceiver::Default(),
-        key,
-        dataset,
-        true,
-        databento::VersionUpgradePolicy::UpgradeToV2,
-        {}
-    };
-    //hardcoding for now will update later
-    std::vector<std::string> symbols;
-    switch (venue) {
-        case api::domain::VENUE_CBOE: symbols= {"SPX.OPT","SPXW.OPT"}; break;
-        case api::domain::VENUE_GLBX: symbols= {"ES.FUT"}; break;
-        default:bts_assert_msg(false, "ONLY GLBX AND CBOE SUPPORTED FOR NOW");
-    }
-    if(load_reference_data) 
-         client.Subscribe(symbols, databento::Schema::Definition, databento::SType::Parent,databento::UnixNanos{});
-    else client.Subscribe(symbols, databento::Schema::Definition, databento::SType::Parent);
-    client.Subscribe(symbols, databento::Schema::Mbp1, databento::SType::Parent);
-    client.Subscribe(symbols, databento::Schema::Trades, databento::SType::Parent);
-    client.Start();
-    while (true) {
-    try {
-    auto* record = client.NextRecord(std::chrono::seconds{500});
-    if (record == nullptr) {
-        LOG(md, Error, "Timed out waiting for data");
-        break;
-    }
-    if (record->Holds<databento::Mbp1Msg>()) {
-        process_bbo(*record);
-    }
-    if (record->Holds<databento::InstrumentDefMsg>()) {
-        process_security(*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",err);
-    client.Reconnect();
-  } catch(const databento::DbnResponseError& err){
-    std::string msg = err.Message();
-    std::string what = err.what();
-    LOG(md, Error, "Caught DbnResponseError what: {} ,msg:{} \nShutting down\n",msg, what);
-    client.Stop();
-    // bts_assert(false);
-  }
-  catch(const databento::Exception& excp){
-    std::string err = excp.Message();
-    std::string err2 = excp.what();
-    LOG(md, Error, "Caught unknown dadtabento::exception what: {}, msg: {} \nShutting down\n",err, err2);
-    client.Stop();
-    // bts_assert(false);
-  }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);
-  }
-    
-    }
+   } else {
+      bts_assert(dataset != "");
+
+      //hardcoding for now will update later
+      std::vector<std::string> symbols;
+      switch (venue) {
+      case api::domain::VENUE_CBOE:
+         symbols = {"SPX.OPT", "SPXW.OPT"};
+         break;
+      case api::domain::VENUE_GLBX:
+         symbols = {"ES.FUT"};
+         break;
+      default:
+         bts_assert_msg(false, "ONLY GLBX AND CBOE SUPPORTED FOR NOW");
+      }
+      //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);
+      }
+
+      //regular flow
+      //read from security dbn, filter contracts that we need and subscribe to those in batches of two thousand
+      LOG(md, Info, "starting loading symbols");
+      databento::DbnFileStore store{securities_path};
+      std::vector<std::vector<std::string>> all_sec;
+
+      all_sec.emplace_back();
+      all_sec.back().reserve(2000);
+      size_t total_count = 0;
+
+      while (auto *record = store.NextRecord()) {
+         if (auto *def = record->GetIf<databento::InstrumentDefMsg>()) {
+            std::string s = def->RawSymbol();
+            if (process_security(*record)) {
+               ++total_count;
+               // Check if the last vector is full
+               if (all_sec.back().size() == 2000) {
+                  all_sec.emplace_back();
+                  all_sec.back().reserve(2000);
+                  LOG(md, Info, "{} symbols added", total_count);
+               }
+               all_sec.back().push_back(std::move(s)); // Move to avoid copying
+            }
+         }
+      }
+      // std::ostringstream output;
+
+      // for (const auto& sec_group : all_sec) {
+      //    for (size_t i = 0; i < sec_group.size(); ++i) {
+      //       if (i > 0) output << ", ";
+      //       output << sec_group[i];
+      //    }
+      //    output << '\n';  // Newline after each group
+      // }
+
+      // // Store the final output in a stringls
+
+      // std::string final_output = output.str();
+      // 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, {}};
+      for (std::vector vec : all_sec) {
+         // client.Subscribe(
+         //     vec, databento::Schema::Tbbo, databento::SType::RawSymbol);
+         // client.Subscribe(
+         //       vec, databento::Schema::Mbp1, databento::SType::RawSymbol);
+         // client.Subscribe(
+            //  vec, databento::Schema::Trades, databento::SType::RawSymbol);
+      }
+      client.Subscribe(symbols, databento::Schema::Definition, databento::SType::Parent,databento::UnixNanos{});
+
+      client.Start();
+      while (true) {
+         try {
+            auto *record = client.NextRecord(std::chrono::seconds{500});
+            if (first) {
+               LOG(md, Info, "first record");
+               first = false;
+            }
+            if (record == nullptr) {
+               LOG(md, Error, "Timed out waiting for data");
+               break;
+            }
+            if (record->Holds<databento::Mbp1Msg>()) {
+               process_bbo(*record);
+            }
+            if (record->Holds<databento::TbboMsg>()) {
+               process_tbbo(*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);
+            }
+         } catch (const databento::TcpError &error) {
+            std::string err = error.what();
+            LOG(md, Error, "Caught TCP error: {} \nAttempting reconnect\n",
+                err);
+            client.Reconnect();
+         } catch (const databento::DbnResponseError &err) {
+            std::string msg = err.Message();
+            std::string what = err.what();
+            LOG(md, Error,
+                "Caught DbnResponseError what: {} ,msg:{} \nShutting down\n",
+                msg, what);
+            client.Stop();
+            // bts_assert(false);
+         } catch (const databento::Exception &excp) {
+            std::string err = excp.Message();
+            std::string err2 = excp.what();
+            LOG(md, Error,
+                "Caught unknown dadtabento::exception what: {}, msg: {} "
+                "\nShutting down\n",
+                err, err2);
+            client.Stop();
+            // bts_assert(false);
+         } 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);
+         }
+      }
    }
 }