all exectuion work

The previous regex was overly ambitious (and needlessly complex as a
result). The new regex is simpler, and doesn't impose bounds on month
and day ints, aside from requiring 1-2 digits.

Bugs (fixed):
  * The DataStreamExpressionInterpretter fails to parse certain date
    constants, such as "8/30/2024"

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

@@ -158,10 +158,6 @@ 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,7 +9,15 @@
 Letter                   [a-zA-Z]
 BooleanConstant          'true'|'false'
 CSStringConstant         \"([^\"\\]*|\\.)*\"
-DateTimeConstant         ([1-9]|'10'|'11'|'12')\/([1-9]|([1-2][0-9])|'30'|'31')\/[0-9][0-9][0-9]*
+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})
 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("bts-cloud://192.168.1.80:12345", ".*")
+            new("bluto://127.0.0.1:12345", ".*")
          },
          SubscriberRoutes = new() {
-            new("bts-cloud://192.168.1.80:12345", ".*")
+            new("bluto://127.0.0.1:12345", ".*")
          }
       };
       return rv;

src/lib/api/md/DatabentoMarketDataConfig.proto

@@ -13,4 +13,5 @@ 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,19 +5,24 @@ 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;
-  //uint32 num_depth_levels = 1;
-  //uint32 depth_throttle_ms = 2;
+  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;
   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,6 +156,8 @@ 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);
@@ -227,6 +229,7 @@ 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();
@@ -236,9 +239,7 @@ 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_CBOE:
-         break;
-         case bts::api::VENUE_TAIFEX:
+      case bts::api::VENUE_TAIFEX:
          break;
       case bts::api::VENUE_SIM:
       case bts::api::VENUE_SIMICE:
@@ -250,6 +251,7 @@ 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);
@@ -571,6 +573,13 @@ 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());
@@ -741,7 +750,6 @@ 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,6 +409,7 @@ 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;
@@ -752,6 +753,7 @@ 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

@@ -0,0 +1,12 @@
+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

@@ -0,0 +1,23 @@
+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,7 +40,6 @@ 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)
@@ -48,8 +47,7 @@ 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()),
-         cpp(s.contract_spec().cash_per_point()) {}
+         min_price_increment(s.contract_spec().low_tick()) {}
 };
 
 void from_json(const nlohmann::json &j, SecurityCommon &sc);
@@ -76,7 +74,6 @@ 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) {
@@ -101,7 +98,6 @@ 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,73 +78,78 @@ struct IncrementalBookData {
    std::map<domain::SecurityID, domain::Book> book_map;
 
    bool update_incremental(const IMarketDataListener::BookOp & op) {
-      // 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
+    // Retrieve the Book from the map and update it
+    domain::Book* book = get_book(op.id);
+    if (book == nullptr)
+        return false;
 
-      //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();
+    // 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";
 
-      size_t insert_level = 0;
-      bool level_found = false;
+    // 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";
+    }
 
-      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;
+    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 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;
+            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];
         }
-        // insert the new level with its price
-        //book_side[insert_level].price.to_double() = op.price_size.price();
+    } 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;
+}
 
-      }
-        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;
-      bts::api::Side side;
+      domain::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();
+            exchange_order_id = native_state->exchange_order_id(); //use this 
             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,9 +4,12 @@
 #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"
@@ -21,23 +24,30 @@ 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());
    }
-   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);
+   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);
+   }
 }
 
 } // namespace bts::orderman

src/lib/orderman/gateways/GeneralExecutionGateway.cpp

@@ -0,0 +1,85 @@
+#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

@@ -0,0 +1,29 @@
+#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,3 +1,6 @@
 #include "IFIXGenericAdapter.h"
 
-namespace bts::orderman {}
+namespace bts::orderman {
+
+    IFIXGenericAdapter::~IFIXGenericAdapter(){}
+}

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 OnOrderUpdate = std::function<void(OrderStateWithFills const &)>;
+using OnExecutionReport = std::function<void(ExecutionReport const &, const trace::CorID &)>;
 
 // 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 OnOrderUpdate &);
