

很多量化研究者的日常是这样的：

- Jupyter Notebook 里跑出一套漂亮的回测曲线。
- 兴致冲冲把代码复制到 `.py` 文件。
- 设了个 cron，每天跑一下。
- 两周后发现某天数据没更新，回测结果还是用的旧数据——**但脚本没报错**。

研究代码和能上线的工程代码之间，差的不只是"把 Notebook 另存为 .py"。最后一篇文章，我们用 200 行代码搭一个**最小可上线闭环**：配置化、质量门禁、自动落库、失败告警，全部可运行。

## 本文你将得到什么

1. **一个"研究→生产"的最小工程化骨架**：`pull -> quality_gate -> clean -> feature -> store -> alert`。
2. **配置驱动的设计**：股票列表、阈值、告警方式都抽成配置，不改代码就能切换策略。
3. **质量门禁前置**：数据不合格直接中断，不让脏数据进入特征和回测。
4. **可扩展的告警钩子**：预留 webhook 位置，接入企业微信/钉钉/飞书只需一行。

## 为什么 Notebook 不能直接跑生产

Notebook 是探索工具，不是工程工具。两者的核心差异：

| 维度 | Notebook | 生产脚本 |
|------|----------|----------|
| 配置 | 硬编码在 cell 里 | 独立配置文件 |
| 错误处理 | 中断手动修 | 自动重试/告警 |
| 状态 | 内存变量 | 数据库/checkpoint |
| 可观测 | 肉眼查看 | 日志 + 监控 |
| 可复现 | 依赖 cell 执行顺序 | 确定性入口函数 |

把研究代码变成生产代码，不需要一步到位用 Airflow，但至少要跨过上面 5 个坎。

## 最小闭环设计

整个闭环只有 6 个函数，每个职责单一：

```
pull()          # 从 API 拉取目标股票数据
quality_gate()  # 数据质量评分，不达标则告警并退出
clean()         # 清洗：去停牌、去异常值
feature()       # 计算特征：收益率、均线、波动
store()         # 落 SQLite，带唯一索引
alert()         # 失败/异常时写入告警日志
```

关键点：**`quality_gate` 必须放在 `feature` 之前**。很多新手把质量检查放在最后，结果发现脏数据已经污染了特征。

## 代码实战：从研究到上线的最小闭环

