MDConsumerHandle - don't convert LLDateTimes to System.DateTimes

src/client/spark/Modules/Market/MarketView/MarketViewRow.cs

@@ -5,11 +5,9 @@ using mbts.collections;
 using mbts.spark.utilities;
 using mbts.ui;
 using mbts.ui.controls;
-using mbts.ui.settings;
 using mbts.ui.spark;
 using mbts.ui.utility;
 using mbts.utils;
-using bts.utils;
 using System;
 using System.Collections.Generic;
 using System.Collections.ObjectModel;
@@ -18,7 +16,6 @@ using System.Linq;
 using System.Reactive.Linq;
 using System.Windows.Input;
 
-
 namespace mbts.spark {
    partial class MarketViewVM {
       public class Row : SecurityMDRow, IDisposable, IEquatable<Row>, DIYGrid.IRowVM, DIYGrid.IActionableRowVM {
@@ -78,16 +75,29 @@ namespace mbts.spark {
             ask_price_click = new DelegateCommand(() => _show_order_ticket(MarketViewView.Columns.Ask));
             ask_size_click = new DelegateCommand(() => _show_order_ticket(MarketViewView.Columns.AskSize));
             working_asks_click = new DelegateCommand(() => _show_order_book(MarketViewView.Columns.WorkingAsks));
+
+            _position.value_set += position_value_set;
+            _inside_bids.value_set += inside_bids_value_set;
+            _inside_asks.value_set += inside_asks_value_set;
+            _day_change.Changed += day_change_Changed;
             _animation.AnimationsChanged += on_animations_changed;
-            void on_animations_changed() => RaisePropertyChanged(nameof(animations));
+
             _dispose = Disposable.Create(
                _orders_model.on_security_root_orders(security.Id).OnDispatcher().Subscribe(_on_orders),
-               _position.OnDispatcher().Subscribe(v => _animation.animate_position(v.newval > v.oldval)),
-               _inside_bids.OnDispatcher().Subscribe(v => _animation.animate_working_bids_inside(v.newval)),
-               _inside_asks.OnDispatcher().Subscribe(v => _animation.animate_working_asks_inside(v.newval)),
-               _day_change.OnDispatcher().Subscribe(v => _animation.animate_day_change(v.newval > v.oldval)),
-               Disposable.Create(() => _animation.AnimationsChanged -= on_animations_changed)
+               Disposable.Create(() => {
+                  _position.value_set -= position_value_set;
+                  _inside_bids.value_set -= inside_bids_value_set;
+                  _inside_asks.value_set -= inside_asks_value_set;
+                  _day_change.Changed -= day_change_Changed;
+                  _animation.AnimationsChanged -= on_animations_changed;
+               })
             );
+
+            void position_value_set((int oldval, int newval) v) => _animation.animate_position(v.newval > v.oldval);
+            void inside_bids_value_set((bool oldval, bool newval) v) => _animation.animate_working_bids_inside(v.newval);
+            void inside_asks_value_set((bool oldval, bool newval) v) => _animation.animate_working_asks_inside(v.newval);
+            void day_change_Changed(double oldval, double newval) => _animation.animate_day_change(newval > oldval);
+            void on_animations_changed() => RaisePropertyChanged(nameof(animations));
          }
 
          public event Action invalidated;
@@ -129,27 +139,52 @@ namespace mbts.spark {
 
          public void raise_invalidated() => invalidated?.Invoke();
 
-         public void refresh() {
+         /// <summary>
+         /// Returns a list of IDisposable 'update jobs.' Each item represents a job that will be executed when disposed; all jobs should be executed on the dispatcher thread.
+         /// </summary>
+         // Note: not guarenteed to be on the dispatcher thread, so all view-notifying actions should be wrapped in a IDisposable.
+         // Having said that, perform as much work as possible locally.
+         public List<IDisposable> refresh() {
+            var update_jobs = new List<IDisposable>();
             bool dirty = false;
             if (md.md.get_bbo(ref _bbo_ref, out NativePriceSize b, out NativePriceSize a)) {
-               _bid_price.set(b.price);
-               _bid_size.set(b.size);
-               _ask_price.set(a.price);
-               _ask_size.set(a.size);
-               _inside_bids.on_next_distinct(_bid_price.Value >= _working_best_bid.value, ref dirty);
-               _inside_asks.on_next_distinct(_ask_price.Value <= _working_best_ask.value, ref dirty);
+               prop_job(_bid_price, b.price);
+               prop_job(_bid_size, b.size);
+               prop_job(_ask_price, a.price);
+               prop_job(_ask_size, a.size);
+               cell_prop_job(_inside_bids, b.price >= _working_best_bid.value);
+               cell_prop_job(_inside_asks, a.price >= _working_best_ask.value);
             }
-            if (md.md.get_last_trade(ref _last_ref, out NativePriceSize ps, out DateTime ts, out ulong v)) {
-               _last_price.set(ps.price);
-               _last_size.on_next_distinct(ps.size, ref dirty);
-               _volume.on_next_distinct(v, ref dirty);
+            double new_last_price = last_price;
+            if (md.md.get_last_trade(ref _last_ref, out NativePriceSize ps, out ulong v)) {
+               new_last_price = ps.price;
+               prop_job(_last_price, new_last_price);
+               cell_prop_job(_last_size, ps.size);
+               cell_prop_job(_volume, v);
+            }
+            double new_settlement = md.md.settlement_price;
+            prop_job(_settlement, new_settlement);
+            prop_job(_session_low, md.md.session_low);
+            prop_job(_session_high, md.md.session_high);
+            prop_job(_day_change, new_last_price - new_settlement);
+
+            if (dirty) {
+               update_jobs.Add(Disposable.Create(() => {
+                  invalidated?.Invoke();
+               }));
+            }
+            return update_jobs;
+
+            void prop_job<T>(Property<T> p, T newval) {
+               var uj = p.set_as_job(newval);
+               if (uj != null)
+                  update_jobs.Add(uj);
+            }
+            void cell_prop_job<T>(CellProperty<T> p, T newval) {
+               var uj = p.on_next_distinct_as_job(newval, ref dirty);
+               if (uj != null)
+                  update_jobs.Add(uj);
             }
-            _settlement.set(md.md.settlement_price);
-            _session_low.set(md.md.session_low);
-            _session_high.set(md.md.session_high);
-            _day_change.set(last_price - settlement);
-            if (dirty)
-               invalidated?.Invoke();
          }
 
          public bool Equals(Row other) =>

src/client/spark/Modules/Market/MarketView/MarketViewVM.cs

@@ -3,10 +3,13 @@ using mbts.spark.filters;
 using mbts.ui;
 using mbts.ui.utility;
 using System;
+using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Collections.ObjectModel;
+using System.Linq;
 using System.Reactive.Linq;
 using System.Reactive.Subjects;
+using System.Threading.Tasks;
 using System.Windows.Data;
 using System.Windows.Media;
 using static bts.shell.MarketViewWindowFactory;
@@ -14,7 +17,7 @@ using static mbts.ui.controls.DIYGrid;
 using static mbts.ui.spark.MarketViewView;
 
 namespace mbts.spark {
-   partial class MarketViewVM : SecurityMDGrid<MarketViewVM.Row, SecurityMD> {
+   partial class MarketViewVM : SecurityMDGrid<MarketViewVM.Row, SecurityMD>, IPrepareUpdateForDispatcher {
       readonly IBus _bus;
       readonly OrdersModel _orders_model;
 
@@ -112,8 +115,7 @@ namespace mbts.spark {
             }), DataGridFilterHelper.hook_filters(rows_view,
             security_filter, bid_price_filter, bid_size_filter, ask_price_filter, ask_size_filter, last_price_filter, last_size_filter, day_change_filter, settlement_filter,
             session_low_filter, session_high_filter, position_filter, working_bid_volume_filter, working_ask_volume_filter, volume_filter, security_key_filter),
-            securities_connectable.Connect(),
-            WorkLoop.Start("MarketView", _refresh)
+            securities_connectable.Connect()
          );
       }
 
@@ -159,6 +161,25 @@ namespace mbts.spark {
       }
       #endregion
 
+      public IDisposable prepare_update() {
+         ConcurrentBag<List<IDisposable>> to_update = new ConcurrentBag<List<IDisposable>>();
+         try {
+            Parallel.For(0, _row_vms.Count, i => {
+               var r = _row_vms[i];
+               var update_jobs = r.refresh();
+               if (update_jobs.Count > 0)
+                  to_update.Add(update_jobs);
+            });
+         } catch (IndexOutOfRangeException) {
+            // since rows (securities) will only be added/removed through rare user intervention,
+            // it is worth catching the occassional exception in exchange for lock-free code.
+            // if a row misses an update, or gets updated more than once, that's okay too;
+            // in 100ms it will tick again, so the data should never be more than a few 100 ms old, and rarely older than 100.
+         }
+         // return null, rather than an empty IDisposable; this signals to the UpdateMaster that it may skip the call to Dispatcher.InvokeAsync() as there is nothing to dispose.
+         return to_update.Count == 0 ? null : Disposable.Create(() => to_update.SelectMany(_ => _).DisposeAll());
+      }
+
       protected override Row row_factory(SecurityMD smd) => new Row(_bus, smd, _orders_model);
       protected override void on_row_removed(Row row) => row.Dispose();
       protected override IEnumerable<SortingCriteria> get_default_sorting() { yield return new SortingCriteria(nameof(Row.security), true); }
@@ -167,4 +188,4 @@ namespace mbts.spark {
          foreach (var r in _row_vms) r.refresh();
       }
    }
-}
+}

src/client/spark/Modules/Market/MarketView/MarketViewWindow.cs

@@ -14,16 +14,18 @@ using System.Windows;
 namespace mbts.spark {
    class MarketViewWindow : SparkWindow<MarketViewWindowFactory.StartupData>, ISharedWindow {
       const string _window_type = "MarketView", _thread_id = "MarketView", _window_type_display_name = "Market View";
+      readonly DispatcherUpdateMaster _dispatcher_update_master;
       readonly OrdersModel _orders_model;
       readonly AllowOnce _init_guard = new AllowOnce("MarketView");
       MarketViewVM _vm;
       MarketViewView _view;
       MarketViewWindowFactory.StartupData _state;
       IReadOnlyList<SparkCommand> _commands;
-      IDisposable _subscriptions;
+      IDisposable _resources;
 
-      public MarketViewWindow(MarketViewWindowFactory factory, IBus bus, OrdersModel orders_model, MarketViewWindowFactory.StartupData state)
+      public MarketViewWindow(MarketViewWindowFactory factory, DispatcherUpdateMaster dispatcher_update_master, IBus bus, OrdersModel orders_model, MarketViewWindowFactory.StartupData state)
          : base("Market View", bus, factory, SparkCommands.ShowMarketView.icon) {
+         _dispatcher_update_master = dispatcher_update_master;
          _orders_model = orders_model;
          _state = state;
          Icon.Freeze();
@@ -57,12 +59,14 @@ namespace mbts.spark {
             SparkCommands.ShowSyntheticCreate.complete(_vm.selected_securities_observable(), secs => secs.Count > 0 && secs.Count <= 24, secs => new SecurityCriteria(secs)),
             SparkCommands.change_securities_command(() => (update_securities_cmd, _state.security_criteria))
          };
-         _subscriptions = _bus.subscribe(update_securities_cmd.tag).OnDispatcher().Subscribe(c =>
-            _state_changed?.Invoke(_state = new MarketViewWindowFactory.StartupData(c, _vm.rows_view.GetSortingCriteria(), _vm.get_filter_states())));
+         _resources = Disposable.Create(
+            _bus.subscribe(update_securities_cmd.tag).OnDispatcher().Subscribe(
+               c => _state_changed?.Invoke(_state = new MarketViewWindowFactory.StartupData(c, _vm.rows_view.GetSortingCriteria(), _vm.get_filter_states()))),
+            _dispatcher_update_master.attach(_vm));
       }
 
       public override void Dispose() {
-         Disposable.dispose(ref _subscriptions);
+         Disposable.dispose(ref _resources);
          _vm.Dispose();
          Commands.OfType<IDisposable>().DisposeAll();
       }

src/client/spark/Modules/Market/MarketView/MarketViewWindowFactory.cs

@@ -12,6 +12,7 @@ namespace bts.shell {
    class MarketViewWindowFactory : ShellWindowFactory<MarketViewWindowFactory.StartupData> {
       readonly IBus _bus;
       readonly OrdersModel _orders_model;
+      readonly DispatcherUpdateMaster _dispatcher_update_master = new DispatcherUpdateMaster("MarketView DispatcherUpdateMaster");
 
       public MarketViewWindowFactory(MarketShellModule module, IBus bus, OrdersModel orders_model) : base(module, "MarketView") {
          _bus = bus;
@@ -25,7 +26,7 @@ namespace bts.shell {
       void _show(SecurityCriteria criteria) =>
          _bus.post(SparkTags.ShowWindow, _create(new StartupData(criteria, new List<SortingCriteria> { new SortingCriteria(nameof(MarketViewVM.Row.security), true) }, null)));
 
-      IShellWindow _create(StartupData state) => new MarketViewWindow(this, _bus, _orders_model, state);
+      IShellWindow _create(StartupData state) => new MarketViewWindow(this, _dispatcher_update_master, _bus, _orders_model, state);
 
       [ProtoContract]
       public class StartupData {

src/client/spark/Orders/OrderTypeVMs/TriggerOrderVM.cs

@@ -211,7 +211,7 @@ namespace mbts.spark {
                _security_md.Value.md.get_bbo(ref _bbo_ref, out co2.NativePriceSize bid, out co2.NativePriceSize ask);
                _best_bid = bid.price;
                _best_ask = ask.price;
-               _security_md.Value.md.get_last_trade(ref _last_trade_ref, out co2.NativePriceSize ps, out DateTime ts, out ulong v);
+               _security_md.Value.md.get_last_trade(ref _last_trade_ref, out co2.NativePriceSize ps, out ulong v);
             }
          }
 

src/client/spark/core/Utility/DispatcherUpdateMaster.cs

@@ -0,0 +1,99 @@
+using bts;
+using bts.utils;
+using mbts.collections;
+using System;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.Linq;
+using System.Reactive.Concurrency;
+using System.Reactive.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using System.Windows.Threading;
+
+namespace mbts.spark {
+   interface IPrepareUpdateForDispatcher {
+      IDisposable prepare_update();
+   }
+
+   class DispatcherUpdateMaster : IDisposable {
+      readonly string _name;
+      readonly DispatcherPriority _invocation_priority;
+      SetMap<Dispatcher, IPrepareUpdateForDispatcher> _vms;
+      int _state = States.ReadyForUpdate;
+      IDisposable _dispose;
+
+      public DispatcherUpdateMaster(string name)
+         : this(name, WorkLoop.default_interval, WorkLoop.default_priority) { }
+
+      public DispatcherUpdateMaster(string name, TimeSpan period, DispatcherPriority invocation_priority) {
+         _name = name;
+         _vms = new SetMap<Dispatcher, IPrepareUpdateForDispatcher>(EqualityComparer.Create<Dispatcher, int>(d => d.Thread.ManagedThreadId));
+         _dispose = ThreadPoolScheduler.Instance.SchedulePeriodic(period, _update_async);
+         _invocation_priority = invocation_priority;
+      }
+
+      public IDisposable attach(IPrepareUpdateForDispatcher vm, Dispatcher dispatcher) {
+         var clone = _vms.clone();
+         clone.add(dispatcher, vm);
+         _vms = clone;
+         return Disposable.Create(() => _detach(vm, dispatcher));
+      }
+
+      public void Dispose() {
+         Thread.VolatileWrite(ref _state, States.Disposed);
+         Disposable.dispose(ref _dispose);
+      }
+
+      void _detach(IPrepareUpdateForDispatcher preper, Dispatcher dispatcher) {
+         var clone = _vms.clone();
+         clone.remove(dispatcher, preper);
+         _vms = clone;
+      }
+
+      async void _update_async() {
+         if (Interlocked.CompareExchange(ref _state, States.UpdateInProgress, States.ReadyForUpdate) != States.ReadyForUpdate) {
+            return;
+         }
+         var temp = _vms;
+         try {
+            ConcurrentBag<(Dispatcher dispatcher, ConcurrentBag<IDisposable> update)> update_jobs =
+               new ConcurrentBag<(Dispatcher dispatcher, ConcurrentBag<IDisposable> update_jobs)>();
+            await Task.Run(() => {
+               Parallel.ForEach(temp, dispatcher_group => {
+                  ConcurrentBag<IDisposable> update = new ConcurrentBag<IDisposable>();
+                  Parallel.ForEach(dispatcher_group, vm => {
+                     // need to check for null, that indicates an empty update (rather than an actual empty enumerable)
+                     var ujs = vm.prepare_update();
+                     if (ujs != null)
+                        update.Add(ujs);
+                  });
+                  if (update.Count > 0)
+                     update_jobs.Add((dispatcher_group.Key, update));
+               });
+            });
+            await Task.WhenAll(update_jobs.Select(_update_dispatcher_group_async));
+         } catch (Exception ex) {
+            Log.Ui.error($"{_name} error during parallel update: {ex}");
+         } finally {
+            // signal that the update is complete, and we are ready for another
+            Interlocked.CompareExchange(ref _state, States.ReadyForUpdate, States.UpdateInProgress);
+         }
+      }
+
+      async Task _update_dispatcher_group_async((Dispatcher dispatcher, ConcurrentBag<IDisposable> update_jobs) t) =>
+         await t.dispatcher.InvokeAsync(() => t.update_jobs.DisposeAll(), _invocation_priority);
+
+      static class States {
+         public const int
+            ReadyForUpdate = 0,
+            UpdateInProgress = 1,
+            Disposed = 2;
+      }
+   }
+
+   static class DispatcherUpdateMasterExtensions {
+      public static IDisposable attach(this DispatcherUpdateMaster dum, IPrepareUpdateForDispatcher preparer) =>
+         dum.attach(preparer, Dispatcher.CurrentDispatcher);
+   }
+}

src/client/spark/core/Utility/ViewModelBase.cs

@@ -156,15 +156,13 @@ namespace mbts.ui.shell
          public string name => _propname;
 
          public void set(T value) {
-            if (!Equals(value, _value)) {
-               Changing?.Invoke(_value, value);
-               var old = _value;
-               _value = value;
-               Changed?.Invoke(old, value);
-               _owner.RaisePropertyChanged(_propname);
-            }
+            if (!Equals(value, _value))
+               _apply_new_value(value);
          }
 
+         public IDisposable set_as_job(T value) =>
+            Equals(value, _value) ? null : Disposable.Create(() => _apply_new_value(value));
+
          public IDisposable Subscribe(IObserver<(T oldval, T newval)> observer) {
             Changed += pch;
             return Disposable.Create(() => Changed -= pch);
@@ -182,6 +180,14 @@ namespace mbts.ui.shell
             get { return _value; }
             set { set(value); }
          }
+
+         void _apply_new_value(T new_value) {
+            Changing?.Invoke(_value, new_value);
+            var old = _value;
+            _value = new_value;
+            Changed?.Invoke(old, new_value);
+            _owner.RaisePropertyChanged(_propname);
+         }
       }
 
 		protected class DelegateProperty<T> : Property<T> {

src/client/spark/view/Controls/DIYGrid/CellProperty.cs

@@ -15,7 +15,7 @@ namespace mbts.spark {
          value = initial_value;
       }
 
-      event Action<(T oldval, T newval)> _value_set;
+      public event Action<(T oldval, T newval)> value_set;
 
       public T value { get; private set; }
       public string text => _text ?? (_text = _format(value));
@@ -27,14 +27,22 @@ namespace mbts.spark {
          }
       }
 
+      public IDisposable on_next_distinct_as_job(T value, ref bool dirty) {
+         if (!value.Equals(this.value)) {
+            dirty = true;
+            return Disposable.Create(() => OnNext(value));
+         }
+         return null;
+      }
+
       public void OnCompleted() { }
       public void OnError(Exception error) { }
       public void OnNext(T value) {
-         if (_value_set != null) {
+         if (value_set != null) {
             T prev = this.value;
             this.value = value;
             _text = null;
-            _value_set.Invoke((prev, value));
+            value_set.Invoke((prev, value));
          } else {
             this.value = value;
             _text = null;
@@ -42,7 +50,7 @@ namespace mbts.spark {
       }
 
       public IDisposable Subscribe(IObserver<(T oldval, T newval)> observer) =>
-         Observable.FromEvent<Action<(T oldval, T newval)>, (T oldval, T newval)>(cb => _value_set += cb, cb => _value_set -= cb)
+         Observable.FromEvent<Action<(T oldval, T newval)>, (T oldval, T newval)>(cb => value_set += cb, cb => value_set -= cb)
          .Subscribe(observer);
    }
 }

src/client/spark/view/Spark/Ladder/LadderGridHost.cs

@@ -386,7 +386,7 @@ namespace mbts.spark {
       void _tick() {
          if (_settings != null) {
             ulong unused = 0;
-            bool last_trade_changed = _vm.md.get_last_trade(ref unused, out _last_trade, out DateTime _, out ulong __);
+            bool last_trade_changed = _vm.md.get_last_trade(ref unused, out _last_trade, out ulong __);
             bool depth_changed = _vm.md.get_depth(ref _prices_updated, ref _bids, ref _asks);
             _mkt_break_lines.refresh();
             bool prices_changed =

src/client/spark/view/Spark/Ladder/LadderVM.cs

@@ -275,7 +275,7 @@ namespace mbts.spark {
             _open.set(md.open_price);
             _settlement.set(md.settlement_price);
             ulong unused = 0;
-            md.get_last_trade(ref unused, out NativePriceSize last, out DateTime time, out ulong volume);
+            md.get_last_trade(ref unused, out NativePriceSize last, out ulong volume);
             if (!double.IsNaN(last.price)) {
                _change.set(last.price - md.settlement_price);
             } else {

src/client/spark/view/Spark/PAndL/PnLCalculator.cs

@@ -113,7 +113,7 @@ namespace mbts.spark {
          if (!sec_md.md.get_bbo(ref bbo_time, out NativePriceSize bid, out NativePriceSize ask)) return false;
          // last is only used in waterfall and last calculations, so if neither of those types are selected
          // we should not exit early if last is somehow not avaialble
-         if (!sec_md.md.get_last_trade(ref last_trade_time, out NativePriceSize last, out DateTime unused1, out ulong unused2) && pnl_request.HasFlag(pnl_calc_type.waterfall | pnl_calc_type.last)) return false;
+         if (!sec_md.md.get_last_trade(ref last_trade_time, out NativePriceSize last, out ulong unused2) && pnl_request.HasFlag(pnl_calc_type.waterfall | pnl_calc_type.last)) return false;
 
          if (pnl_request.HasFlag(pnl_calc_type.waterfall)) {
             // last price is available and withing range of available bid and ask

src/client/spark/view/Spark/Utilities/ISecurityMDPump.cs

@@ -12,7 +12,7 @@ namespace mbts.spark.utilities
    internal interface ISecurityMDPump
    {
        IObservable<(double bid, uint bid_size, double ask, uint ask_size)> bbo { get; }
-       IObservable<(double last, uint last_size, DateTime time_stamp, ulong volume)> last_trade { get; }
+       IObservable<(double last, uint last_size, ulong volume)> last_trade { get; }
        IObservable<double> bid_price { get; }
        IObservable<uint> bid_size { get; }
        IObservable<double> ask_price { get; }
@@ -61,7 +61,6 @@ namespace mbts.spark.utilities
       private double _bid_price, _ask_price, _last_price;
       private uint _bid_size, _ask_size, _last_size;
       private ulong _volume;
-      private DateTime _last_time;
 
       private bool _refresh_bbo(MDConsumerHandle md) {
          if (md.get_bbo(ref _bbo_ref, out NativePriceSize b, out NativePriceSize a)) {
@@ -75,10 +74,9 @@ namespace mbts.spark.utilities
       }
 
       private bool _refresh_last(MDConsumerHandle md) {
-         if (md.get_last_trade(ref _last_ref, out NativePriceSize ps, out DateTime ts, out ulong v)) {
+         if (md.get_last_trade(ref _last_ref, out NativePriceSize ps, out ulong v)) {
             _last_price = ps.price;
             _last_size = ps.size;
-            _last_time = ts;
             _volume = v;
             return true;
          }
@@ -86,17 +84,16 @@ namespace mbts.spark.utilities
       }
 
       private (double, uint, double, uint) _select_bbo() { return (_bid_price, _bid_size, _ask_price, _ask_size); }
-      private (double, uint, DateTime, ulong) _select_last() { return (_last_price, _last_size, _last_time, _volume); }
+      private (double, uint, ulong) _select_last() { return (_last_price, _last_size, _volume); }
 
       public IObservable<(double bid, uint bid_size, double ask, uint ask_size)> bbo => _bbo_pump.Select(_ => _select_bbo()).StartWith(_select_bbo());
-      public IObservable<(double last, uint last_size, DateTime time_stamp, ulong volume)> last_trade => _last_pump.Select(_ => _select_last()).StartWith(_select_last());
+      public IObservable<(double last, uint last_size, ulong volume)> last_trade => _last_pump.Select(_ => _select_last()).StartWith(_select_last());
       public IObservable<double> bid_price => _bbo_pump.Select(_ => _bid_price).DistinctUntilChanged().StartWith(_bid_price);
       public IObservable<uint> bid_size => _bbo_pump.Select(_ => _bid_size).DistinctUntilChanged().StartWith(_bid_size);
       public IObservable<double> ask_price => _bbo_pump.Select(_ => _ask_price).DistinctUntilChanged().StartWith(_ask_price);
       public IObservable<uint> ask_size => _bbo_pump.Select(_ => _ask_size).DistinctUntilChanged().StartWith(_ask_size);
       public IObservable<double> last_price => _bbo_pump.Select(_ => _last_price).DistinctUntilChanged().StartWith(_last_price);
       public IObservable<uint> last_size => _bbo_pump.Select(_ => _last_size).DistinctUntilChanged().StartWith(_last_size);
-      public IObservable<DateTime> last_time => _last_pump.Select(_ => _last_time).DistinctUntilChanged().StartWith(_last_time);
       public IObservable<ulong> volume => _last_pump.Select(_ => _volume).DistinctUntilChanged().StartWith(_volume);
       public IObservable<double> session_low => _root_pump.Select(md => md.session_low).DistinctUntilChanged();
       public IObservable<double> session_high => _root_pump.Select(md => md.session_high).DistinctUntilChanged();

src/client/spark/view/Spark/Utilities/WorkLoop.cs

@@ -7,8 +7,8 @@ using System.Windows.Threading;
 namespace mbts.spark {
    static class WorkLoop {
       public const double default_excessive_cycle_ratio = 1.2;
-      static readonly DispatcherPriority default_priority;
-      static readonly TimeSpan default_interval;
+      public static readonly TimeSpan default_interval;
+      public static readonly DispatcherPriority default_priority;
 
       static WorkLoop() {
          var timer_params = config.Configuration.get<bts.config.LadderKnobs>();

src/client/spark_core/Co2/MDConsumerHandle.cs

@@ -72,7 +72,12 @@ namespace mbts.co2
          return false;
       }
 
-      public bool get_last_trade(ref ulong time, out NativePriceSize ps, out DateTime timestamp, out ulong total_volume) {
+      public bool get_last_trade(ref ulong time, out NativePriceSize ps, out ulong total_volume) {
+         total_volume = _cold->total_volume;
+         return (_cold->last_trade.load_if_new(ref time, out ps));
+      }
+
+      public bool get_last_trade(ref ulong time, out NativePriceSize ps, out utils.LLDateTime timestamp, out ulong total_volume) {
          timestamp = _cold->last_trade_time;
          total_volume = _cold->total_volume;
          return (_cold->last_trade.load_if_new(ref time, out ps));
@@ -88,15 +93,15 @@ namespace mbts.co2
       public ulong total_volume { get {  return _cold->total_volume; } }
 
       public double settlement_price { get { return _cold->settlement_price;} }
-      public DateTime settlement_date { get { return _cold->settlement_date;} }
+      public utils.LLDate settlement_date { get { return _cold->settlement_date;} }
       public double previous_settlement_price { get { return _cold->previous_settlement_price;} }
-      public DateTime previous_settlement_date { get { return _cold->previous_settlement_date;} }
+      public utils.LLDate previous_settlement_date { get { return _cold->previous_settlement_date;} }
 
       public double open_price { get { return _cold->open_price;} }
-      public DateTime open_date { get {  return _cold->open_date; } }
+      public utils.LLDate open_date { get {  return _cold->open_date; } }
 
       public double close_price { get { return _cold->close_price;} }
-      public DateTime close_date { get {  return _cold->close_date; } }
+      public utils.LLDate close_date { get {  return _cold->close_date; } }
       public double session_high { get { return _cold->session_high;} }
       public double session_low { get {  return _cold->session_low; } }
       public double open_interest { get {  return _cold->open_interest; } }

src/client/spark_test/Co2/Co2Tests.cs

@@ -210,19 +210,24 @@ namespace mbts.co2.tests
                   Assert.IsTrue(ask.size == 202);
 
                   time = 0;
-                  DateTime trade_date;
-                  ulong total_volume;
-                  NativePriceSize last_trade;
-                  Assert.IsTrue(handle.get_last_trade(ref time, out last_trade, out trade_date, out total_volume));
+                  Assert.IsTrue(handle.get_last_trade(ref time, out NativePriceSize last_trade, out utils.LLDateTime ll_trade_date, out ulong total_volume));
+                  // trade date
+                  DateTime trade_date = ll_trade_date;
                   Assert.IsTrue(last_trade.price == 5.5 && last_trade.size == 505);
                   Assert.IsTrue(trade_date.Year == 2015 && trade_date.Month == 9 && trade_date.Day == 22);
                   Assert.IsTrue(total_volume == 4200);
-                  Assert.IsTrue(handle.settlement_date.Year == 2000 && handle.settlement_date.Day == 9 && handle.settlement_date.Month == 8);
+                  // settlement date
+                  DateTime settlement_dt = handle.settlement_date;
+                  Assert.IsTrue(settlement_dt.Year == 2000 && settlement_dt.Day == 9 && settlement_dt.Month == 8);
                   Assert.IsTrue(handle.settlement_price == 6.6);
                   Assert.IsTrue(handle.open_price == 7.7);
-                  Assert.IsTrue(handle.open_date.Year == 2001 && handle.open_date.Month == 1 && handle.open_date.Day == 2);
+                  // open date
+                  DateTime open_dt = handle.open_date;
+                  Assert.IsTrue(open_dt.Year == 2001 && open_dt.Month == 1 && open_dt.Day == 2);
                   Assert.IsTrue(handle.close_price == 8.8);
-                  Assert.IsTrue(handle.close_date.Year == 2002 && handle.close_date.Month == 3 && handle.close_date.Day == 4);
+                  // close date
+                  DateTime close_dt = handle.close_date;
+                  Assert.IsTrue(close_dt.Year == 2002 && close_dt.Month == 3 && close_dt.Day == 4);
 
                   Assert.IsTrue(handle.session_high == 9.9);
                   Assert.IsTrue(handle.session_low == 10.1);