第4章调度策略设计

4、调度策略设计：定时调度、事件驱动调度、混合调度模式、优先级队列

调度策略这东西，说白了就是「什么时候该干活」的问题。

我刚开始做量化系统那会儿，觉得调度嘛，不就是定个时跑脚本？后来被生产环境狠狠教育了几次，才明白调度策略设计的好坏，直接决定了因子生产的稳定性和时效性。

今天咱们就把这块掰开揉碎了聊。

4.1 定时调度：最朴素的方案，但坑不少

定时调度，就是按照预设的时间点触发任务。比如每天收盘后16:30开始计算日频因子，或者每小时整点计算分钟级因子。

实现方式其实很简单，Linux的crontab、Python的APScheduler、或者Airflow的定时DAG，都能搞定。

核心要点： 定时调度适合「数据到达时间可预测」的场景。

举个例子，我做过一个日频因子生产流水线。数据源是交易所的收盘行情，通常16:00左右就能拿到。我设了个定时任务，每天16:15启动。为什么留15分钟缓冲？因为网络抖动、数据延迟这些破事，你永远猜不到什么时候来。

# 一个简单的APScheduler定时任务示例
from apscheduler.schedulers.blocking import BlockingScheduler

def daily_factor_job():
    print("开始计算日频因子...")
    # 实际逻辑：读取数据、计算因子、写入数据库
    pass

scheduler = BlockingScheduler()
scheduler.add_job(daily_factor_job, 'cron', hour=16, minute=15)
scheduler.start()

避坑指南： 我曾经犯过一个低级错误——定时任务设在了数据理论到达时间的前5分钟。结果数据没到，任务跑了个寂寞，还报了空指针异常。后来我学乖了，定时任务至少留10-15分钟的冗余窗口。

定时调度的优点很明显：简单、可控、资源占用可预测。缺点也扎心：如果数据提前到了，你得干等着；如果数据延迟了，任务就白跑。

4.2 事件驱动调度：让数据来「敲门」

事件驱动调度，就是「有数据来了，我才干活」。这种方式更灵活，也更贴近真实的生产节奏。

实现方式通常依赖消息队列，比如Kafka、RabbitMQ。数据到达时发布一个事件，调度器监听到事件后，触发对应的因子计算任务。

  核心要点： 事件驱动调度适合「数据到达时间不确定」或「需要实时响应」的场景。

我记得有个项目是做高频因子，数据是逐笔成交数据，每秒可能来几百条。如果用定时调度，要么延迟太高，要么浪费资源。换成事件驱动后，每条数据到达就触发一次增量计算，延迟从秒级降到了毫秒级。

# 一个简化的事件驱动调度示例（伪代码）
def on_data_arrived(event):
    factor_id = event['factor_id']
    data = event['data']
    # 触发因子计算
    compute_factor(factor_id, data)

# 监听Kafka主题
kafka_consumer.subscribe('factor_data_topic')
for message in kafka_consumer:
    on_data_arrived(message.value)

事件驱动的优势是实时性好、资源利用率高。但缺点也明显：依赖消息中间件，系统复杂度上去了；而且如果事件风暴来了（比如瞬间涌入大量数据），处理不过来就容易崩。

我的经验： 事件驱动调度一定要做好背压机制。我曾经遇到过Kafka消费速度跟不上生产速度，结果消息积压了几百万条，因子计算延迟了整整两个小时。后来加了限流和降级策略，才算稳住。

4.3 混合调度模式：取长补短

实际生产中，很少有系统只用一种调度模式。混合调度，就是把定时调度和事件驱动结合起来，各取所长。

我常用的混合模式是这样的：

主调度： 事件驱动。数据来了就触发计算，保证实时性。
兜底调度： 定时调度。比如每30分钟检查一次，如果某个因子还没更新，就强制触发一次计算。

