第三十讲：订单流综合实战项目——构建一个完整的订单流交易系统

终于到了最后一讲。说实话，写到这里我有点感慨。

前面二十九讲，我们从盘口数据讲起，一步步深入到订单流的核心逻辑。现在，是时候把所有东西串起来了。

这一讲，我们要做一个完整的订单流交易系统。从数据获取，到策略执行，全流程用Python实现。你想想看，这其实就是一个迷你版的量化交易平台。

系统架构设计

先画个图，把整体框架理清楚。我个人习惯，做任何系统之前，先画架构图。不然写着写着就乱了。

这个架构分四层：数据层、处理层、策略层、执行层。每层各司其职，数据从下往上流动，信号从上往下执行。

第一步：数据获取模块

数据是订单流系统的命根子。没有高质量的数据，再好的策略也是白搭。

import asyncio
import websockets
import json
from collections import deque

class OrderFlowDataFeed:
    """订单流数据源"""

    def __init__(self, symbol='btcusdt', depth_levels=50):
        self.symbol = symbol
        self.depth_levels = depth_levels
        self.depth_cache = {'bids': deque(maxlen=100), 'asks': deque(maxlen=100)}
        self.trade_buffer = deque(maxlen=1000)  # 逐笔成交缓存

    async def connect_depth(self):
        """连接深度数据WebSocket"""
        url = f'wss://stream.binance.com:9443/ws/{self.symbol}@depth{self.depth_levels}@100ms'
        async with websockets.connect(url) as ws:
            while True:
                data = json.loads(await ws.recv())
                self._update_depth(data)

    async def connect_trades(self):
        """连接逐笔成交WebSocket"""
        url = f'wss://stream.binance.com:9443/ws/{self.symbol}@trade'
        async with websockets.connect(url) as ws:
            while True:
                data = json.loads(await ws.recv())
                self.trade_buffer.append(data)

    def _update_depth(self, data):
        """更新盘口深度"""
        for side in ['bids', 'asks']:
            for price, qty in data.get(side, []):
                qty = float(qty)
                if qty == 0:
                    # 价格为0表示删除该档位
                    self.depth_cache[side] = [
                        d for d in self.depth_cache[side] if d[0] != price
                    ]
                else:
                    self.depth_cache[side].append((float(price), qty))

    def get_order_flow(self):
        """获取当前订单流数据"""
        return {
            'depth': self.depth_cache,
            'recent_trades': list(self.trade_buffer)[-50:]
        }

这里我用了异步WebSocket连接。为什么？因为订单流数据是高频的，同步请求根本跟不上节奏。我之前有个项目，一开始用requests轮询，结果延迟高得离谱，后来改成WebSocket，延迟从秒级降到了毫秒级。

第二步：订单流重构引擎

原始数据拿到手，接下来要重构出订单流的真实面貌。说白了，就是把买卖双方的博弈过程还原出来。

import numpy as np
import pandas as pd

class OrderFlowReconstructor:
    """订单流重构引擎"""

    def __init__(self, tick_size=0.01):
        self.tick_size = tick_size
        self.volume_profile = {}  # 价格 -> 成交量映射

    def reconstruct_from_trades(self, trades):
        """从逐笔成交重构订单流"""
        flow_data = []

        for trade in trades:
            price = float(trade['p'])
            qty = float(trade['q'])
            is_buyer_maker = trade['m']  # True表示卖方主动

            # 判断主动方向
            if is_buyer_maker:
                # 卖方主动卖出，属于被动买入
                side = 'sell'
            else:
                # 买方主动买入，属于被动卖出
                side = 'buy'

            flow_data.append({
                'price': price,
                'volume': qty,
                'side': side,
                'timestamp': trade['T']
            })

            # 更新成交量分布
            price_key = round(price / self.tick_size) * self.tick_size
            if price_key not in self.volume_profile:
                self.volume_profile[price_key] = {'buy': 0, 'sell': 0}
            self.volume_profile[price_key][side] += qty

        return pd.DataFrame(flow_data)

    def calculate_delta(self, df):
        """计算Delta（主动买卖差）"""
        df['delta'] = np.where(df['side'] == 'buy', df['volume'], -df['volume'])
        df['cumulative_delta'] = df['delta'].cumsum()
        return df

    def detect_imbalance(self, df, threshold=2.0):
        """检测失衡信号"""
        df['buy_volume'] = np.where(df['side'] == 'buy', df['volume'], 0)
        df['sell_volume'] = np.where(df['side'] == 'sell', df['volume'], 0)

        # 滚动窗口计算失衡比
        window = 20
        df['buy_sum'] = df['buy_volume'].rolling(window).sum()
        df['sell_sum'] = df['sell_volume'].rolling(window).sum()
        df['imbalance_ratio'] = df['buy_sum'] / (df['sell_sum'] + 1e-8)

        df['imbalance_signal'] = np.where(
            df['imbalance_ratio'] > threshold, 'buy_imbalance',
            np.where(df['imbalance_ratio'] < 1/threshold, 'sell_imbalance', 'neutral')
        )
        return df

