第11章自动化执行引擎

第十一章：自动化执行引擎：定时任务调度、条件单触发、止损止盈自动执行

自动化执行引擎，说白了就是让你的策略自己跑起来，不用你半夜爬起来盯盘。我见过太多交易员，策略逻辑写得漂亮，结果执行环节掉链子——要么忘了挂单，要么止损没及时触发，一夜回到解放前。这一章，咱们就把执行引擎的三大核心模块拆开揉碎：定时任务、条件单、止损止盈。

11.1 定时任务调度：让策略像闹钟一样准时

定时任务，是自动化执行的基础。你想想看，基差交易里很多操作都有固定时间窗口：比如每天收盘前15分钟调仓，或者每周五下午检查展期。手动操作？一次两次还行，时间长了肯定出错。

我个人习惯用 APScheduler 这个库，比 Python 自带的 sched 灵活得多。它支持 cron 表达式，可以精确到秒级调度。

核心要点：定时任务不是简单设个闹钟，要考虑交易时段、节假日、异常重试。

from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.cron import CronTrigger
import datetime

def daily_rebalance():
    """每日调仓任务"""
    print(f"[{datetime.datetime.now()}] 执行基差调仓...")
    # 检查当前持仓
    # 计算目标基差敞口
    # 发送委托单
    pass

def weekly_check():
    """每周展期检查"""
    print(f"[{datetime.datetime.now()}] 检查期货合约展期...")
    # 判断主力合约切换
    # 计算展期成本
    pass

scheduler = BlockingScheduler()

# 每天14:45执行调仓
scheduler.add_job(
    daily_rebalance,
    CronTrigger(hour=14, minute=45, second=0),
    id='rebalance_job',
    misfire_grace_time=60  # 允许60秒延迟
)

# 每周五15:00检查展期
scheduler.add_job(
    weekly_check,
    CronTrigger(day_of_week='fri', hour=15, minute=0),
    id='rollover_job'
)

scheduler.start()

嗯，这里要注意一个坑：misfire_grace_time 参数。我在项目中遇到过，某次服务器负载高，任务延迟了30秒才触发，结果错过了最佳调仓窗口。设置这个参数后，系统会容忍一定延迟，超过阈值就直接跳过，避免堆积执行。

避坑指南：我曾经在生产环境里忘记处理夏令时切换，结果定时任务全部偏移了一小时。建议用 UTC 时间做底层调度，展示时再转本地时间。

11.2 条件单触发：让市场帮你做决策

条件单，就是「如果……那么……」的逻辑。基差交易里最常见的条件单有两种：

价格条件：基差扩大到某个阈值时开仓
时间条件：特定时间点检查持仓状态

为什么需要条件单？因为市场不会等你。基差可能在几秒钟内从-5跳到+10，手动下单根本来不及。我建议把条件单引擎设计成独立模块，和主策略解耦。

class ConditionEngine:
    def __init__(self):
        self.conditions = []
        self.active = True
    
    def add_condition(self, condition_id, check_func, action_func):
        """添加条件单
        check_func: 返回True/False的判断函数
        action_func: 触发后执行的函数
        """
        self.conditions.append({
            'id': condition_id,
            'check': check_func,
            'action': action_func,
            'triggered': False
        })
    
    def run_once(self, market_data):
        """每次行情更新时调用"""
        for cond in self.conditions:
            if cond['triggered']:
                continue
            try:
                if cond['check'](market_data):
                    cond['action'](market_data)
                    cond['triggered'] = True
                    print(f"条件单 {cond['id']} 已触发")
            except Exception as e:
                print(f"条件单 {cond['id']} 检查异常: {e}")
    
    def reset(self, condition_id=None):
        """重置条件单状态"""
        if condition_id:
            for cond in self.conditions:
                if cond['id'] == condition_id:
                    cond['triggered'] = False
        else:
            for cond in self.conditions:
                cond['triggered'] = False

# 使用示例
engine = ConditionEngine()

def check_basis_widening(data):
    """检查基差是否扩大到开仓条件"""
    basis = data['spot_price'] - data['futures_price']
    return basis > 10  # 基差大于10点

def open_position(data):
    """执行开仓"""
    print(f"基差 {data['basis']:.2f}，触发开仓")
    # 发送委托...

engine.add_condition('basis_open', check_basis_widening, open_position)

你可能会问：条件单检查频率多高合适？我个人经验是，对于基差交易，每秒检查一次就够了。基差变化不像高频交易那么快，太频繁反而浪费资源。但如果你的策略涉及秒级套利，那得用事件驱动的方式，而不是轮询。

11.3 止损止盈自动执行：保住利润，截断亏损

止损止盈，是交易员的保命符。基差交易虽然波动相对小，但遇到极端行情（比如交割月逼仓），一天亏20%也不是不可能。我见过不止一个交易员，因为手动止损犹豫，把盈利单拿成了亏损单。

