* added accountSubscriber template class

* removed unnecessary structs

Author: papadpickle

examples/orderbookSubscribe.cpp

@@ -2,33 +2,37 @@
 #include <spdlog/spdlog.h>
 
 #include <chrono>
+#include <mutex>
 #include <websocketpp/client.hpp>
 #include <websocketpp/config/asio_client.hpp>
 
 #include "mango_v3.hpp"
-#include "orderbook/levelOne.hpp"
 #include "solana.hpp"
+#include "subscriptions/accountSubscriber.hpp"
 #include "subscriptions/orderbook.hpp"
-#include "subscriptions/trades.hpp"
 
 class updateLogger {
  public:
-  updateLogger(mango_v3::subscription::orderbook& orderbook,
-               mango_v3::subscription::trades& trades)
+  updateLogger(
+      mango_v3::subscription::orderbook& orderbook,
+      mango_v3::subscription::accountSubscriber<mango_v3::Trades>& trades)
       : orderbook(orderbook), trades(trades) {
     orderbook.registerUpdateCallback(std::bind(&updateLogger::logUpdate, this));
     trades.registerUpdateCallback(std::bind(&updateLogger::logUpdate, this));
-    orderbook.registerCloseCallback(std::bind(&updateLogger::stop, this));
-    trades.registerCloseCallback(std::bind(&updateLogger::stop, this));
+    orderbook.registerCloseCallback(std::bind(&updateLogger::abort, this));
+    trades.registerCloseCallback(std::bind(&updateLogger::abort, this));
+  }
 
+  void start() {
     orderbook.subscribe();
     trades.subscribe();
   }
 
   void logUpdate() {
+    std::scoped_lock lock(logMtx);
     auto level1Snapshot = orderbook.getLevel1();
     if (level1Snapshot->valid()) {
-      auto latestTrade = trades.getLastTrade();
+      auto latestTrade = trades.getAccount()->getLastTrade();
       spdlog::info("============Update============");
       spdlog::info("Latest trade: {}",
                    *latestTrade ? to_string(*latestTrade) : "not received yet");
@@ -43,39 +47,40 @@ class updateLogger {
     }
   }
 
-  void stop() {
+  void abort() {
     spdlog::error("websocket subscription error");
     throw std::runtime_error("websocket subscription error");
   }
 
  private:
   mango_v3::subscription::orderbook& orderbook;
-  mango_v3::subscription::trades& trades;
+  mango_v3::subscription::accountSubscriber<mango_v3::Trades>& trades;
+  std::mutex logMtx;
 };
 
 int main() {
-  const auto& config = mango_v3::MAINNET;
+  using namespace mango_v3;
+  const auto& config = MAINNET;
   const solana::rpc::Connection solCon;
-  const auto group = solCon.getAccountInfo<mango_v3::MangoGroup>(config.group);
+  const auto group = solCon.getAccountInfo<MangoGroup>(config.group);
 
   const auto symbolIt =
       std::find(config.symbols.begin(), config.symbols.end(), "SOL");
   const auto marketIndex = symbolIt - config.symbols.begin();
   assert(config.symbols[marketIndex] == "SOL");
 
   const auto perpMarketPk = group.perpMarkets[marketIndex].perpMarket;
-  auto market =
-      solCon.getAccountInfo<mango_v3::PerpMarket>(perpMarketPk.toBase58());
+  auto market = solCon.getAccountInfo<PerpMarket>(perpMarketPk.toBase58());
   assert(market.mangoGroup.toBase58() == config.group);
 
-  mango_v3::subscription::trades trades(market.eventQueue.toBase58());
-  mango_v3::subscription::orderbook book(market.bids.toBase58(),
-                                         market.asks.toBase58());
+  subscription::accountSubscriber<Trades> trades(market.eventQueue.toBase58());
+  subscription::orderbook book(market.bids.toBase58(), market.asks.toBase58());
 
   updateLogger logger(book, trades);
+  logger.start();
 
   while (true) {
-    std::this_thread::sleep_for(10000s);
+    std::this_thread::sleep_for(100s);
   }
 
   return 0;

include/mango_v3.hpp

@@ -1,13 +1,12 @@
 #pragma once
 
 #include <cstdint>
-#include <mutex>
+#include <memory>
 #include <stack>
 #include <string>
 
 #include "fixedp.h"
 #include "int128.hpp"
-#include "orderbook/order.hpp"
 #include "solana.hpp"
 
 namespace mango_v3 {
@@ -246,20 +245,79 @@ struct FreeNode {
   uint8_t padding[BOOK_NODE_SIZE - 8];
 };
 
+struct Order {
+  Order(uint64_t price, uint64_t quantity) : price(price), quantity(quantity) {}
+  uint64_t price = 0;
+  uint64_t quantity = 0;
+};
+
+struct L1Orderbook {
+  uint64_t highestBid = 0;
+  uint64_t highestBidSize = 0;
+  uint64_t lowestAsk = 0;
+  uint64_t lowestAskSize = 0;
+  double midPoint = 0.0;
+  double spreadBps = 0.0;
+
+  bool valid() const {
+    return ((highestBid && lowestAsk) && (lowestAsk > highestBid)) ? true
+                                                                   : false;
+  }
+};
+
 class BookSide {
  public:
   BookSide(Side side) : side(side) {}
 
+  bool handleMsg(const nlohmann::json &msg) {
+    // ignore subscription confirmation
+    const auto itResult = msg.find("result");
+    if (itResult != msg.end()) {
+      return false;
+    }
+
+    const std::string encoded = msg["params"]["result"]["value"]["data"][0];
+    const std::string decoded = solana::b64decode(encoded);
+    return update(decoded);
+  }
+
+  Order getBestOrder() const {
+    return (!orders->empty()) ? orders->front() : Order(0, 0);
+  }
+
+  uint64_t getVolume(uint64_t price) const {
+    if (side == Side::Buy) {
+      return getVolume<std::greater_equal<uint64_t>>(price);
+    } else {
+      return getVolume<std::less_equal<uint64_t>>(price);
+    }
+  }
+
+ private:
+  template <typename Op>
+  uint64_t getVolume(uint64_t price) const {
+    Op operation;
+    uint64_t volume = 0;
+    for (auto &&order : *orders) {
+      if (operation(order.price, price)) {
+        volume += order.quantity;
+      } else {
+        break;
+      }
+    }
+    return volume;
+  }
+
   bool update(const std::string decoded) {
     if (decoded.size() != sizeof(BookSideRaw)) {
       throw std::runtime_error("invalid response length " +
                                std::to_string(decoded.size()) + " expected " +
                                std::to_string(sizeof(BookSideRaw)));
     }
-    memcpy(&raw, decoded.data(), sizeof(BookSideRaw));
+    memcpy(&(*raw), decoded.data(), sizeof(BookSideRaw));
 
-    auto iter = BookSide::BookSideRaw::iterator(side, raw);
-    book::order_container newOrders;
+    auto iter = BookSide::BookSideRaw::iterator(side, *raw);
+    std::vector<Order> newOrders;
     while (iter.stack.size() > 0) {
       if ((*iter).tag == NodeType::LeafNode) {
         const auto leafNode =
@@ -272,37 +330,21 @@ class BookSide {
             !leafNode->timeInForce ||
             leafNode->timestamp + leafNode->timeInForce < nowUnix;
         if (isValid) {
-          newOrders.orders.emplace_back((uint64_t)(leafNode->key >> 64),
-                                        leafNode->quantity);
+          newOrders.emplace_back((uint64_t)(leafNode->key >> 64),
+                                 leafNode->quantity);
         }
       }
       ++iter;
     }
 
-    if (!newOrders.orders.empty()) {
-      std::scoped_lock lock(mtx);
-      orders = std::move(newOrders);
+    if (!newOrders.empty()) {
+      orders = std::make_shared<std::vector<Order>>(std::move(newOrders));
       return true;
     } else {
       return false;
     }
   }
 
-  book::order getBestOrder() const {
-    std::scoped_lock lock(mtx);
-    return orders.getBest();
-  }
-
-  uint64_t getVolume(uint64_t price) const {
-    std::scoped_lock lock(mtx);
-    if (side == Side::Buy) {
-      return orders.getVolume<std::greater_equal<uint64_t>>(price);
-    } else {
-      return orders.getVolume<std::less_equal<uint64_t>>(price);
-    }
-  }
-
- private:
   struct BookSideRaw {
     MetaData metaData;
     uint64_t bumpIndex;
@@ -348,10 +390,58 @@ class BookSide {
     };
   };
 
-  Side side;
-  BookSideRaw raw;
-  book::order_container orders;
-  mutable std::mutex mtx;
+  const Side side;
+  std::shared_ptr<BookSideRaw> raw = std::make_shared<BookSideRaw>();
+  std::shared_ptr<std::vector<Order>> orders =
+      std::make_shared<std::vector<Order>>();
+};
+
+class Trades {
+ public:
+  auto getLastTrade() const { return latestTrade; }
+
+  bool handleMsg(const nlohmann::json &msg) {
+    // ignore subscription confirmation
+    const auto itResult = msg.find("result");
+    if (itResult != msg.end()) {
+      return false;
+    }
+
+    // all other messages are event queue updates
+    const std::string method = msg["method"];
+    const int subscription = msg["params"]["subscription"];
+    const int slot = msg["params"]["result"]["context"]["slot"];
+    const std::string data = msg["params"]["result"]["value"]["data"][0];
+
+    const auto decoded = solana::b64decode(data);
+    const auto events = reinterpret_cast<const EventQueue *>(decoded.data());
+    const auto seqNumDiff = events->header.seqNum - lastSeqNum;
+    const auto lastSlot =
+        (events->header.head + events->header.count) % EVENT_QUEUE_SIZE;
+
+    bool gotLatest = false;
+    if (events->header.seqNum > lastSeqNum) {
+      for (int offset = seqNumDiff; offset > 0; --offset) {
+        const auto slot =
+            (lastSlot - offset + EVENT_QUEUE_SIZE) % EVENT_QUEUE_SIZE;
+        const auto &event = events->items[slot];
+
+        if (event.eventType == EventType::Fill) {
+          const auto &fill = (FillEvent &)event;
+          latestTrade = std::make_shared<uint64_t>(fill.price);
+          gotLatest = true;
+        }
+        // no break; let's iterate to the last fill to get the latest fill order
+      }
+    }
+
+    lastSeqNum = events->header.seqNum;
+    return gotLatest;
+  }
+
+ private:
+  uint64_t lastSeqNum = INT_MAX;
+  std::shared_ptr<uint64_t> latestTrade = std::make_shared<uint64_t>(0);
 };
 
 #pragma pack(pop)