📚 课程目录第3章批量文件枚举与元数据管理

3. 批量文件枚举与元数据管理

逐笔CSV通常按交易日或按标的分散存放，一个交易年的数据往往对应数百甚至上千个文件。如果每次分析都临时用os.listdir拼路径、逐个打开再判断，既慢又容易漏文件。更稳妥的做法是先做一次“文件盘点”：用glob或pathlib批量发现目标文件，提取每个文件的大小、行数、时间范围等元数据，汇总成一张文件清单DataFrame。后续的读取、校验、增量更新都围绕这张清单展开，而不是反复扫描目录。

一、用glob与pathlib发现文件

glob适合快速按通配符匹配，pathlib则把路径当作对象，链式操作更清晰。假设逐笔数据按“年/月”两级目录存放，文件名为tick_20240102.csv这种形式：

from pathlib import Path

root = Path("./data/tick")
files = sorted(root.glob("2024/*/tick_*.csv"))
print(len(files), files[0])
# 输出示例：248 data/tick/2024/01/tick_20240102.csv

这里有两个细节值得注意。第一，glob的*只匹配单层目录，跨层要用**并配合recursive=True，例如root.glob("**/tick_*.csv")。第二，务必用sorted()排序，因为文件系统返回顺序不保证稳定，排序后清单才可复现。若目录中混有tick_20240102.csv.bak、._tick_20240102.csv这类干扰文件，应在匹配后过滤：

files = [p for p in files
         if p.suffix == ".csv" and not p.name.startswith("._")]

对于需要同时匹配多种命名规则的情况，可以合并多个glob结果，再用set去重，避免同一文件被重复计入。

二、提取元数据：大小、行数、时间范围

文件大小用p.stat().st_size即可，成本极低。行数则有两种口径：物理行数（含表头）和数据行数。逐笔文件通常首行是表头，因此数据行数 = 物理行数 − 1。统计行数最省内存的方式是逐块读取并计数换行符，而不是readlines()：

def count_lines(path, chunk_size=1 << 20):
    n = 0
    with open(path, "rb") as f:
        while True:
            buf = f.read(chunk_size)
            if not buf:
                break
            n += buf.count(b"\n")
    return n

时间范围是逐笔数据最关键的元数据。它不能靠文件名推断，必须读入时间列后取最小值和最大值。为控制开销，可以只读取时间列：

import pandas as pd

def time_range(path, time_col="时间"):
    s = pd.read_csv(path, usecols=[time_col])[time_col]
    return s.min(), s.max(), len(s)

这里len(s)得到的是有效数据行数，可与前面的物理行数互相校验：若两者差异明显，说明文件存在空行、截断或编码异常。时间列的具体名称、格式（如09:30:00.500或20240102 093000500）因数据源而异，示例中统一按字符串比较即可得到当日的起止时刻；若跨日文件，则应先解析为datetime再取范围。

三、构建文件清单DataFrame

把上述信息组织成记录列表，再一次性构造DataFrame，比逐行append高效得多：

records = []
for p in files:
    st = p.stat()
    lo, hi, rows = time_range(p)
    records.append({
        "path": str(p),
        "trade_date": p.stem.split("_")[1],
        "size_mb": round(st.st_size / 1024 / 1024, 2),
        "data_rows": rows,
        "start_time": lo,
        "end_time": hi,
        "mtime": pd.Timestamp(st.st_mtime, unit="s"),
    })

manifest = pd.DataFrame(records).sort_values("trade_date")
manifest.to_parquet("./data/manifest_2024.parquet", index=False)

清单落盘为Parquet后，下次启动直接读取，无需重新扫描。清单的价值在于它把“目录里有什么”变成了“可查询的表”：可以按trade_date筛选区间，按data_rows发现异常小的文件，按start_time检查是否缺失开盘时段。例如下面这段检查能快速定位可疑文件：

suspect = manifest[
    (manifest["data_rows"] < 1000) |
    (manifest["start_time"] > "09:31:00")
]
print(suspect[["path", "data_rows", "start_time"]])

若输出中某个文件行数远低于同月均值，或起始时间晚于9:31，就应优先人工核查，而不是直接进入批量读取环节。

四、清单的维护与增量更新

逐笔数据是持续增长的，清单不能只建一次。合理策略是记录每个文件的mtime和大小，下次扫描时只对“新增或已修改”的文件重新提取元数据，其余直接复用旧记录。判断条件可以写成：

old = pd.read_parquet("./data/manifest_2024.parquet")
old_map = old.set_index("path")[["size_mb", "mtime"]].to_dict("index")

for p in files:
    st = p.stat()
    prev = old_map.get(str(p))
    if prev and prev["size_mb"] == round(st.st_size/1024/1024, 2) \
            and prev["mtime"] == pd.Timestamp(st.st_mtime, unit="s"):
        continue   # 未变化，跳过
    # 否则重新提取元数据并更新记录

这样，日常增量扫描的成本只与新增文件数相关，而不是与全量文件数相关。需要提醒的是，mtime在某些同步盘或解压操作后可能不可靠，因此大小与行数的交叉校验仍然必要。

五、常见问题与处理建议

编码不一致：部分文件为GBK，读取时间列时可能抛UnicodeDecodeError。可在read_csv中显式指定encoding，或先探测编码再统一处理。
表头缺失：个别文件首行直接是数据，导致行数统计偏差1行。可在清单中增加has_header标记，读取时按标记传header=None。
空文件与截断文件：大小为0或行数极少的文件应标记为status="suspect"，而不是直接删除，保留证据便于追溯。
路径过长：Windows下深层目录可能触发路径长度限制，建议在清单中保存相对路径，读取时再与根目录拼接。

完成这一步后，后续的批量读取就有了明确的输入边界：读哪些文件、每个文件预期多少行、时间范围是否完整，都能在读取前判断。清单本身也是数据质量报告的第一层，很多问题在打开文件之前就已经暴露出来。