Add: Incremental Books

For new exchanges, TAIFEX and CBOE

src/DataStreamSuite/DataStream/Primitives/Misc/PubsubEchoTestStream.cs

@@ -158,6 +158,10 @@ namespace bts
          myFireEvent(dp_out);
       }
 
+      private string _get_next_id() {
+         return _table.Count.ToString();
+      }
+
       private string _get_next_id()
       {
          return _table.Count.ToString();

src/DataStreamSuite/DataStreamExpressions/DataStreamExpressionTokens.lex

@@ -9,15 +9,7 @@
 Letter                   [a-zA-Z]
 BooleanConstant          'true'|'false'
 CSStringConstant         \"([^\"\\]*|\\.)*\"
-YearInt                  [0-9]{4}
-MonthInt                 [0-9]{1,2}
-DayInt                   [0-9]{1,2}
-/**
- * DateTimeConstant supports American (mm/dd/yyyy) and European (yyyy/mm/dd)
- * formats. Month and day can be 1 or 2 digits. It does not validate that the
- * numbers represent an actual date.
- */
-DateTimeConstant         ({MonthInt}\/{DayInt}\/{YearInt}|{YearInt}\/{MonthInt}\/{DayInt})
+DateTimeConstant         ([1-9]|'10'|'11'|'12')\/([1-9]|([1-2][0-9])|'30'|'31')\/[0-9][0-9][0-9]*
 IntegerConstant          -?[0-9]+
 NullConstant             'null'
 RealConstant             -?[0-9]+\.[0-9]+

src/DataStreamSuite/PubSub/Config/PubSubConfig.cs

@@ -117,10 +117,10 @@ public class PubSubConfig {
          WaitPeriodAfterGc = 1000,
          SimpleEchoServer = false,
          PublisherRoutes = new() {
-            new("bluto://127.0.0.1:12345", ".*")
+            new("bts-cloud://192.168.1.80:12345", ".*")
          },
          SubscriberRoutes = new() {
-            new("bluto://127.0.0.1:12345", ".*")
+            new("bts-cloud://192.168.1.80:12345", ".*")
          }
       };
       return rv;

src/lib/api/md/DatabentoMarketDataConfig.proto

@@ -13,5 +13,4 @@ message DatabentoMarketDataConfig {
   double replay_sleep_timer_sec = 50;
   string key = 60;
   string dataset = 70;
-  bool filter_securities_one_year= 80;
 }

src/lib/api/md/TaifexMarketDataConfig.proto

@@ -5,24 +5,19 @@ import "lib/api/domain/SecurityCriteria.proto";
 package bts.api.md;
 
 option csharp_namespace = "bts.api.md";
-message TaifexConnectionDetails{
-  string live_data_ip = 30;
-  string reference_data_ip = 31;
-  string refresh_data_ip = 32;
-  string live_data_port = 40;
-  string reference_data_port = 41;
-  string refresh_data_port = 42;
-  string interface_ip = 50;
-}
+
 
 message TaifexMarketDataConfig {
   uint32 unique_check_value = 1;
   bool read_from_file = 20;
-  string ref_data_file_name = 21;
-  string realtime_data_file_name = 22;
-  TaifexConnectionDetails regular_hours_connection_details = 30;
-  TaifexConnectionDetails after_hours_connection_details = 31;
-  bool use_after_hours = 40;
+  //uint32 num_depth_levels = 1;
+  //uint32 depth_throttle_ms = 2;
   bts.api.domain.SecurityCriteria security_criteria = 3;
-  
+  // Market data isn't subscribed for securities listed here (even if they are
+  // included by 'security_criteria'). There are some securities that don't
+  // fit into our system just yet, such as 'BTCBIDR', which trades in prices
+  // larger than our cpp 'domain::Price' class supports.
+  //repeated string excluded_securities = 4;
+  //string url_rest_base = 5;
+  //string url_ws_base = 6;
 }

src/lib/api/service_helpers/AdminServiceUtils.cpp

@@ -156,8 +156,6 @@ AdminServiceUtils::resolve_connection(
    const bts::api::admin::ExchangeConnection *rv = nullptr;
    switch (sec.exchange()) {
    case api::Venue::VENUE_GLBX: {
-      //CME EXECUTION NEEDS TO BE UPDATED, probably merge with VENUE_BINANCE_USDS_FUTURES
-      
       auto deployment_type =
           boost::to_lower_copy(config::get<config::DeploymentType>().type());
       rv = resolve_cme_connection(data, account, sec);
@@ -229,7 +227,6 @@ void AdminServiceUtils::fill_connections_from_config(
       connection.set_name(c.name());
 
       switch (c.venue()) {
-      //CME EXECUTION NEEDS TO BE UPDATED, probably merge default case
       case bts::api::VENUE_GLBX:
          if (c.cme_market_segment_id() == 0)
             connection.mutable_cme_connection()->mutable_cgw_connection();
@@ -239,7 +236,9 @@ void AdminServiceUtils::fill_connections_from_config(
                 ->set_market_segment_id(c.cme_market_segment_id());
          add_connection(admin_data, connection);
          break;
-      case bts::api::VENUE_TAIFEX:
+      case bts::api::VENUE_CBOE:
+         break;
+         case bts::api::VENUE_TAIFEX:
          break;
       case bts::api::VENUE_SIM:
       case bts::api::VENUE_SIMICE:
@@ -251,7 +250,6 @@ void AdminServiceUtils::fill_connections_from_config(
              c.sender_compid());
          add_connection(admin_data, connection);
          break;
-      case bts::api::VENUE_CBOE:
       default:
          connection.mutable_general_connection()->set_exchange(c.venue());
          add_connection(admin_data, connection);
@@ -573,13 +571,6 @@ bts::api::orderman::RoutingRecord AdminServiceUtils::get_routing_record(
       return rv;
    }
    switch (sec.exchange()) {
-   case api::Venue::VENUE_CBOE:{
-       rv.set_session("cboe_1");
-      rv.set_account(account_ptr->name());
-      rv.set_operator_id(user_ptr->name());
-      out_result.set_is_success(true);
-      return rv;
-   }
    case api::Venue::VENUE_DERIBIT: {
       rv.set_session("deribit_1");
       rv.set_account(account_ptr->name());
@@ -750,6 +741,7 @@ bts::api::orderman::RoutingRecord AdminServiceUtils::get_routing_record(
    case api::VENUE_OKX:
    case api::Venue::VENUE_UNSET:
    case api::Venue::VENUE_BTS:
+   case api::Venue::VENUE_CBOE:
    case api::Venue::VENUE_TAIFEX:
    case api::Venue::VENUE_CBOECFE:
    case api::Venue::Venue_INT_MIN_SENTINEL_DO_NOT_USE_:

src/lib/api/services/OrderService.proto

@@ -409,7 +409,6 @@ message RoutingRecord {
 
 message NativeOrderState {
   string exchange_order_id = 1;
-  string external_order_id = 11;  //"string : orderkey"
   uint64 last_msg_seq_num = 4;
   enum SubError {
     SUB_ERROR_NONE = 0;
@@ -753,7 +752,6 @@ message RetrievedExchangeOpenOrder{
   string exchange_sec_desc = 10;
   string exchange_account = 20;
   string exchange_order_id = 30;
-  string external_order_id = 11;  //"string : orderkey" 
   uint32 security_key = 40;  
   bool security_resolved = 50;
   string account_name = 60;

src/lib/config/profiles/default/onix_fix_cfg.conf

@@ -1,12 +0,0 @@
-OnixFixCfg {
-    value{
-        name: "trafix",
-        msg_type: "Order_Single",
-        protocol_version: "FIX_42",
-        handl_inst: 2,
-        sender_sub_id: "BTS",
-        target_sub_id: "TRFXMRX",
-        client_id: "BTS",
-        rule80a: 1
-    }
-}

src/lib/config/protos/onix_fix_cfg.proto

@@ -1,23 +0,0 @@
-syntax = "proto3";
-package config;
-message OnixFixCfg {
-    enum HandlInst {
-        UNDEFINED = 0;
-        DMA = 1;
-        ALGO = 2;
-        CARE = 3;
-    }
-    enum Rule80A {
-        UNUSED = 0;
-        AGENCY = 1;
-        PRINCIPAL = 2;
-    }
-    string name = 9;
-    string msg_type = 1;
-    string protocol_version = 2;
-    HandlInst handl_inst = 3;
-    string sender_sub_id = 4;
-    string target_sub_id = 5;
-    string client_id = 6;
-    Rule80A rule80a = 7;
-}

src/lib/domain/SecurityCommon.h

@@ -40,6 +40,7 @@ class SecurityCommon {
    double min_quantity_increment = 0;
    double min_quantity = 0;
    double max_quantity = std::numeric_limits<double>::max();
+   double cpp = 1;
 
    SecurityCommon() {}
    SecurityCommon(const bts::api::Security &s)
@@ -47,7 +48,8 @@ class SecurityCommon {
          exchange_sec_id(DomainSecurityUtils::get_exchange_id(s)),
          id(SecurityID(s.id())), currency(s.currency()), symbol(s.symbol()),
          desc(exchange_sec_id),
-         min_price_increment(s.contract_spec().low_tick()) {}
+         min_price_increment(s.contract_spec().low_tick()),
+         cpp(s.contract_spec().cash_per_point()) {}
 };
 
 void from_json(const nlohmann::json &j, SecurityCommon &sc);
@@ -74,6 +76,7 @@ inline void from_json(const nlohmann::json &j, SecurityCommon &sc) {
        jr.read("min_price_increment", sc.min_price_increment) &&
        jr.read("min_quantity_increment", sc.min_quantity_increment) &&
        jr.read("min_quantity", sc.min_quantity) &&
+       jr.read("cpp", sc.cpp) &&
        jr.read("max_quantity", sc.max_quantity);
    // clang-format on
    if (!success) {
@@ -98,6 +101,7 @@ inline void to_json(nlohmann::json &j, const SecurityCommon &sc) {
        {"min_quantity_increment", sc.min_quantity_increment},
        {"min_quantity", sc.min_quantity},
        {"max_quantity", sc.max_quantity},
+       {"cpp", sc.cpp},
    };
 }
 

src/lib/md/GeneralMarketDataListener.cpp

@@ -78,78 +78,73 @@ struct IncrementalBookData {
    std::map<domain::SecurityID, domain::Book> book_map;
 
    bool update_incremental(const IMarketDataListener::BookOp & op) {
-    // Retrieve the Book from the map and update it
-    domain::Book* book = get_book(op.id);
-    if (book == nullptr)
-        return false;
+      // TODO: Arham/Adeel, implement this
+      // retreive the Book from the map and udpate it. domain::Book has a lot of functions
+      // already, add more if you want to. but the idea is to update the book in place, ie
+      // if a level needs to be added push the lower levels out
 
-    // Choosing the correct side of the book (bids or asks)
-    auto& book_side = (op.side == domain::Side::Buy) ? book->bids() : book->asks();
-    std::string side_str = (op.side == domain::Side::Buy) ? "Buy" : "Sell";
+      //getting book
+      domain::Book* book = get_book(op.id);
+      if(book == nullptr)
+         return false;
+      
+      //checking if side is Buy or Sell
+      auto& book_side = (op.side == api::Side::SIDE_BUY) ? book->bids() : book->asks();
 
-    // Capturing the book state before the update
-    std::stringstream book_before;
-    book_before << "Book before update (" << side_str << " side):\n";
-    for (const auto& level : book_side) {
-        book_before << "Price: " << level.price << ", Quantity: " << level.direct_quantity << "\n";
-    }
+      size_t insert_level = 0;
+      bool level_found = false;
 
-    size_t insert_level = 0;
-    bool level_found = false;
-
-    // Logic to find where to insert or update the level
-    if (op.side == domain::Side::Buy) {
-        while (insert_level < book_side.size()) {
-            if (op.price == book_side[insert_level].price) {
-                level_found = true;
-                break;
+      if(op.side == api::Side::SIDE_BUY)
+      {
+         //going through book
+         while(insert_level < book_side.size())
+         {
+            //op.price_size?
+            if(op.price == book_side[insert_level].price)
+            {
+               level_found = true;
+               break;
             }
-            if (op.price > book_side[insert_level].price) {
-                break;
+            //if its not equal and its not larger, we go to next level
+            if(op.price > book_side[insert_level].price)
+            {
+               break;
+            }
+            //moving to next level 
+            insert_level++;
+         }
+      }else{
+         while(insert_level < book_side.size())
+         {
+            if(op.price == book_side[insert_level].price)
+            {
+               level_found = true;
+               break;
+            }
+            if (op.price < book_side[insert_level].price)
+            {
+               break;
             }
             insert_level++;
+         }
+      }
+      
+      //inserting and pushing levels
+      if(!level_found)
+      {
+         // shift all lower levels down
+        for (size_t i = book_side.size() - 1; i > insert_level; i--) 
+        {
+            book_side[i] = book_side[i - 1];
         }
-    } else {
-        while (insert_level < book_side.size()) {
-            if (op.price == book_side[insert_level].price) {
-                level_found = true;
-                break;
-            }
-            if (op.price < book_side[insert_level].price) {
-                break;
-            }
-            insert_level++;
-        }
-    }
-
-    // Insert or update the book level
-    if (!level_found) {
-        book_side.insert(book_side.begin() + insert_level, {op.price, op.quantity});
-    } else {
-        book_side[insert_level].direct_quantity = op.quantity;
-    }
-
-    // Capturing the book state after the update
-    std::stringstream book_after;
-    book_after << "Book after update (" << side_str << " side):\n";
-    for (const auto& level : book_side) {
-        book_after << "Price: " << level.price << ", Quantity: " << level.direct_quantity << "\n";
-    }
-
-    // Capturing the operation details
-    std::stringstream op_details;
-    op_details << "Operation: Price: " << op.price << ", Quantity: " << op.quantity 
-               << ", Side: " << side_str << "\n";
-
-    // Combine the before/after and operation details into one string
-    std::string book_before_after_op = book_before.str() + "\n" + book_after.str() + "\n" + op_details.str();
-
-    // Assertion to check the book is sorted
-    bts_assert_msgf(book->is_sorted(), "book not sorted {}", book_before_after_op);
-
-    return true;
-}
+        // insert the new level with its price
+        //book_side[insert_level].price.to_double() = op.price_size.price();
 
+      }
+        book_side[insert_level].price =op.price;
+        book_side[insert_level].direct_quantity =op.quantity;
+      return true;
+   }
    domain::Book const *get_const_book(domain::SecurityID id) const {
       auto it = book_map.find(id);
       if (it == book_map.end()) {

src/lib/md/IMarketDataListener.h

@@ -24,7 +24,7 @@ class IMarketDataListener {
    enum BookOpType { Add = 0, MODIFY, REMOVE  };
    struct BookOp {
       domain::SecurityID id;
-      domain::Side side;
+      bts::api::Side side;
       BookOpType op_type;
       // for op == REMOVE size in price_size is no-op
       domain::Price price;

src/lib/orderman/events/OrderStateUpdateEvent.h

@@ -60,7 +60,7 @@ class OrderStateUpdateEvent : public mercury::Event<OrderStateUpdateEvent> {
       } else {
          auto native_state = get_native_state(state);
          if (native_state) {
-            exchange_order_id = native_state->exchange_order_id(); //use this 
+            exchange_order_id = native_state->exchange_order_id();
             exchange_order_id_u64 = native_state->exchange_order_id_u64();
             last_msg_seq_num = native_state->last_msg_seq_num();
             sub_error_code = native_state->sub_error_code();

src/lib/orderman/gateways/ExecutionGatewayFactory.cpp

@@ -4,12 +4,9 @@
 #include "lib/orderman/gateways/CCAPIExecutionGateway.h"
 // #include "lib/orderman/gateways/EMExecutionGateway.h"
 #include "lib/orderman/gateways/FakeExecutionGateway.h"
-#include "lib/orderman/gateways/GeneralExecutionGateway.h"
 #include "lib/orderman/gateways/GLBXExecutionGateway.h"
 #include "lib/orderman/gateways/GLBXiLink3ExecutionGateway.h"
 #include "lib/orderman/gateways/ICEExecutionGateway.h"
-#include "lib/orderman/gateways/IFIXGenericAdapter.h"
-#include "lib/orderman/gateways/OnixFIXAdapter.h"
 #include "lib/orderman/gateways/sim/SimExecutionGatewayFactory.h"
 #include "lib/orderman/gateways/sim/SimExecutionGatewayUniqueWrapper.h"
 #include "lib/tracing/PerfLog.h"
@@ -24,30 +21,23 @@ std::unique_ptr<IExecutionGateway> ExecutionGatewayFactory::create_gateway(
     const config::JetFIX::Connection &conn,
     trace::PerfLog &pl,
     md::SecurityOracle o) {
-   LOG(OrderMan, Info, "ExecutionGatewayFactory::create_gateway conn: {}",
-       utils::pb::to_string(conn));
    if (SimExecutionGatewayFactory::is_sim_execution_on(conn.venue())) {
       return std::make_unique<SimExecutionGatewayUniqueWrapper>(conn.venue());
    }
-   switch(conn.venue()){
-      case bts::api::VENUE_ICE:
-      case bts::api::VENUE_SIMICE:
-         return std::make_unique<ICEExecutionGateway>(conn, pl, o);
-      case bts::api::VENUE_GLBX:{
-         if (conn.use_ilink3())
-            return std::make_unique<ilink3::GLBXiLink3ExecutionGateway>( conn, pl, o);
-         return std::make_unique<GLBXExecutionGateway>(conn, pl, o);
-      }
-      case bts::api::VENUE_DERIBIT:
-         return std::make_unique<CCAPIExecutionGateway>(conn.venue(), o);
-      case bts::api::VENUE_CBOE:
-         {
-             std::unique_ptr<IFIXGenericAdapter> adapter = std::make_unique<OnixFIXAdapter>();
-            return std::make_unique<GeneralExecutionGateway>(conn.venue(),std::move(adapter),o);
-         }
-      default:
-         return std::make_unique<FakeExecutionGateway>(o);
-   }
+   LOG(OrderMan, Info, "ExecutionGatewayFactory::create_gateway conn: {}",
+       utils::pb::to_string(conn));
+   if (conn.venue() == bts::api::VENUE_ICE ||
+       conn.venue() == bts::api::VENUE_SIMICE)
+      return std::make_unique<ICEExecutionGateway>(conn, pl, o);
+   else if (conn.venue() == bts::api::VENUE_GLBX) {
+      if (conn.use_ilink3())
+         return std::make_unique<ilink3::GLBXiLink3ExecutionGateway>(
+             conn, pl, o);
+      return std::make_unique<GLBXExecutionGateway>(conn, pl, o);
+   } else if (conn.venue() == bts::api::VENUE_DERIBIT)
+      //   return std::make_unique<EMExecutionGateway>(o);
+      return std::make_unique<CCAPIExecutionGateway>(conn.venue(), o);
+   return std::make_unique<FakeExecutionGateway>(o);
 }
 
 } // namespace bts::orderman

src/lib/orderman/gateways/GeneralExecutionGateway.cpp

@@ -1,85 +0,0 @@
-#include "GeneralExecutionGateway.h"
-#include "lib/domain_model/Security.pb.h"
-#include "lib/logging/logging.h"
-#include "lib/orderman/CreateOrderMsg.h"
-#include "lib/orderman/OrderMsg.h"
-#include "lib/orderman/gateways/IExecutionGateway.h"
-#include "lib/orderman/orderman_utils/CancelReject.h"
-#include <memory>
-#include <type_traits>
-namespace bts::orderman {
-struct GeneralExecutionGateway::PImpl {
-   PImpl(bts::api::Venue venue,
-   std::unique_ptr<IFIXGenericAdapter> adapter,
-   md::SecurityOracle o)
-        : venue(venue), _adapter(std::move(adapter)), sec_oracle(o) {}
-   bts::api::Venue venue;
-   std::unique_ptr<IFIXGenericAdapter> _adapter;
-   md::SecurityOracle sec_oracle;
-   OnExeReport on_exe_report;
-   OnGwStatusChange on_gw_status_change;
-   OnCancelReject on_cancel_reject;
-   OnExchangePositionState on_exchange_position_state;
-   OnExchangeOpenOrdersState on_exchange_open_orders;
-};
-OnGwStatusChange &GeneralExecutionGateway::on_gw_status_change() {
-   return me->on_gw_status_change;
-}
-// Modified constructor to accept the adapter from outside
-GeneralExecutionGateway::GeneralExecutionGateway(bts::api::Venue venue, 
-                                                 std::unique_ptr<IFIXGenericAdapter> adapter, md::SecurityOracle o)
-    : IExecutionGateway(), me(new PImpl(venue, std::move(adapter),o)) {
-    LOG(OrderMan, Info, "ExecutionGateway::ExecutionGateway constructor");
-}
-GeneralExecutionGateway::~GeneralExecutionGateway() {}
-void GeneralExecutionGateway::start() { 
-   me->on_gw_status_change(true);
-   me->_adapter->start(); 
-   me->_adapter->register_callback(me->on_exe_report);
-   LOG(OrderMan,Info,"ExecutionGateway::start started" );
-}
-void GeneralExecutionGateway::send_msgs(
-    std::vector<OrderMsg> const &msgs, trace::CorID) {
-   LOG(OrderMan,Info,"ExecutionGateway::send_msgs" );
-   for (const auto &msg : msgs) {
-      std::visit(
-          [this](auto &&order) {
-             using T = std::decay_t<decltype(order)>;
-             if constexpr (std::is_same_v<T, CreateOrderMsg>) {
-                LOG(OrderMan,Warn,"ExecutionGateway::CreateOrderMsg" );
-                me->_adapter->launch_new_order(order);
-             
-             } else if constexpr (std::is_same_v<T, CancelOrderMsg>) {
-                LOG(OrderMan,Warn,"ExecutionGateway::CancelOrderMsg" );
-                me->_adapter->cancel_order(order);
-             } else if constexpr (std::is_same_v<T, CancelReplaceOrderMsg>) {
-                LOG(OrderMan,Warn,"ExecutionGateway::CancelReplaceOrderMsg" );
-                me->_adapter->modify_order(order);
-             } else if constexpr (std::is_same_v<T, RFQMsg>) {
-                 // Handle RFQMsg if necessary
-                 // adapter->handle_rfq(order); // Example placeholder
-             }
-          },
-          msg);
-   }
-}
-void GeneralExecutionGateway::dispatch_incoming_fix_messages() {}
-void GeneralExecutionGateway::on_securities(
-    const mercury::Update<NewSecuritiesEvent> &) {}
-void GeneralExecutionGateway::on_admin_data(
-    const mercury::Update<NewAdminDataEvent> &) {}
-void GeneralExecutionGateway::send_fix(const jetfix::OutFIXPtr &, trace::CorID) {}
-OnExeReport &GeneralExecutionGateway::on_exe_report() { return me->on_exe_report; }
-OnCancelReject &GeneralExecutionGateway::on_cancel_reject() {
-   static OnCancelReject default_cancel_reject;
-   return default_cancel_reject;
-}
-OnExchangePositionState &GeneralExecutionGateway::on_exchange_position_state() {
-   static OnExchangePositionState default_exchange_position_state;
-   return default_exchange_position_state;
-}
-OnExchangeOpenOrdersState &GeneralExecutionGateway::on_exchange_open_orders() {
-   static OnExchangeOpenOrdersState default_exchange_open_orders;
-   return default_exchange_open_orders;
-}
-} //namespace bts::orderman

src/lib/orderman/gateways/GeneralExecutionGateway.h

@@ -1,29 +0,0 @@
-#pragma once
-#include "lib/orderman/gateways/IExecutionGateway.h"
-#include "lib/orderman/gateways/IFIXGenericAdapter.h"
-namespace bts {
-namespace orderman {
-class GeneralExecutionGateway : public IExecutionGateway {
-public:
-    GeneralExecutionGateway(bts::api::Venue venue,std::unique_ptr<IFIXGenericAdapter> adapter, 
-                                                 md::SecurityOracle o);
-    virtual ~GeneralExecutionGateway();
-    void start() override;
-    void send_msgs(std::vector<OrderMsg> const & msgs, trace::CorID) override;
-    void dispatch_incoming_fix_messages() override;
-    void on_securities(const mercury::Update<NewSecuritiesEvent> &) override;
-    void on_admin_data(const mercury::Update<NewAdminDataEvent> &) override;
-    void send_fix(const jetfix::OutFIXPtr &, trace::CorID) override;
-    OnExeReport &on_exe_report() override;
-    OnGwStatusChange &on_gw_status_change() override;
-    OnCancelReject &on_cancel_reject() override;
-    OnExchangePositionState &on_exchange_position_state() override;
-    OnExchangeOpenOrdersState &on_exchange_open_orders() override;
-private:
-    struct PImpl;
-    std::unique_ptr<PImpl> me;
-    
-    
-};
-} //namespace bts
-} //namespace orderman

src/lib/orderman/gateways/IFIXGenericAdapter.cpp

@@ -1,6 +1,3 @@
 #include "IFIXGenericAdapter.h"
 
-namespace bts::orderman {
-
-    IFIXGenericAdapter::~IFIXGenericAdapter(){}
-}
+namespace bts::orderman {}

src/lib/orderman/gateways/IFIXGenericAdapter.h

@@ -15,8 +15,8 @@
 #include "lib/utils/bts_no_copy.h"
 
 #include <functional>
-
-
+#include <memory>
+#include <variant>
 
 namespace bts::orderman {
 
@@ -29,33 +29,33 @@ class NewAdminDataEvent;
 class ExecutionReport;
 class CancelReject;
 
-// struct OrderStateWithFills {
-//    api::orderman::OrderState ordr_state;
-//    std::vector<api::orderman::Fill> fills = {};
-// };
+struct OrderStateWithFills {
+   api::orderman::OrderState ordr_state;
+   std::vector<api::orderman::Fill> fills = {};
+};
 
-using OnExecutionReport = std::function<void(ExecutionReport const &, const trace::CorID &)>;
+using OnOrderUpdate = std::function<void(OrderStateWithFills const &)>;
 
 // fix execution gateway for a single FIX session
 class IFIXGenericAdapter {
  public:
    IFIXGenericAdapter() = default;
-   virtual ~IFIXGenericAdapter();
+   virtual ~IFIXGenericAdapter() {}
 
    virtual void start() = 0;
-//    // this callback should be added in a list, and on receving an update from the other side, it should be passed here
-   virtual void register_callback(const OnExecutionReport &) =0 ;
-   virtual void launch_new_order(const CreateOrderMsg &) =0;
-   virtual void modify_order(const CancelReplaceOrderMsg &) = 0;
-   virtual void cancel_order(const CancelOrderMsg &) =0;
+   // this callback should be added in a list, and on receving an update from the other side, it should be passed here
+   virtual void register_callback(const OnOrderUpdate &);
+   virtual void launch_new_order(const CreateOrderMsg &);
+   virtual void modify_order(const CancelReplaceOrderMsg &);
+   virtual void cancel_order(const CancelOrderMsg &);
    // as new securities are added
-   // virtual void update_securities(
-   //     const mercury::Update<NewSecuritiesEvent> &) = 0;
-   // virtual void update_admin_data(
-   //     const mercury::Update<NewAdminDataEvent> &) = 0;
+   virtual void update_securities(
+       const mercury::Update<NewSecuritiesEvent> &) = 0;
+   virtual void update_admin_data(
+       const mercury::Update<NewAdminDataEvent> &) = 0;
 
-//  private:
-//    BTS_NO_COPY(IFIXGenericAdapter);
+ private:
+   BTS_NO_COPY(IFIXGenericAdapter);
 };
 
 } // namespace bts::orderman

src/lib/orderman/gateways/OnixFIXAdapter.cpp

@@ -1,406 +0,0 @@
-#include "OnixFIXAdapter.h"
-#include "OnixS/FIXEngine/FIX/Engine.h"
-#include "lib/config/config.h"
-#include "lib/config/protos/system_config.pb.h"
-#include "lib/logging/logging.h"
-#include "lib/orderman/CancelReplaceOrderMsg.h"
-#include "lib/orderman/CreateOrderMsg.h"
-#include "lib/orderman/gateways/OnixFIXListener.h"
-#include "lib/orderman/gateways/TrafixOrderStateHandler.h"
-#include "lib/utils/bts_assert.h"
-#include <OnixS/FIXEngine/FIX/Exception.h>
-#include <OnixS/FIXEngine/FIX/FIX42.h>
-#include <OnixS/FIXEngine/FIX/Message.h>
-#include <OnixS/FIXEngine/FIX/ProtocolVersion.h>
-#include <cassert>
-#include <memory>
-#include <quickfix/FixFields.h>
-#include <string>
-
-namespace bts::orderman {
-OnixFIXAdapter::OnixFIXAdapter()
-    : state_handler(std::make_shared<TrafixOrderStateHandler>()), listener(state_handler),      // Initialize listener
-      session(nullptr) // Initialize session pointer to nullptr
-{
-
-   try {
-      // Initialization phase.
-      LOG(OrderMan, Warn, "OnixFIXAdapter::OnixFIXAdapter constructor");
-      OnixS::FIX::Engine::init();
-      session = std::make_unique<Session>(
-          "BTS", "TRFXMRX", OnixS::FIX::ProtocolVersion::FIX_42, &listener);
-      // Session-related logic goes here.
-
-   } catch (const OnixS::FIX::Exception &ex) {
-      std::string errorMessage = ex.what();
-      LOG(OrderMan, Error, "Exception: {}", errorMessage);
-      Engine::shutdown();
-      bts_assert_msg(false, "OnixFIXAdapter constructor failed");
-   }
-}
-OnixFIXAdapter::~OnixFIXAdapter() {
-   try {
-      session->logout();
-      Engine::instance()->unregisterListener();
-      Engine::shutdown();
-      LOG(OrderMan, Warn, "Object removed");
-   } catch (const OnixS::FIX::Exception &ex) {
-      std::string s = ex.what();
-      LOG(OrderMan, Error, "Exception {}", s);
-      bts_assert_msg(false, "OnixFIXAdapter deconstructor failed");
-   }
-}
-void OnixFIXAdapter::start() {
-   try {
-      LOG(OrderMan, Warn, "OnixFIXAdapter::start logging on");
-        const std::string counterpartyHost = "74.217.55.230";
-        const int heartBtInt = 30;
-        const int counterpartyPort = 22754;
-        session->logonAsInitiator(counterpartyHost, counterpartyPort, heartBtInt);
-      LOG(OrderMan, Warn, "OnixFIXAdapter::start logging on done");
-
-   } catch (const OnixS::FIX::Exception &ex) {
-      std::string errorMessage = ex.what();
-      LOG(OrderMan, Error, "OnixFIXAdapter::start Exception: {}", errorMessage);
-        session->logout();
-      Engine::shutdown();
-      bts_assert_msg(false, "OnixFIXAdapter session failed");
-   }
-}
-template <typename T>
-uint translateEnum(T enum_value, const std::map<int, int> &translationMap) {
-   auto it = translationMap.find(enum_value);
-   assert(it != translationMap.end()); // Using assert instead of bts_assert
-   return it->second;
-}
-
-uint translateTimeInForce(api::orderman::TimeInForce time_in_force) {
-   //0 = Day  1 = GTC 2 = At the Open 3 = IOC 4 = FOK 5 = GTX
-   static const std::map<int, int> tifMap = {
-       {api::orderman::TimeInForce::TIME_IN_FORCE_DAY, 0},
-       {api::orderman::TimeInForce::TIME_IN_FORCE_GTC, 1},
-       {api::orderman::TimeInForce::TIME_IN_FORCE_IOC, 3}};
-
-   return translateEnum(time_in_force, tifMap);
-}
-
-uint translateOrderType(api::orderman::NativeOrderType order_type) {
-   //1 = Market 2 = Limit 3 = Stop 4 = Stop Limit 5 = Market on close A = On Close B = Limit on Close
-   static const std::map<int, int> orderTypeMap = {
-       {api::orderman::NativeOrderType::NATIVE_ORDER_TYPE_MARKET, 1},
-       {api::orderman::NativeOrderType::NATIVE_ORDER_TYPE_LIMIT, 2},
-       {api::orderman::NativeOrderType::NATIVE_ORDER_TYPE_STOP, 3},
-       {api::orderman::NativeOrderType::NATIVE_ORDER_TYPE_STOP_LIMIT, 4}};
-
-   return translateEnum(order_type, orderTypeMap);
-}
-uint translateSide(api::Side side) {
-   //1 = Buy 2 = Sell 5 = Sell short 6 = Sell short Exempt
-   static const std::map<int, int> sideMap = {
-       {api::Side::SIDE_BUY, 1}, {api::Side::SIDE_SELL, 2}};
-
-   return translateEnum(side, sideMap);
-}
-
-// void waitUntilEnterKey(const std::string &waitText = std::string()) {
-//    OnixS::Threading::ThisThread::sleep(1000);
-
-//    if (!waitText.empty())
-//       std::clog << std::endl << waitText << std::endl;
-
-// #ifdef _WIN32
-//    _getch();
-// #else
-//    getchar();
-// #endif
-// }
-// static std::string get_value_from_file(const std::string &fallback, const std::string &name) {
-//    std::string filename =
-//        config::get<config::SystemConfig>().data_path() + name;
-//    std::ifstream infile(filename);
-//    if (infile.is_open()) {
-//       std::string file_value;
-//       infile >> file_value;
-//       infile.close();
-//       return file_value;
-//    } else {
-//       std::ofstream outfile(filename);
-//       if (!outfile) {
-//          throw std::runtime_error("Could not create the file.");
-//       }
-//       outfile << fallback;
-//       outfile.close();
-//       return fallback;
-//    }
-// }
-// static std::string get_symbol_from_file(const std::string &symbol) {
-//     return get_value_from_file(symbol, "symbol.txt");
-// }
-// static std::string get_exp_day_from_file(const std::string &exp_day) {
-//     return get_value_from_file(exp_day, "exp_day.txt");
-// }
-// static std::string get_exp_month_from_file(const std::string &exp_month) {
-//     return get_value_from_file(exp_month, "exp_month.txt");
-// }
-// static std::string get_exp_year_from_file(const std::string &exp_year) {
-//     return get_value_from_file(exp_year, "exp_year.txt");
-// }
-// static std::string get_put_from_file(const std::string &put) {
-//     return get_value_from_file(put, "put.txt");
-// }
-// static std::string get_strike_from_file(const std::string &strike) {
-//     return get_value_from_file(strike, "strike.txt");
-// }
-std::string pad_string(const std::string& str, size_t length, char padding = ' ') {
-    if (str.size() > length) {
-        return str.substr(0, length);
-    }
-    return str + std::string(length - str.size(), padding);
-}
-
-// Function to create the OSI Symbol
-static std::string create_osi_symbol(const std::string &symbol, const uint32_t &year, const uint32_t &month,const uint32_t &day,const char &call_put_indicator,const double &strike_price) {
-    if (symbol.size() > 6) {
-        throw std::invalid_argument("Symbol length cannot exceed 6 characters.");
-    }
-
-    std::ostringstream osi_stream;
-
-    // Symbol: 6 bytes, padded with spaces
-    osi_stream << pad_string(symbol, 6);
-
-    // Year: last 2 digits of the year
-    osi_stream << std::setw(2) << std::setfill('0') << (year % 100);
-
-    // Month: 2 digits
-    osi_stream << std::setw(2) << std::setfill('0') << month;
-
-    // Day: 2 digits
-    osi_stream << std::setw(2) << std::setfill('0') << day;
-
-    // Call/Put Indicator: 1 byte
-    osi_stream << call_put_indicator;
-
-    // Strike Price: 5 bytes for dollars, 3 bytes for decimal part
-    int strike_dollars = static_cast<int>(strike_price);
-    int strike_decimals = static_cast<int>((strike_price - strike_dollars) * 1000);
-    osi_stream << std::setw(5) << std::setfill('0') << strike_dollars;
-    osi_stream << std::setw(3) << std::setfill('0') << strike_decimals;
-
-    return osi_stream.str();
-}
-void parse_option(Message &order, const api::Security &sec) {
-    // std::string put_call = get_put_from_file(sec.option().call_put() == bts::api::PutOrCall::PUTORCALL_CALL ? "1" : "0");
-    // std::string exp_day = get_exp_day_from_file(std::to_string(sec.option().expiration_day()));
-    // std::string exp_month = get_exp_month_from_file(std::to_string(sec.option().expiration_month()));
-    // std::string exp_year = get_exp_year_from_file(std::to_string(sec.option().expiration_year()));
-    // std::string maturity_month_year = exp_year + exp_month;
-    // std::string strike_price = get_strike_from_file(std::to_string(sec.option().strike()));
-    auto osi = create_osi_symbol(sec.symbol(),sec.option().expiration_year(),sec.option().expiration_month(), sec.option().expiration_day(), sec.option().call_put()== bts::api::PutOrCall::PUTORCALL_CALL?'c':'p', sec.option().strike());
-    // std::string osi_symbol = get_value_from_file(osi, "osi_symbol.txt");
-
-    order
-        .set(OnixS::FIX::FIX42::Tags::PutOrCall, //put_call)
-        sec.option().call_put() == bts::api::PutOrCall::PUTORCALL_CALL ? "1" : "0")
-        .set(OnixS::FIX::FIX42::Tags::StrikePrice, //strike_price)
-        std::to_string(sec.option().strike()))
-        .set(OnixS::FIX::FIX42::Tags::MaturityMonthYear,// maturity_month_year)
-        std::to_string(sec.option().expiration_month()) +std::to_string(sec.option().expiration_year()))
-        .set(OnixS::FIX::FIX42::Tags::MaturityDay, //exp_day)
-        std::to_string(sec.option().expiration_day()))
-        .set(OnixS::FIX::FIX42::Tags::SecurityType,"OPT")
-        .set(OnixS::FIX::FIX42::Tags::Symbol,// osi_symbol)
-        osi)
-        .set(OnixS::FIX::FIX42::Tags::SecurityExchange, "SMART"); // Example value, adjust accordingly
-}
-void OnixFIXAdapter::modify_order(const CancelReplaceOrderMsg &msg) {
-   try {
-
-      //use this file:///C:/Users/arham/Downloads/OnixS.FixEngineCpp-Ubuntu2204-gcc1210-c++17-x64-4.10.1/OnixS.FixEngineCpp-Ubuntu2204-gcc1210-c++17-x64-4.10.1/doc/html/group__double-decimal-conversion.html
-
-      LOG(OrderMan, Warn, "OnixFIXAdapter::modify_order modifying order");
-      // auto order_cfg = config::get<config::OnixFixCfg>();
-      //message logic
-      //create message
-
-      // Get the symbol from the file or write it if the file doesn't exist
-      std::string symbol =msg.create_fields.security->symbol();
-      Message order(
-          OnixS::FIX::FIX42::Values::MsgType::Order_Cancel_Replace_Request,
-          OnixS::FIX::ProtocolVersion::FIX_42);
-      order
-          .set(OnixS::FIX::FIX42::Tags::HandlInst, 2) // 1: DMA 2: ALGO 3:CARE
-          .set(
-              OnixS::FIX::FIX42::Tags::OrigClOrdID,
-              std::to_string(msg.create_fields.key.value()))
-          .set(
-              OnixS::FIX::FIX42::Tags::ClOrdID,
-              "upd:"+std::to_string(msg.create_fields.key.value())) // ID for order
-          .set(
-              OnixS::FIX::FIX42::Tags::Account,
-              msg.create_fields.routing_record->account())
-          .set(
-              OnixS::FIX::FIX42::Tags::TimeInForce,
-              translateTimeInForce(msg.create_fields.time_in_force))
-          .set(
-              OnixS::FIX::FIX42::Tags::OrdType,
-              translateOrderType(msg.create_fields.order_type))
-          .set(OnixS::FIX::FIX42::Tags::Price, msg.create_fields.price.value())
-          .set(OnixS::FIX::FIX42::Tags::Rule80A, "A") //A = AGENCY P = PRINCIPAL
-          .set(OnixS::FIX::FIX42::Tags::SenderSubID, "BTC")
-          .set(OnixS::FIX::FIX42::Tags::TargetSubID, "TRFXMRX")
-          .set(OnixS::FIX::FIX42::Tags::ClientID, "BTC")
-          .set(OnixS::FIX::FIX42::Tags::SecurityType,"OPT")
-          //   .set(
-          //       OnixS::FIX::FIX42::Tags::Symbol,
-          //       msg.create_fields.security->symbol())
-          .set(OnixS::FIX::FIX42::Tags::CustomerOrFirm,0)
-          .set(OnixS::FIX::FIX42::Tags::OpenClose,"O")
-          .set(
-              OnixS::FIX::FIX42::Tags::Side,
-              translateSide(msg.create_fields.side))
-          .set(
-              OnixS::FIX::FIX42::Tags::OrderQty,
-              msg.create_fields.quantity.value())
-          .set(
-              OnixS::FIX::FIX42::Tags::TransactTime,
-              OnixS::FIX::Timestamp::utc(),
-              OnixS::FIX::TimestampFormat::YYYYMMDDHHMMSSMsec);
-      
-      if (msg.create_fields.security->has_option()) {
-        parse_option(order, *msg.create_fields.security);}
-      else{
-        order.set(OnixS::FIX::FIX42::Tags::Symbol, symbol);
-      }
-      order.validate();
-      LOG(OrderMan, Warn, "order Validated");
-      order.set(847, "SMART");
-      std::string order_string = order.toString('|');
-      updated = true;
-      session->send(&order);
-      LOG(OrderMan, Warn, "Order update sent :{}", order_string);
-      
-
-   }
-   // Shut down all reactors if they exist.
-   catch (const OnixS::FIX::Exception &ex) {
-      std::cout << "Exception: " << ex.what() << std::endl;
-   }
-}
-
-void OnixFIXAdapter::launch_new_order(const CreateOrderMsg &msg) {
-   LOG(OrderMan, Warn, "launch order {}", msg);
-
-   try {
-
-      // Listener listener2;
-
-      LOG(OrderMan, Warn, "OnixFIXAdapter::launch_new_order creating order");
-      // auto order_cfg = config::get<config::OnixFixCfg>() //add this
-
-
-        std::string symbol = msg.security->symbol();
-      Message order(
-          OnixS::FIX::FIX42::Values::MsgType::Order_Single,
-          OnixS::FIX::ProtocolVersion::FIX_42);
-      order
-          .set(OnixS::FIX::FIX42::Tags::HandlInst, 2) // 1: DMA 2: ALGO 3:CARE
-          .set(
-              OnixS::FIX::FIX42::Tags::ClOrdID,
-              std::to_string(msg.key.value())) // ID for order
-          .set(OnixS::FIX::FIX42::Tags::Account, msg.routing_record->account())
-          .set(
-              OnixS::FIX::FIX42::Tags::TimeInForce,
-              translateTimeInForce(msg.time_in_force))
-          .set(
-              OnixS::FIX::FIX42::Tags::OrdType,
-              translateOrderType(msg.order_type))
-          .set(OnixS::FIX::FIX42::Tags::Price, msg.price.value())
-          .set(OnixS::FIX::FIX42::Tags::Rule80A, "A") //A = AGENCY P = PRINCIPAL
-          .set(OnixS::FIX::FIX42::Tags::SenderSubID, "BTC")
-          .set(OnixS::FIX::FIX42::Tags::TargetSubID, "TRFXMRX")
-          .set(OnixS::FIX::FIX42::Tags::ClientID, "BTC")
-          .set(OnixS::FIX::FIX42::Tags::CustomerOrFirm,0)
-          .set(OnixS::FIX::FIX42::Tags::OpenClose,"O")
-          .set(OnixS::FIX::FIX42::Tags::Side, translateSide(msg.side))
-          .set(OnixS::FIX::FIX42::Tags::OrderQty, msg.quantity.value())
-          .set(
-              OnixS::FIX::FIX42::Tags::TransactTime,
-              OnixS::FIX::Timestamp::utc(),
-              OnixS::FIX::TimestampFormat::YYYYMMDDHHMMSSMsec);
-      if (msg.security->has_option()) {
-        parse_option(order, *msg.security);}
-      else{
-        order.set(OnixS::FIX::FIX42::Tags::Symbol, symbol)
-        .set(OnixS::FIX::FIX42::Tags::Currency,msg.security->currency());
-
-      }
-      std::string order_string = order.toString('|');
-      order.validate();
-      LOG(OrderMan, Warn, "order Validated");
-      order.set(847, "SMART");
-      state_handler->launch_order(msg);
-      session->send(&order);
-    //   TrafixOrderStateHandler
-    //   auto er = listener.fire_dummy_report(msg);
-
-      LOG(OrderMan, Warn, "Order sent :{}", order_string);
-      
-   }
-   // Shut down all reactors if they exist.
-   catch (const OnixS::FIX::Exception &ex) {
-      std::cout << "Exception: " << ex.what() << std::endl;
-   }
-}
-void OnixFIXAdapter::register_callback(const OnExecutionReport &cb) {
-   LOG(OrderMan, Warn, "OnixFIXAdapter::register_callback callback registered");
-   listener.setCallback(cb);
-}
-void OnixFIXAdapter::cancel_order(const CancelOrderMsg &msg) {
-   LOG(OrderMan, Warn, "cancel order {}", msg.key);
-//    std::string symbol =
-//           get_symbol_from_file(msg.security->symbol());
-   try {
-
-      Message order_cancel(
-          OnixS::FIX::FIX42::Values::MsgType::Order_Cancel_Request,
-          OnixS::FIX::ProtocolVersion::FIX_42);
-      order_cancel.set(OnixS::FIX::FIX42::Tags::HandlInst, 2);
-      order_cancel.set(
-          OnixS::FIX::FIX42::Tags::ClOrdID, "cancel:"+std::to_string(msg.key.value()));
-      // .set(OnixS::FIX::FIX42::Tags::OrderQty, msg.manual_order_indicator)
-      std::string prefix = updated?"upd:":"";
-      order_cancel
-          .set(
-              OnixS::FIX::FIX42::Tags::OrigClOrdID,
-              prefix+std::to_string(msg.key.value()))
-          .set(OnixS::FIX::FIX42::Tags::Side, translateSide(msg.side))
-        //   .set(OnixS::FIX::FIX42::Tags::Symbol, symbol)
-          .set(OnixS::FIX::FIX42::Tags::SecurityType,"OPT")
-          .set(
-              OnixS::FIX::FIX42::Tags::TransactTime,
-              OnixS::FIX::Timestamp::utc(),
-              OnixS::FIX::TimestampFormat::YYYYMMDDHHMMSSMsec);
-      if (msg.security->has_option()) {
-        parse_option(order_cancel, *msg.security);}
-      else{
-        order_cancel.set(OnixS::FIX::FIX42::Tags::Symbol, msg.security->symbol())
-        .set(OnixS::FIX::FIX42::Tags::Currency,msg.security->currency());
-
-      }
-      std::string order_str = order_cancel.toString('|');
-      LOG(OrderMan, Warn, "order_cancel Created: {}", order_str);
-      // order_cancel.validate(); fails becasuse of HandlInst
-      LOG(OrderMan, Warn, "order Validated");
-      order_cancel.set(847, "SMART");
-      updated = false;
-      session->send(&order_cancel);
-      LOG(OrderMan, Warn, "order cancel sent");
-
-   }
-   // Shut down all reactors if they exist.
-   catch (const OnixS::FIX::Exception &ex) {
-      std::cout << "Exception: " << ex.what() << std::endl;
-   }
-}
-} // namespace bts::orderman

src/lib/orderman/gateways/OnixFIXAdapter.h

@@ -1,27 +0,0 @@
-#pragma once
-
-#include "IFIXGenericAdapter.h"
-#include "lib/orderman/gateways/OnixFIXListener.h"
-#include "lib/orderman/gateways/TrafixOrderStateHandler.h"
-#include <memory>
-
-namespace bts::orderman {
-
-class OnixFIXAdapter : public IFIXGenericAdapter {
- public:
-   OnixFIXAdapter();
-   ~OnixFIXAdapter() override;
-   void register_callback(const OnExecutionReport &report) override;
-   void launch_new_order(const CreateOrderMsg &msg) override;
-   void cancel_order(const CancelOrderMsg &msg) override;
-   void modify_order(const CancelReplaceOrderMsg &) override;
-   void start() override;
-
- private:
-   std::shared_ptr<TrafixOrderStateHandler> state_handler;
-   OnixFixListener listener;
-   std::unique_ptr<Session> session;
-   bool updated = false;
-};
-
-} // namespace bts::orderman

src/lib/orderman/gateways/OnixFIXListener.cpp

@@ -1,175 +0,0 @@
-#include "OnixFIXListener.h"
-
-#include "lib/api/services/OrderService.pb.h"
-#include "lib/domain_model/Security.pb.h"
-#include "lib/domain_model/common.h"
-#include "lib/logging/logging.h"
-#include "lib/orderman/gateways/IExecutionGateway.h"
-#include "lib/orderman/gateways/IFIXGenericAdapter.h"
-#include "lib/orderman/orderman_utils/ExecutionReport.h"
-#include "lib/orderman/orderman_utils/orderman_common.h"
-#include <OnixS/FIXEngine/FIX/FIX42.h>
-#include <OnixS/FIXEngine/FIX/Tag.h>
-#include <string>
-
-namespace bts::orderman {
-ExecutionReport OnixFixListener::fire_dummy_report(CreateOrderMsg msg) {
-    ExecutionReport *er_p = exe_report_pool.construct();
-    ExecutionReport &er = *er_p;
-    std::string account_name = "BTSIM";
-    std::array<char, 12> an_array = {};
-    size_t i = 0;
-    for (const char c : account_name) {
-    an_array[i++] = c;
-    }
-    er.account_name = AccountName(an_array);
-    to_array(msg.routing_record->account(), &er.account_name.value());
-    to_array(
-        msg.routing_record->operator_id(), &er.operator_id.value());
-    er.order_key = msg.key;
-    er.msg_seq_num = 0;
-    er.security = msg.security;
-    er.leg_security = nullptr;
-    er.side = msg.side;
-    er.time_in_force = msg.time_in_force;
-    er.order_type = msg.order_type;
-    er.exec_type = api::orderman::EXEC_TYPE_NEW;
-    er.order_status = api::orderman::NATIVE_ORDER_STATUS_NEW;
-    er.manual_order_indicator = api::orderman::MANUAL_ORDER_INDICATOR_NO;
-    ExchangeOrderID64 eoid = ExchangeOrderID64(exchange_oid++);
-    ExchangeOrderIDVar oid = eoid;
-    er.exchange_order_id = oid;
-    er.price = msg.price;
-    er.quantity = msg.quantity;
-    er.filled_quantity = Quantity(0);
-    er.last_shares = Quantity(0);
-    er.msg = "Message Placed";
-    er.exchange = msg.security->exchange();
-    // Set transaction time
-    ExchangeTimeStamp64 now = ExchangeTimeStamp64(datetime::unix_nano_now());
-    er.transact_time = now;
-    cb(er,{});
-    return er;
-}
-
-ExecutionReport OnixFixListener::get_exe_report(Message msg) {
-     ExecutionReport *er_p = exe_report_pool.construct();
-    ExecutionReport &er = *er_p;
-    auto msg_str = msg.toString();
-    LOG(OrderMan,Info, "OnixFixListener::get_exe_report {}", msg_str);
-    std::string account_name = "BTSIM";
-    std::string ops_id = "op_id_bts";
-    std::array<char, 12> an_array = {};
-    size_t i = 0;
-    for (const char c : account_name) {
-    an_array[i++] = c;
-    }
-    er.account_name = AccountName(an_array);
-    to_array(
-        ops_id, &er.operator_id.value());
-    auto extractUInt = [](const std::string& input) -> uint {
-        std::string numericPart;
-        std::copy_if(input.begin(), input.end(), std::back_inserter(numericPart), [](char c) {
-            return std::isdigit(c);
-        });
-
-        if (numericPart.empty()) {
-            throw std::invalid_argument("Invalid format: no numeric characters found");
-        }
-
-        uint number = std::stoul(numericPart);
-        return number;
-    };
-    er.order_key = OrderKey(extractUInt(msg.get(OnixS::FIX::FIX42::Tags::OrderID).toString()));
-    er.msg_seq_num = 0;
-    // er.security =  ;
-    er.leg_security = nullptr;
-    er.side = msg.get(OnixS::FIX::FIX42::Tags::Side).toString()=="1"?api::Side::SIDE_BUY:api::Side::SIDE_SELL;
-    er.time_in_force = api::orderman::TIME_IN_FORCE_DAY;
-    er.order_type = api::orderman::NATIVE_ORDER_TYPE_LIMIT;
-    er.exec_type =  api::orderman::ExecType(std::stoi(msg.get(OnixS::FIX::FIX42::Tags::ExecType).toString()));
-    er.order_status = api::orderman::NATIVE_ORDER_STATUS_NEW;
-    er.manual_order_indicator = api::orderman::MANUAL_ORDER_INDICATOR_NO;
-    ExchangeOrderID64 eoid = ExchangeOrderID64(exchange_oid++);
-    ExchangeOrderIDVar oid = eoid;
-    er.exchange_order_id = oid;
-    er.price = Price(std::stod(msg.get(OnixS::FIX::FIX42::Tags::Price)));
-    er.quantity = Quantity(std::stoi(msg.get(FIX42::Tags::OrderQty)));
-    er.filled_quantity = Quantity(0);
-    er.last_shares = Quantity(0);
-    er.msg = "Message Placed";
-    er.exchange = api::VENUE_CBOE;
-    // Set transaction time
-    ExchangeTimeStamp64 now = ExchangeTimeStamp64(datetime::unix_nano_now());
-    er.transact_time = now;
-
-    return er;
-}
-
-void OnixFixListener::onInboundApplicationMsg(Message &msg, Session * /*sn*/) {
-    std::string msg_str = msg.toString();
-    std::string msg_json = msg.toJson();
-
-    LOG(OrderMan, Warn, 
-        "Incoming application-level message: {} As json {}",
-        msg_str, msg_json);
-        FieldValueRef t = msg.type();
-        char message_type;
-        if(!t.toChar(message_type)){
-            LOG(OrderMan, Warn, 
-        "no message type message: {} As json {}",
-        msg_str, msg_json);
-            return;
-        }
-
-        switch (message_type) {
-            case 'D':
-                {
-                    //exection report
-                    auto er = get_exe_report(msg);
-                    cb(er,{});
-                }
-        }
-
-    // LOG(OrderMan, Warn,
-    //     "onInboundApplicationMsg::inbound firing er: {}",
-    //     er->to_string());
-
-    // cb(*er, {});
-}
-
-
-void OnixFixListener::onInboundSessionMsg(Message & msg, Session * /*sn*/)
-{
-    std::string msg_str = msg.toString();
-    std::string msg_json = msg.toJson();
-    LOG( OrderMan, Warn,  
-    "Incoming session-level message:\n {} As json {}",
-    msg_str,
-    msg_json);
-}
-
-void OnixFixListener::onStateChange(SessionState::Enum newState, SessionState::Enum prevState,
-                             Session * /*sn*/)
-{
-    std::string prevst = SessionState::toString(prevState);
-    std::string newst = SessionState::toString(newState);
-    LOG(OrderMan, Warn,  
-    "Session's state is changed, prevState = {} , newState= {}", 
-    prevst,
-    newst);
-}
-
-void OnixFixListener::onError(ErrorReason::Enum /*reason*/, const std::string & description, Session * /*sn*/)
-{
-    LOG( OrderMan, Error, "Session-level error: {}", description);
-}
-
-void OnixFixListener::onWarning(WarningReason::Enum /*reason*/, const std::string & description,
-                         Session * /*sn*/)
-{
-    LOG( OrderMan, Error, "Session-level warning: {}", description);
-}
-
-
-}   // namespace bts::orderman

src/lib/orderman/gateways/OnixFIXListener.h

@@ -1,52 +0,0 @@
-#pragma once
-
-#include "lib/orderman/CreateOrderMsg.h"
-#include "lib/orderman/gateways/IExecutionGateway.h"
-#include "lib/orderman/gateways/TrafixOrderStateHandler.h"
-#include "lib/orderman/orderman_utils/ExecutionReport.h"
-#include <OnixS/FIXEngine.h>
-#include <OnixS/FIXEngine/FIX/Message.h>
-#include <boost/pool/object_pool.hpp>
-
-namespace bts::orderman {
-
-using namespace OnixS::FIX;
-class OnixFixListener : public OnixS::FIX::ISessionListener {
- public:
-   OnixFixListener( std::shared_ptr<TrafixOrderStateHandler> state_handler) : state_handler(state_handler),cb(dummy) {
-      LOG(OrderMan, Info, "OnixFixListener::cb set callback registered");
-   }
-   std::shared_ptr<TrafixOrderStateHandler> state_handler;
-   unsigned long exchange_oid = 0;
-   /// Is called when the application-level message is received from the counterparty.
-   void onInboundApplicationMsg(Message &msg, Session *sn)
-       ONIXS_FIXENGINE_FINAL;
-   // Set the real callback function
-   void setCallback(const OnExeReport &callback) {
-      LOG(OrderMan, Info, "OnixFixListener::setCallback callback registered");
-      cb = callback;
-   }
-   /// Is called when the session-level message is received from the counterparty.
-   void onInboundSessionMsg(Message &msg, Session *sn) ONIXS_FIXENGINE_FINAL;
-   void onStateChange(
-       SessionState::Enum newState,
-       SessionState::Enum prevState,
-       Session *sn) ONIXS_FIXENGINE_FINAL;
-   OnExeReport cb;
-   ExecutionReport fire_dummy_report(CreateOrderMsg msg);
-   ExecutionReport get_exe_report(Message msg);
-   static void dummy(const ExecutionReport &, const trace::CorID &) {
-      LOG(OrderMan, Info, "OnixFixListener::Dummy fired");
-   }
-   void onError(
-       ErrorReason::Enum reason,
-       const std::string &description,
-       Session *sn) ONIXS_FIXENGINE_FINAL;
-   void onWarning(
-       WarningReason::Enum reason,
-       const std::string &description,
-       Session *sn) ONIXS_FIXENGINE_FINAL;
-   boost::object_pool<ExecutionReport> exe_report_pool;
-};
-
-} // namespace bts::orderman

src/lib/orderman/gateways/TrafixOrderStateHandler.cpp

@@ -1,82 +0,0 @@
-#include "TrafixOrderStateHandler.h"
-#include "lib/domain_model/Security.pb.h"
-#include "lib/orderman/CreateOrderMsg.h"
-#include "lib/orderman/order_factory/IOrderFactory.h"
-#include "lib/orderman/orderman_utils/orderman_common.h"
-#include <string>
-
-namespace bts::orderman {
-
-// Internal structure for storing order info
-struct TrafixOrderStateHandler::OrderInfo {
-    OrderInfo(const CreateOrderMsg &msg)
-        : update_count(0), routing_record(msg.routing_record) {}
-
-    uint update_count;
-    const api::orderman::RoutingRecord* routing_record;
-
-    std::string update_id(bts::orderman::OrderKey key) {
-        return std::to_string(update_count++) + ":" + std::to_string(key.value());
-    }
-};
-
-// PImpl structure containing the actual implementation details, including SecurityOracle
-struct TrafixOrderStateHandler::PImpl {
-    PImpl() {}
-    std::unordered_map<bts::orderman::OrderKey, OrderInfo> orders;
-    // md::SecurityOracle sec_oracle;
-};
-
-// Constructor
-TrafixOrderStateHandler::TrafixOrderStateHandler()
-    : me(std::make_unique<PImpl>()) {}
-
-// Destructor
-TrafixOrderStateHandler::~TrafixOrderStateHandler() = default;
-
-// Method to launch an order
-bts::orderman::OrderKey TrafixOrderStateHandler::launch_order(const CreateOrderMsg &msg) {
-    me->orders.emplace(msg.key, OrderInfo(msg));
-    return bts::orderman::OrderKey(msg.key);
-}
-
-// Method to modify an order
-bts::orderman::OrderKey TrafixOrderStateHandler::modify_order(const std::string& key) {
-    auto order_key = get_order_key(key);
-    auto it = me->orders.find(order_key);
-    if (it != me->orders.end()) {
-        // Modify the order here
-        return bts::orderman::null_order_key();
-    }
-    return it->first;
-}
-
-// Utility method to get order key from a string
-bts::orderman::OrderKey TrafixOrderStateHandler::get_order_key(const std::string& id) {
-    std::size_t pos = id.find(':');
-    if (pos == std::string::npos) {
-        return null_order_key();
-    }
-    int key_value = std::stoi(id.substr(pos + 1));
-    return bts::orderman::OrderKey(key_value);
-}
-
-// Method to get update number from string
-int TrafixOrderStateHandler::get_update_number(const std::string& id) {
-    std::size_t pos = id.find(':');
-    if (pos == std::string::npos) {
-        return -1; // No colon found
-    }
-    
-    std::string prefix = id.substr(0, pos);
-    
-    // Check if the prefix contains only digits
-    if (prefix.empty() || !std::all_of(prefix.begin(), prefix.end(), ::isdigit)) {
-        return -1; // Not a valid number
-    }
-
-    // Convert the valid numeric prefix to an integer
-    return std::stoi(prefix);
-}
-
-} // namespace b

src/lib/orderman/gateways/TrafixOrderStateHandler.h

@@ -1,28 +0,0 @@
-#pragma once
-#include "IFIXGenericAdapter.h"
-#include "lib/orderman/gateways/OnixFIXListener.h"
-#include "lib/orderman/orderman_utils/orderman_common.h"
-#include <memory>
-
-namespace bts::orderman {
-
-class TrafixOrderStateHandler {
- public:
-  TrafixOrderStateHandler();
-  ~TrafixOrderStateHandler();
-
-  bts::orderman::OrderKey launch_order(const CreateOrderMsg &msg);
-  bts::orderman::OrderKey modify_order(const std::string &key);
-  bts::orderman::OrderKey cancel_order(const std::string &key);
-  std::string get_marex_id(const bts::orderman::OrderKey& key);
-
-  static bts::orderman::OrderKey get_order_key(const std::string& id);
-  static int get_update_number(const std::string& id);
-
- private:
-   struct OrderInfo;
-   struct PImpl;
-   std::unique_ptr<PImpl> me;
-};
-
-} // namespace bts::orderman

src/lib/orderman/order_factory/OrderFactory.cpp

@@ -129,7 +129,7 @@ NativeOrderBase *OrderFactory::make_order(const api::Venue exchange) {
       return new ICENativeOrder();
    else if (exchange == api::VENUE_GLBX || exchange == api::VENUE_SIM)
       return new GLBXNativeOrder();
-   return new EMNativeOrder(exchange);
+   return new EMNativeOrder();
 }
 
 NativeOrderBase *OrderFactory::make_native_order(

src/lib/orderman/orders/EMNativeOrder.cpp

@@ -5,7 +5,6 @@
 #include "lib/config/protos/toggles.pb.h"
 #include "lib/domain/OrderStatus.h"
 #include "lib/domain_model/Security.h"
-#include "lib/domain_model/Security.pb.h"
 #include "lib/domain_model/common.h"
 #include "lib/jetfix/codec/FIXFragment.h"
 #include "lib/jetfix/codec/FieldDefinitions.h"
@@ -53,10 +52,10 @@ struct InflightUpdateState {
 };
 
 struct EMNativeOrder::PImpl {
-   PImpl(const api::Venue exchange)
+   PImpl()
        : instance_id(EMNativeOrder::count++), services(nullptr),
          session_services(), sec(nullptr), previous_working_position(0),
-         definition(), state(), exchange_order_id(), exchange(exchange) {}
+         definition(), state(), exchange_order_id() {}
    const int instance_id;
    bool inited = false;
    OrderServices *services;
@@ -68,7 +67,7 @@ struct EMNativeOrder::PImpl {
    OrderDefinition definition;
    OrderState state;
    std::optional<ExchangeOrderIDVar> exchange_order_id;
-   const api::Venue exchange;
+
    OrderServices::ExeReportSubID exe_report_sid;
    OrderServices::CancelRejectSubID cancel_reject_sid;
    // If a cancel is sent before the order placement is ack'ed this is set to
@@ -83,7 +82,7 @@ struct EMNativeOrder::PImpl {
    RoutingRecord routing_record;
 };
 
-EMNativeOrder::EMNativeOrder(const api::Venue exchange) : me(new PImpl(exchange)) {
+EMNativeOrder::EMNativeOrder() : me(new PImpl()) {
    LOG(OrderMan, Info,
        "EMNativeOrder::EMNativeOrder instance_id: {} t_id: {} t_name: {}",
        me->instance_id, os::get_thread_id(), os::get_application_name());
@@ -502,19 +501,11 @@ OrderOpResult EMNativeOrder::_on_activate(
    me->state.set_filled_quantity(0);
    me->state.set_last_price(0);
    me->state.set_avg_price(0);
+
    me->session_services = &me->services->get_session_services(
        get_native_state(me->state).routing_record().session());
    // TODO: just to get going
-   std::string session_name;
-   switch (me->exchange) {
-      case api::VENUE_CBOE:
-         session_name = "cboe_1";
-      break;
-      default:
-         session_name ="deribit_1";
-
-   }
-   me->session_services = &me->services->get_session_services(session_name);
+   me->session_services = &me->services->get_session_services("deribit_1");
 
    // Checks for the following:
    // 1. Gateway connection

src/lib/orderman/orders/EMNativeOrder.h

@@ -1,7 +1,6 @@
 #pragma once
 
 #include "lib/api/services/OrderService.pb.h"
-#include "lib/domain_model/Security.pb.h"
 #include "lib/domain_model/common.h"
 #include "lib/orderman/CreateOrderMsg.h"
 #include "lib/orderman/orders/NativeOrderBase.h"
@@ -20,7 +19,7 @@ class SessionServices;
 // Wrapper around Emumba's order
 class EMNativeOrder : public NativeOrderBase {
  public:
-   EMNativeOrder(const api::Venue exchange = api::VENUE_DERIBIT);
+   EMNativeOrder();
    ~EMNativeOrder() override;
    api::orderman::OrderDefinition const &definition() const override;
    void on_init_new(

src/lib/orderman/orders/GLBXNativeOrder.cpp

@@ -478,7 +478,7 @@ OrderOpResult GLBXNativeOrder::on_activate(
              now.to_uint64(), (now - corid.app_arrival).nanoseconds());
          // in sim mode, we dont need this,
          me->encoder->send_create(*me->services, *me->session_services);
-         if (me->in_sim_mode) {       //CME EXECUTION NEEDS TO BE UPDATED
+         if (me->in_sim_mode) {
             CreateOrderMsg c_msg = {};
             _create_create_msg(c_msg);
             me->services->queue_order_msg(*me->session_services, c_msg);
@@ -579,7 +579,7 @@ OrderOpResult GLBXNativeOrder::on_update(api::orderman::Update const &update) {
    auto p = Price(update.price());
    auto q = Quantity(update.quantity());
    auto send_update_in_sim_mode = [&]() {
-      if (me->in_sim_mode) {  //CME EXECUTION NEEDS TO BE UPDATED
+      if (me->in_sim_mode) {
          CancelReplaceOrderMsg cxl_repl_msg = {};
          _create_create_msg(cxl_repl_msg.create_fields);
          Price new_p = Price(update.price());
@@ -799,7 +799,7 @@ OrderOpResult GLBXNativeOrder::on_cancel() {
       me->encoder->update_cancel_encoding(*me->services);
       me->encoder->send_cancel(*me->services, *me->session_services);
       // TODO: check for sim mode here
-      if (me->in_sim_mode) { //CME EXECUTION NEEDS TO BE UPDATED
+      if (me->in_sim_mode) {
          CancelOrderMsg cxl_msg = {};
          cxl_msg.key = key();
          LOG(OrderMan, Info, "GLBXNativeOrder queue_order_msg for cxl");
@@ -1381,7 +1381,7 @@ void GLBXNativeOrder::_create_create_msg(CreateOrderMsg &c) {
        bts::orderman::get_quantity(me->state, StateQuantityType::OrderQuantity);
    c.security = me->sec;
    c.side = me->definition.side();
-   c.order_type = get_glbx_native_ord_type(me->definition); //CME EXECUTION NEEDS TO BE UPDATED
+   c.order_type = get_glbx_native_ord_type(me->definition);
    c.time_in_force = me->definition.time_in_force();
    c.order_type = api::orderman::NATIVE_ORDER_TYPE_LIMIT;
    c.manual_order_indicator = me->definition.manual_order_indicator();

src/lib/orderman/orders/glbx/iLink2OrderEncoder.cpp

@@ -277,7 +277,6 @@ void iLink2OrderEncoder::update_exchange_order_id(ExchangeOrderIDVar const &v) {
    auto oid = to_trim_str(std::get<ExchangeOrderID>(v));
    me->update_encoding.add<jetfix::fix::OrderID>(oid);
    me->cancel_encoding.add<jetfix::fix::OrderID>(oid);
-   //CME EXECUTION NEEDS TO BE UPDATED, can be commented out
 }
 
 void iLink2OrderEncoder::update_create_order_encoding(

src/server/sparkd/md/MarketDataGateway.cpp

@@ -35,7 +35,7 @@
 #include "server/sparkd/md/okx/OkxSubscription.h"
 #include "server/sparkd/md/platforms/mdp3_onixs/MDP3Main.h"
 #include "server/sparkd/md/recording/MDReplayer.h"
-#include "server/sparkd/md/taifex/TaifexDataGateway.h"
+#include "server/sparkd/md/taifex/TaifexDataHandler.h"
 
 #include <bits/stdint-uintn.h>
 #include <boost/filesystem.hpp>
@@ -43,7 +43,6 @@
 #include <emumba-deribit-connector/deribit_utils.h>
 #include <memory>
 #include <optional>
-#include <string>
 #include <unordered_map>
 #include <vector>
 
@@ -118,6 +117,10 @@ struct MarketDataGateway::PImpl {
    std::vector<std::unique_ptr<MDReplayData>> replays = {};
    // we create these and pass them to the cc md sources
    std::vector<std::shared_ptr<VenueListenerData>> venue_listeners = {};
+    
+   void start_glbx();
+   void start_cboe();
+   void start_taifex();
 };
 
 MarketDataGateway::MarketDataGateway()
@@ -139,9 +142,9 @@ void MarketDataGateway::run() {
       bts_assert(me_->md_config.emumba_deribit().size() == 1);
       _start_emumba_deribit(me_->md_config.emumba_deribit().at(0));
    }
-   start_glbx(me_->md_config);
-   start_taifex(me_->md_config); 
-   start_cboe(me_->md_config);
+   me_->start_glbx();
+   me_->start_taifex(); 
+   me_->start_cboe();
    _start_replays(me_->md_config);
    while (true) {
       std::this_thread::sleep_for(std::chrono::seconds(1));
@@ -244,16 +247,16 @@ void MarketDataGateway::unsubscribe(MarketDataListenerID id) {
 }
 /// glbx //////////////////////////////////////////////////////////////////////
 //////////////////////////////////////////////////////////////////////////////
-void MarketDataGateway::start_glbx(const bts::api::md::MarketDataConfig &cfg){
-     for (const auto &glbx : cfg.glbx()) {
+void MarketDataGateway::PImpl::start_glbx(){
+     for (const auto &glbx : md_config.glbx()) {
       if(glbx.use_databento()){
-      auto p = me_->venue_listeners.emplace_back(
+      auto p = venue_listeners.emplace_back(
           std::make_shared<VenueListenerData>(bts::api::VENUE_GLBX));
       std::string tid_name = "glbx";
       bts::threading::make_thread(tid_name, [&]() {
         api::md::glbx::GlbxMarketDatahandler::run(
             p->listener,
-            me_->sec_id_store,
+            sec_id_store,
             glbx); 
       }).detach();
       }
@@ -372,31 +375,29 @@ void MarketDataGateway::_start_replays(
        os::get_thread_id(), os::get_thread_name());
 }
 
-void MarketDataGateway::start_taifex(const bts::api::md::MarketDataConfig &cfg) {
+void MarketDataGateway::PImpl::start_taifex() {
    // taifex
-   int counter = 0;
-   for (const auto &taifex : cfg.taifex()) {
-      counter++;
-      auto p = me_->venue_listeners.emplace_back(
+   for (const auto &taifex : md_config.taifex()) {
+      auto p = venue_listeners.emplace_back(
           std::make_shared<VenueListenerData>(bts::api::VENUE_TAIFEX));
-      std::string tid_name = "taifex" +std::to_string(counter);
+      std::string tid_name = "taifex";
       bts::threading::make_thread(tid_name, [&]() {
-         bts::md::taifex::TaifexDataGateway::run(
-          taifex, p->listener, me_->sec_id_store);
+         bts::md::taifex::TaifexDatahandler::run(
+          taifex.read_from_file(), p->listener, sec_id_store);
       }).detach();
       
    }
 }
-void MarketDataGateway::start_cboe(const bts::api::md::MarketDataConfig &cfg){
+void MarketDataGateway::PImpl::start_cboe(){
     // cboe
-    for (const api::md::CboeMarketDataConfig &cboe : cfg.cboe()) {
-       auto p = me_->venue_listeners.emplace_back(
+    for (const api::md::CboeMarketDataConfig &cboe : md_config.cboe()) {
+       auto p = venue_listeners.emplace_back(
            std::make_shared<VenueListenerData>(bts::api::VENUE_CBOE));
        std::string tid_name = "CBOE";
        bts::threading::make_thread(tid_name, [&]() {
           api::md::cboe::CboeMarketDatahandler::run(
             p->listener, 
-            me_->sec_id_store,
+            sec_id_store,
             cboe);
        }).detach();
 

src/server/sparkd/md/MarketDataGateway.h

@@ -36,9 +36,6 @@ class MarketDataGateway {
    static void _start_onix_glbx(
        bts::api::md::GlbxMarketDataConfig cfg);
    void _start_replays(const bts::api::md::MarketDataConfig &cfg);
-   void start_glbx(const bts::api::md::MarketDataConfig &cfg);
-   void start_cboe(const bts::api::md::MarketDataConfig &cfg);
-   void start_taifex(const bts::api::md::MarketDataConfig &cfg);
 };
 
 } // namespace bts::md

src/server/sparkd/md/cboe/CboeMarketDataHandler.cpp

@@ -29,8 +29,7 @@ void CboeMarketDatahandler::run(
         base_path + config.databento_settings().contracts_file_name(),
        base_path + config.databento_settings().bbo_file_name(),
        config.databento_settings().key(),
-       config.databento_settings().dataset(),
-       config.databento_settings().filter_securities_one_year());
+       config.databento_settings().dataset());
 }
 
 } // namespace bts::md::cboe

src/server/sparkd/md/databento/DatabentoDataHandler.cpp

@@ -1,7 +1,6 @@
 #include "DatabentoDataHandler.h"
 #include "databento/datetime.hpp"
 #include "databento/enums.hpp"
-#include "databento/exceptions.hpp"
 #include "databento/live_blocking.hpp"
 #include "databento/log.hpp"
 #include "databento/timeseries.hpp"
@@ -48,7 +47,6 @@ void DatabentoDatahandler::run(
     const std::string &books_path,
     const std::string &key,
     const std::string &dataset,
-    const bool filter_securities,
     const std::chrono::duration<long double> sleep_duration) {
 
    auto sleep_duration_updated = sleep_duration;
@@ -64,6 +62,12 @@ void DatabentoDatahandler::run(
       switch (venue) {
       case api::domain::Venue::VENUE_CBOE:{
          symbol = remove_spaces(std::string(msg.underlying.data(),msg.underlying.size()));
+        //  const auto exp_nanos = msg.expiration;
+        //  std::string exp_string = databento::ToIso8601(exp_nanos);
+        //  const auto expiration_time = bts::core::TimeStamp::from_iso_string(exp_string);
+        //  if(expiration_time.seconds() < expiration_time.now().seconds() || expiration_time.seconds() > expiration_time.now().seconds() + 8*30*24*60){
+        //     return databento::KeepGoing::Continue;
+        //  }
          break;
       }
       case api::domain::Venue::VENUE_GLBX:
@@ -77,16 +81,6 @@ void DatabentoDatahandler::run(
          bts_assert_msg(false, "Exchange not supported on databento");
          return databento::KeepGoing::Stop;
       }
-      if(filter_securities){
-       const auto exp_nanos = msg.expiration;
-       std::string exp_string = databento::ToIso8601(exp_nanos);
-       const auto expiration_time = bts::core::TimeStamp::from_iso_string(exp_string);
-       if(expiration_time.seconds() < expiration_time.now().seconds() || expiration_time.seconds() > (expiration_time.now().seconds() + 60*60*24*365)){
-          return databento::KeepGoing::Continue;
-       }
-      }
-
-
       bts::domain::SecurityID sec_id{sec_id_store->get_or_add(
           venue, std::to_string(msg.hd.instrument_id))};
       bts::domain::Security sec;
@@ -183,7 +177,7 @@ void DatabentoDatahandler::run(
     //hardcoding for now will update later
     switch (venue) {
         case api::domain::VENUE_CBOE:
-            client.Subscribe({"SPX.OPT","SPXW.OPT"}, databento::Schema::Definition, databento::SType::Parent,databento::UnixNanos{});
+            client.Subscribe({"SPX.OPT","SPXW.OPT"}, databento::Schema::Definition, databento::SType::Parent);
             client.Subscribe({"SPX.OPT","SPXW.OPT"}, databento::Schema::Mbp1, databento::SType::Parent);
             break;
         case api::domain::VENUE_GLBX:
@@ -197,7 +191,7 @@ void DatabentoDatahandler::run(
     client.Start();
     while (true) {
     try {
-    auto* record = client.NextRecord(std::chrono::seconds{500});
+    auto* record = client.NextRecord(std::chrono::seconds{5});
     if (record == nullptr) {
         LOG(md, Error, "Timed out waiting for data");
         break;
@@ -212,24 +206,10 @@ void DatabentoDatahandler::run(
     std::string err = error.what();
     LOG(md, Error, "Caught TCP error: {} \nAttempting reconnect\n",err);
     client.Reconnect();
-  } catch(const databento::DbnResponseError& err){
-    std::string msg = err.Message();
-    std::string what = err.what();
-    LOG(md, Error, "Caught DbnResponseError what: {} ,msg:{} \nShutting down\n",msg, what);
-    client.Stop();
-    // bts_assert(false);
-  }
-  catch(const databento::Exception& excp){
-    std::string err = excp.Message();
-    std::string err2 = excp.what();
-    LOG(md, Error, "Caught unknown dadtabento::exception what: {}, msg: {} \nShutting down\n",err, err2);
-    client.Stop();
-    // bts_assert(false);
-  }catch (const std::exception& exc) {
+  } catch (const std::exception& exc) {
     std::string err = exc.what();
     LOG(md, Error, "Caught unknown exception: {} \nShutting down\n",err);
     client.Stop();
-    // bts_assert(false);
   }
     
     }

src/server/sparkd/md/databento/DatabentoDataHandler.h

@@ -14,7 +14,6 @@ class DatabentoDatahandler {
     const std::string& books_path,
     const std::string &dataset="",
     const std::string& key = "db-Q4eCbTQfrAHaG6sAcuFUYB9YqkF6j",
-    const bool filter_securities=false,
     const std::chrono::duration<long double> sleep_duration = 0.04s);
 };
 } // namespace bts::md::databento

src/server/sparkd/md/glbx/GlbxMarketDataHandler.cpp

@@ -30,8 +30,7 @@ void GlbxMarketDatahandler::run(
        base_path + config.databento_settings().contracts_file_name(),
        base_path + config.databento_settings().bbo_file_name(),
        config.databento_settings().key(),
-       config.databento_settings().dataset(),
-       config.databento_settings().filter_securities_one_year());
+       config.databento_settings().dataset());
 }
 
 } // namespace bts::api::md::glbx 

src/server/sparkd/md/taifex/TaifexDataGateway.cpp

@@ -1,848 +0,0 @@
-#include "TaifexDataGateway.h"
-#include "TaifexDataProcessors.h"
-#include "lib/api/md/TaifexMarketDataConfig.pb.h"
-#include "lib/config/config.h"
-#include "lib/config/protos/system_config.pb.h"
-#include "lib/logging/logging.h"
-#include "lib/md/IMarketDataListener.h"
-#include "lib/persistence/IDGenerator.h"
-#include "lib/utils/bts_assert.h"
-#include "lib/utils/bts_chrono.h"
-#include "lib/utils/overloaded.h"
-#include "lib/utils/thread_manager.h"
-#include <memory>
-#include <optional>
-#include <ostream>
-#include <sstream>
-#include <string>
-#include <tbb/concurrent_hash_map.h>
-#include <tbb/concurrent_queue.h>
-#include <thread>
-#include <type_traits>
-#include <variant>
-#include <vector>
-namespace bts::md::taifex {
-
-enum class LIVE_DATA_STATES {
-   START = 0,
-   WAITING_FOR_REFRESH_DATA,
-   REFRESH_DATA_DONE,
-   WAITING_TO_SYNC_LIVE,
-   LIVE_RUNNING // add state to shift to after hours
-};
-constexpr size_t MSG_FOOTER_SIZE = 3;
-constexpr size_t MSG_HEADER_SIZE = 19;
-constexpr size_t FEED_BUFFER_SIZE = 1024;
-std::atomic_bool stop_refresh_thread = true;
-std::atomic_bool stop_reference_thread = true;
-std::atomic_bool stop_live_buffer_thread = true;
-std::atomic<LIVE_DATA_STATES> state;
-using UNPROCESSED_MSG = std::vector<unsigned char>;
-using REFERENCE_MSG =
-    std::variant<TaifexMessages::I010Data, TaifexMessages::I011Data>;
-using LIVE_MSG = std::variant<
-    TaifexMessages::I081Data,
-    TaifexMessages::I083Data,
-    TaifexMessages::I024Data>;
-using REFRESH_MSG = TaifexMessages::I083Data;
-using SNAPSHOT_BUFFER_MSG =
-    std::shared_ptr<std::variant<REFERENCE_MSG, REFRESH_MSG>>;
-
-// Refactored handleReferenceDataParsedMessage
-void handleReferenceDataParsedMessage(
-    const REFERENCE_MSG &msg,
-    const std::shared_ptr<bts::md::IMarketDataListener> &listener,
-    const std::shared_ptr<persistence::IDGenerator> &sec_id_store,
-    const std::shared_ptr<TaifexSecurityDetailsMapper> &security_details) {
-
-   std::visit(
-       [&](auto &&msg) {
-          using T = std::decay_t<decltype(msg)>;
-          if constexpr (std::is_same_v<T, TaifexMessages::I010Data>) {
-             TaifexDataProcessors::processI010Data(
-                 msg, listener, sec_id_store, security_details);
-          } else if constexpr (std::is_same_v<T, TaifexMessages::I011Data>) {
-             TaifexDataProcessors::processI011Data(msg, security_details);
-          }
-       },
-       msg);
-}
-std::optional<REFERENCE_MSG> getParsedReferenceMessage(
-    const UNPROCESSED_MSG &buffer,
-    size_t position,
-    const TaifexMessages::Header_Fields &Header) {
-   switch (Header.message_type.identifier) {
-   case TaifexMessages::MessageIdentifier::I010:
-      return TaifexMessages::I010Data(
-          buffer, position + MSG_HEADER_SIZE,
-          Header.Body_Length + MSG_FOOTER_SIZE);
-   case TaifexMessages::MessageIdentifier::I011:
-      return TaifexMessages::I011Data(
-          buffer, position + MSG_HEADER_SIZE,
-          Header.Body_Length + MSG_FOOTER_SIZE);
-   default:
-      return std::nullopt;
-   }
-}
-void handleReferenceDataUnparsedMessage(
-    const UNPROCESSED_MSG &buffer,
-    size_t position,
-    const TaifexMessages::Header_Fields &Header,
-    std::shared_ptr<bts::md::IMarketDataListener> listener,
-    std::shared_ptr<persistence::IDGenerator> sec_id_store,
-    std::shared_ptr<TaifexSecurityDetailsMapper> security_details) {
-   std::stringstream ss;
-   ss << Header.message_type.identifier;
-   auto s = ss.str();
-   LOG(md, Info, "handleReferenceDataParsedMessage: type:{}", s);
-   auto parsed = getParsedReferenceMessage(buffer, position, Header);
-   if (!parsed.has_value())
-      return;
-   handleReferenceDataParsedMessage(
-       parsed.value(), listener, sec_id_store, security_details);
-}
-void handleLiveDataParsedMessage(
-    const LIVE_MSG &msg,
-    std::shared_ptr<bts::md::IMarketDataListener> listener,
-    std::shared_ptr<TaifexSecurityDetailsMapper> security_details) {
-   std::visit(
-       [&](auto &&msg) {
-          using T = std::decay_t<decltype(msg)>;
-          if constexpr (std::is_same_v<T, TaifexMessages::I024Data>) {
-             TaifexDataProcessors::processTradeRecords(
-                 msg, listener, security_details);
-          } else if constexpr (std::is_same_v<T, TaifexMessages::I081Data>) {
-             TaifexDataProcessors::processBookIncrements(
-                 msg, listener, security_details);
-          } else if constexpr (std::is_same_v<T, TaifexMessages::I083Data>) {
-             TaifexDataProcessors::processBookSnapShotMessage(
-                 msg, listener, security_details);
-          }
-       },
-       msg);
-}
-std::optional<LIVE_MSG> getParsedLiveDataMessage(
-    const UNPROCESSED_MSG &buffer,
-    size_t position,
-    const TaifexMessages::Header_Fields &Header) {
-
-   switch (Header.message_type.identifier) {
-   case TaifexMessages::MessageIdentifier::I024:
-      return TaifexMessages::I024Data(
-          buffer, position + MSG_HEADER_SIZE,
-          Header.Body_Length + MSG_FOOTER_SIZE);
-   case TaifexMessages::MessageIdentifier::I083:
-      return TaifexMessages::I083Data(
-          buffer, position + MSG_HEADER_SIZE,
-          Header.Body_Length + MSG_FOOTER_SIZE);
-   case TaifexMessages::MessageIdentifier::I081:
-      return TaifexMessages::I081Data(buffer, position + MSG_HEADER_SIZE);
-   default:
-      return std::nullopt;
-   }
-}
-void handleLiveDataUnparsedMessage(
-    const UNPROCESSED_MSG &buffer,
-    size_t position,
-    const TaifexMessages::Header_Fields &Header,
-    std::shared_ptr<bts::md::IMarketDataListener> listener,
-    std::shared_ptr<TaifexSecurityDetailsMapper> security_details) {
-
-   auto parsed = getParsedLiveDataMessage(buffer, position, Header);
-   if (!parsed.has_value())
-      return;
-   handleLiveDataParsedMessage(parsed.value(), listener, security_details);
-}
-std::shared_ptr<TaifexMessages::I084Data> handleRefreshDataUnparsedMessage(
-    const UNPROCESSED_MSG &buffer,
-    size_t position,
-    const TaifexMessages::Header_Fields &Header) {
-   switch (Header.message_type.identifier) {
-   case TaifexMessages::MessageIdentifier::I084:
-      return std::make_shared<TaifexMessages::I084Data>(
-          TaifexMessages::I084Data(buffer, position));
-      //handle this
-      break;
-   default:
-      bts_assert_msg(false, "NOT IN REFRESH DATA");
-   }
-   bts_assert_msg(false, "Control should not reach here");
-   return std::make_shared<TaifexMessages::I084Data>(
-       TaifexMessages::I084Data(buffer, position));
-   ;
-}
-
-void HandleRefreshFeed(
-    int socket,
-    unsigned long seq_num,
-    tbb::concurrent_queue<SNAPSHOT_BUFFER_MSG> &snapshot_buffered_messages,
-    unsigned long &refresh_seq_num) {
-   LOG(md, Info, "HandleRefreshFeed looking for {}", seq_num);
-
-   enum {
-      WAITING_FOR_A = 0,
-      WAITING_FOR_Z,
-   } refresh_state = WAITING_FOR_A;
-   size_t counter = 0;
-   while (!stop_refresh_thread && !stop_live_buffer_thread) {
-      UNPROCESSED_MSG buffer(FEED_BUFFER_SIZE);
-      std::string desc = "Refresh Socket";
-      size_t bytes_received =
-          TaifexUtils::recv_socket_data(socket, buffer, FEED_BUFFER_SIZE, desc);
-      if (bytes_received <= 0) {
-         continue;
-      }
-      auto Header = TaifexMessages::Header_Fields(buffer, 0);
-      switch (refresh_state) {
-      case WAITING_FOR_A: {
-         auto i084_message =
-             TaifexMessages::I084Data(buffer, MSG_HEADER_SIZE, true);
-         LOG(md, Info, "HandleRefreshFeed WAITING_FOR_A looking for {}, got {}",
-             seq_num, i084_message.last_seq);
-         if (i084_message.message_type == TaifexMessages::I084Data::A &&
-             i084_message.last_seq > seq_num) {
-            refresh_state = WAITING_FOR_Z;
-         }
-         break;
-      }
-      case WAITING_FOR_Z: {
-         LOG(md, Info, "HandleRefreshFeed WAITING_FOR_Z");
-         auto i084_message = TaifexMessages::I084Data(buffer, MSG_HEADER_SIZE);
-         // bts_assert(i084_message.last_seq > seq_num);
-         if (i084_message.message_type == TaifexMessages::I084Data::O) {
-            for (auto m : i084_message.data) {
-               auto mes =
-                   std::make_shared<std::variant<REFERENCE_MSG, REFRESH_MSG>>(
-                       m);
-               if (std::holds_alternative<REFERENCE_MSG>(*mes)) {
-                  LOG(md, Error, "Adding REFer in refresh");
-               } else if (std::holds_alternative<REFRESH_MSG>(*mes)) {
-                  LOG(md, Info, "Adding REFresh in refresh");
-               } else {
-                  LOG(md, Error,
-                      "Unexpected message type in variant in refresh");
-               }
-               std::vector<std::string> contracts = {
-                   "TXF", "MXF", "TMF", "CDF", "QFF"};
-               bool found = false;
-
-               for (const auto &contract : contracts) {
-                  if (m.prod_id.find(contract) != std::string::npos &&
-                      m.prod_id.find("/") == std::string::npos) {
-                     found = true;
-                     break;
-                  }
-               }
-               if (found) {
-                  snapshot_buffered_messages.push(mes);
-               }
-               counter++;
-            }
-         } else if (i084_message.message_type == TaifexMessages::I084Data::Z) {
-            LOG(md, Info, "HandleRefreshFeed Z found total secs {}", counter);
-            refresh_seq_num = i084_message.last_seq;
-            stop_refresh_thread = true;
-            return;
-         }
-         break;
-      }
-      }
-   }
-}
-void truncateFile(const std::string &filename, size_t validSize) {
-   // Truncate the file to remove invalid data
-   std::fstream file(filename, std::ios::in | std::ios::out | std::ios::binary);
-   if (file) {
-      file.seekp(validSize); // Set the position where truncation should happen
-      file.close();
-      std::cout << "File truncated to " << validSize << " bytes." << std::endl;
-   } else {
-      std::cerr << "Failed to truncate file: " << filename << std::endl;
-   }
-}
-bool LoadReferenceFromFile(
-    const std::string &file_name,
-    tbb::concurrent_queue<SNAPSHOT_BUFFER_MSG> &snapshot_buffered_messages,
-    std::set<std::string> I010_seen,
-    std::set<std::string> I011_seen) {
-   const size_t BUFFER_SIZE = 1024;
-   std::ifstream file(file_name, std::ios::binary);
-   if (!file) {
-      LOG(md, Error, "File not found: {}", file_name);
-      return false;
-   }
-
-   UNPROCESSED_MSG buffer(
-       BUFFER_SIZE + 1024);         // Extra space for possible leftovers
-   UNPROCESSED_MSG prev_data(1024); // Stores leftover data
-   unsigned long total_read = 0;
-   bool start = true;
-   size_t prev_data_len = 0; // Length of leftover data
-   while (file.read(
-       reinterpret_cast<char *>(buffer.data() + prev_data_len), BUFFER_SIZE)) {
-      size_t bytes_read = file.gcount();
-      if (bytes_read < BUFFER_SIZE) {
-         break; // No more data to read
-      }
-      total_read += bytes_read;
-      if (!start) {
-         // Move data from prev_data to the start of the buffer
-         std::memcpy(buffer.data(), prev_data.data(), prev_data_len);
-      }
-
-      size_t position = 0;
-      while (position < (BUFFER_SIZE + prev_data_len - MSG_HEADER_SIZE)) {
-         auto Header = TaifexMessages::Header_Fields(buffer, position);
-         size_t message_length =
-             Header.Body_Length + MSG_HEADER_SIZE + MSG_FOOTER_SIZE;
-
-         // Check if the full message fits within the buffer
-         if ((BUFFER_SIZE + prev_data_len - position) >= message_length) {
-            // Handle the parsed message
-            auto msg = getParsedReferenceMessage(buffer, position, Header);
-            if (!msg.has_value())
-               continue;
-            bool seen = true;
-            std::visit(
-                overloaded{
-                    [&](const TaifexMessages::I010Data &i010) {
-                       if (!I010_seen.contains(i010.prod_id)) {
-                          seen = false;
-                          I010_seen.insert(i010.prod_id);
-                       }
-                    },
-                    [&](const TaifexMessages::I011Data &i011) {
-                       if (!I011_seen.contains(i011.kind_id)) {
-                          seen = false;
-                          I011_seen.insert(i011.kind_id);
-                       }
-                    }},
-                msg.value());
-            if (!seen) {
-               auto mes =
-                   std::make_shared<std::variant<REFERENCE_MSG, REFRESH_MSG>>(
-                       msg.value());
-               snapshot_buffered_messages.push(mes);
-            }
-            position += message_length;
-
-         } else {
-            // Incomplete message, break and handle leftover data
-            break;
-         }
-      }
-
-      // Move remaining (incomplete) data from buffer into prev_data and set prev_data_len
-      prev_data_len = BUFFER_SIZE + prev_data_len - position;
-      if (prev_data_len > 0) {
-         std::memcpy(prev_data.data(), buffer.data() + position, prev_data_len);
-      }
-      start = false;
-   }
-   truncateFile(file_name, total_read);
-   return true;
-}
-void writeBufferWithSeparator(
-    const std::vector<unsigned char> &buffer,
-    size_t bytes_received,
-    unsigned long position = 0) {
-   auto filename = config::get<bts::config::SystemConfig>().data_path() +
-                   "taifex_default_reference";
-   std::ofstream outputFile(
-       filename, std::ios::binary | std::ios::app); // Open file in append mode
-   if (!outputFile) {
-      std::cerr << "Failed to open file for writing: " << filename << std::endl;
-      return;
-   }
-
-   // Write the received bytes from the buffer to the file
-   outputFile.write(
-       reinterpret_cast<const char *>(buffer.data() + position),
-       bytes_received);
-
-   // Write the separator after the message
-   // outputFile.write(MESSAGE_SEPARATOR.c_str(), MESSAGE_SEPARATOR.size());
-
-   if (!outputFile) {
-      std::cerr << "Failed to write to file." << std::endl;
-   } else {
-      std::cout << "Successfully wrote " << bytes_received
-                << " bytes with separator to " << filename << std::endl;
-   }
-}
-void HandleReferenceFeed(
-    int socket,
-    tbb::concurrent_queue<SNAPSHOT_BUFFER_MSG> &snapshot_buffered_messages) {
-   std::set<std::string> I010_seen;
-   std::set<std::string> I011_seen;
-   LoadReferenceFromFile(
-       config::get<bts::config::SystemConfig>().data_path() +
-           "taifex_default_reference",
-       snapshot_buffered_messages, I010_seen, I011_seen);
-   // bool file_done =
-   UNPROCESSED_MSG buffer(FEED_BUFFER_SIZE);
-   LOG(md, Info, "HandleReferenceFeed");
-   std::string desc = "Reference socket";
-   while (!stop_reference_thread) {
-      // LOG(md, Info, "HandleReferenceFeed: Inside While");
-      uint bytes_received =
-          TaifexUtils::recv_socket_data(socket, buffer, FEED_BUFFER_SIZE, desc);
-      // LOG(md, Info, "HandleReferenceFeed: Inside While {}th pakcket recved",counter++);
-      if (bytes_received <= 0) {
-         // LOG(md, Info, "HandleReferenceFeed: Inside While {}th pakcket not valid",counter-1);
-         continue;
-      }
-      auto Header = TaifexMessages::Header_Fields(buffer, 0);
-      size_t message_length =
-          Header.Body_Length + MSG_HEADER_SIZE + MSG_FOOTER_SIZE;
-      if (bytes_received != message_length) {
-         LOG(md, Error, "Reference data recv: {}, should be:{}", bytes_received,
-             message_length);
-      } else {
-         auto msg = getParsedReferenceMessage(buffer, 0, Header);
-         if (!msg.has_value())
-            continue;
-         bool seen = true;
-         std::visit(
-             overloaded{
-                 [&](const TaifexMessages::I010Data &i010) {
-                    if (!I010_seen.contains(i010.prod_id)) {
-                       seen = false;
-                       I010_seen.insert(i010.prod_id);
-                    }
-                 },
-                 [&](const TaifexMessages::I011Data &i011) {
-                    if (!I011_seen.contains(i011.kind_id)) {
-                       seen = false;
-                       I011_seen.insert(i011.kind_id);
-                    }
-                 }},
-             msg.value());
-         if (!seen) {
-            auto mes =
-                std::make_shared<std::variant<REFERENCE_MSG, REFRESH_MSG>>(
-                    msg.value());
-            writeBufferWithSeparator(buffer, bytes_received);
-            snapshot_buffered_messages.push(mes);
-         }
-
-         // if (std::holds_alternative<REFERENCE_MSG>(*mes)) {
-         //    LOG(md, Error, "Adding REFer from refrence data");
-         // } else if (std::holds_alternative<REFRESH_MSG>(*mes)) {
-         //    LOG(md, Info, "Adding REFresh from refrence data");
-         // } else {
-         //    LOG(md, Error,
-         //        "Unexpected message type in variant from refrence data");
-         // }
-         // snapshot_buffered_messages.push(mes);
-      }
-   }
-   // 6. Close the socket after done processing
-   close(socket);
-}
-
-void HandleLiveBufferFeed(
-    int socket,
-    tbb::concurrent_queue<UNPROCESSED_MSG> &live_buffered_messages,
-    std::string desc) {
-   int bytes_received = -1;
-   while (!stop_live_buffer_thread) {
-      UNPROCESSED_MSG b(FEED_BUFFER_SIZE);
-      // auto &b = live_buffered_messages.emplace();
-      bytes_received =
-          TaifexUtils::recv_socket_data(socket, b, FEED_BUFFER_SIZE, desc);
-      if (bytes_received <= 0) {
-         LOG(md, Error, "HandleLiveBufferFeed: recv failed resseting state");
-         state = LIVE_DATA_STATES::START;
-         break;
-      }
-
-      if (bytes_received > 0) {
-         b.resize(bytes_received);
-         live_buffered_messages.push(
-             std::move(b)); // have to use this instead of emplace
-      }
-   }
-   stop_live_buffer_thread = true;
-}
-bool runOnBinaryFile(
-    const std::string &file_name,
-    std::shared_ptr<bts::md::IMarketDataListener> listener,
-    std::shared_ptr<persistence::IDGenerator> sec_id_store,
-    std::shared_ptr<TaifexSecurityDetailsMapper> security_details,
-    TaifexDataGateway::DATA_TYPE data_type) {
-   const size_t BUFFER_SIZE = 1024 * 1024;
-   std::ifstream file(file_name, std::ios::binary);
-   if (!file) {
-      LOG(md, Error, "File not found: {}", file_name);
-      bts_assert(false);
-   }
-
-   UNPROCESSED_MSG buffer(
-       BUFFER_SIZE + 1024);         // Extra space for possible leftovers
-   UNPROCESSED_MSG prev_data(1024); // Stores leftover data
-   bool start = true;
-   size_t prev_data_len = 0; // Length of leftover data
-   while (file.read(
-       reinterpret_cast<char *>(buffer.data() + prev_data_len), BUFFER_SIZE)) {
-      size_t bytes_read = file.gcount();
-      if (bytes_read < BUFFER_SIZE) {
-         break; // No more data to read
-      }
-      if (!start) {
-         // Move data from prev_data to the start of the buffer
-         std::memcpy(buffer.data(), prev_data.data(), prev_data_len);
-      }
-
-      size_t position = 0;
-      while (position < (BUFFER_SIZE + prev_data_len - MSG_HEADER_SIZE)) {
-         auto Header = TaifexMessages::Header_Fields(buffer, position);
-         size_t message_length =
-             Header.Body_Length + MSG_HEADER_SIZE + MSG_FOOTER_SIZE;
-
-         // Check if the full message fits within the buffer
-         if ((BUFFER_SIZE + prev_data_len - position) >= message_length) {
-            // Handle the parsed message
-            switch (data_type) {
-            case TaifexDataGateway::DATA_TYPE::LIVE:
-               handleLiveDataUnparsedMessage(
-                   buffer, position, Header, listener, security_details);
-               break;
-            case TaifexDataGateway::DATA_TYPE::REFERENCE:
-               handleReferenceDataUnparsedMessage(
-                   buffer, position, Header, listener, sec_id_store,
-                   security_details);
-               break;
-            case TaifexDataGateway::DATA_TYPE::REFRESH:
-               handleRefreshDataUnparsedMessage(buffer, position, Header);
-               break;
-            default:
-               bts_assert(false);
-            }
-            position += message_length;
-         } else {
-            // Incomplete message, break and handle leftover data
-            break;
-         }
-      }
-
-      // Move remaining (incomplete) data from buffer into prev_data and set prev_data_len
-      prev_data_len = BUFFER_SIZE + prev_data_len - position;
-      if (prev_data_len > 0) {
-         std::memcpy(prev_data.data(), buffer.data() + position, prev_data_len);
-      }
-
-      start = false; // Only move leftover data on subsequent iterations
-   }
-   return true;
-}
-
-void runLive(
-    std::shared_ptr<IMarketDataListener> listener,
-    std::shared_ptr<persistence::IDGenerator> sec_id_store,
-    std::shared_ptr<TaifexSecurityDetailsMapper> security_details) {
-
-   long reset_counter = -1;
-   state = LIVE_DATA_STATES::START;
-   std::set<int> sockets;
-   unsigned long last_seq_num;
-   std::unordered_map<std::string, REFRESH_MSG> products_waiting;
-   // auto print =
-   //     [](std::unordered_map<std::string, REFRESH_MSG> &products_waiting) {
-   //        std::string s = "\n";
-   //        for (auto &pair : products_waiting) {
-   //           s = s + "Key: " + pair.first + ", value: '" + pair.second.prod_id +
-   //               "'\n";
-   //        }
-   //        return s;
-   //     };
-   tbb::concurrent_queue<SNAPSHOT_BUFFER_MSG>
-       snapshot_buffered_messages; // Shared concurrent queue
-   tbb::concurrent_queue<UNPROCESSED_MSG>
-       live_buffered_messages; // Shared concurrent queue
-
-   // Load from config
-   std::string live_data_ip = "225.0.140.140";
-   const int live_data_port = 14000;
-   std::string futures_market_data_ip =
-       "192.168.201.94";                            // Replace with actual IP
-   std::string reference_data_ip = "225.0.140.141"; // Placeholder
-   const int reference_data_port = 14100;           // Placeholder
-   std::string refresh_data_ip = "225.0.140.147";   // Placeholder
-   const int refresh_data_port = 14700;             // Placeholder
-
-   //   if (use_after_hours) {
-   //     live_data_ip = "225.10.140.140";
-   //     reference_data_ip = "225.10.140.141"; // Placeholder
-   //     refresh_data_ip = "225.10.140.147";   // Placeholder
-   //   }
-   std::vector<TaifexMessages::I083Data> refresh_messages;
-   int live_socket = -1;
-   int reference_socket = -1;
-   int last_channel_id = -1;
-
-   while (true) {
-      switch (state) {
-      case LIVE_DATA_STATES::START: {
-         live_data_ip = TaifexUtils::getIP("realtime");
-         reference_data_ip = TaifexUtils::getIP("reference");
-         refresh_data_ip = TaifexUtils::getIP("refresh");
-         reset_counter++;
-         LOG(md, Error, "State: START counter:{}", reset_counter);
-         stop_reference_thread = stop_refresh_thread = stop_live_buffer_thread =
-             true;
-         security_details->reset();
-         live_buffered_messages.clear();
-         TaifexUtils::close_sockets(sockets);
-         last_seq_num = 0;
-         bts_assert(reset_counter < 5);
-
-         state = LIVE_DATA_STATES::WAITING_FOR_REFRESH_DATA;
-         break;
-      }
-      case LIVE_DATA_STATES::WAITING_FOR_REFRESH_DATA: {
-         LOG(md, Info, "State: WAITING_FOR_REFRESH_DATA");
-
-         reference_socket = TaifexUtils::create_multicast_socket_assert(
-             reference_data_ip, reference_data_port, futures_market_data_ip);
-         sockets.insert(reference_socket);
-         stop_reference_thread = false;
-         bts::threading::make_thread("Taifex_reference_data", [&]() {
-            HandleReferenceFeed(
-                reference_socket, std::ref(snapshot_buffered_messages));
-         }).detach();
-         LOG(md, Info,
-             "State: WAITING_FOR_REFRESH_DATA Reference thread started");
-
-         std::string desc = "WAITING_FOR_REFRESH_DATA Live data";
-         live_socket = TaifexUtils::create_multicast_socket_assert(
-             live_data_ip, live_data_port, futures_market_data_ip);
-         sockets.insert(live_socket);
-         UNPROCESSED_MSG buffer(FEED_BUFFER_SIZE);
-         size_t bytes_received = TaifexUtils::recv_socket_data(
-             live_socket, buffer, FEED_BUFFER_SIZE, desc);
-         if (bytes_received <= 0) {
-            LOG(md, Error,
-                "LIVE_DATA_STATES::WAITING_FOR_REFRESH_DATA recv failed "
-                "resseting state");
-            state = LIVE_DATA_STATES::START;
-            break;
-         }
-
-         auto s = TaifexUtils::getBufferString(buffer);
-         LOG(md, Info,
-             "State: WAITING_FOR_REFRESH_DATA first live message recieved: {}",
-             s);
-
-         TaifexMessages::Header_Fields Header =
-             TaifexMessages::Header_Fields(buffer, 0);
-         if (!Header.valid) {
-            LOG(md, Error, "Header not valid");
-            state = LIVE_DATA_STATES::START;
-            break;
-         }
-
-         last_seq_num = Header.Channel_Seq;
-         stop_refresh_thread = false;
-         stop_live_buffer_thread = false;
-
-         bts::threading::make_thread("Taifex_live_buffer", [&]() {
-            HandleLiveBufferFeed(
-                live_socket, live_buffered_messages, "Taifex Live Buffer");
-         }).detach();
-         LOG(md, Info,
-             "State: WAITING_FOR_REFRESH_DATA refresh thread started");
-
-         auto refresh_socket = TaifexUtils::create_multicast_socket_assert(
-             refresh_data_ip, refresh_data_port, futures_market_data_ip);
-         sockets.insert(refresh_socket);
-
-         unsigned long refresh_seq_num = 0;
-         HandleRefreshFeed(
-             refresh_socket, last_seq_num, snapshot_buffered_messages,
-             refresh_seq_num);
-
-         last_seq_num = refresh_seq_num;
-         if (state != LIVE_DATA_STATES::START)
-            state = LIVE_DATA_STATES::REFRESH_DATA_DONE;
-         break;
-      }
-      case LIVE_DATA_STATES::REFRESH_DATA_DONE: {
-         LOG(md, Info,
-             "State: REFRESH_DATA_DONE snapshot_buffered_messages size{}",
-             snapshot_buffered_messages.unsafe_size());
-         bts_assert(stop_refresh_thread);
-
-         while (!snapshot_buffered_messages.empty() ||
-                !products_waiting.empty()) {
-            SNAPSHOT_BUFFER_MSG msg;
-            if (!snapshot_buffered_messages.try_pop(msg)) {
-               continue;
-            }
-            // LOG(md, Info,
-            //     "State: REFRESH_DATA_DONE Waiting for Reference data products "
-            //     "waiting:{}, buffer_size:{} \n waiting: {}\n sec_deets: {}",
-            //     products_waiting.size(),
-            //     snapshot_buffered_messages.unsafe_size(),
-            //     print(products_waiting), security_details->to_string());
-
-            std::visit(
-                overloaded{
-                    [&](const REFERENCE_MSG &ref_msg) {
-                       handleReferenceDataParsedMessage(
-                           ref_msg, listener, sec_id_store, security_details);
-
-                       std::visit(
-                           overloaded{
-                               [&](const TaifexMessages::I010Data &data) {
-                                  // std::string  s = "looking for: '"+TaifexUtils::remove_spaces(data.prod_id)+"', in"+print(products_waiting);
-                                  //  LOG(md, Error, "{}",s);
-                                  if (products_waiting.contains(
-                                          TaifexUtils::remove_spaces(
-                                              data.prod_id))) {
-                                     auto m = products_waiting.at(
-                                         TaifexUtils::remove_spaces(
-                                             data.prod_id));
-                                     SNAPSHOT_BUFFER_MSG mes =
-                                         std::make_shared<std::variant<
-                                             REFERENCE_MSG, REFRESH_MSG>>(m);
-                                     snapshot_buffered_messages.push(mes);
-                                     products_waiting.erase(data.prod_id);
-                                  }
-                               },
-                               [&](const TaifexMessages::I011Data) {
-                                  return; // Exit early for I011Data
-                               }},
-                           ref_msg);
-                       return;
-                    },
-                    [&](const REFRESH_MSG &refresh_msg) {
-                       if (!TaifexDataProcessors::processBookSnapShotMessage(
-                               refresh_msg, listener, security_details)) {
-                          products_waiting.insert_or_assign(
-                              TaifexUtils::remove_spaces(refresh_msg.prod_id),
-                              refresh_msg);
-                       }
-                       return;
-                    }},
-                *msg);
-         }
-         stop_reference_thread = true;
-         state = LIVE_DATA_STATES::WAITING_TO_SYNC_LIVE;
-         break;
-         // stop reference data
-         // apply all refresh data
-      }
-      case LIVE_DATA_STATES::WAITING_TO_SYNC_LIVE: {
-         LOG(md, Info, "State: WAITING_TO_SYNC_LIVE");
-
-         while (true) {
-
-            // Process messages from the concurrent queue
-            UNPROCESSED_MSG buffer;
-            live_buffered_messages.try_pop(buffer);
-            if (buffer.empty()) {
-               continue;
-            }
-
-            TaifexMessages::Header_Fields Header(buffer, 0);
-            if (!Header.valid) {
-               LOG(md, Error, "Header not valid");
-               state = LIVE_DATA_STATES::START;
-               break;
-            }
-
-            if (Header.Channel_Seq < last_seq_num) {
-               continue; // Skip messages with outdated sequence numbers
-            } else {
-               break;
-            }
-            // If no message was available to pop, break the outer while loop
-            if (live_buffered_messages.empty()) {
-               break;
-            }
-         }
-
-         state = LIVE_DATA_STATES::LIVE_RUNNING;
-         break;
-      }
-         // while 1
-         // process buffer:
-         // if seq_num > last_seq_num + 1
-         // if seq_num <= last_seq_num
-         // skip message
-         // state = LIVE_RUNNING;
-         // break
-      case LIVE_DATA_STATES::LIVE_RUNNING: {
-         // if seq_num != last_seq_num + 1
-         LOG(md, Info, "STATE: LIVE RUNNING !!");
-         while (state == LIVE_DATA_STATES::LIVE_RUNNING) {
-
-            UNPROCESSED_MSG buffer;
-            live_buffered_messages.try_pop(buffer);
-            if (buffer.empty()) {
-               continue;
-            }
-            TaifexMessages::Header_Fields Header =
-                TaifexMessages::Header_Fields(buffer, 0);
-            if (!Header.valid) {
-               LOG(md, Error, "Header not valid");
-               state = LIVE_DATA_STATES::START;
-               break;
-            }
-            if (Header.Channel_Seq != last_seq_num + 1) {
-               LOG(md, Error, "Header.Channel_Seq != last_seq_num+1");
-               LOG(md, Info, "header.channel_seq {}", Header.Channel_Seq);
-               LOG(md, Info, "header.channel_id {}", Header.Channel_Id);
-               LOG(md, Info, "last_seq_number {}", last_seq_num);
-               LOG(md, Info, "last channel id  {}", last_channel_id);
-               state = LIVE_DATA_STATES::START;
-               break;
-            }
-            last_seq_num = Header.Channel_Seq;
-            last_channel_id = Header.Channel_Id;
-            handleLiveDataUnparsedMessage(
-                buffer, 0, Header, listener, security_details);
-         }
-         break;
-      }
-         // else
-         // process live
-      default:
-         bts_assert(false);
-      };
-   }
-}
-//const std::string &filename,
-void bts::md::taifex::TaifexDataGateway::run(
-    const api::md::TaifexMarketDataConfig &config,
-    std::shared_ptr<IMarketDataListener> listner,
-    std::shared_ptr<persistence::IDGenerator> sec_id_store) {
-   std::shared_ptr<TaifexSecurityDetailsMapper> sec_details =
-       std::make_shared<TaifexSecurityDetailsMapper>();
-   if (config.read_from_file()) {
-      bts_assert(!config.ref_data_file_name().empty());
-      bts_assert(!config.realtime_data_file_name().empty());
-      std::string reference_file_name =
-          config.ref_data_file_name(); //"reference_data_20240806_1";
-      std::string realtime_file_name =
-          config.realtime_data_file_name(); //"real_time_data_20240806_274";
-      std::string base_path =
-          config::get<bts::config::SystemConfig>().data_path();
-      runOnBinaryFile(
-          base_path + reference_file_name, listner, sec_id_store, sec_details,
-          REFERENCE);
-      runOnBinaryFile(
-          base_path + realtime_file_name, listner, sec_id_store, sec_details,
-          LIVE);
-      while (1) { //confirm with taha
-         sleep_for(10s);
-      }
-   } else {
-      runLive(
-          listner, sec_id_store,
-          sec_details); //regular_hours_connection_details,after_hours_connection_details);
-   }
-}
-
-} //namespace bts::md::taifex

src/server/sparkd/md/taifex/TaifexDataHandler.h

@@ -1,6 +1,5 @@
 #pragma once
 
-#include "lib/api/md/TaifexMarketDataConfig.pb.h"
 #include <memory>
 
 namespace bts::persistence {
@@ -13,15 +12,10 @@ class IMarketDataListener;
 
 namespace bts::md::taifex {
 //    const std::string &filename,
-class TaifexDataGateway {
+class TaifexDatahandler {
  public:
-  enum DATA_TYPE{
-    LIVE=0,
-    REFERENCE,
-    REFRESH
-  };
    static void run(
-      const api::md::TaifexMarketDataConfig &config,
+       bool use_file,
        std::shared_ptr<IMarketDataListener> listner,
        std::shared_ptr<persistence::IDGenerator> sec_id_store);
 };

src/server/sparkd/md/taifex/TaifexDataProcessors.h

@@ -1,232 +0,0 @@
-#pragma once
-#include "TaifexMessages.h"
-#include "TaifexSecurityDetailsMapper.h"
-#include "lib/domain/Security.h"
-#include "lib/domain/TradeRecord.h"
-#include "lib/md/IMarketDataListener.h"
-#include "lib/persistence/IDGenerator.h"
-#include "lib/utils/bts_assert.h"
-
-namespace bts::md::taifex {
-
-class TaifexDataProcessors {
- public:
-   static bool processBookSnapShotMessage(
-       const TaifexMessages::I083Data &i083_message,
-       std::shared_ptr<bts::md::IMarketDataListener> listener,
-       std::shared_ptr<TaifexSecurityDetailsMapper> security_details) {
-
-      auto [sec_id, dec_loc] = security_details->getSecurityDetails(
-          TaifexUtils::remove_spaces(i083_message.prod_id));
-      if (!sec_id || !dec_loc)
-         return false;
-      domain::Book book;
-      auto &bids = book.bids();
-      auto &asks = book.asks();
-      book.set_security_id(sec_id.value());
-
-      for (const auto &entry : i083_message.entries) {
-         bts::domain::Price price(
-             entry.price *
-             pow(10,
-                 9 - dec_loc
-                         .value())); // Assuming the price needs to be adjusted by a factor of -2
-         bts::domain::Quantity quantity(entry.quantity);
-
-         // Process each MD_ENTRY
-         if (entry.md_entry_type == '0') { // Buy or derived buy
-            // Update the corresponding bid level
-            if (entry.level <= bids.size()) {
-               bids[entry.level - 1].price = price;
-               bids[entry.level - 1].total_quantity = quantity;
-            }
-         } else if (
-             entry.md_entry_type == '1') { // Sell or derived sell
-            // Update the corresponding ask level
-            if (entry.level <= asks.size()) {
-               asks[entry.level - 1].price = price;
-               asks[entry.level - 1].total_quantity = quantity;
-            }
-         }
-      }
-      std::stringstream book_after;
-      book_after << "Book after snapshot update:\n";
-      book_after << "Bids:\n";
-      for (const auto &level : bids) {
-         book_after << "Price: " << level.price.to_double()
-                    << ", Quantity: " << level.total_quantity.value() << "\n";
-      }
-      book_after << "Asks:\n";
-      for (const auto &level : asks) {
-         book_after << "Price: " << level.price.to_double()
-                    << ", Quantity: " << level.total_quantity.value() << "\n";
-      }
-      book_after<<"MSG:\n"<<i083_message;
-      std::string book_snapshot_details = book_after.str() + "\n";
-      bts_assert_msgf(
-          book.is_sorted(), "book not sorted {}", book_snapshot_details);
-      listener->on_book(book);
-      return true;
-   }
-   static void processBookIncrements(
-       const TaifexMessages::I081Data &msg,
-       const std::shared_ptr<bts::md::IMarketDataListener> &listener,
-       const std::shared_ptr<TaifexSecurityDetailsMapper> &security_details) {
-
-      auto [sec_id, dec_loc] = security_details->getSecurityDetails(
-          TaifexUtils::remove_spaces(msg.prod_id));
-      if (!sec_id || !dec_loc)
-         return;
-
-      for (const auto &entry : msg.entries) {
-         IMarketDataListener::BookOp inc;
-         inc.id = *sec_id;
-         inc.price =
-             bts::domain::Price(entry.price * std::pow(10, 9 - *dec_loc));
-         inc.quantity = bts::domain::Quantity(entry.quantity);
-         inc.side = (entry.md_entry_type == '0' || entry.md_entry_type == 'E')
-                        ? bts::domain::Side::Buy
-                        : bts::domain::Side::Sell;
-
-         switch (entry.md_update_action) {
-         case '0':
-            inc.op_type = IMarketDataListener::BookOpType::Add;
-            break;
-         case '1':
-            inc.op_type = IMarketDataListener::BookOpType::MODIFY;
-            break;
-         case '2':
-            inc.op_type = IMarketDataListener::BookOpType::REMOVE;
-            break;
-         default:
-            continue;
-         }
-         listener->on_book_increment(inc);
-      }
-   }
-   // Refactored trade processing function
-   static void processTradeRecords(
-       const TaifexMessages::I024Data &msg,
-       const std::shared_ptr<bts::md::IMarketDataListener> &listener,
-       const std::shared_ptr<TaifexSecurityDetailsMapper> &security_details) {
-
-      auto [sec_id, dec_loc] = security_details->getSecurityDetails(
-          TaifexUtils::remove_spaces(msg.prod_id));
-      if (!sec_id || !dec_loc)
-         return;
-
-      std::vector<bts::domain::TradeRecord> trades;
-      for (const auto &trade_record : msg.trades) {
-         auto &record = trades.emplace_back();
-         record.sec_id = *sec_id;
-         record.time_stamp = bts::core::TimeStamp::from_iso_string(
-             TaifexUtils::convertToISO(msg.match_time, "datetime_no_timezone"));
-         record.trade_quantity = bts::domain::Quantity(trade_record.match_qty);
-         record.trade_price = bts::domain::Price(
-             trade_record.match_price * std::pow(10, 9 - *dec_loc));
-         // TODO: Set the aggressor side (record.agg_side)
-      }
-
-      std::vector<const domain::TradeRecord *> trade_ptrs;
-      trade_ptrs.reserve(trades.size());
-      for (const auto &trade : trades) {
-         trade_ptrs.push_back(&trade);
-      }
-      //   for (const auto *trade_ptr : trade_ptrs) {
-      //          if (trade_ptr != nullptr) {
-      //             LOG(md, Error, "Trade Quantity: {}, Trade Price: {}",
-      //                 trade_ptr->trade_quantity.value(),
-      //                 trade_ptr->trade_price);
-      //          }
-      //       }
-      listener->on_trades(trade_ptrs);
-   }
-   static bts::domain::Security initializeSecurity(
-       const std::string &id,
-       const bts::domain::SecurityID &sec_id,
-       const std::string &symbol,
-       const std::shared_ptr<TaifexSecurityDetailsMapper> &security_details) {
-
-      bts::domain::Security sec;
-      sec.security_common.venue = bts::api::domain::VENUE_TAIFEX;
-      sec.security_common.id = sec_id;
-      sec.security_common.exchange_sec_id = id;
-      sec.security_common.desc = id;
-      sec.security_common.symbol = symbol;
-      sec.security_common.asset = id;
-      sec.security_common.group = id;
-      sec.security_common.min_price_increment =
-          security_details->get_min_inc(id);
-      sec.security_common.min_quantity = security_details->get_min_inc(id);
-      sec.security_common.min_quantity_increment =
-          security_details->get_min_inc(id);
-      sec.security_common.currency = security_details->get_currency(id);
-
-      return sec;
-   }
-
-   // Process I010Data
-   static void processI010Data(
-       const TaifexMessages::I010Data &msg,
-       const std::shared_ptr<bts::md::IMarketDataListener> &listener,
-       const std::shared_ptr<persistence::IDGenerator> &sec_id_store,
-       const std::shared_ptr<TaifexSecurityDetailsMapper> &security_details) {
-
-      const auto id = TaifexUtils::remove_spaces(msg.prod_id);
-      const auto symbol = id.substr(0, 3);
-
-      bts::domain::SecurityID sec_id{
-          sec_id_store->get_or_add(api::domain::VENUE_TAIFEX, id)};
-      auto check =
-          security_details->add_i010_mapping(id, sec_id, msg.decimal_locator);
-      if (!check) {
-         return;
-      }
-
-      // Initialize the security object
-      bts::domain::Security sec =
-          initializeSecurity(id, sec_id, symbol, security_details);
-
-      // Handle product-specific details
-      switch (msg.parsedProductInfo.productType) {
-      case TaifexMessages::I010Data::ProductInfo::ProductType::FUTURES: {
-         auto &futureSec = sec.security.emplace<domain::FutureSecurity>();
-         std::tm tm = {};
-         std::istringstream iss(msg.end_date);
-         iss >> std::get_time(&tm, "%Y%m%d");
-         std::time_t time = std::mktime(&tm);
-         futureSec.expiration =
-             bts::core::TimeStamp(static_cast<uint64_t>(time));
-         break;
-      }
-      case TaifexMessages::I010Data::ProductInfo::ProductType::OPTIONS: {
-         auto &optionSec = sec.security.emplace<domain::OptionSecurity>();
-         std::tm tm = {};
-         std::istringstream iss(msg.end_date);
-         iss >> std::get_time(&tm, "%Y%m%d");
-         std::time_t time = std::mktime(&tm);
-         optionSec.expiration =
-             bts::core::TimeStamp(static_cast<uint64_t>(time));
-         optionSec.strike =
-             domain::Price::from_double(msg.parsedProductInfo.strikePrice);
-         break;
-      }
-      }
-
-      std::vector<const domain::Security *> const_security_list;
-      const_security_list.push_back(&sec);
-      listener->on_securities(const_security_list);
-   }
-
-   // Process I011Data
-   static void processI011Data(
-       const TaifexMessages::I011Data &msg,
-       const std::shared_ptr<TaifexSecurityDetailsMapper> &security_details) {
-
-      const auto id = TaifexUtils::remove_spaces(msg.kind_id);
-      security_details->add_i011_mapping(
-          id, msg.contract_size, msg.getCurrencyString());
-   }
-};
-
-} //namespace bts::md::taifex

src/server/sparkd/md/taifex/TaifexSecurityDetailsMapper.h

@@ -1,140 +0,0 @@
-#pragma once
-#include "lib/domain/SecurityID.h"
-#include "lib/logging/logging.h"
-#include <optional>
-#include <sstream>
-#include <string>
-namespace bts::md::taifex {
-class TaifexSecurityDetailsMapper {
-
-
-   struct product_details {
-      std::string instrument_id; // use this as 
-      bts::domain::SecurityID sec_id;//uint64_t
-      uint dec_loc;
-   };
-   struct contract_details{
-      std::unordered_map<std::string,product_details> products;
-      std::string kind_id;
-      double min_increment;
-      std::string currency;
-   };
-
- public:
-   static std::string get_kind_id(const std::string &prod_id ){
-      return prod_id.substr(0,3);
-   }
-   // Add a mapping from instrument_id to sec_id
-   bool add_i010_mapping(
-       std::string instrument_id,
-       const bts::domain::SecurityID sec_id,
-       uint decimal_locator) {
-      auto kind_id = get_kind_id(instrument_id);
-      if(!sec_deets_map.contains(kind_id)){
-         return false;
-      }
-      auto &contract = sec_deets_map[kind_id];
-      auto &details = contract.products[instrument_id];
-      details.sec_id = sec_id;
-      details.dec_loc = decimal_locator;
-      details.instrument_id = instrument_id;
-      auto s = this->to_string();
-      LOG(md, Info, "add_i010_mapping: SEC_ID STORE= {}",s);
-      return true;
-   }
-
-   // Add a mapping for i011
-   void add_i011_mapping(
-       std::string kind_id, double min_increment, std::string currency) {
-       auto &contract = sec_deets_map[kind_id];
-       contract.currency= currency!=""?currency:"usd";
-       contract.kind_id=kind_id;
-       contract.min_increment=min_increment;
-   }
-
-   // Find sec_id by instrument_id
-   std::optional<bts::domain::SecurityID> get_sec_id(
-       std::string instrument_id) const {
-      auto kind_id = get_kind_id(instrument_id);
-      auto it = sec_deets_map.find(kind_id);
-      if (it != sec_deets_map.end()) {
-         auto &contract = it->second;
-         auto it2 = contract.products.find(instrument_id);
-         if (it2 != contract.products.end()) 
-            return it2->second.sec_id;
-      }
-      return std::nullopt;
-   }
-
-   // Find dec_loc by instrument_id
-   std::optional<uint> get_dec_loc(std::string instrument_id) const {
-      auto kind_id = get_kind_id(instrument_id);
-      auto it = sec_deets_map.find(kind_id);
-      if (it != sec_deets_map.end()) {
-         auto &contract = it->second;
-         auto it2 = contract.products.find(instrument_id);
-         if (it2 != contract.products.end()) 
-            return it2->second.dec_loc;
-      }
-      return std::nullopt;
-   }
-   std::string get_currency(std::string instrument_id) const {
-      auto kind_id = get_kind_id(instrument_id);
-      auto it = sec_deets_map.find(kind_id);
-      if (it != sec_deets_map.end()) {
-         return it->second.currency;
-         
-      }
-      return "usd";
-   }
-
-   double get_min_inc(std::string instrument_id) const {
-      auto kind_id = get_kind_id(instrument_id);
-      auto it = sec_deets_map.find(kind_id);
-      if (it != sec_deets_map.end()) {
-         return it->second.min_increment;
-      }
-      return 1;
-   }
-   std::pair<std::optional<bts::domain::SecurityID>, std::optional<int>> getSecurityDetails(
-    const std::string &prod_id) {
-    auto sec_id = get_sec_id(prod_id);
-    auto dec_loc = get_dec_loc(prod_id);
-
-    if (!sec_id.has_value() || !dec_loc.has_value()) {
-        if (sec_id.has_value() != dec_loc.has_value())
-            LOG(md, Error, "Mismatch in sec_id: {}, dec_loc: {}", sec_id.has_value(), dec_loc.has_value());
-        return {std::nullopt, std::nullopt};
-    }
-    return {sec_id, dec_loc};
-   }
-   // to_string function to print all data in both maps
-   std::string to_string() const {
-    std::stringstream ss;
-    for (const auto &[kind_id, contract] : sec_deets_map) {
-        ss << "Kind ID: '" << kind_id << "'\n"
-           << "  Currency: '" << contract.currency << "'\n"
-           << "  Min Increment: " << contract.min_increment << "\n";
-
-        for (const auto &[instrument_id, details] : contract.products) {
-            ss << "    Instrument ID: '" << instrument_id << "'\n"
-               << "      Sec ID: " << details.sec_id << "\n"
-               << "      Dec Loc: " << details.dec_loc << "\n";
-        }
-    }
-    return ss.str();
-}
-
-   // Reset function to clear the data map
-   void reset() {
-      for(auto &contract:sec_deets_map){
-         contract.second.products.clear();
-      }
-      sec_deets_map.clear();
-   }
-
- private:
-   std::unordered_map<std::string, contract_details> sec_deets_map;
-};
-
-} // namespace bts::md::taifex

src/server/sparkd/md/taifex/TaifexUtils.h

@@ -1,508 +0,0 @@
-#include "lib/logging/logging.h"
-#include "lib/utils/bts_assert.h"
-#include <boost/date_time/posix_time/ptime.hpp>
-#include <boost/date_time/posix_time/time_parsers.hpp>
-#include <chrono>
-#include <cstddef>
-#include <cstdio>
-#include <limits>
-#include <netinet/in.h>
-#include <string>
-#include <vector>
-
-namespace bts::md::taifex {
-
-class TaifexUtils {
-   using UNPROCESSED_MSG = std::vector<unsigned char>;
- public:
-   static int recv_socket_data(int socket, UNPROCESSED_MSG& buffer, const size_t FEED_BUFFER_SIZE, const std::string & socket_desc ="") {
-    int bytes_received = -1;
-    try {
-        bytes_received = recvfrom(socket, buffer.data(), FEED_BUFFER_SIZE, 0, nullptr, nullptr);
-    } catch (...) {
-         std::string s = strerror(errno);
-        LOG(md, Error, "{}: Error receiving data from socket, {}",socket_desc, s);
-    }
-    if (bytes_received < 0) {
-         std::string s = strerror(errno);
-        LOG(md, Error, "{}: Socket error: {}",socket_desc, s);
-    } else if (bytes_received == 0) {
-         std::string s = strerror(errno);
-        LOG(md, Error, "{} Received 0 bytes {}",socket_desc, s);
-    }
-    return bytes_received;
-}
-   
-   // Reusable function to create and join a multicast group
-   static int create_multicast_socket(
-       const std::string &multicast_ip,
-       const int port,
-       const std::string &interface_ip) {
-      struct sockaddr_in addr;
-      memset(&addr, 0, sizeof(addr));
-
-      // Create UDP socket
-      auto sock = socket(AF_INET, SOCK_DGRAM, 0);
-      if (sock < 0) {
-         LOG(md, Error, "Socket creation failed IP{} PORT{}", multicast_ip,
-             port);
-         return -1;
-      }
-
-      // Allow multiple sockets to use the same port
-      int reuse = 1;
-      if (setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse)) <
-          0) {
-         LOG(md, Error, "Setting SO_REUSEADDR failed IP{} PORT{}", multicast_ip,
-             port);
-         close(sock);
-         return -1;
-      }
-
-      // Bind to the multicast port
-      addr.sin_family = AF_INET;
-      addr.sin_port = htons(port);
-      addr.sin_addr.s_addr = htonl(INADDR_ANY);
-
-      if (bind(sock, (struct sockaddr *)&addr, sizeof(addr)) < 0) {
-         LOG(md, Error, "Bind failed IP{} PORT{}", multicast_ip, port);
-         close(sock);
-         return -1;
-      }
-
-      // Specify the network interface for multicast
-      struct in_addr localInterface;
-      localInterface.s_addr = inet_addr(interface_ip.c_str());
-      if (setsockopt(
-              sock, IPPROTO_IP, IP_MULTICAST_IF, (char *)&localInterface,
-              sizeof(localInterface)) < 0) {
-         LOG(md, Error, "Setting local interface failed IP{} PORT{}",
-             multicast_ip, port);
-         close(sock);
-         return -1;
-      }
-
-      // Join the multicast group
-      struct ip_mreq mreq;
-      mreq.imr_multiaddr.s_addr = inet_addr(multicast_ip.c_str());
-      mreq.imr_interface.s_addr = inet_addr(interface_ip.c_str());
-      if (setsockopt(
-              sock, IPPROTO_IP, IP_ADD_MEMBERSHIP, (char *)&mreq,
-              sizeof(mreq)) < 0) {
-         LOG(md, Error, "adding multicast group failed IP{} PORT{}",
-             multicast_ip, port);
-         close(sock);
-         return -1;
-      }
-      LOG(md, Info, "Joined multicast group on IP: {}, Port: {}", multicast_ip,
-          port);
-      // struct timeval tv;
-      // tv.tv_sec = 5;
-      // tv.tv_usec = 0;
-      // setsockopt(sock, SOL_SOCKET, SO_RCVTIMEO, (const char*)&tv, sizeof tv);
-      return sock;
-   }
-   static int create_multicast_socket_assert(
-       const std::string &ip, int port, const std::string &market_data_ip) {
-      int socket =
-          TaifexUtils::create_multicast_socket(ip, port, market_data_ip);
-      if (socket == -1) {
-         bts_assert_msgf(false, "Couldn't open socket for {}:{}", ip, port);
-      }
-      return socket;
-   }
-   static void close_sockets(std::set<int> &sockets) {
-      for (int socket : sockets) {
-         try {
-            close(socket);
-         } catch (...) {
-            LOG(md, Error, "Couldnt properly close socket {}", socket);
-         }
-      }
-      sockets.clear();
-   }
-   static std::string getIP(const std::string &ipType) {
-   auto isLastTradingDay=[](){
-      return false;
-   };
-    const std::string regular_referenceIP = "225.0.140.141";
-    const std::string regular_refreshIP = "225.0.140.147";
-    const std::string regular_realTimeIP = "225.0.140.140";
-
-    const std::string afterHours_referenceIP = "225.10.140.141";
-    const std::string afterHours_refreshIP = "225.10.140.147";
-    const std::string afterHours_realTimeIP = "225.10.140.140";
-
-    // Get current time
-    std::time_t now = std::time(nullptr);
-    std::tm *localTime = std::localtime(&now);
-
-    int hour = localTime->tm_hour;
-   //  int minute = localTime->tm_min;
-
-    bool lastTradingDay = isLastTradingDay();
-
-    // Check if it's regular trading hours
-   //  bool isRegularTrading = (hour == 8 && minute >= 45) ||
-   //                          (hour > 8 && hour < 14 && (lastTradingDay ? minute < 30 : minute < 45));
-
-   //  // Check if it's after-hours trading
-    bool isAfterHoursTrading = !lastTradingDay && 
-                               ((hour == 15) || (hour > 15 && hour < 24) || (hour < 5));
-
-    // Determine the IP based on the time of day and type of IP requested
-    if (isAfterHoursTrading) {
-        if (ipType == "reference")
-            return afterHours_referenceIP;
-        if (ipType == "refresh")
-            return afterHours_refreshIP;
-        if (ipType == "realtime")
-            return afterHours_realTimeIP;
-    } //else if (isAfterHoursTrading) {
-   //  }
-        if (ipType == "reference")
-            return regular_referenceIP;
-        if (ipType == "refresh")
-            return regular_refreshIP;
-        if (ipType == "realtime")
-            return regular_realTimeIP;
-
-    // Return an error message if an invalid IP type is passed
-    return "Invalid IP type!";
-}
-   static std::string getBufferString(
-       const UNPROCESSED_MSG &buffer,
-       size_t offset = 0,
-       size_t length = std::numeric_limits<size_t>::max()) {
-
-      // Adjust length if it exceeds buffer size
-      if (length == std::numeric_limits<size_t>::max() ||
-          offset + length > buffer.size()) {
-         length = buffer.size() - offset;
-      }
-
-      if (offset >= buffer.size()) {
-         offset = buffer.size() - 1;
-      }
-
-      std::stringstream ss; // Use stringstream to build the string
-
-      for (size_t i = 0; i < length; ++i) {
-         ss << " offset: " << offset << ", i: " << i << ", value: " << std::hex
-            << std::uppercase << std::setw(2) << std::setfill('0')
-            << (unsigned int)buffer[offset + i] << "|";
-      }
-
-      return ss.str(); // Return the constructed string
-   }
-
-   static void printBuffer(
-       const UNPROCESSED_MSG &buffer,
-       size_t offset = 0,
-       size_t length = std::numeric_limits<size_t>::max()) {
-      // Adjust length if it exceeds buffer size
-      if (length == std::numeric_limits<size_t>::max() ||
-          offset + length > buffer.size()) {
-         length = buffer.size() - offset;
-      }
-      if (offset >= buffer.size())
-         offset = buffer.size() - 1;
-      for (size_t i = 0; i < length; ++i) {
-         printf(
-             " offset: %zu, i: %zu, value: %.2X|", offset, i,
-             (unsigned int)buffer[offset + i]);
-      }
-      //   std::cout << std::dec << std::endl;
-   }
-
-   //Replace std::localtime with localtime_s (use localtime_r for POSIX)
-   static std::string getCurrentDate() {
-      auto now = std::chrono::system_clock::now();
-      std::time_t now_c = std::chrono::system_clock::to_time_t(now);
-      std::tm local_tm = *std::localtime(&now_c);
-
-      char buffer[11]; // Buffer to hold formatted date
-      std::strftime(buffer, sizeof(buffer), "%Y-%m-%d", &local_tm);
-      return std::string(buffer);
-   }
-
-   static boost::posix_time::ptime convertToPTime(
-       const std::string &match_time) {
-      if (match_time.length() != 12) {
-         throw std::invalid_argument(
-             "match_time must be exactly 12 characters long");
-      }
-
-      // Extract hours, minutes, seconds, and microseconds from match_time
-      std::string hours = match_time.substr(0, 2);
-      std::string minutes = match_time.substr(2, 2);
-      std::string seconds = match_time.substr(4, 2);
-      std::string microseconds = match_time.substr(6, 6);
-
-      // Get the current date
-      std::string date = getCurrentDate();
-
-      // Combine the date and time
-      std::string datetime = date + " " + hours + ":" + minutes + ":" + seconds;
-
-      // Convert to ptime
-      boost::posix_time::ptime pt =
-          boost::posix_time::time_from_string(datetime);
-
-      // Add fractional seconds (microseconds)
-      boost::posix_time::ptime result =
-          pt + boost::posix_time::microseconds(std::stoi(microseconds));
-
-      return result;
-   }
-
-   static std::string convertToISO(
-       const std::string &match_time,
-       const std::string &format = "datetime_no_timezone") {
-      if (match_time.length() != 12) {
-         throw std::invalid_argument(
-             "in function convertToISO: match_time must be exactly 12 "
-             "characters "
-             "long: -> " +
-             match_time);
-      }
-
-      try {
-         // Extract time components
-         std::string hours = match_time.substr(0, 2);
-         std::string minutes = match_time.substr(2, 2);
-         std::string seconds = match_time.substr(4, 2);
-
-         // Validate and sanitize time components
-         int h = std::abs(std::stoi(hours)) % 24;
-         int m = std::abs(std::stoi(minutes)) % 60;
-         int s = std::abs(std::stoi(seconds)) % 60;
-
-         // Get the current date dynamically
-         std::string current_date = getCurrentDate();
-
-         // Construct the desired ISO format
-         std::ostringstream iso_stream;
-         if (format == "datetime_with_timezone") {
-            // Full date-time with time zone: "YYYY-MM-DDTHH:MM:SS+02:00" (for simplicity using UTC "Z")
-            iso_stream << current_date << "T" << std::setfill('0')
-                       << std::setw(2) << h << ":" << std::setfill('0')
-                       << std::setw(2) << m << ":" << std::setfill('0')
-                       << std::setw(2) << s << "Z"; // Assuming UTC for now
-         } else if (format == "datetime_no_timezone") {
-            // Date-time without time zone: "YYYY-MM-DDTHH:MM:SS"
-            iso_stream << current_date << "T" << std::setfill('0')
-                       << std::setw(2) << h << ":" << std::setfill('0')
-                       << std::setw(2) << m << ":" << std::setfill('0')
-                       << std::setw(2) << s;
-         } else if (format == "date_only") {
-            // Date only: "YYYY-MM-DD"
-            iso_stream << current_date;
-         } else {
-            throw std::invalid_argument(
-                "Unknown format specified for convertToISO: " + format);
-         }
-
-         return iso_stream.str();
-      } catch (const std::exception &e) {
-         std::cerr << "Error in convertToISO: " << e.what() << std::endl;
-         throw;
-      }
-   }
-
-   static std::string sanitizeISOString(const std::string &isoString) {
-      std::string result = isoString;
-
-      // Remove any whitespace
-      result.erase(
-          std::remove_if(result.begin(), result.end(), ::isspace),
-          result.end());
-
-      // Ensure it's at least 20 characters long (e.g., YYYY-MM-DDTHH:MM:SS)
-      if (result.size() != 20) {
-         if (result.size() < 20) {
-            // If it's too short, pad with zeros at the end (assuming it's missing time components)
-            result.append(20 - result.size(), '0');
-         } else {
-            // If it's too long, truncate to 20 characters
-            result = result.substr(0, 20);
-         }
-      }
-
-      // Check for and correct any time zone issues or missing Z (UTC indicator)
-      size_t posT = result.find('T');
-      if (posT != std::string::npos) {
-         size_t posZ = result.find('Z', posT);
-         if (posZ == std::string::npos) {
-            size_t posMinus = result.find('-', posT);
-            size_t posPlus = result.find('+', posT);
-            if (posMinus != std::string::npos && posMinus > posT) {
-               // Check if it's "-0" or "-00" or "-000", etc.
-               if (std::all_of(
-                       result.begin() + posMinus + 1, result.end(),
-                       [](char c) { return c == '0'; })) {
-                  result = result.substr(0, posMinus) + "Z";
-               }
-            } else if (posPlus != std::string::npos && posPlus > posT) {
-               // Check if it's "+0" or "+00" or "+000", etc.
-               if (std::all_of(
-                       result.begin() + posPlus + 1, result.end(),
-                       [](char c) { return c == '0'; })) {
-                  result = result.substr(0, posPlus) + "Z";
-               }
-            } else {
-               // If there's no timezone indicator, assume UTC and add 'Z'
-               result += "Z";
-            }
-         }
-      }
-
-      return result;
-   }
-
-   static std::string time_decodePackBCD(
-       const unsigned char *data, size_t length) {
-      std::stringstream ss;
-      for (size_t i = 0; i < length; ++i) {
-         int upper = (data[i] >> 4) & 0x0F;
-         int lower = data[i] & 0x0F;
-
-         // Check if the decoded values are valid digits
-         if (upper > 9 || lower > 9) {
-            // Instead of throwing an error, replace invalid digits with '0'
-            upper = upper > 9 ? 0 : upper;
-            lower = lower > 9 ? 0 : lower;
-         }
-
-         ss << upper << lower;
-      }
-
-      std::string result = ss.str();
-
-      // Ensure the result is exactly 12 characters long (hhmmssxxxxxx)
-      if (result.length() < 12) {
-         // If less than 12 characters, pad with zeros at the end
-         result = result + std::string(12 - result.length(), '0');
-      } else if (result.length() > 12) {
-         // If more than 12 characters, keep the first 12
-         result = result.substr(0, 12);
-      }
-
-      // Ensure the time part (hhmmss) is valid
-      int hours = std::stoi(result.substr(0, 2));
-      int minutes = std::stoi(result.substr(2, 2));
-      int seconds = std::stoi(result.substr(4, 2));
-
-      hours = std::min(23, std::max(0, hours));
-      minutes = std::min(59, std::max(0, minutes));
-      seconds = std::min(59, std::max(0, seconds));
-
-      // Reconstruct the result with valid time
-      std::stringstream final_ss;
-      final_ss << std::setfill('0') << std::setw(2) << hours
-               << std::setfill('0') << std::setw(2) << minutes
-               << std::setfill('0') << std::setw(2) << seconds
-               << result.substr(6); // Keep the last 6 characters (xxxxxx) as is
-
-      return final_ss.str();
-   }
-
-   // need to change parameters
-   static std::string decodePackBCD(const unsigned char *data, size_t length) {
-      std::stringstream ss;
-      for (size_t i = 0; i < length; ++i) {
-         // Extract the upper and lower nibble (4 bits each)
-         int upper = (data[i] >> 4) & 0x0F;
-         int lower = data[i] & 0x0F;
-
-         // Append both nibbles to the stringstream
-         ss << upper << lower;
-      }
-
-      std::string result = ss.str();
-
-      // Remove leading zeros
-      result.erase(0, result.find_first_not_of('0'));
-
-      // If the result is empty (all zeros), return "0"
-      return result.empty() ? "0" : result;
-   }
-   static std::string extractString(
-       const UNPROCESSED_MSG &buffer,
-       size_t &position,
-       size_t length) {
-      // Extract the substring from the buffer
-      std::string result(
-          buffer.begin() + position, buffer.begin() + position + length);
-      position += length;
-
-      // Log the raw result
-      return result;
-   }
-
-   // Helper function to extract a BCD value from the buffer and convert to
-   // unsigned long
-   static unsigned long extractBCD_STOUL(
-       const UNPROCESSED_MSG &buffer,
-       size_t &position,
-       size_t length) {
-      std::string bcd_str = decodePackBCD(&buffer[position], length);
-      position += length;
-      return std::stoul(bcd_str);
-   }
-   static std::string extractBCD_STRING(
-       const UNPROCESSED_MSG &buffer,
-       size_t &position,
-       size_t length) {
-      std::string bcd_str = decodePackBCD(&buffer[position], length);
-      position += length;
-      return bcd_str;
-   }
-   // Helper function to extract a single character and convert to bool
-   static bool extractBool(
-       const UNPROCESSED_MSG &buffer, size_t &position) {
-      bool result = (buffer[position] == 'Y');
-      position += 1;
-      return result;
-   }
-
-   static std::string extractBCDDate(
-       const UNPROCESSED_MSG &buffer,
-       size_t &current_position,
-       size_t size) {
-      if (size != 4) {
-         throw std::invalid_argument("BCD date must be 4 bytes long");
-      }
-
-      std::string date;
-      for (size_t i = 0; i < size; ++i) {
-         unsigned char byte = buffer[current_position++];
-         char high = (byte >> 4) + '0';
-         char low = (byte & 0x0F) + '0';
-         date.push_back(high);
-         date.push_back(low);
-      }
-
-      // Ensure the date is in YYYYMMDD format
-      if (date.length() != 8) {
-         throw std::runtime_error("Extracted date is not in YYYYMMDD format");
-      }
-
-      return date;
-   }
-
-   static std::string remove_spaces(const std::string &name) {
-    std::string removed;
-    removed.reserve(name.size());  // Reserve space to avoid reallocations
-    const char space = ' ';
-    for (auto c : name) {
-        if (c != space) {
-            removed += c;
-        }
-    }
-    return removed;
-   }
-   
-};
-
-} //namespace bts::md::taifex