12 数据管理：从多表合并到可复用工作流

运行完数据获取脚本之后，data_raw/ 文件夹里通常已经有了很多文件：个股日度行情、市场指数、宏观指标、年度财务表、公司信息表等。数据一多，真正麻烦的地方就不再是「如何读入一个 CSV」，而是如何把这些文件组织成一套可以反复查询、合并和复用的数据工作流。

假设你要做这样一个分析：找出 2023 年以来，制造业上市公司中换手率最高的 10 个交易日，同时看看这些日期对应的 Shibor 利率是多少。用散落的原始文件也能做，但你需要先循环读入多个 CSV，再合并公司信息，再把日度行情和月度利率对齐。每次分析都重复一遍，代码会越来越长，也越来越容易出错。

本章讨论的就是这个问题：数据多了之后，如何管好它们。

本章主线可以概括为：

\[\text{source files}\rightarrow \text{keys and grain}\rightarrow \text{analysis base table}\rightarrow \text{storage and workflow}\]

这不是要用数据库替代 pandas。更恰当的理解是：pandas 仍然是核心分析工具，但在它的上游，需要先把多来源数据表的主键、粒度、存储格式和查询方式理清楚。

12.1 本章导读

本章的目标不是让你记住所有数据库术语，而是建立三种意识。

粒度意识：每张表的每一行代表什么？是一家公司、一家公司一年，还是一家公司一个交易日？
主键意识：什么变量或变量组合能唯一识别一行？主键不对，merge 很可能悄悄出错。
复用意识：数据组织好了，每次分析只需要查询和调用，不必反复读文件、反复合并。

本章使用的数据集如下。

数据集	文件	粒度	主键	说明
个股日度行情	`data_raw/stock_daily/*.csv`	firm-date	`code + date`	每只股票每天一行
市场指数日度	`data_raw/index_daily/*.csv`	index-date	`index_code + date`	每个指数每天一行
宏观月度	`data_raw/macro_monthly/*.csv`	month	`date`	每个月一行
年度财务指标	`data_raw/fin_annual/fin_indicators.csv`	firm-year	`code + year`	每家公司每年一行
公司基本信息	`data_raw/company_info.csv`	firm	`code`	每家公司一行

本章从一个 mini-dataset 开始。mini-dataset 的字段名与真实数据保持一致，这样先把概念看清楚，再切换到 data_raw/ 中的真实数据。

12.2 环境准备

本章沿用当前项目中的三个目录：

data_raw/：原始数据及由原始数据生成的本地数据库文件；
data/：后续可以存放处理后的数据和输出结果；
figs/：本章插图。

下面的代码只检查路径和导入常用包，不移动、不重命名任何已有文件。

import glob
import sqlite3
import timeit
import subprocess
import sys
from pathlib import Path

import pandas as pd
from IPython.display import display

BASE = Path('data_raw')
DATA = Path('data')
FIGS = Path('figs')

print('当前工作目录：', Path.cwd())
print('data_raw 是否存在：', BASE.exists())
print('data 是否存在：', DATA.exists())
print('figs 是否存在：', FIGS.exists())

if not BASE.exists():
    raise FileNotFoundError(
        '未找到 data_raw/。请先运行数据获取 notebook，生成本章所需原始数据。'
    )

当前工作目录： d:\github_lianxh\ds2026\Lecture\data_manage
data_raw 是否存在： True
data 是否存在： True
figs 是否存在： True

12.3 一个金融研究中的多表场景

先看三张最小数据表。它们的数据量很小，但涵盖了金融实证中最常见的三种粒度。

表名	粒度	主键	用途
`company_mini`	firm	`code`	公司基本信息
`fin_mini`	firm-year	`code, year`	年度财务指标
`daily_mini`	firm-date	`code, date`	日度行情

这三张表不能随便合并。真正需要先判断的是：最终分析底表的目标粒度是什么？如果目标是 firm-year，那么日度行情表就不能直接并入年度财务表，而应先聚合到年度。

下图把本章前半部分的逻辑放在一起：先识别每张原始表的粒度和主键，再判断哪些表可以直接合并，哪些表必须先聚合，最后形成 analysis_base。

在这个例子中，fin_annual 已经是 firm-year 粒度，可以作为分析底表的基础；stock_daily 是 firm-date 粒度，需要先聚合为 ret_yearly；company_info 是公司层面的静态表，可以按 code 并入。

# ── Mini-dataset：3 只股票，字段名与 data_raw/ 尽量保持一致 ──
# 后面切换到真实数据时，分析逻辑基本不变。

company_mini = pd.DataFrame({
    'code'       : ['000001', '600519', '300750'],
    'name'       : ['平安银行', '贵州茅台', '宁德时代'],
    'industry_l1': ['金融',    '消费',    '制造'],
    'ownership'  : ['国企',    '民企',    '民企'],
})

fin_mini = pd.DataFrame({
    'code'  : ['000001','000001','600519','600519','300750','300750'],
    'year'  : [2022,    2023,    2022,    2023,    2022,    2023   ],
    'pe_ttm': [5.2,     5.8,     28.4,    24.1,    35.6,    22.3  ],
    'pb'    : [0.6,     0.6,     9.1,     8.4,     4.2,     3.1   ],
})

daily_mini = pd.DataFrame({
    'code'    : ['000001']*3 + ['600519']*3,
    'date'    : ['2024-01-02','2024-01-03','2024-01-04'] * 2,
    'close'   : [10.2, 10.5, 10.3, 1680.0, 1695.0, 1702.0],
    'pct_chg' : [-0.5,  2.9, -1.9,    0.3,    0.9,    0.4 ],
    'turnover': [ 0.8,  1.2,  0.9,    0.2,    0.3,    0.2 ],
})
daily_mini['date'] = pd.to_datetime(daily_mini['date'])

print('三张表的粒度一览：')
print(f'  company_mini：{len(company_mini)} 行（每行 = 一家公司）')
print(f'  fin_mini    ：{len(fin_mini)} 行（每行 = 一家公司 × 一年）')
print(f'  daily_mini  ：{len(daily_mini)} 行（每行 = 一家公司 × 一个交易日）')

三张表的粒度一览：
  company_mini：3 行（每行 = 一家公司）
  fin_mini    ：6 行（每行 = 一家公司 × 一年）
  daily_mini  ：6 行（每行 = 一家公司 × 一个交易日）

print('company_mini：公司基本信息')
display(company_mini)

print('fin_mini：年度财务指标')
display(fin_mini)

print('daily_mini：日度行情')
display(daily_mini)

company_mini：公司基本信息

	code	name	industry_l1	ownership
0	000001	平安银行	金融	国企
1	600519	贵州茅台	消费	民企
2	300750	宁德时代	制造	民企

fin_mini：年度财务指标

	code	year	pe_ttm	pb
0	000001	2022	5.2	0.6
1	000001	2023	5.8	0.6
2	600519	2022	28.4	9.1
3	600519	2023	24.1	8.4
4	300750	2022	35.6	4.2
5	300750	2023	22.3	3.1

daily_mini：日度行情

	code	date	close	pct_chg	turnover