```python
import json
import sqlite3
import sys
import time
from datetime import datetime
from pathlib import Path

import pandas as pd
from mairui import Client

# 演示证书，请换成你自己的正式证书
client = Client("LICENCE-66D8-9F96-0C7F0FBCD073")

DB_PATH = Path(__file__).parent / "production.db"
ALERT_LOG = Path(__file__).parent / "alerts.jsonl"

CONFIG = {
    "codes": ["600519", "000001", "000002"],
    "lookback": 60,
    "quality_thresholds": {
        "completeness": 99.0,
        "coverage": 100.0,
        "validity": 95.0,
        "consistency": 99.0,
    },
    "alert_webhook": None,  # 生产环境填 webhook 地址
}


def send_alert(message: str):
    """模拟告警：写入本地日志；生产环境可替换为真实 webhook。"""
    entry = {"ts": datetime.now().isoformat(), "message": message}
    with ALERT_LOG.open("a", encoding="utf-8") as f:
        f.write(json.dumps(entry, ensure_ascii=False) + "\n")
    print(f"[alert] {message}")


def pull(codes, lt):
    frames = []
    for code in codes:
        raw = client.stock_history(code, lt=lt, dividend="n")
        df = pd.DataFrame(raw)
        if df.empty:
            continue
        df["t"] = pd.to_datetime(df["t"])
        df["code"] = code
        for col in ["o", "c", "h", "l", "v"]:
            df[col] = pd.to_numeric(df[col], errors="coerce")
        frames.append(df[["code", "t", "o", "c", "h", "l", "v"]])
    return pd.concat(frames, ignore_index=True) if frames else pd.DataFrame()


def quality_gate(df, codes, thresholds):
    total = len(df)
    comp = 100.0 if total == 0 else round(
        (1 - df[["o", "c", "h", "l", "v"]].isna().sum().sum() / (total * 5)) * 100, 2
    )
    cov = round(len(set(df["code"]) & set(codes)) / len(codes) * 100, 2) if codes else 0.0

    df["ret"] = df.groupby("code")["c"].pct_change() * 100
    valid = round(df["ret"].between(-20, 20).sum() / df["ret"].notna().sum() * 100, 2)
    cons = round(
        ((df["h"] >= df["l"]).sum() + df["c"].between(df["l"], df["h"]).sum()) / (total * 2) * 100, 2
    )

    scores = {"completeness": comp, "coverage": cov, "validity": valid, "consistency": cons}
    for dim, thr in thresholds.items():
        if scores.get(dim, 0) < thr:
            send_alert(f"{dim} 评分 {scores.get(dim, 0)} 低于阈值 {thr}")
            return False, scores
    return True, scores


def clean(df):
    before = len(df)
    df = df[df["v"] > 0].copy()
    df = df.dropna(subset=["o", "c", "h", "l", "v"])
    print(f"[clean] {before} -> {len(df)} 行")
    return df


def feature(df):
    df = df.sort_values(["code", "t"]).copy()
    df["ret_1"] = df.groupby("code")["c"].pct_change()
    df["ma_5"] = df.groupby("code")["c"].transform(lambda x: x.rolling(5).mean())
    df["vol_5"] = df.groupby("code")["ret_1"].transform(lambda x: x.rolling(5).std())
    return df


def store(df, db_path):
    if db_path.exists():
        db_path.unlink()
    con = sqlite3.connect(db_path)
    df2 = df.copy()
    for c in df2.columns:
        if pd.api.types.is_datetime64_any_dtype(df2[c]):
            df2[c] = df2[c].dt.strftime("%Y-%m-%d")
    df2.to_sql("kline_features", con, if_exists="replace", index=False)
    con.execute("CREATE UNIQUE INDEX IF NOT EXISTS idx_code_t ON kline_features(code, t)")
    con.close()
    print(f"[store] {len(df)} 行写入 {db_path}")


def run_once():
    print(f"\n[start] 生产闭环 @ {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    t0 = time.time()

    df = pull(CONFIG["codes"], CONFIG["lookback"])
    ok, scores = quality_gate(df, CONFIG["codes"], CONFIG["quality_thresholds"])
    print(f"[quality] {scores}")
    if not ok:
        print("[abort] 质量门禁未通过，下游不执行")
        return

    df_clean = clean(df)
    df_feat = feature(df_clean)
    store(df_feat, DB_PATH)

    elapsed = round(time.time() - t0, 2)
    print(f"[done] elapsed={elapsed}s, rows={len(df_feat)}")


def main():
    run_once()
    print("\n[note] 生产环境用 schedule/cron 包装 run_once()，每日收盘后执行。")


if __name__ == "__main__":
    main()
```

运行输出示例：

```
[start] 生产闭环 @ 2026-08-30 10:10:36
[quality] {'completeness': 100.0, 'coverage': 100.0, 'validity': 100.0, 'consistency': 100.0}
[clean] 150 -> 150 行
[store] 150 行写入 production.db
[done] elapsed=0.32s, rows=150

[note] 生产环境用 schedule/cron 包装 run_once()，每日收盘后执行。
```

产物 `production.db` 里会有一张 `kline_features` 表，带 `(code, t)` 唯一索引，可以作为下游回测/模型的输入。

## 把它挂到每日调度

最小闭环本身不依赖任何调度框架。生产环境推荐三种方式：

**方式 1：schedule（跨平台，适合本地/服务器）**

```python
import schedule
import time

schedule.every().day.at("16:30").do(run_once)

while True:
    schedule.run_pending()
    time.sleep(60)
```

**方式 2：cron / systemd（Linux 服务器）**

