[源码解析]网格交易总亏钱?试着用Python复现Avellaneda-Stoikov做市模型

01.引言:从“几何图形”到“概率博弈”

在量化交易(Quantitative Trading)的入门阶段，网格策略(Grid Trading)几乎是每个开发者的必修课。写一个while True循环，基于固定的Price Gap挂出Buy/Sell Limit单。

但实盘教训往往很惨痛：静态网格在单边行情下极其脆弱。因为缺乏对库存(Inventory)的感知，程序会在下跌趋势中机械地“接飞刀”，直至耗尽流动性被套牢。

最近重读高频交易(HFT)经典文献Avellaneda-Stoikov(AS)模型，发现它的核心思想完全可以降维打击普通网格——将“死扛”转化为“库存博弈”。

本文将分享如何用Python复现AS模型的核心逻辑，并探讨在Python这种非实时系统下，如何通过高性能行情API来弥补速度短板。

02.理论核心:重新定义“中间价”

传统的网格策略是“几何对称”的：Ask = Mid + GapBid = Mid - Gap

AS模型提出了一个颠覆性的概念：保留价格(Reservation Price, r)。 它认为，交易员不应锚定市场价，而应锚定自己的“心理价”。

其核心计算公式如下(建议直接复制)：

r = s - q × γ × σ²

作为开发者，我们需要深入理解这四个变量的物理含义，以及工程实现上的妥协：

1. s (Mid Price): 市场共识

当前订单薄的买一卖一均价。[进阶注解]：在高频领域，通常会使用微观价格(Micro-price)或考虑订单流不平衡(OFI)来修正s。但作为入门复现，直接使用Mid Price是性价比最高的选择。

2. q (Inventory Factor): 库存压力——策略的灵魂

这是AS模型的重力参数。[工程避坑]：绝对不能直接用持仓数量(如10000)代入公式。必须归一化：q = (当前持仓 - 目标持仓) / 满仓限制

• q > 0(积压)：r < s。挂单整体下移，降价甩卖，拒接新货。

• q < 0(短缺)：r > s。挂单整体上移，溢价抢筹，惜售不卖。

3. γ (Risk Aversion): 风险厌恶系数

策略的性格参数。γ越大，策略越“怂”，稍微拿点货就拼命降价。

4. σ² (Volatility): 市场波动率

[工程妥协]：学术界通常使用GARCH模型或已实现波动率(Realized Volatility)。但在工程落地时，使用ATR或滚动窗口的标准差(Std)通常已经足够捕捉盘面风险。

03.源码实现:封装AS_Grid_Logic类

Talk is cheap, show me the code. 以下是基于Python的算法逻辑封装。为了降低理解门槛，我们对AS模型进行了工程化简化：保留了最核心的库存偏斜(Skew)，而将复杂的**价差宽度(Spread)**计算简化为ATR动态调整。