0	000001	2024-01-02	10.2	-0.5	0.8
1	000001	2024-01-03	10.5	2.9	1.2
2	000001	2024-01-04	10.3	-1.9	0.9
3	600519	2024-01-02	1680.0	0.3	0.2
4	600519	2024-01-03	1695.0	0.9	0.3
5	600519	2024-01-04	1702.0	0.4	0.2

12.4 主键与粒度：合并前先问清楚三件事

数据管理中最容易被忽略的两个概念是主键和粒度。

主键是唯一识别一行数据的变量或变量组合。例如，company_mini 的主键是 code，fin_mini 的主键是 (code, year)，daily_mini 的主键是 (code, date)。

粒度是每一行数据对应的观测层级。例如，公司基本信息是 firm 粒度，年度财务指标是 firm-year 粒度，日度行情是 firm-date 粒度。

在合并数据前，至少要问三个问题：

这张表的一行代表什么？
哪些变量可以唯一识别一行？
最终分析底表的目标粒度是什么？

如果候选主键为 \(K\)，数据表行数为 \(N\)，那么主键检查可以写成：

\[\operatorname{unique}(K)=N\]

若唯一主键数小于行数，就说明存在重复主键。此时继续 merge，往往会导致样本量膨胀或重复匹配。

def check_key(df, keys, table_name):
    """
    检查一张表中给定主键是否唯一。

    参数
    ----
    df : pandas.DataFrame
        待检查的数据表。
    keys : list
        候选主键变量名列表。
    table_name : str
        数据表名称，用于打印提示信息。
    """
    n_rows = len(df)
    n_unique = df[keys].drop_duplicates().shape[0]
    has_dup = df.duplicated(subset=keys).any()

    print(f'{table_name:<15} 行数：{n_rows:<6} 唯一主键数：{n_unique:<6} 是否重复：{has_dup}')

checks = [
    ('company_mini', company_mini, ['code']),
    ('fin_mini',     fin_mini,     ['code', 'year']),
    ('daily_mini',   daily_mini,   ['code', 'date']),
]

for name, df, keys in checks:
    check_key(df, keys, name)

company_mini    行数：3      唯一主键数：3      是否重复：False
fin_mini        行数：6      唯一主键数：6      是否重复：False
daily_mini      行数：6      唯一主键数：6      是否重复：False

12.5 一个典型错误：忽略粒度直接合并

假设目标是构造 firm-year 层面的分析底表。如果把 fin_mini 和 daily_mini 只按 code 直接合并，就会出现一个很隐蔽的错误：年度财务数据会被复制到每一个交易日上。

这个错误本质上不是语法错误，而是粒度错误。代码可以正常运行，但生成的表已经不再是 firm-year 数据。

下面这张图展示了错误做法与正确做法的差异。左侧的问题在于直接把 firm-year 表和 firm-date 表按 code 合并，导致行数膨胀；右侧的做法是先把日度表聚合到 firm-year，再按 (code, year) 合并。

读这张图时，可以重点看两点：一是错误合并后 code, year 不再唯一；二是正确做法始终围绕目标粒度 firm-year 展开。

# ── 错误做法：只按 code 合并不同粒度的表 ──────────────
wrong = fin_mini.merge(daily_mini, on='code', how='left')

print(f'fin_mini    原始行数：{len(fin_mini)}')
print(f'daily_mini  原始行数：{len(daily_mini)}')
print(f'合并后行数：         {len(wrong)}   ← 行数发生膨胀')
print()
display(wrong)

print('合并后 (code, year) 是否重复：', wrong.duplicated(['code', 'year']).any())

fin_mini    原始行数：6
daily_mini  原始行数：6
合并后行数：         14   ← 行数发生膨胀

	code	year	pe_ttm	pb	date	close	pct_chg	turnover
0	000001	2022	5.2	0.6	2024-01-02	10.2	-0.5	0.8
1	000001	2022	5.2	0.6	2024-01-03	10.5	2.9	1.2
2	000001	2022	5.2	0.6	2024-01-04	10.3	-1.9	0.9
3	000001	2023	5.8	0.6	2024-01-02	10.2	-0.5	0.8
4	000001	2023	5.8	0.6	2024-01-03	10.5	2.9	1.2
5	000001	2023	5.8	0.6	2024-01-04	10.3	-1.9	0.9
6	600519	2022	28.4	9.1	2024-01-02	1680.0	0.3	0.2
7	600519	2022	28.4	9.1	2024-01-03	1695.0	0.9	0.3
8	600519	2022	28.4	9.1	2024-01-04	1702.0	0.4	0.2
9	600519	2023	24.1	8.4	2024-01-02	1680.0	0.3	0.2
10	600519	2023	24.1	8.4	2024-01-03	1695.0	0.9	0.3
11	600519	2023	24.1	8.4	2024-01-04	1702.0	0.4	0.2
12	300750	2022	35.6	4.2	NaT	NaN	NaN	NaN
13	300750	2023	22.3	3.1	NaT	NaN	NaN	NaN

合并后 (code, year) 是否重复： True

原本 6 行的年度财务表，合并之后变成了更多行。原因是：同一家公司在 fin_mini 里有多个年度观测，在 daily_mini 里又有多个交易日观测。只按 code 合并，会产生多对多匹配。

正确做法是先把日度数据聚合到目标粒度。如果最终分析底表是 firm-year，日度行情需要先转换为年度特征，例如年均收益率、年波动率、年均换手率。

# ── 正确做法：先把日度数据聚合到 firm-year，再 merge ──

# 第一步：从日期提取年份
daily_agg = daily_mini.copy()
daily_agg['year'] = daily_agg['date'].dt.year

# 第二步：按 code + year 聚合，计算年度统计量
daily_agg = (
    daily_agg
    .groupby(['code', 'year'], as_index=False)
    .agg(
        avg_ret      = ('pct_chg',  'mean'),
        volatility   = ('pct_chg',  'std'),
        avg_turnover = ('turnover', 'mean'),
    )
)

print('按 code + year 聚合后的日度数据：')
display(daily_agg)

# 第三步：两张表粒度一致，才能安全 merge
analysis_mini = fin_mini.merge(daily_agg, on=['code', 'year'], how='left')
analysis_mini = analysis_mini.merge(company_mini, on='code', how='left')

print(f'合并后行数：{len(analysis_mini)}（与 fin_mini 原始行数一致，没有膨胀）')
check_key(analysis_mini, ['code', 'year'], 'analysis_mini')
display(analysis_mini.round(3))

按 code + year 聚合后的日度数据：

	code	year	avg_ret	volatility	avg_turnover
0	000001	2024	0.166667	2.468468	0.966667
1	600519	2024	0.533333	0.321455	0.233333

合并后行数：6（与 fin_mini 原始行数一致，没有膨胀）
analysis_mini   行数：6      唯一主键数：6      是否重复：False

	code	year	pe_ttm	pb	avg_ret	volatility	avg_turnover	name	industry_l1	ownership