为什么要加兜底？因为事件驱动虽然好，但万一消息丢了、消费者挂了、或者数据源没发事件，因子就永远算不出来。定时兜底就像个「安全网」，确保不会漏掉任何一次计算。

  核心要点： 混合调度模式的核心思想是「事件驱动为主，定时兜底为辅」。

我参与过一个多因子生产平台，每天要处理上百个因子。有些因子是实时数据驱动的（比如高频因子），有些是定时计算的（比如日频因子）。我们设计了一个统一的调度框架：

因子类型	主调度模式	兜底策略
高频因子（Tick级）	事件驱动	每5分钟检查一次
分钟级因子	事件驱动	每30分钟检查一次
日频因子	定时调度	无（定时即可）
周频因子	定时调度	无（定时即可）

你看，不同因子用不同的调度策略，灵活搭配。这就是混合调度的精髓。

4.4 优先级队列：让重要的事先干

调度策略设计好了，但还有个问题：多个任务同时触发时，先算哪个？

优先级队列就是用来解决这个问题的。简单说，就是给每个任务打一个优先级标签，调度器按照优先级从高到低依次执行。

  核心要点： 优先级队列确保「高优任务不被低优任务阻塞」。

我举个例子。在因子生产系统中，有些因子是给实盘交易用的，延迟高了就会亏钱；有些因子是给研究团队做分析的，晚几分钟无所谓。如果两者同时触发，肯定先算实盘因子。

# 一个简单的优先级队列实现（Python）
import heapq

class PriorityTaskQueue:
    def __init__(self):
        self._queue = []
        self._index = 0

    def add_task(self, task, priority):
        # 优先级数值越小，优先级越高
        heapq.heappush(self._queue, (priority, self._index, task))
        self._index += 1

    def get_task(self):
        return heapq.heappop(self._queue)[-1]

# 使用示例
queue = PriorityTaskQueue()
queue.add_task("计算实盘因子", priority=1)
queue.add_task("计算研究因子", priority=5)
queue.add_task("计算回测因子", priority=10)

next_task = queue.get_task()  # 返回"计算实盘因子"

避坑指南： 我曾经踩过一个坑——优先级队列没有做饥饿预防。结果低优先级的任务（比如研究因子）一直被高优先级任务（比如实盘因子）抢占，导致研究因子永远算不出来。后来我加了「老化机制」：任务等待时间越长，优先级自动提升。这样低优任务最终也能被执行。

优先级队列的设计，其实就是在「公平」和「效率」之间找平衡。我的建议是：

高优任务： 实盘相关、时效性要求高的因子。优先级设为1-3。
中优任务： 日常监控、风控相关的因子。优先级设为4-6。
低优任务： 研究分析、历史回测相关的因子。优先级设为7-10。

4.5 调度策略的核心逻辑图

说了这么多，咱们用一张图把调度策略的核心逻辑串起来。这张图是我自己画的设计思路，你一看就明白。

这张图把整个调度流程串起来了。数据源进来后，要么走事件驱动（实时触发），要么走定时调度（固定时间触发）。两者都汇聚到混合调度模式里，由它统一管理。然后任务进入优先级队列，按重要性排序，最后交给执行器去算因子。

我的建议： 刚开始做调度系统时，别一上来就搞混合调度。先从定时调度开始，跑通了再加事件驱动。优先级队列也是，先简单分两三级就够了，别搞得太复杂。系统是慢慢演进出来的，不是一步到位的。

好了，调度策略设计这块就聊到这儿。核心就三句话：定时调度保底，事件驱动提效，优先级队列保公平。你想想看，是不是这个理？

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

📚 课程目录

4、调度策略设计：定时调度、事件驱动调度、混合调度模式、优先级队列

4.1 定时调度：最朴素的方案，但坑不少

4.2 事件驱动调度：让数据来「敲门」

4.3 混合调度模式：取长补短

4.4 优先级队列：让重要的事先干

4.5 调度策略的核心逻辑图