import numpy as npclass AS_Grid_Logic:    """    Avellaneda-Stoikov动态网格策略核心逻辑    """    def __init__(self, risk_gamma: float = 0.5, max_pos: int = 1000):        """        :param risk_gamma: 风险厌恶系数(Gamma), 值越大策略越倾向于去库存        :param max_pos: 最大持仓限制, 用于归一化计算        """        self.risk_gamma = risk_gamma        self.max_pos = max_pos    def calculate_skew(self, current_pos: int, volatility: float) -> float:        """        计算价格偏移量(Spread Shift)        """        # 防御性编程: 避免除零错误        if self.max_pos == 0:            return 0.0        # 1.关键步骤:归一化库存q (-1.0 ~ 1.0)        # 如果不归一化，公式中的线性惩罚项会直接溢出        q = current_pos / self.max_pos        # 2.核心公式: Shift = q * gamma * sigma^2        # 物理含义: 库存压力 * 怂的程度 * 市场风浪        shift = q * self.risk_gamma * (volatility ** 2)        return shift    def get_quotes(self, mid_price: float, current_pos: int,                    volatility: float, half_spread: float):        """        生成最终的Bid/Ask价格        :param mid_price: 当前市场中间价        :param current_pos: 当前持仓        :param volatility: 波动率(如ATR或std)        :param half_spread: 基础网格半宽(此处简化处理，未使用AS模型的k参数求解)        :return: (bid_price, ask_price, reservation_price)        """        # 计算偏移        shift = self.calculate_skew(current_pos, volatility)        # 计算保留价格(Reservation Price)        # 这一步将锚点从市场价s切换到了心理价r        reservation_price = mid_price - shift        # 生成围绕保留价格的网格        bid_price = reservation_price - half_spread        ask_price = reservation_price + half_spread        return bid_price, ask_price, reservation_price# --- 单元测试/模拟运行 ---if __name__ == "__main__":    # 初始化策略: 比较激进的去库存设定(Gamma=1.5)    logic = AS_Grid_Logic(risk_gamma=1.5, max_pos=1000)    # 模拟场景: 市场价100，满仓被套(pos=1000)，高波动(vol=2.0)    bid, ask, res_p = logic.get_quotes(        mid_price=100.0,         current_pos=1000,         volatility=2.0,         half_spread=0.5    )    print(f"Market Mid Price: 100.00")    print(f"Inventory Ratio (q): 1.0 (满仓焦虑状态)")    print(f"Reservation Price: {res_p:.2f} (心理中枢大幅下移)")    print(f"Algo Bid: {bid:.2f} (防止接盘)")    print(f"Algo Ask: {ask:.2f} (降价甩卖)")

运行结果解析：在满仓且高波动场景下，算法将Ask报价从理论的100.5压低到了94.5(示例值)。 这在代码层面实现了：“只要我有货且市场不稳，我就比谁跑得都快”。

04.工程挑战:Python跑做市是伪命题吗？

这也是很多资深交易员会质疑的点：“Python有GIL锁，延迟那么高，跑AS模型不是找死吗？”

你是对的，但也不全对。你跑不赢FPGA驱动的顶级HFT团队，但你的对手盘如果是散户，你只需要跑赢HTTP轮询即可。

在实盘落地时，我们需要解决两个工程瓶颈：

1.Maker策略的费率控制AS模型本质是提供流动性(Market Making)。

• 痛点：高频调整挂单容易导致在价格剧烈波动时误成交为Taker。

• 优化：不要每秒都重挂单。在代码中增加min_step_filter(最小变动过滤器)，只有当abs(new_price - old_price) > threshold时才发送Order Update请求。

2.逆向选择(Adverse Selection)与数据源延迟这是最致命的。当Python算出r需要下移时，通常是因为市场已经发生了Micro-crash。

• 瓶颈：如果你使用的是普通的RESTful行情API(轮询机制)，延迟通常在500ms~1000ms。等你的Cancel Order到达交易所，原本的Buy Limit早就被Toxic Flow击穿了。

• 解决方案：必须升级行情API的接入方式。

技术选型建议：TickDB对于Python开发者，如果不想花大量时间维护C++的底层连接，推荐使用TickDB这样的专业数据基础设施：

• WebSocket Stream：实盘必须使用WebSocket订阅全量Tick数据。TickDB提供的行情API可以实现毫秒级的Tick推送。

• Event-Driven(事件驱动)：将策略架构改为OnTick()事件驱动模式。TickDB的Python SDK能够很好地适配这种模式，确保策略在接收到最新Tick的瞬间完成计算和发单。

• Data Consistency：AS模型依赖准确的volatility计算。TickDB提供的清洗后的历史Tick数据，可以方便地预热计算ATR。

05.总结

从“死网格”进化到“AS动态网格”，本质是量化思维的升维：

1. 算法层：引入Inventory(q)和Volatility(σ)因子，使策略具备自我保护能力。

2. 工程层：承认Python的速度局限，通过接入高性能行情API(TickDB)，利用WebSocket低延迟特性构建护城河。

代码已开源在文中，欢迎各位开发者Copy测试。如有关于行情API对接或算法优化的疑问，欢迎在评论区技术交流。