

你做过几个数据项目后，是不是经常在这种场景里挣扎：

- 数据来了用 pandas 拼一下，写到 SQLite，跑了 —— **三天后重启脚本发现 row 又重了一倍**。
- 突然发现某天的数据漏了，但不知道是拉漏了还是合并时丢了 —— **没有监控，你永远不知道数据在哪一步坏的**。
- 加新因子要改三段代码、改三个 SQL、改三个 schema —— **流水线越来越像"打补丁"**。

这篇文章给你一套 **200 行的"端到端数据流水线骨架"**：把前 6 篇的方法（清洗 / 增量 / 合并 / 复权 / 验证 / 筛选）整合成一个 6 阶段的工作流——**拉取 → 入库 → 清洗 → 合并 → 特征 → 监控**——每个阶段独立可测、可独立重跑、留监控报告。读完你能直接复制这套骨架，把它当作模板套到你的实际数据项目里。

## 本文你将得到什么

1. **6 阶段流水线设计模式**：`pull → store → clean → merge → feat → monitor`
2. **每个阶段的最小可用代码**：200 行级，覆盖「拉 - 存 - 清 - 合 - 算 - 监」完整闭环
3. **3 个工程化铁律**：复合主键幂等、阶段独立可重跑、监控报告必落盘
4. **怎么把骨架套到生产**：每个阶段改成"按调度运行 + 失败发告警"

## 一句话讲清流水线骨架的设计哲学

> **流水线骨架 = 「每阶段只做一件事、阶段之间靠 artifact 衔接、每阶段都留监控输出」。**

这句话有三个关键词：

- **只做一件事**：阶段函数职责单一，别一个函数里又拉又清又合
- **靠 artifact 衔接**：阶段之间的输入输出是 DataFrame 或数据库表，不是全局变量
- **留监控输出**：每个阶段结束都输出一份"自检报告"（行数 / NaN 占比 / schema 检查）

这样设计的好处是：**某阶段出错时，你能精确知道是哪一阶段坏的**，而不是看到下游爆了却无从下手。

## 代码实战：6 阶段骨架

下面这段代码演示完整的 6 阶段流水线。骨架可直接复制，按你的数据源改各 stage 函数即可。

