第十六章：订单流与程序化交易

说实话，很多做手工交易的朋友一听到「程序化」三个字就头大。觉得那是量化团队才玩的东西，自己搞不定。

但我个人习惯是：能交给机器的，绝不手动干。尤其是订单流这种高频数据，你盯盘盯到眼睛出血，也不如一段代码跑得精准。

这一章，我就带你走一遍完整的流程——从数据获取，到指标编程，再到策略开发和回测。嗯，都是我在实盘里踩过坑之后总结出来的。

核心逻辑：订单流程序化交易 = 数据管道 + 指标引擎 + 策略逻辑 + 回测验证。四者缺一不可。

数据获取 Level2 / 逐笔成交 指标编程 Delta / CVD / 失衡 策略开发 入场 / 出场 / 风控 回测系统 绩效评估 / 参数优化 反馈优化

16.1 订单流数据的获取

做程序化交易，第一步就是搞定数据源。没有数据，什么策略都是空中楼阁。

我早期吃过这个亏——自己写了个爬虫去抓某平台的逐笔数据，结果人家反爬一升级，整个策略直接瘫痪。后来学乖了，老老实实走正规渠道。

目前主流的订单流数据获取方式有三种：

  • 交易所API直连：比如币安的WebSocket流，实时推送逐笔成交和深度快照。延迟最低，但需要自己维护连接。
  • 第三方数据商：像TickData、Quandl这类，提供清洗好的历史数据。适合回测，但实时性差一些。
  • 券商Level2行情：A股市场用这个，包含十档行情和逐笔委托。我建议用CTP接口接入，稳定。

我的经验：如果你做高频，必须用WebSocket直连。如果做日内波段，第三方数据完全够用。别为了省几百块钱去用免费数据——你想想看，数据延迟一秒，策略可能就从盈利变亏损了。

下面是一个简单的WebSocket订阅代码示例，订阅币安的逐笔成交数据：

import websocket
import json

def on_message(ws, message):
    data = json.loads(message)
    # 提取订单流关键字段
    price = float(data['p'])
    volume = float(data['q'])
    side = 'buy' if data['m'] == False else 'sell'
    print(f"价格: {price}, 量: {volume}, 方向: {side}")

def on_error(ws, error):
    print(f"连接异常: {error}")

def on_close(ws, close_status_code, close_msg):
    print("连接关闭，准备重连...")

def on_open(ws):
    # 订阅BTCUSDT的逐笔成交
    sub_msg = {
        "method": "SUBSCRIBE",
        "params": ["btcusdt@trade"],
        "id": 1
    }
    ws.send(json.dumps(sub_msg))

ws = websocket.WebSocketApp(
    "wss://stream.binance.com:9443/ws",
    on_open=on_open,
    on_message=on_message,
    on_error=on_error,
    on_close=on_close
)
ws.run_forever()

注意：WebSocket连接必须实现重连机制。我曾经因为网络抖动导致数据断流半小时，策略还在按旧数据开仓，结果你懂的...后来我加了心跳检测和自动重连，才算稳下来。

16.2 订单流指标的编程实现

数据拿到手了，接下来就是算指标。订单流里最常用的几个指标，说白了就是：Delta、CVD、失衡比。

我一个个说。

16.2.1 Delta（买卖量差）

Delta = 主动买量 - 主动卖量。正值说明买方更主动，负值反之。

实现起来很简单，但要注意：必须区分主动买和主动卖。在逐笔数据里，一般用「taker buy」标记来判断。

def calculate_delta(trades):
    """
    trades: list of dict, 每笔包含 {'price': float, 'volume': float, 'side': str}
    """
    buy_vol = sum(t['volume'] for t in trades if t['side'] == 'buy')
    sell_vol = sum(t['volume'] for t in trades if t['side'] == 'sell')
    delta = buy_vol - sell_vol
    return delta

16.2.2 CVD（累积成交量差）

CVD就是Delta的累积版本。它能帮你看到资金是在持续流入还是流出。

我个人习惯用CVD来过滤假突破——价格创新高但CVD没跟上，十有八九是诱多。

class CVDIndicator:
    def __init__(self):
        self.cumulative_delta = 0
    
    def update(self, delta):
        self.cumulative_delta += delta
        return self.cumulative_delta

16.2.3 失衡比（Imbalance Ratio）

失衡比 = (买量 - 卖量) / (买量 + 卖量)。范围在-1到1之间。

绝对值大于0.6就算显著失衡，这时候往往意味着短期方向要变了。

