PTrade 策略风控：如何监控全市场成交额与流动性？

在量化交易中，监控市场流动性对于评估交易成本、滑点风险以及市场情绪至关重要。在 PTrade 平台中，我们可以结合**实时行情快照（Snapshot）和历史行情数据（History）**来构建多维度的流动性监控指标。

以下是监控市场流动性变化的几个核心维度及具体的 PTrade 实现方案。

1. 核心流动性指标

在 PTrade 中，我们通常通过以下四个维度来量化流动性：

1. 买卖价差 (Bid-Ask Spread):

   • 定义: 卖一价（Ask 1）与买一价（Bid 1）的差值比率。
   • 意义: 价差越小，流动性越好；价差突然扩大通常意味着流动性枯竭或市场剧烈波动。
   • API: get_snapshot (仅交易模式/模拟盘可用)。

2. 盘口深度 (Market Depth):

   • 定义: 买一和卖一档位的挂单量总和。
   • 意义: 挂单量越大，市场承受大额交易而不引起价格剧烈波动的能力越强。
   • API: get_snapshot。

3. 换手率与量比 (Turnover & Volume Ratio):

   • 定义: 换手率反映交易活跃度，量比反映当前成交量与过去5天平均成交量的关系。
   • 意义: 高换手和高量比通常代表流动性充裕。
   • API: get_snapshot 直接提供 turnover_ratio 和 vol_ratio。

4. Amihud 非流动性指标 (Amihud Illiquidity Ratio):

   • 定义: 绝对收益率与成交额的比值 (|Return| / Dollar Volume)。
   • 意义: 衡量单位成交金额带来的价格冲击。数值越大，流动性越差（价格越容易被少量资金推动）。
   • API: get_history (回测和交易模式均可用)。

2. PTrade 策略代码实现

以下代码展示了如何在一个策略中同时监控上述指标。

注意：

• get_snapshot 函数仅在交易模式（实盘或模拟盘）下有效，回测模式下无法获取盘口挂单数据。
• 回测模式下，建议主要使用 get_history 计算 Amihud 指标和成交量变化。

import numpy as np

def initialize(context):
    # 设置监控标的，以恒生电子为例
    g.security = '600570.SS'
    set_universe(g.security)
    
    # 设置Amihud指标计算的窗口期（如过去5天）
    g.amihud_window = 5
    
    # 设定流动性预警阈值（示例）
    g.spread_threshold = 0.005  # 价差超过0.5%预警
    g.depth_threshold = 100000  # 盘口资金少于10万预警

def handle_data(context, data):
    """
    该函数在回测中根据频率运行，在交易中根据频率运行
    此处演示分钟级别或日线级别的监控
    """
    monitor_realtime_liquidity(context)
    monitor_historical_liquidity(context)

def monitor_realtime_liquidity(context):
    """
    监控实时流动性：价差、盘口深度、量比
    注意：get_snapshot 仅在交易/模拟盘环境有效
    """
    # 判断是否在回测模式，回测模式不支持 snapshot
    if not is_trade():
        return

    # 获取行情快照
    snapshot = get_snapshot(g.security)
    if not snapshot:
        return
        
    tick_data = snapshot.get(g.security)
    if not tick_data:
        return

    # 1. 获取买一、卖一价格和数量
    # offer_grp/bid_grp 结构: {1: [价格, 数量, 笔数], ...}
    try:
        ask_1_price = tick_data['offer_grp'][1][0]
        ask_1_vol = tick_data['offer_grp'][1][1]
        
        bid_1_price = tick_data['bid_grp'][1][0]
        bid_1_vol = tick_data['bid_grp'][1][1]
        
        current_price = tick_data['last_px']
    except:
        # 处理涨跌停导致没有卖一或买一的情况
        log.info("当前可能处于涨跌停状态，无法计算完整价差")
        return

    # 2. 计算相对买卖价差 (Spread)
    # (卖一 - 买一) / 中间价
    if ask_1_price > 0 and bid_1_price > 0:
        mid_price = (ask_1_price + bid_1_price) / 2
        spread_ratio = (ask_1_price - bid_1_price) / mid_price
        
        if spread_ratio > g.spread_threshold:
            log.warning("【流动性预警】%s 价差扩大: %.4f%%" % (g.security, spread_ratio * 100))
        else:
            log.info("当前价差: %.4f%%" % (spread_ratio * 100))

    # 3. 计算盘口深度 (Market Depth - Level 1)
    # 计算买一卖一档位的总金额估算
    depth_value = (ask_1_price * ask_1_vol) + (bid_1_price * bid_1_vol)
    if depth_value < g.depth_threshold:
        log.warning("【流动性预警】%s 盘口深度不足: %.2f 元" % (g.security, depth_value))

    # 4. 获取量比和换手率
    vol_ratio = tick_data.get('vol_ratio', 0)
    turnover_rate = tick_data.get('turnover_ratio', 0)
    
    log.info("量比: %s, 换手率: %s%%" % (vol_ratio, turnover_rate))

