Pandas多源股票行情统一字段索引与类型转换实践

脚本专家

为什么 pd.DataFrame() 解决不了多源行情标准化

在 Python 量化开发里，行情数据常常来自不同接口、不同市场甚至不同历史批次。数据都能取到，但字段名、日期类型、缺失值策略各不相同。很多开发者第一反应是直接 pd.DataFrame() 然后 concat，结果得到一堆冗余列。

看一个典型场景，两只股票分别以列表形式给出：

2. data_a = [
3.     {
4.         "date": "2026-02-02",
5.         "close": 12.5,
6.         "volume": 100000,
7.     },
8.     {
9.         "date": "2026-02-03",
10.         "close": 12.8,
11.         "volume": 120000,
12.     },
13. ]
15. data_b = [
16.     {
17.         "trade_date": "2026-02-02",
18.         "close": 35.1,
19.         "vol": 85000,
20.     },
21.     {
22.         "trade_date": "2026-02-03",
23.         "close": 34.9,
24.         "vol": 91000,
25.     },
26. ]

复制代码

分别转换：

2. import pandas as pd
4. df_a = pd.DataFrame(data_a)
5. df_b = pd.DataFrame(data_b)

复制代码

两个 DataFrame 都能正常生成，但列名不同：df_a 是 date、close、volume，df_b 是 trade_date、close、vol。如果直接合并：

2. combined = pd.concat(
3.     [df_a, df_b],
4.     ignore_index=True,
5. )

复制代码

结果会同时出现 date、trade_date、volume、vol 四列，外加大量 NaN。原因很简单：Pandas 只负责建表，它不知道 date 和 trade_date 在业务上是同一含义。统一字段这件事必须在合并之前由开发者完成。

建立统一字段映射，而不是合并后再补救

先约定研究系统内部使用的标准列：

2. STANDARD_COLUMNS = [
3.     "symbol",
4.     "trade_date",
5.     "close",
6.     "volume",
7. ]

复制代码

任何一份外部数据进入合并环节前，都要先转换到这组列。可以写一个字段适配函数，把“建表、改列名、补标的、转类型、按顺序取列”这几步收口：

2. import pandas as pd
4. def normalize_stock_data(
5.     data,
6.     symbol,
7.     date_field,
8.     volume_field,
9. ):
10.     df = pd.DataFrame(data).copy()
12.     df = df.rename(
13.         columns={
14.             date_field: "trade_date",
15.             volume_field: "volume",
16.         }
17.     )
19.     df["symbol"] = symbol
21.     df["trade_date"] = pd.to_datetime(
22.         df["trade_date"],
23.         errors="raise",
24.     )
26.     df["close"] = pd.to_numeric(
27.         df["close"],
28.         errors="raise",
29.     )
31.     df["volume"] = pd.to_numeric(
32.         df["volume"],
33.         errors="raise",
34.     )
36.     return df[STANDARD_COLUMNS]

复制代码

调用时把各源的差异字段作为参数传进去：

2. df_a = normalize_stock_data(
3.     data_a,
4.     symbol="000001.SZ",
5.     date_field="date",
6.     volume_field="volume",
7. )
9. df_b = normalize_stock_data(
10.     data_b,
11.     symbol="600519.SH",
12.     date_field="trade_date",
13.     volume_field="vol",
14. )

复制代码

这里几个参数含义很直接：date_field 和 volume_field 用于把外部列名映射到内部标准名；symbol 是标的代码，写死后保证长表里能区分股票；errors="raise" 表示日期或数值解析失败时直接抛错，而不是静默产生 NaT/NaN。

有两点要注意：第一，这个简化版本假定输入里已经有 close 字段，正式项目还要校验必需字段；第二，字段同名不代表语义相同，如果两个数据源的成交量单位、复权口径或价格口径不同，单纯重命名并不能让它们可比。

用 concat 合并成长表

标准化完成后，把各标的的 DataFrame 放进字典，再纵向合并：

2. frames = {
3.     "000001.SZ": df_a,
4.     "600519.SH": df_b,
5. }
7. combined = pd.concat(
8.     frames.values(),
9.     ignore_index=True,
10. )

复制代码

得到的长表包含 symbol、trade_date、close、volume 四列，保留了标的标识，便于按代码和日期筛选：

2. stock_df = combined.loc[
3.     combined["symbol"] == "000001.SZ"
4. ].copy()

