第4章事件驱动交易系统搭建

第四章：事件驱动交易系统搭建

系统架构设计、数据层、策略层、执行层——这四个词，说白了就是一套交易系统的骨架、血肉、大脑和手脚。我见过太多人一上来就写策略代码，结果数据源断了都不知道，或者下单接口报错了还在傻等。嗯，今天我们就把这四层拆开揉碎了讲清楚。

4.1 系统架构设计：别让系统成为你的绊脚石

我个人习惯把事件驱动系统分成三层：数据层、策略层、执行层。为什么不是四层五层？因为多了反而容易乱。你想想看，一个事件从发生到交易完成，路径越短越好。

核心架构其实就一句话：数据层负责喂料，策略层负责判断，执行层负责动手。这三层之间通过消息队列或者共享内存通信。我早期用Redis做消息队列，后来发现量大了容易丢数据，现在改用Kafka了。

架构设计要点：

各层之间松耦合——数据层挂了不影响策略层，策略层改了不用动执行层
数据流单向流动——数据→策略→执行，别搞成循环依赖
异常处理要独立——每层都有自己的熔断机制

下面这张图是我自己项目里用的架构，你可以参考一下：

4.2 数据层：选对数据库，少走三年弯路

数据层是整个系统的根基。根基不稳，后面全是白搭。我见过有人用MySQL存行情数据，结果一张表几亿条记录，查询一次要半分钟——这还做什么交易？

数据库选型建议

数据类型	推荐数据库	原因
行情数据（Tick/分钟线）	InfluxDB / TimescaleDB	时序数据库，写入快，查询快，自带降采样
事件数据（新闻/公告）	MongoDB / Elasticsearch	文档型，支持全文检索，适合非结构化数据
交易记录/账户数据	PostgreSQL	关系型，事务支持好，数据一致性有保障
缓存/实时数据	Redis	内存数据库，读写极快，适合做消息队列

我的经验：别想着一个数据库打天下。我早期用MongoDB存所有数据，结果查询行情数据慢得要死。后来拆成InfluxDB存行情、MongoDB存事件、PostgreSQL存交易记录，各司其职，舒服多了。

数据库设计要点

设计表结构时，记住一个原则：写入快于读取。交易系统里，数据写入频率远高于读取频率。你想想看，一秒可能写入几百条Tick数据，但查询可能几分钟才一次。

举个例子，行情数据表的设计：

-- 行情数据表（TimescaleDB超表）
CREATE TABLE market_data (
    time TIMESTAMPTZ NOT NULL,
    symbol VARCHAR(20) NOT NULL,
    price DOUBLE PRECISION,
    volume BIGINT,
    bid_price DOUBLE PRECISION,
    ask_price DOUBLE PRECISION
);

-- 转换为超表，按时间和标的自动分区
SELECT create_hypertable('market_data', 'time');

-- 创建索引，加速查询
CREATE INDEX idx_symbol_time ON market_data (symbol, time DESC);

避坑指南：我曾经犯过一个错误——没有给时间列建索引。结果回测时查询某只股票的历史数据，跑了五分钟还没出来。后来加了索引，同样的查询只要几秒钟。记住：时序数据一定要按时间建索引！

4.3 策略层：事件识别与信号生成

策略层是整个系统的核心。说白了，就是判断「这个事件值不值得交易」。我个人习惯把事件分成三类：

确定性事件：财报发布、分红除权、指数调整——时间可预测，影响相对明确
半确定性事件：并购重组、定增、回购——时间不确定，但影响大
随机事件：突发事件、舆情危机、政策变化——完全不可预测

事件识别的流程，我一般这么走：

数据采集：从数据层获取原始数据
事件提取：用规则或模型提取事件特征
事件分类：判断事件类型和影响程度
信号生成：根据事件类型生成交易信号

下面是一个简单的事件识别代码示例：

import pandas as pd
from datetime import datetime, timedelta

