用 DuckDB 搭建自动化投资研报系统：从数据采集到报告推送全流程

手把手教你用 DuckDB 搭建全自动投资研报系统，从数据采集、指标计算到报告推送，全程无需数据库服务器，一个 Python 脚本搞定。

用 DuckDB 搭建自动化投资研报系统

很多数据分析师每天花数小时重复做研报，但真正能变现的是那些「可产品化」的分析。今天带你用 DuckDB 搭一个自动化投资研报系统——从数据采集到报告生成全自动，你可以拿这个架构去做付费研报服务。

架构图

为什么选 DuckDB？

传统研报流程是这样的：Python 爬数据 → pandas 清洗 → Excel 排版 → 手动发送。一天下来效率极低，且容易出错。

DuckDB 的核心优势在这里体现得淋漓尽致：

内嵌数据库，零部署成本，单文件就能跑
SQL 直接分析 CSV/Parquet/JSON，不需要 ETL 管道
Python/R/Node.js 多语言绑定，无缝嵌入现有工作流
向量化执行，百万行数据秒级查询

我们今天要做的系统，全程不需要数据库服务器，一个 Python 脚本搞定。

项目架构

整个系统分三步：

数据层：用 DuckDB 读取市场数据（CSV/Parquet）
分析层：用 SQL 计算关键指标（PE、ROE、动量等）
输出层：生成结构化研报，推送给读者

先看数据准备。假设你有一份股票历史数据文件 stocks.csv，字段包括日期、代码、收盘价、成交量、公司基本面数据。

第一步：建立数据管道

import duckdb
import pandas as pd
from datetime import datetime, timedelta

# 连接内存数据库，零配置
con = duckdb.connect(':memory:')

# 注册 CSV 文件为虚拟表，无需加载到内存
con.execute("""
    CREATE TABLE stocks AS
    SELECT * FROM read_csv_auto('stocks.csv')
""")

# 同样可以读取多个文件并合并
con.execute("""
    CREATE TABLE fundamentals AS
    SELECT * FROM read_csv_auto('fundamentals/*.csv')
""")

这里的关键点是 read_csv_auto 会自动推断列类型，而且它只在不涉及数据时才真正读取——DuckDB 使用谓词下推优化，只读需要的列和行。

第二步：编写研报分析 SQL

真正的核心价值在于分析逻辑。我们用 SQL 一次性完成所有指标计算：

-- 创建分析视图：计算 PE 分位数、动量、ROE 趋势
CREATE OR REPLACE VIEW daily_analysis AS
WITH price_changes AS (
    SELECT 
        date,
        code,
        close,
        close / LAG(close, 20) OVER (PARTITION BY code ORDER BY date) - 1 AS momentum_20d,
        close / LAG(close, 60) OVER (PARTITION BY code ORDER BY date) - 1 AS momentum_60d,
        close / LAG(close, 120) OVER (PARTITION BY code ORDER BY date) - 1 AS momentum_120d
    FROM stocks
),
ranked_pe AS (
    SELECT 
        code,
        date,
        pe_ratio,
        PERCENT_RANK() OVER (PARTITION BY code ORDER BY pe_ratio) AS pe_percentile,
        AVG(pe_ratio) OVER (
            PARTITION BY code 
            ORDER BY date 
            ROWS BETWEEN 250 PRECEDING AND CURRENT ROW
        ) AS pe_ma5y
    FROM fundamentals
),
combined AS (
    SELECT 
        p.date, p.code, p.close,
        p.momentum_20d, p.momentum_60d,
        r.pe_ratio, r.pe_percentile, r.pe_ma5y,
        f.roe, f.revenue_growth
    FROM price_changes p
    JOIN ranked_pe r ON p.code = r.code AND p.date = r.date
    JOIN fundamentals f ON p.code = f.code AND p.date = f.date
)
SELECT * FROM combined;

注意看这里用了哪些 DuckDB 原生能力：

窗口函数 LAG / PERCENT_RANK / AVG OVER：直接算动量和 PE 分位数，不需要 groupby + join 的复杂写法
时间范围聚合：ROWS BETWEEN 250 PRECEDING AND CURRENT ROW 自然表达"近5年"的概念
视图物化：CREATE OR REPLACE VIEW 让后续查询可以直接引用，逻辑分层清晰

第三步：筛选策略标的

有了分析视图，接下来是核心策略逻辑：

-- 生成每日推荐列表
CREATE OR REPLACE VIEW daily_picks AS
SELECT 
    date, code, name, close,
    ROUND(pe_ratio, 2) AS pe,
    ROUND(pe_percentile * 100, 1) AS pe_percentile,
    ROUND(momentum_60d * 100, 2) AS momentum_60d_pct,
    ROUND(roe * 100, 2) AS roe_pct,
    ROUND(revenue_growth * 100, 2) AS rev_growth_pct,
    CASE 
        WHEN pe_percentile < 0.3 AND roe > 0.15 
             AND momentum_60d > 0 AND revenue_growth > 0.1 
        THEN 'BUY'
        WHEN pe_percentile > 0.7 AND momentum_60d < -0.1
        THEN 'SELL'
        ELSE 'HOLD'
    END AS signal
