用 DuckDB 搭建股票市场分析流水线：从数据导入到自动化报告

import duckdb

# 连接（文件模式：数据持久化）
conn = duckdb.connect('portfolio.db')

# 创建表结构
conn.execute("""
CREATE TABLE IF NOT EXISTS stock_prices (
    date DATE,
    ticker VARCHAR,
    open DOUBLE,
    high DOUBLE,
    low DOUBLE,
    close DOUBLE,
    volume BIGINT
);
""")

# 从 CSV 批量导入（比 INSERT 快 100 倍）
conn.execute("""
COPY stock_prices FROM 'stock_data.csv'
    (FORMAT CSV, HEADER, DELIMITER ',');
""")

# 创建索引加速查询
conn.execute("CREATE INDEX IF NOT EXISTS idx_date ON stock_prices(date);")
conn.execute("CREATE INDEX IF NOT EXISTS idx_ticker ON stock_prices(ticker);")

SELECT 
    ticker,
    ROUND(AVG(close), 2) AS avg_price,
    ROUND(MAX(close), 2) AS peak_price,
    ROUND(MIN(close), 2) AS bottom_price,
    ROUND(LAST(close), 2) AS latest_price,
    ROUND((LAST(close) - FIRST(close)) / FIRST(close) * 100, 2) AS total_return_pct
FROM stock_prices
GROUP BY ticker
ORDER BY total_return_pct DESC;

SELECT 
    ticker,
    ROUND(STDDEV(close), 2) AS vol_abs,
    ROUND(STDDEV(close) / AVG(close) * 100, 2) AS vol_pct,
    ROUND(AVG(volume) / 1000000, 1) AS avg_vol_m
FROM stock_prices
GROUP BY ticker
ORDER BY vol_pct DESC;

WITH monthly AS (
    SELECT 
        ticker,
        FIRST_VALUE(close) OVER w AS month_start,
        LAST_VALUE(close) OVER w AS month_end
    FROM stock_prices
    WHERE EXTRACT(MONTH FROM date) = 6
    WINDOW w AS (
        PARTITION BY ticker 
        ORDER BY date 
        ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING
    )
)
SELECT DISTINCT ticker,
       ROUND((month_end - month_start) / month_start * 100, 2) AS return_pct
FROM monthly
ORDER BY return_pct DESC;

import time

# 100 次聚合查询的耗时
start = time.time()
for _ in range(100):
    conn.execute("""
        SELECT AVG(close), STDDEV(close) 
        FROM stock_prices 
        WHERE date BETWEEN '2026-01-01' AND '2026-06-28'
    """).fetchall()
elapsed_ms = (time.time() - start) * 1000
print(f"100 次查询总耗时: {elapsed_ms:.1f}ms（平均 {elapsed_ms/100:.2f}ms/次）")

import duckdb
import smtplib
from email.mime.text import MIMEText
from datetime import datetime

def generate_report(db_path='portfolio.db'):
    conn = duckdb.connect(db_path)
    
    # 生成报告文本
    lines = []
    lines.append(f"📊 每日市场简报 — {datetime.now().strftime('%Y-%m-%d')}")
    lines.append("")
    
    # 1. 组合概览
    lines.append("═══ 组合概览 ═══")
    for row in conn.execute("""
        SELECT ticker, 
               ROUND(LAST(close), 2),
               ROUND((LAST(close) - FIRST(close)) / FIRST(close) * 100, 2)
        FROM stock_prices
        GROUP BY ticker
        ORDER BY (LAST(close) - FIRST(close)) / FIRST(close) DESC
    """).fetchall():
        direction = "📈" if row[2] > 0 else "📉"
        lines.append(f"  {direction} {row[0]}: ${row[1]} ({row[2]}%)")
    
    lines.append("")
    
    # 2. 波动率排行
    lines.append("═══ 波动率排行 ═══")
    for row in conn.execute("""
        SELECT ticker, ROUND(STDDEV(close) / AVG(close) * 100, 2)
        FROM stock_prices GROUP BY ticker ORDER BY 2 DESC LIMIT 3
    """).fetchall():
        lines.append(f"  🔥 {row[0]}: {row[1]}%")
    
    return "\n".join(lines)

report = generate_report()
print(report)

# 发送报告（邮件 / Telegram Bot API）
# msg = MIMEText(report)
# msg['Subject'] = '每日市场简报'
# msg['From'] = '[email protected]'
# msg['To'] = '[email protected]'
# smtplib.SMTP('smtp.gmail.com', 587).send_message(msg)

# 每天 21:00 自动生成并发送邮件
0 21 * * * /usr/bin/python3 /opt/scripts/stock_report.py