spark/src/lib/md/MBOServiceServer.cpp

#ifndef _WIN32
#include "MBOServiceServer.h" 

#include "lib/api/services/OrderService.client.pb.h"
#include "lib/utils/Dispatcher.h"
#include "lib/utils/TimeStamp.h"
#include "lib/domain_model/Security.h"
#include "MDClient.h"
#include "ExchangeOrderKey.h"
#include "MBOProcessor.h"

#include "lib/utils/LockBox.h"

#include <tuple>
#include <vector>
#include <unordered_map>


using namespace bts::api::orderman;

namespace bts { 
namespace md { 

//
// Simple class to handle MD recovery and maintain an index of SecurityKey=>MDConsumerHandle

class HandleIndex {
   MDClient& _client;
   int _sub;
   std::optional<MDConsumer> _consumer;

public:
   utils::LockBox<std::unordered_map<SecurityKey,MDConsumerHandle>> index;

   HandleIndex(md::MDClient& client): 
      _client(client),
      _sub(0),
      _consumer(std::nullopt),
      index() {
         _sub = _client.on_consumer().subscribe([=](const MDConsumer& c) { on_consumer(c); });
         on_consumer(_client.current_consumer());
      }

   ~HandleIndex() { _client.on_consumer().unsubscribe(_sub); }


private:
   void on_consumer(const MDConsumer& c) {
      MDConsumer temp(c);
      if (temp.all_heartbeats_ok()) {
         auto existing = _consumer ? _consumer->id() : 0;
         LOG(md,Info, "Updateing MBOService MDConsumer. Old {} new {}", existing, temp.id());
         // clear the index
         index.lock([](auto& index) { index.clear(); });
         // Swap to new consumer
         if (_consumer) {
            _consumer->swap(temp);
         }
         else { _consumer = temp; }
         // Subscribe to on_security
         auto id = _consumer->id();
         _consumer->on_security([this,id](const SecurityPtr&, MDConsumerHandle handle) {
            if (_consumer->id() == id) {
               index.lock([handle](auto& index) {
                  index[SecurityKey(handle.key())] = handle;
               });
            }
         });
      }
   }
};

class ActiveOrder {
   SecurityKey _security;
   SparkOrderKey _id;
   ExchangeOrderKey  _ex_id;
   int32_t _position;

public:
   // empty constructor needed for unordered map operator[]
   // ActiveOrder(): _security(), _id(), _ex_id(), _position(-1) { bts_assert(false); }
   ActiveOrder(SecurityKey sec, SparkOrderKey s, ExchangeOrderKey ex): _security(sec), _id(s), _ex_id(ex), _position(-1)  {}
   ActiveOrder(SecurityKey sec, SparkOrderKey s, const std::string& ex): _security(sec), _id(s), _ex_id(), _position(-1) {
      _ex_id = ExchangeOrderKey(std::stoul(ex));
   }

   SecurityKey security_id() const { return _security; }
   SparkOrderKey id() const { return _id; }
   ExchangeOrderKey exchange_id() const { return _ex_id; }
   int32_t& position() { return _position; };
   int32_t position() const { return _position; };
   bool set_position(int32_t pos) {
      if (pos != _position) {
         _position = pos;
         return true;
      }
      return false;
   }
   OrderPIQ to_piq() const {
      OrderPIQ result;
      result.set_order_id(_id.value());
      result.set_position(_position);
      return result;
   }
};

//
// MBOService

struct MBOServiceImpl {
   MBOServiceImpl(
      api::ServiceContext& ctx,
      utils::Dispatcher& d,
      md::MDClient& c,
      MBOServiceBase::PiqPublisher&& pub
      ):
      dispatcher(d),
      client(c),
      orders(ctx.locate<api::orderman::OrderService>()),
      handle_index(c),
      publish(pub),
      timer(d, 200, [&]() {timer_tick();}),
      orders_sub()
   {}

   utils::Dispatcher& dispatcher;
   md::MDClient& client;
   OrderService orders;

   HandleIndex handle_index;
   MBOServiceBase::PiqPublisher publish;
   utils::DispatcherTimer timer;

   rxcpp::composite_subscription orders_sub;
   std::unordered_map<SparkOrderKey, ActiveOrder> active;