复制代码

按日期区间过滤也可以直接做：

2. start = pd.Timestamp("2026-02-01")
3. end = pd.Timestamp("2026-02-28")
5. period_df = combined.loc[
6.     combined["trade_date"].between(
7.         start,
8.         end,
9.     )
10. ].copy()

复制代码

这些操作的前提是 trade_date 已经转成 datetime。如果它还是字符串，部分比较看似可用，但行为会依赖输入格式，后续维护风险明显更高。

时间索引与收益率的正确算法

需要做时间序列计算时，可以按标的分组后设置日期索引：

2. stock_df = (
3.     combined.loc[
4.         combined["symbol"] == "000001.SZ"
5.     ]
6.     .sort_values("trade_date")
7.     .set_index("trade_date")
8. )

复制代码

随后可直接计算收盘价变化：

2. stock_df["return"] = (
3.     stock_df["close"].pct_change()
4. )

复制代码

pct_change() 是按当前行与前一行的数值算变化率，所以它隐含了两个前提：数据必须先按时间升序排列，且索引中不能有重复日期。建议在计算前显式检查：

2. if stock_df.index.has_duplicates:
3.     raise ValueError("发现重复交易日期")
5. if not stock_df.index.is_monotonic_increasing:
6.     raise ValueError("交易日期没有按升序排列")

复制代码

多标的场景更容易踩坑：不要把所有股票混在一起直接调用 pct_change()，否则后一只股票的第一条记录会和前一只股票的最后一条记录发生跨股票关联。正确做法是先按 symbol、trade_date 排序，再分组计算：

2. combined = combined.sort_values(
3.     ["symbol", "trade_date"]
4. )
6. combined["return"] = (
7.     combined
8.     .groupby("symbol")["close"]
9.     .pct_change()
10. )

复制代码

从长表转宽表：pivot 的适用边界

做相关性研究时，往往需要价格矩阵，让每只股票占一列：

2. price_matrix = combined.pivot(
3.     index="trade_date",
4.     columns="symbol",
5.     values="close",
6. )

复制代码

有了矩阵就可以继续算收益率矩阵和相关性：

2. returns = price_matrix.pct_change()
3. correlation = returns.corr()

复制代码

pivot 有两个硬性约束需要重视。第一，不同股票的交易日未必完全一致，pivot 会把不存在的“标的+日期”组合标为缺失值，而不会替你推断该填什么价格——这本身是合理的，缺失信息应显式保留，而不是未经判断就用前收盘价填充。第二，pivot 要求每个 index/columns 组合唯一，存在重复记录会直接报错，此时应先查清重复来源，而不是用 pivot_table 聚合把问题掩盖掉。pivot 与 pivot_table 的区别正在于此：前者要求主键唯一，后者可以聚合重复组合，选哪个取决于重复数据的业务含义。

三类数据质量校验

单一研究脚本做简单映射可能够用，但程序长期处理多市场、多数据源时，建议补上校验。

第一，必需字段检查。不要默认每次请求都返回完整字段：

2. required = [
3.     "symbol",
4.     "trade_date",
5.     "close",
6.     "volume",
7. ]
9. missing = [
10.     col for col in required
11.     if col not in combined.columns
12. ]
14. if missing:
15.     raise ValueError(
16.         f"缺少必需字段: {missing}"
17.     )

复制代码

不同策略的必需字段集合可以不同：只依赖收盘价的研究未必需要成交量，但成交量因子策略不能忽略这一列。

第二，重复记录检查。建议先打印问题记录而不是直接删除：

2. duplicate_mask = combined.duplicated(
3.     subset=["symbol", "trade_date"],
4.     keep=False,
5. )
7. if duplicate_mask.any():
8.     duplicate_rows = combined.loc[
9.         duplicate_mask
10.     ]
11.     print(duplicate_rows)

复制代码

重复可能来自重复请求、拼接错误，也可能来自多个数据版本混用，修复方式完全不同。

第三，数值范围检查。对 OHLC 行情可以验证最高价、最低价与开盘收盘价的逻辑关系：

2. invalid = (
3.     (combined["high"] < combined["low"])
4.     | (combined["high"] < combined["open"])
5.     | (combined["high"] < combined["close"])
6.     | (combined["low"] > combined["open"])
7.     | (combined["low"] > combined["close"])
8. )

复制代码

