1. 基于订单簿不平衡的做市价差动态调整

任务目标

实现一个订单簿不平衡驱动的做市策略核心逻辑：根据订单簿买卖盘深度不平衡度，动态调整买卖报价相对中间价的偏移（价差），在提供流动性的同时控制库存风险。策略输入为订单簿快照（买一/卖一价格与数量），输出为调整后的买价与卖价。

核心逻辑

不平衡度定义为 imbalance = bidVol / (bidVol + askVol)，取值范围 [0,1]。当买单深度显著大于卖单时（imbalance 接近 1），说明卖方流动性不足，做市商应提高卖价（减少卖出风险敞口），同时可适当降低买价以吸引成交、积累库存；反之亦然。价差偏移量由不平衡度与目标库存共同决定。

完整实现

import math

class OrderBookSnapshot:
    def __init__(self, bid_price, bid_vol, ask_price, ask_vol):
        self.bid_price = bid_price
        self.bid_vol = bid_vol
        self.ask_price = ask_price
        self.ask_vol = ask_vol

class MarketMakingStrategy:
    def __init__(self, spread_base=0.01, max_offset=0.005, target_inventory=0.0,
                 inventory_penalty=0.001, position=0.0):
        self.spread_base = spread_base          # 基础价差（价格）
        self.max_offset = max_offset            # 最大单边偏移量
        self.target_inventory = target_inventory  # 目标库存（正=多头）
        self.inventory_penalty = inventory_penalty  # 库存偏离惩罚系数
        self.position = position                # 当前净库存（正=净多头）

    def compute_imbalance(self, book: OrderBookSnapshot) -> float:
        total_vol = book.bid_vol + book.ask_vol
        if total_vol == 0:
            return 0.5
        return book.bid_vol / total_vol

    def inventory_offset(self) -> float:
        """库存偏移：库存偏离目标越远，报价越偏向降低库存方向"""
        return (self.target_inventory - self.position) * self.inventory_penalty

    def calculate_quotes(self, book: OrderBookSnapshot):
        mid_price = (book.bid_price + book.ask_price) / 2.0
        imbalance = self.compute_imbalance(book)

        # 不平衡偏移：买单多时提高卖价，降低买价（偏移方向相反）
        imbalance_offset = (imbalance - 0.5) * 2 * self.max_offset

        # 总偏移 = 不平衡偏移 + 库存偏移
        total_offset = imbalance_offset + self.inventory_offset()
        total_offset = max(-self.max_offset, min(self.max_offset, total_offset))

        half_spread = self.spread_base / 2.0
        bid_price = mid_price - half_spread + total_offset
        ask_price = mid_price + half_spread + total_offset

        return round(bid_price, 4), round(ask_price, 4)

    def update_position(self, trade_price: float, trade_qty: float, side: str):
        """side: 'buy' 表示我们买入（库存增加），'sell' 表示我们卖出（库存减少）"""
        if side == 'buy':
            self.position += trade_qty
        elif side == 'sell':
            self.position -= trade_qty

运行与验证

# 验证脚本
if __name__ == "__main__":
    # 场景1：买卖深度均衡，无库存
    book1 = OrderBookSnapshot(100.0, 10, 100.1, 10)
    mm = MarketMakingStrategy()
    bid1, ask1 = mm.calculate_quotes(book1)
    print(f"均衡场景: bid={bid1}, ask={ask1}")
    # 预期: 价差中心=100.05, 偏移≈0, bid≈100.00, ask≈100.10

    # 场景2：买单深度远大于卖单（imbalance≈0.83）
    book2 = OrderBookSnapshot(100.0, 50, 100.1, 10)
    bid2, ask2 = mm.calculate_quotes(book2)
    print(f"买单倾斜场景: bid={bid2}, ask={ask2}")
    # 预期: 偏移为正（约+0.0033），bid≈100.003, ask≈100.103

    # 场景3：库存为净多头，应降低报价以卖出减库存
    mm.position = 20  # 净多头20
    bid3, ask3 = mm.calculate_quotes(book1)
    print(f"净多头库存场景: bid={bid3}, ask={ask3}")
    # 预期: 偏移为负（约-0.02被截断到-0.005），bid≈99.995, ask≈100.095

    # 场景4：更新库存后重新计算
    mm.update_position(100.05, 5, 'sell')  # 卖出5，库存从20降到15
    bid4, ask4 = mm.calculate_quotes(book1)
    print(f"减仓后: bid={bid4}, ask={ask4}")

验证方法：运行上述脚本，观察输出是否符合以下判定条件：

• 场景1中买卖价差约为 spread_base（0.01），且买卖价围绕中间价对称分布；
• 场景2中卖价相对场景1上移（因买单深度大，卖方流动性不足，提高卖价），买价相对下移，价差扩大；
• 场景3中买卖报价整体下移（偏移为负），使报价更具吸引力以促成卖出、降低净多头库存；
• 场景4中库存减少后，负向偏移幅度较场景3收窄。

若输出满足上述逻辑关系，则策略核心实现正确。实际部署时需将 calculate_quotes 接入交易所订单簿推送流，并在每次成交后调用 update_position 更新库存。