【Python量化实盘工程 #01】回测到实盘，先补一个最小撮合引擎

系列定位：回测通过了，评估也合格，但离真正「上线」还差一个最小可运行实盘工程。本系列用 6 篇文章把「策略信号」推进到「可切换实盘/模拟盘的订单闭环」。

一、为什么回测漂亮，实盘就变形？

很多量化初学者都会遇到一个落差：

• 回测里设定「10 元买入 1000 股」，净值曲线立刻往上走；
• 实盘里同样的信号，成交价可能是 10.03，部分成交 600 股，剩下 400 股变成废单；
• 更隐蔽的是，回测假设所有订单瞬时、全额、按收盘价成交，而真实市场根本没有这个保证。

如果你打算把策略从研究环境搬到可运行系统，第一步不是接券商柜台，而是先在自己代码里补一个最小撮合引擎——它能告诉你：

这个信号发出去，按照市场规则，到底能不能成交、成交多少、成交在什么价位。

二、本文你将得到什么

读完本文，你可以直接复制以下内容到自己的项目里：

1. 一个事件驱动撮合引擎：限价单队列 + 价格-时间优先撮合；
2. 对 LIMIT / MARKET 两种订单的完整处理；
3. partial fill（部分成交） 与成交回报（Fill）的实现；
4. 用历史行情收盘价作为参考价的撮合演示；
5. 一段可直接运行的 Python 代码。

三、撮合引擎的核心概念

最小撮合引擎不需要对接真实交易所，但它必须模拟交易所最核心的规则：

规则	说明
价格优先	买单价越高越先成交；卖单价越低越先成交
时间优先	同价格下，先挂单者优先
partial fill	当对手盘数量不足时，只成交可匹配的部分
成交回报	每笔成交生成唯一的 Fill 记录，含价格、数量、时间

引擎内部维护两个优先队列：

• bids（买单队列）：按价格降序排列，用最大堆实现；
• asks（卖单队列）：按价格升序排列，用最小堆实现。

四、完整代码实战

以下代码不依赖任何外部框架，只需要 mairui 获取历史行情作为参考价。演示使用公开演示证书，实际项目请换成你自己的正式证书。

import heapq
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import List, Dict, Optional

from mairui import Client

DEMO_LICENCE = "LICENCE-66D8-9F96-0C7F0FBCD073"


class Side(Enum):
    BUY = "BUY"
    SELL = "SELL"


class OrderType(Enum):
    LIMIT = "LIMIT"
    MARKET = "MARKET"


class OrderStatus(Enum):
    NEW = "NEW"
    PARTIALLY_FILLED = "PARTIALLY_FILLED"
    FILLED = "FILLED"
    CANCELLED = "CANCELLED"


@dataclass
class Order:
    order_id: str
    side: Side
    order_type: OrderType
    quantity: int
    symbol: str
    price: Optional[float] = None
    timestamp: datetime = field(default_factory=datetime.now)
    filled_quantity: int = 0
    status: OrderStatus = OrderStatus.NEW

    @property
    def remaining(self) -> int:
        return self.quantity - self.filled_quantity

    def fill(self, qty: int):
        self.filled_quantity += qty
        if self.filled_quantity >= self.quantity:
            self.status = OrderStatus.FILLED
        else:
            self.status = OrderStatus.PARTIALLY_FILLED


@dataclass
class Fill:
    fill_id: str
    order_id: str
    symbol: str
    side: Side
    quantity: int
    price: float
    timestamp: datetime


class MatchingEngine:
    def __init__(self, symbol: str):
        self.symbol = symbol
        self.bids: List[tuple] = []   # (-price, timestamp, order)
        self.asks: List[tuple] = []   # (price, timestamp, order)
        self.fills: List[Fill] = []
        self._counter = 0

    def _next_id(self) -> str:
        self._counter += 1
        return f"F{self._counter:04d}"

    def submit(self, order: Order) -> List[Fill]:
        if order.symbol != self.symbol:
            raise ValueError(f"引擎只接受 {self.symbol}")
        fills = self._match(order)
        if order.remaining > 0 and order.order_type == OrderType.LIMIT:
            self._add_to_book(order)
        return fills

    def _match(self, order: Order) -> List[Fill]:
        fills: List[Fill] = []
        if order.side == Side.BUY:
            while order.remaining > 0 and self.asks:
                ask_price, _, ask = self.asks[0]
                ask_price = abs(ask_price)
                if order.order_type == OrderType.LIMIT and order.price < ask_price:
                    break
                fill_qty = min(order.remaining, ask.remaining)
                ask.fill(fill_qty)
                order.fill(fill_qty)
                fill = Fill(self._next_id(), order.order_id, self.symbol,
                            order.side, fill_qty, ask_price, datetime.now())
                fills.append(fill)
                self.fills.append(fill)
                if ask.status == OrderStatus.FILLED:
                    heapq.heappop(self.asks)
        else:
            while order.remaining > 0 and self.bids:
                bid_price_neg, _, bid = self.bids[0]
                bid_price = -bid_price_neg
                if order.order_type == OrderType.LIMIT and order.price > bid_price:
                    break
                fill_qty = min(order.remaining, bid.remaining)
                bid.fill(fill_qty)
                order.fill(fill_qty)
                fill = Fill(self._next_id(), order.order_id, self.symbol,
                            order.side, fill_qty, bid_price, datetime.now())
                fills.append(fill)
                self.fills.append(fill)
                if bid.status == OrderStatus.FILLED:
                    heapq.heappop(self.bids)
        return fills

    def _add_to_book(self, order: Order):
        if order.side == Side.BUY:
            heapq.heappush(self.bids, (-order.price, order.timestamp, order))
        else:
            heapq.heappush(self.asks, (order.price, order.timestamp, order))

    def book_snapshot(self) -> Dict:
        bids = sorted([(-p, o) for p, _, o in self.bids
                       if o.status != OrderStatus.FILLED],
                      key=lambda x: (-x[0], x[1].timestamp))
        asks = sorted([(p, o) for p, _, o in self.asks
                       if o.status != OrderStatus.FILLED],
                      key=lambda x: (x[0], x[1].timestamp))
        return {
            "bids": [(round(price, 2), o.remaining) for price, o in bids],
            "asks": [(round(price, 2), o.remaining) for price, o in asks],
        }