这里有个细节要注意：is_buyer_maker这个字段。很多人会搞反。我曾经就因为这个bug，回测跑出来收益惊人，实盘一跑就亏。后来发现是方向搞反了。嗯，这种坑踩过一次就记住了。

第三步：策略引擎实现

有了重构好的订单流数据，接下来就是策略逻辑。我们实现一个经典的POC（控制点）追踪策略。

class POCStrategy:
    """POC追踪策略"""

    def __init__(self, lookback_bars=50, threshold_pct=0.3):
        self.lookback_bars = lookback_bars
        self.threshold_pct = threshold_pct
        self.position = 0
        self.trades = []

    def find_poc(self, volume_profile):
        """找到成交量最大的价格（控制点）"""
        if not volume_profile:
            return None
        poc_price = max(volume_profile, key=lambda k: volume_profile[k]['buy'] + volume_profile[k]['sell'])
        return poc_price

    def calculate_value_area(self, volume_profile, poc_price, volume_pct=0.7):
        """计算价值区间（VAH/VAL）"""
        total_volume = sum(v['buy'] + v['sell'] for v in volume_profile.values())
        target_volume = total_volume * volume_pct

        sorted_prices = sorted(volume_profile.keys())
        poc_idx = sorted_prices.index(poc_price)

        # 从POC向两边扩展
        current_volume = volume_profile[poc_price]['buy'] + volume_profile[poc_price]['sell']
        left_idx = right_idx = poc_idx

        while current_volume < target_volume:
            left_vol = volume_profile.get(sorted_prices[left_idx - 1], {}).get('buy', 0) + \
                       volume_profile.get(sorted_prices[left_idx - 1], {}).get('sell', 0) if left_idx > 0 else 0
            right_vol = volume_profile.get(sorted_prices[right_idx + 1], {}).get('buy', 0) + \
                        volume_profile.get(sorted_prices[right_idx + 1], {}).get('sell', 0) if right_idx < len(sorted_prices) - 1 else 0

            if left_vol >= right_vol and left_idx > 0:
                left_idx -= 1
                current_volume += left_vol
            elif right_idx < len(sorted_prices) - 1:
                right_idx += 1
                current_volume += right_vol
            else:
                break

        return {
            'val': sorted_prices[left_idx],
            'vah': sorted_prices[right_idx],
            'poc': poc_price
        }

    def generate_signal(self, current_price, value_area):
        """生成交易信号"""
        if current_price > value_area['vah']:
            # 价格突破价值区间上沿，看多
            return 'buy'
        elif current_price < value_area['val']:
            # 价格跌破价值区间下沿，看空
            return 'sell'
        else:
            return 'hold'

    def execute(self, market_data):
        """策略执行入口"""
        volume_profile = market_data.get('volume_profile', {})
        current_price = market_data.get('price')

        if not volume_profile or not current_price:
            return None

        poc_price = self.find_poc(volume_profile)
        if not poc_price:
            return None

        value_area = self.calculate_value_area(volume_profile, poc_price)
        signal = self.generate_signal(current_price, value_area)

        return {
            'signal': signal,
            'poc': poc_price,
            'val': value_area['val'],
            'vah': value_area['vah'],
            'current_price': current_price
        }

第四步：风控与执行模块

策略再好，没有风控也是白搭。我见过太多人，策略赚了九次，一次风控没做好就全亏回去了。

class RiskManager:
    """风控管理器"""

    def __init__(self, max_position=100, max_drawdown=0.05, max_daily_loss=0.03):
        self.max_position = max_position
        self.max_drawdown = max_drawdown
        self.max_daily_loss = max_daily_loss
        self.initial_equity = None
        self.peak_equity = None
        self.daily_pnl = 0

    def check_position_limit(self, current_position, order_qty):
        """检查仓位限制"""
        return abs(current_position + order_qty) <= self.max_position

    def check_drawdown(self, current_equity):
        """检查回撤"""
        if self.peak_equity is None:
            self.peak_equity = current_equity
            return True

        self.peak_equity = max(self.peak_equity, current_equity)
        drawdown = (self.peak_equity - current_equity) / self.peak_equity

        return drawdown <= self.max_drawdown

    def check_daily_loss(self, pnl):
        """检查当日亏损"""
        self.daily_pnl += pnl
        return self.daily_pnl >= -self.max_daily_loss * self.initial_equity

    def can_trade(self, signal, current_position, current_equity, pnl):
        """综合风控检查"""
        checks = {
            'position_limit': self.check_position_limit(current_position, signal['qty']),
            'drawdown': self.check_drawdown(current_equity),
            'daily_loss': self.check_daily_loss(pnl)
        }

        if not all(checks.values()):
            failed_checks = [k for k, v in checks.items() if not v]
            print(f'风控未通过: {failed_checks}')
            return False

        return True