def imbalance_ratio(buy_vol, sell_vol):
    if buy_vol + sell_vol == 0:
        return 0
    return (buy_vol - sell_vol) / (buy_vol + sell_vol)

实战技巧：别只看单一指标。我一般把Delta、CVD和失衡比放在一起看。比如Delta为正、CVD持续上升、失衡比大于0.5——三个信号共振，胜率会高很多。

16.3 自动化交易策略的开发

指标算出来了，接下来就是写策略逻辑。订单流策略的核心思路其实就一句话：跟着聪明钱走。

我常用的一个策略框架是这样的：

  1. 入场条件：Delta连续3根K线为正 + CVD创近期新高 + 价格突破前高
  2. 出场条件：Delta转负 + CVD拐头向下
  3. 止损：最近一根大阳线的低点下方1个ATR
  4. 仓位管理：根据失衡比动态调整，失衡越强仓位越大

下面是一个简化版的策略代码：

class OrderFlowStrategy:
    def __init__(self, delta_threshold=100, cvd_breakout=True):
        self.delta_threshold = delta_threshold
        self.cvd_breakout = cvd_breakout
        self.position = 0
        self.cvd = CVDIndicator()
    
    def on_bar(self, bar_data):
        # bar_data包含该K线的所有逐笔成交
        delta = calculate_delta(bar_data['trades'])
        cvd_value = self.cvd.update(delta)
        
        # 入场逻辑
        if delta > self.delta_threshold and cvd_value > 0:
            if self.position == 0:
                self.enter_long(bar_data['close'])
                self.position = 1
        
        # 出场逻辑
        elif delta < -self.delta_threshold:
            if self.position == 1:
                self.exit_long(bar_data['close'])
                self.position = 0

避坑指南：我曾经在策略里用了未来函数——用当前K线的CVD去判断入场，但实际上CVD是累积值，包含了整个K线的数据。这导致回测曲线漂亮得不像话，实盘直接打脸。记住：只能用已完成的K线数据做判断。

16.4 回测系统的搭建

策略写好了，不回测就上实盘？那是赌博。

一个靠谱的回测系统，至少要有这几个模块：

模块 功能 注意事项
数据引擎 加载历史订单流数据，按时间切片 必须包含逐笔数据，不能用K线模拟
策略引擎 逐笔或逐K线执行策略逻辑 注意滑点和手续费模拟
风控模块 检查仓位、资金、最大回撤 实盘风控参数要和回测一致
绩效评估 计算夏普、胜率、盈亏比等 别只看总收益，要看资金曲线

我习惯用Python的backtrader框架做回测，但订单流数据需要自己写数据源适配器。下面是一个简单的回测框架示例：

import pandas as pd
from datetime import datetime

class OrderFlowBacktest:
    def __init__(self, strategy, data, initial_capital=100000):
        self.strategy = strategy
        self.data = data  # DataFrame, 包含逐笔数据
        self.capital = initial_capital
        self.equity_curve = []
    
    def run(self):
        # 按时间窗口切片，模拟逐K线运行
        for timestamp, bar_data in self.data.groupby(pd.Grouper(freq='1min')):
            self.strategy.on_bar(bar_data)
            # 记录权益
            self.equity_curve.append({
                'time': timestamp,
                'equity': self.capital + self.strategy.position * bar_data['close'].iloc[-1]
            })
        return pd.DataFrame(self.equity_curve)
    
    def evaluate(self):
        df = pd.DataFrame(self.equity_curve)
        total_return = (df['equity'].iloc[-1] - self.capital) / self.capital
        max_drawdown = (df['equity'].cummax() - df['equity']).max() / self.capital
        print(f"总收益率: {total_return:.2%}")
        print(f"最大回撤: {max_drawdown:.2%}")
        return total_return, max_drawdown

我的建议：回测时一定要加入滑点模拟。订单流策略通常在高流动性品种上跑，但滑点依然存在。我一般按0.1%的滑点来算，如果回测还能盈利，实盘才有戏。

嗯，到这里，订单流程序化交易的四个核心环节就都讲完了。从数据获取到指标编程，从策略开发到回测验证，每一步都有坑，但也都有解法。你想想看，把这些串起来，就是一个完整的自动化交易系统了。

最后说一句：程序化交易不是让你躺赚的魔法，而是帮你把交易纪律执行到位的工具。工具好不好，还得看用的人。

交易系统化学习资料 微信Strategy888888

← 上一章 📖 返回目录 下一章 →