自动化止损止盈，核心就三个要素：

触发条件：价格、时间、波动率
执行逻辑：市价单、限价单、分批出场
风控保护：单笔最大亏损、总敞口限制

关键设计：止损单必须独立于主策略运行。即使策略逻辑崩溃，止损引擎也要能正常工作。

class StopManager:
    def __init__(self, account):
        self.account = account
        self.stop_orders = {}  # position_id -> stop_config
    
    def set_stop_loss(self, position_id, price, quantity, stop_type='fixed'):
        """设置止损
        stop_type: 'fixed'固定价格, 'trailing'移动止损
        """
        self.stop_orders[position_id] = {
            'price': price,
            'quantity': quantity,
            'type': stop_type,
            'high_water_mark': None  # 用于移动止损
        }
    
    def check_and_execute(self, market_prices):
        """检查所有止损单"""
        for pos_id, config in self.stop_orders.items():
            position = self.account.get_position(pos_id)
            if not position:
                continue
            
            current_price = market_prices.get(position.symbol)
            if not current_price:
                continue
            
            # 固定止损
            if config['type'] == 'fixed':
                if position.direction == 'long' and current_price <= config['price']:
                    self._execute_stop(pos_id, current_price)
                elif position.direction == 'short' and current_price >= config['price']:
                    self._execute_stop(pos_id, current_price)
            
            # 移动止损
            elif config['type'] == 'trailing':
                if config['high_water_mark'] is None:
                    config['high_water_mark'] = current_price
                else:
                    # 更新最高价
                    if position.direction == 'long':
                        config['high_water_mark'] = max(config['high_water_mark'], current_price)
                        stop_price = config['high_water_mark'] * (1 - config['trail_percent'])
                        if current_price <= stop_price:
                            self._execute_stop(pos_id, current_price)
                    else:
                        config['high_water_mark'] = min(config['high_water_mark'], current_price)
                        stop_price = config['high_water_mark'] * (1 + config['trail_percent'])
                        if current_price >= stop_price:
                            self._execute_stop(pos_id, current_price)
    
    def _execute_stop(self, pos_id, price):
        """执行止损"""
        config = self.stop_orders[pos_id]
        print(f"止损触发: {pos_id} @ {price:.2f}")
        # 发送市价单平仓
        self.account.close_position(pos_id, config['quantity'])
        del self.stop_orders[pos_id]

这里有个实战细节：移动止损的 trail_percent 怎么设？我一般用 ATR（平均真实波幅）来动态调整。比如设置 2 倍 ATR 作为回撤容忍度，这样在波动大的时候止损不会太紧，波动小的时候又能及时锁定利润。

个人经验：止损单最好用市价单，别用限价单。我曾经为了省几个 tick 的滑点，用限价单做止损，结果行情直接跳空穿过去，单子没成交，亏损扩大了一倍。止损的目的不是优化价格，是活着出来。

11.4 执行引擎的整体架构

把上面三个模块整合起来，就是一个完整的自动化执行引擎。我画了一张架构图，你看一眼就明白了：

从图上你能看到，行情数据进来后，核心调度器负责分发到三个子模块。定时任务负责「到点就做」，条件单负责「到了就做」，止损止盈负责「坏了就跑」。三者互不干扰，但最终都汇聚到委托单输出接口。

11.5 实战中的几个关键问题

最后，聊几个我在实战中踩过的坑：

问题	现象	解决方案
任务重复执行	同一个调仓任务在1秒内执行了3次	加分布式锁（Redis锁），确保同一时刻只有一个实例执行
止损单未触发	行情跳空，止损价格没碰到	改用市价单，或者设置「如果价格超过X，立即市价平仓」
条件单堆积	行情剧烈波动时，条件单检查队列积压	设置最大检查频率，行情更新太快时直接丢弃旧数据
系统重启后任务丢失	服务器重启，所有定时任务和条件单都没了	任务持久化到数据库，启动时自动恢复未完成的任务

嗯，自动化执行引擎这东西，看着简单，但细节决定成败。我见过有人把止损逻辑写在策略主循环里，结果策略卡住了，止损也跟着失效。记住一条原则：风控模块必须独立，且优先级最高。哪怕策略逻辑全崩了，止损止盈也得能正常工作。

好了，这一章的内容就这些。自动化执行引擎是基差交易系统的「手脚」，没有它，再好的策略也只是纸上谈兵。下一章咱们聊聊回测系统的搭建——怎么验证你的策略在历史上能不能赚钱。

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

📚 课程目录

第十一章：自动化执行引擎：定时任务调度、条件单触发、止损止盈自动执行

11.1 定时任务调度：让策略像闹钟一样准时

11.2 条件单触发：让市场帮你做决策

11.3 止损止盈自动执行：保住利润，截断亏损

11.4 执行引擎的整体架构

11.5 实战中的几个关键问题