DuckDB Monetization 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