4. 数据清洗与预处理

任务目标

对获取的股票日线行情数据执行标准化清洗：处理缺失值、去除重复记录、统一日期与数值格式，并输出数据质量报告，确保后续策略计算使用的数据干净一致。

实现方案

使用 pandas 构建一个可复用的清洗流水线，覆盖以下核心环节：

• 缺失值处理：对价格类字段（开高低收）采用前向填充，成交量缺失则补 0（停牌日无成交）；若开盘价仍缺失则丢弃该行。
• 去重：按 股票代码 + 日期 去重，保留最后一条记录（应对盘中多次抓取）。
• 格式转换：日期统一为 YYYY-MM-DD，数值列转为 float64，股票代码转为字符串并补零到 6 位。
• 质量检查：输出每列的缺失计数、重复行数、日期范围及价格非正数检查结果。

完整代码

import pandas as pd
import numpy as np

def clean_stock_data(df: pd.DataFrame) -> pd.DataFrame:
    """
    对股票日线数据执行清洗。
    要求输入列：ts_code, trade_date, open, high, low, close, vol
    """
    # 复制避免修改原数据
    df = df.copy()

    # ---------- 1. 格式转换 ----------
    # 日期统一为 datetime 类型
    df['trade_date'] = pd.to_datetime(df['trade_date'], format='%Y%m%d')
    # 股票代码转为 6 位字符串
    df['ts_code'] = df['ts_code'].astype(str).str.zfill(6)
    # 数值列转为 float
    price_cols = ['open', 'high', 'low', 'close']
    for col in price_cols:
        df[col] = pd.to_numeric(df[col], errors='coerce')
    df['vol'] = pd.to_numeric(df['vol'], errors='coerce')

    # ---------- 2. 去重 ----------
    # 按代码+日期去重，保留最后一条
    df = df.drop_duplicates(subset=['ts_code', 'trade_date'], keep='last')

    # ---------- 3. 缺失值处理 ----------
    # 先按股票代码分组，组内按日期排序后前向填充价格
    df = df.sort_values(['ts_code', 'trade_date']).reset_index(drop=True)
    df[price_cols] = df.groupby('ts_code')[price_cols].ffill()
    # 成交量缺失补 0（停牌无成交）
    df['vol'] = df['vol'].fillna(0)
    # 若开盘价仍缺失（该股票首个交易日即缺失），删除该行
    df = df.dropna(subset=['open'])

    # ---------- 4. 质量检查 ----------
    print("=== 数据质量报告 ===")
    print(f"总行数: {len(df)}")
    print(f"股票数量: {df['ts_code'].nunique()}")
    print(f"日期范围: {df['trade_date'].min().date()} ~ {df['trade_date'].max().date()}")
    print("\n缺失值统计:")
    print(df.isnull().sum())
    print("\n重复行数:", df.duplicated(subset=['ts_code', 'trade_date']).sum())
    # 价格非正数检查
    invalid_price = df[(df[price_cols] <= 0).any(axis=1)]
    print(f"价格非正数行数: {len(invalid_price)}")
    # 检查 high >= low 且 close 在 [low, high] 区间内
    invalid_hl = df[(df['high'] < df['low']) | (df['close'] > df['high']) | (df['close'] < df['low'])]
    print(f"高低价逻辑异常行数: {len(invalid_hl)}")

    return df

# ---------- 最小可复现示例 ----------
if __name__ == "__main__":
    # 构造含脏数据的示例输入
    raw_data = pd.DataFrame({
        'ts_code': ['600000', '600000', '600000', '000001', '000001', '000001'],
        'trade_date': ['20240102', '20240102', '20240103', '20240102', '20240103', '20240104'],
        'open': [10.5, 10.8, None, 5.2, 5.4, 5.6],
        'high': [10.9, 11.0, 10.7, 5.5, 5.7, 5.9],
        'low': [10.2, 10.5, 10.1, 5.0, 5.2, 5.4],
        'close': [10.7, 10.6, 10.5, 5.3, 5.5, 5.8],
        'vol': [10000, None, 12000, 8000, 9000, None]
    })

    cleaned = clean_stock_data(raw_data)
    print("\n=== 清洗后数据 ===")
    print(cleaned.to_string(index=False))

运行与验证

环境依赖：pandas 与 numpy（pip install pandas numpy）。

运行方式：直接执行脚本，示例数据会经过完整清洗流程并打印质量报告。

通过条件：

• 质量报告中“重复行数”为 0；
• “缺失值统计”中所有列均为 0；
• “价格非正数行数”为 0；
• “高低价逻辑异常行数”为 0；
• 清洗后数据中 600000 在 20240102 的重复记录仅保留 1 条，且 open 缺失值被前向填充为 10.8（来自前一行），vol 缺失值被填充为 0。

预期输出关键行（由给定输入唯一确定）：清洗后共 5 行数据，其中 600000/20240102 仅保留最后一条（open=10.8），600000/20240103 的 open 被填充为 10.8，000001/20240104 的 vol 被填充为 0。