FROM daily_analysis
WHERE date = (SELECT MAX(date) FROM daily_analysis)
ORDER BY pe_percentile ASC;

这个查询的输出就是每日研报的核心表格——按 PE 分位数排序，低 PE + 高 ROE + 正动量 + 高增长的股票被标记为 BUY。

第四步：生成研报并推送

最后一步是把 SQL 结果变成可读的报告：

# 获取今日推荐
today = datetime.now().strftime('%Y-%m-%d')
picks = con.execute("""
    SELECT code, name, close, pe, pe_percentile, 
           momentum_60d_pct, roe_pct, signal
    FROM daily_picks
""").fetchdf()

# 生成研报文本
report = f"""
📊 投资研报 {today}
{'='*20}

🔥 今日重点关注

{chr(10).join([
    f"• {row['name']}({row['code']}) | 现价:{row['close']:.2f} | PE:{row['pe']}({row['pe_percentile']}%分位) | ROE:{row['roe_pct']}% | 60日动量:{row['momentum_60d_pct']}%"
    for _, row in picks[picks['signal'] == 'BUY'].iterrows()
])}

📈 市场概况
总标的数: {len(picks)}
买入信号: {len(picks[picks['signal']=='BUY'])}
卖出信号: {len(picks[picks['signal']=='SELL'])}

⚠️ 风险提示: 以上分析仅供参考，不构成投资建议。
"""

# 发送到 Telegram
import requests
bot_token = "YOUR_BOT_TOKEN"
chat_id = "YOUR_CHAT_ID"
requests.post(
    f"https://api.telegram.org/bot{bot_token}/sendMessage",
    json={"chat_id": chat_id, "text": report, "parse_mode": "HTML"}
)

第五步：定时自动化

用 cron 或者 Python 的 schedule 库实现每晚自动执行：

import schedule
import time

def daily_report():
    con = duckdb.connect(':memory:')
    # ... 上面的完整逻辑 ...
    print(f"[{datetime.now()}] 研报已推送")

schedule.every().day.at("22:00").do(daily_report)

while True:
    schedule.run_pending()
    time.sleep(60)

与传统方案对比

维度	传统方案	DuckDB 方案
部署成本	需要 MySQL/PostgreSQL 服务器	零部署，内存数据库
数据准备	ETL 管道 + 定时抽取	直接读 CSV/Parquet
查询性能	慢，依赖索引和分区	向量化执行，谓词下推
开发效率	多语言拼接（Python + SQL）	纯 SQL 完成全部逻辑
维护成本	高，需监控数据库状态	低，单文件运行

关键优化技巧

如果你处理的是几百万行的数据，下面几个技巧能让性能提升 10 倍以上：

1. 使用 Parquet 而非 CSV

con.execute("COPY stocks TO 'stocks.parquet' (FORMAT PARQUET)")
# 后续查询速度提升 5-10x
con.execute("SELECT * FROM 'stocks.parquet' WHERE date > '2024-01-01'")

2. 开启内存限制，防止 OOM

con.execute("SET memory_limit='4GB'")
con.execute("SET threads TO 4")

3. 利用分区裁剪

# 只读需要的日期范围，DuckDB 会自动下推谓词
con.execute("SELECT * FROM 'parquet_files/*.parquet' WHERE date > '2024-06-01'")

4. 物化中间结果

con.execute("CREATE TABLE analysis_cache AS SELECT ...")
# 下次直接读缓存，避免重复计算
con.execute("SELECT * FROM analysis_cache WHERE date = '2024-09-19'")

变现思路

这套系统本身就可以变成产品：

付费研报订阅：每周/每月付费获取推荐列表，定价 99-299 元/月
SaaS 化：把系统封装成 API，让其他分析师接入，按调用次数收费
数据产品：把清洗好的 Parquet 数据集卖给量化团队，一份数据反复售卖
培训变现：教别人搭建类似的系统，开设 DuckDB 实战课程

DuckDB 的价值在于：你不需要维护任何数据库服务器，本地跑起来就能用，部署成本趋近于零。这意味着你可以把更多精力放在分析逻辑和变现上，而不是基础设施。

想深入了解 DuckDB 在金融数据场景的完整应用？duckdblab.org 上有从数据采集到报告推送的全套教程，包含真实股票数据文件和可运行的代码模板，帮你快速搭建自己的投研系统。学习更多 DuckDB 实战经验 → duckdblab.org

📺 Watch video tutorials → Olap Studio YouTube

Subscribe for more DuckDB & AI automation tutorials