DuckDB 变现案例：一个数据分析师靠它月入 3 万的「行业报告 SaaS」

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

class IndustryReportEngine:
    """行业报告分析引擎"""
    
    def __init__(self):
        # 使用内存数据库，每次请求独立隔离
        self.con = duckdb.connect(":memory:")
        self._init_schema()
        
    def _init_schema(self):
        """初始化 DuckDB 内置函数和扩展"""
        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]):
        """
        接入多个公开数据源
        data_sources 格式：[{"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:
        """
        市场规模估算 —— 利用 DuckDB 的模糊匹配和聚合能力
        
        核心思路：从多源数据中筛选与关键词相关的记录，
        通过历史数据的复合增长率推算当前市场规模
        """
        query = f"""
            WITH relevant_records AS (
                -- 从所有数据源中筛选关键词相关记录
                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,
                    -- 计算年复合增长率
                    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:
        """
        竞争格局分析 —— 找出头部玩家及其市场份额
        
        利用 DuckDB 的 RANK() 聚合和分箱能力
        """
        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,
                    -- 计算市场份额占比
                    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,
                -- 市场份额等级划分
                CASE
                    WHEN market_share_pct >= 20 THEN '头部玩家'
                    WHEN market_share_pct >= 5 THEN '主要竞争者'
                    ELSE '长尾玩家'
                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:
        """
        增长趋势预测 —— 基于历史数据的简单线性回归
        
        利用 DuckDB 的线性回归函数进行趋势外推
        """
        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():
    """生成行业报告"""
    data = request.json
    keyword = data['keyword']
    
    engine = IndustryReportEngine()
    
    # 模拟从公开数据源加载数据
    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"}
    ])
    
    # 执行分析
    market_info = engine.market_size_estimate(keyword)
    competitors = engine.competitive_landscape(keyword)
    forecasts = engine.trend_forecast(keyword)
    
    # 生成 PDF
    buffer = io.BytesIO()
    c = canvas.Canvas(buffer, pagesize=A4)
    
    # 标题
    c.setFont("Helvetica-Bold", 24)
    c.drawString(50, 750, f"{keyword} 行业趋势报告")
    
    # 市场规模
    c.setFont("Helvetica", 14)
    c.drawString(50, 700, f"最新市场规模: ¥{market_info['market_size']:,}")
    c.drawString(50, 680, f"年复合增长率: {market_info['cagr']}%")
    c.drawString(50, 660, f"竞争对手数量: {market_info['competitors']}")
    
    # 竞争格局
    c.drawString(50, 620, "竞争格局:")
    for _, row in competitors.head(5).iterrows():
        c.drawString(70, 600 - row['rank']*20, 
                     f"{row['rank']}. {row['company']} ({row['market_share_pct']}%)")
    
    # 趋势预测
    c.drawString(50, 450, "未来趋势预测:")
    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}_行业报告.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