订单管理系统¶
1.1 订单管理有限状态机
Mealy 状态机和 Moore 状态机是两种经典的有限状态机（FSM）模型。交易系统通常是Mealy状态机。相比Moore，Mealy将输入也包括其中。
有限状态机的核心要素：
- 状态（State）表示对象在某一时刻的具体状态，是 FSM 的基本单元。例如订单的“新订单”“处理中”“已取消”等状态。状态具有排他性：同一时刻，对象只能处于一个状态。
- 事件（Event）触发状态转换的外部或内部信号。例如 “订单成功”“撤单指令” 等事件会推动状态机从当前状态切换到新状态。
- 转换（Transition）定义 “当前状态 + 事件” 如何映射到 “下一状态” 的规则。例如：“处理中” 状态收到 “订单成交” 事件后，转换到 “成功” 状态。
- 动作（Action）在状态转换过程中执行的具体操作（可选）。例如撤单时记录日志、更新订单状态等。
- 守卫（Guard）状态转换的条件（可选）。只有当条件满足时，事件才会触发状态转换。例如 “撤单间隔超过 5 秒” 才允许执行撤单。
1.2 无分支状态表 无分支状态表是一种状态机设计模式，它使用表驱动的方式替代传统的条件分支语句(if-else/switch-case)来实现状态转换。 在Boost MSM (Meta State Machine)框架中，无分支状态表通过编译期元编程实现，转换逻辑在编译时确定，运行时无需条件判断。
以下代码实现了TWAP算法交易的订单状态机，使用Boost MSM库。核心类是OrderStateMachineImpl，它定义了所有状态、事件、转换规则和业务逻辑，来源：https://github.com/byrnexu/betterquant2
/*!
* \file AlgoTWAPFSM.hpp
* \project BetterQuant
*
* \author byrnexu
* \date 2023/05/17
*
* \brief
*/
#pragma once
#include "AlgoConst.hpp"
#include "AlgoDef.hpp"
#include "AlgoTWAP.hpp"
#include "AlgoTWAPDef.hpp"
#include "AlgoUtil.hpp"
#include "TrdSymParams.hpp"
#include "def/BQConst.hpp"
#include "def/BQDef.hpp"
#include "def/Const.hpp"
#include "def/DataStruOfMD.hpp"
#include "def/DataStruOfTD.hpp"
#include "def/Def.hpp"
#include "def/StatusCode.hpp"
#include "util/Datetime.hpp"
#include "util/Decimal.hpp"
#include "util/Logger.hpp"
#include "util/Pch.hpp"
namespace msm = boost::msm;
namespace msmf = boost::msm::front;
namespace mpl = boost::mpl;
namespace bq::algo {
enum class NextActionOfAlgo { HandleCurOrder, HandleNextOrder, TerminateAlgo };
struct EventOfOrderSuccess {};
struct EventOfOrderFailed {};
struct EventOfCheckIfCancelOrder {};
struct EventOfCancelOrderFailed {};
struct EventOfCancelOrderRetFailed {};
struct EventOfOrderRetFilled {};
struct EventOfOrderRetFailed {};
struct EventOfOrderRetCanceled {};
struct EventOfOrderSizeLTTickerSize {};
struct EventOfCheckIfRestoreOrder {};
/*
* 触发顺序
*
* on_entry fsm
* on_entry StateOfNewOrder
* Enter GuardOfNewOrder
* on_exit StateOfNewOrder
* Enter ActionOfNewOrder
* on_entry StateOfOrderInProc
*
*/
class OrderStateMachineImpl
: public msm::front::state_machine_def<OrderStateMachineImpl> {
private:
//! 新订单
struct StateOfNewOrder : public msm::front::state<> {
template <class Event, class Fsm>
void on_entry(const Event& evt, Fsm& fsm) {
fsm.twap_->getAlgoMgr()->getStgEng()->logInfo(
"[ALGO] Entering StateOfNewOrder. [{}] {}",
{fsm.twap_->toStr(), fsm.toStr()}, fsm.twap_->getStgInstInfo());
fsm.startTime_ = GetTotalMSSince1970();
fsm.curOrderSize_ = fsm.orderSize_;
fsm.curUrgency_ = fsm.twap_->getTWAPParams()->initialUrgency_;
fsm.curTickerOffset_ = fsm.twap_->getTWAPParams()->tickerOffset_;
fsm.order(fsm);
}
template <class Event, class Fsm>
void on_exit(const Event&, Fsm& fsm) {
fsm.twap_->getAlgoMgr()->getStgEng()->logInfo(
"[ALGO] Exiting StateOfNewOrder. [{}] {}",
{fsm.twap_->toStr(), fsm.toStr()}, fsm.twap_->getStgInstInfo());
}
};
//! 正在下单
struct StateOfOrderInProc : public msm::front::state<> {
template <class Event, class Fsm>
void on_entry(const Event&, Fsm& fsm) {
fsm.twap_->getAlgoMgr()->getStgEng()->logInfo(
"[ALGO] Entering StateOfOrderInProc. [{}] {}",
{fsm.twap_->toStr(), fsm.toStr()}, fsm.twap_->getStgInstInfo());
}
template <class Event, class Fsm>
void on_exit(const Event&, Fsm& fsm) {
fsm.twap_->getAlgoMgr()->getStgEng()->logInfo(
"[ALGO] Exiting StateOfOrderInProc. [{}] {}",
{fsm.twap_->toStr(), fsm.toStr()}, fsm.twap_->getStgInstInfo());
}
};
struct GuardOfCancelOrder {
template <class Event, class Fsm, class SourceState, class TargetState>
bool operator()(const Event&, Fsm& fsm, SourceState&, TargetState&) const {
std::uint64_t checkTimepoint = 0;
if (fsm.lastCancelOrderTime_ == 0) {
//! 说明是首次撤单，下次撤单间隔从fsm.lastOrderTime_开始计算
checkTimepoint = fsm.lastOrderTime_;
} else {
//! 说明是重复撤单(撤单出错会有这个情况)，下次撤单间隔从fsm.lastCancelOrderTime_开始计算
checkTimepoint = fsm.lastCancelOrderTime_;
}
const auto now = GetTotalMSSince1970();
const auto timeDur = now - checkTimepoint;
if (timeDur <=
fsm.twap_->getTWAPParams()->minMSIntervalOfOrderAndCancelOrder_) {
//! 如果间隔小于minMSIntervalOfOrderAndCancelOrder_那么不撤单
return false;
}
if (!fsm.twap_->getBid1Ask1()) {
//! 这通常是由于lastPrice超出报单价格区间引起的，那么就在这里一直等待bid1ask1
const auto timeDurOfSec = timeDur / 1000;
static std::decay_t<decltype(timeDurOfSec)> timeDurOfSecOfPrint = 0;
if (timeDurOfSec % 5 == 0) {
if (timeDurOfSec != timeDurOfSecOfPrint) {
//! timeDurOfSecOfPrint确保每5秒只打印一次
fsm.twap_->getAlgoMgr()->getStgEng()->logWarn(
"[ALGO] Bid1Ask1 is null when try to order. [{}] {}",
{fsm.twap_->toStr(), fsm.toStr()}, fsm.twap_->getStgInstInfo());
} else {
timeDurOfSecOfPrint = timeDurOfSec;
}
}
return false;
}
const auto priceOffset =
fsm.curUrgency_ == Urgency::PriceOfMaker
? fsm.twap_->getTrdSymParams()->precOfOrderPrice_ *
fsm.curTickerOffset_
: 0;
const auto priceForComp =
fsm.twap_->getTWAPParams()->side_ == Side::Bid
? fsm.twap_->getBid1Ask1()->bidPrice_ - priceOffset
: fsm.twap_->getBid1Ask1()->askPrice_ + priceOffset;
if (fsm.twap_->getTWAPParams()->side_ == Side::Bid) {
//! 如果是买入且下单价格已经小于买一减去偏移
if (DEC::LT(fsm.orderPrice_, priceForComp)) {
fsm.twap_->getAlgoMgr()->getStgEng()->logInfo(
"[ALGO] Past the cancel order guard because of orderPice {} less "
"than bidPrice {} - priceOfOffset {} . [{}] {}",
{std::to_string(fsm.orderPrice_),
std::to_string(fsm.twap_->getBid1Ask1()->bidPrice_),
std::to_string(priceOffset), fsm.twap_->toStr(), fsm.toStr()},
fsm.twap_->getStgInstInfo());
return true;
}
} else {
//! 如果是卖出且下单价格已经大于卖一加上偏移
if (DEC::GT(fsm.orderPrice_, priceForComp)) {
fsm.twap_->getAlgoMgr()->getStgEng()->logInfo(
"[ALGO] Past the cancel order guard because of orderPice {} "
"greater than askPrice {} + priceOfOffset {}. [{}] {}",
{std::to_string(fsm.orderPrice_),
std::to_string(fsm.twap_->getBid1Ask1()->askPrice_),
std::to_string(priceOffset), fsm.twap_->toStr(), fsm.toStr()},
fsm.twap_->getStgInstInfo());
return true;
}
}
return false;
}
};
struct ActionOfCancelOrder {
template <class Event, class Fsm, class SourceState, class TargetState>
void operator()(const Event&, Fsm& fsm, SourceState&, TargetState&) {
fsm.twap_->getAlgoMgr()->getStgEng()->logInfo(
"[ALGO] Entering ActionOfCancelOrder. [{}] {}",
{fsm.twap_->toStr(), fsm.toStr()}, fsm.twap_->getStgInstInfo());
//! 重置最后一次撤单时间
fsm.lastCancelOrderTime_ = GetTotalMSSince1970();
const auto statusCode =
fsm.twap_->getAlgoMgr()->getStgEng()->cancelOrder(fsm.orderId_);
if (statusCode != 0) {
fsm.twap_->getAlgoMgr()->getStgEng()->logWarn(
"[ALGO] Send cancel order failed. [{}] {} [{} - {}]",
{fsm.twap_->toStr(), fsm.toStr(), std::to_string(statusCode),
GetStatusMsg(statusCode)},
fsm.twap_->getStgInstInfo());
//! 撤单失败，回到StateOfOrderInProc
fsm.process_event(EventOfCancelOrderFailed{});
} else {
fsm.twap_->getAlgoMgr()->getStgEng()->logInfo(
"[ALGO] Send cancel order {} success. [{}] {}",
{std::to_string(fsm.orderId_), fsm.twap_->toStr(), fsm.toStr()},
fsm.twap_->getStgInstInfo());
}
}
};
//! 正在撤单
struct StateOfCancelOrderInProc : public msm::front::state<> {
template <class Event, class Fsm>
void on_entry(const Event&, Fsm& fsm) {
fsm.twap_->getAlgoMgr()->getStgEng()->logInfo(
"[ALGO] Entering StateOfCancelOrderInProc. [{}] {}",
{fsm.twap_->toStr(), fsm.toStr()}, fsm.twap_->getStgInstInfo());
}
template <class Event, class Fsm>
void on_exit(const Event&, Fsm& fsm) {
fsm.twap_->getAlgoMgr()->getStgEng()->logInfo(
"[ALGO] Exiting StateOfCancelOrderInProc. [{}] {}",
{fsm.twap_->toStr(), fsm.toStr()}, fsm.twap_->getStgInstInfo());
}
};
//! 已经撤单
struct StateOfOrderCanceled : public msm::front::state<> {
template <class Event, class Fsm>
void on_entry(const Event&, Fsm& fsm) {
++fsm.canceledTimes_;
fsm.twap_->getAlgoMgr()->getStgEng()->logInfo(
"[ALGO] Entering StateOfOrderCanceled, "
"the order has been canceled {} times. [{}] {}",
{std::to_string(fsm.canceledTimes_), fsm.twap_->toStr(), fsm.toStr()},
fsm.twap_->getStgInstInfo());
}
template <class Event, class Fsm>
void on_exit(const Event&, Fsm& fsm) {
fsm.twap_->getAlgoMgr()->getStgEng()->logInfo(
"[ALGO] Exiting StateOfOrderCanceled. [{}] {}",
{fsm.twap_->toStr(), fsm.toStr()}, fsm.twap_->getStgInstInfo());
}
};
//! 尝试重新报单
struct StateOfRetryOrder : public msm::front::state<> {
template <class Event, class Fsm>
void on_entry(const Event&, Fsm& fsm) {
fsm.twap_->getAlgoMgr()->getStgEng()->logInfo(
"[ALGO] Entering StateOfRetryOrder . [{}] {}",
{fsm.twap_->toStr(), fsm.toStr()}, fsm.twap_->getStgInstInfo());
const auto threshold =
fsm.twap_->getTWAPParams()->cancelTimesOfUpgradeUrgency_;
//! 如果撤单次数达到cancelTimesOfUpgradeUrgency的整数倍，升级urgency
if (fsm.canceledTimes_ % threshold == 0) {
std::tie(fsm.curUrgency_, fsm.curTickerOffset_) =
UpgradeUrgency(fsm.curUrgency_, fsm.curTickerOffset_);
fsm.twap_->getAlgoMgr()->getStgEng()->logInfo(
"[ALGO] Upgrade urgency to {}:{}. [{}] {}",
{ENUM_TO_STR(fsm.curUrgency_), std::to_string(fsm.curTickerOffset_),
fsm.twap_->toStr(), fsm.toStr()},
fsm.twap_->getStgInstInfo());
}
//! 如果在ActionOfRetryOrder里写下单代码的话，transition_table不直观
fsm.order(fsm);
}
template <class Event, class Fsm>
void on_exit(const Event&, Fsm& fsm) {
fsm.twap_->getAlgoMgr()->getStgEng()->logInfo(
"[ALGO] Exiting StateOfRetryOrder . [{}] {}",
{fsm.twap_->toStr(), fsm.toStr()}, fsm.twap_->getStgInstInfo());
}
};
//! 报单出错，重新报单
struct StateOfRestoreOrder : public msm::front::state<> {
template <class Event, class Fsm>
void on_entry(const Event&, Fsm& fsm) {
fsm.twap_->getAlgoMgr()->getStgEng()->logInfo(
"[ALGO] Entering StateOfRestoreOrder. [{}] {}",
{fsm.twap_->toStr(), fsm.toStr()}, fsm.twap_->getStgInstInfo());
}
template <class Event, class Fsm>
void on_exit(const Event&, Fsm& fsm) {
fsm.twap_->getAlgoMgr()->getStgEng()->logInfo(
"[ALGO] Exiting StateOfRestoreOrder. [{}] {}",
{fsm.twap_->toStr(), fsm.toStr()}, fsm.twap_->getStgInstInfo());
}
};
//! 订单成功完结
struct StateOfSuccess : public msm::front::state<> {
template <class Event, class Fsm>
void on_entry(const Event&, Fsm& fsm) {
fsm.twap_->getAlgoMgr()->getStgEng()->logInfo(
"[ALGO] Entering StateOfSuccess. [{}] {}",
{fsm.twap_->toStr(), fsm.toStr()}, fsm.twap_->getStgInstInfo());
fsm.nextActionOfAlgo_ = NextActionOfAlgo::HandleNextOrder;
}
template <class Event, class Fsm>
void on_exit(const Event&, Fsm& fsm) {
fsm.twap_->getAlgoMgr()->getStgEng()->logInfo(
"[ALGO] Exiting StateOfSuccess. [{}] {}",
{fsm.twap_->toStr(), fsm.toStr()}, fsm.twap_->getStgInstInfo());
}
};
//! 订单出错
struct StateOfError : public msm::front::state<> {
template <class Event, class Fsm>
void on_entry(const Event&, Fsm& fsm) {
fsm.twap_->getAlgoMgr()->getStgEng()->logInfo(
"[ALGO] Entering StateOfError. [{}] {}",
{fsm.twap_->toStr(), fsm.toStr()}, fsm.twap_->getStgInstInfo());
}
template <class Event, class Fsm>
void on_exit(const Event&, Fsm& fsm) {
fsm.twap_->getAlgoMgr()->getStgEng()->logInfo(
"[ALGO] Exiting StateOfError. [{}] {}",
{fsm.twap_->toStr(), fsm.toStr()}, fsm.twap_->getStgInstInfo());
}
};
struct GuardOfRestoreOrder {
template <class Event, class Fsm, class SourceState, class TargetState>
bool operator()(const Event&, Fsm& fsm, SourceState&, TargetState&) const {
if (fsm.twap_->needRetoreOrder(fsm.statusCode_)) {
const auto now = GetTotalMSSince1970();
if (now - fsm.lastOrderTime_ > fsm.twap_->getMSIntervalOfRetryOrder()) {
fsm.twap_->getAlgoMgr()->getStgEng()->logInfo(
"[ALGO] Retry order after order failed "
"because statusCode {} in white list. [{}] {}",
{std::to_string(fsm.statusCode_), fsm.twap_->toStr(),
fsm.toStr()},
fsm.twap_->getStgInstInfo());
return true;
} else {
return false;
}
} else {
fsm.nextActionOfAlgo_ = NextActionOfAlgo::TerminateAlgo;
return false;
}
return true;
}
};
struct ActionOfRestoreOrder {
template <class Event, class Fsm, class SourceState, class TargetState>
void operator()(const Event&, Fsm& fsm, SourceState&, TargetState&) {
fsm.twap_->getAlgoMgr()->getStgEng()->logInfo(
"[ALGO] Entering ActionOfRetoreOrder. [{}] {}",
{fsm.twap_->toStr(), fsm.toStr()}, fsm.twap_->getStgInstInfo());
// 这里已经是StateOfRestore状态了，所以 StateOfRestoreOrder =>
// StateOfOrderInProc 起作用
fsm.order(fsm);
}
};
public:
OrderStateMachineImpl(TWAP* twap, std::size_t no, Decimal orderSize)
: twap_(twap), no_(no), orderSize_(orderSize) {}
using initial_state = StateOfNewOrder;
// clang-format off
struct transition_table
: public boost::mpl::vector<
// ------+------------------------+-----------------------------+--------------------------+--------------------+-------------------+
// + Start + Event + Target + Action + Guard +
// ------+------------------------+-----------------------------+--------------------------+--------------------+-------------------+
msmf::Row<StateOfNewOrder , EventOfOrderSuccess , StateOfOrderInProc >,
msmf::Row<StateOfNewOrder , EventOfOrderFailed , StateOfError >,
// ------+------------------------+-----------------------------+--------------------------+--------------------+-------------------+
msmf::Row<StateOfOrderInProc , EventOfCheckIfCancelOrder , StateOfCancelOrderInProc , ActionOfCancelOrder, GuardOfCancelOrder>,
msmf::Row<StateOfOrderInProc , EventOfCancelOrderFailed , StateOfOrderInProc >,
msmf::Row<StateOfOrderInProc , EventOfOrderRetFilled , StateOfSuccess >,
msmf::Row<StateOfOrderInProc , EventOfOrderRetFailed , StateOfError >,
// ------+------------------------+-----------------------------+--------------------------+--------------------+-------------------+
msmf::Row<StateOfCancelOrderInProc, EventOfCancelOrderRetFailed , StateOfOrderInProc >,
msmf::Row<StateOfCancelOrderInProc, EventOfOrderRetFilled , StateOfSuccess >,
msmf::Row<StateOfCancelOrderInProc, EventOfOrderRetFailed , StateOfError >,
msmf::Row<StateOfCancelOrderInProc, EventOfOrderRetCanceled , StateOfRetryOrder >, // 人工撤单会触发此事件从而导致 StateOfOrderInProc => StateOfRetryOrder
msmf::Row<StateOfCancelOrderInProc, EventOfOrderSizeLTTickerSize, StateOfSuccess >, // 人工撤单会触发此事件从而导致 StateOfOrderInProc => StateOfSuccess
// ------+------------------------+-----------------------------+--------------------------+--------------------+-------------------+
msmf::Row<StateOfRetryOrder , EventOfOrderSuccess , StateOfOrderInProc >,
msmf::Row<StateOfRetryOrder , EventOfOrderFailed , StateOfError >,
// ------+------------------------+-----------------------------+--------------------------+--------------------+-------------------+
msmf::Row<StateOfError , EventOfCheckIfRestoreOrder , StateOfRestoreOrder , ActionOfRestoreOrder, GuardOfRestoreOrder>,
// ------+------------------------+-----------------------------+--------------------------+--------------------+-------------------+
msmf::Row<StateOfRestoreOrder , EventOfOrderSuccess , StateOfOrderInProc >,
msmf::Row<StateOfRestoreOrder , EventOfOrderFailed , StateOfError >
// ------+------------------------+-----------------------------+--------------------------+--------------------+-------------------+
> {};
// clang-format on
//! 定义订单状态机的初始行为
template <class Event, class FSM>
void on_entry(Event const&, FSM& fsm) {
fsm.twap_->getAlgoMgr()->getStgEng()->logInfo(
"[ALGO] Entering FSM. [{}] {}", {fsm.twap_->toStr(), fsm.toStr()},
fsm.twap_->getStgInstInfo());
}
//! 定义订单状态机的退出行为
template <class Event, class FSM>
void on_exit(Event const&, FSM& fsm) {
fsm.twap_->getAlgoMgr()->getStgEng()->logInfo(
"[ALGO] Exiting FSM. [{}] {}", {fsm.twap_->toStr(), fsm.toStr()},
fsm.twap_->getStgInstInfo());
}
template <class FSM, class Event>
void no_transition(Event const& e, FSM& fsm, int state) {}
//! 状态机业务代码
public:
template <class Fsm>
void order(Fsm& fsm) {
//! 重置最后一次下单时间
fsm.lastOrderTime_ = GetTotalMSSince1970();
if (!fsm.twap_->getBid1Ask1()) {
fsm.twap_->getAlgoMgr()->getStgEng()->logWarn(
"[ALGO] Bid1Ask1 is null when try to order. [{}] {}",
{fsm.twap_->toStr(), fsm.toStr()}, fsm.twap_->getStgInstInfo());
//! 如果发现价格为空(由于超出报单价格区间引起)，那么通过EventOfOrderSuccess切换到下一订单
fsm.process_event(EventOfOrderSuccess{});
return;
}
//! 根据Urgency计算下单价格
const auto orderPrice = GetPriceOfUrgency(
fsm.twap_->getBid1Ask1(), fsm.twap_->getTWAPParams()->side_,
fsm.curUrgency_, fsm.curTickerOffset_,
fsm.twap_->getTrdSymParams()->precOfOrderPrice_);
fsm.twap_->getAlgoMgr()->getStgEng()->logInfo(
"[ALGO] Calc order price {} by cur urgency. [{}] {}",
{std::to_string(orderPrice), fsm.twap_->toStr(), fsm.toStr()},
fsm.twap_->getStgInstInfo());
//! 开始报单
OrderId orderId;
std::tie(fsm.statusCode_, orderId) =
fsm.twap_->getAlgoMgr()->getStgEng()->order(
fsm.twap_->getStgInstInfo(), //
fsm.twap_->getTWAPParams()->marketCode_, //
fsm.twap_->getTWAPParams()->symbolCode_, //
fsm.twap_->getTWAPParams()->side_, //
fsm.twap_->getTWAPParams()->posDirection_, //
orderPrice, //
fsm.curOrderSize_, //
fsm.twap_->getTWAPParams()->trdAcctId_, //
fsm.twap_->getTWAPParams()->closeTDayStg_, //
fsm.twap_->getAlgoId()); //
//! 下单成功保存下单价格和订单号
if (fsm.statusCode_ == 0) {
fsm.orderId_ = orderId;
fsm.orderPrice_ = orderPrice;
fsm.process_event(EventOfOrderSuccess{});
fsm.twap_->getAlgoMgr()->getStgEng()->logInfo(
"[ALGO] Send order {} success. [{}] {}",
{std::to_string(orderId), std::to_string(orderPrice),
fsm.twap_->toStr(), fsm.toStr()},
fsm.twap_->getStgInstInfo());
} else {
fsm.twap_->getAlgoMgr()->getStgEng()->logWarn(
"[ALGO] Send order failed. [{}] {} [{} - {}]",
{std::to_string(orderPrice), fsm.twap_->toStr(), fsm.toStr(),
std::to_string(fsm.statusCode_), GetStatusMsg(fsm.statusCode_)},
fsm.twap_->getStgInstInfo());
fsm.process_event(EventOfOrderFailed{});
}
return;
}
template <class FSM>
void onTimer(FSM& fsm) {
const auto stateOfNewOrder = fsm.template get_state<StateOfNewOrder&>();
if (fsm.startTime_ == 0) {
//! 说明订单尚未开始
return;
}
// if state = StateOfError => StateOfRestoreOrder
fsm.process_event(EventOfCheckIfRestoreOrder{});
// if state = StateOfOrderInProc => StateOfCancelOrderInProc
fsm.process_event(EventOfCheckIfCancelOrder{});
}
template <class FSM>
void onOrderRet(FSM& fsm, const OrderInfo* orderInfo) {
//! 防护代码，确保是当前订单的下单回报
if (orderInfo->orderId_ != fsm.orderId_) {
fsm.twap_->getAlgoMgr()->getStgEng()->logInfo(
"[ALGO] Recv order ret of another order {}. [{}] {}",
{std::to_string(orderInfo->orderId_), fsm.twap_->toStr(),
fsm.toStr()},
fsm.twap_->getStgInstInfo());
return;
}
fsm.twap_->getAlgoMgr()->getStgEng()->logInfo(
"[ALGO] Order status of order {} is {}. [progress: {}/{}] [{}] {} {}",
{std::to_string(orderInfo->orderId_),
ENUM_TO_STR(orderInfo->orderStatus_),
std::to_string(orderInfo->dealSize_),
std::to_string(orderInfo->orderSize_), fsm.twap_->toStr(), fsm.toStr(),
orderInfo->toShortStr()},
fsm.twap_->getStgInstInfo());
switch (orderInfo->orderStatus_) {
case OrderStatus::Filled:
handleOrderFilled(fsm, orderInfo);
break;
case OrderStatus::Canceled:
case OrderStatus::PartialFilledCanceled:
handleOrderCanceled(fsm, orderInfo);
break;
case OrderStatus::Failed:
handleOrderFailed(fsm, orderInfo);
break;
default:
break;
}
}
template <class FSM>
void handleOrderFilled(FSM& fsm, const OrderInfo* orderInfo) {
recalcDealInfo(fsm, orderInfo);
fsm.twap_->getAlgoMgr()->getStgEng()->logInfo(
"[ALGO] Update deal size of order {} to {}. [{}] {} {}",
{std::to_string(orderInfo->orderId_), std::to_string(fsm.dealSize_),
fsm.twap_->toStr(), fsm.toStr(), orderInfo->toShortStr()},
fsm.twap_->getStgInstInfo());
fsm.process_event(EventOfOrderRetFilled{});
}
template <class FSM>
void handleOrderCanceled(FSM& fsm, const OrderInfo* orderInfo) {
recalcDealInfo(fsm, orderInfo);
fsm.twap_->getAlgoMgr()->getStgEng()->logInfo(
"[ALGO] Update deal size of order {} to {}. [{}] {} {}",
{std::to_string(orderInfo->orderId_), std::to_string(fsm.dealSize_),
fsm.twap_->toStr(), fsm.toStr(), orderInfo->toShortStr()},
fsm.twap_->getStgInstInfo());
//! 开始计算curOrderSize_，也就是下次报单的数量
const auto origOrdereSize = fsm.curOrderSize_;
//! 这里必须减去orderInfo->dealSize_而不是ftm.dealSize_，因为假如当前状态机经
//! 过反复报撤之后已经成交了很多导致ftm.dealSize_已经是一个很大的数字了，但是
//! curOrderSize_已经几乎没有了，也就是说是一个很小的数字，这样减出来的结果明
//! 显是不对的，所以未成交的数量应该是当前订单数减去当前订单成交数
fsm.curOrderSize_ -= orderInfo->dealSize_;
fsm.twap_->getAlgoMgr()->getStgEng()->logInfo(
"[ALGO] Adjust order size "
"from {} to {} after minus deal size {}. [{}] {}",
{std::to_string(origOrdereSize), std::to_string(fsm.curOrderSize_),
std::to_string(fsm.dealSize_), fsm.twap_->toStr(), fsm.toStr()},
fsm.twap_->getStgInstInfo());
//! 修正报单数量为precOfOrderVol的整数倍
auto precOfOrderVol = fsm.twap_->getTrdSymParams()->precOfOrderVol_;
if (IsCNMarketOfSpots(fsm.twap_->getTWAPParams()->marketCode_)) {
precOfOrderVol = 100;
}
fsm.curOrderSize_ =
std::floor(fsm.curOrderSize_ / precOfOrderVol) * precOfOrderVol;
fsm.twap_->getAlgoMgr()->getStgEng()->logInfo(
"[ALGO] Adjust order size to {} after "
"set to an integer multiple of precOfOrderVol {}. [{}] {}",
{std::to_string(fsm.curOrderSize_), std::to_string(precOfOrderVol),
fsm.twap_->toStr(), fsm.toStr()},
fsm.twap_->getStgInstInfo());
if (DEC::ZERO(fsm.curOrderSize_)) {
fsm.twap_->getAlgoMgr()->getStgEng()->logInfo(
"[ALGO] The remaining order size {} "
"less than 1 ticker size {}, switch to next order. [{}] {}",
{std::to_string(fsm.curOrderSize_), std::to_string(precOfOrderVol),
fsm.twap_->toStr(), fsm.toStr()},
fsm.twap_->getStgInstInfo());
//! 如果报单数量小于一个ticker那么结束当前订单，零股在TWAP里重新分配
fsm.process_event(EventOfOrderSizeLTTickerSize{});
} else {
//! 否则进入StateOfOrderCanceled，尝试重新报单
fsm.process_event(EventOfOrderRetCanceled{});
}
}
template <class FSM>
void recalcDealInfo(FSM& fsm, const OrderInfo* orderInfo) {
//! 更新成交数量，在完结态累加确保只加一次，同时确保反复撤报也能正确计算dealSize
const auto newDealSize = fsm.dealSize_ + std::fabs(orderInfo->dealSize_);
if (!DEC::ZERO(newDealSize)) {
fsm.avgDealPrice_ =
(fsm.avgDealPrice_ * fsm.dealSize_ +
orderInfo->avgDealPrice_ * std::fabs(orderInfo->dealSize_)) /
newDealSize;
fsm.dealSize_ = newDealSize;
}
}
template <class FSM>
void handleOrderFailed(FSM& fsm, const OrderInfo* orderInfo) {
if (orderInfo->statusCode_ != 0) {
fsm.statusCode_ = orderInfo->statusCode_;
}
fsm.process_event(EventOfOrderRetFailed{});
}
template <class FSM>
void onCancelOrderRet(FSM& fsm, const OrderInfo* orderInfo) {
//! 防护代码，确保是当前订单的撤单应答
if (orderInfo->orderId_ != fsm.orderId_) {
fsm.twap_->getAlgoMgr()->getStgEng()->logInfo(
"[ALGO] Recv cancel order ret of another order {}. [{}] {}",
{std::to_string(orderInfo->orderId_), fsm.twap_->toStr(),
fsm.toStr()},
fsm.twap_->getStgInstInfo());
return;
}
//! 撤单失败，继续StateOrderInProc状态
if (orderInfo->statusCode_ != 0) {
fsm.twap_->getAlgoMgr()->getStgEng()->logWarn(
"[ALGO] Cancel order failed. [{}] {} [{} - {}]",
{fsm.twap_->toStr(), fsm.toStr(),
std::to_string(orderInfo->statusCode_),
GetStatusMsg(orderInfo->statusCode_)},
fsm.twap_->getStgInstInfo());
//! 回到StateOrderInProc状态
fsm.process_event(EventOfCancelOrderRetFailed{});
}
}
public:
TWAP* twap_{nullptr};
const std::size_t no_{0};
const Decimal orderSize_{0};
std::uint64_t startTime_{0};
std::uint64_t lastOrderTime_{0};
std::uint64_t lastCancelOrderTime_{0};
OrderId orderId_{0};
Decimal orderPrice_{0};
//! 因为部成部撤后再次报单不能用orderSize_所以有了这个变量用于计算每次报单数量
Decimal curOrderSize_{0};
//! 成交数量有了两个作用：
//! 1. 用于部成部撤后重新下单的时候计算新的下单数量
//! 2. 计算当前订单结束时因为反复撤报导致的未成交的零股
Decimal dealSize_{0};
//! 成交均价，通知算法单进度用到
Decimal avgDealPrice_{0};
Urgency curUrgency_;
std::uint32_t curTickerOffset_{0};
std::uint32_t canceledTimes_{0};
int statusCode_{0};
NextActionOfAlgo nextActionOfAlgo_{NextActionOfAlgo::HandleCurOrder};
std::string toStr() const {
return fmt::format("no: {}; orderId:{}; orderSize: {}; orderPrice: {}", no_,
orderId_, orderSize_, orderPrice_);
}
};
using OrderStateMachine = msm::back::state_machine<OrderStateMachineImpl>;
using OrderStateMachineSPtr = std::shared_ptr<OrderStateMachine>;
using OrderStateMachineGroup = std::deque<OrderStateMachineSPtr>;
} // namespace bq::algo
1.3 分层状态机
分层状态机通过引入状态嵌套和继承机制，有效解决复杂状态结构下转换关系的维护问题
状态嵌套(State Nesting)：在分层状态机中，状态可以包含子状态，形成树状结构：
复合状态(Composite State): 包含子状态的状态
原子状态(Atomic State): 不包含子状态的叶节点状态
区域(Region): 并行执行的子状态集合
状态继承 (State Inheritance)：子状态可以继承父状态的行为：
进入/退出行为继承: 子状态进入/退出时自动触发父状态的进入/退出动作
事件处理继承: 子状态未处理的事件会委托给父状态处理
转换继承: 父状态定义的转换对所有子状态有效
正交区域 (Orthogonal Regions)：允许状态机同时处于多个互不影响的状态，非常适合描述并行行为，如：
交易执行与风险管理并行
市场监控与订单管理并行
数据采集与策略计算并行
做市商系统分层状态机示例：
MarketMakingSystem (根状态)
├── NormalOperation (正常运行)
│ ├── QuoteManagement (报价管理)
│ │ ├── AggressiveQuoting (激进报价)
│ │ ├── NeutralQuoting (中性报价)
│ │ └── ConservativeQuoting (保守报价)
│ └── InventoryManagement (库存管理)
│ ├── BalancedInventory (平衡库存)
│ ├── LongBias (偏多头寸)
│ └── ShortBias (偏空头寸)
├── RiskControlMode (风控模式)
│ ├── ReduceExposure (降低风险)
│ └── HaltTrading (暂停交易)
└── SystemMaintenance (系统维护)
├── DataRefresh (数据刷新)
└── StrategyUpdate (策略更新)
这种层次结构使系统能够：
在市场波动时，仅改变报价策略而不影响库存管理
当风险指标超标时，整个系统切换到风控模式
在维护期间，可以单独更新数据或策略，而不中断整个系统
1.4 会话状态机与报单状态机分离
1.分离模型：
2.重连恢复策略：
1.5 订单状态机的幂等与乱序回报处理
1.幂等处理原则
同一执行回报重复到达，最终状态不可二次推进。
使用稳定主键（如订单键 + 交易所回报序号）去重。
2.乱序处理原则
ACK/TRADE/CANCEL/REJECT 到达顺序不保证，状态机必须允许“延迟补齐”。
通过中间态（如 CANCEL_PENDING）表达时序竞争，避免非法跃迁。
1.6 其他订单管理系统接口设计
#include <cassert>
#include <string>
#include <unordered_map>
using namespace std;
enum OrderStatus {
PENDING,
REJECTED,
CONFIRMED,
COMPLETED,
};
struct OrderRequest {
string order_id;
string account;
string item_id;
string side;
double price;
int quantity;
};
struct OrderUpdate {
string order_id;
OrderStatus status;
};
unordered_map<string, OrderStatus> id_map;
unordered_map<string, unordered_map<string, OrderRequest>> account_map;
bool checkAndSaveOrder(const OrderRequest& req) {
if (id_map.find(req.order_id) == id_map.end()) {
id_map[req.order_id] = PENDING;
}
auto& orders_of_acc = account_map[req.account];
orders_of_acc[req.order_id] = req;
for (const auto& kv : orders_of_acc) {
const string& other_order_id = kv.first;
const OrderRequest& other_req = kv.second;
if (other_order_id == req.order_id) continue;
if (other_req.item_id != req.item_id) continue;
auto other_status_it = id_map.find(other_order_id);
OrderStatus other_status = PENDING;
if (other_status_it != id_map.end()) other_status = other_status_it->second;
if (other_status == REJECTED || other_status == COMPLETED) continue;
// opposite sides
if (req.side == "BUY" && other_req.side == "SELL") {
if (req.price >= other_req.price) return false;
return true;
} else if (req.side == "SELL" && other_req.side == "BUY") {
if (req.price <= other_req.price) return false;
return true;
}
}
return true;
}
bool updateOrderStatus(const OrderUpdate& update) {
auto it = id_map.find(update.order_id);
if (it == id_map.end()) {
return false;
}
OrderStatus& current_status = it->second;
OrderStatus new_status = update.status;
switch (current_status) {
case PENDING:
if (new_status == CONFIRMED || new_status == REJECTED) {
current_status = new_status;
return true;
}
break;
case CONFIRMED:
if (new_status == COMPLETED) {
current_status = new_status;
return true;
}
break;
case REJECTED:
case COMPLETED:
break;
}
return false;
}
int main() {
OrderRequest order1 = {"1", "U1", "A", "BUY", 100, 10};
assert(checkAndSaveOrder(order1));
OrderRequest order2 = {"2", "U1", "A", "SELL", 102, 5};
assert(checkAndSaveOrder(order2));
OrderRequest order3 = {"3", "U1", "A", "SELL", 99, 5};
assert(checkAndSaveOrder(order3));
// 测试状态流转：PENDING -> CONFIRMED -> COMPLETED
OrderUpdate update_pending_to_completed = {"1", COMPLETED};
assert(!updateOrderStatus(
update_pending_to_completed)); // 不能直接从PENDING到COMPLETED
OrderUpdate update_pending_to_confirmed = {"1", CONFIRMED};
assert(updateOrderStatus(
update_pending_to_confirmed)); // 先从PENDING到CONFIRMED
OrderUpdate update_confirmed_to_completed = {"1", COMPLETED};
assert(updateOrderStatus(
update_confirmed_to_completed)); // 再从CONFIRMED到COMPLETED
// 测试已拒绝的订单不能再处理
OrderUpdate update_rejected_order = {"1", REJECTED};
assert(!updateOrderStatus(update_rejected_order)); // 已完成的订单不能再拒绝
OrderRequest order4 = {"4", "U1", "A", "SELL", 98, 5};
assert(checkAndSaveOrder(order4));
return 0;
}
1.5 Event Sourcing 和 CQRS
事件溯源(Event Sourcing)：系统不直接存储当前状态，而是持久化所有导致状态变化的业务事件，通过重放这些不可变事件流来重建任意时间点的状态。
- 状态由事件驱动：不直接保存系统当前状态，而是将所有状态变更（如订单创建、撮合、取消） 以 “不可变事件” 的形式完整、有序地持久化。
- 状态可重建：通过 “重放” 历史事件序列，可以精确还原系统在任意时刻的状态。
- 天然可审计：事件日志完整记录了所有操作历史，便于故障排查、合规审计和回测。
在交易系统中，Event Sourcing 特别适合处理订单生命周期、资金流水、持仓变更等核心流程，能解决传统 “状态覆盖” 模式下的历史追溯难题。
#include <iostream>
#include <memory>
#include <unordered_map>
#include <vector>
// ====================== 1. 事件定义（不可变对象） ======================
// 事件类型
enum class EventType { ORDER_CREATED, ORDER_CANCELLED };
// 事件基类（多态用于统一存储）
class Event {
public:
virtual ~Event() = default;
virtual EventType type() const = 0;
virtual void print() const = 0; // 仅用于调试
};
// 具体事件：订单创建
class OrderCreated : public Event {
public:
OrderCreated(uint64_t order_id, std::string symbol, double price,
uint64_t qty)
: order_id_(order_id),
symbol_(std::move(symbol)),
price_(price),
qty_(qty) {}
EventType type() const override { return EventType::ORDER_CREATED; }
void print() const override {
std::cout << "[OrderCreated] ID: " << order_id_ << " | " << symbol_ << " @ "
<< price_ << " x " << qty_ << "\n";
}
// 事件数据不可变，仅提供 getter
uint64_t order_id_;
std::string symbol_;
double price_;
uint64_t qty_;
};
// 具体事件：订单取消
class OrderCancelled : public Event {
public:
OrderCancelled(uint64_t order_id) : order_id_(order_id) {}
EventType type() const override { return EventType::ORDER_CANCELLED; }
void print() const override {
std::cout << "[OrderCancelled] ID: " << order_id_ << "\n";
}
uint64_t order_id_;
};
// ====================== 2. 事件存储（只追加，不修改/删除）
// ======================
class EventStore {
public:
// 核心操作1：追加事件（Append-Only）
void append(std::unique_ptr<Event> event) {
events_.push_back(std::move(event));
}
// 核心操作2：获取所有历史事件（用于状态重建）
const std::vector<std::unique_ptr<Event>>& get_all_events() const {
return events_;
}
private:
std::vector<std::unique_ptr<Event>> events_; // 有序的事件日志
};
// ====================== 3. 状态投影（通过事件重放重建状态）
// ======================
class OrderState {
public:
// 核心：根据事件更新状态
void apply(const Event& event) {
if (event.type() == EventType::ORDER_CREATED) {
const auto& e = static_cast<const OrderCreated&>(event);
active_orders_[e.order_id_] = {e.symbol_, e.price_, e.qty_};
} else if (event.type() == EventType::ORDER_CANCELLED) {
const auto& e = static_cast<const OrderCancelled&>(event);
active_orders_.erase(e.order_id_);
}
}
// 打印当前状态
void print() const {
std::cout << "\n=== Current Active Orders ===\n";
for (const auto& [id, info] : active_orders_) {
std::cout << "ID: " << id << " | " << info.symbol << " @ " << info.price
<< " x " << info.qty << "\n";
}
}
private:
struct OrderInfo {
std::string symbol;
double price;
uint64_t qty;
};
std::unordered_map<uint64_t, OrderInfo> active_orders_; // 当前状态
};
// ====================== 4. 主流程演示 ======================
int main() {
// 1. 初始化事件存储和状态
EventStore store;
OrderState state;
// 2. 模拟业务操作：生成事件并追加到存储
store.append(std::make_unique<OrderCreated>(1001, "AAPL", 175.5, 200));
store.append(std::make_unique<OrderCreated>(1002, "GOOGL", 140.2, 150));
store.append(std::make_unique<OrderCancelled>(1001));
// 3. 打印事件日志（体现“完整历史记录”）
std::cout << "=== Event Log (Full History) ===\n";
for (const auto& e : store.get_all_events()) {
e->print();
}
// 4. 状态重建：重放所有事件
std::cout << "\n=== Rebuilding State from Events ===\n";
for (const auto& e : store.get_all_events()) {
state.apply(*e);
}
// 5. 打印最终状态
state.print();
return 0;
}
| 代码模块 | 对应概念 | 说明 |
|---|---|---|
OrderCreated / OrderCancelled |
不可变事件 | 事件对象仅包含“发生了什么”的数据，无 setter，一旦创建不可修改。 |
EventStore::append() |
只追加存储 | 事件仅能按顺序追加到日志，绝不修改或删除已有事件，保证历史完整可追溯。 |
OrderState::apply() + 事件重放 |
状态重建 | 当前状态不直接存储，而是通过“重放”历史事件序列动态计算得出。 |
=== Event Log (Full History) ===
[OrderCreated] ID: 1001 | AAPL @ 175.5 x 200
[OrderCreated] ID: 1002 | GOOGL @ 140.2 x 150
[OrderCancelled] ID: 1001
=== Rebuilding State from Events ===
=== Current Active Orders ===
ID: 1002 | GOOGL @ 140.2 x 150
- 左侧是完整的事件日志（记录“发生了什么”）；
- 右侧是通过日志重建的当前状态（记录“现在是什么样”）。
CQRS(命令查询职责分离)：将系统拆分为两个独立模型——命令端专注处理写操作和业务规则（优化事务处理），查询端专责读操作（优化数据检索），两者可使用不同技术栈、独立扩展，结合事件溯源后能实现高频交易与复杂分析的性能隔离。 CQRS与 Event Sourcing配合方式如下：
| 角色 | 职责 | 与 Event Sourcing 的配合 |
|---|---|---|
| Command 端（写模型） | 处理写操作、执行业务规则 | 以 Event Store 为唯一事实源： 1. 接收 Command 后，重放 Event Store 中的历史事件重建当前状态； 2. 执行业务逻辑，产生新的“不可变事件”； 3. 将新事件只追加到 Event Store（不修改/删除已有事件）。 |
| Event Store（事件存储） | 持久化所有事件 | 作为连接 Command 端和 Query 端的单一数据源，保证数据一致性。 |
| Query 端（读模型） | 处理读操作、优化数据展示 | 通过事件构建读模型： 1. 启动时从 Event Store 加载所有历史事件，“投影”（Project）为针对查询优化的读模型； 2. 运行时订阅 Event Store 的新事件，实时更新读模型。 |
Command → Event Store（唯一事实源）→ Query Model（投影）
#include <iostream>
#include <memory>
#include <string>
#include <unordered_map>
#include <unordered_set>
#include <vector>
// ====================== 1. 事件定义（不可变，ES + CQRS 的核心纽带）
// ======================
enum class EventType { ORDER_CREATED, ORDER_CANCELLED };
class Event {
public:
virtual ~Event() = default;
virtual EventType type() const = 0;
virtual void print() const = 0; // 仅用于调试
};
class OrderCreated : public Event {
public:
OrderCreated(uint64_t order_id, std::string symbol, double price,
uint64_t qty)
: order_id_(order_id),
symbol_(std::move(symbol)),
price_(price),
qty_(qty) {}
EventType type() const override { return EventType::ORDER_CREATED; }
void print() const override {
std::cout << "[Event] OrderCreated: ID=" << order_id_ << ", " << symbol_
<< " @ " << price_ << " x " << qty_ << "\n";
}
const uint64_t order_id_;
const std::string symbol_;
const double price_;
const uint64_t qty_;
};
class OrderCancelled : public Event {
public:
OrderCancelled(uint64_t order_id) : order_id_(order_id) {}
EventType type() const override { return EventType::ORDER_CANCELLED; }
void print() const override {
std::cout << "[Event] OrderCancelled: ID=" << order_id_ << "\n";
}
const uint64_t order_id_;
};
// ====================== 2. Event Store（唯一事实源，ES 核心）
// ======================
class EventStore {
public:
// ES 核心操作1：只追加事件（Append-Only）
void append(std::unique_ptr<Event> event) {
events_.push_back(std::move(event));
}
// ES 核心操作2：获取所有历史事件（用于 Command 端状态重建 + Query 端投影）
const std::vector<std::unique_ptr<Event>>& get_all_events() const {
return events_;
}
private:
std::vector<std::unique_ptr<Event>> events_; // 有序的事件日志
};
// ====================== 3. Command 端（写模型：结合 ES，以事件为事实源）
// ====================== Command 定义
class CreateOrderCommand {
public:
CreateOrderCommand(uint64_t order_id, std::string symbol, double price,
uint64_t qty)
: order_id_(order_id),
symbol_(std::move(symbol)),
price_(price),
qty_(qty) {}
const uint64_t order_id_;
const std::string symbol_;
const double price_;
const uint64_t qty_;
};
class CancelOrderCommand {
public:
CancelOrderCommand(uint64_t order_id) : order_id_(order_id) {}
const uint64_t order_id_;
};
// Command 处理器（结合 ES：状态通过重放事件重建）
class CommandHandler {
public:
CommandHandler(EventStore& store) : store_(store) {}
// 处理 CreateOrder Command
void handle(const CreateOrderCommand& cmd) {
// 1. 从 Event Store 重放事件，重建当前状态（ES 核心）
rebuild_state();
// 2. 执行业务逻辑（如：订单ID不能重复）
if (existing_order_ids_.find(cmd.order_id_) != existing_order_ids_.end()) {
std::cout << "[Command] Rejected: Order " << cmd.order_id_
<< " already exists\n";
return;
}
// 3. 产生新事件
auto event = std::make_unique<OrderCreated>(cmd.order_id_, cmd.symbol_,
cmd.price_, cmd.qty_);
// 4. 将事件追加到 Event Store（唯一事实源）
std::cout << "[Command] Accepted: Creating Order " << cmd.order_id_ << "\n";
store_.append(std::move(event));
}
// 处理 CancelOrder Command
void handle(const CancelOrderCommand& cmd) {
// 1. 从 Event Store 重放事件，重建当前状态（ES 核心）
rebuild_state();
// 2. 执行业务逻辑（如：订单必须存在才能取消）
if (existing_order_ids_.find(cmd.order_id_) == existing_order_ids_.end()) {
std::cout << "[Command] Rejected: Order " << cmd.order_id_
<< " not found\n";
return;
}
// 3. 产生新事件
auto event = std::make_unique<OrderCancelled>(cmd.order_id_);
// 4. 将事件追加到 Event Store（唯一事实源）
std::cout << "[Command] Accepted: Cancelling Order " << cmd.order_id_
<< "\n";
store_.append(std::move(event));
}
private:
// ES 核心：通过重放 Event Store 的历史事件重建状态
void rebuild_state() {
existing_order_ids_.clear();
for (const auto& e : store_.get_all_events()) {
if (e->type() == EventType::ORDER_CREATED) {
const auto& created = static_cast<const OrderCreated&>(*e);
existing_order_ids_.insert(created.order_id_);
} else if (e->type() == EventType::ORDER_CANCELLED) {
const auto& cancelled = static_cast<const OrderCancelled&>(*e);
existing_order_ids_.erase(cancelled.order_id_);
}
}
}
EventStore& store_;
std::unordered_set<uint64_t>
existing_order_ids_; // 临时状态，仅用于业务逻辑校验
};
// ====================== 4. Query 端（读模型：通过事件投影构建）
// ====================== Query 结果定义
struct OrderView {
uint64_t order_id;
std::string symbol;
double price;
uint64_t qty;
bool is_active;
};
// Query 处理器（结合 ES：读模型通过事件投影构建）
class QueryHandler {
public:
QueryHandler(EventStore& store) : store_(store) {
// 启动时：从 Event Store 加载所有历史事件，投影为读模型（ES + CQRS
// 配合核心）
rebuild_projection();
}
// 当 Event Store 有新事件时，调用此方法更新读模型
void on_new_event() { rebuild_projection(); }
// Query 1：获取活跃订单列表
std::vector<OrderView> get_active_orders() const {
std::vector<OrderView> result;
for (const auto& [id, view] : order_views_) {
if (view.is_active) result.push_back(view);
}
return result;
}
// Query 2：获取订单详情
OrderView get_order_details(uint64_t order_id) const {
auto it = order_views_.find(order_id);
if (it != order_views_.end()) return it->second;
return {};
}
private:
// CQRS + ES 配合核心：通过事件投影构建读模型
void rebuild_projection() {
order_views_.clear();
for (const auto& e : store_.get_all_events()) {
if (e->type() == EventType::ORDER_CREATED) {
const auto& created = static_cast<const OrderCreated&>(*e);
order_views_[created.order_id_] = {created.order_id_, created.symbol_,
created.price_, created.qty_, true};
} else if (e->type() == EventType::ORDER_CANCELLED) {
const auto& cancelled = static_cast<const OrderCancelled&>(*e);
if (order_views_.find(cancelled.order_id_) != order_views_.end()) {
order_views_[cancelled.order_id_].is_active = false;
}
}
}
}
EventStore& store_;
std::unordered_map<uint64_t, OrderView> order_views_; // 读模型：针对查询优化
};
// ====================== 5. 主流程：完整演示 CQRS + ES 配合链路
// ======================
int main() {
// 1. 初始化核心组件（Event Store 是唯一事实源）
EventStore store;
CommandHandler command_handler(store);
QueryHandler query_handler(store);
// ---------------------- 链路 1：Command 端 → Event Store
// ----------------------
std::cout << "=== Step 1: Sending Commands (Write Side) ===\n";
command_handler.handle(CreateOrderCommand(1001, "AAPL", 175.5, 200));
command_handler.handle(CreateOrderCommand(1002, "GOOGL", 140.2, 150));
command_handler.handle(CancelOrderCommand(1001));
command_handler.handle(CancelOrderCommand(9999)); // 测试业务逻辑拒绝
// 打印 Event Store 中的完整事件日志（ES 核心：完整历史）
std::cout << "\n=== Step 2: Event Store (Full History) ===\n";
for (const auto& e : store.get_all_events()) e->print();
// ---------------------- 链路 2：Event Store → Query 端（投影）
// ----------------------
std::cout << "\n=== Step 3: Query Side (Projection from Events) ===\n";
query_handler.on_new_event(); // 模拟订阅新事件
// 执行查询（读模型与写模型完全分离）
auto active_orders = query_handler.get_active_orders();
std::cout << "\n[Query] Active Orders:\n";
for (const auto& o : active_orders) {
std::cout << " ID: " << o.order_id << " | " << o.symbol << " @ " << o.price
<< " x " << o.qty << "\n";
}
auto order_1001 = query_handler.get_order_details(1001);
std::cout << "\n[Query] Order 1001 Details:\n";
std::cout << " ID: " << order_1001.order_id
<< " | Active: " << (order_1001.is_active ? "Yes" : "No") << "\n";
return 0;
}
| 代码位置 | 对应概念 | 说明 |
|---|---|---|
EventStore::append() |
ES 只追加存储 | 事件仅能按顺序追加，绝不修改/删除，保证历史完整。 |
CommandHandler::rebuild_state() |
ES 状态重建 | Command 端不保存当前状态，每次处理命令前通过重放事件重建。 |
CommandHandler::handle() 产生事件并追加到 EventStore |
CQRS 写模型 | Command 端仅负责“产生事件”，以 Event Store 为唯一事实源。 |
QueryHandler::rebuild_projection() |
CQRS 读模型 + 事件投影 | Query 端通过事件构建针对查询优化的读模型，与写模型完全分离。 |
主流程 Command → EventStore → Query |
CQRS + ES 完整链路 | 两者通过 Event Store 解耦，实现最终一致性。 |
=== Step 1: Sending Commands (Write Side) ===
[Command] Accepted: Creating Order 1001
[Command] Accepted: Creating Order 1002
[Command] Accepted: Cancelling Order 1001
[Command] Rejected: Order 9999 not found
=== Step 2: Event Store (Full History) ===
[Event] OrderCreated: ID=1001, AAPL @ 175.5 x 200
[Event] OrderCreated: ID=1002, GOOGL @ 140.2 x 150
[Event] OrderCancelled: ID=1001
=== Step 3: Query Side (Projection from Events) ===
[Query] Active Orders:
ID: 1002 | GOOGL @ 140.2 x 150
[Query] Order 1001 Details:
ID: 1001 | Active: No