spark/src/lib/orderman/orders/NativeOrder.cpp

#include "NativeOrder.h"

#include "lib/api/services/OrderService.pb.h"
#include "lib/domain_model/Security.h"
#include "lib/jetfix/codec/FIXFragment.h"
#include "lib/jetfix/codec/FieldDefinitions.h"
#include "lib/logging/logging.h"
#include "lib/orderman/order_factory/IOrderFactory.h"
#include "lib/orderman/order_router/OrderRouter.h"
#include "lib/orderman/orderman_utils/ExecutionReport.h"
#include "lib/orderman/orderman_utils/fix_order_codec.h"
#include "lib/orderman/orders/PreActivationOrderServices.h"
#include "lib/orderman/securities/ISecurityDB.h"
#include "lib/utils/PBUtils.h"
#include "lib/utils/bts_chrono.h"

#include "lib/orderman/actors/order_manager/OrderServices.h"

#include <algorithm>
#include <sys/time.h>

namespace bts {
namespace orderman {

using namespace api::orderman;

OrderStatus map_external_status(NativeOrderStatus);

struct NativeOrder::PImpl {
   PImpl()
       : services(nullptr), session_services(), sec(nullptr),
         previous_position_contribution(0), definition(), state(),
         price_encoding_precision(7), exchange_order_id() {
      exchange_order_id.reserve(100);
   }

   OrderServices *services;
   SessionServices *session_services;

   const api::Security *sec;
   Quantity previous_position_contribution;

   OrderDefinition definition;
   OrderState state;
   int price_encoding_precision;
   std::string exchange_order_id;

   jetfix::FIXFragment update_encoding;
   jetfix::FIXFragment::Bookmark<jetfix::fix::Price> update_price_bm;
   jetfix::FIXFragment::Bookmark<jetfix::fix::OrderQty> update_quantity_bm;
   jetfix::FIXFragment::Bookmark<jetfix::fix::TransactTimeStr>
       update_transact_time_bm;

   jetfix::FIXFragment create_encoding;
   jetfix::FIXFragment::Bookmark<jetfix::fix::Price> create_price_bm;
   jetfix::FIXFragment::Bookmark<jetfix::fix::OrderQty> create_quantity_bm;
   jetfix::FIXFragment::Bookmark<jetfix::fix::TransactTimeStr>
       create_transact_time_bm;

   jetfix::FIXFragment cancel_encoding;
   jetfix::FIXFragment::Bookmark<jetfix::fix::TransactTimeStr>
       cancel_transact_time_bm;

   OrderServices::ExeReportSubID exe_report_sid;
   OrderServices::CancelRejectSubID cancel_reject_sid;

