引言与学习目标
学习目标
完成本章后，你应该能够：
用集合关系解释内连接、左连接、外连接与多对多连接的行数变化；
构造、切片并审计 MultiIndex，验证面板键的唯一性和排序要求；
用 merge、join 与 concat 整合本地行情和估值数据，并用 validate 检查键基数；
用 stack、unstack、melt 与透视操作在宽表和长表之间转换；
识别发布日期与报告期的区别，按信息可得日完成时点安全的连接；
说明连接和重塑是描述性数据工程步骤，本身不产生预测或因果结论。
目标—活动/示例—核心练习—答案证据映射
解释四类连接与多对多行数
小节 7.1.1 习题 7.1、7.3
按键频数、预期行数和实际 merge 行数一致
构造并审计 MultiIndex
小节 7.2 习题 7.4
层级名称、排序、切片结果和键唯一性检查
使用 merge、join、concat 并验证基数
小节 7.3 习题 7.2、7.6
validate 合同、行数变化和重复键报告
完成长宽表重塑
小节 7.4 stack、unstack、melt 活动习题 7.5
往返重塑后的键集合、形状与数值一致性
按信息可得日完成时点安全连接
发布日与报告期辨析及 merge_asof 活动
习题 7.8
publish_date <= trade_date 断言、发布日前不可见断言与反例说明
限定连接的推断边界
本章数据与推断边界、综合整合报告
习题 7.7、7.8
区分数据工程、预测验证与因果识别所需额外证据
数据与推断边界
本章直接写入的小型报价、键表和评级记录均为“机制演示题设值”，用于暴露重复键、笛卡尔积与时点错配，不属于真实研报或公司观测。真实案例读取本地 HDF5，并保留证券代码、日期和字段来源。任何模型输入仍须遵守第 15—19 章的训练期拟合、披露时点和样本外验证边界。
在大多数真实世界的应用中，你所需要的数据往往分散在多个文件、数据库或API中。这些数据很少会以一种立即可用于分析的格式呈现。本章将重点介绍 pandas 中用于合并、连接和重排数据，从而将其整理成干净、便于分析的格式的核心工具。
首先，我们将介绍分层索引 (Hierarchical Indexing) 的概念，这是 pandas 的一个强大功能，它允许我们构建更复杂的数据结构。这将是后续许多操作的基石。然后，我们将深入探讨数据操作的具体机制，并通过源自中国经济与金融的实际案例，将这些抽象的工具付诸实践。
理论基础：数据合并的数学原理
数据合并的集合论基础
数据合并操作本质上是基于键（keys）的集合运算。给定两个数据集：
左表 \(D_L\) ，键集合 \(K_L = \{k_1, k_2, ..., k_m\}\)
右表 \(D_R\) ，键集合 \(K_R = \{k'_1, k'_2, ..., k'_n\}\)
四种连接类型的数学定义 ：
内连接 (Inner Join) ： $ K_{inner} = K_L K_R $ 结果仅包含两个表中都存在的键。
左连接 (Left Join) ： $ K_{left} = K_L $ 结果包含左表的所有键，右表中不匹配的键填充缺失值。
右连接 (Right Join) ： $ K_{right} = K_R $ 结果包含右表的所有键，左表中不匹配的键填充缺失值。
外连接 (Outer Join) ： $ K_{outer} = K_L K_R $ 结果包含两个表的所有键，不匹配的位置填充缺失值。
数据合并的行对表示
令 \(R_L(k)\) 、\(R_R(k)\) 分别表示左右表中键为 \(k\) 的行的多重集合，\(\bot\) 表示缺失的一侧，\(\uplus\) 表示保留重复行的多重集合并。不同 how 参数对应的输出行对集合必须分别定义：
\[
J_{\mathrm{inner}}
=\mathop{\biguplus}_{k\in K_L\cap K_R}R_L(k)\times R_R(k),
\]
\[
J_{\mathrm{left}}
=J_{\mathrm{inner}}
\uplus\mathop{\biguplus}_{k\in K_L\setminus K_R}\{(r_L,\bot):r_L\in R_L(k)\},
\]
\[
J_{\mathrm{right}}
=J_{\mathrm{inner}}
\uplus\mathop{\biguplus}_{k\in K_R\setminus K_L}\{(\bot,r_R):r_R\in R_R(k)\},
\]
\[
J_{\mathrm{outer}}
=J_{\mathrm{left}}
\uplus\mathop{\biguplus}_{k\in K_R\setminus K_L}\{(\bot,r_R):r_R\in R_R(k)\}.
\]
因此，内连接只保留匹配行对；左、右连接分别额外保留对应一侧的未匹配行；外连接同时保留两侧未匹配行。实际 merge 会把每个行对展开为一行，并在 \(\bot\) 对应的字段中填入缺失值。
按键连接基数与最坏复杂度
令 \(n_L(k)\) 与 \(n_R(k)\) 分别表示键 \(k\) 在左、右表中的行数。对共同键 \(k\) ，连接会生成 \(n_L(k)n_R(k)\) 个配对，而不是无条件生成全表笛卡尔积。因此内连接基数为
\[
|D_{\mathrm{inner}}|=\sum_{k\in K_L\cap K_R}n_L(k)n_R(k).
\tag{7.1}\]
左、右、外连接还要保留未匹配行：
\[
|D_{\mathrm{left}}|
=\sum_{k\in K_L}n_L(k)\max\{1,n_R(k)\},
\tag{7.2}\]
式 式 7.2 给出本节后续实现与解释采用的数学关系。
\[
|D_{\mathrm{right}}|
=\sum_{k\in K_R}n_R(k)\max\{1,n_L(k)\},
\tag{7.3}\]
式 式 7.3 给出本节后续实现与解释采用的数学关系。
\[
|D_{\mathrm{outer}}|
=\sum_{k\in K_L\cap K_R}n_L(k)n_R(k)
+\sum_{k\in K_L\setminus K_R}n_L(k)
+\sum_{k\in K_R\setminus K_L}n_R(k).
\tag{7.4}\]
式 式 7.4 给出本节后续实现与解释采用的数学关系。
若所有左、右行都共享同一个键，输出才达到 \(|D_L||D_R|\) ，即关于两表行数的最坏乘积增长；当两表规模都为 \(n\) 时是 \(O(n^2)\) ，不是指数增长。哈希连接在通常哈希假设下的处理时间可写为期望 \(O(|D_L|+|D_R|+|D_M|)\) ，其中输出物化本身就至少需要 \(\Omega(|D_M|)\) 时间与空间。排序合并还会引入排序成本，具体实现不能只由集合定义推出。
分层索引
分层索引是 pandas 的一个关键特性，它允许你在单个轴上拥有多个（两个或更多）索引层级。你可以将其视为一种在熟悉的二维 DataFrame 结构中表示更高维度数据的方法。这在经济学中对于处理面板数据（Panel Data）尤其有用，面板数据旨在追踪多个主体（如不同省份或公司）在一段时间内的变化。
下面使用本地上证综指、沪深 300 和深证成指的真实日度点位。它们天然适合两级索引：一级是指数名称，另一级是交易日期。
核心概念：分层索引 (MultiIndex) 的直观逻辑
在处理复杂的金融数据集时，你可以将多重索引想象成电子表格中的“合并单元格”架构。
高维数据的二维表示 ：外层索引（级别 0，如：经济指标、股票代码）作为主要分类，其下嵌套了内层索引（级别 1，如：日期、财务科目）。面板数据结构 ：这种结构允许我们在二维表格中自然地表达三维甚至更高维度的数据。例如，一个主体的多项指标在多个时间点上的值。计算优势 ：在计量经济学中，这是处理面板数据 (Panel Data) 的核心基础，使得针对特定截面或特定时间段的选择性聚合变得极其高效。
首先，让我们创建一个带有这种多级索引的 Series。
下面的小表用重复键直接验证 式 7.1 。键 A 在两侧分别出现 2 次和 3 次，贡献 6 行；键 B 只在左侧，内连接不保留它。
= pd.DataFrame({'join_key' : ['A' , 'A' , 'B' ], 'left_value' : [1 , 2 , 3 ]}) # 左表键频数为A两次、B一次 = pd.DataFrame({'join_key' : ['A' , 'A' , 'A' , 'C' ], 'right_value' : [4 , 5 , 6 , 7 ]}) # 右表键频数为A三次、C一次 = left_key_table['join_key' ].value_counts() # 统计公式中的左侧n_L(k) = right_key_table['join_key' ].value_counts() # 统计公式中的右侧n_R(k) = left_counts.index.intersection(right_counts.index) # 只保留内连接的共同键集合 = sum (left_counts[key] * right_counts[key] for key in common_keys) # 按键乘积后求和 = left_key_table.merge(right_key_table, on= 'join_key' , how= 'inner' , validate= 'many_to_many' ) # A键在左右2×3配对生成6行，B/C非共同键不进入内连接 assert expected_inner_rows == 6 # 核对手算的A键贡献为2乘3 assert len (inner_key_pairs) == expected_inner_rows # 核对pandas输出行数符合按键基数公式 # 展示六个A键配对作为可见答案证据
0
A
1
4
1
A
1
5
2
A
1
6
3
A
2
4
4
A
2
5
5
A
2
6
= [] # 暂存三个指数的日收盘序列，随后组成外层指数、内层日期的长轴 for sym in symbols: # 逐一处理每支指数的数据 = index_data[index_data['symbol' ] == sym].copy() # 循环每次隔离一个 symbol，仅在副本上解析时间与设置层级 'datetime' ] = pd.to_datetime(temp['datetime' ].astype(str ), format = '%Y%m %d %H%M%S' ) # 按十四位来源合同解析日键，使三个指数可按同一时间层拼接 'datetime' , inplace= True ) # 将日期时间列设为行索引 = temp['close' ].sort_index().loc['2023-01-01' :'2023-12-31' ] # 提取2023年收盘价并按时间排序 = 'Value' # 将Series命名为'Value'以便后续合并 # 外层保留真实指数名称，内层保留交易日期 = pd.MultiIndex.from_product( # 将单个指数名称与其真实交易日组成两层笛卡尔索引 = ['index_name' , 'date' ] # 固定外层业务名称和内层日期名称 # 完成当前指数的两层索引构造 # 收集当前指数的日期内层序列，循环后按外层指数名纵向拼接 # 将所有 Series 连接成一个并排序索引 = pd.concat(all_series).sort_index() # 纵向拼接三个指数并按层级键排序 # 抽查外层指数名、内层交易日及日收益值三者的层级关系 print (china_market_indices_series.head()) # 输出层级索引头部作为可见核验结果
index_name date
上证综指 2023-01-03 3116.5119
2023-01-04 3123.5164
2023-01-05 3155.2162
2023-01-06 3157.6365
2023-01-09 3176.0845
Name: Value, dtype: float64
列表 7.1 的输出使用 MultiIndex。外层 index_name 的空白表示沿用上方标签。下面检查索引对象本身。
# 核对层级名称及指数—日期的嵌套顺序
MultiIndex([('上证综指', '2023-01-03'),
('上证综指', '2023-01-04'),
('上证综指', '2023-01-05'),
('上证综指', '2023-01-06'),
('上证综指', '2023-01-09'),
('上证综指', '2023-01-10'),
('上证综指', '2023-01-11'),
('上证综指', '2023-01-12'),
('上证综指', '2023-01-13'),
('上证综指', '2023-01-16'),
...
('深证成指', '2023-12-18'),
('深证成指', '2023-12-19'),
('深证成指', '2023-12-20'),
('深证成指', '2023-12-21'),
('深证成指', '2023-12-22'),
('深证成指', '2023-12-25'),
('深证成指', '2023-12-26'),
('深证成指', '2023-12-27'),
('深证成指', '2023-12-28'),
('深证成指', '2023-12-29')],
names=['index_name', 'date'], length=726)
对于一个分层索引的对象，部分索引 (Partial Indexing) 是常用特性之一。它允许你像操作普通索引一样，通过外层标签直接提取一个完整的数据子集。
例如，从包含多个指数的长序列中，可以直接取出沪深 300 的历史点位：
'沪深300' ] # 仅选择一个外层指数并观察内层日期索引降维
date
2023-01-03 3887.8992
2023-01-04 3892.9477
2023-01-05 3968.5782
2023-01-06 3980.8888
2023-01-09 4013.1196
...
2023-12-25 3347.4508
2023-12-26 3324.7902
2023-12-27 3336.3572
2023-12-28 3414.5403
2023-12-29 3431.1099
Name: Value, Length: 242, dtype: float64
可以看到，返回的结果是一个标准的 Series，其索引仅保留了原先的 date 层级。这种层级降维（Dimensionality Reduction）特性极大地简化了多维数据的分析。
你也可以像对标准索引一样，对外层索引进行切片操作。
# 同时选择两个外层标签，验证 MultiIndex 可保留指定指数集合 # 使用显式 IndexSlice 同时筛选两个外层标签 '上证综指' , '沪深300' ], :] # 内层冒号保留两个指数的全部交易日 # 返回仍保留两层索引的选定指数集合
index_name date
上证综指 2023-01-03 3116.5119
2023-01-04 3123.5164
2023-01-05 3155.2162
2023-01-06 3157.6365
2023-01-09 3176.0845
...
沪深300 2023-12-25 3347.4508
2023-12-26 3324.7902
2023-12-27 3336.3572
2023-12-28 3414.5403
2023-12-29 3431.1099
Name: Value, Length: 484, dtype: float64
甚至可以从“内层”级别进行选择。在这里，我们可以选择所有指数在特定日期的数据。为此，我们使用 .loc 索引器。第一个位置的冒号 : 表示我们想要选择第一个（外层）索引级别中的所有项。
实战技巧：.loc 索引器的高级选择逻辑
在处理分层索引时，.loc 是实现精确切片的最强工具。
全能语法 ：.loc[(level_0_selection, level_1_selection, ...)]占位符机制 ：使用冒号 : 作为占位符，代表“选择该层级的所有标签”。示例 ：data.loc[:, '2023-01-03'] 表示保留全部指数，提取 2023 年 1 月 3 日的横截面观测。
# 选择 2023 年 1 月 3 日的数据 '2023-01-03' ] # 提取三个指数在指定交易日的横截面值
index_name
上证综指 3116.5119
沪深300 3887.8992
深证成指 11117.1303
Name: Value, dtype: float64
使用 stack 和 unstack 进行重塑
分层索引在数据重塑中扮演着重要角色。一个常见的操作是将带有 MultiIndex 的 Series 重排成 DataFrame。这可以通过 unstack() 方法完成，该方法会将最内层的索引级别旋转到列索引中。
表 表 7.1 汇总本节的计算或审计结果，解释时应遵循正文给出的口径与限制。
= china_market_indices_series.unstack() # 将日期层旋转为列以比较不同交易日点位 # 检查宽表的指数行与日期列是否符合预期
unstack() 的逆操作是 stack()，它将列旋转到行中，再次生成一个 Series。这些操作对于在“长”格式和“宽”格式数据之间转换至关重要，我们将在 小节 7.4
# `stack` 默认移除宽表中的缺失单元，因此往返行数取决于三指数的共同交易日覆盖 # 将非缺失宽表单元还原为指数—日期长序列并抽查头部
index_name date
上证综指 2023-01-03 3116.5119
2023-01-04 3123.5164
2023-01-05 3155.2162
2023-01-06 3157.6365
2023-01-09 3176.0845
dtype: float64
DataFrame 上的分层索引
对于 DataFrame，行轴和列轴都可以拥有分层索引。让我们构建一个更复杂的例子。我们将使用本地 HDF5 中的真实 A 股市场数据——宁波港（601018） 和 宁波银行（002142） 的开盘价、收盘价和日交易量。这将允许我们创建一个在行（股票代码、日期）和列（度量类型）上都具有 MultiIndex 的 DataFrame。
import pandas as pd # 组装宁波港与宁波银行的双轴 MultiIndex 样例 from pathlib import Path # 解析复权行情文件位置，使双轴 MultiIndex 示例可跨运行环境读取同一 HDF5 # 两只证券共用复权日行情数据合同，便于纵向拼接后比较 = pd.read_hdf( # 取得宁波港的 date 与 OHLCV 记录 f' { DATA_ROOT} /stock/stock_price_pre_adjusted.h5' , # 来源为复权股票日行情 HDF5 表 = "order_book_id='601018.XSHG'" # 在读取端限定宁波港代码 = {'date' : 'trade_date' , 'vol' : 'volume' }) # 恢复宁波港交易日列并统一 volume 字段 = pd.read_hdf( # 从同源复权行情取得银行股 OHLCV，与港口股纵向拼接 f' { DATA_ROOT} /stock/stock_price_pre_adjusted.h5' , # 宁波银行沿用同一价格调整与字段口径 = "order_book_id='002142.XSHE'" # 在读取端锁定 002142.XSHE，不让其他证券进入银行子表 = {'date' : 'trade_date' , 'vol' : 'volume' }) # 恢复宁波银行交易日列并统一 volume 字段 # 选择2024年1月的前几个交易日用于演示 = ningbo_port[ # 截取港口股 2024-01-01 至 01-10 的实际交易日作索引外层样本 'trade_date' ] >= '2024-01-01' ) & # 宁波港演示窗口从2024年1月起 'trade_date' ] <= '2024-01-10' ) # 宁波港演示窗口截至1月10日 # 独立处理宁波港子表，不改写来源行情 = ningbo_bank[ # 银行股使用相同日期边界，以检查两品种的内层 Date 对齐 'trade_date' ] >= '2024-01-01' ) & # 宁波银行使用相同窗口起点 'trade_date' ] <= '2024-01-10' ) # 宁波银行使用相同窗口终点 # 独立处理宁波银行子表，保留原数据不变 # 选择需要的列并重命名 = ningbo_port_demo[['trade_date' , 'open' , 'high' , 'low' , 'close' , 'volume' ]] # 投影宁波港的日期键与 OHLCV 指标 = ningbo_bank_demo[['trade_date' , 'open' , 'high' , 'low' , 'close' , 'volume' ]] # 以相同 schema 投影宁波银行行情 # 两只证券统一为 Date 与首字母大写 OHLCV 合同，保证纵向拼接不产生同义列 = ['Date' , 'Open' , 'High' , 'Low' , 'Close' , 'Volume' ] # 宁波港列名转为分层列演示使用的首字母大写合同 = ['Date' , 'Open' , 'High' , 'Low' , 'Close' , 'Volume' ] # 宁波银行对齐同一列合同，确保纵向拼接不扩列 # 为每个 DataFrame 添加一个 'ticker' 列 'ticker' ] = 'NingboPort' # 港口子表使用 NingboPort 作双层行索引的外层键 'ticker' ] = 'NingboBank' # 银行子表标记为另一外层 ticker，与港口行分区
# 连接并设置 MultiIndex = pd.concat([ningbo_port_demo, ningbo_bank_demo]) # 将两只股票数据纵向拼接为一个DataFrame = stocks.set_index(['ticker' , 'Date' ]) # 设置(股票代码, 日期)为双层行索引 # 为列创建 MultiIndex = stocks[['Open' , 'Close' , 'Volume' ]].copy() # 值轴仅保留价格、成交量三项指标，行轴继续使用证券—日期复合键 = pd.MultiIndex.from_tuples([ # 为列创建两级索引(Category, Metric) 'Price' , 'Open' ), # 价格类别-开盘价 'Price' , 'Close' ), # 价格类别-收盘价 'Volume' , 'Total' ) # 成交量类别-总量 # 完成列的MultiIndex构建 # 核对行轴为 ticker—Date、列轴为 Category—Metric 的双层 schema
分层索引的每个级别都可以有名称。让我们来为它们命名。
= ['Ticker' , 'TradeDate' ] # 为行索引的两个层级分别命名为股票代码和交易日期 = ['Category' , 'Metric' ] # 为列索引的两个层级分别命名为类别和指标 # 核对四个层级名称已写入轴元数据，便于后续按层选择和聚合
Metric
Open
Close
Total
NingboPort
2024-01-02
3.3400
3.3777
24852644.0
2024-01-03
3.3683
3.4059
17502674.0
2024-01-04
3.4153
3.4059
16564281.0
2024-01-05
3.4153
3.3871
15904201.0
2024-01-08
3.3965
3.3494
16029300.0
2024-01-09
3.3306
3.3588
20051699.0
2024-01-10
3.3588
3.3777
19764948.0
NingboBank
2024-01-02
18.7598
18.3594
39575754.0
2024-01-03
18.3222
18.1546
37235409.0
2024-01-04
18.1453
17.7636
50880362.0
2024-01-05
17.6891
18.1546
56130083.0
2024-01-08
18.0243
17.7170
38023194.0
2024-01-09
17.7729
18.0708
42221761.0
2024-01-10
17.9870
18.0987
33169881.0
我们可以通过访问索引的 nlevels 属性来查看它有多少个级别：
print (f'行索引级别数: { stock_price_volume_multiindex_df. index. nlevels} ' ) # 验证行轴由证券代码、交易日两级键共同定位观测 print (f'列索引级别数: { stock_price_volume_multiindex_df. columns. nlevels} ' ) # 验证列轴由指标类别、具体度量两级标签组织数值
通过部分列索引，你同样可以选择列的分组。例如，要选择所有的 ‘Price’ 数据：
表 表 7.3 汇总本节的计算或审计结果，解释时应遵循正文给出的口径与限制。
'Price' ] # 通过外层列标签选取Price类别下的所有子列(Open和Close)
重排序和排序级别
有时，你可能需要重新排列轴上级别的顺序，或者根据某个特定级别中的值对数据进行排序。swaplevel() 方法接收两个级别编号或名称，并返回一个级别互换的新对象（但数据本身保持不变）。
表 表 7.4 汇总本节的计算或审计结果，解释时应遵循正文给出的口径与限制。
'Ticker' , 'TradeDate' ) # 交换多级索引层级
sort_index() 默认使用所有索引级别按字典顺序对数据进行排序。你可以选择按特定级别排序。例如，让我们按 TradeDate 排序。
表 表 7.5 汇总本节的计算或审计结果，解释时应遵循正文给出的口径与限制。
= 'TradeDate' ) # 按索引排序
性能优化：词典排序 (Lexicographical Sorting) 的重要性
在分层索引的对象上进行数据检索时，索引的排序状态直接决定了底层搜索算法的效率。
排序逻辑 ：pandas 默认首先按最外层索引排序，然后在每个外层分类内部对次级索引进行排序。性能差异 ：经过 sort_index() 排序后的索引可以利用二分查找等高效算法 ，避免昂贵的线性扫描。最佳实践 ：在执行复杂的切片（Slicing）或部分索引选择之前，习惯性地调用 sort_index() 能够显著缩短在大规模数据集上的响应时间。
按级别进行汇总统计
DataFrame 和 Series 上的许多描述性和汇总统计都有一个 level 选项，你可以用它来指定在特定轴上进行聚合的级别。思考一下我们在 表 7.2 中创建的股票 DataFrame。我们可以在行或列上按级别进行聚合。例如，让我们计算每个股票代码的所有指标的平均值。
表 表 7.6 汇总本节的计算或审计结果，解释时应遵循正文给出的口径与限制。
= 'Ticker' ).mean() # 按股票代码分组，计算每只股票各指标的均值
现在，让我们在列轴上计算每天每个类别的总和。
表 表 7.7 汇总本节的计算或审计结果，解释时应遵循正文给出的口径与限制。
# 这个操作只需要数值型数据 = 'Category' ).sum ().T.head() # 转置后按原列轴的Category级别聚合，再转回Ticker—TradeDate行索引
使用 DataFrame 的列进行索引
从文件中加载数据时，期望的索引通常存储在一个或多个普通列中。set_index 可以把这些列提升为索引。下面直接使用本地中国股票指数数据，保留每个字段原本的经济含义：close 始终表示指数收盘点位，volume 始终表示成交量，绝不把它们改称 GDP、人口或其他宏观变量。
表 表 7.8 汇总本节的计算或审计结果，解释时应遵循正文给出的口径与限制。
import pandas as pd # 复用 pandas 完成索引构造、日期解析和真实行情筛选 # 限定四个代表性指数并保留代码到名称的可追溯映射 = { # 为复用子集中的四个指数建立可读名称 '000001.XSHG' : '上证综指' , # 映射上证综合指数 '000300.XSHG' : '沪深300' , # 映射沪深300指数 '399001.XSHE' : '深证成指' , # 映射深证成份指数 '399006.XSHE' : '创业板指' # 映射创业板指数 # 完成扁平表展示所需名称字典 assert set (index_names).issubset(set (index_data['symbol' ].unique())) # 核验本章首次读取保留了扁平表示所需的四个指数 # 只投影构造分层索引所需字段，减少无关列进入示例 = index_data.loc[ # 从本章已缩减并复用的指数子集构造扁平表 'symbol' ].isin(index_names), # 只保留名称字典覆盖的四个指数 'symbol' , 'datetime' , 'close' , 'volume' ] # 投影层级索引和市场变量所需字段 # 隔离副本，避免后续日期解析修改复用对象 # 按数据源固定格式解析日期，避免字符串排序误当时间排序 'date' ] = pd.to_datetime( # 按来源合同创建日期列 'datetime' ).astype(str ), format = '%Y%m %d %H%M%S' # 解析十四位日期时间字符串 # 完成扁平表时间键标准化 'index_name' ] = index_flat_df['symbol' ].map (index_names) # 添加可读名称供后续层级展示 # 抽查关键字段完整的扁平行情记录
现在使用 set_index 从 symbol 和 date 创建 MultiIndex。第一层回答“哪一个指数”，第二层回答“哪一个交易日”。
表 表 7.9 汇总本节的计算或审计结果，解释时应遵循正文给出的口径与限制。
= index_flat_df.set_index(['symbol' , 'date' ]).sort_index() # 以指数代码和日期建立有序二维键 # 检查外层代码、内层日期与数据列的布局
默认情况下，用作索引的列会从 DataFrame 中移除。你可以通过传递 drop=False 来保留它们。
表 表 7.10 汇总本节的计算或审计结果，解释时应遵循正文给出的口径与限制。
'symbol' , 'date' ], drop= False ).head() # 保留键列以对照索引与原始字段是否一致
另一方面，reset_index 的作用与 set_index 相反；分层索引的级别会被移回到列中。
表 表 7.11 汇总本节的计算或审计结果，解释时应遵循正文给出的口径与限制。
# 将两层键还原为列，验证索引转换可逆
合并与连接数据集
pandas 对象中的数据可以通过几种方式进行组合：
pandas.mergeDataFrame 中的行。这是数据库 join 操作的基石。pandas.concatcombine_first
我们将通过应用实例来逐一介绍这些方法。
数据库风格的 DataFrame 连接
合并 (Merge) 或 连接 (Join) 操作通过使用一个或多个键来链接行，从而组合数据集。这些是关系型数据库（例如 SQL）中的基本操作。pandas.merge 函数是在你的数据上使用这些算法的主要入口点。
让我们构建一个不会改变变量含义的场景：分别准备沪深 300 指数的季度末点位表和季度成交量表，再按季度键合并。两个表来自同一真实数据源，但承担不同的数据管理职责，这与企业中“行情表”和“流动性表”分别维护的情形相似。
# 从全市场指数表提取沪深 300 的点位和成交量作为同源连接输入 = index_data.loc[ # 继续复用同一内存子集提取沪深300日行情 'symbol' ].eq('000300.XSHG' ), # 仅保留沪深300代码 'datetime' , 'close' , 'volume' ] # 投影季度点位和成交量所需字段 # 隔离副本供时间键转换与重采样 # 按十四位来源合同解析沪深 300 时间键，使季度重采样边界可复核 'date' ] = pd.to_datetime( # 将来源时间字段转换为重采样日期索引 'datetime' ).astype(str ), format = '%Y%m %d %H%M%S' # 依十四位来源合同解析时间 # 完成沪深300日期列构造 = hs300_daily_df.set_index('date' ).sort_index() # 建立单调时间索引以保证季度聚合顺序正确 = hs300_daily_df.loc['2015-01-01' :'2023-12-31' ] # 固定案例样本期以便结果复现 = hs300_daily_df['close' ].resample('QE' ).last().to_frame('quarter_end_close' ) # 取每季最后观测表示季末点位 = hs300_daily_df['volume' ].resample('QE' ).sum ().to_frame('quarter_volume' ) # 汇总季度成交量衡量交易活跃度 'year' ] = quarterly_close_df.index.year # 从季末日期派生年度连接键，同时保留原季度粒度 'year' ] = quarterly_volume_df.index.year # 为成交量表建立相同连接口径 print (quarterly_close_df.head()) # 核对季末点位表的日期唯一性和年份列 print (quarterly_volume_df.head()) # 核对季度成交量表使用同一季度键
quarter_end_close year
date
2015-03-31 4051.2040 2015
2015-06-30 4472.9976 2015
2015-09-30 3202.9475 2015
2015-12-31 3731.0047 2015
2016-03-31 3218.0879 2016
quarter_volume year
date
2015-03-31 1.532171e+12 2015
2015-06-30 2.678763e+12 2015
2015-09-30 1.804145e+12 2015
2015-12-31 1.079313e+12 2015
2016-03-31 7.219930e+11 2016
这是一个一对一 (one-to-one) 连接的例子，因为在按季度重采样后，两个DataFrame的日期索引都是唯一的。我们可以直接在日期索引上进行合并。
# 按唯一季度索引合并价格与流动性指标，并让 validate 审计键关系 = pd.merge( # 按唯一季度索引连接季末点位与季度成交量 = True ,= True ,= 'one_to_one' # 检查一对一合并后的列和季度覆盖
请注意，当在索引上合并时，我不需要指定要连接的列。如果是在列上连接，pandas.merge 会使用重叠的列名作为键。一个好的做法是使用 on 参数明确指定。让我们尝试在 year 列上合并。
表 表 7.13 汇总本节的计算或审计结果，解释时应遵循正文给出的口径与限制。
# 仅按年度会令两侧季度记录组内全组合；下方同时使用季度日键阻止笛卡尔扩张 # 同时使用日期和年份列，避免仅按年份形成季度笛卡尔积 # 显式使用日期与年份复合键完成一对一连接 = ['date' , 'year' ],= 'one_to_one' 8 )
默认情况下，pandas.merge 执行内连接 (inner join) ，结果只保留两张表共同拥有的键。其他选项是 'left'、'right' 和 'outer'。下面故意让沪深 300 点位表与成交量表覆盖不同的季度范围，用真实市场变量展示键的交集与并集。
= quarterly_close_df.loc['2018-01-01' :'2021-12-31' , ['quarter_end_close' ]] # 构造较早结束的左表以展示非共同键 = quarterly_volume_df.loc['2020-01-01' :'2023-12-31' , ['quarter_volume' ]] # 构造较晚开始的右表以形成部分重叠 print (close_window_df.head()) # 核对左表覆盖期起点 print (volume_window_df.head()) # 核对右表覆盖期起点及键差异
quarter_end_close
date
2018-03-31 3898.4977
2018-06-30 3510.9845
2018-09-30 3438.8649
2018-12-31 3010.6536
2019-03-31 3872.3412
quarter_volume
date
2020-03-31 9.479566e+11
2020-06-30 6.582232e+11
2020-09-30 1.249509e+12
2020-12-31 8.420723e+11
2021-03-31 1.115499e+12
现在，让我们执行一个 outer 连接。
表 表 7.14 汇总本节的计算或审计结果，解释时应遵循正文给出的口径与限制。
# 外连接保留两表季度并集，使未匹配键显式呈现为缺失值 # 对覆盖期不同的季度表执行外连接 = True ,= True ,= 'outer' ,= 'one_to_one'
在 outer 连接中，如果某个季度只出现在一张表中，另一张表对应的列就显示为 NaN。这不是“坏数据”，而是键集合不同的可见结果。表 7.15 总结了 how 选项。
多对多 (Many-to-many) 合并会形成匹配键的笛卡尔积。这是一个至关重要的概念。想象一下，你有一个按季度报告的公司财务数据集，和另一个这些公司的分析师评级数据集，其中多个分析师可能在任何给定的季度发布评级。
风险警示：量化建模中的笛卡尔积 (Cartesian Product) 隐患
在执行多对多合并（Many-to-many join）时，左表中每个键的每一行都会与右表中对应键的每一行进行笛卡尔式组合。
数据爆炸 ：如果左侧有 \(m\) 行键 \(k\) ，右侧有 \(n\) 行键 \(k\) ，合并结果将产生 \(m \times n\) 行。这在处理包含重复时间戳的高频订单簿数据或未清洗的财务报告时，容易导致内存瞬间耗尽。逻辑污染 ：笛卡尔积会在不经意间引入重复的计算样本，导致后续回归模型（如 Fama-French 三因子模型）的权重失真。避坑指南 ：在执行 merge() 之前，强烈建议使用 duplicated() 检查连接键的唯一性，或通过 validate='one_to_one' 等参数进行强制断言校验。
下面用一个明确标注为“结构演示”的最小订单—报价例子展示笛卡尔积。数值不用于任何经验结论；真实分析中应优先从第19章介绍的逐笔数据构造对应表。
表 表 7.16 汇总本节的计算或审计结果，解释时应遵循正文给出的口径与限制。
# 构造同一秒含两笔订单的机制题设以显示多对多行数膨胀 = pd.DataFrame({ # 构造含重复秒键的两笔演示订单 'symbol' : ['DEMO' , 'DEMO' ],'second' : ['09:30:01' , '09:30:01' ],'order_id' : ['O1' , 'O2' ],# 构造同一秒含两条报价的机制题设，与订单形成二乘二匹配 = pd.DataFrame({ # 构造含重复秒键的两条演示报价 'symbol' : ['DEMO' , 'DEMO' ],'second' : ['09:30:01' , '09:30:01' ],'quote_id' : ['Q1' , 'Q2' ],= ['symbol' , 'second' ], validate= 'many_to_many' ) # 显示普通等值连接产生的四行笛卡尔积
同一秒内左表有2行、右表也有2行，结果产生 \(2\times2=4\) 行。若业务规则要求“每笔订单只匹配最近一条报价”，这里就不应使用普通merge，而应先定义时间容忍度与方向，再考虑merge_asof。
连接时，你可能会有不属于连接键的重叠列名。pandas.merge 有一个 suffixes 选项，可以为这些列名附加字符串。
表 表 7.17 汇总本节的计算或审计结果，解释时应遵循正文给出的口径与限制。
= pd.DataFrame({'key' : ['a' , 'b' ], 'data' : [1 , 2 ]}) # 左表：含重叠列名 data = pd.DataFrame({'key' : ['a' , 'b' ], 'data' : [3 , 4 ]}) # 右表：同样含重叠列名 data = 'key' , suffixes= ('_left_data' , '_right_data' )) # 按key合并，重叠列名自动添加后缀以区分来源
pd.merge 的完整参数列表在 表 7.18 中提供。
按索引合并
正如 表 7.12 所示，合并键也可以位于索引中。传递 left_index=True 和 right_index=True 即可按索引连接。下面重新合并覆盖期不同的季度点位表与成交量表。
表 表 7.19 汇总本节的计算或审计结果，解释时应遵循正文给出的口径与限制。
# 按共同日期索引取交集，并强制检查每季仅有一条记录 # 按共同季度索引取得两表交集 = True ,= True ,= 'inner' ,= 'one_to_one'
对于分层索引的数据，按索引连接等同于多键合并。
DataFrame.join 方法提供了一种便捷的按索引合并的方式。它默认执行左连接。
方法进阶：merge() 与 .join() 在工程实践中的取舍
pandas 提供的这两种接口虽然功能交织，但在高性能量化系统开发中有明确的应用边界：
pd.merge() (通用底层函数)
核心特性 ：基于列名 的灵活连接。适用场景 ：处理复杂的异构数据集（如将“高管持股变动表”与“股价行情表”按 symbol 关联）。支持多键、不同列名匹配，以及通过 validate 参数进行严密的逻辑校验。 df.join() (高效实例方法)
核心特性 ：默认基于索引 的快速合并。性能逻辑 ：在已排序的日期索引上，join 的执行效率通常高于基于非索引列的 merge，因为它在底层利用了索引的指针加速。适用场景 ：多支股票收盘价序列（Panel Data）的快速横向对齐，或将主表与已构建好索引的静态特征表合并。
经验法则 ：当需要业务维度的关联（如外键映射）时，用 pd.merge()；当需要时间维度的对齐（如多指标拼接）时，优先考虑 df.join()。
让我们使用 .join() 来重现之前的合并操作。
表 表 7.20 汇总本节的计算或审计结果，解释时应遵循正文给出的口径与限制。
# .join() 默认执行左连接 # 默认左连接，保留点位表全部季度
为了得到与之前相同的 inner 连接结果，我们指定 how='inner'。
表 表 7.21 汇总本节的计算或审计结果，解释时应遵循正文给出的口径与限制。
= 'inner' , validate= 'one_to_one' ) # 用索引连接复现共同季度的一对一结果
沿轴连接
另一种组合方式是连接 (concatenation) 或堆叠 (stacking) 。pandas.concat 是主要函数。下面把沪深 300 的两个相邻真实日期片段重新拼成一条点位序列。
# 连接两个本地真实数据片段 = hs300_daily_df.loc['2018-01-01' :'2019-12-31' , 'close' ] # 提取较早的沪深300真实收盘片段 = hs300_daily_df.loc['2020-01-01' :'2021-12-31' , 'close' ] # 提取后一相邻样本片段 # 纵向恢复完整序列，并检查两个片段不存在重复日期 = pd.concat( # 纵向拼回两个相邻日期片段并审计重复键 = True print (hs300_close_full.head()) # 前部应来自 first_half，确认上半年日期未被后一片段覆盖 print ('...' ) # 省略中间部分 print (hs300_close_full.tail()) # 核对拼接序列延伸至后一片段末端
date
2018-01-02 4087.4012
2018-01-03 4111.3925
2018-01-04 4128.8119
2018-01-05 4138.7505
2018-01-08 4160.1595
Name: close, dtype: float64
...
date
2021-12-27 4919.3238
2021-12-28 4955.9644
2021-12-29 4883.4804
2021-12-30 4921.5109
2021-12-31 4940.3733
Name: close, dtype: float64
默认情况下，pd.concat 沿 axis=0（行索引）工作。如果你传递 axis=1（axis='columns'），结果将是一个 DataFrame。
表 表 7.22 汇总本节的计算或审计结果，解释时应遵循正文给出的口径与限制。
# 横向对齐两个不重叠时期，观察列拼接产生的结构性缺失 # 横向对齐两个不重叠时期的收盘序列 'close_2018_2019' ),'close_2020_2021' )],= 'columns'
一个潜在的问题是，在结果中无法识别连接的各个部分。假设你希望在连接轴上创建一个分层索引。为此，可以使用 keys 参数。
# 用样本期键标记每段来源，避免纵向拼接后丢失血缘 = pd.concat( # 用样本期键标记纵向拼接片段的来源 = ['2018—2019' , '2020—2021' ],= ['sample_period' , 'date' ]# 检查纵向拼接后 outer 层保留 first/second，内层保留原日期索引
sample_period date
2018—2019 2018-01-02 4087.4012
2018-01-03 4111.3925
2018-01-04 4128.8119
2018-01-05 4138.7505
2018-01-08 4160.1595
...
2020—2021 2021-12-27 4919.3238
2021-12-28 4955.9644
2021-12-29 4883.4804
2021-12-30 4921.5109
2021-12-31 4940.3733
Name: close, Length: 973, dtype: float64
同样的逻辑也适用于 DataFrame。下面把 2020 年的季度末点位与季度成交量沿列轴对齐。
= quarterly_close_df.loc['2020-01-01' :'2020-12-31' , ['quarter_end_close' ]] # 截取 2020 年季末点位 = quarterly_volume_df.loc['2020-01-01' :'2020-12-31' , ['quarter_volume' ]] # 截取同季度成交量 = 'columns' ) # 依日期横向对齐两类季度指标
date
2020-03-31
3686.1551
9.479566e+11
2020-06-30
4163.9637
6.582232e+11
2020-09-30
4587.3953
1.249509e+12
2020-12-31
5211.2885
8.420723e+11
最后一个需要考虑的问题是，DataFrame 的行索引不包含任何相关数据的情况。在这种情况下，你可以传递 ignore_index=True，这将丢弃原始索引并分配一个新的默认整数索引。
表 表 7.23 汇总本节的计算或审计结果，解释时应遵循正文给出的口径与限制。
= temp_close_chunk.reset_index(drop= True ) # 丢弃日期键以演示纯位置拼接 = temp_volume_chunk.reset_index(drop= True ) # 对右表采用相同整数位置口径 = True ) # 纵向合并异名列并生成连续行号
合并重叠数据
还有一种数据组合情况，既不能表示为合并也不能表示为连接操作。你可能有索引完全或部分重叠的两个数据集，并且你希望用一个数据集中的值来“修补”另一个数据集中的缺失值。这在处理来自不同来源且可能存在数据缺口的数据时是一项常见任务。combine_first 方法就是为此设计的。
让我们创建两个 Series，它们有一些重叠的索引和 NaN 值。
= pd.Series([np.nan, 10 , np.nan, 30 ], index= ['600276' , '002415' , '002230' , '002142' ], name= 'bid' ) # 买入价序列，部分股票缺失报价 = pd.Series([100 , np.nan, 300 , 400 ], index= ['600276' , '002415' , '002230' , '002142' ], name= 'ask' ) # 卖出价序列，另一部分股票缺失报价 # 当 bid_prices 中有 NaN 时，使用 ask_prices 中的值 # 用ask价格填补bid中的缺失值
600276 100.0
002415 10.0
002230 300.0
002142 30.0
Name: bid, dtype: float64
对于 DataFrame，combine_first 对每一列执行相同的操作。因此，你可以将其视为用你传递的对象中的数据“修补”调用对象中的缺失数据。
表 表 7.24 汇总本节的计算或审计结果，解释时应遵循正文给出的口径与限制。
= pd.DataFrame({'600276' : [150. , np.nan, 155. , np.nan], # 交易所A的报价数据，部分日期缺失 '002415' : [np.nan, 280. , np.nan, 285. ]}) # 交易所A的002415(海康威视)报价，也有缺失值 = pd.DataFrame({'600276' : [152. , 148. , np.nan, 153. , 158. ], # 交易所B的报价数据，行数不同 '002415' : [np.nan, 281. , 282. , 286. , 290. ]}) # 交易所B的002415(海康威视)报价 # 用交易所B的数据填补交易所A的缺失值
使用 DataFrame 对象时，combine_first 的输出将包含所有列名的并集。
重塑与透视
有许多用于重排表格数据的基本操作。这些操作被称为重塑 (reshape) 或透视 (pivot) 操作。
使用分层索引进行重塑
正如我们在 小节 7.2.1 DataFrame 中重排数据提供了一种一致的方式。两个主要操作是：
stackunstack
下面用沪深 300 季度末点位变化率和成交量变化率说明这些操作。两列都是市场变量；点位变化反映价格表现，成交量变化反映交易活跃度。
# 在共同季度样本上计算两类市场指标的环比变化 = quarterly_market_df.loc[ # 提取共同样本并计算点位与成交量环比变化 '2020-01-01' :'2021-12-31' , ['quarter_end_close' , 'quarter_volume' ]= None ).dropna()= 'Quarter' # 命名行维度以便堆叠后可按季度还原 = 'MarketMeasure' # 命名指标维度以保留变量语义 # 检查季度×指标宽表及首期差分删除结果
Quarter
2020-06-30
0.129622
-0.305640
2020-09-30
0.101690
0.898306
2020-12-31
0.136002
-0.326077
2021-03-31
-0.031264
0.324707
2021-06-30
0.034799
-0.258996
2021-09-30
-0.068464
0.513029
2021-12-31
0.015204
-0.293357
对这个数据使用 stack 方法会将列透视到行，生成一个带有 MultiIndex 的 Series。
= market_change_panel_df.stack() # 将指标列压入第二层行索引形成长序列 # 核对每个季度对应两项市场变化率
Quarter MarketMeasure
2020-06-30 quarter_end_close 0.129622
quarter_volume -0.305640
2020-09-30 quarter_end_close 0.101690
quarter_volume 0.898306
2020-12-31 quarter_end_close 0.136002
quarter_volume -0.326077
2021-03-31 quarter_end_close -0.031264
quarter_volume 0.324707
2021-06-30 quarter_end_close 0.034799
quarter_volume -0.258996
2021-09-30 quarter_end_close -0.068464
quarter_volume 0.513029
2021-12-31 quarter_end_close 0.015204
quarter_volume -0.293357
dtype: float64
从这个分层索引的 Series 中，你可以使用 unstack 将数据重新排列回 DataFrame。
表 表 7.25 汇总本节的计算或审计结果，解释时应遵循正文给出的口径与限制。
# 将指标层还原为列以验证重塑可逆
默认情况下，最内层的级别被 unstack。你可以通过传递级别编号或名称来 unstack 不同的级别。例如，让我们 unstack ‘Quarter’ 级别。
表 表 7.26 汇总本节的计算或审计结果，解释时应遵循正文给出的口径与限制。
= 'Quarter' ) # 指定季度层旋转为列以比较不同层级选择
将“长”格式透视为“宽”格式
在数据库和CSV文件中存储多个时间序列的一种常见方式是所谓的长格式 (long) 或堆叠格式 (stacked) 。在这种格式中，每一行都是一个单独的观测值。这通常是用于存储和某些类型分析的首选“整洁”数据格式。
继续使用真实的季度市场表创建长格式数据集。
# 保留完整季度的点位和成交量，作为宽长转换的统一输入 = quarterly_market_df.loc[ # 选择宽长转换所需两项真实季度指标 'quarter_end_close' , 'quarter_volume' ]= 'date' # 明确每行对应季度日期 = 'item' # 明确每列对应市场指标 # 堆叠以创建 long 格式 = ( # 依次堆叠、恢复索引列并命名数值列 = {0 : 'value' })10 ) # 核对每个日期—指标组合仅占一行
0
2015-03-31
quarter_end_close
4.051204e+03
1
2015-03-31
quarter_volume
1.532171e+12
2
2015-06-30
quarter_end_close
4.472998e+03
3
2015-06-30
quarter_volume
2.678763e+12
4
2015-09-30
quarter_end_close
3.202948e+03
5
2015-09-30
quarter_volume
1.804145e+12
6
2015-12-31
quarter_end_close
3.731005e+03
7
2015-12-31
quarter_volume
1.079313e+12
8
2016-03-31
quarter_end_close
3.218088e+03
9
2016-03-31
quarter_volume
7.219930e+11
在这种长格式中，每一行代表一个单一的观测值（特定日期的特定项目）。这是关系型数据库中常见的结构。然而，对于时间序列分析或某些类型的建模，你可能更喜欢宽格式 (wide) ，即每个不同的 item 都有一列。pivot 方法正是执行这种转换。
表 表 7.27 汇总本节的计算或审计结果，解释时应遵循正文给出的口径与限制。
= market_long_df.pivot(index= 'date' , columns= 'item' , values= 'value' ) # 以日期和指标唯一键恢复宽表 # 核对日期键成为行、指标名成为列，单元值仍为原 value
pivot 的前两个参数分别是将用作行索引和列索引的列。values 参数是包含要填充 DataFrame 的数据的列。如果省略 values，并且你有多个剩余的列，你将得到一个带有分层列的 DataFrame。
将“宽”格式透视为“长”格式
对于 DataFrame，pivot 的逆操作是 pandas.melt。它不是将一列转换为多列，而是将多列合并为一列，生成一个比输入更长的 DataFrame。
继续使用前面由本地沪深300数据形成的季度市场宽表，截取最近六个季度作为可核验的melt输入。
表 表 7.28 汇总本节的计算或审计结果，解释时应遵循正文给出的口径与限制。
= quarterly_market_wide_df.tail(6 ).reset_index() # 固定最近六个完整季度作为 melt 输入 # 核对标识列和两项待展开指标
date列是标识变量，收盘点位与成交量是待转成长格式的值列。使用pandas.melt时，必须明确id_vars，并可用value_vars限制进入结果的指标。
表 表 7.29 汇总本节的计算或审计结果，解释时应遵循正文给出的口径与限制。
# 保留日期标识，将两个指标归并为“指标—数值”两列 = market_melt_source_df.melt( # 将两项季度指标展开为指标—数值长表 = 'date' , var_name= 'market_measure' , value_name= 'value' # 检查长表行数等于季度数乘指标数
你可以使用 pivot 将其重塑回原始布局。
表 表 7.30 汇总本节的计算或审计结果，解释时应遵循正文给出的口径与限制。
# 依据唯一日期—指标组合恢复原始宽表布局 = market_melt_long_df.pivot( # 按唯一日期—指标组合恢复宽表 = 'date' , columns= 'market_measure' , values= 'value' # 还原日期列以便与 melt 输入逐列比较
你还可以指定一个列的子集作为值列（value_vars）。
表 表 7.31 汇总本节的计算或审计结果，解释时应遵循正文给出的口径与限制。
# 仅展开季末点位列，演示 value_vars 对指标范围的约束 # 仅将季末点位列展开为长表 = 'date' , value_vars= ['quarter_end_close' ],= 'market_measure' , value_name= 'value'
习题
习题 7.1: 数据合并基础
问题描述 ：
假设你有两个数据集：
数据集A：宁波港 (601018.SH) 的日度交易数据（日期、开盘价、收盘价）
数据集B：宁波港的日度成交量数据（日期、成交量）
请完成以下任务：
使用 merge 函数按日期合并两个数据集
使用不同类型的连接（inner、left、right、outer）并比较结果
使用 join 方法合并两个数据集
使用 concat 函数合并两个数据集
完整解答 ：
import pandas as pd # 用宁波港价格表与成交量子表对照四种连接 import numpy as np # 为连接产生的结构性缺失提供统一数值表示 # 从复权日行情中构造共用的宁波港交易日母表 = pd.read_hdf( # 取得 601018.XSHG 的日度价格与成交量 f' { DATA_ROOT} /stock/stock_price_pre_adjusted.h5' , # 来源表以 order_book_id—date 区分证券日记录 = "order_book_id='601018.XSHG'" # 在 HDF 层只下推宁波港记录 = {'date' : 'trade_date' , 'vol' : 'volume' }) # 恢复日期键并将 vol 统一为 volume 'datetime' ] = pd.to_datetime(stock_data['trade_date' ], format = '%Y%m %d ' ) # 按8位交易日合同解析连接键 # 两张连接输入共用宁波港2023年1月真实交易日母表 = stock_data[(stock_data['datetime' ] >= '2023-01-01' ) & (stock_data['datetime' ] <= '2023-01-31' )].copy() # 固定宁波港2023年1月样本，不改写来源底表 # 创建数据集A：价格数据 = port_data_jan[['datetime' , 'open' , 'close' ]].copy() # 提取日期、开盘价、收盘价列 = port_price_data.rename(columns= {'datetime' : 'date' }) # 将价格表的交易日键命名为 date # 从第六个交易日起保留成交量，使左连时前五个价格日产生可解释的结构性缺失 = port_data_jan[['datetime' , 'volume' ]].iloc[5 :].copy() # 从第6行开始截取，模拟数据缺失 = port_volume_data.rename(columns= {'datetime' : 'date' }) # 将截短成交量表的连接键同样命名为 date print ('=== 数据集信息 ===' ) # 先报告两张输入表规模，为连接后的行数核验提供基准 print (f'数据集A形状: { port_price_data. shape} ' ) # 记录完整价格日历的基准行数与 date/open/close 三列 schema print (f'数据集B形状: { port_volume_data. shape} ' ) # 确认成交量表少五个 date 键，为后续连接的样本差异提供基准
=== 数据集信息 ===
数据集A形状: (16, 3)
数据集B形状: (11, 2)
1. 使用 merge 函数
print (' \n === 1. Merge 不同连接类型 ===' ) # 分隔基于列键的四种连接结果 # Inner join（默认） = pd.merge(port_price_data, port_volume_data, on= 'date' , how= 'inner' ) # 内连接：仅保留两表共有日期 print (f' \n Inner join 形状: { merged_inner. shape} ' ) # 验证内连接仅保留共同交易日 print (merged_inner.head()) # 抽查内连接中价格与成交量均非缺失的最早共同日 # 价格日历为主样本：前5日无成交量匹配时保留为 NA = pd.merge(port_price_data, port_volume_data, on= 'date' , how= 'left' ) # 价格表 date 集合全部保留，成交量表缺少的前5日填为 NA print (f' \n Left join 形状: { merged_left. shape} ' ) # 验证左连接保留全部价格日期 print (f'Left join 缺失值: { merged_left. isna(). sum (). sum ()} ' ) # 缺失数应对应成交量表被截去的五个 date 键 # 截短成交量日历为主样本：不保留右表缺少的前5日 = pd.merge(port_price_data, port_volume_data, on= 'date' , how= 'right' ) # 右连接：保留右表全部日期 print (f' \n Right join 形状: { merged_right. shape} ' ) # 验证右连接以成交量日期集合为准 print (f'Right join 缺失值: { merged_right. isna(). sum (). sum ()} ' ) # 右表 date 均能命中价格表，因此本例不应新增字段缺失 # 取两个日期键集合的并集，使非共同日的结构性缺失可见 = pd.merge(port_price_data, port_volume_data, on= 'date' , how= 'outer' ) # 取价格与成交量日期键并集；本例右表是左表子集，行数应等于价格表 print (f' \n Outer join 形状: { merged_outer. shape} ' ) # 验证外连接保留两表日期并集 print (f'Outer join 缺失值: { merged_outer. isna(). sum (). sum ()} ' ) # 右键集合是左键子集，外连接的缺失仍只来自最早五日 volume
=== 1. Merge 不同连接类型 ===
Inner join 形状: (11, 4)
date open close volume
0 2023-01-10 3.2669 3.2485 5882400.0
1 2023-01-11 3.2394 3.2394 4453499.0
2 2023-01-12 3.2394 3.2302 4491700.0
3 2023-01-13 3.2302 3.2577 4929600.0
4 2023-01-16 3.2669 3.2761 8098816.0
Left join 形状: (16, 4)
Left join 缺失值: 5
Right join 形状: (11, 4)
Right join 缺失值: 0
Outer join 形状: (16, 4)
Outer join 缺失值: 5
2. 使用 join 方法（基于索引）
print (' \n === 2. Join 方法 ===' ) # 单列展示索引连接，便于与 merge 的列键语义对照 = port_price_data.set_index('date' ) # 将date列设为索引，便于基于索引合并 = port_volume_data.set_index('date' ) # 同样将date设为索引 = port_price_data_indexed.join(port_volume_data_indexed, how= 'outer' ) # 基于日期索引外连接两表 print (f'Joined 形状: { joined. shape} ' ) # 索引外连接应保留 date 并集并生成 open、close、volume 三个值列 print (joined.head()) # 最早价格日对应的 volume 应为 NA，以显示右表截短造成的缺口
=== 2. Join 方法 ===
Joined 形状: (16, 3)
open close volume
date
2023-01-03 3.2761 3.2761 NaN
2023-01-04 3.2669 3.2944 NaN
2023-01-05 3.2944 3.2852 NaN
2023-01-06 3.2852 3.2669 NaN
2023-01-09 3.2669 3.2669 NaN
3. 使用 concat 函数
print (' \n === 3. Concat 方法 ===' ) # 单列展示按轴拼接，不将其误解为数据库键连接 # 沿着列方向合并 = pd.concat([port_price_data_indexed, port_volume_data_indexed], axis= 1 ) # 按列方向拼接价格与成交量数据 print (f'Concatenated 形状: { concatenated. shape} ' ) # 核对按索引横拼得到 date 行轴与 open/close/volume 三列 schema print (concatenated.head()) # 核对 axis=1 与索引外连接保留相同日期集合和结构性 NA
=== 3. Concat 方法 ===
Concatenated 形状: (16, 3)
open close volume
date
2023-01-03 3.2761 3.2761 NaN
2023-01-04 3.2669 3.2944 NaN
2023-01-05 3.2944 3.2852 NaN
2023-01-06 3.2852 3.2669 NaN
2023-01-09 3.2669 3.2669 NaN
4. 比较不同方法
print (' \n === 4. 方法对比 ===' ) # 汇总各方法的输出规模和日期保留规则 = pd.DataFrame({ # 构建合并方法对比表 '方法' : ['merge_inner' , 'merge_left' , 'merge_right' , 'merge_outer' , 'join' , 'concat' ], # 六种合并方式名称 '形状' : [ # 各方法结果的行列维度 str (merged_inner.shape), # 共同 date 键产生的行列数 str (merged_left.shape), # 保留完整价格日历后的表形状 str (merged_right.shape), # 以成交量日期为准时的样本大小 str (merged_outer.shape), # 日期并集下的最大行数与字段数 str (joined.shape), # 索引外连接产生的 date 集合和四列 str (concatenated.shape) # concat结果维度 '保留所有日期' : ['否' , '是（来自A）' , '是（来自B）' , '是' , '是' , '是' ] # 各方法对日期的保留策略 print (comparison) # 对照六种方法的输出形状与 date 键保留规则，识别 inner/left/right 的样本变化
=== 4. 方法对比 ===
方法 形状 保留所有日期
0 merge_inner (11, 4) 否
1 merge_left (16, 4) 是（来自A）
2 merge_right (11, 4) 是（来自B）
3 merge_outer (16, 4) 是
4 join (16, 3) 是
5 concat (16, 3) 是
关键要点 ： - merge 基于列值合并，最灵活，适合不同索引的数据集 - join 基于索引合并，语法更简洁 - concat 沿着轴简单拼接，不做键值匹配 - Inner join 只保留两个数据集都有的键 - Outer join 保留所有键，可能产生缺失值 - Left/Right join 保留左/右数据集的所有键
习题 7.2: 多对一合并
问题描述 ：
在实际应用中，经常需要将包含多个观测的数据集与包含参考信息的数据集合并。请：
创建包含股票代码和公司名称的参考数据
创建包含多只股票（宁波港、宁波银行、恒瑞医药）交易数据的数据集
将公司名称映射到交易数据上
分析合并后的数据
完整解答 ：
import pandas as pd # 将多行交易事实表连接到唯一公司维度表 import numpy as np # 支持连接后缺失属性的统计诊断
1. 创建参考数据：股票代码与公司名称映射
= pd.DataFrame({ # 构建股票代码与公司信息的参考表 'stock_code' : ['601018.SH' , '002142.SZ' , '600276.SH' , '600000.SH' , '002230.SZ' ], # 五只长三角上市公司代码 'company_name' : ['宁波港' , '宁波银行' , '恒瑞医药' , '浦发银行' , '科大讯飞' ], # 对应公司中文名称 'sector' : ['交通运输' , '金融服务' , '医药' , '金融服务' , '计算机/软件' ], # 使用本题统一的行业分类口径 'city' : ['宁波' , '宁波' , '连云港' , '上海' , '合肥' ] # 使用公司总部所在长三角城市 = {'交通运输' , '金融服务' , '医药' , '计算机/软件' } # 固定本题维表允许出现的行业类别集合 = ['stock_code' , 'company_name' , 'sector' , 'city' ] # 集中声明连接所依赖的主键与维度属性 assert reference_data['stock_code' ].is_unique # 主键重复会把多对一连接意外扩张为多对多连接 assert reference_data[dimension_columns].notna().all ().all () # 禁止主键和维度属性出现缺失值 assert reference_data[dimension_columns].apply (lambda dimension_values: dimension_values.astype('string' ).str .strip().ne('' ).all ()).all () # 禁止空字符串或纯空白维度值 assert reference_data['sector' ].isin(allowed_sectors).all () # 阻止拼写漂移或未批准类别污染行业分组 print ('=== 参考数据 ===' ) # 展示多对一连接中每个代码唯一对应的公司维度表 print (reference_data) # 核对 stock_code 在维表中唯一且一行携带 company_name、sector、city 三项属性
=== 参考数据 ===
stock_code company_name sector city
0 601018.SH 宁波港 交通运输 宁波
1 002142.SZ 宁波银行 金融服务 宁波
2 600276.SH 恒瑞医药 医药 连云港
3 600000.SH 浦发银行 金融服务 上海
4 002230.SZ 科大讯飞 计算机/软件 合肥
本题把这五行作为固定的教学维表快照，并统一采用“医药”和“计算机/软件”口径。生产环境不应手工维护这种无血缘常量，而应从版本化公司基础表（例如由 stock_basic_data.h5 定期生成的快照）读取，并至少保存分类标准版本、来源快照标识、effective_from 与 effective_to。若行业会随时间变化，应以交易日落入的有效区间连接相应版本，避免用当前分类回填历史记录。
2. 创建交易数据
# 从同一行情底表组装三只证券的多行事实表 = pd.read_hdf( # 交通运输事实子表，后续匹配唯一公司属性 f' { DATA_ROOT} /stock/stock_price_pre_adjusted.h5' , # 三个子表共享复权行情口径 = "order_book_id='601018.XSHG'" # 取 601018.XSHG，该代码在参考表应只出现一次 = {'date' : 'trade_date' , 'vol' : 'volume' }) # 恢复宁波港事实日键，并与维度连接输入对齐 volume = pd.read_hdf( # 金融服务事实子表，用于测试同城不同公司映射 f' { DATA_ROOT} /stock/stock_price_pre_adjusted.h5' , # 沿用相同复权价格与字段合同 = "order_book_id='002142.XSHE'" # 取 002142.XSHE，规范为 .SZ 后再连公司维度 = {'date' : 'trade_date' , 'vol' : 'volume' }) # 恢复宁波银行事实日键，并与维度连接输入对齐 volume = pd.read_hdf( # 医药制造事实子表，增加跨行业维度检验 f' { DATA_ROOT} /stock/stock_price_pre_adjusted.h5' , # 第三子表保持同源复权口径 = "order_book_id='600276.XSHG'" # 取 600276.XSHG，核对它不会误匹配另一个 .SH 代码 = {'date' : 'trade_date' , 'vol' : 'volume' }) # 恢复恒瑞医药日期键并对齐 volume 列 'symbol' ] = '601018.SH' # 将宁波港来源代码规范为维度表键 'symbol' ] = '002142.SZ' # 将宁波银行来源代码规范为维度表键 'symbol' ] = '600276.SH' # 将恒瑞医药来源代码规范为公司维度键 = pd.concat([port_stock_data, bank_stock_data, hengrui_stock_data]) # 合并三只股票的行情数据 'datetime' ] = pd.to_datetime(stock_data['trade_date' ], format = '%Y%m %d ' ) # 将交易日期转为datetime格式 = stock_data[(stock_data['datetime' ] >= '2023-01-01' ) & (stock_data['datetime' ] <= '2023-01-31' )] # 公司维度连接只使用2023年1月多行交易事实 # 投影 datetime、stock_code、close 与 volume，形成多对一连接的左侧事实 schema = ['601018.SH' , '002142.SZ' , '600276.SH' ] # 事实表需要匹配公司维度的三个代码键 = stock_data[stock_data['symbol' ].isin(symbols)][['datetime' , 'symbol' , 'close' , 'volume' ]].copy() # 提取目标股票的日期、代码、收盘价、成交量 = trading_data.rename(columns= {'symbol' : 'stock_code' }) # 将symbol列重命名为stock_code以便合并 print (' \n === 交易数据示例 ===' ) # 抽查连接前的多行交易事实表及其代码键 print (trading_data.head(10 )) # 抽查事实表为一代码多交易日，且连接前 schema 仅含日期、代码、收盘价与成交量
=== 交易数据示例 ===
datetime stock_code close volume
2981 2023-01-03 601018.SH 3.2761 10743034.0
2982 2023-01-04 601018.SH 3.2944 8787221.0
2983 2023-01-05 601018.SH 3.2852 10407133.0
2984 2023-01-06 601018.SH 3.2669 10450610.0
2985 2023-01-09 601018.SH 3.2669 6518398.0
2986 2023-01-10 601018.SH 3.2485 5882400.0
2987 2023-01-11 601018.SH 3.2394 4453499.0
2988 2023-01-12 601018.SH 3.2302 4491700.0
2989 2023-01-13 601018.SH 3.2577 4929600.0
2990 2023-01-16 601018.SH 3.2761 8098816.0
3. 多对一合并
# 将公司信息映射到每一笔交易记录 = pd.merge( # 把唯一公司维度映射到每一笔交易事实 = 'stock_code' ,= 'left' ,= 'many_to_one' print (' \n === 合并后数据 ===' ) # 验证多对一连接保持交易事实行数，并为每行补入唯一公司属性 print (merged_data.head(15 )) # 抽查每笔交易的 stock_code 是否唯一带入 company_name、sector 和 city
=== 合并后数据 ===
datetime stock_code close volume company_name sector city
0 2023-01-03 601018.SH 3.2761 10743034.0 宁波港 交通运输 宁波
1 2023-01-04 601018.SH 3.2944 8787221.0 宁波港 交通运输 宁波
2 2023-01-05 601018.SH 3.2852 10407133.0 宁波港 交通运输 宁波
3 2023-01-06 601018.SH 3.2669 10450610.0 宁波港 交通运输 宁波
4 2023-01-09 601018.SH 3.2669 6518398.0 宁波港 交通运输 宁波
5 2023-01-10 601018.SH 3.2485 5882400.0 宁波港 交通运输 宁波
6 2023-01-11 601018.SH 3.2394 4453499.0 宁波港 交通运输 宁波
7 2023-01-12 601018.SH 3.2302 4491700.0 宁波港 交通运输 宁波
8 2023-01-13 601018.SH 3.2577 4929600.0 宁波港 交通运输 宁波
9 2023-01-16 601018.SH 3.2761 8098816.0 宁波港 交通运输 宁波
10 2023-01-17 601018.SH 3.2761 4898283.0 宁波港 交通运输 宁波
11 2023-01-18 601018.SH 3.2852 7279200.0 宁波港 交通运输 宁波
12 2023-01-19 601018.SH 3.3036 6843700.0 宁波港 交通运输 宁波
13 2023-01-20 601018.SH 3.3403 15032977.0 宁波港 交通运输 宁波
14 2023-01-30 601018.SH 3.3403 14658336.0 宁波港 交通运输 宁波
4. 分析合并后的数据
print (' \n === 按公司统计 ===' ) # 分隔公司层面的价格区间与成交量汇总 = merged_data.groupby('company_name' ).agg({ # 每个公司维表键汇总其多行交易事实，输出价格水平与成交量统计 'close' : ['mean' , 'min' , 'max' ], # 收盘价的均值、最小、最大 'volume' : 'sum' # 累加每家公司在样本期的日成交量 round (2 ) # 公司价格区间与成交量汇总统一两位展示精度 print (company_stats) # 核对 company_name 分组继承多对一连接的公司边界，并输出两层聚合列索引 print (' \n === 按行业统计 ===' ) # 分隔行业层面的合并后聚合结果 = merged_data.groupby('sector' ).agg({ # 将多个公司归入行业维度，比较跨公司平均价格与累计成交量 'close' : 'mean' , # 行业平均收盘价 'volume' : 'sum' # 行业总成交量 round (2 ) # 行业聚合表保留两位便于横向对照 print (sector_stats) # 对照各行业的公司数、平均收盘价与总成交量，检查维表分组是否生效 print (' \n === 按城市统计 ===' ) # 分隔总部城市层面的样本描述统计 = merged_data.groupby('city' ).agg({ # 以总部城市归并公司事实行，检查地理维表属性的分组覆盖 'close' : 'mean' , # 城市平均收盘价 'volume' : 'sum' # 城市总成交量 round (2 ) # 城市层描述统计用两位小数输出 print (city_stats) # 核对 city 维度已成功映射到全部事实行，再比较城市层价格与成交量汇总
=== 按公司统计 ===
close volume
mean min max sum
company_name
宁波港 3.28 3.23 3.37 132777579.0
宁波银行 30.37 29.38 31.20 447477998.0
恒瑞医药 40.14 37.34 43.12 885924199.0
=== 按行业统计 ===
close volume
sector
交通运输 3.28 132777579.0
医药 40.14 885924199.0
金融服务 30.37 447477998.0
=== 按城市统计 ===
close volume
city
宁波 16.83 580255577.0
连云港 40.14 885924199.0
5. 验证合并完整性
print (' \n === 合并完整性检查 ===' ) # 对照合并前后行数、缺失和公司覆盖以审计连接质量 print (f'交易数据行数: { len (trading_data)} ' ) # 记录多对一连接前的事实表基数 print (f'合并数据行数: { len (merged_data)} ' ) # 与交易事实表行数对照，确认多对一维表连接未扩张样本 print (f'缺失值数量: { merged_data. isna(). sum (). sum ()} ' ) # 验证每个股票代码均命中公司维表，连接字段没有空缺 print (f'唯一公司数: { merged_data. company_name. nunique()} ' ) # 统计合并后包含的不同公司数量 assert len (merged_data) == len (trading_data) # 多对一连接必须保持左侧交易事实行数不变 assert merged_data[dimension_columns].notna().all ().all () # 每条交易记录都必须命中完整的公司维度属性 assert set (merged_data['sector' ]).issubset(allowed_sectors) # 合并后的行业值仍须服从维表类别合同
=== 合并完整性检查 ===
交易数据行数: 48
合并数据行数: 48
缺失值数量: 0
唯一公司数: 3
关键要点 ：
多对一合并是最常见的场景之一
参考数据通常包含维度属性（行业、地区等）
how='left' 确保保留所有交易记录合并后可以进行多维度分析
注意检查合并的完整性（缺失值、行数等）
习题 7.3: 多对多合并
问题描述 ：
多对多合并会产生两个数据集键的笛卡尔积。请：
创建员工-项目关系数据（一个员工可以参与多个项目）
创建项目-技能要求数据（一个项目需要多种技能）
创建员工已掌握技能数据，并明确连接键
执行多对多合并，汇总员工技能需求
按“所需技能集合减去已掌握技能集合”识别技能缺口
完整解答 ：
import pandas as pd # 构造员工—项目与项目—技能两张关系表并按项目键展开 import numpy as np # 支持连接结果中的数值计数与缺失表示
1. 创建员工-项目数据
= pd.DataFrame({ # 构建员工-项目关联关系表 'employee_id' : ['E001' , 'E001' , 'E002' , 'E002' , 'E003' , 'E003' , 'E004' ], # 员工编号，一人可对应多个项目 'employee_name' : ['张三' , '张三' , '李四' , '李四' , '王五' , '王五' , '赵六' ], # 员工姓名 'project_id' : ['P001' , 'P002' , 'P001' , 'P003' , 'P002' , 'P003' , 'P001' ], # 参与的项目编号 'role' : ['开发者' , '项目经理' , '开发者' , '分析师' , '测试员' , '开发者' , '设计师' ] # 员工在项目中的角色 print ('=== 员工-项目关系 ===' ) # 展示多对多连接左表中的员工—项目关系 print (employee_projects) # 核对左表一名员工可有多个 project_id，且员工—项目组合构成一行
=== 员工-项目关系 ===
employee_id employee_name project_id role
0 E001 张三 P001 开发者
1 E001 张三 P002 项目经理
2 E002 李四 P001 开发者
3 E002 李四 P003 分析师
4 E003 王五 P002 测试员
5 E003 王五 P003 开发者
6 E004 赵六 P001 设计师
2. 创建项目-技能需求数据
= pd.DataFrame({ # 构建项目-技能需求关联表 'project_id' : ['P001' , 'P001' , 'P001' , 'P002' , 'P002' , 'P003' , 'P003' ], # 项目编号，一个项目可需多种技能 'required_skill' : ['Python' , 'SQL' , '机器学习' , '项目管理' , '沟通能力' , '数据分析' , '可视化' ], # 项目所需技能 'proficiency_level' : ['高级' , '中级' , '中级' , '高级' , '中级' , '高级' , '中级' ] # 技能熟练度要求 print (' \n === 项目-技能需求 ===' ) # 展示多对多连接右表中的项目—技能关系 print (project_skills) # 核对右表一个 project_id 可对应多项技能，为多对多展开提供键频数
=== 项目-技能需求 ===
project_id required_skill proficiency_level
0 P001 Python 高级
1 P001 SQL 中级
2 P001 机器学习 中级
3 P002 项目管理 高级
4 P002 沟通能力 中级
5 P003 数据分析 高级
6 P003 可视化 中级
3. 创建员工已掌握技能数据
= pd.DataFrame({ # 构建一行一个员工—已掌握技能的关系表 'employee_id' : ['E001' , 'E001' , 'E001' , 'E002' , 'E002' , 'E002' , 'E003' , 'E003' , 'E003' , 'E004' , 'E004' ], # employee_id 连接到员工主键 'owned_skill' : ['Python' , 'SQL' , '项目管理' , 'Python' , 'SQL' , '数据分析' , '沟通能力' , '数据分析' , '可视化' , 'Python' , '机器学习' ] # 技能值与 required_skill 采用同一词表 assert not employee_owned_skills.duplicated(['employee_id' , 'owned_skill' ]).any () # 一个员工—技能键只能出现一次 assert set (employee_owned_skills['employee_id' ]).issubset(set (employee_projects['employee_id' ])) # 每条技能记录都必须命中员工主键 print (' \n === 员工已掌握技能 ===' ) # 展示后续 required-minus-owned 运算的右侧关系表 print (employee_owned_skills) # 核对连接键由 employee_id 与标准化技能值共同组成
=== 员工已掌握技能 ===
employee_id owned_skill
0 E001 Python
1 E001 SQL
2 E001 项目管理
3 E002 Python
4 E002 SQL
5 E002 数据分析
6 E003 沟通能力
7 E003 数据分析
8 E003 可视化
9 E004 Python
10 E004 机器学习
员工—项目表与项目—技能表先按 project_id 连接；技能需求再与已掌握技能表按 employee_id 和技能值连接。这里将 owned_skill 重命名为 required_skill 只是为了显式对齐第二个连接键，不改变业务含义。
4. 多对多合并
= pd.merge( # 按项目键展开每位参与者承担的全部技能需求 = 'project_id' ,= 'inner' ,= 'many_to_many' print (' \n === 多对多合并结果 ===' ) # 对照输入与输出行数以观察键内笛卡尔积 print (f'合并前: 员工-项目 { len (employee_projects)} 行, 项目-技能 { len (project_skills)} 行' ) # 记录两表按 project_id 分组前的键频数基准 print (f'合并后: { len (merged)} 行' ) # 核对输出行数等于各项目左右键频数乘积之和 print (' \n 合并数据示例:' ) # 展示每位员工因项目技能展开后的明细行 print (merged) # 逐项目检查每位参与者与每项技能的组内组合，解释连接后的行数扩张
=== 多对多合并结果 ===
合并前: 员工-项目 7 行, 项目-技能 7 行
合并后: 17 行
合并数据示例:
employee_id employee_name project_id role required_skill proficiency_level
0 E001 张三 P001 开发者 Python 高级
1 E001 张三 P001 开发者 SQL 中级
2 E001 张三 P001 开发者 机器学习 中级
3 E001 张三 P002 项目经理 项目管理 高级
4 E001 张三 P002 项目经理 沟通能力 中级
5 E002 李四 P001 开发者 Python 高级
6 E002 李四 P001 开发者 SQL 中级
7 E002 李四 P001 开发者 机器学习 中级
8 E002 李四 P003 分析师 数据分析 高级
9 E002 李四 P003 分析师 可视化 中级
10 E003 王五 P002 测试员 项目管理 高级
11 E003 王五 P002 测试员 沟通能力 中级
12 E003 王五 P003 开发者 数据分析 高级
13 E003 王五 P003 开发者 可视化 中级
14 E004 赵六 P001 设计师 Python 高级
15 E004 赵六 P001 设计师 SQL 中级
16 E004 赵六 P001 设计师 机器学习 中级
5. 汇总员工技能需求
print (' \n === 按员工统计所需技能 ===' ) # 汇总员工因参与项目而对应的技能集合 = merged.groupby('employee_name' )['required_skill' ].apply (list ).reset_index() # 按员工汇总项目带来的技能需求列表 = ['员工' , '所需技能' ] # 将分组键与需求列表改为报告用中文表头 print (employee_required_skills) # 核对该表只表示需求，不能误读为员工已经掌握的技能 print (' \n === 按项目统计参与员工 ===' ) # 汇总每个项目去重后的参与人数 = merged.groupby('project_id' )['employee_name' ].nunique().reset_index() # 按项目分组，统计参与的不同员工数 = ['项目ID' , '参与员工数' ] # 明确两列分别是项目键与去重员工数 print (project_employees) # 核对项目键聚合后每行是一项项目，计数口径为去重员工数 print (' \n === 项目技能需求明细 ===' ) # 联合展示项目、员工、技能及熟练度要求 = merged.groupby('project_id' ).agg({ # 按项目分组进行多列聚合 'employee_name' : lambda x: ', ' .join(sorted (set (x))), # 拼接去重后的员工姓名 'required_skill' : lambda x: ', ' .join(sorted (x)), # 拼接全部所需技能名称 'proficiency_level' : lambda x: ', ' .join(x) # 拼接对应的熟练度要求 # 将项目ID从索引还原为普通列 = ['项目ID' , '参与员工' , '所需技能' , '技能等级' ] # 明确一行一项目及三列逗号拼接明细的输出模式 print (project_detail) # 核对输出 schema 为项目键加员工、技能、等级三列汇总文本
=== 按员工统计所需技能 ===
员工 所需技能
0 张三 [Python, SQL, 机器学习, 项目管理, 沟通能力]
1 李四 [Python, SQL, 机器学习, 数据分析, 可视化]
2 王五 [项目管理, 沟通能力, 数据分析, 可视化]
3 赵六 [Python, SQL, 机器学习]
=== 按项目统计参与员工 ===
项目ID 参与员工数
0 P001 3
1 P002 2
2 P003 2
=== 项目技能需求明细 ===
项目ID 参与员工 所需技能 \
0 P001 张三, 李四, 赵六 Python, Python, Python, SQL, SQL, SQL, 机器学习, 机...
1 P002 张三, 王五 沟通能力, 沟通能力, 项目管理, 项目管理
2 P003 李四, 王五 可视化, 可视化, 数据分析, 数据分析
技能等级
0 高级, 中级, 中级, 高级, 中级, 中级, 高级, 中级, 中级
1 高级, 中级, 高级, 中级
2 高级, 中级, 高级, 中级
6. 创建技能需求矩阵
= merged.pivot_table( # 创建员工—所需技能交叉矩阵 = 'employee_name' , # 行索引为员工姓名 = 'required_skill' , # 列索引为技能名称 = 'project_id' , # 用项目ID作为填充值 = 'count' , # 计数聚合，表示涉及项目数 = 0 # 以 0 表示员工—技能组合未参与任何项目，避免空值被误作未知 print (' \n === 技能需求矩阵 ===' ) # 展示员工—技能需求计数以定位训练任务 print ('(数字表示该员工需要该技能的项目数)' ) # 说明矩阵的含义 print (required_skill_matrix) # 检查每个员工—技能单元的项目需求计数，无需求的组合应为0
=== 技能需求矩阵 ===
(数字表示该员工需要该技能的项目数)
required_skill Python SQL 可视化 数据分析 机器学习 沟通能力 项目管理
employee_name
张三 1 1 0 0 1 1 1
李四 1 1 1 1 1 0 0
王五 0 0 1 1 0 1 1
赵六 1 1 0 0 1 0 0
7. 计算 required-minus-owned 技能缺口
= employee_owned_skills.rename(columns= {'owned_skill' : 'required_skill' }).assign(has_owned_skill= True ) # 对齐员工与技能两列连接键 = merged.merge( # 左连接保留全部项目需求，并标记已掌握的员工—技能组合 = ['employee_id' , 'required_skill' ],= 'left' ,= 'many_to_one' 'has_owned_skill' ] = skill_coverage_detail['has_owned_skill' ].notna() # 未命中已掌握技能表的需求即为缺口 = skill_coverage_detail.loc[ # 反连接得到所需技能集合减去已掌握技能集合 ~ skill_coverage_detail['has_owned_skill' ],'employee_id' , 'employee_name' , 'required_skill' ]= skill_gap_detail.groupby( # 把明细缺口整理为一行一员工的可读集合 'employee_id' , 'employee_name' ], sort= False 'required_skill' ].agg(lambda required_skills: sorted (set (required_skills))).reset_index(name= '技能缺口' )print (' \n === 员工技能缺口（required minus owned）===' ) # 明示结果是集合差而非技能需求频次 print (employee_skill_gaps) # 输出员工主键、姓名和去重后的技能缺口
=== 员工技能缺口（required minus owned）===
employee_id employee_name 技能缺口
0 E001 张三 [机器学习, 沟通能力]
1 E002 李四 [可视化, 机器学习]
2 E003 王五 [项目管理]
3 E004 赵六 [SQL]
张三参与 P001 与 P002，需要 {Python, SQL, 机器学习, 项目管理, 沟通能力}，已掌握 {Python, SQL, 项目管理}，所以人工复算缺口为 {机器学习, 沟通能力}；赵六只参与 P001，已掌握 {Python, 机器学习}，所以缺口为 {SQL}。下面把这两项人工计算和全表缺口数固化为断言。
= set (employee_skill_gaps.loc[employee_skill_gaps['employee_id' ].eq('E001' ), '技能缺口' ].iloc[0 ]) # 取张三缺口集合供人工结果对照 = set (employee_skill_gaps.loc[employee_skill_gaps['employee_id' ].eq('E004' ), '技能缺口' ].iloc[0 ]) # 取赵六缺口集合供单项目结果对照 assert zhang_skill_gaps == {'机器学习' , '沟通能力' } # 核验张三的五项需求减三项已掌握技能等于两项缺口 assert zhao_skill_gaps == {'SQL' } # 核验赵六的三项需求减两项已掌握技能只剩SQL assert len (skill_gap_detail) == 6 # 四名员工去重后的缺口总数应为2加2加1加1
8. 识别技能需求热点
print (' \n === 最热门技能需求 ===' ) # 按关联记录频次汇总样本中的技能需求热度 = merged['required_skill' ].value_counts().reset_index() # 统计各技能出现频次并还原为DataFrame = ['技能' , '涉及项目数' ] # 将频数表的类别与计数列改为可读报告表头 print (skill_demand) # 核对频数按连接后的员工—项目—技能行计数，不能误读为唯一项目数
=== 最热门技能需求 ===
技能 涉及项目数
0 Python 3
1 SQL 3
2 机器学习 3
3 项目管理 2
4 沟通能力 2
5 数据分析 2
6 可视化 2
关键要点 ：
多对多合并会产生键的笛卡尔积
结果行数通常大于任一输入数据集
适合分析关系型数据（员工-项目-技能）
技能需求不等于技能覆盖；覆盖判断必须同时连接员工主键与标准化技能键
技能缺口是“所需技能集合减去已掌握技能集合”，不能用需求频次代替
pivot_table 可以将长格式需求转换为易读的矩阵
习题 7.4: 分层索引操作
问题描述 ：
分层索引可用于在二维表中表达多层键。请：
创建包含股票、日期、指标的分层索引数据
使用 stack 和 unstack 进行数据重塑
使用 swaplevel 和 reorder_levels 调整索引层级
执行分层索引的选择和分组聚合
完整解答 ：
import pandas as pd # 构造证券—交易日 MultiIndex 并演示层级重塑、选择与聚合 import numpy as np # 支持分层行情数值列与聚合结果中的缺失表示 # 从同源行情表取两只证券，构造“证券—交易日”唯一索引 = pd.read_hdf( # 取得宁波港日度 OHLCV f' { DATA_ROOT} /stock/stock_price_pre_adjusted.h5' , # 来源为复权日行情底表 = "order_book_id='601018.XSHG'" # 在读取端限定宁波港 = {'date' : 'trade_date' , 'vol' : 'volume' }) # 港口子表将 date 从索引转回内层候选键，成交量列改为 volume = pd.read_hdf( # 取得宁波银行日度 OHLCV f' { DATA_ROOT} /stock/stock_price_pre_adjusted.h5' , # 保持与宁波港一致的复权口径 = "order_book_id='002142.XSHE'" # 在读取端限定宁波银行 = {'date' : 'trade_date' , 'vol' : 'volume' }) # 银行子表对齐 trade_date/OHLCV 列合同，避免 concat 生成额外列 'symbol' ] = '601018.SH' # 规范宁波港代码为分层索引外层键 'symbol' ] = '002142.SZ' # 规范宁波银行代码为分层索引外层键 = pd.concat([port_stock_data, bank_stock_data]) # 纵向拼接两只股票的行情数据 'datetime' ] = pd.to_datetime(stock_data['trade_date' ], format = '%Y%m %d ' ) # 将交易日期转换为datetime格式 = stock_data[(stock_data['datetime' ] >= '2023-01-01' ) & (stock_data['datetime' ] <= '2023-01-15' )] # MultiIndex 演示限制在半月窗口，使层级切片输出保持紧凑 # 仅保留两个 symbol 外层标签与 OHLCV 值列，内层为 datetime = ['601018.SH' , '002142.SZ' ] # MultiIndex 外层仅接受宁波港与宁波银行两个键 = stock_data[stock_data['symbol' ].isin(symbols)][['datetime' , 'symbol' , 'open' , 'high' , 'low' , 'close' , 'volume' ]].copy() # 筛选目标股票并提取指定列 # 将两只证券代码改为 MultiIndex 外层的中文品种标签 'symbol' ] = filtered['symbol' ].map ({ # 将两个规范代码改为 MultiIndex 外层的中文标签 '601018.SH' : '宁波港' , # 宁波港股票代码映射 '002142.SZ' : '宁波银行' # 宁波银行股票代码映射
1. 创建分层索引
= filtered.set_index(['symbol' , 'datetime' ]).sort_index() # 设置股票+日期为分层索引并排序 print ('=== 分层索引数据结构 ===' ) # 展示股票—日期层级及行情字段布局 print (f'索引层级: { hierarchical_data. index. names} ' ) # 核对外层为 symbol、内层为 datetime，避免后续切片指定错层 print (f'数据形状: { hierarchical_data. shape} ' ) # 记录两只证券半月交易行数与 OHLCV 五列 schema print (f' \n 数据示例:' ) # 抽查排序后的分层索引记录 print (hierarchical_data.head(10 )) # 抽查排序结果先按证券、再按交易日递增，并确认值列未进入索引
=== 分层索引数据结构 ===
索引层级: ['symbol', 'datetime']
数据形状: (18, 5)
数据示例:
open high low close volume
symbol datetime
宁波港 2023-01-03 3.2761 3.2944 3.2577 3.2761 10743034.0
2023-01-04 3.2669 3.2944 3.2669 3.2944 8787221.0
2023-01-05 3.2944 3.3219 3.2761 3.2852 10407133.0
2023-01-06 3.2852 3.2944 3.2577 3.2669 10450610.0
2023-01-09 3.2669 3.2852 3.2577 3.2669 6518398.0
2023-01-10 3.2669 3.2761 3.2485 3.2485 5882400.0
2023-01-11 3.2394 3.2669 3.2394 3.2394 4453499.0
2023-01-12 3.2394 3.2577 3.2302 3.2302 4491700.0
2023-01-13 3.2302 3.2669 3.2302 3.2577 4929600.0
宁波银行 2023-01-03 29.3540 29.5547 28.7338 29.3814 27552512.0
2. 使用 unstack 将数据从长格式转换为宽格式
print (' \n === Unstack: 将股票索引转为列 ===' ) # 展示同日两只股票横向对齐的宽表结果 = hierarchical_data['close' ].unstack(level= 'symbol' ) # 将symbol层级从long转wide，每只股票成为一列 print (f'Unstacked 形状: { unstacked. shape} ' ) # 核对 symbol 层转为两列，datetime 层保留为唯一行索引 print (unstacked.head()) # 抽查同一交易日两只证券收盘价是否在同一行对齐 # 使用 stack 将数据从宽格式转换为长格式 print (' \n === Stack: 将列转回索引 ===' ) # 标记宽表重新堆叠为长索引的输出段 = unstacked.stack() # 将股票列重新堆叠为行索引 print (f'Restacked 形状: { restacked. shape} ' ) # 对照宽表非缺失单元数，确认 stack 恢复的 datetime—symbol 观测数 print (restacked.head()) # 核对堆叠后外层为 datetime、内层为 symbol，层序不同于原表
=== Unstack: 将股票索引转为列 ===
Unstacked 形状: (9, 2)
symbol 宁波港 宁波银行
datetime
2023-01-03 3.2761 29.3814
2023-01-04 3.2944 30.5399
2023-01-05 3.2852 30.1933
2023-01-06 3.2669 29.7645
2023-01-09 3.2669 29.9470
=== Stack: 将列转回索引 ===
Restacked 形状: (18,)
datetime symbol
2023-01-03 宁波港 3.2761
宁波银行 29.3814
2023-01-04 宁波港 3.2944
宁波银行 30.5399
2023-01-05 宁波港 3.2852
dtype: float64
3. 使用 swaplevel 交换索引层级
print (' \n === Swaplevel: 交换索引层级 ===' ) # 展示日期优先与股票优先两种检索顺序 = hierarchical_data.swaplevel('symbol' , 'datetime' ) # 交换股票与日期的索引顺序 print (f'原始索引顺序: { hierarchical_data. index. names} ' ) # 确认原表先按 symbol、再按 datetime 定位观测 print (f'交换后索引顺序: { swapped. index. names} ' ) # 确认新外层为 datetime、新内层为 symbol print (swapped.head()) # 抽查交换后按日期外层展示多个证券，值列与观测内容保持不变 # 使用 reorder_levels 重新排序 = hierarchical_data.reorder_levels(['datetime' , 'symbol' ]) # 将索引顺序调整为日期→股票 print (f' \n Reorder 后索引顺序: { reordered. index. names} ' ) # 确认重排后的层级顺序
=== Swaplevel: 交换索引层级 ===
原始索引顺序: ['symbol', 'datetime']
交换后索引顺序: ['datetime', 'symbol']
open high low close volume
datetime symbol
2023-01-03 宁波港 3.2761 3.2944 3.2577 3.2761 10743034.0
2023-01-04 宁波港 3.2669 3.2944 3.2669 3.2944 8787221.0
2023-01-05 宁波港 3.2944 3.3219 3.2761 3.2852 10407133.0
2023-01-06 宁波港 3.2852 3.2944 3.2577 3.2669 10450610.0
2023-01-09 宁波港 3.2669 3.2852 3.2577 3.2669 6518398.0
Reorder 后索引顺序: ['datetime', 'symbol']
4. 分层索引的选择
print (' \n === 选择特定股票的所有数据 ===' ) # 验证按外层股票标签可取得完整时间序列 = hierarchical_data.loc['宁波港' ] # 用loc选取宁波港的全部交易数据 print (nbz_data.head()) # 核对选择外层证券后索引降为 datetime 单层，OHLCV schema 保持不变 print (' \n === 选择特定日期的所有股票数据 ===' ) # 验证按内层日期可取得当日横截面 = hierarchical_data.loc[(slice (None ), '2023-01-03' ), :] # 用切片选取某天所有股票数据 print (specific_date) # 核对内层日期切片保留两只证券，结果仍显示 symbol—datetime 复合键 print (' \n === 使用 xs 进行跨层级选择 ===' ) # 对照 xs 与 loc 的跨层级切片结果 # xs 允许你从多层索引中选择特定值 = hierarchical_data.xs('宁波港' , level= 'symbol' ) # 用xs按symbol层级截取宁波港数据 print (nbz_via_xs.head()) # 与 loc 结果对照，确认 xs 指定 symbol 层后同样降为日期索引
=== 选择特定股票的所有数据 ===
open high low close volume
datetime
2023-01-03 3.2761 3.2944 3.2577 3.2761 10743034.0
2023-01-04 3.2669 3.2944 3.2669 3.2944 8787221.0
2023-01-05 3.2944 3.3219 3.2761 3.2852 10407133.0
2023-01-06 3.2852 3.2944 3.2577 3.2669 10450610.0
2023-01-09 3.2669 3.2852 3.2577 3.2669 6518398.0
=== 选择特定日期的所有股票数据 ===
open high low close volume
symbol datetime
宁波港 2023-01-03 3.2761 3.2944 3.2577 3.2761 10743034.0
宁波银行 2023-01-03 29.3540 29.5547 28.7338 29.3814 27552512.0
=== 使用 xs 进行跨层级选择 ===
open high low close volume
datetime
2023-01-03 3.2761 3.2944 3.2577 3.2761 10743034.0
2023-01-04 3.2669 3.2944 3.2669 3.2944 8787221.0
2023-01-05 3.2944 3.3219 3.2761 3.2852 10407133.0
2023-01-06 3.2852 3.2944 3.2577 3.2669 10450610.0
2023-01-09 3.2669 3.2852 3.2577 3.2669 6518398.0
5. 分层分组聚合
print (' \n === 按股票分组统计 ===' ) # 汇总每只股票在样本期的价格与成交量特征 = hierarchical_data.groupby(level= 'symbol' ).agg({ # 沿 MultiIndex 的证券层汇总，日期层观测作为组内样本 'close' : ['mean' , 'std' , 'min' , 'max' ], # 收盘价的均值、标准差、最小、最大 'volume' : 'sum' # 累加每只证券样本期内成交量 round (2 ) # 证券层价格与成交量统计保留两位 print (by_symbol) # 核对 symbol 层聚合后每只证券一行，列轴由原字段与统计量构成两层 print (' \n === 按日期分组统计 ===' ) # 汇总同一交易日跨股票的横截面指标 = hierarchical_data.groupby(level= 'datetime' ).agg({ # 按日期层级分组聚合 'close' : ['mean' , 'std' ], # 收盘价的均值和标准差 'volume' : 'sum' # 累加同日两只证券的成交量 round (2 ) # 交易日横截面统计统一两位展示精度 print (by_date.head()) # 核对 datetime 层聚合后每个交易日一行，统计量来自当日两只证券横截面
=== 按股票分组统计 ===
close volume
mean std min max sum
symbol
宁波港 3.26 0.02 3.23 3.29 66663595.0
宁波银行 30.21 0.56 29.38 30.99 257009431.0
=== 按日期分组统计 ===
close volume
mean std sum
datetime
2023-01-03 16.33 18.46 38295546.0
2023-01-04 16.92 19.27 46717935.0
2023-01-05 16.74 19.03 35977710.0
2023-01-06 16.52 18.74 50212483.0
2023-01-09 16.61 18.87 32938017.0
6. 多级聚合
print (' \n === 高级多级聚合 ===' ) # 展示股票×自然周的 OHLCV 分层聚合 = hierarchical_data.groupby(['symbol' , pd.Grouper(level= 'datetime' , freq= 'W' )]).agg({ # 按股票和自然周分组聚合 'open' : 'first' , # 每周的开盘价（取周内第一个） 'high' : 'max' , # 周窗口内取 high 最大值，与首日 open/末日 close 共同形成周K线 'low' : 'min' , # 每周最低价 'close' : 'last' , # 每周的收盘价（取周内最后一个） 'volume' : 'sum' # 累加周内各交易日成交量，保持流量指标的可加性 round (2 ) # 周度 OHLCV 报告表保留两位精度 print (multi_agg) # 核对输出行索引为 symbol—周末日期，列 schema 为周度 OHLCV
=== 高级多级聚合 ===
open high low close volume
symbol datetime
宁波港 2023-01-08 3.28 3.32 3.26 3.27 40387998.0
2023-01-15 3.27 3.29 3.23 3.26 26275597.0
宁波银行 2023-01-08 29.35 30.93 28.73 29.76 130815676.0
2023-01-15 29.89 31.30 29.42 30.99 126193755.0
关键要点 ： - 分层索引允许在单个 DataFrame 中表示多维数据 - stack 将列转换为索引级别（宽→长） - unstack 将索引级别转换为列（长→宽） - swaplevel 交换两个索引级别的位置 - xs 提供了跨层级选择的便捷方法 - groupby 可以基于特定索引层级进行聚合
习题 7.5: 数据重塑与透视表
问题描述 ：
使用宁波港、宁波银行和恒瑞医药的交易数据，请：
创建透视表，显示不同股票在不同日期的收盘价
使用 pivot 创建交叉表
使用 melt 将宽格式数据转换为长格式
计算各股票的涨跌幅并创建热力图数据
完整解答 ：
import pandas as pd # 在三只证券的日长表上演示 pivot、melt 与 crosstab import numpy as np # 核验三类收益方向比例逐证券合计为1 # 三只证券共用价格调整口径，拼接后以日期—证券作重塑键 = pd.read_hdf( # 港口股提供交通运输列，收盘价与成交量同时进入重塑 f' { DATA_ROOT} /stock/stock_price_pre_adjusted.h5' , # 三个输入均来自复权行情 HDF5 表 = "order_book_id='601018.XSHG'" # 601018.XSHG 决定宽表中“宁波港”列的来源行 = {'date' : 'trade_date' , 'vol' : 'volume' }) # 恢复宁波港重塑日期键并对齐 volume 字段 = pd.read_hdf( # 银行股补充金融服务列，与港口股按共同日期对齐 f' { DATA_ROOT} /stock/stock_price_pre_adjusted.h5' , # 沿用与宁波港一致的数据合同 = "order_book_id='002142.XSHE'" # 002142.XSHE 将被映射为宽表的“宁波银行”列 = {'date' : 'trade_date' , 'vol' : 'volume' }) # 恢复宁波银行重塑日期键并对齐 volume 字段 = pd.read_hdf( # 医药股作第三个品种，检查三列 pivot 和三组 crosstab f' { DATA_ROOT} /stock/stock_price_pre_adjusted.h5' , # 第三输入仍使用同源复权行情 = "order_book_id='600276.XSHG'" # 600276.XSHG 对应“恒瑞医药”列，不与 601018.XSHG 共用标签 = {'date' : 'trade_date' , 'vol' : 'volume' }) # 恢复恒瑞医药日期键并对齐 volume 字段 'symbol' ] = '601018.SH' # 将宁波港代码规范为重塑外层标签 'symbol' ] = '002142.SZ' # 将宁波银行代码规范为重塑外层标签 'symbol' ] = '600276.SH' # 将恒瑞医药代码统一为透视表证券键 = pd.concat([stock_data_nbz, stock_data_nby, stock_data_m]) # 纵向拼接三只证券，保留同一行情 schema 'datetime' ] = pd.to_datetime(stock_data['trade_date' ], format = '%Y%m %d ' ) # 转换为日期时间格式 = stock_data[(stock_data['datetime' ] >= '2023-01-01' ) & (stock_data['datetime' ] <= '2023-01-31' )] # 将透视表样本固定在2023年1月交易日 # 按 symbols_map 的三个键投影长表，它将被重塑为日期×证券矩阵 = { # 为透视列准备代码到中文证券名的映射 '601018.SH' : '宁波港' , # 交通运输品种用于透视列与分组行 '002142.SZ' : '宁波银行' , # 金融品种检查深市代码规范化结果 '600276.SH' : '恒瑞医药' # 医药品种提供跨行业的第三类
= stock_data[stock_data['symbol' ].isin(symbols_map.keys())][['datetime' , 'symbol' , 'close' , 'volume' ]].copy() # 筛选目标股票并提取日期、代码、收盘价、成交量 'symbol' ] = filtered['symbol' ].map (symbols_map) # 将三个代码转为透视表的中文证券标签 # 计算涨跌幅 = filtered.sort_values(['symbol' , 'datetime' ]) # 按股票和日期排序，确保时间序列连续 'return' ] = filtered.groupby('symbol' )['close' ].pct_change(fill_method= None ) # 分证券计算且不跨价格缺口填补 print ('=== 原始数据示例 ===' ) # 抽查排序后的日期—证券长表及首日收益缺口 print (filtered.head(10 )) # 抽查日期—证券长表排序及每只证券首个收益率为空的分组边界
=== 原始数据示例 ===
datetime symbol close volume return
2981 2023-01-03 宁波港 3.2761 10743034.0 NaN
2982 2023-01-04 宁波港 3.2944 8787221.0 0.005586
2983 2023-01-05 宁波港 3.2852 10407133.0 -0.002793
2984 2023-01-06 宁波港 3.2669 10450610.0 -0.005570
2985 2023-01-09 宁波港 3.2669 6518398.0 0.000000
2986 2023-01-10 宁波港 3.2485 5882400.0 -0.005632
2987 2023-01-11 宁波港 3.2394 4453499.0 -0.002801
2988 2023-01-12 宁波港 3.2302 4491700.0 -0.002840
2989 2023-01-13 宁波港 3.2577 4929600.0 0.008513
2990 2023-01-16 宁波港 3.2761 8098816.0 0.005648
1. 创建透视表
print (' \n === 1. 透视表：收盘价 ===' ) # 检查三只证券是否每日各占一列且共用交易日行 = pd.pivot_table( # 创建收盘价透视表（行=日期，列=股票） # 输入含 datetime、symbol、close 和 volume 的长表 = 'close' , # 聚合的值列为收盘价 = 'datetime' , # 收盘价宽表每个交易日占一行 = 'symbol' , # 三个中文证券名展开为独立收盘价列 = 'mean' # 唯一日期—证券键下均值等于原收盘价，同时可暴露意外重复键 print (pivot_close.head()) # 核对 datetime 行索引与三只证券列，确认日期—证券唯一键未被均值聚合改变
=== 1. 透视表：收盘价 ===
symbol 宁波港 宁波银行 恒瑞医药
datetime
2023-01-03 3.2761 29.3814 37.9796
2023-01-04 3.2944 30.5399 38.3056
2023-01-05 3.2852 30.1933 39.0763
2023-01-06 3.2669 29.7645 38.8392
2023-01-09 3.2669 29.9470 39.0763
2. 使用 pivot 创建交叉表
print (' \n === 2. Pivot: 涨跌幅交叉表 ===' ) # 验证唯一日期—证券键可无聚合地展开收益宽表 = filtered.pivot( # 将 datetime、symbol、return 三列长表展开为日期行、证券列的收益宽表 = 'datetime' , # 收益宽表沿用相同交易日行键 = 'symbol' , # 每只证券形成一列日简单收益 = 'return' # 值为日收益率 print (pivot_returns.head()) # 核对无聚合 pivot 得到相同日期行、三证券收益列及各列首期缺失
=== 2. Pivot: 涨跌幅交叉表 ===
symbol 宁波港 宁波银行 恒瑞医药
datetime
2023-01-03 NaN NaN NaN
2023-01-04 0.005586 0.039430 0.008584
2023-01-05 -0.002793 -0.011349 0.020120
2023-01-06 -0.005570 -0.014202 -0.006068
2023-01-09 0.000000 0.006131 0.006105
3. 使用 melt 将宽格式转换为长格式
print (' \n === 3. Melt: 宽格式→长格式 ===' ) # 检查重塑后 schema 由一日三价列变为日期、股票、收盘价三列 # pivot_close 是宽格式，每列是一只股票 = pivot_close.reset_index().melt( # 恢复日期列后转成长表，使证券成为可分组的观测维度 = 'datetime' , # 日期仍是每条长表观测的识别键，不融入收盘价值列 = '股票' , # 将原列名存入“股票”列 = '收盘价' # 三个证券价格列的单元值统一下推到长表“收盘价”字段 print (melted.head(10 )) # 核对宽表列名下推为“股票”值，输出固定为 datetime、股票、收盘价三列
=== 3. Melt: 宽格式→长格式 ===
datetime 股票 收盘价
0 2023-01-03 宁波港 3.2761
1 2023-01-04 宁波港 3.2944
2 2023-01-05 宁波港 3.2852
3 2023-01-06 宁波港 3.2669
4 2023-01-09 宁波港 3.2669
5 2023-01-10 宁波港 3.2485
6 2023-01-11 宁波港 3.2394
7 2023-01-12 宁波港 3.2302
8 2023-01-13 宁波港 3.2577
9 2023-01-16 宁波港 3.2761
4. 创建热力图数据
print (' \n === 4. 热力图数据：按周统计涨跌幅 ===' ) # 输出“ISO周—证券”二维均值矩阵作热力图输入 'week' ] = pd.to_datetime(filtered['datetime' ]).dt.isocalendar().week # 提取每个交易日所属的ISO周数 = pd.pivot_table( # 创建周平均涨跌幅热力图数据 # 输入已附加 ISO 周号和日收益的长表 = 'return' , # 聚合的值列为日收益率 = 'week' , # 行索引为ISO周数 = 'symbol' , # 将三只证券展开为热力图列 = 'mean' # 同一 ISO 周若有多个日收益率，压缩为证券周均收益 round (4 ) # 周度收益矩阵以小数口径展示到万分位，便于比较微小差异 print ('周平均涨跌幅:' ) # 标明随后矩阵的单元是周内日收益均值 print (heatmap_data) # 核对 ISO 周为行、证券为列的二维 schema，并观察跨周样本数变化后的均值 print (' \n === 5. 多值透视表 ===' ) # 检查按证券聚合后的两层列索引：指标在外、统计量在内 = pd.pivot_table( # 创建多值聚合透视表（同时对多列应用不同聚合函数） # 输入含价格、成交量和收益的证券长表 = ['close' , 'volume' , 'return' ], # 同时输出价格水平、交易活跃度与收益风险三类统计 = 'symbol' , # 每只证券汇总为一行多统计量 = { # 为不同列指定不同的聚合函数 'close' : ['mean' , 'std' , 'min' , 'max' ], # 收盘价：均值、标准差、最小值、最大值 'volume' : ['mean' , 'sum' ], # 成交量：日均、总计 'return' : ['mean' , 'std' ] # 收益率：均值、标准差 print (multi_value_pivot.round (2 )) # 输出价格、成交量与收益的两层列聚合表，统一展示精度
=== 4. 热力图数据：按周统计涨跌幅 ===
周平均涨跌幅:
symbol 宁波港 宁波银行 恒瑞医药
week
1 -0.0009 0.0046 0.0075
2 -0.0006 0.0082 0.0000
3 0.0050 -0.0045 0.0219
5 0.0041 -0.0060 -0.0154
=== 5. 多值透视表 ===
close return volume
max mean min std mean std mean sum
symbol
宁波港 3.37 3.28 3.23 0.04 0.00 0.01 8298598.69 132777579.0
宁波银行 31.20 30.37 29.38 0.52 0.00 0.02 27967374.88 447477998.0
恒瑞医药 43.12 40.14 37.34 2.06 0.01 0.03 55370262.44 885924199.0
6. 使用 crosstab 创建频数表
print (' \n === 6. CrossTab: 收益方向天数统计 ===' ) # 输出证券×下跌/平盘/上涨的频数和边际合计 = pd.Series(pd.NA, index= filtered.index, dtype= 'string' ) # 首个不可计算收益保持缺失方向 'return' ] < 0 ] = '下跌' # 负收益归入下跌 'return' ] == 0 ] = '平盘' # 零收益单列，避免高估下跌天数 'return' ] > 0 ] = '上涨' # 正收益归入上涨 'direction' ] = pd.Categorical( # 固定三类显示顺序并保留缺失值 = ['下跌' , '平盘' , '上涨' ], ordered= True = filtered.loc[ # 显式排除每只证券首个不可计算收益 'direction' ].notna(), ['symbol' , 'direction' ]= pd.crosstab( # 创建证券×三类收益方向的交叉频数表 = valid_direction_rows['symbol' ], # 每只证券占一行有效方向频数 = valid_direction_rows['direction' ], # 下跌、平盘与上涨分别占一列 = True , # 添加行/列合计 = '总计' , # 使用稳定标签定位边际总计 = False # 即使某类当前为零也保留完整三分类schema = len (valid_direction_rows) # 统计可计算收益方向的有效记录 assert crosstab_result.index[- 1 ] == '总计' and crosstab_result.columns[- 1 ] == '总计' # 核验边际位置 assert int (crosstab_result.iloc[- 1 , - 1 ]) == valid_direction_count # 核验三类频数合计等于非缺失收益记录数 print (crosstab_result) # 核对证券行、三类方向列与边际合计，首日缺失收益不进入任一方向 # 计算三类收益方向比例 = pd.crosstab( # 创建按行归一化的三类收益方向交叉表 = valid_direction_rows['symbol' ], # 按证券逐行归一化有效收益方向天数 = valid_direction_rows['direction' ], # 保留下跌、平盘与上涨三类比例列 = 'index' , # 按行归一化（每行合计为1） = False # 与频数表保持相同三分类schema sum (axis= 1 ).to_numpy(), 1.0 ) # 核验每只证券三类比例合计为1 print (' \n 收益方向比例:' ) # 说明随后交叉表已按证券行归一化为比例 print (crosstab_pct.round (3 )) # 以千分位显示，并与频数表使用同一有效样本
=== 6. CrossTab: 收益方向天数统计 ===
direction 下跌 平盘 上涨 总计
symbol
宁波港 5 3 7 15
宁波银行 9 0 6 15
恒瑞医药 9 0 6 15
总计 23 3 19 45
收益方向比例:
direction 下跌 平盘 上涨
symbol
宁波港 0.333 0.2 0.467
宁波银行 0.600 0.0 0.400
恒瑞医药 0.600 0.0 0.400
关键要点 ： - pivot_table 灵活，可以指定聚合函数 - pivot 简单，要求索引-列值唯一 - melt 将宽格式转为长格式，便于某些分析 - crosstab 专门用于计算频数和交叉表 - 透视表是数据分析和报表的核心工具 - 热力图数据通常是二维矩阵形式
习题 7.6: 数据合并策略与性能优化
问题描述 ：
在处理大规模数据时，合并操作的效率很重要。请：
比较不同合并方法的性能
使用索引键合并 vs 列键合并
处理重复键名
优化大数据集的合并操作
完整解答 ：
import pandas as pd # 整合日行情、季度估值、行业维度与题设评级事件 import numpy as np # 承载多源对齐后的数值缺失与统计量 import time # 分别计时列键 merge 与索引 join；结果只适用于当次样本与环境 # 性能样本使用三只证券的真实日行情，不用随机生成的键频数 = { # 定义HDF5代码与简短代码的映射 '601018.XSHG' : '601018.SH' , # 沪市港口股：来源后缀 XSHG 转为展示后缀 SH '002142.XSHE' : '002142.SZ' , # 深市银行股：映射为公司维度使用的 SZ 键 '600276.XSHG' : '600276.SH' # 沪市医药股：保留与港口股不同的六位代码 = [] # 收集三只证券的同 schema 行情子表，循环后纵向形成事实表 for hdf_code, short_code in stock_code_map.items(): # 每个来源证券键生成独立日行情事实子表，并同步写入展示代码连接键 = pd.read_hdf( # 循环中每次下推一个来源代码，保留真实的键重复分布 f' { DATA_ROOT} /stock/stock_price_pre_adjusted.h5' ,= f"order_book_id=' { hdf_code} '" # 循环的 hdf_code 依次为港口、银行和医药来源键，每次不跨证券读取 # 把单股交易日恢复为事实表时间键 'stock_code' ] = short_code # 添加简短格式的股票代码 # 收集当前证券子表，待循环后纵向拼接 = pd.concat(stock_frames, ignore_index= True ) # 合并三只股票的真实交易数据 'date' ] = pd.to_datetime(all_stock_trades['date' ]) # 确保date列为datetime格式
# 构建交易数据集A：使用真实收盘价和成交量，最多抽取10000条 = all_stock_trades[['date' , 'stock_code' , 'close' , 'volume' ]].rename( # 重命名价格列后固定种子抽取事实表样本 = {'close' : 'price' } # 将收盘价列重命名为price = min (10000 , len (all_stock_trades)), random_state= 42 ).reset_index(drop= True )0 , 'transaction_id' , range (len (data_a))) # 在首列插入交易ID编号
# 数据集B：公司基本信息（真实数据） = pd.read_hdf(f' { DATA_ROOT} /stock/stock_basic_data.h5' ) # 读取一次公司基本信息用于两项维表练习 # 从本地公司表构造每个股票代码唯一对应的维度记录 = ( # 从公司基本信息筛选并整理唯一证券维度记录 'order_book_id' ].isin(stock_code_map),'order_book_id' , 'symbol' , 'industry_name' , 'province' ],= lambda x: x['order_book_id' ].map (stock_code_map))= {'symbol' : 'company_name' , 'industry_name' : 'sector' , 'province' : 'region' 'stock_code' , 'company_name' , 'sector' , 'region' ]]print ('=== 数据集信息 ===' ) # 报告抽样交易事实行数与每代码唯一的公司维度行数 print (f'数据集A形状: { data_a. shape} ' ) # 记录抽样交易事实表的行数及 transaction/date/code/price/volume schema print (f'数据集B形状: { company_info. shape} ' ) # 维度表应只有3个唯一 stock_code，值列为 company_name、sector、region
=== 数据集信息 ===
数据集A形状: (10000, 5)
数据集B形状: (3, 4)
1. 性能比较：列键合并 vs 索引键合并
print (' \n === 1. 性能比较 ===' ) # 在相同 one-to-many 业务键上比较列合并与索引连接耗时 # 方法1: 使用列键合并 = time.time() # 记录列键合并开始时间 = pd.merge(data_a, company_info, on= 'stock_code' , how= 'left' ) # 基于stock_code列键进行左合并 = time.time() - start # 计算列键合并耗时 print (f'列键合并耗时: { time_col:.4f} 秒' ) # 报告当次抽样事实表按 stock_code 列左连接唯一维度表的墙钟时间 # 方法2: 使用索引键合并 = data_a.set_index('stock_code' ) # 将stock_code设置为数据集A的索引 = company_info.set_index('stock_code' ) # 将stock_code设置为公司信息表的索引 = time.time() # 记录索引键合并开始时间 = data_a_indexed.join(company_info_indexed, how= 'left' ) # 基于索引进行左连接 = time.time() - start # 计算索引键合并耗时 print (f'索引键合并耗时: { time_idx:.4f} 秒' ) # 报告同一事实表和维表在索引连接下的当次墙钟时间 print (f' \n 性能提升: { (time_col / max (time_idx, 1e-6 )):.2f} x' ) # 防止除零错误
=== 1. 性能比较 ===
列键合并耗时: 0.0025 秒
索引键合并耗时: 0.0026 秒
性能提升: 0.97x
2. 处理重复键名
print (' \n === 2. 处理重复列名 ===' ) # 验证公司省份与办公地址两个 region 字段被后缀区分 # 提取办公地址并故意命名为 region，用于演示重名列后缀 = ( # 从同一公司表提取办公地址并构造重名列示例 'order_book_id' ].isin(stock_code_map),'order_book_id' , 'office_address' ],= lambda x: x['order_book_id' ].map (stock_code_map))= {'office_address' : 'region' })'stock_code' , 'region' ]]# 合并时两个数据集都有 'region' 列 = pd.merge(data_a, company_info, on= 'stock_code' , how= 'left' ) # 以行情代码左连公司名与公司地址，保持事实表行数 = pd.merge(merged_suffix, data_c, on= 'stock_code' , how= 'left' , # 第二次左合并：添加地区信息 = ('_company' , '_location' )) # 重复列名加后缀区分 print (f'合并后列名: { merged_suffix. columns. tolist()} ' ) # 核对重名 region 已按来源拆成 company/location 两列且其余 schema 未被改名 print (merged_suffix[['stock_code' , 'region_company' , 'region_location' ]].drop_duplicates()) # 每个代码保留一行，对照注册省份与办公地址的来源差异
=== 2. 处理重复列名 ===
合并后列名: ['transaction_id', 'date', 'stock_code', 'price', 'volume', 'company_name', 'sector', 'region_company', 'region_location']
stock_code region_company region_location
0 600276.SH 江苏省 江苏连云港市经济技术开发区昆仑山路7号
4 601018.SH 浙江省 宁波市鄞州区宁东路269号环球航运广场
7 002142.SZ 浙江省 浙江省宁波市鄞州区宁东路345号
3. 使用验证参数检查合并质量
print (' \n === 3. 合并验证 ===' ) # 用 validate 明示审计维度表键唯一与事实表键可重复 # 先将公司维度表与自身连接，要求 stock_code 在两侧均唯一 try : # 若任一侧 stock_code 重复，捕获 MergeError 并暴露维表键违约 = pd.merge( # 公司维表自连接要求两侧 stock_code 均唯一，结果行数不应扩张 # 左侧数据框：公司信息 # 右侧数据框：同一个公司信息表 = 'stock_code' , # 以两侧均唯一的证券代码做一对一键 = 'one_to_one' # 验证两侧键均唯一 print ('One-to-one 验证通过' ) # 明示左右 stock_code 均唯一，连接不会产生笛卡尔扩张 except pd.errors.MergeError as e: # 捕获合并验证失败的异常 print (f'One-to-one 验证失败: { e} ' ) # 显示验证失败的具体原因 # one_to_many: 左侧键唯一，右侧可重复 = pd.merge( # 以唯一公司键连接右侧多条交易事实，验证一对多基数合同 # 左侧：公司信息表（每只股票一行） # 右侧：交易数据（每只股票多行） = 'stock_code' , # 用维度表唯一代码匹配事实表重复代码 = 'one_to_many' # 验证左侧唯一、右侧可重复 print (f'One-to-many 合并成功: { merged_otm. shape} ' ) # 行数应等于交易事实行数，列数增加公司属性字段
=== 3. 合并验证 ===
One-to-one 验证通过
One-to-many 合并成功: (10000, 8)
4. 优化大数据集合并
print (' \n === 4. 优化建议 ===' ) # 将性能经验与键基数、内存和血缘诊断一并输出 # 将连接优化建议集中为多行文本，避免把性能经验误写成无条件规则 = ( # 组织多行优化提示并保留适用边界说明 """ 大数据集合并优化技巧： 1. 使用索引键合并通常比列键合并更快 2. 如果可能，先对键进行排序 3. 使用 'indicator' 参数检查合并结果 4. 对于大型数据集，考虑使用 Dask 或 modin 5. 避免在合并前进行不必要的操作 6. 使用适当的数据类型减少内存占用 """ # 多行字符串：大数据集合并优化技巧摘要 # 完成优化提示文本赋值 print (tips) # 输出键排序、indicator 与内存优化清单，作为性能试验结果的适用边界 # 演示 indicator 参数 = pd.merge( # 带indicator参数的左合并 100 ), # 取交易数据前100行作为示例 # 公司信息查找表 = 'stock_code' , # 用前100笔交易的证券代码查找公司属性 = 'left' , # 左合并方式 = True # 添加_merge列标记每行的合并来源 print (' \n 使用 indicator 检查合并来源:' ) # 声明随后计数用于确认交易代码是否全部匹配维度表 print (merged_indicator['_merge' ].value_counts()) # 统计各合并来源的记录数
=== 4. 优化建议 ===
大数据集合并优化技巧：
1. 使用索引键合并通常比列键合并更快
2. 如果可能，先对键进行排序
3. 使用 'indicator' 参数检查合并结果
4. 对于大型数据集，考虑使用 Dask 或 modin
5. 避免在合并前进行不必要的操作
6. 使用适当的数据类型减少内存占用
使用 indicator 检查合并来源:
_merge
both 100
left_only 0
right_only 0
Name: count, dtype: int64
5. 批量合并多个数据集
print (' \n === 5. 批量合并策略 ===' ) # 对照顺序左连接与 reduce 折叠后的列集合及行顺序 # 汇总多个已构造的数据源 = [company_info, data_c] # 按顺序登记公司属性表与办公地址表，二者均以 stock_code 唯一 # 方法1: 逐步合并 = data_a.copy() # 以交易事实表作为连续左连接的固定行集 for df in data_sources: # 依次把公司属性和地区维表按 stock_code 左连到交易事实，始终保留左侧行数 = pd.merge(result_step, df, on= 'stock_code' , how= 'left' ) # 按代码依次追加公司属性与办公地址列 print (f'逐步合并结果: { result_step. shape} ' ) # 核对连续左连接最终保留的行列规模 # 方法2: 使用 reduce（需要 functools） from functools import reduce # 用二元归并函数折叠相同的维表连接序列 def merge_on_stock(left, right): # 固化按 stock_code 左连且保留事实行的二元归并规则，供 reduce 重用 return pd.merge(left, right, on= 'stock_code' , how= 'left' ) # 折叠每次保留左侧交易行，并扩展右表属性列 = reduce (merge_on_stock, [data_a] + data_sources) # 折叠全部维表后应与逐步连接具有相同行数、字段数和键基数 print (f'Reduce 合并结果: { result_reduce. shape} ' ) # 用相同行列规模检验reduce实现的连接口径 # 验证两种方法结果相同 print (f' \n 两种方法结果一致: { result_step. equals(result_reduce)} ' ) # 同时核对值、行序、列序和缺失位置，确认两种归并策略等价
=== 5. 批量合并策略 ===
逐步合并结果: (10000, 9)
Reduce 合并结果: (10000, 9)
两种方法结果一致: True
关键要点 ： - 索引键合并通常比列键合并更快 - 使用 validate 参数可以检查合并质量 - indicator 参数帮助识别合并来源 - suffixes 参数处理重复列名 - 对于多个数据集，使用 reduce 更优雅 - 大数据集合并需要考虑内存和性能优化
习题 7.7: 综合数据整合项目
问题描述 ：
假设你正在构建一个股票分析系统，需要从多个数据源整合信息：
数据源A：本地 HDF5 文件的日度行情数据
数据源B：财务指标数据（真实估值因子）
数据源C：行业分类数据
数据源D：分析师评级数据
请将这四个数据源整合成一个完整的分析数据集。
完整解答 ：
import pandas as pd # 组装行情、估值、行业与评级四源表，并审计连接后 schema import numpy as np # 支持四源表中数值缺失、聚合与完整率计算 from datetime import datetime # 保留整合项目时间字段使用的标准日期类型 print ('=== 开始数据整合项目 === \n ' ) # 标明四源数据从键规范、连接到时点填充的流水线起点 # 项目初始化结束；后续六步按数据源和连接合同展开
1. 加载数据源A：日度行情数据
# 数据源A：行情事实表定义证券—交易日主样本 print ('[1/6] 加载数据源A：日度行情数据' ) # 首先建立最终数据集的证券—交易日粒度 # 三只证券的复权日行情是事实表，后续维度与事件不得改变其行数 = pd.read_hdf( # 整合主表的港口价量行，定义交通运输样本 f' { DATA_ROOT} /stock/stock_price_pre_adjusted.h5' , # 四源项目的行情底表路径 = "order_book_id='601018.XSHG'" # 事实表仅接受 601018.XSHG，以确保证券—日期键唯一 = {'date' : 'trade_date' , 'vol' : 'volume' }) # 将港口行转为整合管道约定的 trade_date 与 volume = pd.read_hdf( # 同城银行价量行与港口股共享日频 schema f' { DATA_ROOT} /stock/stock_price_pre_adjusted.h5' , # 沿用同一复权价格口径 = "order_book_id='002142.XSHE'" # 只保留 002142.XSHE，为后续行业维度提供金融样本 = {'date' : 'trade_date' , 'vol' : 'volume' }) # 银行子表对齐时间键与成交量合同，便于纵向拼接 = pd.read_hdf( # 医药价量行为三证券整合加入跨行业对照 f' { DATA_ROOT} /stock/stock_price_pre_adjusted.h5' , # 第三只证券仍遵循同源数据合同 = "order_book_id='600276.XSHG'" # 限定 600276.XSHG，防止与同为 XSHG 后缀的港口股混淆 = {'date' : 'trade_date' , 'vol' : 'volume' }) # 医药子表完成同样字段投影后才进入行情事实表 'symbol' ] = '601018.SH' # 港口股改用报表维度的 SH 键，与后续中文名映射对应 'symbol' ] = '002142.SZ' # 银行股从 XSHE 来源后缀转为 SZ 业务后缀 'symbol' ] = '600276.SH' # 医药股保留六位主体码，仅规范交易所后缀 = pd.concat([port_stock_data, bank_stock_data, hengrui_stock_data]) # 纵向拼接三只股票的行情数据 'datetime' ] = pd.to_datetime(stock_data['trade_date' ], format = '%Y%m %d ' ) # 按8位来源日期解析三只证券的共同时间键 = stock_data[(stock_data['datetime' ] >= '2023-01-01' ) & (stock_data['datetime' ] <= '2023-01-31' )] # 四源整合以2023年1月行情事实行为左侧主样本
# 按三个规范代码保留行情主样本，随后映射为 stock_name 复合键的证券维度 = { # 股票代码到中文简称的映射字典 '601018.SH' : '宁波港' , # 宁波港：长三角核心港口企业 '002142.SZ' : '宁波银行' , # 宁波银行：优质城商行代表 '600276.SH' : '恒瑞医药' # 恒瑞医药：医药制造业代表 = price_data[price_data['symbol' ].isin(symbols_map.keys())].copy() # 筛选目标股票并保留副本 'symbol' ] = price_data['symbol' ].map (symbols_map) # 将三个行情代码映射为四源连接的 stock_name 业务键 # 提取关键字段 = price_data[['datetime' , 'symbol' , 'open' , 'high' , 'low' , 'close' , 'volume' ]].copy() # 提取日期、代码、OHLC和成交量 = price_data.rename(columns= {'symbol' : 'stock_name' }) # 将symbol列重命名为stock_name print (f' 形状: { price_data. shape} ' ) # A表每行是一个证券交易日，列为 stock_name 与 OHLCV print (f' 日期范围: { price_data["datetime" ]. min ()} 至 { price_data["datetime" ]. max ()} ' ) # 确认A表覆盖当月首末实际交易日，而非自然日每日一行 print (f' 股票数量: { price_data["stock_name" ]. nunique()} ' ) # 确认数据源A仍覆盖三个唯一证券键 # 输出A表行列数、1月样本窗口和证券覆盖诊断
形状: (48, 7)
日期范围: 2023-01-03 00:00:00 至 2023-01-31 00:00:00
股票数量: 3
2. 创建数据源B：财务指标数据
# 数据源B：低频估值按证券分组与周频时间键向后对齐 print (' \n [2/6] 创建数据源B：财务指标数据（真实估值因子）' ) # 将季度估值按“证券—最近已知季末”映射到周频键 # B表从长期季度估值底表取 PE、PB、股息率与 PS，频率低于行情A表 = pd.read_hdf( # 读取全量估值因子数据 f' { DATA_ROOT} /stock/valuation_factors_quarterly_15_years.h5' # 定义目标股票及其中文名称映射 = { # HDF5代码到中文名的映射 '601018.XSHG' : '宁波港' ,'002142.XSHE' : '宁波银行' ,'600276.XSHG' : '恒瑞医药' = pd.date_range(start= '2023-01-01' , end= '2023-01-31' , freq= '7D' ) # 创建每7天的日期序列 = [] # 收集三只证券各自完成向后时点匹配的周频估值子表
[2/6] 创建数据源B：财务指标数据（真实估值因子）
for code, name in stock_name_map.items(): # 为每只证券分别生成周频估值表，避免跨证券前向匹配 = valuation_all.loc[code][ # 提取该股票的季度估值因子 'pe_ratio_ttm' , 'pb_ratio_ttm' , 'dividend_yield_ttm' , 'ps_ratio_ttm' ]# 把季末索引恢复为估值发生日键，供 asof 对齐 = ['date' , 'pe_ratio' , 'pb_ratio' , 'dividend_yield' , 'ps_ratio' ] # 将原 TTM 字段映射为周频整合表约定的四个估值列名 'date' ] = pd.to_datetime(quarterly['date' ]) # 确保日期列为datetime格式 = pd.DataFrame({'date' : financial_dates}) # 一行一周的目标日历作为 asof 左表，季度估值作为低频右表 = pd.merge_asof( # 使用merge_asof将季度数据对齐到周频日期 'date' ), # 按日期排序的周频日期 'date' ), # 按日期排序的季度数据 = 'date' , direction= 'backward' # 向后查找最近的季度数据 'stock_name' ] = name # 添加股票中文名称列 # 收集当前证券的周频估值子表 = pd.concat(financial_data, ignore_index= True ) # 合并三只股票的财务数据 print (f' 形状: { financial_df. shape} ' ) # B表行数应等于3只证券×5个周频日，值列为四项估值因子 print (f' 列: { financial_df. columns. tolist()} ' ) # 核对B表包含日期、四项估值因子与 stock_name 连接键
形状: (15, 6)
列: ['date', 'pe_ratio', 'pb_ratio', 'dividend_yield', 'ps_ratio', 'stock_name']
# 数据源C：每个 stock_name 唯一对应行业、板块与地区维度 print (' \n [3/6] 创建数据源C：机制演示题设的行业分类数据' ) # 明示该小表不是本地基础表观测 = pd.DataFrame({ # 创建三只股票的行业分类信息表 'stock_name' : ['宁波港' , '宁波银行' , '恒瑞医药' ], # 股票中文简称 'industry' : ['港口航运' , '银行' , '医药制造' ], # 三个值分别对应交通、金融和医药公司，确保 stock_name 键一对一 'sector' : ['交通运输' , '金融服务' , '医药健康' ], # 设置对应一级板块 'region' : ['华东-宁波' , '华东-宁波' , '华东-连云港' ] # 三家公司总部均位于长三角 print (f' 形状: { industry_data. shape} ' ) # C表应为三行唯一 stock_name，每行携带 industry、sector 和 region print (industry_data) # 核对C表每个 stock_name 唯一对应 industry、sector、region 三项维度属性 # 输出C表三个唯一证券键及其题设分类属性
[3/6] 创建数据源C：机制演示题设的行业分类数据
形状: (3, 4)
stock_name industry sector region
0 宁波港 港口航运 交通运输 华东-宁波
1 宁波银行 银行 金融服务 华东-宁波
2 恒瑞医药 医药制造 医药健康 华东-连云港
4. 创建数据源D：分析师评级数据
# 数据源D：题设评级事件以证券—发布日为复合键 print (' \n [4/6] 创建数据源D：分析师评级数据' ) # 创建带明确评级日的题设事件表，便于审计可用时点 # 以下评级和目标价均为机制演示题设值，不代表任何真实研报或投资观点 = pd.DataFrame([ # 构造按证券名—评级日唯一的机制表，用于演示稀疏事件左连接 'stock_name' : '宁波港' , 'rating_date' : '2023-01-03' , 'rating' : '增持' , 'target_price' : 5.80 },'stock_name' : '宁波港' , 'rating_date' : '2023-01-10' , 'rating' : '买入' , 'target_price' : 6.20 },'stock_name' : '宁波港' , 'rating_date' : '2023-01-17' , 'rating' : '增持' , 'target_price' : 5.90 },'stock_name' : '宁波港' , 'rating_date' : '2023-01-24' , 'rating' : '买入' , 'target_price' : 6.10 },'stock_name' : '宁波银行' , 'rating_date' : '2023-01-05' , 'rating' : '买入' , 'target_price' : 38.00 },'stock_name' : '宁波银行' , 'rating_date' : '2023-01-12' , 'rating' : '买入' , 'target_price' : 40.50 },'stock_name' : '宁波银行' , 'rating_date' : '2023-01-19' , 'rating' : '增持' , 'target_price' : 37.00 },'stock_name' : '宁波银行' , 'rating_date' : '2023-01-26' , 'rating' : '买入' , 'target_price' : 39.50 },'stock_name' : '恒瑞医药' , 'rating_date' : '2023-01-04' , 'rating' : '买入' , 'target_price' : 2100.00 },'stock_name' : '恒瑞医药' , 'rating_date' : '2023-01-11' , 'rating' : '买入' , 'target_price' : 2200.00 },'stock_name' : '恒瑞医药' , 'rating_date' : '2023-01-18' , 'rating' : '增持' , 'target_price' : 2050.00 },'stock_name' : '恒瑞医药' , 'rating_date' : '2023-01-25' , 'rating' : '买入' , 'target_price' : 2150.00 },'rating_date' ] = pd.to_datetime(rating_df['rating_date' ]) # 将评级日期转换为datetime格式 print (f' 形状: { rating_df. shape} ' ) # D表按证券—评级日记录12个题设事件及 rating/target_price print (f' 评级分布:' ) # 计数题设增持/买入事件，核对评级表输入合同 print (rating_df['rating' ].value_counts()) # 核对原始题设事件中买入与增持的条数 # 输出D表行列数与原始评级事件频数
[4/6] 创建数据源D：分析师评级数据
形状: (12, 4)
评级分布:
rating
买入 8
增持 4
Name: count, dtype: int64
5. 数据整合
# 四源整合：保留A表全部事实行，先连维度再连低频事件 print (' \n [5/6] 整合所有数据源' ) # 以行情事实行为左表，分步追加维度、估值与评级列 # 步骤1: 整合价格数据和行业信息 = pd.merge(price_data, industry_data, on= 'stock_name' , how= 'left' ) # 左合并行情数据与行业分类 # 步骤2: 整合财务数据（基于日期匹配） # 由于财务数据频率较低，使用前向填充 'date_only' ] = pd.to_datetime(merged['datetime' ]).dt.normalize() # 提取日期部分（去除时间）用于匹配 'date_only' ] = pd.to_datetime(financial_df['date' ]).dt.normalize() # 将财务数据日期归一化为日期格式 = pd.merge( # 左合并估值因子数据 # 左侧：已合并的行情+行业数据 'stock_name' , 'date_only' , 'pe_ratio' , 'pb_ratio' , 'dividend_yield' , 'ps_ratio' ]], # 右侧：估值因子列 = ['stock_name' , 'date_only' ], # 按股票名称和日期双键合并 = 'left' # 左合并，保留所有交易日记录 # 步骤3: 整合评级数据 # 确保合并键类型一致（统一为 datetime64[ns]） 'date_only' ] = pd.to_datetime(merged['date_only' ]) # 确保行情日键与评级日键均为 datetime64[ns] 'rating_date' ] = pd.to_datetime(rating_df['rating_date' ]).dt.normalize() # 将评级日期归一化为日期格式 = pd.merge( # 左合并分析师评级数据 # 左侧：已合并的多源数据 'stock_name' , 'rating_date' , 'rating' , 'target_price' ]], # 右侧：评级相关列 = ['stock_name' , 'date_only' ], # 左侧合并键：股票名和日期 = ['stock_name' , 'rating_date' ], # 右侧合并键：股票名和评级日期 = 'left' # 左合并，保留所有交易日 # 步骤4: 填充缺失值 # 财务指标前向填充 'pe_ratio' , 'pb_ratio' , 'dividend_yield' , 'ps_ratio' ]] = merged.groupby('stock_name' )[ # 按股票分组 'pe_ratio' , 'pb_ratio' , 'dividend_yield' , 'ps_ratio' ] # 选取需要前向填充的估值因子列 # 对估值因子进行前向填充（最近可用值延续到后续交易日） # 评级前向填充，使历史交易日只能使用当时已经发布的题设评级 'rating' , 'target_price' ]] = merged.groupby('stock_name' )[ # 按股票分组以阻止跨证券传播 'rating' , 'target_price' ] # 选取需要按信息时点延续的评级列 # 将最近已知题设评级延续到后续日期，避免未来信息回填
# 清理辅助列 = merged.drop(columns= ['date_only' , 'rating_date' ]) # 删除合并过程中产生的辅助日期列 print (f' 整合后形状: { merged. shape} ' ) # 核对左连接没有改变行情事实行数，仅追加维度、估值与评级列 print (f' 缺失值: { merged. isna(). sum (). sum ()} ' ) # 汇总估值频率差与评级事件稀疏性在填充后仍留下的字段缺口 # 输出删除辅助日期键后的最终 schema 形状与缺失总数
6. 创建整合报告
# 整合报告：从字段完整性、样本窗口与分组统计三方面审计 print (' \n [6/6] 数据整合报告' ) # 审计最终表的完整性、时间范围、证券覆盖和分组分布 print ('=' * 60 ) # 用视觉分隔符将整合过程日志与最终 schema 审计区分开 print (' \n 数据完整性:' ) # 逐列报告连接与前向填充后的非缺失比例 = (1 - merged.isna().mean()) * 100 # 计算每列的数据完整性百分比 for col in merged.columns: # 以整合表总行数为分母报告每个行情、估值、行业与评级字段完整率 print (f' { col} : { completeness[col]:.2f} %' ) # 逐列定位连接或时点填充后仍存在缺口的字段 print (' \n 数据范围:' ) # 核对最终记录数、2023年1月窗口和三只证券覆盖 print (f' 总记录数: { len (merged)} ' ) # 与A表事实行数对照，确认连续左连接没有扩张或丢失证券—交易日键 print (f' 日期范围: { merged["datetime" ]. min ()} 至 { merged["datetime" ]. max ()} ' ) # 核对输出仍限定在2023年1月的实际交易日边界 print (f' 股票数量: { merged["stock_name" ]. nunique()} ' ) # 确认四源连接与填充未丢失任一证券 print (' \n 按股票统计:' ) # 按证券汇总收盘价区间与样本期累计成交量 = merged.groupby('stock_name' ).agg({ # 按股票分组计算多列聚合统计 'close' : ['mean' , 'min' , 'max' ], # 收盘价：均值、最小值、最大值 'volume' : 'sum' # 成交量：累计总和 round (2 ) # 价格与成交量汇总表保留两位用于报告展示 print (stock_stats) # 对照每只证券的价格均值、波动、极值及总成交量，检查证券键分组 print (' \n 按行业统计:' ) # 检查三个细分行业映射后的平均价格与累计成交量 = merged.groupby('industry' ).agg({ # 按行业分组计算聚合统计 'close' : 'mean' , # 收盘价均值 'volume' : 'sum' # 成交量累计总和 round (2 ) # 行业层统计统一保留两位展示精度 print (industry_stats) # 核对一证券一行业映射后各行业的证券数与平均估值水平 print (' \n 评级分布:' ) # 计数按证券前向延续后的每日可用评级类别 = merged['rating' ].value_counts() # 计数延续到交易日后的可用评级日记录数 print (rating_dist) # 核对题设评级按证券向后延续后的交易日覆盖，而非原始事件条数
[6/6] 数据整合报告
============================================================
数据完整性:
datetime: 100.00%
stock_name: 100.00%
open: 100.00%
high: 100.00%
low: 100.00%
close: 100.00%
volume: 100.00%
industry: 100.00%
sector: 100.00%
region: 100.00%
pe_ratio: 0.00%
pb_ratio: 0.00%
dividend_yield: 0.00%
ps_ratio: 0.00%
rating: 93.75%
target_price: 93.75%
数据范围:
总记录数: 48
日期范围: 2023-01-03 00:00:00 至 2023-01-31 00:00:00
股票数量: 3
按股票统计:
close volume
mean min max sum
stock_name
宁波港 3.28 3.23 3.37 132777579.0
宁波银行 30.37 29.38 31.20 447477998.0
恒瑞医药 40.14 37.34 43.12 885924199.0
按行业统计:
close volume
industry
医药制造 40.14 885924199.0
港口航运 3.28 132777579.0
银行 30.37 447477998.0
评级分布:
rating
买入 25
增持 20
Name: count, dtype: int64
# 抽查前10行的 stock_name—datetime 键与行业、PE/PB、评级是否正确共存 print (' \n 最终整合数据样本（前10行）:' ) # 抽查行情键、行业、估值与评级列是否共存于同一行 = ['datetime' , 'stock_name' , 'industry' , 'close' , 'volume' , # 展示列：日期、股票、行业、价格、量 'pe_ratio' , 'pb_ratio' , 'rating' , 'target_price' ] # 展示列：估值因子、评级、目标价 print (merged[display_cols].head(10 ).to_string()) # 抽查证券—交易日键与行业、估值、评级字段在同一输出 schema 中正确共存 # 按保真存储、跨平台交换和人工查阅三种下游用途给出导出路径 print (' \n === 导出建议 ===' ) # 区分保真存储、跨平台交换和人工复核三类下游交付用途 print ('整合后的数据可以:' ) # 说明随后三种导出建议面向保真、交换与人工查阅场景 print ('1. 推荐导出为HDF5: merged.to_hdf("integrated_data.h5", key="data")' ) # 推荐使用HDF5格式存储数据 print ('2. 导出为CSV: merged.to_csv("integrated_data.csv")' ) # CSV格式可用于跨平台交换 print ('3. 导出为Excel: merged.to_excel("integrated_data.xlsx")' ) # Excel 保留二维整合表供人工复核日期—股票键及指标列
最终整合数据样本（前10行）:
datetime stock_name industry close volume pe_ratio pb_ratio rating target_price
0 2023-01-03 宁波港 港口航运 3.2761 10743034.0 NaN NaN 增持 5.8
1 2023-01-04 宁波港 港口航运 3.2944 8787221.0 NaN NaN 增持 5.8
2 2023-01-05 宁波港 港口航运 3.2852 10407133.0 NaN NaN 增持 5.8
3 2023-01-06 宁波港 港口航运 3.2669 10450610.0 NaN NaN 增持 5.8
4 2023-01-09 宁波港 港口航运 3.2669 6518398.0 NaN NaN 增持 5.8
5 2023-01-10 宁波港 港口航运 3.2485 5882400.0 NaN NaN 买入 6.2
6 2023-01-11 宁波港 港口航运 3.2394 4453499.0 NaN NaN 买入 6.2
7 2023-01-12 宁波港 港口航运 3.2302 4491700.0 NaN NaN 买入 6.2
8 2023-01-13 宁波港 港口航运 3.2577 4929600.0 NaN NaN 买入 6.2
9 2023-01-16 宁波港 港口航运 3.2761 8098816.0 NaN NaN 买入 6.2
=== 导出建议 ===
整合后的数据可以:
1. 推荐导出为HDF5: merged.to_hdf("integrated_data.h5", key="data")
2. 导出为CSV: merged.to_csv("integrated_data.csv")
3. 导出为Excel: merged.to_excel("integrated_data.xlsx")
关键要点 ： - 真实项目往往需要整合多个异构数据源 - 分步整合比一次性合并更容易调试 - 不同数据源可能需要不同的填充策略 - 在合并时注意键的类型和格式一致性 - 创建详细的整合报告有助于验证数据质量 - 选择合适的存储格式取决于后续使用场景 - 数据整合是数据科学项目的核心环节
习题 7.8: 报告期与发布日期的时点安全连接
问题描述 ：
下面给出同一公司的四个收盘后信息时点，以及两份机制演示财务记录。report_period 表示财务数字归属的会计期末，publish_time 表示该数字开始可被市场使用的时间。数值是固定题设，不代表真实公司业绩。
以 trade_information_time 为左键、publish_time 为右键，使用向后 merge_asof 连接最近一份已发布报告。
断言每个已匹配报告满足 publish_time <= trade_information_time，且 2023-04-27 收盘时还看不到 2023-03-31 报告。
说明若按 report_period 连接并向后填充，会在什么意义上产生前视偏差。
评分标准：安全连接 4 分，两项时点断言 3 分，错误连接反例 2 分，推断边界说明 1 分。
完整解答 ：
代码清单 列表 7.2 用两个日内信息时点断言检验发布日期连接。
= pd.DataFrame({ # 构造四个收盘后可用信息时点的机制演示行情表 'order_book_id' : ['600276.XSHG' ] * 4 , # 固定同一证券以隔离时间键逻辑 'trade_information_time' : pd.to_datetime(['2023-04-27 15:00' , '2023-04-28 15:00' , '2023-08-24 15:00' , '2023-08-25 15:00' ]), # 明确日内信息截止时刻 'close' : [45.0 , 45.4 , 46.2 , 46.8 ], # 使用固定题设价格承载连接结果 = pd.DataFrame({ # 构造报告期与真实可得时点分离的机制演示报告表 'order_book_id' : ['600276.XSHG' , '600276.XSHG' ], # 保持证券键与行情表一致 'report_period' : pd.to_datetime(['2023-03-31' , '2023-06-30' ]), # 记录财务数字所属会计期末 'publish_time' : pd.to_datetime(['2023-04-28 08:00' , '2023-08-25 08:00' ]), # 记录市场首次可用的题设发布时间 'net_profit_cny_million' : [1200.0 , 2500.0 ], # 固定题设利润仅用于检验连接语义 = pd.merge_asof( # 对每个行情信息时点只回看此前已发布报告 'trade_information_time' ), # 左表必须按asof时间键排序 'publish_time' ), # 右表按信息可得时点排序而非报告期排序 = 'trade_information_time' , right_on= 'publish_time' , by= 'order_book_id' , direction= 'backward' , # 指定证券内向后匹配 = safe_panel['publish_time' ].notna() # 仅在已有报告可见的行上检查时间不等式 assert (safe_panel.loc[matched_rows, 'publish_time' ] <= safe_panel.loc[matched_rows, 'trade_information_time' ]).all () # 禁止任何未来报告进入特征行 assert pd.isna(safe_panel.loc[0 , 'report_period' ]) # 核验首份报告发布前一天仍不可见 assert safe_panel.loc[1 , 'report_period' ] == pd.Timestamp('2023-03-31' ) # 核验发布日收盘后可使用首份报告 # 展示每个收盘信息时点实际可见的报告期与利润
若把 2023-03-31 当作信息可得日并向后填充，那么 4 月 1 日至 4 月 27 日的行会提前读到 4 月 28 日才发布的数字。report_period 回答“这份报告描述哪个期间”，publish_time 才回答“分析者何时能知道它”。本题只证明连接满足信息时点约束；利润与价格同表出现并不构成预测有效性或因果关系证据。
结论
本章建立了 pandas 数据导入、清洗和重组的核心方法。分层索引、合并、连接与重塑不是孤立的机械步骤，而是把原始信息转换为可验证分析样本的结构约束；键的唯一性、连接基数和缺失模式会直接影响后续实证结论是否可靠。
章节导航 ：上一章为数据清洗与准备 ，下一章为数据聚合与分组操作 。