```python
import sqlite3
from pathlib import Path
import numpy as np
import pandas as pd
from mairui import Client

# ===== 全局：演示证书 + 数据库路径 =====
client = Client("LICENCE-66D8-9F96-0C7F0FBCD073")  # 演示证书，请换成你自己的正式证书
DB_PATH = Path(__file__).parent / "pipeline.db"

# ===== 1) PULL：从 API 拉取 =====
def stage_pull(codes, lt=80):
    klines, profits, anns = [], [], []
    for code in codes:
        raw = client.stock_history(code, lt=lt, dividend="n")
        df = pd.DataFrame(raw); df["t"] = pd.to_datetime(df["t"]); df["code"] = code
        klines.append(df)
        # 财务 + 公告 ...
    return pd.concat(klines, ignore_index=True)  # 同样处理 profits/anns

# ===== 2) STORE：用复合 PRIMARY KEY 落 SQLite（幂等） =====
def _create_with_pk(con, table, df, pk_cols):
    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 %H:%M:%S")
    placeholders = ", ".join(["?"] * len(df2.columns))
    col_list    = ", ".join([f'"{c}"' for c in df2.columns])
    type_map    = {c: "TEXT" if (pd.api.types.is_datetime64_any_dtype(df[c]) or pd.api.types.is_object_dtype(df2[c])) else "REAL" for c in df2.columns}
    schema_cols = ", ".join([f'"{c}" {type_map[c]}' for c in df2.columns])
    pk_clause   = ", ".join([f'"{c}"' for c in pk_cols])
    con.execute(f'DROP TABLE IF EXISTS "{table}"')
    con.execute(f'CREATE TABLE "{table}" ({schema_cols}, PRIMARY KEY ({pk_clause}))')
    con.executemany(f'INSERT OR REPLACE INTO "{table}" ({col_list}) VALUES ({placeholders})', df2.values.tolist())

def stage_store(df_k, df_p, df_a, db_path=DB_PATH):
    if db_path.exists(): db_path.unlink()
    con = sqlite3.connect(db_path)
    _create_with_pk(con, "kline", df_k[["code","t","o","c","h","l","v"]], ["code","t"])
    _create_with_pk(con, "profit", df_p[["code","date","profit","income"]], ["code","date"])
    if not df_a.empty:
        _create_with_pk(con, "announcement", df_a[["code","t"]], ["code","t"])
    con.close()

# ===== 3) CLEAN：剔除停牌 + 3σ异常 =====
def stage_clean(df_k):
    df = df_k.copy()
    df["ret"] = df.groupby("code")["c"].pct_change()
    before = len(df)
    df = df[df["v"] > 0]                                # 停牌日：成交量 0
    thr = df["ret"].std() * 3
    df = df[df["ret"].between(-thr, thr)]               # 3σ 异常剔除
    print(f"[clean] {before} -> 停牌剔除 -> 3σ剔除 -> {len(df)}")
    return df

# ===== 4) MERGE：统一时间轴 =====
def stage_merge(df_k_clean, df_p, df_a):
    out = df_k_clean[["t","code","c","v"]].sort_values("t")
    df_p2 = df_p.rename(columns={"date":"t"})[["code","t","profit"]]
    parts = []
    for code, g in out.groupby("code"):
        p = df_p2[df_p2["code"]==code].sort_values("t")
        parts.append(pd.merge_asof(g, p[["t","profit"]], on="t", direction="backward"))
    out = pd.concat(parts, ignore_index=True)
    if not df_a.empty:
        cnt = df_a.groupby(["code","t"]).size().reset_index(name="ann_count")
        out = out.merge(cnt, on=["code","t"], how="left")
        out["ann_count"] = out["ann_count"].fillna(0).astype(int)
    return out

# ===== 5) FEAT：动量 / 波动因子 =====
def stage_feat(df_merged):
    df = df_merged.copy()
    df["ret_1"]    = df.groupby("code")["c"].pct_change(1)
    df["ret_5"]    = df.groupby("code")["c"].pct_change(5)
    df["vol_5"]    = df.groupby("code")["ret_1"].rolling(5).std().reset_index(level=0, drop=True)
    df["vol_20"]   = df.groupby("code")["ret_1"].rolling(20).std().reset_index(level=0, drop=True)
    df["vol_chg_5"] = df.groupby("code")["v"].pct_change(5)
    return df

# ===== 6) MONITOR：schema + NaN 占比 =====
def stage_monitor(df_final):
    expected = {"t","code","c","v","profit","ann_count","ret_1","ret_5","vol_5","vol_20","vol_chg_5"}
    missing = expected - set(df_final.columns)
    assert not missing, f"schema mismatch: {missing}"
    nan_pct = (df_final.isna().mean() * 100).round(1)
    rep = {"rows": len(df_final), "codes": df_final["code"].nunique(),
           "nan_pct_each_col": nan_pct.to_dict(), "schema_check": "PASS"}
    Path(__file__).parent.joinpath("monitor_report.json").write_text(
        __import__("json").dumps(rep, ensure_ascii=False, indent=2), encoding="utf-8")
    return rep

# ===== MAIN =====
def main():
    codes = ["600519","000001","000002","600036"]
    df_k, df_p, df_a = stage_pull(codes, lt=80)
    stage_store(df_k, df_p, df_a)
    df_clean  = stage_clean(df_k)
    df_merged = stage_merge(df_clean, df_p, df_a)
    df_final  = stage_feat(df_merged)
    rep       = stage_monitor(df_final)
    print(rep)

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

跑完会输出：

```
[clean] 200 -> 停牌剔除 -> 3σ剔除 -> 196
{rows: 196, codes: 4, schema_check: 'PASS',
 nan_pct: {ret_1: 2.0, ret_5: 10.2, vol_5: 10.2, vol_20: 40.8, ...}}
