独立开发者如何用 DuckDB 构建个人数据分析 SaaS

import duckdb

class EcomAnalyzer:
    def __init__(self, db_path="ecom.db"):
        self.con = duckdb.connect(db_path)
        self._init_schema()

    def _init_schema(self):
        """初始化表结构"""
        self.con.execute("""
            CREATE TABLE IF NOT EXISTS orders (
                order_id VARCHAR,
                order_date DATE,
                product_id VARCHAR,
                category VARCHAR,
                quantity BIGINT,
                unit_price DOUBLE,
                customer_id VARCHAR,
                region VARCHAR,
                status VARCHAR
            )
        """)

    def load_csv(self, file_path: str):
        """自动检测 Schema 并加载 CSV"""
        self.con.execute(f"""
            INSERT INTO orders
            SELECT * FROM read_csv_auto('{file_path}', 
                auto_detect=true,
                header=true,
                sep=','
            )
        """)

    def sales_trend_30d(self) -> str:
        """最近 30 天的销售趋势"""
        result = self.con.execute("""
            SELECT
                order_date,
                COUNT(*) AS order_count,
                SUM(quantity * unit_price) AS daily_revenue,
                AVG(quantity * unit_price) AS avg_order_value
            FROM orders
            WHERE order_date >= CURRENT_DATE - INTERVAL 30 DAY
                AND status != 'refunded'
            GROUP BY order_date
            ORDER BY order_date
        """).fetchdf()
        return result.to_json(orient='records', date_format='iso')

    def profit_ranking(self, top_n: int = 20) -> str:
        """商品利润排行榜：找出最赚钱和最亏钱的商品"""
        result = self.con.execute("""
            SELECT
                product_id,
                category,
                SUM(quantity) AS total_sold,
                SUM(quantity * unit_price) AS total_revenue,
                COUNT(DISTINCT customer_id) AS unique_customers,
                COUNT(DISTINCT customer_id) * 1.5 AS estimated_cogs,
                SUM(quantity * unit_price) 
                    - COUNT(DISTINCT customer_id) * 1.5 AS estimated_profit
            FROM orders
            WHERE status != 'refunded'
            GROUP BY product_id, category
            ORDER BY estimated_profit DESC
            LIMIT ?
        """, [top_n])
        return result.fetchdf().to_json(orient='records')

    def weekly_pattern(self) -> str:
        """每周各天的销量模式"""
        result = self.con.execute("""
            SELECT
                CASE EXTRACT(DOW FROM order_date)
                    WHEN 0 THEN '周日'
                    WHEN 1 THEN '周一'
                    WHEN 2 THEN '周二'
                    WHEN 3 THEN '周三'
                    WHEN 4 THEN '周四'
                    WHEN 5 THEN '周五'
                    WHEN 6 THEN '周六'
                END AS day_name,
                SUM(quantity) AS total_quantity,
                SUM(quantity * unit_price) AS total_revenue,
                COUNT(*) AS order_count
            FROM orders
            GROUP BY EXTRACT(DOW FROM order_date)
            ORDER BY EXTRACT(DOW FROM order_date)
        """).fetchdf()
        return result.to_json(orient='records')

    def repurchase_rate(self) -> str:
        """客户复购率分析"""
        result = self.con.execute("""
            WITH customer_orders AS (
                SELECT
                    customer_id,
                    COUNT(DISTINCT DATE_TRUNC('month', order_date)) AS active_months,
                    COUNT(*) AS total_orders,
                    SUM(quantity * unit_price) AS lifetime_value
                FROM orders
                WHERE status != 'refunded'
                GROUP BY customer_id
            )
            SELECT
                AVG(CASE WHEN total_orders > 1 THEN 1.0 ELSE 0.0 END) AS repurchase_rate,
                AVG(lifetime_value) AS avg_customer_value,
                PERCENTILE_CONT(0.95) WITHIN GROUP (ORDER BY lifetime_value) AS top_5p_value
            FROM customer_orders
        """).fetchdf()
        return result.to_json(orient='records')

    def generate_report(self) -> dict:
        """生成完整分析报告"""
        return {
            "summary": self.sales_trend_30d(),
            "profit_ranking": self.profit_ranking(),
            "weekly_pattern": self.weekly_pattern(),
            "repurchase_rate": self.repurchase_rate(),
            "generated_at": str(duckdb.query("SELECT NOW()").fetchone()[0])
        }

# 加载 Excel
self.con.execute("SELECT * FROM read_xlsx('orders.xlsx')")

# 加载 Parquet（性能更好）
self.con.execute("SELECT * FROM read_parquet('orders_*.parquet')")

# 加载 JSON
self.con.execute("SELECT * FROM read_json_auto('orders.json')")

    def data_quality_check(self) -> dict:
        """数据质量检查"""
        checks = {
            "missing_order_ids": self.con.execute("""
                SELECT COUNT(*) FROM orders WHERE order_id IS NULL
            """).fetchone()[0],
            "invalid_dates": self.con.execute("""
                SELECT COUNT(*) FROM orders 
                WHERE order_date IS NULL OR order_date > CURRENT_DATE
            """).fetchone()[0],
            "negative_prices": self.con.execute("""
                SELECT COUNT(*) FROM orders WHERE unit_price < 0
            """).fetchone()[0],
        }
        return checks

    def cache_monthly_summary(self, year: int, month: int):
        """缓存月度汇总"""
        self.con.execute("""
            CREATE OR REPLACE VIEW monthly_summary_vw AS
            SELECT
                DATE_TRUNC('month', order_date) AS month,
                COUNT(*) AS total_orders,
                SUM(quantity * unit_price) AS total_revenue,
                COUNT(DISTINCT customer_id) AS unique_customers
            FROM orders
            WHERE EXTRACT(YEAR FROM order_date) = ?
              AND EXTRACT(MONTH FROM order_date) = ?
            GROUP BY 1
        """, [year, month])