4、做市策略基础框架：策略循环架构

做市策略听起来高大上，其实核心就四个字：低买高卖。但怎么在毫秒级的市场里做到这一点？你需要一个靠谱的框架。

我个人习惯把做市策略拆成四个环节：报价 → 成交 → 风控 → 调整。这四个环节循环往复，构成了策略的“心跳”。

4.1 策略循环架构：报价-成交-风控-调整

先看一张图，这是我做策略时必画的框架。

说白了，这就是一个闭环。你报出价格，市场成交了，你检查一下风险，然后调整报价。周而复始。

报价阶段：根据当前持仓、订单簿深度、波动率等数据，计算出买卖报价。我在项目中遇到过一个问题——报价频率太高，导致交易所限流。后来我加了个最小间隔控制，才稳定下来。

成交阶段：处理交易所返回的成交信息。这里有个坑：成交回报可能延迟，也可能乱序。我曾经因为没处理好成交顺序，导致持仓计算错误，亏了一笔。

风控阶段：检查净头寸、最大订单数、资金使用率等。嗯，这里要注意，风控必须独立于策略逻辑，不能因为策略出bug就把风控带崩了。

调整阶段：根据成交情况和风控结果，更新策略参数。比如库存多了，就调低卖价、调高买价，加速去库存。

4.2 事件驱动模型

做市策略本质上是一个事件驱动的系统。你想想看，市场每分钟都在产生海量数据——新订单、成交、撤单、行情更新……你不能用轮询的方式去处理，那样CPU会炸。

事件驱动模型的核心思想是：有事件来了才处理，没事件就休息。

我常用的架构是这样的：

class EventEngine:
    def __init__(self):
        self.handlers = {}  # 事件类型 -> 处理函数列表

    def register(self, event_type, handler):
        if event_type not in self.handlers:
            self.handlers[event_type] = []
        self.handlers[event_type].append(handler)

    def emit(self, event):
        event_type = event['type']
        if event_type in self.handlers:
            for handler in self.handlers[event_type]:
                handler(event)

# 使用示例
engine = EventEngine()
engine.register('tick', on_tick)      # 行情更新
engine.register('trade', on_trade)    # 成交回报
engine.register('order', on_order)    # 订单状态变化

为什么用事件驱动？因为做市策略需要同时处理多个数据源。行情推送、成交回报、风控信号，这些是异步到达的。用事件模型，你可以把不同来源的数据统一到一个处理管道里。

4.3 定时任务与异步处理

事件驱动解决了“被动响应”的问题，但做市策略还需要“主动触发”的能力。比如：

• 每隔100ms重新计算一次报价
• 每隔1分钟检查一次持仓风险
• 每隔5分钟向交易所发送一次心跳

这些就需要定时任务。我个人习惯用 asyncio 来实现，因为Python的异步框架对这类场景支持得很好。

import asyncio

class MarketMaker:
    def __init__(self):
        self.running = True

    async def quote_loop(self):
        """报价循环：每100ms执行一次"""
        while self.running:
            self.calculate_quotes()
            await asyncio.sleep(0.1)  # 100ms

    async def risk_check_loop(self):
        """风控检查：每1秒执行一次"""
        while self.running:
            self.check_risk()
            await asyncio.sleep(1.0)

    async def heartbeat_loop(self):
        """心跳：每5秒执行一次"""
        while self.running:
            self.send_heartbeat()
            await asyncio.sleep(5.0)

    async def run(self):
        """启动所有定时任务"""
        tasks = [
            self.quote_loop(),
            self.risk_check_loop(),
            self.heartbeat_loop()
        ]
        await asyncio.gather(*tasks)

# 启动
mm = MarketMaker()
asyncio.run(mm.run())

异步处理还有一个好处：不会阻塞。比如你在处理成交回报时，行情更新还能继续进来。如果用同步方式，一个慢操作就会卡住整个系统。

4.4 三者如何配合

事件驱动、定时任务、异步处理，这三者不是孤立的。在实际系统中，它们是这样配合的：

组件	职责	触发方式	示例
事件驱动	处理外部输入	被动（市场推送）	行情更新、成交回报
定时任务	执行周期性操作	主动（时间触发）	报价计算、风控检查
异步处理	管理并发任务	混合	同时运行多个循环

举个例子：行情推送来了（事件驱动），触发报价计算函数。但这个函数本身是在一个定时任务里跑的（每100ms一次）。而报价计算需要查询数据库，这个查询是异步的，不会阻塞其他任务。

我记得有一次，我忘了加定时任务，全靠事件驱动来触发报价。结果市场平静的时候，半天没有新事件，策略就“睡着”了。等行情突然爆发，报价还是几分钟前的，直接亏穿。从那以后，我每个策略都至少保留一个定时任务作为“心跳”。

好了，这一章的内容就这些。框架搭好了，后面才能往里填具体的策略逻辑。别急，一步一步来。

---

公众号:蓝海资料掘金营，微信deep3321