def fetch_last_price(symbol: str) -> float:
    client = Client(DEMO_LICENCE)
    bars = client.stock_history(symbol, period="d", dividend="f")
    return float(bars[-1]["c"])


def main():
    symbol = "600519"
    last_price = fetch_last_price(symbol)
    print(f"参考收盘价：{last_price:.2f} 元\n")

    engine = MatchingEngine(symbol)

    # 挂限价买单（低于最新价，进入订单簿）
    bid = Order(order_id="B001", side=Side.BUY, order_type=OrderType.LIMIT,
                quantity=100, symbol=symbol, price=round(last_price * 0.98, 2))
    engine.submit(bid)
    print(f"B001 LIMIT BUY 100 @ {bid.price}, 状态={bid.status.value}")

    # 高价卖单，不撮合，进入订单簿
    ask_high = Order(order_id="S001", side=Side.SELL, order_type=OrderType.LIMIT,
                     quantity=50, symbol=symbol, price=round(last_price * 1.02, 2))
    engine.submit(ask_high)
    print(f"S001 LIMIT SELL 50 @ {ask_high.price}, 状态={ask_high.status.value}")

    # 与 B001 同价卖单，触发 partial fill
    ask = Order(order_id="S002", side=Side.SELL, order_type=OrderType.LIMIT,
                quantity=60, symbol=symbol, price=bid.price)
    fills = engine.submit(ask)
    for f in fills:
        print(f"  成交 {f.fill_id}: 价={f.price:.2f}, 量={f.quantity}")
    print(f"B001 状态={bid.status.value}, 已成交={bid.filled_quantity}, 剩余={bid.remaining}")
    print(f"S002 状态={ask.status.value}, 已成交={ask.filled_quantity}, 剩余={ask.remaining}")

    # 市价买单，与最优卖价 S001 成交
    market_buy = Order(order_id="B002", side=Side.BUY, order_type=OrderType.MARKET,
                       quantity=30, symbol=symbol)
    fills = engine.submit(market_buy)
    for f in fills:
        print(f"  成交 {f.fill_id}: 价={f.price:.2f}, 量={f.quantity}")
    print(f"B002 状态={market_buy.status.value}, 已成交={market_buy.filled_quantity}")

    print("\n订单簿快照：", engine.book_snapshot())
    total = sum(f.quantity * f.price for f in engine.fills)
    print(f"总成交额：{total:.2f} 元")


if __name__ == "__main__":
    main()

五、运行结果（演示环境）

参考收盘价：11.26 元

B001 LIMIT BUY 100 @ 11.03, 状态=NEW
S001 LIMIT SELL 50 @ 11.49, 状态=NEW
  成交 F0001: 价=11.03, 量=60
B001 状态=PARTIALLY_FILLED, 已成交=60, 剩余=40
S002 状态=FILLED, 已成交=60, 剩余=0
  成交 F0002: 价=11.49, 量=30
B002 状态=FILLED, 已成交=30

订单簿快照： {'bids': [(11.03, 40)], 'asks': [(11.49, 20)]}
总成交额：1006.50 元

结果解读：

• B001 限价买 100 股 @11.03，因低于最新价先挂进订单簿；
• S001 限价卖 @11.49，价格高于买价，不撮合；
• S002 限价卖 @11.03，与 B001 撮合，但数量只有 60 股，触发 partial fill；
• B002 市价买 30 股，按最优卖价 11.49 成交，体现了市价单的滑点成本；
• 最终订单簿剩下 B001 未成交的 40 股和 S001 未成交的 20 股。

六、常见坑与注意

1. MARKET 单没有价格保护：本示例中市价单以最优卖一价成交；真实系统里建议加「价格笼子」或「最大可接受价」，防止极端行情下以离谱价格成交。
2. 订单簿要持久化：进程重启后订单簿会丢失，生产环境需要落库（Redis/SQLite）并做快照恢复。
3. 部分成交后的状态机：订单从 NEW → PARTIALLY_FILLED → FILLED，风控和仓位模块要根据这个状态判断真实持仓，而不是看下单数量。
4. 撤单与改单：本文的最小引擎没有实现 cancel/replace，但 OrderStatus.CANCELLED 已预留；下一篇将在状态机里展开。

七、小结 + 下篇预告

本文实现了一个最小撮合引擎，核心目的是让你在把策略送上实盘之前，先在代码里回答一个问题：

这个信号，按交易所规则，到底会成交成什么样？

下篇我们将在这个引擎之上，补上仓位与风控系统——止损、止盈、最大持仓、单日亏损上限、波动率降仓，把这 5 条红线用代码锁死。

免责声明：本文仅供技术学习与交流，不构成任何投资建议。市场有风险，决策需谨慎。

---

代码与文档：https://github.com/MaiRuiApi

转载自 CSDN-专业IT技术社区

原文链接：https://blog.csdn.net/2501_94338261/article/details/167586999