Pull data from ftp and parse

Joel Grunbaum

2022-01-12 611ad7b12fcc6b34229316099ae66bccbcf24ad9

 protocol.cpp

@@ -2,13 +2,17 @@

#include "book.hpp"
#include "cpp-httplib/httplib.h"
#include "easywsclient/easywsclient.hpp"
#include "json.hpp"
#include "secrets.hpp"

#include <chrono>
#include <cstddef>
#include <iostream>
#include <memory>
#include <queue>
#include <sstream>
#include <stdexcept>
#include <string>
#include <unordered_map>

@@ -18,6 +22,8 @@
   mapAnnounce;
std::string server = std::string(HOST) + ":" + std::string(PORT);
httplib::Client cli("http://" + server);

double lastime = 0;

void initialise()
{
@@ -40,26 +46,54 @@
   l = res->body;
   std::queue<json::Message*> a(json::parse(l));
   while (!a.empty()) {
      protocol::handleMessage(bs, a.front());
      if (static_cast<json::FromExchange*>(a.front()) != nullptr ||
          static_cast<json::FromExchange*>(a.front())->timestamp > lastime) {
         lastime = static_cast<json::FromExchange*>(a.front())->timestamp;
         protocol::handleMessage(bs, a.front());
      }
      delete a.front();
      a.pop();
   }
   return bs;
}

void catchUp(std::unordered_map<std::string, book::Book>& bs)
{
   static std::unique_ptr<easywsclient::WebSocket> ws(
      easywsclient::WebSocket::from_url("ws://" + server + "/information"));
   std::string feed;
   bool gotMessage = false;
   ws->poll();
   ws->dispatch([gotMessageOut = &gotMessage, messageOut = &feed,
                 ws = ws.get()](const std::string& message) {
      *gotMessageOut = true;
      *messageOut = message;
   });
   if (gotMessage) {
      std::queue<json::Message*> a(json::parse(feed));
      while (!a.empty()) {
         if (static_cast<json::FromExchange*>(a.front()) != nullptr ||
             static_cast<json::FromExchange*>(a.front())->timestamp >
                 lastime) {
            lastime =
               static_cast<json::FromExchange*>(a.front())->timestamp;
         }
         protocol::handleMessage(bs, a.front());
         delete a.front();
         a.pop();
      }
   }
}

json::Message* addOrder(json::AddMessage& order)
{
   std::string message = "{\"message\": " + order.as_string() + ", " +
                         "\"username\": \"" + std::string(USER) +
                         "\", \"password\": \"" + std::string(PASS) + "\"}";
   std::string message = order.as_string();
   return send(message);
}

json::Message* deleteOrder(json::DeleteMessage& order)
{
   std::string message = "{\"message\": " + order.as_string() +
                         ", \"username\": \"" + std::string(USER) +
                         "\", \"password\": \"" + std::string(PASS) + "\"}";
   std::string message = order.as_string();
   return send(message);
}