0	000001	2022	5.2	0.6	NaN	NaN	NaN	平安银行	金融	国企
1	000001	2023	5.8	0.6	NaN	NaN	NaN	平安银行	金融	国企
2	600519	2022	28.4	9.1	NaN	NaN	NaN	贵州茅台	消费	民企
3	600519	2023	24.1	8.4	NaN	NaN	NaN	贵州茅台	消费	民企
4	300750	2022	35.6	4.2	NaN	NaN	NaN	宁德时代	制造	民企
5	300750	2023	22.3	3.1	NaN	NaN	NaN	宁德时代	制造	民企

这个模式在真实数据里会反复出现：日度数据 → 年度聚合 → 与财务表 merge。它是金融实证研究中最典型的数据整合流程之一。

AI 提示词：检查粒度与主键

我有两张 pandas DataFrame：

df_annual（字段：code, year, roe, leverage，粒度为 firm-year）
df_daily（字段：code, date, pct_chg, volume，粒度为 firm-date）

我想构造一张 firm-year 分析底表，把日度收益率的年均值和年波动率并入财务数据。请帮我：
(1) 先验证两张表的主键是否唯一；
(2) 把日度数据聚合为年度；
(3) 按 (code, year) 合并，用 left join；
(4) 输出合并后的行数，确认没有行数膨胀。

12.6 存储格式的选择

理清主键和粒度之后，下一步才是选择存储方式。这里没有唯一正确答案，关键看使用场景。

格式	最适合	不适合	关键特点
CSV	分享、人工查看、小数据	大文件、频繁读写	无类型信息，通用性强
Excel	人工录入、少量展示	可复现批量分析	易读，但不适合自动化流程
Parquet	大表、重复读取、列查询	人工直接查看	列式存储，压缩好，速度快
SQLite	多表关联、反复查询	高并发写入	单文件数据库，支持 SQL
DuckDB	大数据量分析查询	频繁小事务写入	可直接查询 Parquet / CSV

图中把格式选择拆成几个判断问题。实际项目中，先判断数据是用于人工查看、反复读取、多表查询，还是本地大文件分析，再决定使用 CSV、Excel、Parquet、SQLite 或 DuckDB。

Parquet 如何发音？

Parquet 通常读作 par-kay，接近 /pɑːrˈkeɪ/。这个词原本也有「拼花木地板」的意思。放到数据存储中，可以把它理解为一种面向分析场景的列式表格文件格式。它不像 CSV 那样适合用文本编辑器直接打开查看，但更适合机器读取、压缩和反复分析。

# ── CSV 与 Parquet 的文件大小和读取速度对比 ───────────
# 下面使用 data_raw/stock_daily/000001.csv 作为样本。

csv_path     = BASE / 'stock_daily' / '000001.csv'
parquet_path = BASE / 'stock_daily' / '000001.parquet'

if not csv_path.exists():
    raise FileNotFoundError(f'未找到示例文件：{csv_path}')

df_sample = pd.read_csv(csv_path)

# 若本地没有 pyarrow，则尝试安装；如果安装失败，会给出清晰提示。
try:
    import pyarrow as pa
except ImportError:
    print('当前环境未安装 pyarrow，正在尝试安装。')
    subprocess.check_call([sys.executable, '-m', 'pip', 'install', 'pyarrow', '-q'])
    import pyarrow as pa

try:
    try:
        pa.unregister_extension_type('pandas.period')
    except Exception:
        pass
    df_sample.to_parquet(parquet_path, index=False)
except Exception as e:
    raise RuntimeError(f'Parquet 写入失败，请检查 pyarrow 安装状态。原始错误：{e}')

# ── 文件大小对比 ──────────────────────────────────────
csv_kb = csv_path.stat().st_size / 1024
pq_kb  = parquet_path.stat().st_size / 1024
print(f'文件大小对比（平安银行 {len(df_sample):,} 行）：')
print(f'  CSV     ：{csv_kb:.1f} KB')
print(f'  Parquet ：{pq_kb:.1f} KB')
print(f'  压缩比  ：{csv_kb / pq_kb:.1f}x')

# ── 读取速度对比（各测 30 次取均值）──────────────────
t_csv = timeit.timeit(lambda: pd.read_csv(csv_path),         number=30) / 30
t_pq  = timeit.timeit(lambda: pd.read_parquet(parquet_path), number=30) / 30

print(f'\n读取速度对比（30 次均值）：')
print(f'  CSV     ：{t_csv*1000:.1f} ms')
print(f'  Parquet ：{t_pq*1000:.1f} ms')
print(f'  速度提升：{t_csv / t_pq:.1f}x')

文件大小对比（平安银行 1,510 行）：
  CSV     ：197.9 KB
  Parquet ：105.1 KB
  压缩比  ：1.9x

读取速度对比（30 次均值）：
  CSV     ：8.1 ms
  Parquet ：4.1 ms
  速度提升：2.0x

单只股票 1510 行的数据量还不大，差距通常不会特别夸张。Parquet 的优势在数据量达到几十万行以上时会更明显。除此之外，Parquet 还会保留字段类型：CSV 里日期列读进来默认可能是字符串，Parquet 可以保存更明确的类型信息，减少每次手工转换的麻烦。

下图把 CSV 和 Parquet 的差异画成两种存储方式。CSV 更像逐行写下来的文本表，Parquet 更像按列分块保存的分析型文件。

什么时候应该用 Parquet 而不是 CSV？

这份数据会被反复读取，每次分析都要用；
文件超过 10 MB；
经常只读取其中几列，而不是每次读取全部字段；
需要保留日期、布尔值等类型信息；
这份数据已经是清洗后的标准表，不再主要用于人工查看。

如果只是偶尔查看，或者需要发给别人、上传到系统，CSV 依然是更方便的选择。

12.7 SQLite：把多张结构化表放进一个数据库文件

SQLite 是 Python 标准库内置的关系型数据库，不需要安装服务器，不需要配置账号密码，整个数据库就是一个 .db 文件。对分析者来说，它解决的核心问题是：把多张表统一放在一个地方，用 SQL 查询替代反复的 read_csv + merge。

SQLite 这个名字是什么意思？

SQLite 可以粗略理解为 SQL + lite。

其中，SQL 指 Structured Query Language，即结构化查询语言；lite 是 lightweight 的常见表达，意思是「轻量级」。因此，SQLite 的直观含义就是一种轻量级 SQL 数据库。它不需要单独启动数据库服务器，整个数据库通常就是一个本地 .db 或 .sqlite 文件。

下面先用 mini-dataset 建一个内存数据库，把基本操作看清楚；再把 data_raw/ 中的真实文件导入一个持久化数据库。

下图概括了 SQLite 在本章中的位置：它把多张结构化表放进一个本地数据库文件，用 SQL 完成筛选、聚合和连接，再把结果读回 pandas。

12.7.1 用 mini-dataset 理解建库操作

# ':memory:' 表示在内存中建库，不写磁盘，适合演示和测试。
# 真实项目通常使用文件路径，例如 sqlite3.connect('data_raw/finance.db')。

conn_mini = sqlite3.connect(':memory:')

