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.connect(':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
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 = """
📊 投研数据简报 | {{ date }}
━━━━━━━━━━━━━━━━━━
【宏观环境】
CPI: {{ cpi }} | PMI: {{ pmi }} | 状态: {{ regime }}
━━━━━━━━━━━━━━━━━━
【低估值高 ROE 标的】（PE < 行业均值 & ROE ≥ 15%）
{% for row in stocks %}
{{ loop.index }}. {{ row.name }}（{{ row.symbol }}）
行业: {{ row.sector }} | PE: {{ row.pe }}（行业均: {{ row.sector_pe_avg }}）
ROE: {{ row.roe_pct }}% | 20日动量: {{ row.momentum_20d }}%
{% endfor %}
━━━━━━━━━━━━━━━━━━
⚠️ 数据截至 {{ date }}，仅供参考，不构成投资建议。
Generated by DuckDB Auto-Report
"""
template = Template(template_text)
report = template.render(
date=datetime.now().strftime('%Y-%m-%d'),
cpi=float(screener['cpi'].iloc[0]) if len(screener) > 0 else None,
pmi=float(screener['pmi'].iloc[0]) if len(screener) > 0 else None,
regime=str(screener['market_regime'].iloc[0]) if len(screener) > 0 else 'unknown',
stocks=[
{
'name': r['name'],
'symbol': r['symbol'],
'sector': r['sector'],
'pe': r['pe'],
'sector_pe_avg': r['sector_pe_avg'],
'roe_pct': r['roe_pct'],
'momentum_20d': r['momentum_20d'] if pd.notna(r['momentum_20d']) else '--',
}
for _, r in screener.iterrows()
]
)
print(report)
五、完整自动化脚本（cron 定时执行）
把整个流程封装成入口脚本：
# auto_report.py —— 完整入口
import duckdb
from jinja2 import Template
from datetime import datetime
import smtplib
from email.mime.text import MIMEText
import os
EMAIL_CONFIG = {
'smtp_host': os.environ.get('SMTP_HOST'),
'smtp_port': int(os.environ.get('SMTP_PORT', '587')),
'sender': os.environ.get('EMAIL_USER'),
'password': os.environ.get('EMAIL_PASS'),
'recipients': os.environ.get('EMAIL_RECIPIENTS', '').split(','),
}
def generate_report() -> str:
conn = duckdb.connect(':memory:')
# 读取数据
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')")
# 执行分析查询
query = """
WITH pe_rank AS (
SELECT symbol, name, pe_ttm, roe, revenue_yoy, sector,
AVG(pe_ttm) OVER (PARTITION BY sector) AS sector_pe_avg,
RANK() OVER (PARTITION BY sector ORDER BY pe_ttm) AS pe_rank_in_sector
FROM financials WHERE pe_ttm > 0 AND roe > 0
),
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)
),
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
"""
screener = conn.execute(query).fetchdf()
conn.close()
# 渲染报告
template = Template("""
📊 投研数据简报 | {{ date }}
━━━━━━━━━━━━━━━━━━
【宏观环境】CPI: {{ cpi }} | PMI: {{ pmi }} | 状态: {{ regime }}
━━━━━━━━━━━━━━━━━━
【低估值高 ROE 标的】（PE < 行业均值 & ROE ≥ 15%）
{% for row in stocks %}
{{ loop.index }}. {{ row.name }}（{{ row.symbol }}）
行业: {{ row.sector }} | PE: {{ row.pe }}（行业均: {{ row.sector_pe_avg }}）
ROE: {{ row.roe_pct }}% | 20日动量: {{ row.momentum_20d }}%
{% endfor %}
━━━━━━━━━━━━━━━━━━
⚠️ 数据截至 {{ date }}，仅供参考，不构成投资建议。
Generated by DuckDB Auto-Report
""")
return template.render(
date=datetime.now().strftime('%Y-%m-%d'),
cpi=float(screener['cpi'].iloc[0]) if len(screener) > 0 else None,
pmi=float(screener['pmi'].iloc[0]) if len(screener) > 0 else None,
regime=str(screener['market_regime'].iloc[0]) if len(screener) > 0 else 'unknown',
stocks=[
{'name': r['name'], 'symbol': r['symbol'], 'sector': r['sector'],
'pe': r['pe'], 'sector_pe_avg': r['sector_pe_avg'],
'roe_pct': r['roe_pct'],
'momentum_20d': r['momentum_20d'] if pd.notna(r['momentum_20d']) else '--'}
for _, r in screener.iterrows()
]
)
def send_report(report: str):
msg = MIMEText(report, 'plain', 'utf-8')
msg['Subject'] = f'投研数据简报 | {datetime.now().strftime("%Y-%m-%d")}'
msg['From'] = EMAIL_CONFIG['sender']
msg['To'] = ', '.join(EMAIL_CONFIG['recipients'])
with smtplib.SMTP(EMAIL_CONFIG['smtp_host'], EMAIL_CONFIG['smtp_port']) as s:
s.starttls()
s.login(EMAIL_CONFIG['sender'], EMAIL_CONFIG['password'])
s.sendmail(EMAIL_CONFIG['sender'], EMAIL_CONFIG['recipients'], msg.as_string())
if __name__ == '__main__':
report = generate_report()
send_report(report)
print('✅ 报告已发送')
Cron 定时配置
# 每个交易日 16:00 执行（港股/ A 股收盘后）
0 16 * * 1-5 cd /opt/report && python3 auto_report.py
六、与传统方案的性能对比
| 维度 | 传统 Pandas 方案 | DuckDB 方案 |
|---|---|---|
| 代码量 | 80-100 行（数据读取 + 清洗 + 合并 + 计算） | 30 行（一条 SQL + Jinja2 模板） |
| 内存占用 | 多 DataFrame 并行，1GB+ 数据易 OOM | 列式存储，1GB 数据 ~200MB |
| 日期解析 | 需手动 parse_dates 处理混合格式 | read_csv_auto 自动推断 |
| 窗口函数 | 需 groupby + transform 手动实现 | 原生 SQL 窗口函数，一行搞定 |
| 执行速度 | 百万行聚合 ~5-10 秒 | 百万行聚合 <1 秒 |
| 部署复杂度 | 需维护 Python 环境 + 依赖 | 单文件脚本，Python + DuckDB 即可 |
七、变现路径：这套系统能赚多少钱？
路径 1：数据服务订阅（月入 2000-8000 元）
把日报发给付费订阅者（理财顾问、独立投资者），每人每月收费 99-299 元。100 个订阅者就是 1-3 万/月。
路径 2：API 化服务（月入 5000-20000 元）
封装成 REST API，给中小私募、投顾团队提供每日选股信号推送。按调用量收费，API 化后边际成本趋近于零。
路径 3：SaaS 数据产品（年入 10 万+）
加上 Web 界面和用户管理，做成 SaaS 产品。定价 999 元/年/用户，100 个客户就是 10 万/年。
路径 4：外包接单（单次 2000-10000 元）
很多中小机构有类似需求但不会搭建，你可以接这类外包项目。从数据源对接到报告生成，一套系统收 5000-20000 元很常见。
八、避坑指南：常见踩坑点
- 数据源路径问题：cron 执行时工作目录变化，务必用绝对路径或
os.chdir()确保脚本能找到 CSV 文件。 - 日期格式混杂：
read_csv_auto虽然鲁棒，但如果某列同时包含 “2026-01-01” 和 “2026/01/01” 两种格式，需要先统一格式。 - 时区问题：宏观数据（PMI、CPI）通常是月度或季度发布，要注意最新数据的日期对齐，避免用旧数据做判断。
- 空值处理：动量计算中
LAG(close, 20)在数据不足 20 天时会返回 NULL，记得在 SQL 中做过滤。 - 邮件发送失败：SMTP 密码变更后需同步更新环境变量，建议用
.env文件管理敏感配置。
总结
用 DuckDB 搭建投研简报系统的核心思路是：数据直读 + SQL 表达 + 模板渲染。三步走完，一个原本需要 2 小时的手工流程被压缩到 30 秒。这套系统的真正价值不在于技术本身，而在于它把重复劳动变成了可复制、可售卖的数据产品。
学习更多 DuckDB 实战经验 → duckdblab.org