3天吃透新时代证券交易软件架构，避开80%的高频面试题
别去啃那几万字官方文档了，没人有空。面试官问“新时代证券交易软件”的核心逻辑，你翻书找答案？直接凉凉。
我见过太多转岗做量化或交易系统的开发者，卡死在文档迷宫里。其实核心就三点：行情推送、订单撮合、风控前置。
把这三块代码逻辑跑通，高频面试题里的80%场景你就覆盖了。
项目目标与业务边界
做交易系统，别一上来就搞分布式。先定死边界。
本项目模拟“新时代证券交易软件”的核心交易引擎。目标不是做App，而是做服务端撮合引擎。
核心功能拆解：
行情接入：模拟Level-1行情推送，延迟控制在毫秒级。
订单管理：支持限价单、市价单，处理订单状态机（New, PartiallyFilled, Filled, Cancelled）。
实时风控：在订单进入撮合队列前，进行资金与持仓校验。
数据持久化：订单流水落库，支持对账。
为什么这么设计？
因为面试高频考点集中在“一致性”和“低延迟”。
如果只写一个Web API接收订单，那是CRUD，不是交易软件。
交易软件的本质是高并发下的状态同步。
目录结构规划
工程化是区分初级和高级的分水岭。别把所有代码堆在一个文件里。
推荐采用领域驱动设计（DDD）的简化版结构，清晰划分职责。
trading-engine/
├── config/
│ └── settings.yaml # 配置中心：行情源、风控阈值、DB连接
├── core/
│ ├── engine.py # 主引擎：事件循环入口
│ ├── matcher.py # 撮合器：价格优先、时间优先
│ ├── risk_manager.py # 风控器：资金/持仓/频率检查
│ └── order_book.py # 订单簿：买卖盘口数据结构
├── models/
│ ├── order.py # 订单模型：状态机定义
│ └── quote.py # 行情模型
├── data/
│ ├── db.py # 数据库连接池
│ └── repository.py # 数据访问层：订单持久化
├── utils/
│ ├── logger.py # 日志：结构化日志，便于排查
│ └── time_utils.py # 时间工具：纳秒级时间戳
├── main.py # 启动脚本
└── tests/
└── test_matcher.py # 单元测试：撮合逻辑
关键点：
order_book.py 是性能核心，必须用内存数据结构，严禁查库。
risk_manager.py 必须独立，风控不过，订单连队列都进不去。
models/order.py 的状态机要严谨，防止非法状态流转。
核心代码实现
这是最干货的部分。逐行讲解，直接可跑。
1. 订单模型与状态机
订单不是字典，是对象。状态流转必须受控。
# models/order.py
from enum import Enum
from dataclasses import dataclass, field
from typing import Optional
import time
class OrderStatus(Enum):
PENDING = PENDING # 待处理
OPEN = OPEN # 已报入订单簿
PARTIALLY_FILLED = PARTIALLY_FILLED # 部分成交
FILLED = FILLED # 全部成交
CANCELLED = CANCELLED # 已撤销
REJECTED = REJECTED # 被风控拒绝
@dataclass
class Order:
order_id: str
symbol: str # 股票代码，如 600519
side: str # BUY 或 SELL
price: float # 限价
quantity: int # 委托数量
status: OrderStatus = OrderStatus.PENDING
filled_quantity: int = 0
create_time: float = field(default_factory=time.time_ns)
# 关键：防止状态非法流转
def can_transition_to(self, new_status: OrderStatus) - bool:
if self.status == OrderStatus.FILLED:
return False # 成交后不可变
if self.status == OrderStatus.CANCELLED:
return False # 撤销后不可变
return True
避坑指南：
很多新手用字典存订单，状态改来改去。面试时被问“如何保证订单状态一致性”，直接答不上来。
用枚举+方法封装，从代码层面杜绝非法状态。
2. 订单簿：性能的核心
撮合引擎的瓶颈在订单簿。必须用双向链表+堆或者TreeMap思想。
这里为了代码简洁，用Python的heapq模拟，但在生产环境Go/Rust会用更复杂的数据结构。
# core/order_book.py
import heapq
from typing import List, Dict
from models.order import Order, OrderStatus
class OrderBook:
def __init__(self, symbol: str):
self.symbol = symbol
# 买盘：最小堆，价格高的优先级高，所以取负值
self.bids: List[Order] = []
# 卖盘：最小堆，价格低的优先级高
self.asks: List[Order] = []
# 索引：用于快速查找特定订单（撤销用）
self.order_index: Dict[str, Order] = {}
def add_order(self, order: Order):
# 添加到索引
self.order_index[order.order_id] = order
order.status = OrderStatus.OPEN
if order.side == BUY:
# 买盘：价格高优先。Python heap是最小堆，所以存负价格
heapq.heappush(self.bids, (-order.price, order.create_time, order))
else:
# 卖盘：价格低优先
heapq.heappush(self.asks, (order.price, order.create_time, order))
def get_best_bid(self) - Optional[Order]:
if not self.bids:
return None
# 懒删除：堆顶可能已成交或撤销
while self.bids:
_, _, order = self.bids[0]
if order.status == OrderStatus.OPEN:
return order
else:
heapq.heappop(self.bids) # 清理无效订单
return None
def get_best_ask(self) - Optional[Order]:
if not self.asks:
return None
while self.asks:
_, _, order = self.asks[0]
if order.status == OrderStatus.OPEN:
return order
else:
heapq.heappop(self.asks)
return None
高频考点解析：
面试官常问：“如何保证时间优先？”
答案：同一价格下，先报的单子先成交。
代码里 create_time 是堆的第二个比较键。Python的堆比较元组时，先比第一个元素，相等再比第二个。这完美实现了“价格优先、时间优先”。
3. 撮合引擎：逻辑闭环
撮合器是心脏。它连接订单簿和成交回报。
# core/matcher.py
from core.order_book import OrderBook
from models.order import Order, OrderStatus
from typing import List
class Matcher:
def __init__(self, order_book: OrderBook):
self.book = order_book
self.trades: List[dict] = [] # 模拟成交记录
def match(self, incoming_order: Order):
核心撮合逻辑
# 1. 风控检查（简化版，实际应异步或前置）
if not self._risk_check(incoming_order):
incoming_order.status = OrderStatus.REJECTED
return
# 2. 加入订单簿
self.book.add_order(incoming_order)
# 3. 尝试撮合
while True:
if incoming_order.side == BUY:
best_ask = self.book.get_best_ask()
# 买价 = 卖价，成交
if best_ask and incoming_order.price = best_ask.price:
self._execute_trade(incoming_order, best_ask)
if incoming_order.status == OrderStatus.FILLED:
break
else:
break
else: # SELL
best_bid = self.book.get_best_bid()
# 卖价 = 买价，成交
if best_bid and incoming_order.price = best_bid.price:
self._execute_trade(incoming_order, best_bid)
if incoming_order.status == OrderStatus.FILLED:
break
else:
break
def _execute_trade(self, buy_order: Order, sell_order: Order):
执行单笔撮合
注意：实际中buy_order可能是新进来的，也可能是挂单
# 1. 确定成交量：取两者剩余量的最小值
buy_remaining = buy_order.quantity - buy_order.filled_quantity
sell_remaining = sell_order.quantity - sell_order.filled_quantity
trade_qty = min(buy_remaining, sell_remaining)
if trade_qty = 0:
return
# 2. 成交价：以先报入的订单价格为准（这里简化为卖方价格，实际看谁先报）
# 严格来说，应该比较create_time
if buy_order.create_time sell_order.create_time:
price = buy_order.price
else:
price = sell_order.price
# 3. 更新状态
buy_order.filled_quantity += trade_qty
sell_order.filled_quantity += trade_qty
# 记录成交
self.trades.append({
price: price,
quantity: trade_qty,
buy_id: buy_order.order_id,
sell_id: sell_order.order_id
})
# 4. 更新订单状态
if buy_order.filled_quantity == buy_order.quantity:
buy_order.status = OrderStatus.FILLED
else:
buy_order.status = OrderStatus.PARTIALLY_FILLED
if sell_order.filled_quantity == sell_order.quantity:
sell_order.status = OrderStatus.FILLED
else:
sell_order.status = OrderStatus.PARTIALLY_FILLED
def _risk_check(self, order: Order) - bool:
# 简化风控：价格不能偏离最新成交价5%
# 实际项目这里要查数据库或Redis获取最新价
return True
逐行逻辑解析：
循环撮合：一个订单可能吃掉多个对手盘。while True 循环直到订单全成或无法成交。
成交量计算：min(buy_remaining, sell_remaining)，这是铁律。
价格确定：这里简化了。在真实系统里，价格优先意味着新来的订单如果价格更优，它应该以旧订单的价格成交，而不是新订单的价格。代码中用 create_time 判断，谁先报，就用谁的价格。这是面试超高频细节。
运行与测试
代码写完了，不跑等于白写。
1. 启动脚本
# main.py
import uuid
from core.order_book import OrderBook
from core.matcher import Matcher
from models.order import Order
def main():
symbol = 600519
book = OrderBook(symbol)
matcher = Matcher(book)
print(f--- 开始测试 {symbol} 撮合 ---)
# 场景1：买单挂盘
buy1 = Order(order_id=str(uuid.uuid4()), symbol=symbol, side=BUY, price=1680.0, quantity=100)
matcher.match(buy1)
print(fBuy1: {buy1.status}, Filled: {buy1.filled_quantity})
# 预期：PENDING - OPEN (无对手盘)
# 场景2：卖单挂盘
sell1 = Order(order_id=str(uuid.uuid4()), symbol=symbol, side=SELL, price=1682.0, quantity=50)
matcher.match(sell1)
print(fSell1: {sell1.status}, Filled: {sell1.filled_quantity})
# 预期：PENDING - OPEN (卖价1682 买价1680，不成交)
# 场景3：买单吃单
buy2 = Order(order_id=str(uuid.uuid4()), symbol=symbol, side=BUY, price=1682.0, quantity=30)
matcher.match(buy2)
print(fBuy2: {buy2.status}, Filled: {buy2.filled_quantity})
# 预期：FILLED (1682 = 1682, 成交30股)
print(fSell1 Update: {sell1.status}, Filled: {sell1.filled_quantity})
# 预期：PARTIALLY_FILLED (100-30=70 remaining? No, sell1 qty 50, buy2 qty 30. Sell1 filled 30)
# 场景4：卖单吃单
sell2 = Order(order_id=str(uuid.uuid4()), symbol=symbol, side=SELL, price=1680.0, quantity=20)
matcher.match(sell2)
print(fSell2: {sell2.status}, Filled: {sell2.filled_quantity})
# 预期：FILLED (1680 = 1680, 成交20股)
print(fBuy1 Update: {buy1.status}, Filled: {buy1.filled_quantity})
# 预期：PARTIALLY_FILLED (100-20=80 remaining)
print(f\n--- 成交记录 ---)
for trade in matcher.trades:
print(trade)
if __name__ == __main__:
main()
2. 单元测试
测试必须覆盖边界情况：部分成交、全部成交、价格相等、时间优先。
# tests/test_matcher.py
import unittest
from core.order_book import OrderBook
from core.matcher import Matcher
from models.order import Order, OrderStatus
class TestMatcher(unittest.TestCase):
def setUp(self):
self.book = OrderBook(TEST)
self.matcher = Matcher(self.book)
def test_price_priority(self):
# 买单1：100元，100股
b1 = Order(b1, TEST, BUY, 100.0, 100)
self.matcher.match(b1)
# 买单2：101元，100股
b2 = Order(b2, TEST, BUY, 101.0, 100)
self.matcher.match(b2)
# 卖单1：100.5元，50股
s1 = Order(s1, TEST, SELL, 100.5, 50)
self.matcher.match(s1)
# 预期：卖单1应该和买单2（价格高）成交，而不是买单1
self.assertEqual(s1.status, OrderStatus.FILLED)
self.assertEqual(b2.filled_quantity, 50)
self.assertEqual(b1.filled_quantity, 0)
# 验证成交价：应该是买单2的价格101.0（因为b2先于s1报入？不，b2报入时s1还没报。
# 这里逻辑：s1进来，看到best_bid是b2(101)。100.5 = 101，成交。
# 价格：b2.create_time s1.create_time，所以价格是b2的101.0
self.assertEqual(self.matcher.trades[0][price], 101.0)
if __name__ == __main__:
unittest.main()
运行结果：
--- 开始测试 600519 撮合 ---
Buy1: OrderStatus.OPEN, Filled: 0
Sell1: OrderStatus.OPEN, Filled: 0
Buy2: OrderStatus.FILLED, Filled: 30
Sell1 Update: OrderStatus.PARTIALLY_FILLED, Filled: 30
Sell2: OrderStatus.FILLED, Filled: 20
Buy1 Update: OrderStatus.PARTIALLY_FILLED, Filled: 20
--- 成交记录 ---
{'price': 1682.0, 'quantity': 30, 'buy_id': '...', 'sell_id': '...'}
{'price': 1680.0, 'quantity': 20, 'buy_id': '...', 'sell_id': '...'}
看到没？Buy2 以 1682 成交，Sell2 以 1680 成交。
这就是“价格优先”的体现。 如果代码写错，价格取平均或者取新订单价格，面试直接挂。
优化扩展与避坑
代码能跑，不代表能上生产。
1. 并发安全
上面的代码是单线程的。实际交易是高并发。
问题：多线程同时修改 OrderBook，堆结构会乱。
方案：
Python：threading.Lock 保护 add_order 和 match。或者用 asyncio 单线程事件循环，避免锁开销。
Go/Rust：用 Mutex 或 RwLock。更高级的是无锁队列（Lock-Free Queue）。
面试高频问法：
“如何保证高并发下订单簿的线程安全？”
标准答案：
“采用单线程事件循环模型（如Go的Goroutine + Channel，或Python的Asyncio）。所有订单操作通过Channel发送到主协程，主协程串行处理撮合。这样天然避免锁竞争，且吞吐量极高。”
2. 持久化与对账
内存数据重启就没了。
方案：
订单状态：每次状态变更，异步写入Redis（缓存）和MySQL（持久化）。
成交记录：实时写入消息队列（Kafka），由下游服务落库。
避坑：
别在撮合主循环里同步写数据库！
一定要异步。 主循环只负责内存计算，耗时操作全部扔到后台队列。
3. 风控前置
代码里的 _risk_check 是同步的。
实际中，风控应该前置。
架构：订单进来 - 风控网关（查资金、查持仓、查频率） - 风控通过 - 撮合引擎。
目的：保护撮合引擎。如果非法订单都进撮合引擎，内存爆了怎么办？
4. 日志与监控
交易软件，日志就是生命线。
记录：订单ID、用户ID、价格、数量、状态变更、耗时。
格式：JSON结构化，方便ELK检索。
监控：监控订单延迟（从接收到成交回报的时间）、队列积压长度。
小结与互动
这套代码，从模型定义到撮合逻辑，完整覆盖了“新时代证券交易软件”的核心骨架。
你掌握了什么？
订单状态机：防止非法流转。
订单簿数据结构：价格优先、时间优先的实现细节。
撮合逻辑：部分成交、全部成交、价格确定规则。
架构思想：单线程事件循环、风控前置、异步持久化。
这些知识点，覆盖了高频面试题中关于交易系统设计的90%。
剩下的10%是分布式事务、容灾切换，那是高级岗位的事。
最后问一句：
这个知识点你面试被问过吗？
特别是“价格优先、时间优先”的具体实现，以及“如何保证撮合引擎的线程安全”。
留言说说，你当时怎么答的？或者你踩过什么坑？咱们评论区见。 
长河君 · 2026/9/22 13:50:42 · 阅读 1.2k 3天吃透新时代证券交易软件架构，避开80%的高频面试题
3天吃透新时代证券交易软件架构，避开80%的高频面试题
别去啃那几万字官方文档了，没人有空。面试官问“新时代证券交易软件”的核心逻辑，你翻书找答案？直接凉凉。
我见过太多转岗做量化或交易系统的开发者，卡死在文档迷宫里。其实核心就三点：行情推送、订单撮合、风控前置。
把这三块代码逻辑跑通，高频面试题里的80%场景你就覆盖了。
项目目标与业务边界
做交易系统，别一上来就搞分布式。先定死边界。
本项目模拟“新时代证券交易软件”的核心交易引擎。目标不是做App，而是做服务端撮合引擎。
核心功能拆解：
行情接入：模拟Level-1行情推送，延迟控制在毫秒级。
订单管理：支持限价单、市价单，处理订单状态机（New, PartiallyFilled, Filled, Cancelled）。
实时风控：在订单进入撮合队列前，进行资金与持仓校验。
数据持久化：订单流水落库，支持对账。
为什么这么设计？
因为面试高频考点集中在“一致性”和“低延迟”。
如果只写一个Web API接收订单，那是CRUD，不是交易软件。
交易软件的本质是高并发下的状态同步。
目录结构规划
工程化是区分初级和高级的分水岭。别把所有代码堆在一个文件里。
推荐采用领域驱动设计（DDD）的简化版结构，清晰划分职责。
trading-engine/
├── config/
│ └── settings.yaml # 配置中心：行情源、风控阈值、DB连接
├── core/
│ ├── engine.py # 主引擎：事件循环入口
│ ├── matcher.py # 撮合器：价格优先、时间优先
│ ├── risk_manager.py # 风控器：资金/持仓/频率检查
│ └── order_book.py # 订单簿：买卖盘口数据结构
├── models/
│ ├── order.py # 订单模型：状态机定义
│ └── quote.py # 行情模型
├── data/
│ ├── db.py # 数据库连接池
│ └── repository.py # 数据访问层：订单持久化
├── utils/
│ ├── logger.py # 日志：结构化日志，便于排查
│ └── time_utils.py # 时间工具：纳秒级时间戳
├── main.py # 启动脚本
└── tests/
└── test_matcher.py # 单元测试：撮合逻辑
关键点：
order_book.py 是性能核心，必须用内存数据结构，严禁查库。
risk_manager.py 必须独立，风控不过，订单连队列都进不去。
models/order.py 的状态机要严谨，防止非法状态流转。
核心代码实现
这是最干货的部分。逐行讲解，直接可跑。
1. 订单模型与状态机
订单不是字典，是对象。状态流转必须受控。
# models/order.py
from enum import Enum
from dataclasses import dataclass, field
from typing import Optional
import time
class OrderStatus(Enum):
PENDING = PENDING # 待处理
OPEN = OPEN # 已报入订单簿
PARTIALLY_FILLED = PARTIALLY_FILLED # 部分成交
FILLED = FILLED # 全部成交
CANCELLED = CANCELLED # 已撤销
REJECTED = REJECTED # 被风控拒绝
@dataclass
class Order:
order_id: str
symbol: str # 股票代码，如 600519
side: str # BUY 或 SELL
price: float # 限价
quantity: int # 委托数量
status: OrderStatus = OrderStatus.PENDING
filled_quantity: int = 0
create_time: float = field(default_factory=time.time_ns)
# 关键：防止状态非法流转
def can_transition_to(self, new_status: OrderStatus) - bool:
if self.status == OrderStatus.FILLED:
return False # 成交后不可变
if self.status == OrderStatus.CANCELLED:
return False # 撤销后不可变
return True
避坑指南：
很多新手用字典存订单，状态改来改去。面试时被问“如何保证订单状态一致性”，直接答不上来。
用枚举+方法封装，从代码层面杜绝非法状态。
2. 订单簿：性能的核心
撮合引擎的瓶颈在订单簿。必须用双向链表+堆或者TreeMap思想。
这里为了代码简洁，用Python的heapq模拟，但在生产环境Go/Rust会用更复杂的数据结构。
# core/order_book.py
import heapq
from typing import List, Dict
from models.order import Order, OrderStatus
class OrderBook:
def __init__(self, symbol: str):
self.symbol = symbol
# 买盘：最小堆，价格高的优先级高，所以取负值
self.bids: List[Order] = []
# 卖盘：最小堆，价格低的优先级高
self.asks: List[Order] = []
# 索引：用于快速查找特定订单（撤销用）
self.order_index: Dict[str, Order] = {}
def add_order(self, order: Order):
# 添加到索引
self.order_index[order.order_id] = order
order.status = OrderStatus.OPEN
if order.side == BUY:
# 买盘：价格高优先。Python heap是最小堆，所以存负价格
heapq.heappush(self.bids, (-order.price, order.create_time, order))
else:
# 卖盘：价格低优先
heapq.heappush(self.asks, (order.price, order.create_time, order))
def get_best_bid(self) - Optional[Order]:
if not self.bids:
return None
# 懒删除：堆顶可能已成交或撤销
while self.bids:
_, _, order = self.bids[0]
if order.status == OrderStatus.OPEN:
return order
else:
heapq.heappop(self.bids) # 清理无效订单
return None
def get_best_ask(self) - Optional[Order]:
if not self.asks:
return None
while self.asks:
_, _, order = self.asks[0]
if order.status == OrderStatus.OPEN:
return order
else:
heapq.heappop(self.asks)
return None
高频考点解析：
面试官常问：“如何保证时间优先？”
答案：同一价格下，先报的单子先成交。
代码里 create_time 是堆的第二个比较键。Python的堆比较元组时，先比第一个元素，相等再比第二个。这完美实现了“价格优先、时间优先”。
3. 撮合引擎：逻辑闭环
撮合器是心脏。它连接订单簿和成交回报。
# core/matcher.py
from core.order_book import OrderBook
from models.order import Order, OrderStatus
from typing import List
class Matcher:
def __init__(self, order_book: OrderBook):
self.book = order_book
self.trades: List[dict] = [] # 模拟成交记录
def match(self, incoming_order: Order):
核心撮合逻辑
# 1. 风控检查（简化版，实际应异步或前置）
if not self._risk_check(incoming_order):
incoming_order.status = OrderStatus.REJECTED
return
# 2. 加入订单簿
self.book.add_order(incoming_order)
# 3. 尝试撮合
while True:
if incoming_order.side == BUY:
best_ask = self.book.get_best_ask()
# 买价 = 卖价，成交
if best_ask and incoming_order.price = best_ask.price:
self._execute_trade(incoming_order, best_ask)
if incoming_order.status == OrderStatus.FILLED:
break
else:
break
else: # SELL
best_bid = self.book.get_best_bid()
# 卖价 = 买价，成交
if best_bid and incoming_order.price = best_bid.price:
self._execute_trade(incoming_order, best_bid)
if incoming_order.status == OrderStatus.FILLED:
break
else:
break
def _execute_trade(self, buy_order: Order, sell_order: Order):
执行单笔撮合
注意：实际中buy_order可能是新进来的，也可能是挂单
# 1. 确定成交量：取两者剩余量的最小值
buy_remaining = buy_order.quantity - buy_order.filled_quantity
sell_remaining = sell_order.quantity - sell_order.filled_quantity
trade_qty = min(buy_remaining, sell_remaining)
if trade_qty = 0:
return
# 2. 成交价：以先报入的订单价格为准（这里简化为卖方价格，实际看谁先报）
# 严格来说，应该比较create_time
if buy_order.create_time sell_order.create_time:
price = buy_order.price
else:
price = sell_order.price
# 3. 更新状态
buy_order.filled_quantity += trade_qty
sell_order.filled_quantity += trade_qty
# 记录成交
self.trades.append({
price: price,
quantity: trade_qty,
buy_id: buy_order.order_id,
sell_id: sell_order.order_id
})
# 4. 更新订单状态
if buy_order.filled_quantity == buy_order.quantity:
buy_order.status = OrderStatus.FILLED
else:
buy_order.status = OrderStatus.PARTIALLY_FILLED
if sell_order.filled_quantity == sell_order.quantity:
sell_order.status = OrderStatus.FILLED
else:
sell_order.status = OrderStatus.PARTIALLY_FILLED
def _risk_check(self, order: Order) - bool:
# 简化风控：价格不能偏离最新成交价5%
# 实际项目这里要查数据库或Redis获取最新价
return True
逐行逻辑解析：
循环撮合：一个订单可能吃掉多个对手盘。while True 循环直到订单全成或无法成交。
成交量计算：min(buy_remaining, sell_remaining)，这是铁律。
价格确定：这里简化了。在真实系统里，价格优先意味着新来的订单如果价格更优，它应该以旧订单的价格成交，而不是新订单的价格。代码中用 create_time 判断，谁先报，就用谁的价格。这是面试超高频细节。
运行与测试
代码写完了，不跑等于白写。
1. 启动脚本
# main.py
import uuid
from core.order_book import OrderBook
from core.matcher import Matcher
from models.order import Order
def main():
symbol = 600519
book = OrderBook(symbol)
matcher = Matcher(book)
print(f--- 开始测试 {symbol} 撮合 ---)
# 场景1：买单挂盘
buy1 = Order(order_id=str(uuid.uuid4()), symbol=symbol, side=BUY, price=1680.0, quantity=100)
matcher.match(buy1)
print(fBuy1: {buy1.status}, Filled: {buy1.filled_quantity})
# 预期：PENDING - OPEN (无对手盘)
# 场景2：卖单挂盘
sell1 = Order(order_id=str(uuid.uuid4()), symbol=symbol, side=SELL, price=1682.0, quantity=50)
matcher.match(sell1)
print(fSell1: {sell1.status}, Filled: {sell1.filled_quantity})
# 预期：PENDING - OPEN (卖价1682 买价1680，不成交)
# 场景3：买单吃单
buy2 = Order(order_id=str(uuid.uuid4()), symbol=symbol, side=BUY, price=1682.0, quantity=30)
matcher.match(buy2)
print(fBuy2: {buy2.status}, Filled: {buy2.filled_quantity})
# 预期：FILLED (1682 = 1682, 成交30股)
print(fSell1 Update: {sell1.status}, Filled: {sell1.filled_quantity})
# 预期：PARTIALLY_FILLED (100-30=70 remaining? No, sell1 qty 50, buy2 qty 30. Sell1 filled 30)
# 场景4：卖单吃单
sell2 = Order(order_id=str(uuid.uuid4()), symbol=symbol, side=SELL, price=1680.0, quantity=20)
matcher.match(sell2)
print(fSell2: {sell2.status}, Filled: {sell2.filled_quantity})
# 预期：FILLED (1680 = 1680, 成交20股)
print(fBuy1 Update: {buy1.status}, Filled: {buy1.filled_quantity})
# 预期：PARTIALLY_FILLED (100-20=80 remaining)
print(f\n--- 成交记录 ---)
for trade in matcher.trades:
print(trade)
if __name__ == __main__:
main()
2. 单元测试
测试必须覆盖边界情况：部分成交、全部成交、价格相等、时间优先。
# tests/test_matcher.py
import unittest
from core.order_book import OrderBook
from core.matcher import Matcher
from models.order import Order, OrderStatus
class TestMatcher(unittest.TestCase):
def setUp(self):
self.book = OrderBook(TEST)
self.matcher = Matcher(self.book)
def test_price_priority(self):
# 买单1：100元，100股
b1 = Order(b1, TEST, BUY, 100.0, 100)
self.matcher.match(b1)
# 买单2：101元，100股
b2 = Order(b2, TEST, BUY, 101.0, 100)
self.matcher.match(b2)
# 卖单1：100.5元，50股
s1 = Order(s1, TEST, SELL, 100.5, 50)
self.matcher.match(s1)
# 预期：卖单1应该和买单2（价格高）成交，而不是买单1
self.assertEqual(s1.status, OrderStatus.FILLED)
self.assertEqual(b2.filled_quantity, 50)
self.assertEqual(b1.filled_quantity, 0)
# 验证成交价：应该是买单2的价格101.0（因为b2先于s1报入？不，b2报入时s1还没报。
# 这里逻辑：s1进来，看到best_bid是b2(101)。100.5 = 101，成交。
# 价格：b2.create_time s1.create_time，所以价格是b2的101.0
self.assertEqual(self.matcher.trades[0][price], 101.0)
if __name__ == __main__:
unittest.main()
运行结果：
--- 开始测试 600519 撮合 ---
Buy1: OrderStatus.OPEN, Filled: 0
Sell1: OrderStatus.OPEN, Filled: 0
Buy2: OrderStatus.FILLED, Filled: 30
Sell1 Update: OrderStatus.PARTIALLY_FILLED, Filled: 30
Sell2: OrderStatus.FILLED, Filled: 20
Buy1 Update: OrderStatus.PARTIALLY_FILLED, Filled: 20
--- 成交记录 ---
{'price': 1682.0, 'quantity': 30, 'buy_id': '...', 'sell_id': '...'}
{'price': 1680.0, 'quantity': 20, 'buy_id': '...', 'sell_id': '...'}
看到没？Buy2 以 1682 成交，Sell2 以 1680 成交。
这就是“价格优先”的体现。 如果代码写错，价格取平均或者取新订单价格，面试直接挂。
优化扩展与避坑
代码能跑，不代表能上生产。
1. 并发安全
上面的代码是单线程的。实际交易是高并发。
问题：多线程同时修改 OrderBook，堆结构会乱。
方案：
Python：threading.Lock 保护 add_order 和 match。或者用 asyncio 单线程事件循环，避免锁开销。
Go/Rust：用 Mutex 或 RwLock。更高级的是无锁队列（Lock-Free Queue）。
面试高频问法：
“如何保证高并发下订单簿的线程安全？”
标准答案：
“采用单线程事件循环模型（如Go的Goroutine + Channel，或Python的Asyncio）。所有订单操作通过Channel发送到主协程，主协程串行处理撮合。这样天然避免锁竞争，且吞吐量极高。”
2. 持久化与对账
内存数据重启就没了。
方案：
订单状态：每次状态变更，异步写入Redis（缓存）和MySQL（持久化）。
成交记录：实时写入消息队列（Kafka），由下游服务落库。
避坑：
别在撮合主循环里同步写数据库！
一定要异步。 主循环只负责内存计算，耗时操作全部扔到后台队列。
3. 风控前置
代码里的 _risk_check 是同步的。
实际中，风控应该前置。
架构：订单进来 - 风控网关（查资金、查持仓、查频率） - 风控通过 - 撮合引擎。
目的：保护撮合引擎。如果非法订单都进撮合引擎，内存爆了怎么办？
4. 日志与监控
交易软件，日志就是生命线。
记录：订单ID、用户ID、价格、数量、状态变更、耗时。
格式：JSON结构化，方便ELK检索。
监控：监控订单延迟（从接收到成交回报的时间）、队列积压长度。
小结与互动
这套代码，从模型定义到撮合逻辑，完整覆盖了“新时代证券交易软件”的核心骨架。
你掌握了什么？
订单状态机：防止非法流转。
订单簿数据结构：价格优先、时间优先的实现细节。
撮合逻辑：部分成交、全部成交、价格确定规则。
架构思想：单线程事件循环、风控前置、异步持久化。
这些知识点，覆盖了高频面试题中关于交易系统设计的90%。
剩下的10%是分布式事务、容灾切换，那是高级岗位的事。
最后问一句：
这个知识点你面试被问过吗？
特别是“价格优先、时间优先”的具体实现，以及“如何保证撮合引擎的线程安全”。
留言说说，你当时怎么答的？或者你踩过什么坑？咱们评论区见。