   bool buffered_cancel = false;
   std::optional<std::pair<Price, Quantity>> buffered_update;
};

NativeOrder::NativeOrder() : me(new PImpl()) {}

OrderDefinition const &NativeOrder::definition() const {
   return me->definition;
}

void NativeOrder::on_init_new(OrderDefinition const &def,
                              const PreActivationOrderServices &serv) {
   // order was just created.
   me->definition = def;

   switch (def.custom_def().custom_case()) {
   case CustomOrderDefinition::kMarket:
      me->state.mutable_market();
      break;
   case CustomOrderDefinition::kLimit:
      me->state.mutable_limit();
      break;
   case CustomOrderDefinition::kStop:
      me->state.mutable_stop();
      break;
   case CustomOrderDefinition::kStopLimit:
      me->state.mutable_stop_limit();
      break;
   default:
      bts_assert(false);
      break;
   }

   me->sec = serv.sec_db.resolve(SecurityKey(def.security_key()));

   if (!me->sec) {
      auto msg = fmt::format("Unknown security: {}", def.security_key());
      me->state.set_last_msg(msg);
      transition(ORDER_STATUS_REJECTED_BY_SPARK, msg);
   } else {
      auto rq = serv.order_router.get_routing_record(
          UserID(def.user_id()), AccountID(def.account_id()), *me->sec);
      if (rq.has_route) {
         transition(ORDER_STATUS_INACTIVE);
         auto &native_state = get_native_state(me->state);
         *native_state.mutable_routing_record() = rq.routing_record;

         init_fix();
      } else {
         auto msg = fmt::format("Could not route order: {}", rq.error_msg);
         LOG(OrderMan, Error, "{}", msg);
         transition(ORDER_STATUS_REJECTED_BY_SPARK, msg);
      }
   }
}

void NativeOrder::on_init_recovered(OrderRecord const &rec,
                                    const PreActivationOrderServices &serv) {
   // order was recovered from our orders database or from an execution report

   auto const &def = rec.def();

   me->definition = def;
   me->state = rec.state();
   me->sec = serv.sec_db.resolve(SecurityKey(def.security_key()));
   me->previous_position_contribution = Quantity(rec.state().filled_quantity());

   auto const &native_state = get_native_state(me->state);
   switch (me->state.status()) {

   case ORDER_STATUS_WORKING:
   case ORDER_STATUS_PENDING_WORKING:
   case ORDER_STATUS_DISCONNECTED: {

      // bail if we can't find the security for the order
      if (!me->sec) {
         LOG(OrderMan, Error,
             "Unknown security {} referenced by recovered "
             "order {}. This is a serious "
             "problem which has to be corrected manually. ",
             def.security_key(), rec.key());
         transition_to_error(
             NativeOrderState::SUB_ERROR_UNKNOWN_SECURITY,
             fmt::format("Unknown security: {}", def.security_key()));
         return;
      }

      // bail if the session we used for the order is no longer valid
      auto const &session = native_state.routing_record().session();
      if (!serv.order_router.is_valid_session(session)) {
         LOG(OrderMan, Error,
             "Unknown JetFIX session {} referenced by recovered "
             "order {}. This is a serious "
             "problem which has to be corrected manually. ",
             session, rec.key());
         transition_to_error(NativeOrderState::SUB_ERROR_UNKNOWN_FIX_SESSION,
                             fmt::format("Unknown FIX session: {}", session));
         return;
      }

      // if we made it this far we are good.
      init_fix();
      if (me->state.status() != ORDER_STATUS_PENDING_WORKING) {
         me->exchange_order_id = native_state.exchange_order_id();
         add_exchange_order_id_to_fix();
      }
      return;
   }

   case ORDER_STATUS_REJECTED_BY_SPARK:
   case ORDER_STATUS_ERROR:
   case ORDER_STATUS_COMPLETED:
   case ORDER_STATUS_CANCELED:
      // terminal states, nothing to do
      return;

   case ORDER_STATUS_INACTIVE:
   case OrderStatus_INT_MIN_SENTINEL_DO_NOT_USE_:
   case OrderStatus_INT_MAX_SENTINEL_DO_NOT_USE_:
      bts_assert(false);
      return;
   }
}

api::orderman::OrderDefinition
NativeOrder::on_init_from_exe_report(ExecutionReport const &er,
                                     const PreActivationOrderServices &os) {
   auto k = er.order_key;

   OrderRecord orecord;
   orecord.set_key(k.value());
   orecord.set_parent_key(null_order_key().value());

   auto &def = *orecord.mutable_def();
   def.set_security_key(er.security->id());
   def.set_side(er.side);

   auto sec = os.sec_db.resolve(SecurityKey(def.security_key()));
   auto rev_rec = os.order_router.get_reverse_routing_record(
       er.cfg->name(), to_trim_str(er.operator_id),
       to_trim_str(er.account_name), sec);

   if (rev_rec.ok) {
      def.set_user(rev_rec.user);
      def.set_user_id(rev_rec.user_id.value());
      def.set_account(rev_rec.account);
      def.set_account_id(rev_rec.account_id.value());
   } else {
      LOG(OrderMan, Error,
          "Could not recover account and user information from order {}"
          ". The following information is available: {}. ",
          k.value(), rev_rec.error_msg);
      def.set_user("");
      def.set_user_id(0);
      def.set_account("");
      def.set_account_id(0);
   }

   def.set_time_in_force(er.time_in_force);
   def.set_manual_order_indicator(er.manual_order_indicator);
   def.set_cti_code(CTICode(er.cfg->cti_code()[0]));
   def.set_customer_or_firm(CustomerOrFirm(er.cfg->customer_or_firm()));

   auto &state = *orecord.mutable_state();
   state.set_status(map_external_status(er.order_status));
   state.set_price(er.price.value());
   state.set_quantity(er.quantity.value());
   state.set_filled_quantity(er.filled_quantity.value());
   state.set_last_price(er.last_price.value());
   state.set_avg_price(er.last_price.value());
   state.set_last_msg(er.msg);
   *state.mutable_last_change() = utils::pb::pb_now();

   switch (er.order_type) {
   case NATIVE_ORDER_TYPE_MARKET:
      def.mutable_custom_def()->mutable_market();
      state.mutable_market();
      break;
   case NATIVE_ORDER_TYPE_LIMIT:
      def.mutable_custom_def()->mutable_limit();
      state.mutable_market();
      break;
   case NATIVE_ORDER_TYPE_STOP: {
      auto &stop_def = *def.mutable_custom_def()->mutable_stop();
      stop_def.set_stop_price(er.stop_price.value());
      state.mutable_stop();
      break;
   }
   case NATIVE_ORDER_TYPE_STOP_LIMIT: {
      auto &stop_limit_def = *def.mutable_custom_def()->mutable_stop_limit();
      stop_limit_def.set_stop_price(er.stop_price.value());
      state.mutable_stop_limit();
      break;
   }
   case NATIVE_ORDER_TYPE_MARKET_LIMIT:
   case NATIVE_ORDER_TYPE_UNDEFINED:
   case NativeOrderType_INT_MIN_SENTINEL_DO_NOT_USE_:
   case NativeOrderType_INT_MAX_SENTINEL_DO_NOT_USE_:
      bts_assert(false);
      break;
   }

   auto &native_state = get_native_state(state);
   native_state.set_exchange_order_id(to_trim_str(er.exchange_order_id));
   native_state.set_last_msg_seq_num(er.msg_seq_num);
   native_state.set_native_type(er.order_type);

   auto &rr = *native_state.mutable_routing_record();
   rr.set_session(er.cfg->name());
   rr.set_account(to_trim_str(er.account_name));
   rr.set_operator_id(to_trim_str(er.operator_id));
   rr.set_cti_code(CTICode(er.cfg->cti_code()[0]));
   rr.set_customer_or_firm(CustomerOrFirm(er.cfg->customer_or_firm()));
   rr.set_location(er.cfg->sender_locationid());
   on_init_recovered(orecord, os);
   return orecord.def();
}

NativeOrder::~NativeOrder() {}

void NativeOrder::warm_up_activate(OrderServicesData &data,
                                   OrderServices &services) {
   bts_assert(!me->services);
   me->services = &services;

   init_order_services(data, services);

   // save these so we can re-instate them after we are done with the warmup
   auto before_state = me->state.status();
   auto before_msg = me->state.last_msg();

   _on_activate(NoLog, NoLog, Price(0), Quantity(0), services);

   // undo warm up side effects
   me->state.set_status(before_state);
   me->state.set_last_msg(before_msg);

   me->services->unsubscribe_from_exe_report(*me->session_services,
                                             me->exe_report_sid);
   me->services->unsubscribe_from_cancel_reject(*me->session_services,
                                                me->cancel_reject_sid);
   me->services->unsubscribe_from_gateway_status(*me->session_services);
   me->services = nullptr;
}

OrderOpResult
NativeOrder::_on_activate(log::Level reject_level, log::Level transition_level,
                          Price p, Quantity q,
                          bts::orderman::OrderServices &services) {
   me->state.set_price(p.value());
   me->state.set_quantity(q.value());
   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());
   me->price_encoding_precision =
       me->services->price_encoding_precision(*me->session_services);

   // is our gateway connected?
   if (!me->services->is_gateway_connected(*me->session_services)) {
      LOG(OrderMan, reject_level,
          "Attempt to send order to a disconnected gateway. "
          "Dropping order {}",
          key().value());
      auto msg = "Exchange gateway is down";
      transition(transition_level, ORDER_STATUS_REJECTED_BY_SPARK, msg);
      return OrderOpResult{ORDER_ERROR_GATEWAY_NOT_READY, msg};
   }

