

> 系列前情：#01 我们选定了 SQLite 作为本地存储方案（单文件、零部署、查询快）。本篇解决「每天数据怎么自动进来」——告别手动 `python main.py`。

## 一、为什么需要自动调度

研究阶段你可以每天开终端跑一次脚本，但真到生产环境：

- 节假日忘了跑 → 数据断了 → 策略信号计算错位
- 每天手动重复 30 分钟 → 时间都浪费在机械操作上
- 策略代码更新后忘了重启 → 旧代码继续跑 → 莫名亏损

把「定时 + 拉取 + 入库」全链路打包成自动化任务，由调度器按计划触发。本文用 **APScheduler** 演示最小可运行版本。

## 二、本文你将得到什么

1. **APScheduler 最小可用代码**：10 行完成「每 15 秒触发一次任务」
2. **生产模式 cron 模板**：每个交易日 15:30 收盘后自动跑
3. **增量去重 upsert**：`INSERT OR IGNORE` 保证同一只股票同一天只入库一次
4. **证书 fallback + 重试机制**：内部证书失效时自动切换演示证书，保证流水线不中断

## 三、整体架构

```
APScheduler 触发 → run_pipeline() 
                  → fetch_one(stock)        # API 拉取（含 fallback）
                  → upsert_kline(rows)      # SQLite 批量入库（去重）
                  → task_log 记录           # 可追溯的历史触发日志
```

四块各司其职，调度器只管「几点几分触发」，业务代码只管「拉 + 存」，互不耦合。

## 四、核心代码（可复制即用）

```python
from apscheduler.schedulers.blocking import BlockingScheduler
from mairui import Client
import sqlite3, time

STOCKS = ["600519", "000001", "300750"]
DEMO_LICENCE = "LICENCE-66D8-9F96-0C7F0FBCD073"

def fetch_one(stock: str, licence: str) -> list:
    for attempt in range(2):
        try:
            client = Client(licence)
            return client.stock_history(stock, "d", "f") or []
        except Exception as e:
            print(f"[retry {attempt+1}] {stock} {str(e)[:50]}")
            time.sleep(0.5)
    return []

def upsert(rows):
    conn = sqlite3.connect("kline.db")
    cur = conn.cursor()
    for stock, r in rows:
        cur.execute("""
        INSERT OR IGNORE INTO kline
            (date, stock, open, high, low, close, pre_close, volume)
        VALUES (?, ?, ?, ?, ?, ?, ?, ?)
        """, (
            r["t"][:10], stock,
            r["o"], r["h"], r["l"], r["c"],
            r["pc"], r["v"]
        ))
    conn.commit()
    conn.close()

def run_pipeline():
    """单次任务：拉全市场 → 入库"""
    all_rows = []
    for stock in STOCKS:
        rows = fetch_one(stock, DEMO_LICENCE)
        all_rows.extend([(stock, r) for r in rows])
    if all_rows:
        upsert(all_rows)
        print(f"[task] inserted {len(all_rows)} rows")

# ========== 演示模式：每 15 秒触发一次 ==========
sched = BlockingScheduler(timezone="Asia/Shanghai")
sched.add_job(run_pipeline, "interval", seconds=15,
              id="demo_pipeline", max_instances=1)
sched.start()
```

## 五、证书 fallback 与重试

正式项目里证书可能因为额度到期、权限调整而失效。Demo 加了「内部证书 → 演示证书」的两级 fallback + 2 次重试，确保流水线不轻易中断：

```python
def fetch_one(stock: str) -> list:
    """拉单只股票（含证书 fallback + 重试）"""
    licences = [INTERNAL_LICENCE, DEMO_LICENCE]
    for lic in licences:
        label = "internal" if lic == INTERNAL_LICENCE else "demo"
        for attempt in range(2):
            try:
                client = Client(lic)
                raw = client.stock_history(stock, "d", "f")
                if raw:
                    return raw
            except Exception as e:
                print(f"[retry {attempt+1}] {stock} cert={label} {str(e)[:50]}")
                time.sleep(0.5)
        print(f"[fallback] {stock} {label} cert 失败，尝试下一个")
    return []
```

**关键设计**：

- 内层循环：retry（同一证书重试 2 次，应对网络抖动）
- 外层循环：fallback（换证书，应对权限失效）
- 全部失败才放弃 → 不掩盖问题，但也不会因为单一证书就停摆

## 六、增量去重：PRIMARY KEY + INSERT OR IGNORE

SQLite 建表时把 `(date, stock)` 设为复合主键，配合 `INSERT OR IGNORE` 实现「存在则跳过，不存在则插入」：

```sql
CREATE TABLE kline (
    date TEXT NOT NULL,
    stock TEXT NOT NULL,
    open REAL, high REAL, low REAL, close REAL,
    pre_close REAL, volume INTEGER,
    PRIMARY KEY (date, stock)
);
```

