用 DuckDB 搭建自动化股票分析报告：从数据拉到报告生成全流程
很多数据分析师和量化爱好者都有一个痛点：周末想放松，但周一早上的数据报表必须出。重复劳动不仅浪费时间，还容易因疲劳出错。
今天，我来带你用 DuckDB + Python 搭建一个全自动化的股票分析报告系统。整个过程完全自动化：周五收盘后自动拉取数据、DuckDB 本地分析、生成 Markdown 报告，然后通过 cron 定时任务触发。
你需要的全部基础设施就是一个本地 DuckDB 数据库 + 几个 Python 脚本。
一、整体架构设计
整个系统分为四个层次：
- 数据拉取层：用 yfinance 从 Yahoo Finance 获取股票数据，保存为 CSV
- 分析层：DuckDB 读取 CSV，执行 SQL 分析（涨幅排行、波动率、均线信号）
- 报告层：将分析结果组装成 Markdown 格式的报告
- 调度层：用 cron 实现定时自动运行
核心思路是：DuckDB 擅长处理 CSV 文件，而 yfinance 可以直接导出 CSV。两者结合，不需要部署任何数据库服务，纯本地运行，成本低到极致。
二、数据拉取层
环境准备
pip install yfinance duckdb pandas matplotlib
核心脚本：data_fetcher.py
import yfinance as yf
import duckdb
from datetime import datetime, timedelta
import os
TICKERS = ["AAPL", "MSFT", "GOOGL", "AMZN", "NVDA", "TSLA", "META"]
DATA_DIR = "stock_data"
os.makedirs(DATA_DIR, exist_ok=True)
def fetch_and_save():
end_date = datetime.now()
start_date = end_date - timedelta(days=400)
for ticker in TICKERS:
df = yf.download(ticker, start=start_date, end=end_date, progress=False)
if df.empty:
continue
csv_path = os.path.join(DATA_DIR, f"{ticker}.csv")
df.to_csv(csv_path)
print(f"✅ {ticker} 数据已保存: {csv_path}")
# 将 CSV 加载到 DuckDB
conn = duckdb.connect("stocks.db")
conn.execute("DROP TABLE IF EXISTS metadata")
tables = []
for ticker in TICKERS:
csv_path = os.path.join(DATA_DIR, f"{ticker}.csv")
if os.path.exists(csv_path):
conn.execute(f"CREATE TABLE {ticker} AS SELECT * FROM read_csv_auto('{csv_path}')")
tables.append(ticker)
conn.execute("CREATE TABLE metadata AS SELECT 'stocks' as source, current_date() as last_updated")
conn.commit()
conn.close()
print(f"🗄️ 数据库已更新，共 {len(tables)} 只股票")
if __name__ == "__main__":
fetch_and_save()
关键技术点：
- yfinance 下载的 DataFrame 列名是 MultiIndex 格式（如
('Close', '')），但read_csv_auto会自动处理，不需要手动 flatten - 先用
CREATE TABLE ... AS SELECT * FROM read_csv_auto()将 CSV 加载到 DuckDB 内存数据库中，后续分析直接查询即可 metadata表记录数据源和更新时间，方便后续判断数据是否新鲜
三、分析层：DuckDB SQL 核心
这是整个系统的核心。所有分析逻辑全部用 SQL 完成，逻辑清晰且可复用。
1. 近 30 日涨幅排行
import duckdb
conn = duckdb.connect("stocks.db")
top_gainers = conn.execute("""
SELECT ticker,
first(close) as close_30d_ago,
last(close) as close_current,
round((last(close) - first(close)) / first(close) * 100, 2) as pct_change
FROM (
SELECT ticker, date, close,
ROW_NUMBER() OVER (PARTITION BY ticker ORDER BY date DESC) as rn_desc,
ROW_NUMBER() OVER (PARTITION BY ticker ORDER BY date ASC) as rn_asc
FROM (
UNPIVOT stocks ON columns INCLUDE (date) INTO name, value
)
WHERE name = 'close'
) sub
WHERE rn_desc <= 1 OR rn_asc <= 1
GROUP BY ticker
ORDER BY pct_change DESC
LIMIT 5
""").fetchdf()
print("📈 近 30 日涨幅 TOP5:")
for _, row in top_gainers.iterrows():
print(f" {row.ticker}: {row.pct_change}%")
技巧说明：这里用到了 DuckDB 的 UNPIVOT 操作。yfinance 输出的 CSV 是多列格式（Open/High/Low/Close/Volume），UNPIVOT 将其转为长格式（ticker, date, name, value），方便后续做跨列分析。这是 DuckDB 处理金融时序数据的经典模式。
2. 波动率分析（风险维度）
volatility = conn.execute("""
SELECT ticker,
round(stddev(close), 2) as volatility,
round(avg(volume) / 1e6, 1) as avg_volume_m,
round(avg(close), 2) as avg_price
FROM (
SELECT ticker, date, close, volume
FROM (
UNPIVOT stocks ON columns INCLUDE (date) INTO name, value
)
WHERE name IN ('close', 'volume')
PIVOT_agg(avg) ON name
)
GROUP BY ticker
ORDER BY volatility DESC
LIMIT 5
""").fetchdf()
print("\n📊 波动率 TOP5（高风险高收益）:")
for _, row in volatility.iterrows():
print(f" {row.ticker}: 波动率={row.volatility}, 均价=${row.avg_price}")
3. 均线信号检测（金叉/死叉）
ma_signals = conn.execute("""
WITH daily AS (
SELECT ticker, date, close,
AVG(close) OVER (PARTITION BY ticker ORDER BY date ROWS BETWEEN 19 PRECEDING AND CURRENT ROW) as ma20,
AVG(close) OVER (PARTITION BY ticker ORDER BY date ROWS BETWEEN 49 PRECEDING AND CURRENT ROW) as ma50
FROM (
SELECT ticker, date, close
FROM (
UNPIVOT stocks ON columns INCLUDE (date) INTO name, value
)
WHERE name = 'close'
)
),
signals AS (
SELECT ticker, date, close, ma20, ma50,
LAG(ma20, 1) OVER (PARTITION BY ticker ORDER BY date) as ma20_prev,
LAG(ma50, 1) OVER (PARTITION BY ticker ORDER BY date) as ma50_prev
FROM daily
)
SELECT ticker, date,
CASE
WHEN ma20 > ma50 AND ma20_prev <= ma50_prev THEN '🟢 金叉'
WHEN ma20 < ma50 AND ma20_prev >= ma50_prev THEN '🔴 死叉'
ELSE '⚪ 无信号'
END as signal,
round(close, 2) as price
FROM signals
WHERE date >= CURRENT_DATE - INTERVAL '5' DAY
ORDER BY date DESC, ticker
""").fetchdf()
print("\n📉 最新均线信号:")
for _, row in ma_signals.iterrows():
print(f" {row.ticker} ({row.date}): {row.signal} @ ${row.price}")
conn.close()
进阶技巧：这里组合使用了 DuckDB 的 窗口函数（AVG OVER + LAG）和 CTE（公用表表达式）。LAG 函数用来获取前一天的均线值，从而判断是否发生金叉或死叉。整个逻辑完全用 SQL 表达，无需 Python 循环，执行效率极高。
四、报告生成层
分析完成后，需要将结果组装成易读的 Markdown 报告。
report_generator.py
import duckdb
from datetime import datetime
conn = duckdb.connect("stocks.db")
report_date = datetime.now().strftime("%Y-%m-%d")
week_number = datetime.now().isocalendar()[1]
# 近 30 日涨幅榜
gainers = conn.execute("""
SELECT ticker,
round(last(close) - first(close), 2) as change,
round((last(close) / first(close) - 1) * 100, 2) as pct
FROM (
SELECT ticker, date, close
FROM (UNPIVOT stocks ON columns INCLUDE (date) INTO name, value)
WHERE name = 'close'
)
WHERE date >= CURRENT_DATE - INTERVAL '30' DAY
GROUP BY ticker
ORDER BY pct DESC
LIMIT 5
""").fetchall()
conn.close()
# 组装报告
lines = [
f"# 📊 股票周报 Week {week_number} ({report_date})",
"",
"## 🏆 近 30 日涨幅榜",
"",
]
for i, row in enumerate(gainers, 1):
lines.append(f"{i}. **{row[0]}** —— 涨跌幅: {row[2]}% (绝对值: ${row[1]})")
lines += [
"",
"## 💡 本周策略提示",
"",
"> 以上内容仅供参考，不构成投资建议。投资有风险，入市需谨慎。",
"",
f"*报告由 DuckDB + yfinance 自动生成 | {report_date}*",
]
report = "\n".join(lines)
# 写入文件
output_path = f"report_week{week_number}.md"
with open(output_path, "w") as f:
f.write(report)
print(f"✅ 报告已生成: {output_path}")
五、自动化调度
用 cron 实现每周五收盘后自动运行：
# 编辑 cron 任务
crontab -e
# 添加以下内容（美国东部时间周五 16:00，即北京时间周六 4:00）
# 时区转换为 UTC
0 20 * * 5 cd /path/to/your/project && /usr/bin/python3 data_fetcher.py && /usr/bin/python3 analyzer.py && /usr/bin/python3 report_generator.py
如果你希望用更灵活的方式调度，可以封装成一个统一的 run.py：
#!/usr/bin/env python3
"""统一入口：拉取数据 → 分析 → 生成报告"""
import subprocess
import sys
steps = [
("拉取数据", "data_fetcher.py"),
("执行分析", "analyzer.py"),
("生成报告", "report_generator.py"),
]
for name, script in steps:
print(f"\n{'='*40}")
print(f"▶ {name}: {script}")
print('='*40)
result = subprocess.run([sys.executable, script], capture_output=True, text=True)
print(result.stdout)
if result.returncode != 0:
print(f"❌ {name} 失败: {result.stderr}")
sys.exit(1)
print("\n🎉 全部流程执行完毕！")
然后在 crontab 中简化为一行：
0 20 * * 5 cd /path/to/your/project && /usr/bin/python3 run.py
六、进阶：邮件 / Telegram 推送
报告生成后，可以自动发送到你的邮箱或 Telegram：
邮件推送
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
def send_report(email_to, report_path, smtp_server, smtp_port, smtp_user, smtp_pass):
with open(report_path, "r") as f:
body = f.read()
msg = MIMEMultipart()
msg['From'] = smtp_user
msg['To'] = email_to
msg['Subject'] = f"📊 股票周报 - {datetime.now().strftime('%Y-%m-%d')}"
msg.attach(MIMEText(body, 'plain', 'utf-8'))
server = smtplib.SMTP(smtp_server, smtp_port)
server.starttls()
server.login(smtp_user, smtp_pass)
server.send_message(msg)
server.quit()
print("📧 邮件已发送")
Telegram 推送
import requests
def send_to_telegram(chat_id, bot_token, message):
url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
requests.post(url, json={
"chat_id": chat_id,
"text": message,
"parse_mode": "Markdown"
})
print("📱 Telegram 消息已发送")
七、与传统方案对比
| 维度 | 传统方案（手动操作） | DuckDB 自动化方案 |
|---|---|---|
| 数据获取 | 手动登录交易平台/网站下载 | yfinance 自动拉取 |
| 数据处理 | Excel 手动计算指标 | DuckDB SQL 一行搞定 |
| 分析逻辑 | Python 循环 + Pandas | 纯 SQL，可复用可审计 |
| 报告生成 | 手动整理粘贴 | 脚本自动生成 Markdown |
| 定时执行 | 手动触发 | cron 全自动 |
| 出错率 | 高（人为疲劳） | 低（脚本执行一致） |
| 维护成本 | 每次都要重做 | 一次配置，长期受益 |
八、变现建议
这套系统的价值不仅在于节省时间，更在于可以产品化：
付费订阅服务：将周报打包成付费 Telegram 频道或 Newsletter，每月收费 $5-20。DuckDB 的本地分析能力意味着零服务器成本，利润率极高。
数据产品后端：很多小型 SaaS 产品需要数据处理能力，但不想维护数据库。你可以用 DuckDB 作为后端引擎，提供"数据分析即服务"，按查询量收费。
自动化报表服务：面向中小企业提供自动化周报/月报服务。用 DuckDB 处理他们的 CSV/Excel 数据，自动生成报告。单个客户月费 $100-500，边际成本趋近于零。
量化策略回测工具：将这套系统扩展为策略回测平台，用户提交策略参数，DuckDB 自动计算收益率、最大回撤等指标，按次收费。
核心思路：DuckDB 让你用极低的成本（本地运行、零服务器费用）提供原本需要昂贵基础设施才能完成的数据服务。这就是你的竞争壁垒。
📖 本文的完整可运行代码（包含邮件/Telegram 推送模块）已发布在 duckdblab.org，包含更详细的部署步骤和错误排查指南。