   // can we trade?
   auto lqr = me->services->limiter().can_trade(
       UserID(me->definition.user_id()),
       AccountID(me->definition.account_id()));
   if (!lqr.ok) {
      LOG(OrderMan, reject_level,
          "Unauthorized order attempted. User: {} on "
          "Account: {}. Msg: {} Order: {}. Dropping order",
          me->definition.user(), me->definition.account(), lqr.error_msg,
          key().value());

      auto msg = fmt::format("Unauthorized order: {}", lqr.error_msg);
      transition(transition_level, ORDER_STATUS_REJECTED_BY_SPARK, msg);
      return OrderOpResult{ORDER_ERROR_UNAUTHORIZED, msg};
   }

   // can we trade this security, with this size?
   lqr = me->services->limiter().can_send_new_order(
       AccountID(me->definition.account_id()), *me->sec, me->definition.side(),
       q);

   if (!lqr.ok) {
      LOG(OrderMan, reject_level,
          "New order would breach limit. User: {} Account: "
          "{} Msg: {} Order: {}. Dropping order",
          me->definition.user(), me->definition.account(), lqr.error_msg,
          key().value());

      auto msg = fmt::format("Order would breach limit: {}", lqr.error_msg);
      transition(transition_level, ORDER_STATUS_REJECTED_BY_SPARK, msg);
      return OrderOpResult{ORDER_ERROR_LIMIT_BREACH, msg};
   }

   // are we under the messaging threshold?
   if (!me->services->governor(*me->session_services).can_send()) {
      LOG(OrderMan, reject_level,
          "New order would cause breach of messaging "
          "limits. Dropping order {}",
          key().value());
      auto msg = "Messaging threshold breached";
      transition(transition_level, ORDER_STATUS_REJECTED_BY_SPARK, msg);
      return OrderOpResult(ORDER_ERROR_MESSAGING_LIMIT_BREACH, msg);
   }

   // if we are here we are good.
   // subscribe to usefull events
   me->exe_report_sid = me->services->subscribe_to_exe_report(
       *me->session_services, key(),
       [this](auto const &r) { this->on_exe_report(r); });
   me->cancel_reject_sid = me->services->subscribe_to_cancel_reject(
       *me->session_services, key(),
       [this](auto const &cr) { this->on_cancel_reject(cr); });
   me->services->subscribe_to_gateway_status(
       *me->session_services,
       [this](auto const &gw) { this->on_gateway_status(gw); });

   // send out order
   auto &encoding = me->create_encoding;

   switch (me->definition.custom_def().custom_case()) {
   case CustomOrderDefinition::kLimit:
   case CustomOrderDefinition::kStopLimit:
      me->create_price_bm = encoding.overwrite(
          me->create_price_bm,
          securities::encode_exchange_price(*me->sec, p).value(),
          me->price_encoding_precision, price_sz());
      break;
   case CustomOrderDefinition::kStop:
   case CustomOrderDefinition::kMarket:
      break;
   default:
      bts_assert(false);
      break;
   }