-   virtual void launch_new_order(const CreateOrderMsg &);
-   virtual void modify_order(const CancelReplaceOrderMsg &);
-   virtual void cancel_order(const CancelOrderMsg &);
+//    // 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;
    // 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

@@ -0,0 +1,406 @@
+#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

@@ -0,0 +1,27 @@
+#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

@@ -0,0 +1,175 @@
+#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

@@ -0,0 +1,52 @@
+#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

@@ -0,0 +1,82 @@
+#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

@@ -0,0 +1,28 @@
+#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();
+   return new EMNativeOrder(exchange);
 }
 
 NativeOrderBase *OrderFactory::make_native_order(

src/lib/orderman/orders/EMNativeOrder.cpp

@@ -5,6 +5,7 @@
 #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"
@@ -52,10 +53,10 @@ struct InflightUpdateState {
 };
 
 struct EMNativeOrder::PImpl {
-   PImpl()
+   PImpl(const api::Venue exchange)
        : instance_id(EMNativeOrder::count++), services(nullptr),
          session_services(), sec(nullptr), previous_working_position(0),
-         definition(), state(), exchange_order_id() {}
+         definition(), state(), exchange_order_id(), exchange(exchange) {}
    const int instance_id;
    bool inited = false;
    OrderServices *services;
@@ -67,7 +68,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
@@ -82,7 +83,7 @@ struct EMNativeOrder::PImpl {
    RoutingRecord routing_record;
 };
 
-EMNativeOrder::EMNativeOrder() : me(new PImpl()) {
+EMNativeOrder::EMNativeOrder(const api::Venue exchange) : me(new PImpl(exchange)) {
    LOG(OrderMan, Info,
        "EMNativeOrder::EMNativeOrder instance_id: {} t_id: {} t_name: {}",
        me->instance_id, os::get_thread_id(), os::get_application_name());
@@ -501,11 +502,19 @@ 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
-   me->session_services = &me->services->get_session_services("deribit_1");
+   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);
 
    // Checks for the following:
    // 1. Gateway connection

src/lib/orderman/orders/EMNativeOrder.h

@@ -1,6 +1,7 @@
 #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"
@@ -19,7 +20,7 @@ class SessionServices;
 // Wrapper around Emumba's order
 class EMNativeOrder : public NativeOrderBase {
  public:
-   EMNativeOrder();
+   EMNativeOrder(const api::Venue exchange = api::VENUE_DERIBIT);
    ~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) {
+         if (me->in_sim_mode) {       //CME EXECUTION NEEDS TO BE UPDATED
             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) {
+      if (me->in_sim_mode) {  //CME EXECUTION NEEDS TO BE UPDATED
          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) {
+      if (me->in_sim_mode) { //CME EXECUTION NEEDS TO BE UPDATED
          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);
+   c.order_type = get_glbx_native_ord_type(me->definition); //CME EXECUTION NEEDS TO BE UPDATED
    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,6 +277,7 @@ 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/TaifexDataHandler.h"
+#include "server/sparkd/md/taifex/TaifexDataGateway.h"
 
 #include <bits/stdint-uintn.h>
 #include <boost/filesystem.hpp>
@@ -43,6 +43,7 @@
 #include <emumba-deribit-connector/deribit_utils.h>
 #include <memory>
 #include <optional>
+#include <string>
 #include <unordered_map>
 #include <vector>
 
@@ -117,10 +118,6 @@ 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()
@@ -142,9 +139,9 @@ void MarketDataGateway::run() {
       bts_assert(me_->md_config.emumba_deribit().size() == 1);
       _start_emumba_deribit(me_->md_config.emumba_deribit().at(0));
    }
-   me_->start_glbx();
-   me_->start_taifex(); 
-   me_->start_cboe();
+   start_glbx(me_->md_config);
+   start_taifex(me_->md_config); 
+   start_cboe(me_->md_config);
    _start_replays(me_->md_config);
    while (true) {
       std::this_thread::sleep_for(std::chrono::seconds(1));
@@ -247,16 +244,16 @@ void MarketDataGateway::unsubscribe(MarketDataListenerID id) {
 }
 /// glbx //////////////////////////////////////////////////////////////////////
 //////////////////////////////////////////////////////////////////////////////
-void MarketDataGateway::PImpl::start_glbx(){
-     for (const auto &glbx : md_config.glbx()) {
+void MarketDataGateway::start_glbx(const bts::api::md::MarketDataConfig &cfg){
+     for (const auto &glbx : cfg.glbx()) {
       if(glbx.use_databento()){
-      auto p = venue_listeners.emplace_back(
+      auto p = me_->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,
-            sec_id_store,
+            me_->sec_id_store,
             glbx); 
       }).detach();
       }
@@ -375,29 +372,31 @@ void MarketDataGateway::_start_replays(
        os::get_thread_id(), os::get_thread_name());
 }
 
-void MarketDataGateway::PImpl::start_taifex() {
+void MarketDataGateway::start_taifex(const bts::api::md::MarketDataConfig &cfg) {
    // taifex
-   for (const auto &taifex : md_config.taifex()) {
-      auto p = venue_listeners.emplace_back(
+   int counter = 0;
+   for (const auto &taifex : cfg.taifex()) {
+      counter++;
+      auto p = me_->venue_listeners.emplace_back(
           std::make_shared<VenueListenerData>(bts::api::VENUE_TAIFEX));
-      std::string tid_name = "taifex";
+      std::string tid_name = "taifex" +std::to_string(counter);
       bts::threading::make_thread(tid_name, [&]() {
-         bts::md::taifex::TaifexDatahandler::run(
-          taifex.read_from_file(), p->listener, sec_id_store);
+         bts::md::taifex::TaifexDataGateway::run(
+          taifex, p->listener, me_->sec_id_store);
       }).detach();
       
    }
 }
-void MarketDataGateway::PImpl::start_cboe(){
+void MarketDataGateway::start_cboe(const bts::api::md::MarketDataConfig &cfg){
     // cboe
-    for (const api::md::CboeMarketDataConfig &cboe : md_config.cboe()) {
-       auto p = venue_listeners.emplace_back(
+    for (const api::md::CboeMarketDataConfig &cboe : cfg.cboe()) {
+       auto p = me_->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, 
-            sec_id_store,
+            me_->sec_id_store,
             cboe);
        }).detach();
 

src/server/sparkd/md/MarketDataGateway.h

@@ -36,6 +36,9 @@ 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,7 +29,8 @@ 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().dataset(),
+       config.databento_settings().filter_securities_one_year());
 }
 
 } // namespace bts::md::cboe

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

@@ -1,6 +1,7 @@
 #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"
@@ -47,6 +48,7 @@ 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;
@@ -62,12 +64,6 @@ 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:
@@ -81,6 +77,16 @@ 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;
@@ -177,7 +183,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);
+            client.Subscribe({"SPX.OPT","SPXW.OPT"}, databento::Schema::Definition, databento::SType::Parent,databento::UnixNanos{});
             client.Subscribe({"SPX.OPT","SPXW.OPT"}, databento::Schema::Mbp1, databento::SType::Parent);
             break;
         case api::domain::VENUE_GLBX:
@@ -191,7 +197,7 @@ void DatabentoDatahandler::run(
     client.Start();
     while (true) {
     try {
-    auto* record = client.NextRecord(std::chrono::seconds{5});
+    auto* record = client.NextRecord(std::chrono::seconds{500});
     if (record == nullptr) {
         LOG(md, Error, "Timed out waiting for data");
         break;
@@ -206,10 +212,24 @@ void DatabentoDatahandler::run(
     std::string err = error.what();
     LOG(md, Error, "Caught TCP error: {} \nAttempting reconnect\n",err);
     client.Reconnect();
-  } catch (const std::exception& exc) {
+  } 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) {
     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,6 +14,7 @@ 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,7 +30,8 @@ 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().dataset(),
+       config.databento_settings().filter_securities_one_year());
 }
 
 } // namespace bts::api::md::glbx 

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

@@ -0,0 +1,848 @@
+#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/TaifexDataGateway.h

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

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

@@ -0,0 +1,232 @@
+#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

@@ -0,0 +1,140 @@
+#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

@@ -0,0 +1,508 @@
+#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