本文介绍使用 pandas 按 `sym`（交易对）分组，将高频获取的 ohlcv 时序数据批量、高效地追加写入对应名称的 csv 文件（如 ethusdt.csv、idusdt.csv），避免逐行低效操作。
在实时行情数据采集场景中（例如从 Binance 流式获取 OHLCV 数据），你常会得到一个包含多个交易对（sym）、带时间戳索引的 DataFrame。每条记录属于某个特定币对（如 'ETHUSDT'、'IDUSDT'），且同一币对的数据可能跨不同时间点（甚至跨天）。目标是：将每个币对的所有历史与新增数据，持续追加写入其专属 CSV 文件，并确保格式一致、性能可靠。
直接遍历 df['sym'] 并对每个值调用 to_csv()（如 for coin in df.sym: ...）是严重错误的做法——它不仅重复打开/关闭文件、引发 I/O 瓶颈，还会因未传入完整行数据导致 CSV 内容错乱（仅写入 sym 列，丢失 OHLCV 等关键字段）。
✅ 正确做法是：先按 sym 分组，再对每个分组 DataFrame 整体追加写入对应文件。核心代码如下：
Miller (mlr) 是一个命令行工具，用于查询、整形和重新格式化名称索引数据，如 CSV、TSV、JSON 和 JSON Lines。它将 awk、sed、cut、join 和 sort 的功能整合到一个专为结构化数据处理而构建的单一工具中。
for symbol, group in df.groupby('sym'):
filename = f"{symbol}.csv"
# 首次写入需带表头；后续追加不写表头，且确保 index（时间戳）也被保存
group.to_csv(filename, mode='a', header=not os.path.exists(filename), index=True)? 关键说明：
- df.groupby('sym') 将原始 DataFrame 按 sym 列值自动切分为多个子 DataFrame，每个子集包含该币对全部行；
- mode='a' 启用追加模式，避免覆盖历史数据；
- header=not os.path.exists(filename) 实现「首次创建时写表头，后续追加时不重复写」——这是保证 CSV 结构规范的关键；
- index=True 显式保留时间戳索引（即原始 2024-02-09 11:15:59.594），否则时间信息将丢失；
? 进阶建议（提升吞吐量）： 若数据以小批次持续流入（如每秒一批），频繁磁盘 I/O 仍可能成为瓶颈。推荐采用「缓冲 + 批量落盘」策略：
import pandas as pd
import os
# 初始化缓存字典：symbol → list of DataFrames
buffered_dfs = {}
def append_to_buffer(new_df):
for symbol, group in new_df.groupby('sym'):
if symbol not in buffered_dfs:
buffered_dfs[symbol] = []
buffered_dfs[symbol].append(group)
def flush_buffers():
for symbol, groups in buffered_dfs.items():
if not groups:
continue
# 合并当前所有缓冲数据
full_group = pd.concat(groups, ignore_index=False)
filename = f"{symbol}.csv"
full_group.to_csv(
filename,
mode='a',
header=not os.path.exists(filename),
index=True
)
# 清空缓冲区
buffered_dfs.clear()
# 使用示例：每 N 条或每 T 秒调用一次 flush_buffers()
# append_to_buffer(current_batch_df)
# if len(buffered_dfs) > 1000 or time_since_last_flush > 5:
# flush_buffers()⚠️ 注意事项：
- 确保 sym 列不含非法文件名字符（如 /, , :），建议预处理：df['sym'] = df['sym'].str.replace(r'[<>:"/\|?*]', '_', regex=True)；
- 若时间戳索引存在时区信息，写入前建议统一转为 UTC 或本地无时区格式（df.index = df.index.tz_localize(None)），避免 CSV 解析歧义；
- 生产环境强烈建议添加异常捕获与日志（如 try...except 包裹 to_csv），防止单个文件写入失败阻塞整体流程。
总结：groupby + to_csv(mode='a') 是解决“多符号、多时间戳、追加存储”需求的标准范式。它逻辑清晰、性能可控、易于维护，是量化数据管道中的基础但关键的一环。