   me->create_quantity_bm =
       encoding.overwrite(me->create_quantity_bm, q.value(), quantity_sz());
   auto const &ts = services.get_time_str();
   me->create_transact_time_bm = encoding.overwrite(
       me->create_transact_time_bm, &ts.value()[0], ts.value().size());
   return OrderOpResult{};
}

OrderOpResult NativeOrder::on_activate(Price p, Quantity q,
                                       bts::orderman::OrderServices &services) {
   bts_assert(!me->services);
   me->services = &services;
   auto result = OrderOpResult{};
   switch (me->state.status()) {

   case ORDER_STATUS_INACTIVE: {
      auto scoped_log =
          services.log_scope("NativeOrder::on_activate", key().value());
      // this is a brand new order

      result = _on_activate(Warn, Info, p, q, services);
      if (result.ok()) {
         me->services->queue_fix(*me->session_services,
                                 jetfix::MessageTypes::NewOrderSingle,
                                 me->create_encoding);

         me->services->governor(*me->session_services).record_sent();
         me->state.set_status(ORDER_STATUS_PENDING_WORKING);

         LOG(OrderMan, Info, "Sent new order with key {} to gateway",
             key().value());

         // Early return so we skip the expensive report_state()
         // call. It is OK to spkip it because we are bound to report
         // our state once the excange aknowledges the order.
         update_working_position();
         return OrderOpResult{};
      }

      // When the order fails its limits checks we fall through the
      // switch to report the error state at the end of this function
      break;
   }

      // all cases below are for orders recovered from disk
   case ORDER_STATUS_WORKING:
      me->session_services = &me->services->get_session_services(
          get_native_state(me->state).routing_record().session());
      me->services->subscribe_to_exe_report(
          *me->session_services, key(),
          [this](auto const &r) { this->on_exe_report(r); });
      me->services->subscribe_to_cancel_reject(
          *me->session_services, key(),
          [this](auto const &cr) { this->on_cancel_reject(cr); });
      me->services->subscribe_to_gateway_status(
          *me->session_services,
          [this](auto const &gw) { this->on_gateway_status(gw); });

      if (!me->services->is_gateway_connected(*me->session_services)) {
         result.error = ORDER_ERROR_GATEWAY_NOT_READY;
         result.error_msg = "Gateway is disconnected";
         transition(ORDER_STATUS_DISCONNECTED);
      }
      break;

   case ORDER_STATUS_DISCONNECTED:
      me->session_services = &me->services->get_session_services(
          get_native_state(me->state).routing_record().session());
      me->services->subscribe_to_exe_report(
          *me->session_services, key(),
          [this](auto const &r) { this->on_exe_report(r); });
      me->services->subscribe_to_cancel_reject(
          *me->session_services, key(),
          [this](auto const &cr) { this->on_cancel_reject(cr); });
      me->services->subscribe_to_gateway_status(
          *me->session_services,
          [this](auto const &gw) { this->on_gateway_status(gw); });

      if (me->services->is_gateway_connected(*me->session_services))
         transition(ORDER_STATUS_WORKING);
      break;

   case ORDER_STATUS_PENDING_WORKING:
      me->session_services = &me->services->get_session_services(
          get_native_state(me->state).routing_record().session());
      me->services->subscribe_to_exe_report(
          *me->session_services, key(),
          [this](auto const &r) { this->on_exe_report(r); });
      me->services->subscribe_to_cancel_reject(
          *me->session_services, key(),
          [this](auto const &cr) { this->on_cancel_reject(cr); });
      me->services->subscribe_to_gateway_status(
          *me->session_services,
          [this](auto const &gw) { this->on_gateway_status(gw); });
      break;

   case ORDER_STATUS_REJECTED_BY_SPARK:
   case ORDER_STATUS_COMPLETED:
   case ORDER_STATUS_CANCELED:
   case ORDER_STATUS_ERROR:
      break;

   case OrderStatus_INT_MIN_SENTINEL_DO_NOT_USE_:
   case OrderStatus_INT_MAX_SENTINEL_DO_NOT_USE_:
      bts_assert(false);
      break;
   }
   report_state();
   return result;
}

OrderOpResult NativeOrder::on_update(Price p, Quantity q) {
   auto scoped_log =
       me->services->log_scope("NativeOrder::on_update", key().value());
   auto result = OrderOpResult{};

   switch (me->state.status()) {

   case ORDER_STATUS_WORKING: {

      auto previous_working_qty = 0;
      if (me->state.quantity() > me->state.filled_quantity())
         previous_working_qty =
             me->state.quantity() - me->state.filled_quantity();

      auto current_working_qty = 0;
      if (q.value() > me->state.filled_quantity())
         current_working_qty = q.value() - me->state.filled_quantity();

      auto lqr = me->services->limiter().can_update_order(
          AccountID(me->definition.account_id()), *me->sec,
          me->definition.side(), Quantity(previous_working_qty),
          Quantity(current_working_qty));

      // are we under the position limits?
      if (!lqr.ok) {
         auto msg = fmt::format("Cannot update order due to limit breach: {}",
                                lqr.error_msg);
         LOG(OrderMan, Error,
             "Order update failed due to limit breach. Order: {} Msg: {}",
             key().value(), lqr.error_msg);

         // if we are not being driven by an algo we just signal the
         // error but keep running. otherwise we stop.
         if (parent_key() == null_order_key())
            transition(me->state.status(), msg);
         else {
            me->state.set_last_msg(msg);
            on_cancel();
         }
         result.error = api::orderman::ORDER_ERROR_LIMIT_BREACH;
         result.error_msg = lqr.error_msg;
         break;
      }

      // are we under the messaging threshold?
      if (!me->services->governor(*me->session_services).can_send()) {
         LOG(OrderMan, Warn,
             "Cannot update order {} because of messaging limits. ",
             key().value());
         auto msg = "Messaging threshold breached";

         // if we are not being driven by an algo we just signal the
         // error but keep running. otherwise we stop.
         if (parent_key() == null_order_key())
            transition(me->state.status(), msg);
         else {
            me->state.set_last_msg(msg);
            on_cancel();
         }
         result.error = api::orderman::ORDER_ERROR_MESSAGING_LIMIT_BREACH;
         result.error_msg = msg;
         break;
      }

      // all good
      auto &encoding = me->update_encoding;
      me->update_price_bm = encoding.overwrite(
          me->update_price_bm,
          securities::encode_exchange_price(*me->sec, p).value(),
          me->price_encoding_precision, price_sz());
      me->update_quantity_bm =
          encoding.overwrite(me->update_quantity_bm, q.value(), quantity_sz());

      auto const &ts = me->services->get_time_str();
      me->update_transact_time_bm = encoding.overwrite(
          me->update_transact_time_bm, &ts.value()[0], ts.value().size());

      me->services->queue_fix(*me->session_services,
                              jetfix::MessageTypes::CancelReplace,
                              me->update_encoding);

      me->services->governor(*me->session_services).record_sent();
      update_working_position();
      if (me->state.last_msg().size())
         me->state.set_last_msg("");

      // early return so we skip report_state (for speed)
      return OrderOpResult{};
   }

   case ORDER_STATUS_DISCONNECTED:
   case ORDER_STATUS_PENDING_WORKING:
      LOG(OrderMan, Warn,
          "Cannot apply update to order {}. Update: {}@{}. Buffering",
          key().value(), q.value(), p.value());
      me->buffered_update.emplace(p, q);
      break;

   case ORDER_STATUS_REJECTED_BY_SPARK:
   case ORDER_STATUS_COMPLETED:
   case ORDER_STATUS_CANCELED:
   case ORDER_STATUS_ERROR:
      break;

   case ORDER_STATUS_INACTIVE:
   case OrderStatus_INT_MIN_SENTINEL_DO_NOT_USE_:
   case OrderStatus_INT_MAX_SENTINEL_DO_NOT_USE_:
      bts_assert(false);
      break;
   }
   report_state();
   return result;
}

OrderOpResult NativeOrder::on_update(api::orderman::Update const &update) {
   return on_update(Price(update.price()), Quantity(update.quantity()));
}

OrderOpResult NativeOrder::on_detach() { return OrderOpResult{}; }

void NativeOrder::on_attached(
    api::orderman::OrderDefinition const &new_parent) {
   switch (me->state.status()) {

   case ORDER_STATUS_PENDING_WORKING:
   case ORDER_STATUS_DISCONNECTED:
   case ORDER_STATUS_WORKING: {
      LOG(OrderMan, Info, "Native order {} has been attached to by {}",
          key().value(), utils::pb::to_string(new_parent));

      // Update FIX tags to reflect new parent's information
      auto rq = me->services->route_order(UserID(new_parent.user_id()),
                                          AccountID(new_parent.account_id()),
                                          *me->sec);

      if (!rq.has_route) {
         auto msg = fmt::format("Cannot update order routing information: {}",
                                rq.error_msg);
         transition(me->state.status(), msg);
         report_state();
         return;
      }

      auto &native_state = get_native_state(me->state);
      *native_state.mutable_routing_record() = rq.routing_record;
      init_fix();
      if (has_exchange_order_id())
         add_exchange_order_id_to_fix();

      // Update user, account and manual order indicator based on new parent
      me->definition.set_manual_order_indicator(
          api::orderman::ManualOrderIndicator::MANUAL_ORDER_INDICATOR_NO);
      me->definition.set_user(new_parent.user());
      me->definition.set_user_id(new_parent.user_id());
      me->definition.set_account(new_parent.account());
      me->definition.set_account_id(new_parent.account_id());
      me->services->report_definition(me->definition);
      report_state();
      break;
   }

   case ORDER_STATUS_CANCELED:
   case ORDER_STATUS_REJECTED_BY_SPARK:
   case ORDER_STATUS_COMPLETED:
   case ORDER_STATUS_ERROR:
      break;

   case ORDER_STATUS_INACTIVE:
   case OrderStatus_INT_MIN_SENTINEL_DO_NOT_USE_:
   case OrderStatus_INT_MAX_SENTINEL_DO_NOT_USE_:
      bts_assert(false);
      break;
   }
}

OrderOpResult NativeOrder::on_cancel() {
   auto scoped_log =
       me->services->log_scope("NativeOrder::on_cancel", key().value());
   auto result = OrderOpResult{};

   switch (me->state.status()) {

   case ORDER_STATUS_WORKING: {
      auto &encoding = me->cancel_encoding;
      auto const &ts = me->services->get_time_str();
      me->cancel_transact_time_bm = encoding.overwrite(
          me->cancel_transact_time_bm, &ts.value()[0], ts.value().size());

      me->services->queue_fix(*me->session_services,
                              jetfix::MessageTypes::Cancel,
                              me->cancel_encoding);

      break;
   }

   case ORDER_STATUS_CANCELED:
   case ORDER_STATUS_REJECTED_BY_SPARK:
   case ORDER_STATUS_COMPLETED:
   case ORDER_STATUS_ERROR:
      break;

   case ORDER_STATUS_PENDING_WORKING: {
      me->buffered_cancel = true;
      LOG(OrderMan, Warn,
          "Cannot cancel order {} while in the pending working state. "
          "Buffering cancel",
          key().value());
      break;
   }

   case ORDER_STATUS_DISCONNECTED: {
      me->buffered_cancel = true;
      LOG(OrderMan, Warn,
          "Cannot cancel order {} while in the disconnected state. Buffering "
          "cancel",
          key().value());
      auto msg = "Order will be canceled when re-connected";
      result.error = ORDER_ERROR_GATEWAY_NOT_READY;
      result.error_msg = msg;
      transition(me->state.status(), msg);
      break;
   }
   case ORDER_STATUS_INACTIVE:
   case OrderStatus_INT_MIN_SENTINEL_DO_NOT_USE_:
   case OrderStatus_INT_MAX_SENTINEL_DO_NOT_USE_:
      bts_assert(false);
      break;
   }
   report_state();
   return result;
}

bool NativeOrder::is_old_exe_report(
    api::orderman::NativeOrderState &native_state,
    const NewExecutionReportEvent &er) const {
   if (native_state.last_transact_time().size() !=
       sizeof(mercury::TimeStampStr))
      return false;

   if (std::lexicographical_compare(native_state.last_transact_time().begin(),
                                    native_state.last_transact_time().end(),
                                    er.data->transact_time.value().begin(),
                                    er.data->transact_time.value().end()))
      return false;

   if (std::equal(native_state.last_transact_time().begin(),
                  native_state.last_transact_time().end(),
                  er.data->transact_time.value().begin(),
                  er.data->transact_time.value().end()))
      return native_state.last_msg_seq_num() >= er.data->msg_seq_num;

   return true;
}

void NativeOrder::on_exe_report(const NewExecutionReportEvent &er) {
   auto scoped_log =
       me->services->log_scope("NativeOrder::on_exe_report", key().value());

   auto &state = me->state;
   auto &native_state = get_native_state(state);

   // if we have a previous state we want to ensure this execution
   // report was  not already considered.
   if (is_old_exe_report(native_state, er)) {
      LOG(OrderMan, Info,
          "Ignoring already seen execution report {} in order {}",
          er.data->msg_seq_num, key().value());
      return;
   }

   // we do not care about spread leg fills for purposes of order
   // state maintenance
   if (er.data->leg_security) {
      LOG(OrderMan, Diag, "Ignoring leg execution report for order {}",
          key().value());
      return;
   }

   native_state.set_last_msg_seq_num(er.data->msg_seq_num);
   native_state.set_last_transact_time(to_str(er.data->transact_time));

   auto external_status = map_external_status(er.data->order_status);

   // filled quantity is reset back to 0 when the order is canceled
   if (external_status != ORDER_STATUS_CANCELED) {
      if (er.data->exec_type == api::orderman::EXEC_TYPE_PARTIAL_FILL ||
          er.data->exec_type == api::orderman::EXEC_TYPE_FILL) {

         bts_assert(er.data->last_shares.value());
         auto previous_avg = state.avg_price();
         auto previous_sz = state.filled_quantity();
         auto fill_price = er.data->last_price.value();
         auto fill_sz = er.data->last_shares.value();

         auto new_avg =
             ((previous_avg * previous_sz) + (fill_price * fill_sz)) /
             (previous_sz + fill_sz);

         state.set_avg_price(new_avg);
      }
      state.set_filled_quantity(er.data->filled_quantity.value());
      state.set_last_price(er.data->last_price.value());
   }

   state.set_last_msg(er.data->msg);

   native_state.set_native_status(er.data->order_status);
   native_state.set_native_type(er.data->order_type);
   bool got_first_exchange_order_id = !has_exchange_order_id();
   if (got_first_exchange_order_id) {
      me->exchange_order_id = to_trim_str(er.data->exchange_order_id);
      native_state.set_exchange_order_id(me->exchange_order_id);
      add_exchange_order_id_to_fix();
   }

   if (!terminal_state(me->state.status())) {
      state.set_price(er.data->price.value());
      state.set_quantity(er.data->quantity.value());
   }

   switch (me->state.status()) {
      // terminal states, nothing can change them
   case ORDER_STATUS_COMPLETED:
   case ORDER_STATUS_CANCELED:
      break;

      // receiving an execution report does does not mean
      // we are connected
   case ORDER_STATUS_DISCONNECTED:
      break;

   case ORDER_STATUS_PENDING_WORKING:
      if (external_status == ORDER_STATUS_WORKING &&
          !me->services->is_gateway_connected(*me->session_services))
         transition(ORDER_STATUS_DISCONNECTED);
      else {
         transition(external_status);
         if (external_status == ORDER_STATUS_WORKING) {
            if (me->buffered_cancel) {
               LOG(OrderMan, Warn,
                   "Applying buffered cancel to order {}, on first ack",
                   key().value());
               on_cancel();
            }
            else if (me->buffered_update.has_value()) {
               auto [p, q] = me->buffered_update.value();
               me->buffered_update.reset();
               LOG(OrderMan, Warn,
                   "Applying buffered update ({}@{}) to order {}, on "
                   "first ack",
                   q.value(),p.value(),
                   key().value());
               on_update(p, q);
            }
         }
      }
      break;

   case ORDER_STATUS_WORKING:
      transition(external_status);
      break;

   case ORDER_STATUS_INACTIVE:
   case ORDER_STATUS_ERROR:
   case ORDER_STATUS_REJECTED_BY_SPARK:
   case OrderStatus_INT_MIN_SENTINEL_DO_NOT_USE_:
   case OrderStatus_INT_MAX_SENTINEL_DO_NOT_USE_:
      bts_assert(false);
      break;
   }
   report_state();

   // we only need the exhange order id and the routing record
   // reported once, since they are both immutable, so we clear them
   // after their first report to make subsquent state updates faster
   if (got_first_exchange_order_id) {
      native_state.clear_exchange_order_id();
      native_state.mutable_routing_record()->Clear();
   }
}

void NativeOrder::on_cancel_reject(const orderman::CancelRejectEvent &cr) {
   auto &state = me->state;
   state.set_last_msg(cr.reject_msg);
   LOG(OrderMan, Warn, "Cancel reject received for order {}", key().value());

   switch (me->state.status()) {

   case ORDER_STATUS_COMPLETED:
   case ORDER_STATUS_CANCELED:
   case ORDER_STATUS_DISCONNECTED:
   case ORDER_STATUS_PENDING_WORKING:
   case ORDER_STATUS_WORKING:
      break;

   case ORDER_STATUS_INACTIVE:
   case ORDER_STATUS_ERROR:
   case ORDER_STATUS_REJECTED_BY_SPARK:
   case OrderStatus_INT_MIN_SENTINEL_DO_NOT_USE_:
   case OrderStatus_INT_MAX_SENTINEL_DO_NOT_USE_:
      bts_assert(false);
      break;
   }
   report_state();
}

void NativeOrder::on_gateway_status(const GatewayStatusChangeEvent &ev) {
   if (ev.is_connected) {
      switch (me->state.status()) {
      case ORDER_STATUS_COMPLETED:
      case ORDER_STATUS_CANCELED:
      case ORDER_STATUS_ERROR:
      case ORDER_STATUS_PENDING_WORKING:
      case ORDER_STATUS_WORKING:
         break;
      case ORDER_STATUS_DISCONNECTED: {
         auto const &native_state = get_native_state(me->state);
         transition(map_external_status(native_state.native_status()));

         if (me->state.status() == ORDER_STATUS_WORKING) {
            if (me->buffered_cancel) {
               LOG(OrderMan, Warn,
                   "Applying buffered cancel to order {}, on re-connection",
                   key().value());
               on_cancel();
            } else if (me->buffered_update.has_value()) {
               auto [p, q] = me->buffered_update.value();
               me->buffered_update.reset();
               LOG(OrderMan, Warn,
                   "Applying buffered update ({}@{}) to order {}, on "
                   "re-connection",
                   q.value(),p.value(),
                   key().value());
               on_update(p, q);
            }
         }
         break;
      }
      case ORDER_STATUS_INACTIVE:
      case ORDER_STATUS_REJECTED_BY_SPARK:
      case OrderStatus_INT_MIN_SENTINEL_DO_NOT_USE_:
      case OrderStatus_INT_MAX_SENTINEL_DO_NOT_USE_:
         bts_assert(false);
         break;
      }
   } else {
      switch (me->state.status()) {
      case ORDER_STATUS_COMPLETED:
      case ORDER_STATUS_CANCELED:
      case ORDER_STATUS_ERROR:
      case ORDER_STATUS_DISCONNECTED:
      case ORDER_STATUS_PENDING_WORKING:
         break;
      case ORDER_STATUS_WORKING:
         transition(ORDER_STATUS_DISCONNECTED);
         break;
      case ORDER_STATUS_INACTIVE:
      case ORDER_STATUS_REJECTED_BY_SPARK:
      case OrderStatus_INT_MIN_SENTINEL_DO_NOT_USE_:
      case OrderStatus_INT_MAX_SENTINEL_DO_NOT_USE_:
         bts_assert(false);
         break;
      }
   }
   report_state();
}

void NativeOrder::update_working_position() {
   if (me->sec) {
      auto current_position = position_contribution();
      auto previous_position = me->previous_position_contribution;
      me->previous_position_contribution = current_position;

      if (current_position.value() > previous_position.value())
         me->services->limiter().add(
             *me->sec, AccountID(me->definition.account_id()),
             me->definition.side(),
             Quantity(current_position.value() - previous_position.value()));
      else if (previous_position.value() > current_position.value())
         me->services->limiter().remove(
             *me->sec, AccountID(me->definition.account_id()),
             me->definition.side(),
             Quantity(previous_position.value() - current_position.value()));
   } else
      LOG(OrderMan, Error,
          "Order {} ignored for purposes of limit enforcing "
          "because it references unknown security {} ",
          key().value(), me->definition.security_key());
}

void NativeOrder::report_state() {
   update_working_position();
   me->services->report_state(me->state);
}

void NativeOrder::init_fix() {
   bts_assert(me->sec);

   using namespace jetfix;
   me->create_encoding = FIXFragment{};
   me->update_encoding = FIXFragment{};
   me->cancel_encoding = FIXFragment{};

   auto const &def = me->definition;
   auto const &sec = *me->sec;
   auto const &rr = get_native_state(me->state).routing_record();

   std::string clordid(clordid_sz(), ' ');
   encode_clorid(key(), SecurityKey(sec.id()), &clordid);

   // initialize create FIX
   auto &create_encoding = me->create_encoding;
   create_encoding.add<fix::SenderSubID>(rr.operator_id());
   create_encoding.add<fix::SenderLocationID>(rr.location());
   create_encoding.add<fix::Account>(rr.account());
   create_encoding.add<fix::Side>(def.side());
   create_encoding.add<fix::TimeInForce>(def.time_in_force());
   create_encoding.add<fix::ManualOrderIndicator>(def.manual_order_indicator());
   create_encoding.add<fix::CustomerOrFirm>(rr.customer_or_firm());
   create_encoding.add<fix::CtiCode>(rr.cti_code());
   create_encoding.add<fix::SecurityDesc>(sec.glbx().security_desc());
   create_encoding.add<fix::ClOrdID>(clordid);
   if (rr.smp_id().size()) {
      create_encoding.add<fix::SelfMatchPreventionID>(rr.smp_id());
      if (rr.smp_instruction() != SELF_MATCH_PREVENTION_INSTRUCTION_UNDEFINED)
         create_encoding.add<fix::SelfMatchPreventionInstruction>(
             rr.smp_instruction());
   }

   me->create_transact_time_bm = create_encoding.add<fix::TransactTimeStr>(
       datetime::FIX_timestamp(datetime::now()));
   me->create_quantity_bm =
       create_encoding.add<fix::OrderQty>(0, quantity_sz());
   // initialize update FIX
   auto &update_encoding = me->update_encoding;
   update_encoding.add<fix::SenderSubID>(rr.operator_id());
   update_encoding.add<fix::SenderLocationID>(rr.location());
   update_encoding.add<fix::Account>(rr.account());
   update_encoding.add<fix::Side>(def.side());
   update_encoding.add<fix::TimeInForce>(def.time_in_force());
   update_encoding.add<fix::ManualOrderIndicator>(def.manual_order_indicator());
   update_encoding.add<fix::CtiCode>(rr.cti_code());
   update_encoding.add<fix::CustomerOrFirm>(rr.customer_or_firm());
   update_encoding.add<fix::SecurityDesc>(sec.glbx().security_desc());
   update_encoding.add<fix::ClOrdID>(clordid);
   update_encoding.add<fix::OrigClOrdID>(clordid);
   update_encoding.add<fix::CorrelationClOrdID>(clordid);
   update_encoding.add<fix::InFlightMitigation>(true);
   // Self Match Protection
   if (rr.smp_id().size()) {
      update_encoding.add<fix::SelfMatchPreventionID>(rr.smp_id());
      if (rr.smp_instruction() != SELF_MATCH_PREVENTION_INSTRUCTION_UNDEFINED)
         update_encoding.add<fix::SelfMatchPreventionInstruction>(
             rr.smp_instruction());
   }
   me->update_transact_time_bm = update_encoding.add<fix::TransactTimeStr>(
       datetime::FIX_timestamp(datetime::now()));
   me->update_quantity_bm =
       update_encoding.add<fix::OrderQty>(0, quantity_sz());

   // initialize cancel FIX
   auto &cancel_encoding = me->cancel_encoding;
   cancel_encoding.add<fix::SenderSubID>(rr.operator_id());
   cancel_encoding.add<fix::SenderLocationID>(rr.location());
   cancel_encoding.add<fix::Account>(rr.account());
   cancel_encoding.add<fix::Side>(def.side());
   cancel_encoding.add<fix::TimeInForce>(def.time_in_force());
   cancel_encoding.add<fix::ManualOrderIndicator>(def.manual_order_indicator());
   cancel_encoding.add<fix::SecurityDesc>(sec.glbx().security_desc());
   cancel_encoding.add<fix::ClOrdID>(clordid);
   cancel_encoding.add<fix::OrigClOrdID>(clordid);
   cancel_encoding.add<fix::CorrelationClOrdID>(clordid);
   me->cancel_transact_time_bm = cancel_encoding.add<fix::TransactTimeStr>(
       datetime::FIX_timestamp(datetime::now()));

   // special tags
   switch (def.custom_def().custom_case()) {
   case CustomOrderDefinition::kMarket:
      // type
      create_encoding.add<fix::OrdType>(NATIVE_ORDER_TYPE_MARKET);
      update_encoding.add<fix::OrdType>(NATIVE_ORDER_TYPE_MARKET);
      cancel_encoding.add<fix::OrdType>(NATIVE_ORDER_TYPE_MARKET);
      break;

   case CustomOrderDefinition::kLimit:
      // type
      create_encoding.add<fix::OrdType>(NATIVE_ORDER_TYPE_LIMIT);
      update_encoding.add<fix::OrdType>(NATIVE_ORDER_TYPE_LIMIT);
      cancel_encoding.add<fix::OrdType>(NATIVE_ORDER_TYPE_LIMIT);

      // price
      me->create_price_bm = create_encoding.add<fix::Price>(
          0, me->price_encoding_precision, price_sz());
      me->update_price_bm = update_encoding.add<fix::Price>(
          0, me->price_encoding_precision, price_sz());
      break;

   case CustomOrderDefinition::kStop: {
      // type
      create_encoding.add<fix::OrdType>(NATIVE_ORDER_TYPE_STOP);
      update_encoding.add<fix::OrdType>(NATIVE_ORDER_TYPE_STOP);
      cancel_encoding.add<fix::OrdType>(NATIVE_ORDER_TYPE_STOP);

      // stop px
      auto stop_px = def.custom_def().stop().stop_price();
      auto encoded_stop_px =
          securities::encode_exchange_price(*me->sec, Price(stop_px)).value();
      create_encoding.add<fix::StopPx>(encoded_stop_px,
                                       me->price_encoding_precision, -1);
      update_encoding.add<fix::StopPx>(encoded_stop_px,
                                       me->price_encoding_precision, -1);
      break;
   }

   case CustomOrderDefinition::kStopLimit: {
      // type
      create_encoding.add<fix::OrdType>(NATIVE_ORDER_TYPE_STOP_LIMIT);
      update_encoding.add<fix::OrdType>(NATIVE_ORDER_TYPE_STOP_LIMIT);
      cancel_encoding.add<fix::OrdType>(NATIVE_ORDER_TYPE_STOP_LIMIT);

      // stop px
      auto stop_px = def.custom_def().stop_limit().stop_price();
      auto encoded_stop_px =
          securities::encode_exchange_price(*me->sec, Price(stop_px)).value();
      create_encoding.add<fix::StopPx>(encoded_stop_px,
                                       me->price_encoding_precision, -1);
      update_encoding.add<fix::StopPx>(encoded_stop_px,
                                       me->price_encoding_precision, -1);

      // price
      me->create_price_bm = create_encoding.add<fix::Price>(
          0, me->price_encoding_precision, price_sz());
      me->update_price_bm = update_encoding.add<fix::Price>(
          0, me->price_encoding_precision, price_sz());
      break;
   }
   default:
      bts_assert(false);
      break;
   }
}

void NativeOrder::add_exchange_order_id_to_fix() {
   bts_assert(has_exchange_order_id());
   me->update_encoding.add<jetfix::fix::OrderID>(me->exchange_order_id);
   me->cancel_encoding.add<jetfix::fix::OrderID>(me->exchange_order_id);
}

OrderStatus map_external_status(NativeOrderStatus status) {
   switch (status) {
   case NATIVE_ORDER_STATUS_NEW:
   case NATIVE_ORDER_STATUS_PENDING_NEW:
   case NATIVE_ORDER_STATUS_PARTIALLY_FILLED:
   case NATIVE_ORDER_STATUS_REPLACED:
   case NATIVE_ORDER_STATUS_PENDING_CANCEL:
   case NATIVE_ORDER_STATUS_STOPPED:
   case NATIVE_ORDER_STATUS_SUSPENDED:
   case NATIVE_ORDER_STATUS_CANCEL_ACK:
   case NATIVE_ORDER_STATUS_ACCEPTED_FOR_BIDDING:
   case NATIVE_ORDER_STATUS_PENDING_REPLACE:
   case NATIVE_ORDER_STATUS_CALCULATED:
      return ORDER_STATUS_WORKING;

   case NATIVE_ORDER_STATUS_FILLED:
      return ORDER_STATUS_COMPLETED;

   case NATIVE_ORDER_STATUS_EXPIRED:
   case NATIVE_ORDER_STATUS_CANCELED:
   case NATIVE_ORDER_STATUS_REJECTED:
   case NATIVE_ORDER_STATUS_DONE:
   case NATIVE_ORDER_STATUS_ORDER_NOT_FOUND:
      return ORDER_STATUS_CANCELED;

   case NATIVE_ORDER_STATUS_UNDEFINED:
      return ORDER_STATUS_ERROR;

   case NATIVE_ORDER_STATUS_UNACKNOWLEDGED:
   case NativeOrderStatus_INT_MIN_SENTINEL_DO_NOT_USE_:
   case NativeOrderStatus_INT_MAX_SENTINEL_DO_NOT_USE_:
      bts_assert(false);
      return ORDER_STATUS_ERROR;
   }
   return ORDER_STATUS_ERROR;
}

bool NativeOrder::has_exchange_order_id() {
   return me->exchange_order_id.size();
}

void NativeOrder::transition(api::orderman::OrderStatus new_status,
                             std::string const &msg) {
   transition(Info, new_status, msg);
}

void NativeOrder::transition(log::Level level,
                             api::orderman::OrderStatus new_status,
                             std::string const &msg) {
   auto const &from = OrderStatus_Name(me->state.status());
   auto const &to = OrderStatus_Name(new_status);
   if (from != to)
      LOG(OrderMan, level, "Transitioning order {} from {} to {}. Msg: {}",
          key().value(), from, to, msg);
   me->state.set_status(new_status);
   me->state.set_last_msg(msg);
}

void NativeOrder::transition(api::orderman::OrderStatus new_status) {
   auto const &from = OrderStatus_Name(me->state.status());
   auto const &to = OrderStatus_Name(new_status);
   if (from != to)
      LOG(OrderMan, Info, "Transitioning order {} from {} to {}", key().value(),
          from, to);
   else
      LOG(OrderMan, Diag, "Transitioning order {} from {} to {}", key().value(),
          from, to);

   me->state.set_status(new_status);
}

void NativeOrder::transition_to_error(
    api::orderman::NativeOrderState::SubError sub_error,
    std::string const &msg) {
   me->state.mutable_limit()->mutable_native_state()->set_sub_error_code(
       sub_error);
   transition(ORDER_STATUS_ERROR, msg);
}

Quantity NativeOrder::position_contribution() {
   switch (me->state.status()) {

   case ORDER_STATUS_WORKING:
   case ORDER_STATUS_PENDING_WORKING:
   case ORDER_STATUS_DISCONNECTED:
      return Quantity(
          std::max(me->state.quantity(), me->state.filled_quantity()));

   case ORDER_STATUS_REJECTED_BY_SPARK:
   case ORDER_STATUS_ERROR:
   case ORDER_STATUS_COMPLETED:
   case ORDER_STATUS_CANCELED:
      return Quantity(me->state.filled_quantity());

   case ORDER_STATUS_INACTIVE:
   case OrderStatus_INT_MIN_SENTINEL_DO_NOT_USE_:
   case OrderStatus_INT_MAX_SENTINEL_DO_NOT_USE_:
      bts_assert(false);
      return Quantity(0);
   }
   bts_assert(false);
   return Quantity(0);
}

bool NativeOrder::terminal_state(api::orderman::OrderStatus status) const {
   switch (status) {

   case ORDER_STATUS_COMPLETED:
   case ORDER_STATUS_CANCELED:
   case ORDER_STATUS_ERROR:
   case ORDER_STATUS_REJECTED_BY_SPARK:
      return true;

   case ORDER_STATUS_DISCONNECTED:
   case ORDER_STATUS_PENDING_WORKING:
   case ORDER_STATUS_WORKING:
      return false;

   case OrderStatus_INT_MIN_SENTINEL_DO_NOT_USE_:
   case OrderStatus_INT_MAX_SENTINEL_DO_NOT_USE_:
   case ORDER_STATUS_INACTIVE:
      bts_assert(false);
      return false;
      break;
   }
   bts_assert(false);
   return false;
}

api::orderman::NativeOrderState &
NativeOrder::get_native_state(api::orderman::OrderState &st) {
   switch (st.custom_case()) {
   case OrderState::kMarket:
      return *st.mutable_market()->mutable_native_state();
   case OrderState::kLimit:
      return *st.mutable_limit()->mutable_native_state();
   case OrderState::kStop:
      return *st.mutable_stop()->mutable_native_state();
   case OrderState::kStopLimit:
      return *st.mutable_stop_limit()->mutable_native_state();
   default:
      break;
   }

   // just to appease compiler
   bts_assert(false);
   return *st.mutable_stop_limit()->mutable_native_state();
}
} // namespace orderman
} // namespace bts