add derived data support and start.py

Author: open-trade

Diff for: .gitignore

@@ -17,3 +17,4 @@ ticks
 *.txt
 *.con
 opentrade-dbg
+nohup.out

Diff for: src/opentrade/account.h

@@ -72,6 +72,8 @@ struct User : public AccountBase {
   SubAccountMapPtr sub_accounts_ = std::make_shared<SubAccountMap>();
 };
 
+inline const User kEmptyUser;
+
 class AccountManager : public Singleton<AccountManager> {
  public:
   static void Initialize();

Diff for: src/opentrade/algo.cc

@@ -177,6 +177,19 @@ void AlgoManager::Run(int nthreads) {
     runners_[i].tid_ = threads_[i].get_id();
   }
 #endif
+
+  for (auto& pair : adapters()) {
+    if (pair.first.at(0) != '_') continue;
+    auto user_name = pair.second->config("user");
+    auto user = AccountManager::Instance().GetUser(user_name);
+    auto algo = Spawn(std::make_shared<Algo::ParamMap>(), pair.first,
+                      user ? *user : kEmptyUser, "{}", "");
+    if (algo) {
+      LOG_INFO("Started " << pair.first << " , id=" << algo->id());
+    } else {
+      LOG_ERROR("Failed to start" << pair.first);
+    }
+  }
 }
 
 void AlgoManager::Handle(Confirmation::Ptr cm) {
@@ -361,7 +374,7 @@ Instrument* Algo::Subscribe(const Security& sec, DataSrc src, bool listen) {
   assert(adapter);
   auto inst = new Instrument(this, sec, DataSrc(adapter->src()));
   inst->md_ = &MarketDataManager::Instance().Get(sec, adapter->src());
-  inst->id_ = ++Instrument::kIdCounter;
+  inst->id_ = ++Instrument::id_counter_;
   inst->listen_ = listen;
   instruments_.insert(inst);
   if (listen) AlgoManager::Instance().Register(inst);

Diff for: src/opentrade/algo.h

@@ -158,7 +158,7 @@ class Instrument {
   bool listen_ = true;
   friend class AlgoManager;
   friend class Algo;
-  static inline std::atomic<size_t> kIdCounter = 0;
+  static inline std::atomic<size_t> id_counter_ = 0;
 };
 
 class AlgoRunner {

Diff for: src/opentrade/connection.cc

@@ -1106,10 +1106,11 @@ void Connection::OnLogin(const std::string& action, const json& j) {
       Send(json{"broker_account", pair.first, pair.second->name});
     }
     for (auto& pair : AlgoManager::Instance().adapters()) {
+      if (pair.first.at(0) == '_') continue;
       auto& params = pair.second->GetParamDefs();
       json j = {
           "algo_def",
-          pair.second->name(),
+          pair.first,
       };
       for (auto& p : params) {
         json j2 = {

Diff for: src/opentrade/main.cc

@@ -152,6 +152,7 @@ int main(int argc, char *argv[]) {
   for (auto &section : prop_tree) {
     if (!section.second.size()) continue;
     auto section_name = section.first;
+    if (section_name.empty()) continue;
     opentrade::Adapter::StrMap params;
     for (auto &item : section.second) {
       auto name = item.first;
@@ -206,7 +207,7 @@ int main(int argc, char *argv[]) {
          boost::make_iterator_range(fs::directory_iterator(kAlgoPath), {})) {
       auto path = entry.path();
       auto fn = path.filename().string();
-      if (fn[0] == '_' && fn[0] == '.') continue;
+      if (fn[0] == '.') continue;
       opentrade::Algo *algo = nullptr;
       auto algoname = fn.substr(0, fn.length() - 3);
       if (path.extension() == ".py") {

Diff for: src/opentrade/market_data.h

@@ -2,6 +2,7 @@
 #define OPENTRADE_MARKET_DATA_H_
 
 #include <tbb/concurrent_unordered_map.h>
+#include <any>
 #include <map>
 #include <string>
 
@@ -55,6 +56,30 @@ struct MarketData {
 
   Trade trade;
   Depth depth;
+
+  void SetDerived(const std::string& name, std::any value) {
+    std::lock_guard<std::mutex> lock(mutex_);
+    if (!derived) derived = new AnyMap;
+    derived->emplace(name, value);
+  }
+
+  template <typename T>
+  const T* GetDerived(const std::string& name) const {
+    std::lock_guard<std::mutex> lock(mutex_);
+    if (!derived) return {};
+    auto it = derived->find(name);
+    if (it == derived->end()) return {};
+    try {
+      return std::any_cast<const T*>(it->second);
+    } catch (const std::bad_any_cast& e) {
+      return {};
+    }
+  }
+
+ private:
+  typedef std::map<std::string, std::any> AnyMap;
+  AnyMap* derived = nullptr;
+  static inline std::mutex mutex_;
 };
 
 struct DataSrc {

Diff for: src/opentrade/server.cc

@@ -272,6 +272,9 @@ void Server::Start(int port, int nthreads) {
     for (auto i = 0; i < nthreads; ++i) {
       threads.emplace_back([]() { kIoService->run(); });
     }
+    if (fs::exists(fs::path("start.py"))) {
+      system(("nohup ./start.py " + std::to_string(port) + " &").c_str());
+    }
     for (auto& t : threads) t.join();
   } catch (std::runtime_error& e) {
     LOG_ERROR("failed to start web server: " << e.what());