Building an Automated Investment Research Report Generator with DuckDB: From Idea to SaaS

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

# Create an in-memory DuckDB database
con = duckdb.connect(":memory:")

# Register httpfs extension
con.execute("INSTALL httpfs; LOAD httpfs;")

# Fetch historical data for multiple tickers
tickers = ["AAPL", "GOOGL", "MSFT", "TSLA", "NVDA"]
all_data = []

for ticker in tickers:
    print(f"Fetching data for {ticker}...")
    stock = yf.Ticker(ticker)
    hist = stock.history(period="1y")
    hist['ticker'] = ticker
    hist.reset_index(inplace=True)
    all_data.append(hist)

import pyarrow as pa
import pyarrow.parquet as pq

# Convert to Arrow table and write as Parquet
table = pa.Table.from_pandas(df)
pq.write_table(table, "stocks_daily.parquet")

# Read with DuckDB (supports column pruning and predicate pushdown)
con = duckdb.connect()
con.execute("CREATE TABLE stocks AS SELECT * FROM 'stocks_daily.parquet'")

# Data overview
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)

-- Calculate 12-day and 26-day EMA, detect golden/dead crosses
WITH ema_calcs AS (
    SELECT
        ticker,
        date,
        close,
        -- 12-day EMA
        EXP(AVG(LOG(close)) OVER (
            PARTITION BY ticker 
            ORDER BY date 
            ROWS BETWEEN 11 FOLLOWING AND 0 FOLLOWING
        )) AS ema_12,
        -- 26-day 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,
        -- Annualized Sharpe = mean daily return / std daily return * 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 'High Volume Anomaly'
        WHEN (volume - vol_ma_20) / NULLIF(vol_std_20, 0) < -2 THEN 'Low Volume Anomaly'
        ELSE 'Normal'
    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
import pandas as pd

def generate_report(tickers: list[str]) -> str:
    con = duckdb.connect(":memory:")
    
    # Fetch and process data
    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"""# 📊 Daily Research Report — {report_date}

## Market Overview

"""
    
    # Query latest snapshot
    overview = con.execute("""
        WITH latest AS (
            SELECT DISTINCT ON (ticker) 
                ticker, date, close,
                LAG(close) OVER w AS prev_close
            FROM stocks
            WINDOW w AS (PARTITION BY ticker ORDER BY date)
        )
        SELECT ticker, close, 
               ROUND((close - prev_close)/prev_close * 100, 2) as change_pct
        FROM latest
        ORDER BY 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}%)\n"
    
    report += f"""

## Conclusion

Report generated automatically by DuckDB Investment Research Engine.

---
*Generated by DuckDB Auto-Research Engine | Data: yfinance*
"""
    
    return report

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

app = FastAPI(title="Research Report 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:")
        # ... data fetching and analysis ...
        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):
    """Get technical signals history for a single ticker"""
    con = duckdb.connect(":memory:")
    # ... query logic ...
    return {"ticker": ticker, "signals": signals}

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

# Start the service
pip install fastapi uvicorn duckdb yfinance pyarrow
uvicorn main:app --reload

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