```bash
# cron 示例：每天 16:30 执行
30 16 * * 1-5 /path/to/venv/bin/python /path/to/main.py >> /var/log/quant/run.log 2>&1
```

**方式 3：Windows 任务计划程序**

定时执行 `python main.py` 即可。

## 从最小闭环到完整系统的升级路径

这个闭环只有 200 行，但它预留了清晰的扩展点：

- **配置升级**：把 `CONFIG` 移到 `config.yaml`，支持多策略、多数据源。
- **checkpoint 升级**：每个阶段输出 Parquet，失败可从中间阶段恢复。
- **告警升级**：把 `send_alert` 替换为真实 webhook，分级告警（P0 电话 / P1 短信 / P2 群消息）。
- **存储升级**：SQLite 适合本地研究，生产可切 MySQL/PostgreSQL/ClickHouse。
- **调度升级**：单机 schedule → Airflow / Prefect / Dagster。

## 常见坑

- **不要跳过质量门禁**：很多人为了"跑通"先把 `quality_gate` 注释掉，这是埋下大坑。宁可今天不跑，也别跑脏数据。
- **不要把证书写进配置文件**：演示证书可以明文出现在学习代码里，但正式证书一定要走环境变量或密钥管理服务。
- **数据库要先删再建？** 教学版为了可复现每次重建数据库；生产环境应该用 `INSERT OR REPLACE` 做幂等更新，参考本系列 #04。

## 系列小结

《Python 量化数据工程实战》到这里就收官了。10 篇文章覆盖了一条完整的数据流水线：

1. **清洗**：停牌、除权、异常值处理
2. **特征**：动量、波动、资金流因子
3. **验证**：交叉校验、逻辑审计、前后视偏差
4. **增量更新**：只拉新数据，避免全量重跑
5. **多源合并**：行情 + 基本面 + 公告的统一时间轴
6. **复权**：前复权/后复权的工程实现
7. **特征筛选**：相关性去重 + RFE
8. **流水线骨架**：6 阶段可观测工作流
9. **质量评分卡**：5 维度量化监控
10. **上线闭环**：从 Notebook 到每日自动运行

数据工程没有银弹，但有一个核心原则：**让数据可靠、可观测、可重跑**。掌握这套模式，你的研究脚本就具备了工业化基础。

**系列结束，但工程化实践才刚刚开始**。建议下一步把这套闭环套到你的实际策略上跑一周，观察评分卡和告警，再逐步扩展成完整系统。

---

## 免责声明

本文仅供技术学习交流，不构成任何投资建议。市场有风险，决策需谨慎。

---

## 了解更多

完整示例代码与文档：<https://github.com/MaiRuiApi>


研究代码和能上线的工程代码之间，差的不只是"把 Notebook 另存为 .py"。最后一篇文章，我们用 200 行代码搭一个最小可上线闭环：配置化、质量门禁、自动落库、失败告警，全部可运行。

【Python量化数据工程实战 #10】研究脚本能直接上线吗？从 Notebook 到每日自动跑的最小闭环

编程语言

基于200行Python代码构建最小可上线量化数据闭环，覆盖数据拉取、质量门禁、清洗、特征计算、SQLite落库及告警全流程。配置驱动设计支持灵活切换策略，质量门禁前置防止脏数据污染特征，预留Webhook告警接口，从Notebook脚本升级为生产级每日自动运行系统。

密钥管理

代码解释

PostgreSQL

数据库

服务器

数据流

Windows

Python

短信

SQLite

Linux

MySQL

2026上云采购 | AI焕新·智启新局

tchousec

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

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

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

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

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

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

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

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

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

社区新版编辑器体验调研

【Python量化数据工程实战 #10】研究脚本能直接上线吗？从 Notebook 到每日自动跑的最小闭环-腾讯云开发者社区-腾讯云

【Python量化数据工程实战 #10】研究脚本能直接上线吗？从 Notebook 到每日自动跑的最小闭环

【Python量化数据工程实战 #10】研究脚本能直接上线吗？从 Notebook 到每日自动跑的最小闭环

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