4. 库存状态监控：实时库存计算、库存变化率、库存阈值设定

做市商最怕什么？怕库存失控。

我见过太多新手，策略跑得挺漂亮，但库存监控没做好，一夜回到解放前。说白了，库存就是你的命根子。你想想看，一个做市商手里拿着大量现货，价格突然暴跌，那画面太美我不敢看。

这一章，我们就来聊聊怎么把库存盯死、算准、控住。

4.1 实时库存计算：别让数据骗了你

实时库存计算，听起来简单？不就是「初始库存 + 买入 - 卖出」吗？

嗯，理论上是这样。但实际项目中，坑多得很。

我记得有一次，交易所的成交回报延迟了200毫秒，我的库存计算器直接算错了，导致重复下单。那次教训让我明白：库存计算必须考虑「已提交未成交」的订单。

来看一个我常用的实时库存计算模型：

class InventoryTracker:
    def __init__(self, initial_inventory: dict):
        # 实际持仓（已成交）
        self.actual = initial_inventory.copy()
        # 挂单中的买卖量
        self.pending_buy = defaultdict(float)
        self.pending_sell = defaultdict(float)
        # 时间戳
        self.last_update = time.time()
    
    def on_order_submitted(self, symbol: str, side: str, qty: float):
        """订单提交时，先记入挂单"""
        if side == 'buy':
            self.pending_buy[symbol] += qty
        else:
            self.pending_sell[symbol] += qty
    
    def on_trade(self, symbol: str, side: str, qty: float):
        """成交时，更新实际持仓，并扣除挂单"""
        if side == 'buy':
            self.actual[symbol] += qty
            self.pending_buy[symbol] -= qty
        else:
            self.actual[symbol] -= qty
            self.pending_sell[symbol] -= qty
    
    def get_effective_inventory(self, symbol: str) -> float:
        """有效库存 = 实际持仓 + 挂单买入 - 挂单卖出"""
        return (self.actual[symbol] 
                + self.pending_buy[symbol] 
                - self.pending_sell[symbol])

4.2 库存变化率：发现异常的第一道防线

库存变化率，说白了就是「库存变动的速度」。为什么重要？

你想想看，如果库存突然加速变化，说明什么？要么市场在剧烈波动，要么你的策略出了问题。

我个人习惯用滑动窗口法来计算变化率。窗口大小选多少？这得看你的交易频率。

• 高频做市：窗口选 1-5 秒
• 中频做市：窗口选 30-60 秒
• 低频做市：窗口选 5-15 分钟

来看代码实现：

class InventoryChangeRate:
    def __init__(self, window_seconds: int = 60):
        self.window = window_seconds
        self.snapshots = []  # (timestamp, inventory)
    
    def add_snapshot(self, inventory: float):
        now = time.time()
        self.snapshots.append((now, inventory))
        # 清理过期数据
        cutoff = now - self.window
        self.snapshots = [(t, inv) for t, inv in self.snapshots 
                         if t >= cutoff]
    
    def get_change_rate(self) -> float:
        """返回每秒库存变化率（正数表示增加，负数表示减少）"""
        if len(self.snapshots) < 2:
            return 0.0
        
        first = self.snapshots[0]
        last = self.snapshots[-1]
        time_diff = last[0] - first[0]
        
        if time_diff < 0.001:
            return 0.0
        
        return (last[1] - first[1]) / time_diff

4.3 库存阈值设定：给库存上「紧箍咒」

库存阈值，就是给库存设个上下限。超过上限，你得主动卖出；低于下限，你得主动买入。

但阈值怎么设？拍脑袋？不行。

我一般用动态阈值，基于以下几个因素：

因素	说明	典型值
账户总资产	库存不能超过总资产的一定比例	20%-50%
日均成交量	库存不能超过日均成交量的倍数	5-20倍
波动率	波动越大，阈值越紧	动态调整
流动性	流动性越差，阈值越紧	动态调整

来看一个动态阈值计算器：

class DynamicThreshold:
    def __init__(self, base_ratio: float = 0.3):
        self.base_ratio = base_ratio  # 基础库存比例
    
    def calculate_threshold(self, 
                           total_asset: float,
                           avg_daily_volume: float,
                           volatility: float,
                           liquidity_score: float) -> dict:
        """计算动态阈值"""
        # 基于总资产的阈值
        asset_limit = total_asset * self.base_ratio
        
        # 基于成交量的阈值
        volume_limit = avg_daily_volume * 0.1  # 10%的日均成交量
        
        # 波动率调整因子（波动越大，阈值越小）
        vol_factor = max(0.5, 1.0 - volatility * 10)
        
        # 流动性调整因子（流动性越差，阈值越小）
        liq_factor = min(1.0, liquidity_score)
        
        # 综合阈值
        combined_limit = min(asset_limit, volume_limit) * vol_factor * liq_factor
        
        return {
            'upper': combined_limit,
            'lower': -combined_limit,
            'warning_upper': combined_limit * 0.8,
            'warning_lower': -combined_limit * 0.8
        }

4.4 实战中的库存监控架构

好了，把上面三个模块串起来，就是一个完整的库存监控系统。

我习惯用三层架构：

1. 数据层：实时采集成交、挂单、行情数据
2. 计算层：实时库存、变化率、阈值计算
3. 告警层：超阈值触发告警，自动执行风控动作

来看一个完整的监控循环：

class InventoryMonitor:
    def __init__(self, initial_inventory: dict):
        self.tracker = InventoryTracker(initial_inventory)
        self.rate_calc = InventoryChangeRate(window_seconds=60)
        self.threshold = DynamicThreshold(base_ratio=0.3)
        
        # 告警回调
        self.on_warning = None
        self.on_critical = None
    
    def update(self, 
               symbol: str,
               trade_qty: float,
               trade_side: str,
               total_asset: float,
               avg_volume: float,
               volatility: float,
               liquidity: float):
        """每次成交或行情更新时调用"""
        # 1. 更新库存
        self.tracker.on_trade(symbol, trade_side, trade_qty)
        
        # 2. 计算有效库存
        effective = self.tracker.get_effective_inventory(symbol)
        
        # 3. 记录库存快照
        self.rate_calc.add_snapshot(effective)
        
        # 4. 计算变化率
        rate = self.rate_calc.get_change_rate()
        
        # 5. 计算阈值
        limits = self.threshold.calculate_threshold(
            total_asset, avg_volume, volatility, liquidity
        )
        
        # 6. 检查是否超阈值
        if effective > limits['upper']:
            self._trigger_alert('critical', f'库存超上限: {effective}')
        elif effective > limits['warning_upper']:
            self._trigger_alert('warning', f'库存接近上限: {effective}')
        elif effective < limits['lower']:
            self._trigger_alert('critical', f'库存超下限: {effective}')
        elif effective < limits['warning_lower']:
            self._trigger_alert('warning', f'库存接近下限: {effective}')
    
    def _trigger_alert(self, level: str, msg: str):
        """触发告警"""
        if level == 'critical' and self.on_critical:
            self.on_critical(msg)
        elif level == 'warning' and self.on_warning:
            self.on_warning(msg)

嗯，这一章就到这里。下一章我们聊聊「库存对冲策略」，那才是真正考验功力的地方。