📚 课程目录第3章开发环境与工具链

3. 开发环境与工具链

金融数据清洗流水线的第一道“隐形工序”，不是写清洗逻辑，而是把开发环境与工具链搭稳。环境不稳，后续的缺失值填充、去重、对账、复权、时区对齐都会变成“玄学问题”。本节围绕 Python 生态、Pandas、NumPy、SQL、Airflow、Docker 以及项目目录结构，给出一套可直接落地的工程化搭建方案。

一、核心原理：为什么金融清洗需要“工具链”而非“脚本”

金融数据清洗的典型特征是：数据源多（行情、财报、公告、交易流水）、频率杂（Tick、分钟、日频、季度）、质量参差（缺失、重复、异常值、口径不一致）、合规要求高（可追溯、可复现）。单文件脚本能跑通一次，但无法保证每天稳定运行。工具链的本质，是把“人肉操作”转化为“可调度、可隔离、可观测”的工程能力。

Python 生态：统一语言层，连接数据读取、计算、调度与测试。
Pandas / NumPy：承担表格级清洗与向量化数值计算，是清洗逻辑的主战场。
SQL：承担大数据量下的过滤、聚合、关联与落库，减少内存压力。
Airflow：把清洗任务编排为 DAG，实现依赖管理、重试与回填。
Docker：锁定 Python、依赖库与系统环境版本，保证“本地能跑、线上也能跑”。

  核心原则：计算归计算，调度归调度，环境归环境。三者解耦，清洗逻辑才能被测试和复用。

二、方法流程：从零搭建可运行环境

建议按“依赖锁定 → 目录规划 → 容器化 → 调度接入”的顺序推进。下面给出关键步骤与命令。

创建项目骨架与虚拟环境，锁定依赖版本。
按分层目录组织代码、配置、数据与测试。
编写 Dockerfile 与 docker-compose，统一运行环境。
接入 Airflow，将清洗步骤定义为 DAG 任务。

# 1. 创建虚拟环境并安装核心依赖
python -m venv .venv
source .venv/bin/activate        # Windows: .venv\Scripts\activate
pip install pandas numpy sqlalchemy psycopg2-binary apache-airflow pytest

# 2. 锁定依赖，保证环境可复现
pip freeze > requirements.txt

依赖锁定后，目录结构决定了项目能否长期维护。推荐如下分层：

fin_clean_pipeline/
├── dags/                  # Airflow DAG 定义
│   └── daily_clean_dag.py
├── src/
│   ├── extract/           # 数据抽取：行情、财报、流水
│   ├── transform/         # 清洗逻辑：缺失、去重、对齐
│   ├── load/              # 落库与导出
│   └── utils/             # 日志、配置、校验工具
├── sql/                   # 清洗与聚合 SQL
├── configs/               # YAML/ENV 配置
├── data/
│   ├── raw/               # 原始数据（只读）
│   ├── interim/           # 中间结果
│   └── processed/         # 清洗后数据
├── tests/                 # 单元测试与数据质量测试
├── Dockerfile
├── docker-compose.yml
└── requirements.txt

建议将 raw 目录设为只读挂载，任何清洗都写入 interim 或 processed，避免“污染源数据”导致无法回溯。

三、案例/场景：一条日频行情清洗链路

假设每天 18:00 需要清洗当日 A 股日频行情：从 CSV 抽取 → 去重与缺失处理 → 复权因子对齐 → 写入 PostgreSQL。用 Airflow 定义 DAG，核心任务如下：

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

with DAG(
    dag_id="daily_market_clean",
    start_date=datetime(2024, 1, 1),
    schedule="0 18 * * *",
    catchup=False,
    tags=["finance", "clean"],
) as dag:
    extract = PythonOperator(task_id="extract", python_callable=run_extract)
    transform = PythonOperator(task_id="transform", python_callable=run_transform)
    load = PythonOperator(task_id="load", python_callable=run_load)
    extract >> transform >> load

Docker 侧用 compose 同时拉起 Airflow 与数据库，保证调度与存储在同一网络内：

services:
  postgres:
    image: postgres:15
    environment:
      POSTGRES_USER: fin
      POSTGRES_PASSWORD: fin123
      POSTGRES_DB: market
    ports: ["5432:5432"]
  airflow:
    build: .
    depends_on: [postgres]
    ports: ["8080:8080"]

清洗逻辑中，Pandas 负责表格级处理，NumPy 负责向量化数值运算，SQL 负责大表过滤与落库：

import pandas as pd
import numpy as np

df = pd.read_csv("data/raw/market_20240101.csv")
df = df.drop_duplicates(subset=["symbol", "trade_date"])
df["close"] = df["close"].replace(0, np.nan)
df["close"] = df["close"].ffill()
df.to_sql("market_clean", engine, if_exists="append", index=False)

四、误区与复盘：环境搭建中最容易踩的坑

误区	表现	正确做法
依赖不锁版本	本地能跑，线上报错	使用 requirements.txt 或 poetry.lock 锁定
数据目录混用	原始数据被覆盖，无法回溯	raw / interim / processed 三层隔离
清洗逻辑写在 DAG 里	无法单测，DAG 臃肿	逻辑放 src，DAG 只做编排
忽略时区与交易日历	跨市场数据错位	统一 UTC 存储，交易日历单独维护
无数据质量校验	脏数据静默入库	在 load 前增加断言与质量报告

不要用“先跑起来再说”替代环境设计。金融清洗一旦上线，回填、重跑、审计都会放大环境缺陷的代价。

完成本节搭建后，你应得到一个可复现、可调度、可测试的清洗工程底座。后续所有清洗规则、质量校验与回填策略，都将在这个底座上迭代，而不是散落在临时脚本里。