AW2 now gets its time from the server

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

@@ -13,21 +13,28 @@ namespace mbts.spark {
          bts.Log.Gen.info($"SingleVolSurfaceTimeSource cons vs_id:{vs_id}");
          _vs_id = vs_id;
          _bb = bb;
-         _bb.timer_callback += _on_timer;
+         _bb.subscribe(SparkTags.VolSurfaceStandardServerTimeTag);
+         _bb.data_callback += _on_data;
+         _read_and_publish();
       }
 
-      private void _on_timer() {
-         if (DateTime.Now - _last_publish_time >= _publish_period) {
-            _last_publish_time = DateTime.Now;
-            VolSurfaceStandardTime time = VolSurfaceStandardTime.get_local_default();
-            bts.Log.Gen.info($"SingleVolSurfaceTimeSource vs_id:{_vs_id} posting time:{time.ToString()}");
-            _bb.post(SparkTags.VolSurfaceStandardTimeTag(_vs_id), time);
+      private void _on_data(object tag, object data) {
+         if (tag.Equals(SparkTags.VolSurfaceStandardServerTimeTag)) {
+            _read_and_publish();
+         }
+      }
+
+      private void _read_and_publish() {
+         VolSurfaceStandardTime vs_time = _bb.read_state(SparkTags.VolSurfaceStandardServerTimeTag);
+         if (vs_time != null) {
+            bts.Log.Gen.info($"SingleVolSurfaceTimeSource vs_id:{_vs_id} posting time:{vs_time.ToString()}");
+            _bb.post(SparkTags.VolSurfaceStandardTimeTag(_vs_id), vs_time);
          }
       }
 
       #region IDisposable
       public void Dispose() {
-         _bb.timer_callback -= _on_timer;
+         _bb.data_callback -= _on_data;
       }
       #endregion
    }

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

@@ -96,6 +96,10 @@ namespace mbts.spark {
          var vs_usage_sub = _analytics_service.subscribei_volsurface_usage()
             .ObserveOn(SynchronizationContext.Current)
             .Subscribe(_on_vs_usage_update);
+         var vs_time_sub = _analytics_service.subscribei_volsurface_time()
+            .ObserveOn(SynchronizationContext.Current)
+            .Subscribe(_on_vs_time);
+         _subs.Add(vs_time_sub);
          _subs.Add(vs_sub);
          _subs.Add(vs_usage_sub);
          _subs.Add(new AnalyticsParametersManager(bus, _analytics_service));
@@ -163,6 +167,17 @@ namespace mbts.spark {
          _init_vss(to_publish.Keys);
       }
 
+      void _on_vs_time(bts.api.core.Update<VolSurfaceServerTimeImmutable, int> update) {
+         if (update.all.Count() != 1) {
+            return;
+         }
+         VolSurfaceStandardTime vs_time =
+            new VolSurfaceStandardTime(update.all.ElementAt(0).UtcTime.Seconds);
+         bts.Log.Gen.info(String.Format("VSBusManager._on_vs_time vs_time: {0}",
+                              vs_time.ToString()));
+         _bb.post(SparkTags.VolSurfaceStandardServerTimeTag, vs_time);
+      }
+
       private void _init_vss(IEnumerable<int> vs_ids) {
          // we consider every vs in the db as "activated" which might prove to be expensive
          // (will take care of it later)

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

@@ -17,6 +17,13 @@ namespace mbts.analytics {
          this.server_time_utc = server_time_utc;
       }
 
+      public VolSurfaceStandardTime(long seconds_from_epoch) {
+         timezone = TimeZoneInfo.Local;
+         server_time_utc =
+            DateTimeOffset.FromUnixTimeSeconds(seconds_from_epoch).DateTime;
+         server_time = server_time_utc.ToLocalTime();
+      }
+
       public static VolSurfaceStandardTime get_local_default() {
          DateTime local_rounded = VolSurfaceUtilityFunctions.round_time_mins(
             DateTime.Now, VolSurfaceConstants.vol_time_sample_mins);

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

@@ -261,6 +261,9 @@ namespace mbts.spark {
       public static ITag<VolSurfaceStandardTime> VolSurfaceStandardTimeTag(int vs_id) =>
          _mk_tag<VolSurfaceStandardTime, int>(vs_id);
 
+      public static ITag<VolSurfaceStandardTime> VolSurfaceStandardServerTimeTag { get; } =
+         _mk_tag<VolSurfaceStandardTime>();
+
       public static ITag<IDictionary<SecurityExpiration, double>> VolSurfaceDriverSymbolPriceTag(int vs_id) =>
          _mk_tag<IDictionary<SecurityExpiration, double>, int>(vs_id);
 

src/lib/api/services/AnalyticsService.proto

@@ -7,6 +7,7 @@ import "lib/api/analytics/ExpirationDriver.proto";
 import "lib/api/domain/SecurityExpiration.proto";
 import "lib/api/domain/SecurityCriteria.proto";
 import "lib/domain_model/Security.proto";
+import "google/protobuf/timestamp.proto";
 
 option csharp_namespace = "bts.api.analytics";
 
@@ -40,6 +41,11 @@ message VolSurfaceUsageList {
   repeated VolSurfaceUsageEntry usage_list = 20;
 }
 
+message VolSurfaceServerTime{
+  int32 id = 10 [ (bts.api.key) = true ];
+  google.protobuf.Timestamp utc_time = 20; 
+}
+
 message ArchivedSecurityTheo {
   uint32 id = 10;
   double driver = 20;
@@ -184,6 +190,8 @@ service AnalyticsService {
       returns (bts.api.ResultMessage);
   rpc volsurface_usage(bts.api.VoidMessage)
       returns (stream VolSurfaceUsageList);
+  rpc volsurface_time(bts.api.VoidMessage)
+      returns (stream VolSurfaceServerTime); 
   // trims volsurface history
   rpc clean_up(IntWrapper) returns (bts.api.VoidMessage);
 

src/server/sparkd/analytics/AnalyticsServiceServer.cpp

@@ -10,10 +10,13 @@
 #include "lib/utils/Puid.h"
 #include "lib/utils/bts_assert.h"
 #include "server/sparkd/algo/AlgoDefinitionServiceServer.h"
+#include <bits/stdint-uintn.h>
 #include <boost/algorithm/string.hpp>
 #include <boost/tokenizer.hpp>
 #include <chrono>
 #include <cstdint>
+#include <google/protobuf/timestamp.pb.h>
+#include <google/protobuf/util/time_util.h>
 #include <locale>
 #include <memory>
 #include <string>
@@ -31,6 +34,12 @@ uint64_t get_epoch() {
    return ms_u;
 }
 
+uint64_t minutes_from_epoch() {
+   using namespace std::chrono;
+   auto mins = duration_cast<minutes>(system_clock::now().time_since_epoch());
+   return mins.count();
+}
+
 // assumes key are ints
 int get_max_id(IStore *const store, const std::string &ns) {
    auto ns_map = store->retrieve_all();
@@ -99,6 +108,9 @@ struct AnalyticsServiceServer::PImpl {
    AnalyticsServiceBase::Exp_parametersStatefulPublisher exp_params_publisher;
    AnalyticsServiceBase::Volsurface_usageStatefulPublisher vs_usage_publisher;
    AnalyticsServiceBase::ReportsStatefulPublisher reports_publisher;
+   AnalyticsServiceBase::Volsurface_timeStatefulPublisher time_publisher;
+   const int64_t poll_period_ms = 500;
+   uint64_t last_vs_time_mins = 0;
    PImpl(
        bts::api::ServiceContext &context,
        const AnalyticsServiceBase::VolsurfacesStatefulPublisher &vs_publisher,
@@ -106,7 +118,9 @@ struct AnalyticsServiceServer::PImpl {
            &exp_params_publisher,
        const AnalyticsServiceBase::Volsurface_usageStatefulPublisher
            &vs_usage_publisher,
-       const AnalyticsServiceBase::ReportsStatefulPublisher &reports_publisher)
+       const AnalyticsServiceBase::ReportsStatefulPublisher &reports_publisher,
+       const AnalyticsServiceBase::Volsurface_timeStatefulPublisher
+           time_publisher)
        : context(context),
          vs_store(std::make_unique<persistence::SqliteStore>("volsurfaces")),
          exp_params_store(std::make_unique<persistence::SqliteStore>(
@@ -118,7 +132,7 @@ struct AnalyticsServiceServer::PImpl {
              std::make_unique<persistence::SqliteStore>("archived_theos")),
          vs_publisher(vs_publisher), exp_params_publisher(exp_params_publisher),
          vs_usage_publisher(vs_usage_publisher),
-         reports_publisher(reports_publisher) {}
+         reports_publisher(reports_publisher), time_publisher(time_publisher) {}
 };
 
 AnalyticsServiceServer::AnalyticsServiceServer(
@@ -129,7 +143,8 @@ AnalyticsServiceServer::AnalyticsServiceServer(
           make_volsurfaces_stateful_publisher(),
           make_exp_parameters_stateful_publisher(),
           make_volsurface_usage_stateful_publisher(),
-          make_reports_stateful_publisher())) {
+          make_reports_stateful_publisher(),
+          make_volsurface_time_stateful_publisher())) {
    // publishing all vol surfaces
    LOG(Gen, Info, "Starting AnalyticsServiceServer");
    auto vs_table = me->vs_store->retrieve_all();
@@ -186,6 +201,7 @@ AnalyticsServiceServer::AnalyticsServiceServer(
       }
       me->reports_publisher.flush();
    }
+   me->context.post([this]() { _poll(); }, me->poll_period_ms);
 }
 
 void AnalyticsServiceServer::get_new_volsurface_id(
@@ -428,4 +444,20 @@ void AnalyticsServiceServer::get_archived_theos(
 
 AnalyticsServiceServer::~AnalyticsServiceServer() {}
 
+void AnalyticsServiceServer::_poll() {
+   uint64_t now_mins = minutes_from_epoch();
+   if (now_mins > me->last_vs_time_mins) {
+      me->last_vs_time_mins = now_mins;
+      VolSurfaceServerTime vs_time = {};
+      vs_time.set_id(0);
+      vs_time.mutable_utc_time()->set_nanos(0);
+      vs_time.mutable_utc_time()->set_seconds(now_mins * 60);
+      me->time_publisher.add_or_update(vs_time);
+      me->time_publisher.flush();
+      LOG(Gen, Info, "AnalyticsServiceServer publishing vs_time: {}",
+          utils::pb::to_string(vs_time));
+   }
+   me->context.post([this]() { _poll(); }, me->poll_period_ms);
+}
+
 } // namespace bts::analytics

src/server/sparkd/analytics/AnalyticsServiceServer.h

@@ -69,5 +69,8 @@ class AnalyticsServiceServer
        override;
 
    ~AnalyticsServiceServer();
+
+ private:
+   void _poll();
 };
 } // namespace bts::analytics