用 DuckDB 搭建自动投研报告生成器：从想法到 SaaS 的完整路径

import duckdb
import yfinance as yf
from datetime import datetime, timedelta

# 创建 DuckDB 内存数据库
con = duckdb.connect(":memory:")

# 注册 yfinance 为 DuckDB 扩展（可选）
con.execute("INSTALL httpfs; LOAD httpfs;")

# 拉取多只股票的历史数据
tickers = ["AAPL", "GOOGL", "MSFT", "TSLA", "NVDA"]
all_data = []

for ticker in tickers:
    print(f"正在拉取 {ticker} 的数据...")
    stock = yf.Ticker(ticker)
    hist = stock.history(period="1y")
    hist['ticker'] = ticker
    hist.reset_index(inplace=True)
    all_data.append(hist)

# 合并为一张表
df = duckdb.sql("SELECT * FROM read_auto('combined.parquet')")

import pyarrow as pa
import pyarrow.parquet as pq

# 转换为 Arrow 表后写入 Parquet
table = pa.Table.from_pandas(df)
pq.write_table(table, "stocks_daily.parquet")

# 用 DuckDB 读取（支持列裁剪和谓词下推）
con = duckdb.connect()
con.execute("CREATE TABLE stocks AS SELECT * FROM 'stocks_daily.parquet'")

# 查看数据概况
result = con.execute("""
    SELECT ticker, MIN(date) as start_date, MAX(date) as end_date, COUNT(*) as rows
    FROM stocks GROUP BY ticker ORDER BY ticker
""").fetchdf()
print(result)

-- 计算 12日 和 26日 EMA，检测金叉死叉
WITH ema_calcs AS (
    SELECT
        ticker,
        date,
        close,
        -- 12日 EMA
        EXP(AVG(LOG(close)) OVER (
            PARTITION BY ticker 
            ORDER BY date 
            ROWS BETWEEN 11 FOLLOWING AND 0 FOLLOWING
        )) AS ema_12,
        -- 26日 EMA
        EXP(AVG(LOG(close)) OVER (
            PARTITION BY ticker 
            ORDER BY date 
            ROWS BETWEEN 25 FOLLOWING AND 0 FOLLOWING
        )) AS ema_26
    FROM stocks
),
signals AS (
    SELECT
        ticker,
        date,
        close,
        ema_12,
        ema_26,
        LAG(ema_12) OVER (PARTITION BY ticker ORDER BY date) AS prev_ema_12,
        LAG(ema_26) OVER (PARTITION BY ticker ORDER BY date) AS prev_ema_26,
        CASE
            WHEN ema_12 > ema_26 AND LAG(ema_12) OVER (PARTITION BY ticker ORDER BY date) 
                          <= LAG(ema_26) OVER (PARTITION BY ticker ORDER BY date)
            THEN 'GOLDEN_CROSS'  -- 金叉
            WHEN ema_12 < ema_26 AND LAG(ema_12) OVER (PARTITION BY ticker ORDER BY date) 
                          >= LAG(ema_26) OVER (PARTITION BY ticker ORDER BY date)
            THEN 'DEAD_CROSS'    -- 死叉
            ELSE 'HOLD'
        END AS signal
    FROM ema_calcs
)
SELECT ticker, date, close, ema_12, ema_26, signal
FROM signals
WHERE signal != 'HOLD'
ORDER BY ticker, date;

WITH daily_returns AS (
    SELECT
        ticker,
        date,
        LOG(close / LAG(close) OVER (PARTITION BY ticker ORDER BY date)) AS daily_return
    FROM stocks
),
sharpe_calc AS (
    SELECT
        ticker,
        -- 年化夏普比率 = 日均收益率 / 日标准差 × sqrt(252)
        ROUND(
            AVG(daily_return) / NULLIF(STDDEV(daily_return), 0) * SQRT(252),
            2
        ) AS sharpe_ratio,
        ROUND(AVG(daily_return) * 252, 4) AS annual_return,
        ROUND(STDDEV(daily_return) * SQRT(252), 4) AS annual_volatility,
        COUNT(*) AS trading_days
    FROM daily_returns
    GROUP BY ticker
)
SELECT * FROM sharpe_calc
ORDER BY sharpe_ratio DESC;

