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"]}'