DuckDB Case Study: Building a ¥30K/Month Report SaaS

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

class IndustryReportEngine:
 """Industry Report Analysis Engine"""
 
 def __init__(self):
 # Use in-memory database, isolated per request
 self.con = duckdb.connect(":memory:")
 self._init_schema()
 
 def _init_schema(self):
 """Initialize DuckDB built-in functions and extensions"""
 self.con.execute("INSTALL httpfs; LOAD httpfs")
 self.con.execute("INSTALL json; LOAD json")
 self.con.execute("INSTALL parquet; LOAD parquet")
 self.con.execute("INSTALL rds; LOAD rds")
 
 def ingest_public_data(self, data_sources: list[dict]):
 """
 Ingest multiple public data sources
 data_sources format: [{"url": "...", "format": "csv"}, ...]
 """
 for i, source in enumerate(data_sources):
 fmt = source.get("format", "csv")
 path = source["path"] or source["url"]
 
 if fmt == "csv":
 self.con.execute(f"""
 CREATE OR REPLACE TABLE source_{i} AS 
 SELECT * FROM read_csv_auto('{path}', auto_detect=true)
 """)
 elif fmt == "json":
 self.con.execute(f"""
 CREATE OR REPLACE TABLE source_{i} AS 
 SELECT * FROM read_json_auto('{path}')
 """)
 elif fmt == "parquet":
 self.con.execute(f"""
 CREATE OR REPLACE TABLE source_{i} AS 
 SELECT * FROM read_parquet('{path}')
 """)
 
 def market_size_estimate(self, keyword: str) -> dict:
 """
 Market Size Estimation — leveraging DuckDB's fuzzy matching and aggregation
 
 Core idea: filter records related to the keyword from multiple data sources,
 then project current market size through historical CAGR calculation
 """
 query = f"""
 WITH relevant_records AS (
 -- Filter keyword-related records from all data sources
 SELECT 'source_0' AS src, name, revenue, year, quarter, region
 FROM source_0 WHERE name ILIKE '%{keyword}%'
 UNION ALL
 SELECT 'source_1' AS src, name, revenue, year, quarter, region
 FROM source_1 WHERE description ILIKE '%{keyword}%'
 UNION ALL
 SELECT 'source_2' AS src, product_name, market_value, year, NULL, region
 FROM source_2 WHERE product_name ILIKE '%{keyword}%'
 ),
 yearly_revenue AS (
 SELECT
 year,
 SUM(revenue) AS total_revenue,
 COUNT(DISTINCT name) AS company_count,
 ROUND(AVG(revenue), 2) AS avg_company_revenue
 FROM relevant_records
 WHERE revenue IS NOT NULL AND year IS NOT NULL
 GROUP BY year
 ORDER BY year
 ),
 growth_calc AS (
 SELECT
 year,
 total_revenue,
 company_count,
 avg_company_revenue,
 -- Calculate year-over-year compound growth rate
 CASE WHEN LAG(total_revenue) OVER (ORDER BY year) > 0 THEN
 ROUND(
 POWER(total_revenue / LAG(total_revenue) OVER (ORDER BY year), 
 1.0 / (year - LAG(year) OVER (ORDER BY year))) - 1,
 4
 ) * 100
 ELSE NULL END AS cagr_pct
 FROM yearly_revenue
 )
 SELECT
 MAX(year) AS latest_year,
 ROUND(MAX(total_revenue), 2) AS latest_market_size,
 ROUND(AVG(cagr_pct)) AS avg_cagr,
 MAX(company_count) AS max_competitors,
 ROUND(AVG(avg_company_revenue), 2) AS avg_firm_size
 FROM growth_calc
 """
 result = self.con.execute(query).fetchone()
 return {
 "latest_year": int(result[0]),
 "market_size": float(result[1]),
 "cagr": float(result[2]) if result[2] else None,
 "competitors": int(result[3]),
 "avg_firm_size": float(result[4])
 }
 
 def competitive_landscape(self, keyword: str) -> pd.DataFrame:
 """
 Competitive Landscape Analysis — identify top players and their market shares
 
 Leverages DuckDB's RANK() aggregation and binning capabilities
 """
 query = f"""
 WITH ranked AS (
 SELECT
 COALESCE(name, product_name) AS company,
 region,
 SUM(COALESCE(revenue, market_value, 0)) AS total_revenue,
 RANK() OVER (ORDER BY SUM(COALESCE(revenue, market_value, 0)) DESC) AS rank,
 -- Calculate market share percentage
 ROUND(
 100.0 * SUM(COALESCE(revenue, market_value, 0)) 
 / NULLIF(SUM(SUM(COALESCE(revenue, market_value, 0))) OVER (), 0),
 2
 ) AS market_share_pct
 FROM (
 SELECT name, NULL AS product_name, region, revenue, NULL AS market_value FROM source_0
 UNION ALL
 SELECT NULL, product_name, region, NULL, market_value FROM source_2
 ) combined
 WHERE name ILIKE '%{keyword}%' OR product_name ILIKE '%{keyword}%'
 GROUP BY company, region
 )
 SELECT
 rank,
 company,
 region,
 ROUND(total_revenue, 2) AS revenue,
 market_share_pct,
 -- Tier classification by market share
 CASE
 WHEN market_share_pct >= 20 THEN 'Top Player'
 WHEN market_share_pct >= 5 THEN 'Major Competitor'
 ELSE 'Long Tail'
 END AS tier
 FROM ranked
 ORDER BY rank
 """
 return self.con.execute(query).df()
 
 def trend_forecast(self, keyword: str, periods: int = 3) -> pd.DataFrame:
 """
 Trend Forecasting — simple linear regression based on historical data
 
 Uses DuckDB's linear regression functions for trend extrapolation
 """
 query = f"""
 WITH relevant_data AS (
 SELECT year, SUM(revenue) AS total_revenue
 FROM source_0
 WHERE name ILIKE '%{keyword}%'
 GROUP BY year
 ORDER BY year
 ),
 regression AS (
 SELECT
 REGR_SLOPE(total_revenue, year) AS slope,
 REGR_INTERCEPT(total_revenue, year) AS intercept,
 CORR(total_revenue, year) AS correlation
 FROM relevant_data
 ),
 forecast_years AS (
 SELECT UNNEST(GENERATE_SERIES(
 (SELECT MAX(year) FROM relevant_data) + 1,
 (SELECT MAX(year) FROM relevant_data) + {periods},
 1
 )) AS year
 )
 SELECT
 fy.year,
 ROUND(LEAST(0, r.slope * fy.year + r.intercept), 2) AS predicted_revenue,
 ROUND(r.correlation * 100, 1) AS confidence_pct
 FROM forecast_years fy
 CROSS JOIN regression r
 ORDER BY fy.year
 """
 return self.con.execute(query).df()

