src/coin.cpp - develop - Stormbery/CEX

src/coin.cpp
437 строк21 KB
stormbery
deleted plots objects
20 сен 2024, 21:17
20 сен 2024, 21:174e0e245
#include "coin.h"

  Coin::Coin(std::string   name):
    m_websocketstream(),
    m_name(name),
    m_Price_Old {0},
    m_Value_Old {0},
    m_Value_Temp {0},
    m_Price_Old_Old {0},
    m_Value_Old_Old {0},
    m_Value_Accumulate_ToWrite {0},
    m_count {0}

   {
    m_PricePublic.reserve(ITEM_REGRESS);
    m_ValuePublic.reserve(ITEM_REGRESS);
    m_Value_SellBuy.reserve(ITEM_REGRESS);
    m_Value_Accumulate.reserve(ITEM_REGRESS);
    m_time.reserve(ITEM_REGRESS);
    m_PricePublicRegress.reserve(LAG_REGRESS);

    for (auto i = 1;i <= ITEM_REGRESS; i++) {/// формируем временной диапазон
        m_time.push_back(i);
      }

    m_websocketstream.SetPair(m_name);
    Initlog();// инициализация системы логирования
    boost::log::add_common_attributes();// инициализация системы логирования

    for (int i = 0; i < MAX_NUMBER_FILES; ++i) {/// создание лог файла
        m_fileOrder [i] = boost::filesystem::current_path().string() + m_fileName [i];
        outf[i] = open(m_fileOrder [i].c_str(), O_WRONLY | O_CREAT | O_APPEND  , 0644);
          if ( outf[i] < 0 ) std::cout << "Cant open file   " << m_fileOrder [i] << std::endl;
            else   WebSocketLoggingTrace ("Opened file " + m_fileOrder [i]);
    }

    memset(&m_IO_uring_params, 0, sizeof(m_IO_uring_params));
       /**
     * Создаем инстанс io_uring, не используем никаких кастомных опций.
     */
    uint16_t ret = io_uring_queue_init_params(4, &m_IO_uring, &m_IO_uring_params);
    WebSocketLoggingTrace ("Coin constructor starting....");
   }

  Coin::~Coin() {
    WebSocketLoggingTrace ("Coin destructor starting....");
    io_uring_queue_exit(&m_IO_uring);

    for (int i = 0; i < MAX_NUMBER_FILES; ++i) {
        close( outf[i]);
      }
  }
   /**
    * Вычисление полиномиальной регрессии
    */
  std::vector<double> Coin::Regress(const std::vector<double> &x, const std::vector<double> &y, uint32_t count, uint32_t degree){

    std::vector<double> result;
    result.reserve(count);
    std::vector<std::vector<double>> p(degree + 1, std::vector<double>(ITEM_REGRESS));
    std::vector<double> s(degree + 1), q(degree + 1);

    for (int i = 0; i < degree + 1; ++i) {
        if (i == 0) {
            fill(p[i].begin(), p[i].end(), 1.0);
        } else if (i == 1) {
            for (int j = 0; j < ITEM_REGRESS; ++j) {
                p[i][j] = (x[j] - q[i - 1] / s[i - 1]);
            }
        } else {
            for (int j = 0; j < ITEM_REGRESS; ++j) {
                p[i][j] = (x[j] - q[i - 1] / s[i - 1]) * p[i - 1][j] - s[i - 1] / s[i - 2] * p[i - 2][j];
            }
        }
        s[i] = 0.0;
        q[i] = 0.0;

    for (int j = 0; j < ITEM_REGRESS; ++j) {
        s[i] += p[i][j] * p[i][j];
        q[i] += p[i][j] * p[i][j] * x[j];
        }
    }

    std::vector<std::vector<double>> c(degree + 1);

    for (int k = 0; k <= degree; ++k) {
        c[k].resize(k + 1);
        c[k][k] = 1.0;
    }

    for (int k = 1; k < degree + 1; ++k) {
        for (int j = 0; j < k; ++j) {
            c[k][j] = (j > 0 ? c[k - 1][j - 1] : 0) -
                      (q[k - 1] / s[k - 1] * c[k - 1][j]) -
                      ((k > 1) && (j < k - 1) ? (s[k - 1] / s[k - 2] * c[k - 2][j]) : 0);
        }
    }

    std::vector<double> b(degree+ 1);

    for (int k = 0; k <= degree; ++k) {
        b[k] = 0.0;
        for (int i = 0; i < ITEM_REGRESS; ++i) {
            b[k] += y[i] * p[k][i];
        }
        b[k] /= s[k];
    }

    std::vector<double> a(degree + 1);

    for (int j = 0; j <= degree; ++j) {
        a[j] = 0.0;
        for (int k = j; k <= degree; ++k) {
            a[j] += b[k] * c[k][j];
        }
    }

    for (auto i = x.size()- count; i < x.size(); i++) {
      double sum = 0;
      for (auto j = a.size()-1; j < -1; j--){
        sum = sum* x[i];
        sum = sum + a[j];
      }
    result.emplace_back(sum);
    }
    return result;
  }

  void Coin::Show(){

     std::cout << "Coin String==" << m_name <<std::endl;
  }

  void Coin::Initlog() {

    boost::log::add_file_log
      (
        boost::log::keywords::file_name = "sample_%N.log",                                                                /*< название лог файла >*/
        boost::log::keywords::rotation_size = 10 * 1024 * 1024,                                                           /*< перезапись каждые 10 Мб... >*/
        boost::log::keywords::time_based_rotation = boost::log::sinks::file::rotation_at_time_point(0, 0, 0),             /*< ...или в 00:00 >*/
        boost::log::keywords::format = "%LineID%   <%Severity%> [%TimeStamp%]: %Message%" ,                                /*< формат >*/
        boost::log::keywords::open_mode =  (std::ios::out | std::ios_base::app),
        boost::log::keywords::auto_flush = true
    );
  }

  void Coin::WebSocketLoggingError (std::string str) {
   BOOST_LOG_TRIVIAL(error) << str;
  }

  void Coin::WebSocketLoggingTrace (std::string str) {
   BOOST_LOG_TRIVIAL(trace) << str;
  }

  void Coin::GetOrder50Value(){
    m_bid50_value = 0;
    m_ask50_value = 0;
    int b = 0;
    int a = 0;

    for (const auto& element : m_bid50) {
        m_bid50_value = m_bid50_value + element.second;
          if (b == 0) {
              bid50_0_price = element.first;
              bid50_0_value = element.second;
          }

          if (b == 1) {
              bid50_1_price = element.first;
              bid50_1_value = element.second;
          }
    b++;
    }

    for (const auto& element : m_ask50) {
        m_ask50_value = m_ask50_value + element.second;
          if (a == 0) {
              ask50_0_price = element.first;
              ask50_0_value = element.second;
          }

          if (a == 1) {
              ask50_1_price = element.first;
              ask50_1_value = element.second;
          }
      a++;
    }
  }

  void Coin::ParsePublicStream(beast::flat_buffer buffer){
    json::object req_obj;
    json::value  req;
    std::string firstkey;
    std::string firstvalue;
    std::string type;
    std::array<std::string,2> ss;
    std::array< long double,2> ss1;
   //   std::cout << beast::make_printable(buffer.data()) << std::endl;

      // для оценки времени выполнения------------------------
      using boost::chrono::duration_cast;
      using boost::chrono::microseconds;
      typedef boost::chrono::high_resolution_clock clock;
      typedef boost::chrono::time_point <clock> time_point;
      time_point tp_start, tp_end;
      long long duration = 0;
      uint16_t temp =0;
//----------------------------------JSON Parsing----------------------------
    auto defualt_precision{std::cout.precision()};
    setlocale(LC_NUMERIC , "C");
    tp_start = clock::now();
    std::string str = beast::buffers_to_string(buffer.data());
    boost::system::error_code errcode;
    req = json::parse(str, errcode ).as_object();
    req_obj = json::parse(str, errcode ).as_object();

    if (errcode) std::cout << "Parsing JSON failed: " << errcode.message() << std::endl;

    auto const& obj = req.get_object();
    auto it = obj.begin();
    firstkey = json::serialize(it->key());
      if (firstkey == R"("topic")") {
          firstvalue = json::serialize(it -> value());
            if (firstvalue == R"("orderbook.50.)" + m_name +R"(")") {
                m_bid50_hyst.push_back(m_bid50);
                m_ask50_hyst.push_back(m_ask50);
                auto& data_obj = req_obj["data"].as_object();
                m_ts50 = req_obj["ts"].as_int64();
                m_cts50 = req_obj["cts"].as_int64();
                m_u50 = data_obj["u"].as_int64();
                m_seq50 = data_obj["seq"].as_int64();
                type =  req_obj["type"].as_string();
                  if ( type == "snapshot" ) {
                      m_bid50.clear();
                      m_ask50.clear();

                            for (auto&bid:data_obj["b"].as_array()) {
                                ss[0] = bid.at(0).as_string();
                                ss[1] = bid.at(1).as_string();
                                ss1[0] = std::stold(std::move(ss[0]));
                                ss1[1] = std::stold(std::move(ss[1]));
                                m_bid50[ss1[0]] = ss1[1];
                            }

                            for (auto&ask:data_obj["a"].as_array()) {
                                ss[0] = ask.at(0).as_string();
                                ss[1] = ask.at(1).as_string();
                                ss1[0] = std::stold(std::move(ss[0]));
                                ss1[1] = std::stold(std::move(ss[1]));
                                m_ask50[ss1[0]] = ss1[1];
                            }
                        }

                        if ( type == "delta" ) {

                            for (auto&bid:data_obj["b"].as_array()) {
                                ss[0] = bid.at(0).as_string();
                                ss[1] = bid.at(1).as_string();
                                ss1[0] = std::stold(std::move(ss[0]));
                                ss1[1] = std::stold(std::move(ss[1]));
                                  if (ss1[1] !=0 ) m_bid50[ss1[0]] = ss1[1]; else m_bid50.erase(ss1[0]);
                            }

                            for (auto&ask:data_obj["a"].as_array()) {
                                ss[0] = ask.at(0).as_string();
                                ss[1] = ask.at(1).as_string();
                                ss1[0] = std::stold(std::move(ss[0]));
                                ss1[1] = std::stold(std::move(ss[1]));
                                  if (ss1[1] !=0 ) m_ask50[ss1[0]] = ss1[1]; else m_ask50.erase(ss1[0]);
                            }

                        }
    GetOrder50Value();
/*
                               tp_end = clock::now();
                               duration = duration_cast <microseconds> (tp_end - tp_start).count();
                               std::cout << "Duration = " << duration << std::endl;


                               for (const auto& element:test) {
                                   std::cout << "bid[0]== " << std::fixed << std::setprecision(6) << element.first << "bid[1]== " << element.second << std::endl;
                                }

                                   for (const auto& element:test1) {
                                   std::cout << "ask[0]== " << std::fixed << std::setprecision(6) << element.first << "ask[1]== " << element.second << std::endl;
                                }

                                std::cout << "Value bid=  "  << m_bid50_value << std::endl;
                                std::cout << "Value ask=  "  << m_ask50_value << std::endl;
                                std::cout << "Bid0=  "  << bid50_0_price << " " << bid50_0_value << std::endl;
                                std::cout << "Bid1=  "  << bid50_1_price << " " << bid50_1_value << std::endl;
                                std::cout << "Ask0=  "  << ask50_0_price << " " << ask50_0_value << std::endl;
                                std::cout << "ASk1=  "  << ask50_1_price << " " << ask50_1_value << std::endl;
                                std::cout << std::setprecision(defualt_precision);
*/
            }
            if (firstvalue == R"("publicTrade.)" + m_name +R"(")") {
//    std::cout << "Parse string  =  " << str << std::endl;

              for (auto&ask:req_obj["data"].as_array()) {
                  auto PublicTrade = ask.as_object();
                  m_tsPublicTrade = PublicTrade["T"].as_int64();
                  auto PricePublicTrade = PublicTrade["p"].as_string();
                  auto ValuePublicTrade = PublicTrade["v"].as_string();
                  m_SideTaker = PublicTrade["S"].as_string();
                  m_BTPublicTrade = PublicTrade["BT"].as_bool();
                  m_ValuePublicTrade = std::stold(static_cast<std::string>(ValuePublicTrade) );
                  m_PricePublicTrade = std::stold(static_cast<std::string>(PricePublicTrade) );
/*
    m_write_to_file = static_cast<std::string> (PricePublicTrade) +  "   " + static_cast<std::string>(ValuePublicTrade) + "   " + boost::lexical_cast<std::string>(m_tsPublicTrade) + "   " + static_cast<std::string>(m_SideTaker) +  "   " + std::to_string(m_ask50_value) + "   " + std::to_string(m_bid50_value) + "   " +
    std::to_string(bid50_0_price) +  "   " + std::to_string(bid50_0_value) +  "   " + std::to_string(bid50_1_price) +  "   " + std::to_string(bid50_1_value)+  "   " +
    std::to_string(ask50_0_price) +  "   " + std::to_string(ask50_0_value) +  "   " + std::to_string(ask50_1_price) +  "   " + std::to_string(ask50_1_value)+"\n";
*/
                  if (m_ValuePublicTrade > 0.0015) {//фильтр минимального обЪема
                      m_SideTaker_Str = static_cast<std::string>(m_SideTaker);
                        if ((m_Price_Old != m_Price_Old_Old) && (m_Price_Old != m_PricePublicTrade) && (m_Value_Temp == 0)) {//если цена изменилась
                            temp = 1;
                            m_ValuePublicToWrite =  m_Value_Old;
                              if ( m_SideTaker_Str == "Buy") m_Value_SellBuy_ToWrite = m_ValuePublicToWrite; else m_Value_SellBuy_ToWrite = ( 0 - m_ValuePublicToWrite );
                        }

                        if (m_Price_Old == m_PricePublicTrade) {//если цена не изменилась складываем обЪем
                            m_Value_Temp = m_Value_Temp +   m_Value_Old ;

                        }

                        if ((m_Price_Old == m_Price_Old_Old) && (m_Price_Old != m_PricePublicTrade) && (m_Value_Temp != 0)) {//если цена изменилась
                            m_Value_Temp = m_Value_Temp +   m_Value_Old ;
                            temp = 1;
                            m_ValuePublicToWrite =  m_Value_Temp;
                        if ( m_SideTaker_Str == "Buy") m_Value_SellBuy_ToWrite = m_ValuePublicToWrite; else m_Value_SellBuy_ToWrite = ( 0 - m_ValuePublicToWrite );
                            m_Value_Temp = 0;
                  }


        if ( temp == 1) {//есть новые данные для анализа
            m_Value_Accumulate_ToWrite = m_Value_Accumulate_ToWrite + m_Value_SellBuy_ToWrite * m_Price_Old;//накопительная цена*обЪем
            m_PriceVecSmall.emplace_back(m_Price_Old);
            m_ValueVecSmall.emplace_back(m_ValuePublicToWrite);
            m_Value_SellBuy_VecSmall.emplace_back(m_Value_SellBuy_ToWrite);
            m_Value_Accumulate_VecSmall.emplace_back(m_Value_Accumulate_ToWrite);

            //формирование векторов данных для анализа
              if (m_PriceVecSmall.size() >= LAG_REGRESS) {
                  if (m_PricePublic.size() != ITEM_REGRESS) {
                      m_PricePublic.insert(m_PricePublic.end(), m_PriceVecSmall.begin(), m_PriceVecSmall.end());
                      m_ValuePublic.insert( m_ValuePublic.end(), m_ValueVecSmall.begin(), m_ValueVecSmall.end());
                      m_Value_SellBuy.insert( m_Value_SellBuy.end(), m_Value_SellBuy_VecSmall.begin(), m_Value_SellBuy_VecSmall.end());
                      m_Value_Accumulate.insert( m_Value_Accumulate.end(), m_Value_Accumulate_VecSmall.begin(), m_Value_Accumulate_VecSmall.end());
                  }
                  else {
                      m_PricePublicRegress = Regress(m_time, m_PricePublic, LAG_REGRESS, 4);
                      m_write_to_file_2.clear();

                      for (auto i = 0; i <LAG_REGRESS; i++) {
                          m_count++;
                          m_write_to_file_2 = m_write_to_file_2 + std::to_string(m_count) +  "   "  + std::to_string(m_PricePublic[ITEM_REGRESS - LAG_REGRESS + i]) +  "   " + std::to_string(m_PricePublicRegress[i]) +   "   " +std::to_string( m_ValuePublic[ITEM_REGRESS - LAG_REGRESS + i]) + "   " + std::to_string( m_Value_SellBuy[ITEM_REGRESS - LAG_REGRESS + i]) + "   " + std::to_string( m_Value_Accumulate[ITEM_REGRESS - LAG_REGRESS + i]) + "\n";
                          }

                      std::shift_left(m_PricePublic.begin(), m_PricePublic.end(), LAG_REGRESS);
                      m_PricePublic.erase(m_PricePublic.begin()+(ITEM_REGRESS-LAG_REGRESS-1), m_PricePublic.end()-1);
                      m_PricePublic.insert(m_PricePublic.end(), m_PriceVecSmall.begin(), m_PriceVecSmall.end());

                      std::shift_left(m_ValuePublic.begin(), m_ValuePublic.end(), LAG_REGRESS);
                      m_ValuePublic.erase(m_ValuePublic.begin()+(ITEM_REGRESS-LAG_REGRESS-1), m_ValuePublic.end()-1);
                      m_ValuePublic.insert(m_ValuePublic.end(), m_ValueVecSmall.begin(), m_ValueVecSmall.end());

                      std::shift_left(m_Value_SellBuy.begin(), m_Value_SellBuy.end(), LAG_REGRESS);
                      m_Value_SellBuy.erase(m_Value_SellBuy.begin()+(ITEM_REGRESS-LAG_REGRESS-1), m_Value_SellBuy.end()-1);
                      m_Value_SellBuy.insert(m_Value_SellBuy.end(), m_Value_SellBuy_VecSmall.begin(), m_Value_SellBuy_VecSmall.end());

                      std::shift_left(m_Value_Accumulate.begin(), m_Value_Accumulate.end(), LAG_REGRESS);
                      m_Value_Accumulate.erase(m_Value_Accumulate.begin()+(ITEM_REGRESS-LAG_REGRESS-1), m_Value_Accumulate.end()-1);
                      m_Value_Accumulate.insert(m_Value_Accumulate.end(), m_Value_Accumulate_VecSmall.begin(), m_Value_Accumulate_VecSmall.end());


          // Указатель на следующее готовое SQE в SQ.
                      for (int i = 0; i < MAX_NUMBER_FILES; ++i) {
                          m_sqe[i] = io_uring_get_sqe(&m_IO_uring); //для каждого файла свой
                      }
          // Добавляем операцию write в очередь SQ.
                      io_uring_prep_write(m_sqe[0], outf[0], m_write_to_file_2.c_str(), m_write_to_file_2.length(), 0);
          // Сообщаем io_uring о новых SQE в SQ.
                      io_uring_submit(&m_IO_uring);
          // Ждем пока в CQ появится новое CQE.
                      for (int i = 0; i < MAX_NUMBER_FILES; ++i) {
                          auto ret = io_uring_wait_cqe(&m_IO_uring, &m_cqe[i]);
                            if (ret < 0) std::cout << "io_uring_wait_cqe error... "  << std::endl;
                      }

          // Помечаем CQE как прочитанное (удаляем из очереди CQ).
                      for (int i = 0; i < MAX_NUMBER_FILES; ++i) {
                          io_uring_cqe_seen(&m_IO_uring, m_cqe[i]);
                      }
                  }
              m_PriceVecSmall.erase(m_PriceVecSmall.begin(), m_PriceVecSmall.end());
              m_ValueVecSmall.erase(m_ValueVecSmall.begin(), m_ValueVecSmall.end());
              m_Value_SellBuy_VecSmall.erase(m_Value_SellBuy_VecSmall.begin(), m_Value_SellBuy_VecSmall.end());
              m_Value_Accumulate_VecSmall.erase(m_Value_Accumulate_VecSmall.begin(), m_Value_Accumulate_VecSmall.end());
            }
          temp = 0;

                              tp_end = clock::now();
                               duration = duration_cast <microseconds> (tp_end - tp_start).count();
                               std::cout << "Duration = " << duration << std::endl;
        }
        m_Value_Old_Old = m_Value_Old;
        m_Price_Old_Old = m_Price_Old;
        m_Price_Old =  m_PricePublicTrade;
        m_Value_Old = m_ValuePublicTrade ;
                  }
                    /*для быстродействия
                    std::cout << "m_ValuePublicTrade = " << m_ValuePublicTrade << std::endl;
                    std::cout << "m_PricePublicTrade = " << m_PricePublicTrade << std::endl;
                    std::cout << "m_SideTicker = " << m_SideTaker << std::endl;
                    std::cout << "m_BT = " << m_BTPublicTrade << std::endl;
                    std::cout << "m_tsPublicTrade = " << m_tsPublicTrade << std::endl;
                    */
              }
            }
      }
}

  void Coin::AddPublicStream(){

    m_websocketstream.SetPtr(shared_from_this());
    m_subscription = R"({"op": "subscribe", "args": ["orderbook.50.)" + m_name + R"("]})";
    m_websocketstream.AddPublicStream(m_host, m_streamspot, m_subscription);
    m_subscription = R"({"op": "subscribe", "args": ["publicTrade.)" + m_name + R"("]})";
    m_websocketstream.AddPublicStream(m_host, m_streamspot, m_subscription);
    WebSocketLoggingTrace ("AddPublicStream function running...");
}

  void Coin::StopWebSocket(){
    WebSocketLoggingTrace ("StopWebSocket function running...");
    m_websocketstream.StopWebSocket();
}