1. 数据获取与清洗

任务目标

构建一个可复现的股票日线数据管道：从公开数据源获取指定股票的日线行情，完成缺失值与异常值清洗，输出标准化格式的CSV文件，为后续量化分析提供可靠数据基础。本案例以贵州茅台（600519.SH）2024年1月1日至2024年3月31日数据为例。

环境与依赖

需要Python 3.9+，安装以下库：

pip install akshare pandas numpy

akshare为开源财经数据接口库，若网络受限可替换为本地CSV数据源，但需保持字段名一致。

完整实现代码

import akshare as ak
import pandas as pd
import numpy as np
from datetime import datetime

def fetch_stock_data(symbol: str, start_date: str, end_date: str) -> pd.DataFrame:
    """从akshare获取股票日线数据（前复权）"""
    df = ak.stock_zh_a_hist(
        symbol=symbol,
        period="daily",
        start_date=start_date.replace("-", ""),
        end_date=end_date.replace("-", ""),
        adjust="qfq"
    )
    # 统一列名为英文标准格式
    df.rename(columns={
        "日期": "date", "开盘": "open", "收盘": "close",
        "最高": "high", "最低": "low", "成交量": "volume",
        "成交额": "amount", "振幅": "amplitude",
        "涨跌幅": "pct_change", "涨跌额": "change",
        "换手率": "turnover"
    }, inplace=True)
    df["date"] = pd.to_datetime(df["date"])
    return df[["date", "open", "high", "low", "close", "volume", "amount"]]

def clean_data(df: pd.DataFrame) -> pd.DataFrame:
    """清洗缺失值与异常值"""
    df = df.sort_values("date").reset_index(drop=True)
    # 1. 删除全空行
    df = df.dropna(how="all")
    # 2. 数值列缺失值用前向填充（交易日数据连续性）
    numeric_cols = ["open", "high", "low", "close", "volume", "amount"]
    df[numeric_cols] = df[numeric_cols].fillna(method="ffill")
    # 若仍存在缺失（如首行），用后向填充兜底
    df[numeric_cols] = df[numeric_cols].fillna(method="bfill")
    # 3. 异常值清洗：价格非正、high<low、high<max(open,close)等逻辑错误
    df = df[
        (df["high"] >= df["low"]) &
        (df["high"] >= df[["open", "close"]].max(axis=1)) &
        (df["low"] <= df[["open", "close"]].min(axis=1)) &
        (df["open"] > 0) & (df["close"] > 0) &
        (df["volume"] >= 0)
    ].copy()
    # 4. 去除重复日期
    df = df.drop_duplicates(subset="date", keep="last")
    return df.reset_index(drop=True)

def save_standard_csv(df: pd.DataFrame, output_path: str) -> None:
    """保存为标准格式CSV（日期索引，数值保留4位小数）"""
    df = df.set_index("date")
    df.to_csv(output_path, float_format="%.4f")
    print(f"数据已保存至: {output_path}, 共 {len(df)} 条记录")

if __name__ == "__main__":
    SYMBOL = "600519"      # 贵州茅台
    START = "2024-01-01"
    END = "2024-03-31"
    OUTPUT = "stock_daily_600519.csv"
    
    raw_df = fetch_stock_data(SYMBOL, START, END)
    print(f"原始数据获取: {len(raw_df)} 条")
    
    cleaned_df = clean_data(raw_df)
    print(f"清洗后数据: {len(cleaned_df)} 条")
    
    save_standard_csv(cleaned_df, OUTPUT)
    # 控制台预览前3行
    print(cleaned_df.head(3).to_string())

运行方式

python stock_pipeline.py

验证方法与通过条件

运行脚本后，按以下标准验证管道正确性：

  1. 数据完整性：输出CSV文件存在且非空，记录数在50~65之间（2024年一季度约58个交易日，停牌日除外）。
  2. 格式标准化：CSV首列为date索引，格式为YYYY-MM-DD；数值列包含open/high/low/close/volume/amount，价格保留4位小数。
  3. 清洗有效性：控制台打印的清洗前后记录数对比中，清洗后数量≤清洗前数量；若原始数据无缺失/异常，则数量相等。
  4. 逻辑自洽性：抽查CSV中任意一行，满足low ≤ min(open, close) ≤ max(open, close) ≤ high，且所有价格为正数。
  5. 数据真实性：对比2024年1月2日（首个交易日）收盘价，应约为1685元（前复权），与实际行情一致。

若以上条件全部满足，则数据管道构建成功，可用于后续策略回测或因子计算。

常见问题

  • 网络超时：akshare依赖东方财富接口，若请求失败，重试或更换start_date为近30日数据测试连通性。
  • 字段缺失：若接口返回缺少振幅等列，代码已仅保留核心6列，不影响运行。
  • 停牌日处理：停牌日无数据记录，属正常现象，无需填充。
← 上一章 📖 返回目录 下一章 →