class OrderExecutor:
    """订单执行器"""

    def __init__(self, exchange_api):
        self.api = exchange_api
        self.pending_orders = []
        self.filled_orders = []

    async def execute_market_order(self, symbol, side, qty):
        """执行市价单"""
        order = {
            'symbol': symbol,
            'side': side,
            'qty': qty,
            'type': 'market',
            'status': 'pending'
        }

        try:
            # 模拟下单
            result = await self.api.place_order(**order)
            order['status'] = 'filled'
            order['fill_price'] = result['price']
            self.filled_orders.append(order)
            return order
        except Exception as e:
            order['status'] = 'failed'
            order['error'] = str(e)
            self.pending_orders.append(order)
            return None

    def calculate_position_size(self, signal_strength, account_balance, risk_per_trade=0.02):
        """计算仓位大小"""
        base_qty = account_balance * risk_per_trade
        # 信号强度调整仓位
        adjusted_qty = base_qty * (1 + signal_strength)
        return round(adjusted_qty, 2)

第五步：主循环与系统集成

最后，把所有模块拼起来，形成一个完整的运行循环。

class OrderFlowTradingSystem:
    """订单流交易系统主控"""

    def __init__(self, symbol='btcusdt'):
        self.data_feed = OrderFlowDataFeed(symbol)
        self.reconstructor = OrderFlowReconstructor()
        self.strategy = POCStrategy()
        self.risk_manager = RiskManager()
        self.executor = OrderExecutor(exchange_api=None)  # 实际使用时传入API

        self.position = 0
        self.equity = 100000  # 初始资金
        self.pnl = 0

    async def run(self):
        """系统主循环"""
        print('订单流交易系统启动...')

        # 启动数据采集
        depth_task = asyncio.create_task(self.data_feed.connect_depth())
        trade_task = asyncio.create_task(self.data_feed.connect_trades())

        while True:
            # 1. 获取最新数据
            raw_data = self.data_feed.get_order_flow()

            # 2. 重构订单流
            trades_df = self.reconstructor.reconstruct_from_trades(
                raw_data['recent_trades']
            )
            trades_df = self.reconstructor.calculate_delta(trades_df)
            trades_df = self.reconstructor.detect_imbalance(trades_df)

            # 3. 构建市场数据
            market_data = {
                'volume_profile': self.reconstructor.volume_profile,
                'price': float(raw_data['recent_trades'][-1]['p']) if raw_data['recent_trades'] else None,
                'delta': trades_df['cumulative_delta'].iloc[-1] if len(trades_df) > 0 else 0,
                'imbalance': trades_df['imbalance_signal'].iloc[-1] if len(trades_df) > 0 else 'neutral'
            }

            # 4. 策略决策
            decision = self.strategy.execute(market_data)

            if decision and decision['signal'] != 'hold':
                # 5. 风控检查
                order_qty = self.executor.calculate_position_size(
                    signal_strength=0.5,
                    account_balance=self.equity
                )

                if self.risk_manager.can_trade(
                    signal={'qty': order_qty},
                    current_position=self.position,
                    current_equity=self.equity,
                    pnl=self.pnl
                ):
                    # 6. 执行交易
                    side = 'buy' if decision['signal'] == 'buy' else 'sell'
                    order = await self.executor.execute_market_order(
                        symbol=self.data_feed.symbol,
                        side=side,
                        qty=order_qty
                    )

                    if order:
                        self.position += order_qty if side == 'buy' else -order_qty
                        print(f'执行交易: {side} {order_qty} @ {order["fill_price"]}')

            # 7. 等待下一个周期
            await asyncio.sleep(0.1)  # 100ms一个周期

好了，整个订单流交易系统就搭建完成了。从数据获取到策略执行，每一步都是我自己在实战中踩过坑、优化过的方案。你拿回去跑一跑，有问题随时交流。

记住一点：系统是死的，市场是活的。再好的系统也需要你根据市场变化不断调整。别指望一劳永逸。

---

无相订单流研究社 微信Lucian808555