# pandas 的 to_sql() 可以把 DataFrame 写入数据库表。
# 需要说明的是，to_sql() 不会自动建立主键约束；这里先用于教学演示。
company_mini.to_sql('company',     conn_mini, if_exists='replace', index=False)
fin_mini.to_sql(    'fin_annual',  conn_mini, if_exists='replace', index=False)
daily_mini.to_sql(  'stock_daily', conn_mini, if_exists='replace', index=False)

# 查看数据库中有哪些表
tables = pd.read_sql(
    "SELECT name FROM sqlite_master WHERE type='table'",
    conn_mini
)
print('数据库中的表：')
display(tables)

数据库中的表：

	name
0	company
1	fin_annual
2	stock_daily

# ── SELECT + WHERE + ORDER BY ──────────────────────────
# pd.read_sql() 直接把查询结果返回为 DataFrame。

query = """
SELECT code, year, pe_ttm, pb
FROM   fin_annual
WHERE  pe_ttm < 10
ORDER  BY pe_ttm ASC
"""
pd.read_sql(query, conn_mini)

	code	year	pe_ttm	pb
0	000001	2022	5.2	0.6
1	000001	2023	5.8	0.6

# ── GROUP BY + JOIN ─────────────────────────────────────
# 按行业计算平均 pe_ttm，先 join 公司信息获取行业字段。

query = """
SELECT   c.industry_l1,
         COUNT(*)               AS n,
         ROUND(AVG(f.pe_ttm), 2) AS avg_pe
FROM     fin_annual f
JOIN     company    c  ON f.code = c.code
WHERE    f.year = 2023
GROUP BY c.industry_l1
ORDER BY avg_pe
"""
pd.read_sql(query, conn_mini)

	industry_l1	n	avg_pe
0	金融	1	5.8
1	制造	1	22.3
2	消费	1	24.1

上面这段 SQL 对应的 pandas 写法是：

merged = fin_mini.merge(company_mini, on='code')
merged[merged['year'] == 2023].groupby('industry_l1')['pe_ttm'].agg(['count', 'mean'])

两种写法都能得到同样的结果。当数据已经在数据库里、需要多表关联时，SQL 的表达往往更直接。

12.7.2 真实数据建库

下面把 data_raw/ 中的多个原始文件写入同一个 SQLite 数据库 data_raw/finance.db。这一步完成之后，后续分析就不必每次从多个文件夹中循环读入 CSV，而可以直接写 SQL 查询。

DB_PATH = BASE / 'finance.db'
conn = sqlite3.connect(DB_PATH)

# ── 1. 个股日度：10 个 CSV 合并写入一张表 ─────────────
files = sorted(glob.glob(str(BASE / 'stock_daily/*.csv')))
if not files:
    raise FileNotFoundError('未在 data_raw/stock_daily/ 中找到 CSV 文件。')

df_stock = pd.concat([pd.read_csv(f) for f in files], ignore_index=True)
df_stock.to_sql('stock_daily', conn, if_exists='replace', index=False)
print(f'stock_daily 写入完成：{len(df_stock):,} 行')

# ── 2. 市场指数 ──────────────────────────────────────
files = sorted(glob.glob(str(BASE / 'index_daily/*.csv')))
if files:
    df_idx = pd.concat([pd.read_csv(f) for f in files], ignore_index=True)
    df_idx.to_sql('index_daily', conn, if_exists='replace', index=False)
    print(f'index_daily 写入完成：{len(df_idx):,} 行')
else:
    print('未找到 index_daily CSV 文件，跳过 index_daily。')

# ── 3. 宏观月度：文件名作为表名 ─────────────────────
for f in sorted(glob.glob(str(BASE / 'macro_monthly/*.csv'))):
    tbl = Path(f).stem
    df = pd.read_csv(f)
    df.to_sql(tbl, conn, if_exists='replace', index=False)
    print(f'{tbl} 写入完成：{len(df)} 行')

# ── 4. 年度财务 ──────────────────────────────────────
fin_path = BASE / 'fin_annual' / 'fin_indicators.csv'
if not fin_path.exists():
    raise FileNotFoundError(f'未找到年度财务文件：{fin_path}')

df_fin = pd.read_csv(fin_path)
df_fin.to_sql('fin_annual', conn, if_exists='replace', index=False)
print(f'fin_annual 写入完成：{len(df_fin)} 行')

# ── 5. 公司信息：兼容 key-value 长格式和普通宽格式 ─────
co_path = BASE / 'company_info.csv'
if not co_path.exists():
    raise FileNotFoundError(f'未找到公司信息文件：{co_path}')

df_co_raw = pd.read_csv(co_path)
if len(df_co_raw) > 20:
    key_map = {
        '行业': 'industry_l1',
        '地区': 'province',
        '上市时间': 'list_date',
        '实控人': 'controller'
    }
    col0, col1 = df_co_raw.columns[0], df_co_raw.columns[1]
    records = []
    code_col = [c for c in df_co_raw.columns if 'code' in c.lower()]

    if code_col:
        for code_val in df_co_raw[code_col[0]].unique():
            sub = df_co_raw[df_co_raw[code_col[0]] == code_val]
            row = {'code': code_val}
            for _, r in sub.iterrows():
                k = str(r[col0])
                if k in key_map:
                    row[key_map[k]] = r[col1]
            if 'controller' in row:
                row['ownership'] = '国企' if any(
                    kw in str(row['controller'])
                    for kw in ['国资', '国有', '国家', '政府']
                ) else '民企'
            records.append(row)
        df_co = pd.DataFrame(records)
    else:
        df_co = df_co_raw.head(10)
else:
    df_co = df_co_raw

# 补充股票名称
names = df_stock[['code', 'name']].drop_duplicates('code')
df_co = df_co.merge(names, on='code', how='left')
df_co.to_sql('company_info', conn, if_exists='replace', index=False)
print(f'company_info 写入完成：{len(df_co)} 行')
print(f'\n数据库已保存至 {DB_PATH}')

stock_daily 写入完成：15,100 行
index_daily 写入完成：4,533 行
cpi_monthly 写入完成：74 行
shibor_3m 写入完成：75 行
usd_cny 写入完成：74 行
fin_annual 写入完成：60 行
company_info 写入完成：10 行

数据库已保存至 data_raw\finance.db

# ── 验证：查看各表行数 ──────────────────────────────
tables = pd.read_sql(
    "SELECT name FROM sqlite_master WHERE type='table' ORDER BY name",
    conn
)

print(f'数据库大小：{DB_PATH.stat().st_size / 1024 / 1024:.1f} MB\n')
print(f'{"表名":<20} {"行数":>8}')
print('-' * 30)
for tbl in tables['name']:
    n = pd.read_sql(f'SELECT COUNT(*) AS n FROM {tbl}', conn)['n'][0]
    print(f'{tbl:<20} {n:>8,}')

数据库大小：1.8 MB

表名                         行数
------------------------------
company_info               10
cpi_monthly                74
fin_annual                 60
index_daily             4,533
shibor_3m                  75
stock_daily            15,100
usd_cny                    74

多个散落的 CSV 文件现在被统一放入一个 .db 文件中，可以随时用 SQL 查询和关联。

