用 DuckDB 搭建股票周报自动生成器：10行代码到付费订阅产品
昨晚我用 DuckDB + Python 搭建了一个股票周报自动生成器，现在每天自动发邮件给客户。这套系统从数据采集到产品交付只需要不到50行核心代码，但背后是一套完整的变现闭环。
为什么选择 DuckDB？
在做这个产品之前，我用的是 Pandas + SQLite 的组合。处理几百兆的 CSV 文件需要30多秒，周末还经常忘记定时执行，导致客户投诉。
换成 DuckDB 之后，同样的逻辑，耗时降到2秒以内。更重要的是，DuckDB 可以直接查询 CSV 文件而无需全部加载到内存，利用列式存储让聚合查询快5-10倍。
对于任何需要快速处理大量数据并生成报告的场景，DuckDB 都是比传统方案更好的选择。
完整实现代码
第一步：安装依赖
pip install duckdb pandas python-dotenv yfinance
第二步：核心查询（只用了8行）
import duckdb
import pandas as pd
from dotenv import load_dotenv
import os
load_dotenv()
con = duckdb.connect()
query = """
SELECT
ticker,
date,
close,
round(avg(close) over (partition by ticker order by date rows between 20 preceding and current row), 2) as ma20,
round(stddev(close) over (partition by ticker order by date rows between 20 preceding and current row), 2) as vol20
FROM read_csv_auto('market_data_*.csv')
WHERE date >= '2025-08-01'
"""
df = con.execute(query).fetchdf()
这段代码的核心是 SQL 窗口函数：
avg(close) over (... rows between 20 preceding and current row)计算20日均线stddev(close) over (...)计算20日波动率read_csv_auto('market_data_*.csv')自动合并所有匹配的 CSV 文件并推断类型
第三步：生成交易信号
signals = df.groupby('ticker').apply(lambda g: {
'ticker': g['ticker'].iloc[-1],
'signal': 'BUY' if g['close'].iloc[-1] > g['ma20'].iloc[-1] * 1.02 else 'HOLD',
'score': round((g['close'].iloc[-1] - g['ma20'].iloc[-1]) / g['vol20'].iloc[-1], 2) if g['vol20'].iloc[-1] > 0 else 0
})
这里用 groupby + lambda 为每只股票生成买入/持有信号，并计算一个标准化分数。当收盘价突破20日均线2%以上且波动率较低时，标记为买入信号。
第四步：发送邮件
import smtplib
from email.mime.text import MIMEText
def send_weekly_report(signals_df):
html = signals_df.to_html(index=False, classes='signals')
msg = MIMEText(html, 'html')
msg['Subject'] = f'📊 本周数据简报 {pd.Timestamp.now().strftime("%m月%d日")}'
msg['From'] = os.getenv('EMAIL_USER')
msg['To'] = ','.join(os.getenv('SUBSCRIBERS').split(','))
with smtplib.SMTP_SSL('smtp.gmail.com', 465) as s:
s.login(os.getenv('EMAIL_USER'), os.getenv('EMAIL_PASS'))
s.send_message(msg)
send_weekly_report(signals)
邮件内容直接用 Pandas DataFrame 的 to_html() 生成 HTML 表格，简单高效。
关键技术优化点
1. 直接查询 CSV，避免内存爆炸
# 比 read_csv + groupby 快 8 倍
result = con.execute("""
SELECT category, sum(amount) as total
FROM big_table.csv
GROUP BY category
""").fetchdf()
DuckDB 的列式存储和向量化执行，让它在处理大文件时比 Pandas 快得多，而且不会 OOM。
2. read_csv_auto 省去类型猜测的麻烦
# 自动推断 schema，无需手动定义列类型
df = con.read_csv_auto('data/*.csv')
glob 模式支持批量读取，对于按日期分割的 CSV 文件特别友好。
3. 用 UDF 做自定义计算
con.register('my_data', df)
con.execute("""
SELECT
my_func(close, ma20) as signal_score
FROM my_data
""")
复杂逻辑可以用 SQL UDF 封装，保持代码整洁。
DuckDB vs 传统方案对比
| 维度 | Pandas + SQLite | DuckDB |
|---|---|---|
| 内存占用 | 全量加载，易OOM | 列式扫描，按需读取 |
| CSV 查询速度 | 基准 | 快 5-10 倍 |
| 多文件合并 | 手动 concat | glob + auto merge |
| SQL 窗口函数 | 支持但慢 | 原生优化 |
| 零配置 | 需创建连接 | 单文件数据库 |
| 适用场景 | 中小数据 | 海量数据实时分析 |
变现思路拆解
这个项目已经跑通了以下闭环：
- 数据源：用 yfinance 免费拉取股票数据
- 处理：DuckDB 做所有清洗和计算
- 产品：HTML 周报邮件
- 分发：定时任务每天凌晨自动发送
- 变现：付费订阅（99元/月），目前 127 个付费用户
关键是把"数据"变成"可执行的洞察"——这才是付费用户愿意买单的核心。他们买的不是数据，而是节省的时间和明确的交易信号。
进阶：封装成可复用模板
把整个流程封装成通用组件，任何人都能快速搭建自己的数据产品：
# 封装后的调用方式
from duckdb_product import DataProduct
dp = DataProduct(
data_src='*.csv',
query_template='weekly_signals.sql',
output_format='html_email',
schedule='cron 0 8 * * 1-5'
)
dp.run()
总结
用 DuckDB 搭建自动化数据产品的核心在于：
- SQL 优先：能用 SQL 解决的问题不用 Python 循环
- 零配置：单文件数据库，部署简单
- 高性能：列式存储 + 向量化，大数据也不慌
- 可扩展：从 CSV 到 Parquet 到 Delta Lake，无缝切换
如果你也有想法想把数据变成产品，但卡在数据处理环节，可以来 duckdblab.org 看看我的完整系列教程，里面有从数据采集到变现的全链路案例，都是可以直接复用的代码模板。
学习更多 DuckDB 实战经验 → duckdblab.org