```

200 行不到的骨架，已经覆盖了完整的数据工程闭环。

## 核心设计：3 个工程化铁律

### 铁律 1：复合 PRIMARY KEY 落库天然幂等

上一篇文章讲的"增量更新"很容易踩一个坑：脚本重复跑时数据翻倍。**复合 PRIMARY KEY + INSERT OR REPLACE** 是最简的解决方案：

```sql
CREATE TABLE kline (
    code TEXT, t TEXT,
    o REAL, c REAL, h REAL, l REAL, v REAL,
    PRIMARY KEY (code, t)
);
INSERT OR REPLACE INTO kline (...) VALUES (...);
```

- `(code, t)` 是天然的"唯一键"——一只股票一天只能有一行
- `INSERT OR REPLACE` 遇到重复 PK 自动覆盖，不会新增行
- 重跑脚本不会爆数据，永不爆

### 铁律 2：每阶段独立可重跑

骨架里的每个 `stage_X` 都接受输入 + 产 artifact，要重跑「特征阶段」只需要重新构造 DataFrame；要重跑「入库阶段」只需要 DROP + CREATE。**每个阶段都可以独立调试、独立单元测试**。

实战中这意味着：当某天发现监控告警爆了，能精确知道是哪一阶段挂了：

| 监控告警 | 几乎一定是 |
|----------|------------|
| schema_check FAIL | `feat` 阶段代码改了字段名 |
| NaN 占比突增 | `pull` 阶段某接口超时部分 row 为空 |
| 行数暴跌 | `clean` 阶段阈值过严，剔太多 |
| 财务列 NaN 大增 | `merge` 阶段 merge_asof 方向错了 |

### 铁律 3：监控报告必落盘

最后阶段的 `stage_monitor` 看似可有可无，**实际上是最重要的**。它的输出（`monitor_report.json`）会进版本控制 / 写监控 dashboard / 触发告警——是整套流水线的"心电图"。

最实用的几个监控指标：

- **行数**：每个 stage 的输出行数（`200 → 200 → 196 → 196 → 196`）
- **NaN 占比**：每列 NaN 比例（rolling 起点/复权缺失/接口超时都会反映）
- **schema 检查**：列名 + 数据类型是否符合预期
- **耗时**：每阶段用时（用以发现性能退化）

## 怎么把骨架套到生产

骨架是教学版（4 只股票 × 80 天），生产环境需要改 3 个地方：

### 1) stock list 从配置文件读

```python
import yaml
codes = yaml.safe_load(open("config/codes.yaml"))["stocks"]
```

### 2) 阶段之间加 checkpoint

```python
df_k.to_parquet(checkpoint_dir / "stage1_kline.parquet")
df_clean.to_parquet(checkpoint_dir / "stage3_clean.parquet")
```

每个阶段写一份 Parquet 文件到 `checkpoint/`，重跑时先检查 checkpoint 是否有效（mtime + 行数），避免无谓重跑。

### 3) 监控升级：加告警

把 `monitor_report.json` 接到告警系统（飞书 / 钉钉 / 企微）：

```python
if any(v > 50 for k, v in rep["nan_pct_each_col"].items()):
    send_alert(f"WARN: {k} NaN 占比 = {v}%")
```

最后用 Airflow / schedule / cron 把整套调度起来，按"日终一次 / 盘中每 5 分钟"两种节奏跑。

## 小结

数据工程的本质不是写多花哨的代码，而是 **让数据可靠、可观测、可重跑**。记住三件事：

1. **复合 PK 落库**：幂等性是自动的，不靠手动去重
2. **每阶段独立 artifact**：故障可定位、阶段可重跑、单元可测试
3. **监控必落盘**：没有监控的流水线都是定时炸弹

掌握这套模式，你的"写一次跑一次"脚本就能升级成"全市场每日自动 + 失败告警"的工业级数据系统。

**下篇预告**：流水线骨架有了，但怎么知道数据本身"质量好不好"？#09 带你搭一套**数据质量评分卡**：用完整性、及时性、一致性、可信度四个维度量化监控每个数据源，问题早发现早处理。

---

## 免责声明

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

---

## 了解更多

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


数据来了用 pandas 拼一下，写到 SQLite，跑了 —— 三天后重启脚本发现 row 又重了一倍。

【Python量化数据工程实战 #08】从拉取到入库到特征：一键数据流水线骨架

编程语言

大数据

后端

一套200行的端到端数据流水线骨架，整合拉取、入库、清洗、合并、特征与监控6阶段，覆盖复合主键幂等、阶段独立可重跑、监控落盘三大工程铁律，提供可复制的模板代码，助力数据工程从脚本升级为可观测、可告警的工业级系统。

单元测试

代码解释

数据库

数据流

SQLite

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

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

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

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

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

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

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

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

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

社区新版编辑器体验调研

【Python量化数据工程实战 #08】从拉取到入库到特征：一键数据流水线骨架-腾讯云开发者社区-腾讯云

【Python量化数据工程实战 #08】从拉取到入库到特征：一键数据流水线骨架

【Python量化数据工程实战 #08】从拉取到入库到特征：一键数据流水线骨架

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