@@ -67,29 +101,32 @@
                   json::Message* message)
{
   if (mapAnnounce.empty()) initialise();
   if (dynamic_cast<json::FromExchange*>(message) != nullptr) {
      lastime = dynamic_cast<json::FromExchange*>(message)->timestamp;
   }
   switch (message->type) {
   case json::FUTURE_TYPE:
   case json::SPREAD_TYPE:
   case json::CALL_TYPE:
   case json::PUT_TYPE:
      announce(bs, dynamic_cast<json::AnnounceMessage*>(message));
      announce(bs, static_cast<json::AnnounceMessage*>(message));
      break;
   case json::SETTLEMENT:
      settle(bs, dynamic_cast<json::SettleMessage*>(message));
      settle(bs, static_cast<json::SettleMessage*>(message));
      break;
   case json::ADDED:
      addedOrder(bs, dynamic_cast<json::AddedMessage*>(message));
      addedOrder(bs, static_cast<json::AddedMessage*>(message));
      break;
   case json::DELETED:
      deletedOrder(bs, dynamic_cast<json::DeletedMessage*>(message));
      deletedOrder(bs, static_cast<json::DeletedMessage*>(message));
      break;
   case json::TRADE:
      tradeOrder(bs, dynamic_cast<json::TradeMessage*>(message));
      tradeOrder(bs, static_cast<json::TradeMessage*>(message));
      break;
   case json::BROKER_REQUEST:
   case json::BROKER_ACK:
   case json::BROKER_CONFIRM:
      broker(bs, dynamic_cast<json::Broker*>(message));
      broker(bs, static_cast<json::Broker*>(message));
      break;
   default:;
   }
@@ -111,11 +148,11 @@
void addedOrder(std::unordered_map<std::string, book::Book>& bs,
                json::AddedMessage* message)
{
   if (message->side == book::Buy) {
   if (message->side == book::Buy && message->resting) {
      book::Order t(message->price, book::Buy, message->resting,
                    message->timestamp, message->id);
      bs[message->product].bid(t);
   } else {
   } else if (message->side == book::Sell && message->resting) {
      book::Order t(message->price, book::Sell, message->resting,
                    message->timestamp, message->id);
      bs[message->product].ask(t);
@@ -125,42 +162,33 @@
                  json::DeletedMessage* message)
{
   if (message->side == book::Buy) {
      for (std::size_t i = 0; i < bs[message->product].bidSide.size(); i++) {
         if (bs[message->product].bidSide[i].id == message->id) {
            bs[message->product].bidSide[i] =
               bs[message->product].bidSide.back();
            bs[message->product].bidSide.pop_back();
            std::make_heap(bs[message->product].bidSide.begin(),
                           bs[message->product].bidSide.end(),
                           std::less<book::Level>());
         }
      }
      bs[message->product].bidSide.erase(message->id);
   } else {
      for (std::size_t i = 0; i < bs[message->product].askSide.size(); i++) {
         if (bs[message->product].askSide[i].id == message->id) {
            bs[message->product].askSide[i] =
               bs[message->product].askSide.back();
            bs[message->product].askSide.pop_back();
            std::make_heap(bs[message->product].askSide.begin(),
                           bs[message->product].askSide.end(),
                           std::greater<book::Level>());
         }
      }
      bs[message->product].askSide.erase(message->id);
   }
}
void tradeOrder(std::unordered_map<std::string, book::Book>& bs,
                json::TradeMessage* message)
{
   if (message->tradeType == json::BUY_AGGRESSOR) {
      book::Order t(message->price, book::Buy, message->volume,
                    message->timestamp, message->passiveOrder);
      bs[message->product].bid(t);
   } else {
      book::Order t(message->price, book::Sell, message->volume,
                    message->timestamp, message->passiveOrder);
      bs[message->product].ask(t);
   if (bs.find(message->passiveOrder) != bs.end()) {
      if (message->tradeType == json::BUY_AGGRESSOR) {
         if (message->passiveOrderRemaining > 0) {
            bs[message->product].askSide.at(message->passiveOrder).volume =
               message->passiveOrderRemaining;
         } else {
            bs[message->product].askSide.erase(message->passiveOrder);
         }
      } else if (message->tradeType == json::SELL_AGGRESSOR) {
         if (message->passiveOrderRemaining > 0) {
            bs[message->product].bidSide.at(message->passiveOrder).volume =
               message->passiveOrderRemaining;
         } else {
            bs[message->product].bidSide.erase(message->passiveOrder);
         }
      }
   }
}

void broker(std::unordered_map<std::string, book::Book>& bs,
            json::Broker* message)
{
@@ -168,9 +196,13 @@

json::Message* send(std::string& message)
{
   auto res = cli.Post("/execution", message, "text/plain");
   std::queue<json::Message*> a = json::parse(res->body);
   return a.front();
   httplib::MultipartFormDataItems a = {{"message", message, "", ""},
                                        {"username", USER, "", ""},
                                        {"password", PASS, "", ""}};
   auto res = cli.Post("/execution", a);
   std::string b = res->body;
   std::queue<json::Message*> c = json::parse(b);
   return c.front();
}

} // namespace protocol