class EventDetector:
    def __init__(self):
        self.event_rules = {
            'earnings': self._check_earnings,
            'dividend': self._check_dividend,
            'index_adjust': self._check_index_adjust
        }
    
    def detect_events(self, market_data, news_data):
        events = []
        for rule_name, rule_func in self.event_rules.items():
            detected = rule_func(market_data, news_data)
            if detected:
                events.append({
                    'type': rule_name,
                    'time': datetime.now(),
                    'data': detected
                })
        return events
    
    def _check_earnings(self, market_data, news_data):
        # 检查是否有财报发布
        earnings_news = news_data[news_data['type'] == 'earnings']
        if len(earnings_news) > 0:
            return earnings_news.iloc[0]
        return None
    
    def _check_dividend(self, market_data, news_data):
        # 检查是否有分红除权
        dividend_news = news_data[news_data['type'] == 'dividend']
        if len(dividend_news) > 0:
            return dividend_news.iloc[0]
        return None

信号生成的核心逻辑：

事件强度：影响越大，信号越强
事件确定性：确定性越高，仓位可以越重
市场情绪：结合市场整体情绪调整信号
风险预算：每个事件分配的仓位不能超过风险预算

4.4 执行层：自动化下单接口

执行层是最后一步，也是最容易出问题的一步。我见过太多人策略写得好好的，一到实盘就卡在下单接口上。嗯，这里有几个关键点要注意：

下单接口选型

接口类型	适用场景	延迟	稳定性
CTP（期货）	商品期货、股指期货	<10ms	高
XTP（股票）	A股、港股通	<20ms	高
券商API	普通散户交易	100ms-1s	中等
量化平台API	回测、模拟交易	1s-5s	低

下单的核心逻辑其实不复杂，但细节很多：

class OrderExecutor:
    def __init__(self, api_connector):
        self.api = api_connector
        self.pending_orders = []
        self.filled_orders = []
    
    def execute_order(self, signal):
        # 根据信号生成订单
        order = {
            'symbol': signal['symbol'],
            'direction': signal['direction'],  # 'buy' or 'sell'
            'quantity': signal['quantity'],
            'order_type': 'limit',  # 限价单
            'price': signal['price'],
            'time': datetime.now()
        }
        
        # 发送订单
        try:
            order_id = self.api.place_order(order)
            self.pending_orders.append(order_id)
            return order_id
        except Exception as e:
            self._handle_error(e)
            return None
    
    def _handle_error(self, error):
        # 错误处理：重试、报警、熔断
        if 'timeout' in str(error):
            # 超时重试
            self._retry_order()
        elif 'rejected' in str(error):
            # 订单被拒，记录日志
            self._log_rejection(error)
        else:
            # 其他错误，触发熔断
            self._circuit_breaker()

避坑指南：我曾经在实盘时遇到过一个坑——下单接口返回了成功，但实际没有成交。后来发现是网络延迟导致订单状态没更新。从那以后，我加了一个订单状态轮询机制，每100ms检查一次订单状态，直到确认成交或超时。

4.5 系统集成与测试

三层都搭好了，接下来就是集成测试。我个人习惯按这个顺序来：

单元测试：每层单独测试，确保功能正常
集成测试：三层联调，模拟真实数据流
压力测试：模拟高并发场景，看系统扛不扛得住
回测验证：用历史数据跑一遍，看策略表现

我的经验：集成测试时，一定要模拟异常情况。比如数据源断了怎么办？下单接口超时怎么办？这些场景在实盘里一定会遇到。提前做好预案，比到时候手忙脚乱强得多。

好了，系统搭建的核心内容就这些。记住：架构设计要简洁，数据层要稳，策略层要准，执行层要快。把这四层搞明白了，你的事件驱动交易系统就成功了一大半。

公众号:蓝海资料掘金营，微信deep3321

📚 课程目录

第四章：事件驱动交易系统搭建

4.1 系统架构设计：别让系统成为你的绊脚石

4.2 数据层：选对数据库，少走三年弯路

数据库选型建议

数据库设计要点

4.3 策略层：事件识别与信号生成

4.4 执行层：自动化下单接口

下单接口选型

4.5 系统集成与测试