A股量化交易系统开发：从架构设计到高频交易实战

1次阅读

共计 3626 个字符，预计需要花费 10 分钟才能阅读完成。

开发 A 股量化交易系统时，工程师常遇到三个核心挑战：

交易所协议解析复杂度 ：上交所 FAST 协议与深交所 STEP 协议的二进制结构差异，需要处理动态字段和嵌套组结构。例如委托簿消息中 Price 字段可能以整数或浮点形式出现，需按行情类别动态解析。
订单执行延迟敏感度 ：实测显示，在涨停板抢单场景中，10ms 的延迟会导致成交率下降 35%。系统需保证从信号生成到报单的全链路延迟稳定在 3ms 以内。
Tick 级数据吞吐量 ：早盘集合竞价时段，单个证券的 Tick 峰值可达 8000+/ 秒，全市场订阅时网络带宽可能突破 1Gbps。

C++：实测订单处理延迟最低（0.7μs/ 单），但开发周期长。关键代码需考虑内存对齐，例如：

#pragma pack(push, 1)
struct Order {
    uint64_t order_id;
    int32_t  price; // 价格单位 0.01 元
    uint32_t quantity;
    char     side; // 'B' 或 'S'
};
#pragma pack(pop)

Rust：无 GC 且线程安全，实测延迟 1.2μs/ 单。但 borrow checker 增加协议解析代码复杂度，需大量使用 Arc<Mutex<T>>。

Python：配合 Cython 优化后延迟可达 5μs/ 单。推荐方案：

# cython: boundscheck=False, wraparound=False
cdef class OrderGateway:
    cdef dict _order_map  # 使用 C 级 dict 加速查询
    cdef void submit(self, Order order) nogil:
        # 无 GIL 锁的订单提交
        ...

指标	Kafka	Pulsar
99% 延迟	2.1ms	1.7ms
吞吐量	550MB/s	620MB/s
持久化保证	异步刷盘	同步多副本

测试环境：20 生产者 /50 消费者，消息体 200 字节，上海同机房部署

基于 asyncio 的 TCP 流重组方案：

class FASTProtocol:
    def __init__(self):
        self._buffer = bytearray()
        self._pmap = {}  # 模板缓存

    async def feed_data(self, data: bytes):
        self._buffer.extend(data)
        while len(self._buffer) >= 2:  # 至少读取消息头
            msg_len = int.from_bytes(self._buffer[:2], 'big')
            if len(self._buffer) < msg_len + 2:
                break
            raw_msg = self._buffer[2:2+msg_len]
            await self._parse_message(raw_msg)
            self._buffer = self._buffer[2+msg_len:]

    async def _parse_message(self, raw: bytes):
        template_id = raw[0]
        if template_id not in self._pmap:
            await self._load_template(template_id)
        ...  # 按模板解析字段

采用分层锁策略减少竞争：

from threading import Lock

class OrderBook:
    def __init__(self):
        self._price_lock = Lock()  # 价格树大锁
        self._level_locks = {}     # 价格档位细粒度锁

    def update(self, price: int, is_bid: bool):
        # 先获取价格档位锁
        with self._get_level_lock(price):
            # 再短暂获取大锁更新树结构
            with self._price_lock:
                ...  # 更新 RB 树

    def _get_level_lock(self, price: int) -> Lock:
        if price not in self._level_locks:
            with self._price_lock:  # 双检锁模式
                if price not in self._level_locks:
                    self._level_locks[price] = Lock()
        return self._level_locks[price]

滑动窗口计数器实现：

from collections import deque

class SlidingWindow:
    def __init__(self, window_sec: int, max_count: int):
        self.window = window_sec * 1e9  # 转为纳秒
        self.max = max_count
        self.events = deque()

    def check(self, timestamp: int) -> bool:
        # 清理过期事件
        while self.events and timestamp - self.events[0] > self.window:
            self.events.popleft()

        if len(self.events) >= self.max:
            return False
        self.events.append(timestamp)
        return True

百分位	行情→策略 (μs)	策略→报单 (μs)
50%	42	28
90%	55	39
99%	210	85
99.9%	1500	300

测试条件：沪深 300 成分股 Tick 数据，策略为简单均线交叉

采用线程池 + 任务窃取方案：

import concurrent.futures
from multiprocessing import cpu_count

class OrderExecutor:
    def __init__(self):
        self._workers = cpu_count()
        self._pools = [
            concurrent.futures.ThreadPoolExecutor(1, 
                thread_name_prefix=f'OrderWorker-{i}')
            for i in range(self._workers)
        ]

    def submit_order(self, order: Order):
        # 按股票代码哈希选择线程池
        idx = hash(order.symbol) % self._workers
        future = self._pools[idx].submit(self._real_submit, order)
        ...

撤单比例限制 （上交所技术细则 5.3.2 条）：
动态监控撤单 / 成交比，超过 1:1 时自动降频
采用指数退避重试，初始间隔 500ms
委托频率限制 （深交所技术指引第 17 条）：
单个账户每秒不超过 300 笔

在网关层实现令牌桶算法

from time import monotonic

class RateLimiter:
    def __init__(self, rate: int):
        self._per_sec = rate
        self._allowance = rate
        self._last_check = monotonic()

    def acquire(self) -> bool:
        now = monotonic()
        elapsed = now - self._last_check
        self._last_check = now
        self._allowance += elapsed * self._per_sec
        if self._allowance > self._per_sec:
            self._allowance = self._per_sec
        if self._allowance < 1:
            return False
        self._allowance -= 1
        return True

典型 case 分析：

滑点模型缺失 ：回测假设立即成交，实盘需考虑盘口变化。建议在回测中引入：

def apply_slippage(price: float, is_buy: bool) -> float:
    # 买方向上加 3 档，卖方向下减 2 档
    tick_size = 0.01  # 根据证券类型动态获取
    offset = 3 if is_buy else -2
    return round(price + offset * tick_size, 2)