def monitor_historical_liquidity(context):
    """
    监控基于历史数据的流动性：Amihud非流动性指标
    适用于回测和交易环境
    """
    # 获取过去 N 天的历史数据
    # 需要获取收盘价(close)和成交额(money)
    hist = get_history(g.amihud_window + 1, '1d', ['close', 'money'], g.security, fq=None, include=False)
    
    if len(hist) < g.amihud_window + 1:
        return

    # 提取数据
    closes = hist['close'].values
    moneys = hist['money'].values
    
    # 计算每日收益率 (绝对值)
    # pct_change = (今日收盘 - 昨日收盘) / 昨日收盘
    returns = np.diff(closes) / closes[:-1]
    abs_returns = np.abs(returns)
    
    # 对应的成交额 (取后N天)
    valid_moneys = moneys[1:]
    
    # 防止除以0
    valid_moneys[valid_moneys == 0] = 1.0 
    
    # 计算每日 Amihud 指标: |R| / Volume
    # 为了数值可读性，通常乘以 10^6 或其他系数
    daily_amihud = (abs_returns / valid_moneys) * 100000000
    
    # 计算平均 Amihud 指标
    avg_amihud = np.mean(daily_amihud)
    
    log.info("Amihud非流动性指标(5日均值): %.4f" % avg_amihud)
    
    # 简单的逻辑判断：如果非流动性指标突增，说明价格受资金冲击变大
    if avg_amihud > 0.5: # 阈值需根据具体股票池调整
        log.info("注意：当前标的非流动性较高，大额交易可能产生较大滑点")

3. 代码解析

1. monitor_realtime_liquidity (实时监控):

   • 使用 get_snapshot(g.security) 获取 L1 行情数据。
   • 价差计算: 提取 offer_grp (卖方) 和 bid_grp (买方) 的第一档价格。计算公式为 (Ask - Bid) / Mid。价差越大，交易成本越高，流动性越差。
   • 深度计算: 估算买一和卖一档位的资金总额。如果资金量过小，说明市场承接力弱。
   • 异常处理: 增加了 try-except 块，因为在涨停（无卖单）或跌停（无买单）时，offer_grp 或 bid_grp 可能为空或价格为0。

2. monitor_historical_liquidity (历史/趋势监控):

   • 使用 get_history 获取过去几天的日线数据。
   • Amihud 指标: 计算公式为 |收益率| / 成交额。这个指标衡量了“每交易一单位金额，价格变动了多少”。
   • 数值含义: Amihud 值越大，说明少量的资金就能把价格打上去或砸下来，意味着流动性越差。

4. 总结

• 高频/日内交易: 重点关注 monitor_realtime_liquidity 中的价差和盘口深度。如果价差突然扩大，应暂停交易或使用限价单（Limit Order）而非市价单。
• 低频/大资金交易: 重点关注 monitor_historical_liquidity 中的 Amihud 指标。如果该指标处于高位，建议分批下单（TWAP/VWAP 算法）以减少对市场的冲击。

Q&A

Q: 为什么 get_snapshot 在回测中返回的数据很多是 0？
A: PTrade 的回测机制主要基于 K 线数据（OHLCV）。get_snapshot 是获取当前时刻的 Level-1/Level-2 实时切片数据（如五档盘口），历史 K 线数据中并不包含这些盘口细节。因此，回测中无法模拟真实的盘口价差，该函数仅在实盘或模拟交易中有效。

Q: 如何在回测中模拟流动性不足导致的滑点？
A: 在 initialize 中使用 set_slippage (百分比滑点) 或 set_fixed_slippage (固定价差滑点)。例如 set_slippage(0.002) 表示双边各 0.1% 的滑点，以此来模拟流动性成本。

Q: Amihud 指标的阈值应该设为多少？
A: Amihud 指标没有绝对的标准阈值，它与股票的市值和板块有关。通常建议计算该股票过去一年的 Amihud 均值和标准差，当当前值超过 均值 + 2 * 标准差 时，视为流动性异常枯竭。