python案例如何结合盘口做出最终判断？

wen python案例 2026-11-14 12

Python 结合盘口数据做最终判断的实战案例

说明：以下内容为技术方法演示，不构成任何投资建议，盘口数据噪音大、易被操纵，实盘决策需谨慎。

盘口数据的核心字段

以股票 Level-2 / 加密货币订单簿为例，通常包含：

字段	含义
bid_price[i], bid_volume[i]	买 i 档价格与挂单量
ask_price[i], ask_volume[i]	卖 i 档价格与挂单量
last_price, volume	最新成交价与成交量
timestamp	时间戳

盘口特征工程（Python 实现）

import numpy as np
import pandas as pd
def orderbook_features(ob: dict) -> dict:
    """
    ob = {
        'bids': [(price, vol), ...] 降序,
        'asks': [(price, vol), ...] 升序
    }
    """
    bids = np.array(ob['bids'], dtype=float)
    asks = np.array(ob['asks'], dtype=float)
    bid_p, bid_v = bids[:, 0], bids[:, 1]
    ask_p, ask_v = asks[:, 0], asks[:, 1]
    best_bid, best_ask = bid_p[0], ask_p[0]
    mid = (best_bid + best_ask) / 2
    spread = best_ask - best_bid
    # 1) 盘口不平衡度 OBI (Order Book Imbalance)
    obi = (bid_v.sum() - ask_v.sum()) / (bid_v.sum() + ask_v.sum() + 1e-9)
    # 2) 加权不平衡（按价格距离衰减，越近权重越大）
    w = np.exp(-np.abs(np.arange(len(bid_v))))  # 简化衰减
    obi_w = (bid_v @ w - ask_v @ w) / (bid_v @ w + ask_v @ w + 1e-9)
    # 3) 买卖价差率
    spread_ratio = spread / mid
    # 4) 盘口厚度（前5档）
    depth_bid = bid_v[:5].sum()
    depth_ask = ask_v[:5].sum()
    depth_ratio = depth_bid / (depth_ask + 1e-9)
    # 5) 微观价格 (Microprice)，比中间价更能反映压力
    microprice = (best_bid * ask_v[0] + best_ask * bid_v[0]) / (bid_v[0] + ask_v[0] + 1e-9)
    # 6) 大单墙检测（某档量 > 均值 * k）
    k = 3
    wall_bid = bid_v.max() > k * bid_v.mean()
    wall_ask = ask_v.max() > k * ask_v.mean()
    return {
        'mid': mid, 'spread': spread, 'spread_ratio': spread_ratio,
        'obi': obi, 'obi_weighted': obi_w, 'depth_ratio': depth_ratio,
        'microprice': microprice, 'microprice_bias': (microprice - mid) / mid,
        'wall_bid': wall_bid, 'wall_ask': wall_ask,
    }

多信号融合决策框架

单纯看 OBI 容易被打脸（诱多/诱空），应结合 动量、成交量、价差做加权投票：

def final_decision(feats: dict, recent_returns: list, vol_zscore: float) -> str:
    score = 0.0
    # ① 盘口不平衡方向（近端权重更高）
    score += 0.35 * np.tanh(feats['obi_weighted'] * 5)
    # ② 微观价格偏置
    score += 0.15 * np.tanh(feats['microprice_bias'] * 10000)
    # ③ 短期动量
    mom = np.mean(recent_returns[-10:]) if recent_returns else 0
    score += 0.25 * np.tanh(mom * 200)
    # ④ 深度比例
    score += 0.15 * np.tanh((feats['depth_ratio'] - 1) * 2)
    # ⑤ 风险惩罚：价差过宽 → 不确定，拉回中性
    if feats['spread_ratio'] > 0.002:   # 阈值按品种调整
        score *= 0.5
    # ⑥ 假墙过滤：若大单墙在卖方且 OBI 偏多 → 可能诱多
    if feats['wall_ask'] and score > 0:
        score *= 0.6
    if feats['wall_bid'] and score < 0:
        score *= 0.6
    # ⑦ 放量确认
    if vol_zscore > 2 and abs(score) > 0.2:
        score *= 1.2
    # 最终阈值判定（三档：偏多 / 中性 / 偏空）
    if score > 0.25:
        return f"偏多 (score={score:.2f})"
    elif score < -0.25:
        return f"偏空 (score={score:.2f})"
    else:
        return f"中性/观望 (score={score:.2f})"

一个完整调用示例

# 模拟一帧盘口
ob = {
    'bids': [(100.00, 50), (99.99, 30), (99.98, 20), (99.97, 15), (99.96, 10)],
    'asks': [(100.02, 20), (100.03, 15), (100.04, 10), (100.05, 8),  (100.06, 5)],
}
feats = orderbook_features(ob)
recent = [0.0002, -0.0001, 0.0003, 0.0005]     # tick 收益率
vol_z = 2.5                                     # 成交量 z-score
result = final_decision(feats, recent, vol_z)
print(feats)
print(result)

输出会给出偏多/偏空/观望，并附带可解释的分数。

实盘落地的几个关键细节

时序化处理：单帧盘口噪音极大，建议对 OBI 做 EMA 平滑（如 obi_ema = 0.2*obi + 0.8*obi_ema），或取最近 N 帧平均。
样本回测：盘口因子必须做前视偏差防范——用 t 时刻盘口预测 t+1~t+k 收益，且 tick 数据时间戳要严格对齐。
阈值需要按品种/时段校准：A股集合竞价、加密货币亚洲时段流动性差异巨大，固定阈值会失效。
诱单识别：大单墙常在几帧内撤单（spoofing），可用"挂单存活时长"作为额外特征。
组合使用：盘口是微观信号，应与日线/小时线级别的趋势、波动率 regime 结合——趋势向上时 OBI 偏多信号可信度更高，震荡市则降低权重。