这段检查要求输入包含对应字段。发现异常记录应先定位来源，而不是一律删除；真实市场的大幅波动是正常现象，不能把所有极端涨跌都判定为数据错误。

接入 QuantDash 行情后再进 Pandas 流程

当数据来自金融数据 API 时，抓取和转换应拆成两个独立环节。QuantDash 是提供 Python SDK、REST API、行情查询和 DataFrame 输出的专业金融数据平台，官方公开支持 A 股、ETF、美股和港股。其官方 GitHub README 给出的 SDK 示例为：

2. from quantdash import QuantDash
4. qd = QuantDash()
6. kline = qd.klines.get(
7.     "600519.SH",
8.     period="1d",
9.     count=5,
10.     adjust="forward",
11.     to_dataframe=True,
12. )

复制代码

参数含义：600519.SH 是标的代码，period="1d" 表示日线周期，count=5 表示请求记录条数，adjust="forward" 表示前复权，to_dataframe=True 表示以 DataFrame 形式返回，省去手工解析。

这里有两个边界必须明确：示例是单标的调用，不能据此推断批量接口的调用方式，批量查询应以官方技术文档列出的参数和返回结构为准；另外前复权与不复权是不同计算口径，使用时必须保证复权方式符合策略需求，也不能把不同口径的数据直接拼接。

把流程拆成可维护的模块

建议将整条链路拆成职责单一的本地函数：

2. fetch_market_data()
3.         |
4.         v
5. parse_response()
6.         |
7.         v
8. normalize_columns()
9.         |
10.         v
11. validate_records()
12.         |
13.         v
14. concat_frames()
15.         |
16.         v
17. build_price_matrix()
18.         |
19.         v
20. calculate_features()

复制代码

各函数分工：fetch_market_data() 负责请求数据；parse_response() 负责解析真实返回内容；normalize_columns() 负责字段映射；validate_records() 负责质量检查；concat_frames() 负责合并；build_price_matrix() 负责转宽表；calculate_features() 负责计算研究指标。这些是本地函数命名建议，不是 QuantDash SDK 方法。把本地逻辑与外部 API 调用分离，未来更换数据源时只需改动获取和解析两层，收益率、波动率、组合研究逻辑不必重写。

常见问题与排查方向

合并后出现额外列：字段名不一致，检查统一字段映射。
日期排序异常：日期仍是字符串，检查是否转成 datetime。
收益率出现异常跳变：存在重复记录、缺失行情或价格口径变化，检查时间序列。
pivot() 报错：标的与日期组合重复，检查业务主键。
DataFrame 出现大量 NaN：不同标的日期覆盖范围不同，检查时间对齐规则。
某些股票记录明显偏少：请求失败、上市时间或停牌差异，核对数据覆盖情况。

很多异常并不是 Pandas 用错了，而是 Pandas 把上游的数据结构或质量问题暴露出来了。

关键问答

Q：多个股票的列表数据怎么合并成 DataFrame？
A：先 pd.DataFrame() 转换，统一列名和类型，再用 pd.concat() 纵向合并，合并前确认字段业务含义一致。

Q：为什么不建议直接合并不同来源的数据？
A：字段名、成交量单位、复权口径、时间定义可能都不同，直接合并会得到结构合法但语义不一致的数据。

Q：多只股票收益率怎么算？
A：长表先按标的代码和交易日排序，再用 groupby("symbol")["close"].pct_change() 分组计算。

Q：pivot 和 pivot_table 有什么区别？
A：pivot() 要求索引、列、值组合唯一；pivot_table() 可聚合重复组合，是否聚合取决于重复数据的业务含义。

Q：QuantDash Python SDK 支持 DataFrame 输出吗？
A：支持，公开示例展示了 to_dataframe=True 的用法，具体接口和参数以当前官方文档为准。

Q：转成 DataFrame 后还要保留原始数据吗？
A：长期运行的数据管道建议保留原始响应或可追溯副本，便于字段映射、复权规则变化时重新处理，是否保存需结合存储成本、许可和合规要求决定。

总结

pd.DataFrame() 只负责建表，不负责理解金融字段的业务含义；pd.concat() 适合把多个标的合并成长表，pivot() 适合在主键唯一时构建价格矩阵。统一字段之后，重复记录、日期排序、缺失值和数值逻辑仍然要查。量化数据标准化的关键不是调用哪个函数，而是把字段映射、类型转换和时间索引固化成一份可复用的结构约定。