金融数据清洗那些坑：我的数据处理经验与避坑指南

import pandas as pd
import akshare as ak

# 获取A股日线数据（含复权因子）
def get_stock_data(code, start_date):
    df = ak.stock_zh_a_hist(symbol=code, adjust="hfq", start_date=start_date)
    # 处理东方财富接口的异常日期格式
    df['日期'] = pd.to_datetime(df['日期'], errors='coerce') 
    df = df.dropna(subset=['日期']).set_index('日期')
    return df

# 示例：获取贵州茅台数据
df = get_stock_data("600519", "20200101")

# 金融数据缺失的特殊场景：停牌、集合竞价无成交
def handle_missing(df):
    # 价格缺失：用前复权价向前填充
    price_cols = ['开盘', '最高', '最低', '收盘']
    df[price_cols] = df[price_cols].fillna(method='ffill')
    
    # 成交量缺失：填充0（停牌期间无交易）
    df['成交量'] = df['成交量'].fillna(0)
    return df

# 验证：检查停牌期间是否填充正确
assert df.loc['2023-02-01':'2023-02-10', '成交量'].sum() == 0, "停牌期成交量处理错误！"

import numpy as np

def detect_outliers(s, n=5):
    """改进版标准差法，解决涨停板干扰"""
    median = np.median(s)
    mad = np.median(np.abs(s - median))  # 中位数绝对偏差
    return ~((s > median + n*mad) | (s < median - n*mad))

# 应用：过滤日收益率异常值
daily_return = df['收盘'].pct_change()
valid_mask = detect_outliers(daily_return.dropna(), n=5)
clean_df = df[valid_mask.reindex(df.index, fill_value=True)]

# 识别真正需要去重的场景
duplicate_mask = df.duplicated(subset=['代码', '日期'], keep=False)
if duplicate_mask.sum() > 0:
    # 优先保留收盘价最接近成交均价的记录
    df['价差'] = (df['收盘'] - df['成交均价']).abs()
    df = df.sort_values('价差').drop_duplicates(['代码', '日期'], keep='first')

def handle_limit_up_down(df):
    # 识别涨停（close=high且收益率>=9.7%）
    df['is_limit_up'] = (df['收盘'] == df['最高']) & (df['收盘'].pct_change() >= 0.097)
    # 涨停日特征：用次日开盘价修正流动性偏差
    df['修正收盘价'] = np.where(df['is_limit_up'].shift(1), df['开盘'], df['收盘'])
    return df

from talib import RSI, BBANDS

# 计算技术指标（典型错误示范）
df['rsi'] = RSI(df['收盘'])           # 错误！未处理停牌期
df['upper'], _, _ = BBANDS(df['收盘']) # 错误！窗口包含未来数据

# 正确做法：滚动窗口计算
df['rsi'] = df.groupby('代码', group_keys=False)['收盘'].apply(
    lambda x: x.rolling(14, min_periods=5).apply(lambda x: RSI(x)[-1]))

# 最佳存储格式（Parquet + 分区）
df.to_parquet(
    path='stock_data/',
    partition_cols=['date_part=YYYYMM'],  # 按年月分区
    schema_version='v1.2',
    metadata={'author': 'Your Name', 'source': 'AKShare'}
)

# 处理某私募基金净值数据
raw_df = pd.read_excel("fund_data.xlsx", skiprows=3, parse_dates=['估值日期'])
clean_df = (
    raw_df
    .rename(columns={'估值日': 'date', '单位净值': 'nav'})
    .pipe(handle_missing)
    .pipe(handle_limit_up_down)
    .query("nav > 0")  # 剔除清算期异常值
    .sort_values('date')
)
clean_df.to_feather("cleaned_fund_data.feather")