WITH volume_stats AS (
    SELECT
        ticker,
        date,
        volume,
        AVG(volume) OVER (PARTITION BY ticker ORDER BY date ROWS BETWEEN 19 FOLLOWING AND -1 FOLLOWING) AS vol_ma_20,
        STDDEV(volume) OVER (PARTITION BY ticker ORDER BY date ROWS BETWEEN 19 FOLLOWING AND -1 FOLLOWING) AS vol_std_20
    FROM stocks
)
SELECT
    ticker,
    date,
    volume,
    ROUND(vol_ma_20, 0) AS avg_volume_20d,
    ROUND((volume - vol_ma_20) / NULLIF(vol_std_20, 0), 2) AS z_score,
    CASE
        WHEN (volume - vol_ma_20) / NULLIF(vol_std_20, 0) > 2 THEN '异常放量'
        WHEN (volume - vol_ma_20) / NULLIF(vol_std_20, 0) < -2 THEN '异常缩量'
        ELSE '正常'
    END AS anomaly_flag
FROM volume_stats
WHERE date >= CURRENT_DATE - INTERVAL '5 days'
ORDER BY ABS(z_score) DESC
LIMIT 20;

import duckdb
import datetime

def generate_report(tickers: list[str]) -> str:
    con = duckdb.connect(":memory:")
    
    # 拉取数据
    all_data = []
    for ticker in tickers:
        stock = yf.Ticker(ticker)
        hist = stock.history(period="6mo")
        hist['ticker'] = ticker
        hist.reset_index(inplace=True)
        all_data.append(hist)
    
    df = pd.concat(all_data, ignore_index=True)
    df.to_parquet("/tmp/stocks.parquet")
    con.execute("CREATE TABLE stocks AS SELECT * FROM '/tmp/stocks.parquet'")
    
    # 生成报告
    report_date = datetime.date.today().strftime("%Y-%m-%d")
    
    report = f"""# 📊 投研日报 — {report_date}

## 市场概览

| 股票 | 最新价 | 日涨跌 | 夏普比率 | 信号 |
|------|--------|--------|---------|------|
"""
    
    # 查询每个股票的概况
    overview = con.execute("""
        WITH latest AS (
            SELECT DISTINCT ON (ticker) 
                ticker, date, close, volume,
                LAG(close) OVER w AS prev_close,
                LAG(volume) OVER w AS prev_volume
            FROM stocks
            WINDOW w AS (PARTITION BY ticker ORDER BY date)
        ),
        sharpe AS (
            SELECT ticker, 
                   ROUND(AVG(LOG(close/LAG(close) OVER w))*252, 4) as ann_ret,
                   ROUND(STDDEV(LOG(close/LAG(close) OVER w)) * SQRT(252), 4) as ann_vol
            FROM stocks
            WINDOW w AS (PARTITION BY ticker ORDER BY date)
            GROUP BY ticker
        )
        SELECT l.ticker, l.close, 
               ROUND((l.close - l.prev_close)/l.prev_close * 100, 2) as change_pct,
               ROUND(s.ann_ret / NULLIF(s.ann_vol, 0) * SQRT(252), 2) as sharpe
        FROM latest l
        JOIN sharpe s ON l.ticker = s.ticker
        ORDER BY l.ticker
    """).fetchdf()
    
    for _, row in overview.iterrows():
        change = row['change_pct']
        arrow = "📈" if change >= 0 else "📉"
        report += f"| {row['ticker']} | {row['close']:.2f} | {arrow}{abs(change):.2f}% | {row['sharpe']:.2f} | - |\n"
    
    report += """

## 结论

以上为今日自动生成报告，仅供参考。

---
*本报告由 DuckDB 自动投研引擎生成 | 数据来源: yfinance*
"""
    
    return report

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import duckdb
import json

app = FastAPI(title="投研报告 SaaS", version="1.0.0")

class ReportRequest(BaseModel):
    tickers: list[str]
    date_range: str = "1y"
    include_signals: bool = True

@app.get("/health")
async def health():
    return {"status": "ok", "engine": "DuckDB"}

@app.post("/report")
async def generate_report(req: ReportRequest):
    try:
        con = duckdb.connect(":memory:")
        
        # 拉取数据并分析（复用上面的逻辑）
        # ... 省略中间步骤 ...
        
        # 返回 JSON 格式的报告
        return {
            "status": "success",
            "generated_at": datetime.datetime.now().isoformat(),
            "tickers": req.tickers,
            "report_markdown": report_content
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/tickers/{ticker}/signals")
async def get_signals(ticker: str):
    """获取单只股票的技术信号历史"""
    con = duckdb.connect(":memory:")
    # ... 查询逻辑 ...
    return {"ticker": ticker, "signals": signals}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

# 启动服务
pip install fastapi uvicorn duckdb yfinance pyarrow
uvicorn main:app --reload

# 测试 API
curl -X POST http://localhost:8000/report \
  -H "Content-Type: application/json" \
  -d '{"tickers": ["AAPL", "NVDA", "TSLA"]}'