与「每次 concat 再 merge」相比，建库的价值在于：第一次建库需要一些时间；之后每次分析只需要写 SQL，不再循环读文件。查询时也可以只取需要的行和列，不必把全量数据都读进内存。

AI 提示词：把多个 CSV 导入 SQLite

我有一个文件夹 data_raw/stock_daily/，里面有 10 个 CSV，字段包括 date, open, close, high, low, volume, amount, pct_chg, turnover, code, name。请帮我写 Python 代码，把这 10 个文件合并后写入 SQLite 数据库 data_raw/finance.db 的 stock_daily 表，写入后验证总行数和时间范围，并用 try/except 处理读取错误。

12.8 SQL 核心操作

SQL 不是一套完全陌生的新技能，而是你已经掌握的很多 pandas 操作的另一种表达方式。

操作	pandas 写法	SQL 写法
选列	`df[['code','close']]`	`SELECT code, close`
筛选行	`df[df['pct_chg'] > 5]`	`WHERE pct_chg > 5`
排序	`df.sort_values('close', ascending=False)`	`ORDER BY close DESC`
取前 N 行	`df.nlargest(10, 'volume')`	`ORDER BY volume DESC LIMIT 10`
分组均值	`df.groupby('code')['pct_chg'].mean()`	`GROUP BY code` + `AVG(pct_chg)`
多表合并	`pd.merge(df1, df2, on='code')`	`JOIN ... ON code`
新建列	`df['x'] = df['pct_chg'] * 252`	`pct_chg * 252 AS x`
提取年份	`df['date'].dt.year`	`STRFTIME('%Y', date)`

下面用真实数据演示四个有实际分析意义的查询。

# 如果重新打开 notebook，可以从这里重新连接数据库。
conn = sqlite3.connect(BASE / 'finance.db')

# ── 查询 1：各股票 2024 年表现概览 ─────────────────────
# 计算年化收益率、年化波动率、平均换手率，并关联公司信息。

q1 = """
SELECT
    s.code,
    c.name,
    c.industry_l1,
    c.ownership,
    ROUND(AVG(s.pct_chg), 3)                   AS avg_daily_ret,
    ROUND(AVG(s.pct_chg) * 252, 1)             AS ret_annualized,
    ROUND(AVG(ABS(s.pct_chg)) * 15.87, 2)      AS vol_annualized,
    ROUND(AVG(s.turnover), 3)                  AS avg_turnover
FROM   stock_daily  s
JOIN   company_info c  ON s.code = c.code
WHERE  s.date BETWEEN '2024-01-01' AND '2024-12-31'
GROUP  BY s.code
ORDER  BY ret_annualized DESC
"""

# Check actual columns in company_info
cols = pd.read_sql("SELECT * FROM company_info LIMIT 1", conn).columns.tolist()
print("company_info 字段：", cols)

# Rebuild query using only available columns
q1 = """
SELECT
    s.code,
    c.name,
    ROUND(AVG(s.pct_chg), 3)                   AS avg_daily_ret,
    ROUND(AVG(s.pct_chg) * 252, 1)             AS ret_annualized,
    ROUND(AVG(ABS(s.pct_chg)) * 15.87, 2)      AS vol_annualized,
    ROUND(AVG(s.turnover), 3)                  AS avg_turnover
FROM   stock_daily  s
JOIN   company_info c  ON s.code = c.code
WHERE  s.date BETWEEN '2024-01-01' AND '2024-12-31'
GROUP  BY s.code
ORDER  BY ret_annualized DESC
"""

df_q1 = pd.read_sql(q1, conn)
print(f'查询结果：{len(df_q1)} 只股票')
df_q1

company_info 字段： ['code', 'name']
查询结果：10 只股票

	code	name	avg_daily_ret	ret_annualized	vol_annualized	avg_turnover
0	300750	宁德时代	0.253	63.8	29.23	0.655
1	600036	招商银行	0.179	45.1	17.81	0.338
2	2594	比亚迪	0.175	44.1	24.59	1.097
3	333	美的集团	0.164	41.3	20.04	0.473
4	601318	中国平安	0.149	37.6	19.44	0.572
5	1	平安银行	0.142	35.7	17.17	0.731
6	600276	恒瑞医药	0.030	7.5	23.14	0.520
7	600519	贵州茅台	-0.022	-5.5	17.40	0.268
8	2	万科A	-0.109	-27.6	33.61	2.075
9	601012	隆基绿能	-0.112	-28.2	32.47	1.621

# ── 查询 2：找出异常成交日 ───────────────────────────
# 这里把「异常成交日」定义为成交量超过当年均值 3 倍的交易日。

q2 = """
SELECT
    s.code,
    c.name,
    s.date,
    ROUND(s.volume / 10000, 2)  AS volume_wan,
    ROUND(s.pct_chg, 2)         AS pct_chg,
    ROUND(s.turnover, 2)        AS turnover
FROM   stock_daily  s
JOIN   company_info c  ON s.code = c.code
JOIN (
    SELECT code, AVG(volume) AS avg_vol
    FROM   stock_daily
    WHERE  date BETWEEN '2024-01-01' AND '2024-12-31'
    GROUP  BY code
) avg ON s.code = avg.code
WHERE  s.date BETWEEN '2024-01-01' AND '2024-12-31'
  AND  s.volume > avg.avg_vol * 3
ORDER  BY s.volume DESC
LIMIT  20
"""

df_q2 = pd.read_sql(q2, conn)
print(f'2024 年成交量超过年均 3 倍的交易日：{len(df_q2)} 条')
df_q2

2024 年成交量超过年均 3 倍的交易日：20 条

	code	name	date	volume_wan	pct_chg	turnover
0	2	万科A	2024-10-08	109723.0	6.79	11.29
1	2	万科A	2024-05-17	92471.0	10.02	9.52
2	2	万科A	2024-05-20	88096.0	2.00	9.07
3	2	万科A	2024-05-22	71937.0	2.29	7.40
4	2	万科A	2024-09-27	70021.0	9.95	7.21
5	2	万科A	2024-05-16	66358.0	5.82	6.83
6	2	万科A	2024-04-30	63079.0	-1.98	6.49
7	2	万科A	2024-05-23	60622.0	1.70	6.24
8	1	平安银行	2024-10-08	58889.0	5.49	3.03
9	1	平安银行	2024-09-30	54302.0	6.92	2.80
10	1	平安银行	2024-02-21	50552.0	9.98	2.61
11	601012	隆基绿能	2024-10-23	49562.0	6.53	6.54
12	1	平安银行	2024-02-22	49223.0	0.93	2.54
13	601012	隆基绿能	2024-10-08	49121.0	9.79	6.48
14	601012	隆基绿能	2024-10-25	48093.0	10.00	6.35
15	1	平安银行	2024-10-09	42693.0	-9.32	2.20
16	601012	隆基绿能	2024-10-28	42516.0	-1.86	5.61
17	601318	中国平安	2024-10-08	39610.0	8.13	3.68
18	601012	隆基绿能	2024-09-30	39595.0	10.03	5.22
19	600036	招商银行	2024-10-08	29730.0	6.35	1.44

