

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

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

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

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

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

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

## 二、本文你将得到什么

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

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

## 三、撮合引擎的核心概念

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

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

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

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

## 四、完整代码实战

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

```python
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()
```

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

```text
参考收盘价：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


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

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

编程语言

开发工具

算法

本文实现一个最小撮合引擎，模拟限价单与市价单的价格-时间优先撮合、部分成交及成交回报。通过事件驱动队列与订单状态机，填补回测到实盘的规则缺口，提供可运行Python代码，验证信号能否按市场规则成交，为订单闭环与风控系统奠定基础。

代码解释

SQLite

Python

GitHub

Redis

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

【Python量化实盘工程 #01】回测到实盘，先补一个最小撮合引擎-腾讯云开发者社区-腾讯云

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

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

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