DuckDB 投研简报自动化：从多源数据到每日投资报告的全流程实战
难度：⭐⭐⭐｜预计耗时：2 小时搭建，之后每天 30 秒生成专业投研简报
一、为什么投研简报系统是值钱的 Data Product？
在金融圈，投研简报是最高频的需求之一——基金经理、理财顾问、量化团队每天都要看市场概况和选股信号。
传统做法：手动打开十几张 Excel，复制粘贴数据，写 VLOOKUP，拼出一个报告。耗时 2-3 小时，而且每天重复同样的劳动。
这个流程的变现价值非常大：
- 你开发的简报系统可以卖给理财顾问（他们愿意为每日市场简报付费）
- 封装成 API 服务给中小私募做后台数据支撑
- 做成付费 Newsletter，订阅制月入数千
- 自己用，省下的时间做更深的数据挖掘
今天我们用 DuckDB 搭建一套全自动投研简报系统，从多源数据读取到报告生成，全流程只需一条命令。
二、系统架构：三层数据，一条 SQL 搞定
我们的投研简报依赖三类数据源：
- 行情数据 —
stock_prices.csv：每日股票收盘价 - 财务数据 —
financials.csv：PE、ROE、营收增长率等 - 宏观数据 —
macro.csv：CPI、PMI、利率等
传统做法需要先把三个 CSV 分别读入 Pandas，做清洗，再 merge。DuckDB 的做法是直接原生读取，一个查询搞定所有计算。
import duckdb
import pandas as pd
from pathlib import Path
# 一个内存数据库，直接读取所有 CSV
conn = duckdb.ATTACH (':memory:')
# read_csv_auto 自动推断日期、数值类型，无需手动处理格式
conn.execute("CREATE TABLE stocks AS SELECT * FROM read_csv_auto('data/stock_prices.csv')")
conn.execute("CREATE TABLE financials AS SELECT * FROM read_csv_auto('data/financials.csv')")
conn.execute("CREATE TABLE macro AS SELECT * FROM read_csv_auto('data/macro.csv')")
print(f"✅ stocks: {conn.execute('SELECT COUNT(*) FROM stocks').fetchone()[0]} 行")
print(f"✅ financials: {conn.execute('SELECT COUNT(*) FROM financials').fetchone()[0]} 行")
print(f"✅ macro: {conn.execute('SELECT COUNT(*) FROM macro').fetchone()[0]} 行")
关键洞察：
read_csv_auto会自动处理日期格式、推断数值类型、跳过空行。对于从 Excel 导出的 CSV（常含混合类型列），它也比 Pandas 的read_csv更鲁棒。省掉的数据清洗时间，就是实打实的收入。
三、核心分析逻辑：三条 CTE，表达完整投研框架
投研简报的核心逻辑可以拆解为三个模块：
模块 1：估值筛选
找出 PE < 行业均值 且 ROE > 15% 的低估值优质标的：
WITH pe_rank AS (SELECT
SELECT
f.symbol,
f.name,
f.pe_ttm,
f.roe,
f.revenue_yoy,
f.sector,
-- 行业 PE 平均值（窗口函数，无需 GROUP BY）
AVG(f.pe_ttm) OVER (PARTITION BY f.sector) AS sector_pe_avg,
-- 行业 PE 排名
RANK() OVER (PARTITION BY f.sector ORDER BY f.pe_ttm) AS pe_rank_in_sector
FROM financials f
WHERE f.pe_ttm > 0 AND f.roe > 0
)
模块 2：动量确认
计算最新日期的 20 日价格动量：
momentum AS (
SELECT
symbol,
((close - LAG(close, 20) OVER (PARTITION BY symbol ORDER BY date))
/ LAG(close, 20) OVER (PARTITION BY symbol ORDER BY date)) * 100 AS momentum_20d
FROM stocks
WHERE date = (SELECT MAX(date) FROM stocks)
)
模块 3：宏观风险判断
基于 PMI 和利率变化判断当前市场 regime：
macro_filter AS (
SELECT
date,
cpi,
pmi,
rate,
CASE
WHEN pmi > 50 AND rate <= LAG(rate) OVER (ORDER BY date) THEN 'risk_on'
WHEN pmi < 49 AND rate >= LAG(rate) OVER (ORDER BY date) THEN 'risk_off'
ELSE 'neutral'
END AS market_regime
FROM macro
ORDER BY date DESC
LIMIT 1
)
组合查询
把三个模块组合起来，完成最终筛选：
SELECT
p.symbol,
p.name,
p.sector,
ROUND(p.pe_ttm, 2) AS pe,
ROUND(p.sector_pe_avg, 2) AS sector_pe_avg,
ROUND(p.roe, 2) AS roe_pct,
ROUND(m.momentum_20d, 2) AS momentum_20d,
ROUND(r.cpi, 3) AS cpi,
r.pmi,
r.market_regime
FROM pe_rank p
LEFT JOIN momentum m ON p.symbol = m.symbol
CROSS JOIN macro_filter r
WHERE p.pe_ttm < p.sector_pe_avg
AND p.roe >= 15
AND p.pe_rank_in_sector <= 5
ORDER BY p.pe_ttm ASC
LIMIT 20
性能对比：同样的逻辑用 Pandas 实现需要写 50+ 行代码，而且遇到大文件时会内存溢出。DuckDB 用一条 SQL 完成所有计算，处理百万行数据只需秒级。
四、报告生成：Jinja2 模板输出 Telegram 可读格式
分析结果需要变成可直接发送的简报格式。我们用 Jinja2 模板，输出适合 Telegram/邮件阅读的文本：
from jinja2 import Template
from datetime import datetime
template_text = """