# ── 查询 3：混频 JOIN，日度行情 × 月度 Shibor ──────────
# 关键：用 STRFTIME 把日度日期截取为 YYYY-MM，与月度数据的 date 列做匹配。

q3 = """
SELECT
    s.date,
    s.code,
    c.name,
    ROUND(s.pct_chg, 2)             AS pct_chg,
    r.shibor_3m,
    STRFTIME('%Y-%m', s.date)       AS ym
FROM   stock_daily  s
JOIN   company_info c   ON s.code = c.code
LEFT JOIN shibor_3m r   ON STRFTIME('%Y-%m', s.date) = r.date
WHERE  s.code = '000001'
  AND  s.date >= '2024-01-01'
ORDER  BY s.date
LIMIT  20
"""

df_q3 = pd.read_sql(q3, conn)
print('平安银行日度收益率 + 当月 Shibor（前 20 行）：')
df_q3

平安银行日度收益率 + 当月 Shibor（前 20 行）：

	date	code	name	pct_chg	shibor_3m	ym
0	2024-01-02	1	平安银行	-1.92	2.45	2024-01
1	2024-01-03	1	平安银行	-0.11	2.45	2024-01
2	2024-01-04	1	平安银行	-0.98	2.45	2024-01
3	2024-01-05	1	平安银行	1.76	2.45	2024-01
4	2024-01-08	1	平安银行	-1.29	2.45	2024-01
5	2024-01-09	1	平安银行	0.33	2.45	2024-01
6	2024-01-10	1	平安银行	-0.98	2.45	2024-01
7	2024-01-11	1	平安银行	0.88	2.45	2024-01
8	2024-01-12	1	平安银行	0.22	2.45	2024-01
9	2024-01-15	1	平安银行	0.22	2.45	2024-01
10	2024-01-16	1	平安银行	1.41	2.45	2024-01
11	2024-01-17	1	平安银行	-1.18	2.45	2024-01
12	2024-01-18	1	平安银行	-0.33	2.45	2024-01
13	2024-01-19	1	平安银行	-0.33	2.45	2024-01
14	2024-01-22	1	平安银行	-0.76	2.45	2024-01
15	2024-01-23	1	平安银行	0.66	2.45	2024-01
16	2024-01-24	1	平安银行	1.86	2.45	2024-01
17	2024-01-25	1	平安银行	1.82	2.45	2024-01
18	2024-01-26	1	平安银行	1.26	2.45	2024-01
19	2024-01-29	1	平安银行	0.83	2.45	2024-01

查询 3 展示了混频数据连接的标准做法：用 STRFTIME('%Y-%m', date) 把日度日期截取为月份字符串，与月度数据的 date 列做 JOIN。这样，每个交易日都能找到对应月份的 Shibor，实现「月度数据广播到每个交易日」。

等价的 pandas 写法要更繁琐：先提取月份列，再 merge，再处理键格式不一致。SQL 擅长的正是这类筛选、聚合和多表连接。

# ── 查询 4：日度数据聚合到年度，再与财务表合并 ───────
# 这是金融实证最常见的数据整合模式：日度行情 → 年度风险指标 → 并入财务数据。

fin_cols = pd.read_sql("PRAGMA table_info(fin_annual)", conn)["name"].tolist()

metric_cols = [col for col in ["pe_ttm", "pb", "roe", "current_ratio", "asset_turnover"] if col in fin_cols]
metric_select = ",\n    ".join([f"ROUND(f.{col}, 2) AS {col}" for col in metric_cols])

q4 = f"""
SELECT
    f.code,
    c.name{", c.industry_l1" if "industry_l1" in company_cols else ""},
    f.year{"," if metric_cols else ""}
    {metric_select if metric_cols else ""}
    {"," if metric_cols else ""}
    ROUND(r.avg_ret_daily * 252 * 100, 2) AS ret_ann_pct,
    r.trading_days AS trading_days
FROM fin_annual f
JOIN company_info c ON f.code = c.code
JOIN (
    SELECT
        code,
        CAST(STRFTIME('%Y', date) AS INTEGER) AS year,
        AVG(pct_chg / 100) AS avg_ret_daily,
        COUNT(*) AS trading_days
    FROM stock_daily
    GROUP BY code, STRFTIME('%Y', date)
) r ON f.code = r.code AND f.year = r.year
ORDER BY f.year, ret_ann_pct DESC
"""

df_q4 = pd.read_sql(q4, conn)
print(f'firm-year 分析底表：{len(df_q4)} 行')
df_q4.head(15)

firm-year 分析底表：60 行

	code	name	year	roe	current_ratio	asset_turnover	ret_ann_pct	trading_days
0	2594	比亚迪	2020	0.07	1.05	3.68	162.87	243
1	601012	隆基绿能	2020	0.27	1.28	5.48	144.54	243
2	300750	宁德时代	2020	0.11	2.05	2.57	138.55	243
3	600519	贵州茅台	2020	0.31	4.06	63.37	64.76	243
4	333	美的集团	2020	0.25	1.31	10.99	60.58	243
5	600276	恒瑞医药	2020	0.23	7.44	5.28	49.10	243
6	600036	招商银行	2020	0.15	NaN	NaN	20.31	243
7	601318	中国平安	2020	0.20	NaN	NaN	7.87	243
8	1	平安银行	2020	0.09	NaN	NaN	4.28	243
9	2	万科A	2020	0.20	1.17	167.01	-3.83	243
10	300750	宁德时代	2021	0.21	1.19	5.62	68.18	243
11	2594	比亚迪	2021	0.04	0.97	5.58	48.94	243
12	601012	隆基绿能	2021	0.22	1.39	8.17	43.05	243
13	600036	招商银行	2021	0.15	NaN	NaN	18.75	243
14	600519	贵州茅台	2021	0.30	3.81	NaN	10.59	243

12.8.1 什么时候用 SQL，什么时候用 pandas？

优先用 SQL：数据在数据库里，需要筛选、聚合、多表关联、混频数据连接，或者只需要读取部分行列。

优先用 pandas：数据已经在 DataFrame 中，需要复杂的向量化计算、滚动窗口、信号生成、绘图和可视化。

实际项目中最常见的模式是：SQL 负责数据整合和过滤，pandas 负责最终计算和可视化。

AI 提示词：SQL 多表关联分析

我有一个 SQLite 数据库（data_raw/finance.db），包含 stock_daily（code, date, close, pct_chg, volume, turnover）、company_info（code, name, industry_l1, ownership）、fin_annual（code, year, pe_ttm, pb）、shibor_3m（date 格式 YYYY-MM, shibor_3m）。请帮我写一个 SQL 查询：找出 2023-2024 年间，国有企业中 pe_ttm 连续两年下降且年均日换手率超过 1% 的股票，输出股票代码、名称、两年的 pe_ttm 和平均换手率。用 pd.read_sql() 执行。

12.9 DuckDB + Parquet：直接查询本地大文件