from flask import Flask, request, jsonify
from reportlab.lib.pagesizes import A4
from reportlab.pdfgen import canvas
import io

app = Flask(__name__)

@app.route('/generate_report', methods=['POST'])
def generate_report():
 """Generate an industry report"""
 data = request.json
 keyword = data['keyword']
 
 engine = IndustryReportEngine()
 
 # Simulate loading data from public sources
 engine.ingest_public_data([
 {"path": "/data/company_data.csv", "format": "csv"},
 {"path": "/data/market_data.json", "format": "json"},
 {"path": "/data/industry_trends.parquet", "format": "parquet"}
 ])
 
 # Run analyses
 market_info = engine.market_size_estimate(keyword)
 competitors = engine.competitive_landscape(keyword)
 forecasts = engine.trend_forecast(keyword)
 
 # Generate PDF
 buffer = io.BytesIO()
 c = canvas.Canvas(buffer, pagesize=A4)
 
 # Title
 c.setFont("Helvetica-Bold", 24)
 c.drawString(50, 750, f"{keyword} Industry Trend Report")
 
 # Market size
 c.setFont("Helvetica", 14)
 c.drawString(50, 700, f"Latest Market Size: ¥{market_info['market_size']:,}")
 c.drawString(50, 680, f"CAGR: {market_info['cagr']}%")
 c.drawString(50, 660, f"Competitors: {market_info['competitors']}")
 
 # Competitive landscape
 c.drawString(50, 620, "Competitive Landscape:")
 for _, row in competitors.head(5).iterrows():
 c.drawString(70, 600 - row['rank']*20, 
 f"{row['rank']}. {row['company']} ({row['market_share_pct']}%)")
 
 # Trend forecast
 c.drawString(50, 450, "Future Trend Forecast:")
 for _, row in forecasts.iterrows():
 c.drawString(70, 430 - (row['year'] - forecasts['year'].min())*20,
 f"{int(row['year'])}: ¥{row['predicted_revenue']:,.0f}")
 
 c.save()
 buffer.seek(0)
 
 return send_file(
 buffer,
 mimetype='application/pdf',
 as_attachment=True,
 download_name=f'{keyword}_report.pdf'
 )

# Dockerfile
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
EXPOSE 5000
CMD ["python", "app.py"]

# requirements.txt
flask==3.0.0
duckdb==1.5.0
pandas==2.2.0
reportlab==4.0.0
gunicorn==21.2.0