实测：第一次跑 inserted=150，第二次跑 inserted=0 / skipped=150 —— 完全幂等，重复触发不会污染数据。

## 七、生产模式：cron + 进程守护

演示模式每 15 秒跑一次仅供测试，生产环境每个交易日 15:30 收盘后跑一次就够：

```python
sched = BlockingScheduler(timezone="Asia/Shanghai")
sched.add_job(
    run_pipeline,
    "cron",
    day_of_week="mon-fri",   # 仅交易日触发
    hour=15, minute=30,      # 收盘后半小时
    id="daily_kline",
    max_instances=1
)
sched.start()
```

**进程守护（关键）**：

- **Linux**：`systemd` 或 `supervisord` 守护，挂了自动重启
- **Windows**：`nssm` 把脚本注册成系统服务，或用「任务计划程序」触发
- **Docker**：用 `--restart=always` 参数；容器内只跑 scheduler

## 八、任务可追溯：task_log 表

每次任务执行写一行到 SQLite，3 个月后出问题可以一键追溯：

```sql
CREATE TABLE task_log (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    run_at TEXT NOT NULL,    -- 'YYYY-MM-DD HH:MM:SS'
    status TEXT,             -- 'ok' / 'fail'
    fetched INTEGER,         -- 拉到多少条
    inserted INTEGER,        -- 新增多少条
    skipped INTEGER,         -- 跳过（去重）多少条
    note TEXT                -- 详情 '600519:50/0, ...'
);
```

配合监控告警（详见 #03）就能实时发现「任务没跑 / 跑了没拉数据 / 拉到了 0 条」等异常。

## 九、常见坑

1. **节假日/周末 cron 不会自动跳过**：`day_of_week="mon-fri"` 只过滤周末，节假日需要另外判断（建议接入交易日历表）。
2. **`max_instances=1` 不能省**：如果任务耗时超过调度间隔（如 5 秒任务 + 5 秒间隔），没这个参数会并发跑 → 数据库锁死。
3. **证书 fallback 是兜底、不是常态**：fallback 到演示证书说明正式证书失效，**必须当天告警 + 续费**，否则生产数据会被演示证限制（如演示证只能拉 60 个交易日）。
4. **多实例部署需引入分布式锁**：同一份调度任务在两台机器上跑会重复入库。简单方案是 Redis `SETNX` 或 `apscheduler` 的 `SQLAlchemyJobStore`。

## 十、小结

本文用 APScheduler 搭了一条「自动触发 + 拉取 + 入库 + 去重 + 日志」的最小流水线。核心要点：

- **调度频率**：生产用 `cron`，演示用 `interval`
- **证书 fallback**：内部 → 演示，至少保证 demo 可复现
- **去重靠 SQLite 主键**：`PRIMARY KEY (date, stock)` + `INSERT OR IGNORE`
- **可追溯靠 task_log 表**：每次任务写一行，出问题能回放

下一篇（#03）我们解决「任务跑挂了没人知道」——给量化系统加监控告警：收益回撤异常、数据过期、任务心跳超时三类自动检测 + 邮件/桌面推送。

---

**免责声明**：本文仅供技术学习交流，不构成任何投资建议。量化策略回测表现不代表未来收益，投资有风险，决策需谨慎。

**代码与文档**：https://github.com/MaiRuiApi

系列前情：#01 我们选定了 SQLite 作为本地存储方案（单文件、零部署、查询快）。本篇解决「每天数据怎么自动进来」——告别手动 python main.py。

【Python量化系统工程实战 #02】每天手动拉数据太烦？用 APScheduler 搭一条「自动采集 + 增量去重」的流水线

编程语言

基于APScheduler构建股票数据自动入库方案：通过cron定时触发（每交易日15:30），实现API拉取、SQLite增量去重存储及任务日志追溯。内置证书fallback重试机制保障流水线稳定，搭配INSERT OR IGNORE复合主键确保数据幂等。生产环境建议配合systemd/Docker守护进程，实现无人值守量化数据管道。

代码解释

数据库

Windows

容器

SQLite

Linux

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

基于APScheduler构建股票数据自动入库方案：通过cron定时触发（每交易日15:30），实现API拉取、SQLite增量去重存储及任务日志追溯。内置证书fallback重试机制保障流水线稳定，搭配INSERT OR IGNORE复合主键确保数据幂等。生产环境建议配合systemd/Docker守护进程，实现无人...

【Python量化系统工程实战 #02】每天手动拉数据太烦？用 APScheduler 搭一条「自动采集 + 增量去重」的流水线-腾讯云开发者社区-腾讯云

【Python量化系统工程实战 #02】每天手动拉数据太烦？用 APScheduler 搭一条「自动采集 + 增量去重」的流水线

【Python量化系统工程实战 #02】每天手动拉数据太烦？用 APScheduler 搭一条「自动采集 + 增量去重」的流水线

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