本章的数据量对 SQLite 来说并不大。但在实际工作中，可能会遇到更大规模的数据：全市场 5000 只股票 × 5 年日度数据约 900 万行，或者逐笔成交数据动辄数亿行。这时候，DuckDB 与 Parquet 的组合会更适合分析型查询。

DuckDB 有两个适合金融数据分析的特点：

可以直接查询 Parquet 文件，不必先导入数据库；
面向分析型查询优化，适合筛选、聚合和分组统计。

可以粗略区分三者的位置：

工具	更适合的任务
Parquet	保存清洗后的标准化分析数据
SQLite	管理多张结构化关系表
DuckDB	直接查询本地 Parquet / CSV 文件，并做分析型 SQL

# 如果没有安装 duckdb，代码会给出提示并跳过本节示例。
try:
    import duckdb
    HAS_DUCKDB = True
except ImportError:
    HAS_DUCKDB = False
    print('当前环境没有安装 duckdb。如需运行本节示例，请先执行：pip install duckdb')

PARQUET_PATH = BASE / 'stock_daily' / 'all_stocks.parquet'

if HAS_DUCKDB:
    # 把多只股票合并保存为单个 Parquet 文件
    if not PARQUET_PATH.exists():
        files = sorted(glob.glob(str(BASE / 'stock_daily/*.csv')))
        df_all = pd.concat([pd.read_csv(f) for f in files], ignore_index=True)
        df_all.to_parquet(PARQUET_PATH, index=False)
        print(f'已保存：{PARQUET_PATH.stat().st_size / 1024:.0f} KB')
    else:
        print(f'文件已存在：{PARQUET_PATH.stat().st_size / 1024:.0f} KB')

    # DuckDB 直接查 Parquet，不需要任何导入步骤。
    result = duckdb.query(f"""
        SELECT code, COUNT(*) AS days
        FROM   '{PARQUET_PATH}'
        GROUP  BY code
        ORDER  BY code
    """).df()

    print('各股票交易日数：')
    print(result.to_string(index=False))

已保存：1011 KB
各股票交易日数：
  code  days
     1  1510
     2  1510
   333  1510
  2594  1510
300750  1510
600036  1510
600276  1510
600519  1510
601012  1510
601318  1510

# ── 性能基准测试 ─────────────────────────────────────
# 任务：计算每只股票 2024 年的平均日收益率。
# 三种方式做同一件事，比较时间差异。

if HAS_DUCKDB:
    N = 20
    csv_files = sorted(glob.glob(str(BASE / 'stock_daily/*.csv')))

    def method_csv():
        dfs = [pd.read_csv(f) for f in csv_files]
        df = pd.concat(dfs, ignore_index=True)
        return df[df['date'].str.startswith('2024')].groupby('code')['pct_chg'].mean()

    def method_sqlite():
        return pd.read_sql(
            "SELECT code, AVG(pct_chg) AS avg_ret "
            "FROM stock_daily WHERE date LIKE '2024%' GROUP BY code",
            conn
        )

    def method_duckdb():
        return duckdb.query(f"""
            SELECT code, AVG(pct_chg) AS avg_ret
            FROM   '{PARQUET_PATH}'
            WHERE  date LIKE '2024%'
            GROUP  BY code
        """).df()

    t1 = timeit.timeit(method_csv,    number=N) / N * 1000
    t2 = timeit.timeit(method_sqlite, number=N) / N * 1000
    t3 = timeit.timeit(method_duckdb, number=N) / N * 1000

    print(f'性能对比（{N} 次均值，任务：10 只股票 × 2024 年均收益率）：')
    print(f'  方法 1  循环读 CSV + pandas  {t1:6.1f} ms')
    print(f'  方法 2  SQLite               {t2:6.1f} ms')
    print(f'  方法 3  DuckDB + Parquet     {t3:6.1f} ms')
    print(f'\n  SQLite 比 CSV 快 {t1 / t2:.1f}x')
    print(f'  DuckDB 比 CSV 快 {t1 / t3:.1f}x')
    print('\n注：当数据量达到百万行以上时，DuckDB 的优势通常会更加明显。')

性能对比（20 次均值，任务：10 只股票 × 2024 年均收益率）：
  方法 1  循环读 CSV + pandas    95.6 ms
  方法 2  SQLite                  4.7 ms
  方法 3  DuckDB + Parquet        4.3 ms

  SQLite 比 CSV 快 20.2x
  DuckDB 比 CSV 快 22.0x

注：当数据量达到百万行以上时，DuckDB 的优势通常会更加明显。

在约几万行的数据量下，三种方法的差距可能有限。DuckDB 的价值更多体现在更大规模的数据场景中。面对全市场多年日度数据时，循环读 CSV 可能需要较长时间，而 DuckDB 可以直接读取 Parquet 并完成分组聚合。

三种方法的适用场景

方法	适合	不适合
循环读 CSV + pandas	文件少、一次性分析	文件多、反复查询
SQLite	多表关联、中等数据量、教学	超大数据量
DuckDB + Parquet	大数据量分析查询	频繁小事务写入

进入券商、基金等机构后，可能会接触到 ClickHouse、Greenplum 等更专业的分析型数据库，使用逻辑与 DuckDB 相似：先用 SQL 整合和过滤数据，再接 pandas 做最后处理。本章的 SQL 基础在这些场景下同样适用。

AI 提示词：DuckDB 查询 Parquet

我有一个 Parquet 文件（data_raw/stock_daily/all_stocks.parquet），字段包括 code, date, close, pct_chg, volume, turnover。请帮我用 DuckDB 计算每只股票 2023-2024 年的年化收益率（日均收益率 × 252）和年化波动率（日收益率标准差 × √252），用 duckdb.query(...).df() 返回 DataFrame，按年化收益率降序排列。

12.10 数据管理规范

技术工具之外，数据项目能不能长期维护，很大程度上取决于一些非技术习惯：目录怎么组织、文件怎么命名、数据来源怎么记录。

本讲沿用当前项目结构，不在代码中移动文件。对学生而言，可以先理解每个目录在数据工作流中的位置：

project/
├── data_raw/              ← 原始数据，只写入、不手工修改
│   ├── stock_daily/
│   ├── index_daily/
│   ├── macro_monthly/
│   ├── fin_annual/
│   ├── company_info.csv
│   ├── finance.db         ← 由代码生成，可重新生成
│   └── download_log.txt
├── data/                  ← 处理后的数据和输出结果
│   ├── processed/
│   └── outputs/
├── figs/                  ← 本章插图
├── codes/                 ← 辅助脚本和数据获取代码
├── codes_get_data.ipynb
└── lecture_data_management.ipynb

最重要的一条是：data_raw/ 只写入、不手工修改。原始数据一旦下载，所有处理结果都应保存到其他位置，保证以后可以从原始数据重新开始。

图中最需要记住的是 data_raw/ 和 data/ 的分工：前者保留原始数据和可重新生成的数据库文件，后者保存处理后的中间数据和输出结果。

12.10.1 文件命名规范

一个好的文件名应尽量包含三类信息：数据主题 + 粒度 + 时间范围。

