上一篇我们用 akshare 拉到了行情数据，但你有没有想过一个问题：每次跑代码都重新去网上拉一遍，又慢又不稳，万一哪天接口抽风，你的策略研究就全停摆了。成熟的做法是：数据拉一次，洗干净，存到本地，以后随取随用。这一步听起来不像「量化」那么酷，却是所有靠谱量化工作的地基——数据错一步，后面全错。
本篇做两件事：第一，处理真实数据的三大脏问题——缺失值、重复值、异常值，外加让新人头疼的类型转换；第二，把清洗好的数据分别存成 CSV、SQLite 和 Parquet 三种格式，跑通各自的存和取，最后用一张对比表帮你记住「什么场景该用哪个」。上一篇我们拉的是 2020-01-01 至 2024-12-31 的日行情，本篇的清洗和存储就接着用这份数据。
本篇目标
学完本篇，你能够做到：
- 用
isna定位缺失值，并根据场景选择fillna、ffill或dropna处理，说清楚三者的适用条件。 - 用
duplicated找重复行、用涨跌幅阈值筛查极端异常值，并决定「删除还是保留」。 - 把日期列转成 datetime、价格列转成 float，理解
dtype错误是如何悄悄污染计算的。 - 分别用 CSV、SQLite（
sqlite3+pd.read_sql）和 Parquet（需pyarrow）完成行情数据的落盘与读回。 - 面对一份新数据，能独立走完「检查 → 清洗 → 存储哪种格式」的决策流程。
核心概念
缺失值：先问「为什么缺」，再决定「怎么补」
数据缺一格，原因可能完全不同：新股上市首日没有昨收、节假日接口返回空行、网络抖动导致某天没拉到。处理之前先看「缺在哪」，再选手段：
df.isna().sum()：每列缺失多少个，永远是处理缺失值的第一步。fillna(值)：用固定值填，适合「确实该是这个值」的场景，比如用 0 填「无成交」。ffill()（forward fill，前向填充）：用前一个交易日的值向后补。行情里最常用——停牌期间价格按停牌前最后价处理，符合直觉。dropna()：直接删掉含缺失的行。适合缺失占比很小、且补了反而误导的场景。
划重点：价格类缺失优先考虑
ffill，但成交量类缺失慎用ffill——停牌日成交量是 0 而不是「沿用昨天的量」，直接填 0 才符合事实。补什么值取决于业务含义，没有万能公式。
重复值与异常值：一个全删，一个慎删
重复行多半是爬取时「同一接口请求了两次」造成的，用 df[df.duplicated()] 找出来，df.drop_duplicates() 删掉即可，几乎没有争议。
异常值就复杂了。A 股有涨跌停制度，主板单日涨跌幅限制在 ±10%（ST 股 ±5%，创业板/科创板 ±20%），所以「涨跌幅超出制度限制」的记录大概率是数据错误；但如果只是「跌幅接近跌停」，那可能是真实行情。本篇教一个保守的筛查思路：把「超过制度可能上限」的记录找出来人工复核，而不是自动删除——删错了真实的大涨大跌，你的回测就错过了最重要的行情。
类型转换：看不见的坑最多
从 CSV 读回来的数据最容易出类型问题：日期变成了字符串（object）、价格混进一个 "16.5 " 带空格的脏值导致整列变成字符串。判断依据是 df.dtypes。常用手段：
pd.to_datetime(df["date"])：字符串转日期。pd.to_numeric(df["close"], errors="coerce")：转数字，转不了的脏值（比如"--"）会被置为 NaN，交给缺失值流程处理，比直接报错友好得多。df.astype({"volume": "int64"})：整列改指定类型，适合确认干净后的收尾。
三种存储格式，三种性格
- CSV（Comma-Separated Values，逗号分隔值文本）：任何软件都能打开的「普通话」，人类可读，Excel 也能开；缺点是读写慢、文件大、不保存 dtype——日期存进去再读回来就变字符串，类型全靠读的时候重新转。
- SQLite：一个文件就是一个迷你数据库，用 SQL 查询，支持按条件取数（
WHERE close > 20），多只股票放一张表里管理很方便；缺点是存取速度一般，需要写一点 SQL。 - Parquet：列式存储的二进制格式（需安装
pyarrow），保存 dtype、压缩率高、读写快，是量化数据落地的首选；缺点是文本编辑器打不开，「人不可读」。
完整代码
安装命令（pyarrow 只在 Parquet 部分用到，一并装上）：
uv pip install pandas numpy pyarrow
代码分三段。第一段：造一份「带脏数据」的行情——优先拉真实数据，然后人为注入缺失、重复、异常和类型问题，让清洗环节有事可做（真实数据太干净，教学效果反而差）。
import numpy as np
import pandas as pd
START, END = "20200101", "20241231"
def fetch_raw(symbol: str) -> pd.DataFrame:
"""拉数据并人为制造脏数据，供清洗教学。优先真实数据，失败则本地造。"""
try:
import akshare as ak
df = ak.stock_zh_a_hist(symbol=symbol, period="daily",
start_date=START, end_date=END, adjust="qfq")
df = df.rename(columns={"日期": "date", "开盘": "open", "收盘": "close",
"最高": "high", "最低": "low",
"成交量": "volume", "涨跌幅": "pct_chg"})
except Exception:
rng = np.random.default_rng(int(symbol))
n = 1215
idx = pd.bdate_range("2020-01-02", periods=n)
close = 16 * np.exp(np.cumsum(rng.normal(0.0002, 0.02, n)))
df = pd.DataFrame({
"open": close * (1 + rng.normal(0, 0.005, n)),
"close": close,
"high": close * (1 + np.abs(rng.normal(0, 0.008, n))),
"low": close * (1 - np.abs(rng.normal(0, 0.008, n))),
"volume": rng.integers(30, 500, n) * 1e4,
"pct_chg": np.append(np.nan, np.diff(close) / close[:-1] * 100),
}, index=idx)
df.index.name = "date"
df = df.reset_index() # 故意先不转 datetime，模拟「刚拿到的原始数据」
# ---- 人为注入四类脏数据（真实数据自带的脏，这里模拟出来）----
df.loc[10, "close"] = np.nan # 缺失：某天收盘价丢了
df.loc[20:21] = df.loc[19:20].values # 重复：同一行被写了两次
df.loc[30, "pct_chg"] = 55.0 # 异常：超出涨跌停制度的涨幅
df["date"] = df["date"].astype(str) # 类型：日期被存成字符串
print(f"原始数据：{len(df)} 行")
print(df.dtypes)
return df
raw = fetch_raw("000001")
第二段：按「查缺失 → 补缺失 → 删重复 → 筛异常 → 转类型」的顺序清洗，每步都打印前后对比，让你看到每一步到底改变了什么。
# ---- 第 1 步：缺失值 ----
print("各列缺失数：\n", raw.isna().sum())
df = raw.copy()
df["date"] = pd.to_datetime(df["date"]) # 类型先修好，后面才能按日期排序
df = df.sort_values("date").set_index("date")
df["close"] = df["close"].ffill() # 价格用前向填充
print("填充后 close 是否还有缺失：", df["close"].isna().any())
# ---- 第 2 步：重复值 ----
dup_mask = df.index.duplicated(keep="first") # 保留第一次出现
print(f"发现 {dup_mask.sum()} 条重复日期，删除")
df = df[~dup_mask]
# ---- 第 3 步：异常值筛查（只标记人工复核，不自动删除）----
suspicious = df[df["pct_chg"].abs() > 21] # A 股制度最大 ±20%（创业板/科创板）
print("超出制度限制的可疑记录：")
print(suspicious[["open", "close", "pct_chg"]])
# ---- 第 4 步：类型转换收尾 ----
df["close"] = pd.to_numeric(df["close"], errors="coerce")
df["volume"] = pd.to_numeric(df["volume"], errors="coerce").fillna(0).astype("int64")
df["pct_chg"] = pd.to_numeric(df["pct_chg"], errors="coerce")
print(df.dtypes)
print(f"清洗完成：{len(df)} 行")
第三段：把清洗好的数据分别用 CSV、SQLite、Parquet 存取一遍，并验证读回来的数据是否「原样」。
import sqlite3
# ---- 方式一：CSV ----
df.to_csv("000001_clean.csv") # 存
df_csv = pd.read_csv("000001_clean.csv", index_col="date", parse_dates=["date"])
print("CSV 读回 dtypes：\n", df_csv.dtypes) # 注意：数值能回来，但要看日期有没有转
# ---- 方式二：SQLite ----
conn = sqlite3.connect("quant.db")
df.reset_index().to_sql("daily_bar", conn, if_exists="replace", index=False)
df_sql = pd.read_sql("SELECT date, close, volume FROM daily_bar "
"WHERE close > 20 ORDER BY date LIMIT 5",
conn, parse_dates=["date"]) # SQL 条件查询是它的强项
conn.close()
print("SQLite 条件查询结果：\n", df_sql)
# ---- 方式三：Parquet ----
df.to_parquet("000001_clean.parquet") # 需要 pyarrow
df_pq = pd.read_parquet("000001_clean.parquet")
print("Parquet 读回 dtypes 与原数据完全一致：", df_pq.dtypes.equals(df.dtypes))
print("Parquet 行数一致：", len(df_pq) == len(df))
代码逐段解释
第 1 段：数据准备与「投毒」。 fetch_raw 沿用第 7 篇的 try/except + fallback 结构，保证没装 akshare、断网都能跑。然后我们故意干四件坏事：把第 11 行（loc[10]）的收盘价挖掉、把 20–21 行复制成 19–20 行的内容制造重复、在第 31 行塞一个 55% 的「不可能涨幅」、再把日期列转成字符串。这四步对应真实世界最常见的四种脏。打印 dtypes 时你会看到 date 是 object——这就是「日期其实是字符串」的信号。示例输出示意（数值为示意）：
原始数据：1215 行
date object
open float64
close float64
...
各列缺失数：
close 1
...
发现 2 条重复日期，删除
超出制度限制的可疑记录：
open close pct_chg
date
2020-02-05 15.8 16.4 55.0
第 2 步顺序有讲究。 先转 datetime 并按日期排序，是一切时间序列操作的前提——ffill 的「前」指的是时间上的前一行，如果日期乱序，「前向填充」就填错了方向。ffill 之后 close 不再缺失；重复用 index.duplicated(keep="first") 标记后取反删除；异常值我们只打印出来「标记待人工复核」，代码注释里写明了不自动删——这是本篇刻意传递的纪律：可疑 ≠ 错误。最后 to_numeric 收尾，成交量列若有缺失按「无成交 = 0」处理后再转整数。每一步都有打印，你在自己的机器上能逐行看到数据的变化过程。
划重点：清洗顺序记住四个字「先正后清」——先把日期和类型修正（正），再做填充去重筛查（清）。顺序错了，后面的操作会在错误的地基上跑。
第 3 段：三种存取对照。 CSV 用 to_csv/read_csv，注意 read_csv 必须传 parse_dates=["date"] 才能把日期读回 datetime——这正是「CSV 不保存类型」的代价；SQLite 用 to_sql 一行入库存进 quant.db，用 pd.read_sql 直接写 SQL 做条件查询（示例里取收盘价大于 20 元的前 5 条），parse_dates 同样要手动指定；Parquet 用 to_parquet/read_parquet，读回后 dtypes 与原始数据完全一致，一个参数都不用传。示例输出示意：
CSV 读回 dtypes：
open float64 ...
SQLite 条件查询结果：
date close volume
0 2020-03-02 20.1 1200000
...
Parquet 读回 dtypes 与原数据完全一致： True
Parquet 行数一致： True
常见坑与小练习
坑 1：ffill 把成交量也填了。 现象：停牌日的成交量变成了前一天的天量。原因：对整表无差别 ffill。避免方式：只对价格列 ffill，成交量缺失按业务含义填 0 或单独处理。
坑 2：CSV 读回后画图画不出时间序列。 现象：X 轴变成一堆字符串、切片 loc["2020"] 报错。原因：CSV 不保存 dtype，日期读成了字符串。避免方式：read_csv 时传 parse_dates=["date", ...]，或读回后立刻 pd.to_datetime。
坑 3：dropna 一口气删掉大半数据。 现象：本来只缺 1 个收盘价，结果删了几百行。原因：默认删除「任何一列含缺失」的行，而 pct_chg 首日天然是 NaN。避免方式：用 subset 参数限定列，如 df.dropna(subset=["close"])，或改用 ffill。
坑 4：SQLite 表里日期变成毫秒大整数。 现象：read_sql 读出来的 date 是一串数字。原因：写库时 datetime 被转成时间戳，读时没指定解析。避免方式：pd.read_sql(..., parse_dates=["date"])，或写入前先转成 ISO 字符串。
坑 5：读 Parquet 报 ImportError。 现象：提示缺少 pyarrow。原因：Parquet 依赖独立的 pyarrow 库，pandas 只是调用方。避免方式：uv pip install pyarrow 装上即可，代码不用改。
小练习 1： 把异常值筛查阈值从 ±21% 改成「主板 ±10%」，并对你自己拉的股票跑一遍：如果它是主板股票，筛选出可疑记录后想一想，哪些是真的触顶、哪些可能是脏数据。
思路提示：区分「接近制度上限的真实行情」和「超出上限的错误记录」——代码里用
> 21只抓后者，练习里可以用(10, 21]区间单独看「触顶」记录。
小练习 2： 把三只股票（第 7 篇的 symbols 列表）的清洗后数据都写进 SQLite 的同一张表，加一列 symbol 区分，再用一条 SQL 查出「每只股票最新一天的收盘价」。
思路提示：每只
df先赋df["symbol"] = sym再to_sql；查询用GROUP BY symbol配合每组的最大日期。
小练习 3： 分别统计三种格式存同一份 1215 行数据的文件大小（用 os.path.getsize），直观感受 Parquet 的压缩率。
思路提示：
import os; os.path.getsize("000001_clean.parquet")，单位是字节，除以 1024 换算 KB；数据行数越大，差距越明显，可以试着把三只股票拼成三千多行再对比。
本文策略仅用于教学，不构成投资建议，实盘前请充分测试。