   void timer_tick();
   void flush();
};



///////////////////////

std::optional<uint32_t> search_orders(const ExchangeOrderKey oid, const std::vector<SnapshotOrder>& orders) {
   IntPrice price = {};
   size_t position = 0;
   for (size_t x = 0; x < orders.size(); ++x) {
      const auto& oo = orders[x];
      if (oo.id == oid) {
         if (oo.price == price) return { position };
         else return 0;
      }
      else if (price == oo.price) {
         position += oo.size;
      }
      else {
         price = oo.price;
         position = oo.size;
      }
   }
   return std::nullopt;
}

bool update_position(const HandleIndex& handles, ActiveOrder& order) {
   const auto& index = handles.index;
   auto handle = index.lock([id=order.security_id()](const auto& index) {
      auto iter = index.find(id);
      return iter != index.end() ? iter->second : MDConsumerHandle {};
   });

   if (handle.is_initialized()) {
      auto mbo = handle.get_mbo();
      if (mbo) {
         auto search_result = search_orders(order.exchange_id(), mbo->bids);
         if (!search_result) {
            search_result = search_orders(order.exchange_id(), mbo->asks);
         }
         if (search_result) {
            return order.set_position(search_result.value());
         }
      }
   }
   return false;
}

void MBOServiceImpl::timer_tick() {
   //Dispatcher thread
   std::vector<OrderPIQ> rows;
   for (auto& pair : active) {
      if (update_position(handle_index, pair.second)) {
         rows.push_back(pair.second.to_piq());
      }
   }
   if (!rows.empty()) {
      publish(nullptr, &rows, nullptr);
   }
}

const NativeOrderState* native_order_state(const OrderRecord& record) {
   switch(record.state().custom_case()) {
      case OrderState::kMarket:
         return &record.state().market().native_state();
      case OrderState::kLimit:
         return &record.state().limit().native_state();
      case OrderState::kStop:
         return &record.state().stop().native_state();
      case OrderState::kStopLimit:
         return &record.state().stop_limit().native_state();
      case OrderState::kGlbxIceberg:
         return &record.state().glbx_iceberg().native_state();
      default:
      return nullptr;
   }
}

bool is_active_exchange_order(const OrderRecord& record) {
   return record.state().status() == ORDER_STATUS_WORKING &&
          native_order_state(record) != nullptr;
}

void try_remove(MBOServiceImpl& me, std::set<uint32_t>& to_remove, const OrderRecord& record) {
   auto key = SparkOrderKey(record.key());
   auto iter = me.active.find(key);
   if (iter != me.active.end()) {
      me.active.erase(key);
      to_remove.insert(key.value());
   }
}
void try_add(MBOServiceImpl& me, std::vector<OrderPIQ>& to_add, const OrderRecord& record) {
   if (is_active_exchange_order(record)) {
      auto native = native_order_state(record);
      bts_assert(native);
      ActiveOrder order(
         SecurityKey(record.def().security_key()), 
         SparkOrderKey(record.key()), 
         native->exchange_order_id());
      update_position(me.handle_index, order);
      to_add.push_back(order.to_piq());
      me.active.insert(std::make_pair(order.id(), order));
   }
}

MBOServiceServer::MBOServiceServer(
      api::ServiceContext& ctx,
      utils::Dispatcher& d,
      md::MDClient& c
      ) : 
         MBOServiceBase(ctx),
         _me(new MBOServiceImpl(ctx,d,c,make_piq_publisher()))
      {
         // subscribe to mdclient ✔
         // make publishers ✔
         // subscribe to orders => stash active orders dict
         _me->orders_sub =
         _me->orders.subscribe_orders().subscribe([=](const auto& update) {
            // api thread
            const auto& added = update.added();
            const auto& changed = update.changed();
            const auto& removed = update.removed();
            _me->dispatcher.post([this,added,changed,removed]() {
               // dispatcher thread
               std::vector<OrderPIQ> to_add, to_change;
               std::set<uint32_t> to_remove;

               // Remove
               for(const auto& r : removed) {
                  // If it was active remove it and add to to_remove
                  try_remove(*_me, to_remove, *r);
               }
               // Added
               for(const auto& add : added) {
                  // If active add look up the position and add to the state
                  try_add(*_me, to_add, *add);
               }
               // Changed
               for(const auto& ch : changed) {
                  // new order is active and old is not look up piq and add to state
                  bool is_active = is_active_exchange_order(*ch.current());
                  bool was_active = is_active_exchange_order(*ch.previous());
                  if (is_active && !was_active) {
                     try_add(*_me, to_add, *ch.current());
                  }
                  else if (was_active && !is_active) {
                     try_remove(*_me, to_remove, *ch.current());
                  }
                  // is_active && was_active => don't care
                  // !is_active && !was_active => don't care
               }

               _me->publish(&to_add, &to_change, &to_remove);
               if (!_me->timer.running()) _me->timer.start();
            });
         });
      }

MBOServiceServer::~MBOServiceServer() { }


void MBOServiceServer::get_orders_for_security(const MBOSecurityQuery& request, const api::Response<PositionData>& reply) {
   PositionData response;
   response.set_security_id(request.security_id());
   auto handle = _me->handle_index.index.lock([id=request.security_id()](const auto& index) {
      auto iter = index.find(SecurityKey(id));
      return iter != index.end() ? iter->second : MDConsumerHandle {};
   });

   std::unordered_map<ExchangeOrderKey, SparkOrderKey> reverse;
   for (const auto& pair : _me->active) {
      reverse.insert(std::make_pair(pair.second.exchange_id(), pair.first));
   }

   size_t count = 0;
   auto tick = handle.get_minimum_tick();
   if(handle.is_initialized()) {
      auto mbo = handle.get_mbo();
      if (mbo) {
         for (const auto& o : mbo->bids) {
            double price = o.price.price();
            bool isnan = os::is_nan(request.price_level());
            if (isnan || securities::test_price_equality(request.price_level(), price, tick)) {
               ++count;
               if (request.limit() && count > request.limit()) break;
               auto* entry = response.add_bid();
               entry->set_exchange_id(o.id.value());
               if (isnan) entry->set_price(price);
               entry->set_size(o.size);
               auto iter = reverse.find(o.id);
               if (iter != reverse.end()) {
                  entry->set_order_id(iter->second.value());
               }
            }
         }
         for (const auto& o : mbo->asks) {
            double price = o.price.price();
            bool isnan = os::is_nan(request.price_level());
            if (isnan || securities::test_price_equality(request.price_level(), price, tick)) {
               ++count;
               if (request.limit() && count > request.limit()) break;
               auto* entry = response.add_ask();
               entry->set_exchange_id(o.id.value());
               if (isnan) entry->set_price(price);
               entry->set_size(o.size);
               auto iter = reverse.find(o.id);
               if (iter != reverse.end()) {
                  entry->set_order_id(iter->second.value());
               }
            }
         }
      }
   }
   
   reply(response);
}

}} //namespace 
#endif