好的命名	差的命名
`stock_daily_firmdate_2020_2026.parquet`	`data1.csv`
`fin_annual_firmyear_2020_2025.csv`	`new_data_final.xlsx`
`macro_shibor_monthly_2020_2026.csv`	`利率数据最新版.csv`
`analysis_base_firmyear_v1.csv`	`分析底表（用这个）.csv`

文件名里带上粒度和时间范围，以后一眼就能判断这份数据覆盖到什么时候，不必打开文件再查。

# ── 生成数据字典 ─────────────────────────────────────
# 记录每张表的主键、粒度、字段、行数，方便后续查阅和交接。

datasets = [
    ('stock_daily',   '个股日度行情',   'code + date',       'firm-date'),
    ('index_daily',   '市场指数日度',   'index_code + date', 'index-date'),
    ('fin_annual',    '年度财务指标',   'code + year',       'firm-year'),
    ('company_info',  '公司基本信息',   'code',              'firm'),
    ('shibor_3m',     'Shibor 3 个月期', 'date',             'monthly'),
    ('usd_cny',       '人民币兑美元',   'date',              'monthly'),
    ('cpi_monthly',   'CPI 月度同比',   'date',              'monthly'),
]

lines = [f'# 数据字典\n\n生成时间：{pd.Timestamp.now().strftime("%Y-%m-%d %H:%M")}\n']

print(f'{"表名":<20} {"粒度":<12} {"主键":<20} {"行数":>6} 字段')
print('-' * 80)

for tbl, desc, pk, gran in datasets:
    try:
        df = pd.read_sql(f'SELECT * FROM {tbl} LIMIT 1', conn)
        cnt = pd.read_sql(f'SELECT COUNT(*) AS n FROM {tbl}', conn)['n'][0]
        cols = list(df.columns)
        print(f'{tbl:<20} {gran:<12} {pk:<20} {cnt:>6} {cols}')
        lines.append(
            f'\n## {tbl}\n'
            f'- 描述：{desc}\n'
            f'- 粒度：{gran}\n'
            f'- 主键：{pk}\n'
            f'- 行数：{cnt:,}\n'
            f'- 字段：{cols}\n'
        )
    except Exception as e:
        print(f'{tbl:<20} 读取失败：{e}')

dict_path = BASE / 'data_dictionary.md'
with open(dict_path, 'w', encoding='utf-8') as f:
    f.writelines(lines)

print(f'\n数据字典已保存至 {dict_path}')

表名                   粒度           主键                       行数 字段
--------------------------------------------------------------------------------
stock_daily          firm-date    code + date           15100 ['date', 'open', 'high', 'low', 'close', 'volume', 'amount', 'turnover', 'code', 'name', 'pct_chg']
index_daily          index-date   index_code + date      4533 ['date', 'open', 'high', 'low', 'close', 'volume', 'amount', 'pct_chg', 'code']
fin_annual           firm-year    code + year              60 ['code', 'name', 'year', 'roe', 'net_profit_margin', 'revenue_yoy', 'profit_yoy', 'debt_ratio', 'current_ratio', 'asset_turnover']
company_info         firm         code                     10 ['code', 'name']
shibor_3m            monthly      date                     75 ['date', 'shibor_3m']
usd_cny              monthly      date                     74 ['date', 'usd_cny']
cpi_monthly          monthly      date                     74 ['date', 'cpi_yoy']

数据字典已保存至 data_raw\data_dictionary.md

提示词：生成数据字典和 README 文档

可以使用以下提示词生成数据字典和 README 文档：

我有以下数据表：

stock_daily_firmdate_2020_2026.parquet：主键为 (code, date)，粒度为 firm-date，时间范围为 2020-2026，核心变量包括 close、pct_chg、volume、turnover，数据来源于公开行情接口。

fin_annual_firmyear_2020_2025.csv：主键为 (code, year)，粒度为 firm-year，核心变量包括 pe_ttm、pb，数据来源于财务指标接口。

macro_shibor_monthly_2020_2026.csv：主键为 date，粒度为 month，核心变量包括 shibor_3m，数据来源于宏观利率数据。

请帮我根据这些信息生成一个数据字典表格，并写一段简短的 README 文档，说明这个项目的目标、目录结构、数据处理流程，以及这些数据表的用途。

如果使用 AI Agent，也可以让它读取具体文件并自动生成数据字典。例如：

请读取 data_raw/finance.db 中的所有表，输出每张表的字段名、行数、可能的主键、粒度和时间范围，并生成一份 data_dictionary.md。

12.10.2 机构环境简介

本章介绍的工具链（SQLite + DuckDB + Parquet）适合个人分析和小团队项目。进入金融机构之后，会遇到更大规模的数据基础设施，但底层逻辑是相通的。

MySQL / PostgreSQL：服务器型关系数据库，支持多人并发访问。券商、基金的行情数据库和因子数据库常用这类工具。
数据仓库：大型机构把各业务线数据统一接入平台，分析师通过 SQL 接口取数。
ClickHouse / TimescaleDB：针对分析查询或时间序列数据优化，处理逐笔成交、行情快照等高频数据时可能会用到。

这些工具现在不需要系统掌握。先理解主键、粒度、SQL 查询和本地数据工作流，进入更复杂的数据环境后会更容易迁移。

12.11 一个可复用的数据管理工作流

到这里，本章已经完成了从多张源表到分析底表，再到存储和查询的完整流程。面对一个新的多来源数据项目，可以按以下顺序思考：

Step 1: 列出所有原始数据表，写清楚每张表的粒度和主键；
Step 2: 确定最终分析底表的目标粒度；
Step 3: 判断哪些高频表需要先聚合再合并；
Step 4: 根据数据量和查询复杂度，选择 CSV、Parquet、SQLite 或 DuckDB；
Step 5: 建库、写 SQL 查询，用 pd.read_sql() 接入 pandas；
Step 6: 把分析底表持久化保存，为后续清洗和建模做好准备。

下图把本章内容压缩成一个完整流程。一个可复用的数据项目，应从原始文件出发，经过主键和粒度检查、必要的聚合与清洗、合适的存储格式选择，再进入 SQL 查询和 pandas 分析。

12.12 本章小结

本章讨论的不是某个具体模型，而是数据分析的前置工作：数据多了之后怎么管好它们。核心结论可以归纳为四点。

1. 粒度和主键比语法更重要。 merge 之前先想清楚「每张表的每行代表什么」，再验证主键唯一性，能避免绝大多数数据合并错误。

2. 格式选择取决于使用场景。 CSV 用于分享和查看，Parquet 用于大表的高效存储和读取，SQLite 用于多表关联查询，DuckDB 用于大数据量的分析型查询。

3. SQL 是 pandas 的上游。 用 SQL 做数据整合和过滤，用 pandas 做最终计算和可视化，这是实际项目中很常见的工作模式。

4. 规范比工具更持久。 data_raw/ 只写不改，文件名包含粒度和时间范围，维护下载日志和数据字典。这些习惯不需要复杂工具，但能让项目在三个月后仍然可以被别人，以及未来的自己，理解和复用。

简言之，本章解决的是一个完整问题：从多张原始数据表出发，如何合并、保存、查询，并形成可复用的数据工作流。