30行代码用DuckDB搭建个人数据产品后端，月入过万不是梦

import duckdb
import pandas as pd
import numpy as np
from datetime import datetime, timedelta

def generate_daily_sales(days=30):
    """模拟生成30天销售数据"""
    np.random.seed(42)
    stores = [f"store_{i:03d}" for i in range(1, 11)]
    products = [f"prod_{i:03d}" for i in range(1, 51)]
    
    records = []
    for day_offset in range(days):
        date = datetime.now() - timedelta(days=day_offset)
        n_records = np.random.randint(15000, 25000)
        
        df = pd.DataFrame({
            'date': np.random.choice(
                pd.date_range(date, periods=1), n_records
            ),
            'store_id': np.random.choice(stores, n_records),
            'product_id': np.random.choice(products, n_records),
            'quantity': np.random.randint(1, 20, n_records),
            'unit_price': np.round(np.random.uniform(10, 500, n_records), 2),
            'region': np.random.choice(['CN', 'US', 'EU'], n_records, p=[0.6, 0.3, 0.1])
        })
        df['revenue'] = df['quantity'] * df['unit_price']
        records.append(df)
    
    return pd.concat(records, ignore_index=True)

# 生成 150 万行模拟数据
df = generate_daily_sales(30)
print(f"生成 {len(df)} 行数据")
print(df.head())

import duckdb

class SalesAnalyticsEngine:
    def __init__(self, parquet_path="s3://bucket/sales/"):
        self.path = parquet_path
        self.con = duckdb.connect("memory:")
        # 直接扫描 Parquet，零数据拷贝
        self.con.execute(f"""
            CREATE TABLE sales AS 
            SELECT * FROM read_parquet('{parquet_path}*.parquet')
        """)
    
    def daily_revenue(self, store_id: str, days: int = 30) -> list:
        """某店铺近N天每日营收"""
        result = self.con.execute(f"""
            SELECT date::DATE as dt,
                   SUM(revenue) as revenue,
                   SUM(quantity) as orders
            FROM sales 
            WHERE store_id = '{store_id}'
              AND date >= CURRENT_DATE - INTERVAL '{days} DAYS'
            GROUP BY dt
            ORDER BY dt DESC
            LIMIT {days}
        """).fetchall()
        return [{"date": r[0].isoformat(), "revenue": r[1], "orders": r[2]} for r in result]
    
    def top_products(self, store_id: str, n: int = 10) -> list:
        """TOP N 畅销商品"""
        result = self.con.execute(f"""
            SELECT product_id,
                   SUM(quantity) as total_qty,
                   SUM(revenue) as total_rev,
                   COUNT(DISTINCT date) as active_days
            FROM sales 
            WHERE store_id = '{store_id}'
              AND date >= CURRENT_DATE - INTERVAL '30 DAYS'
            GROUP BY product_id
            ORDER BY total_rev DESC
            LIMIT {n}
        """).fetchall()
        return [{"product_id": r[0], "qty": r[1], "revenue": r[2], "days": r[3]} for r in result]
    
    def regional_breakdown(self, store_id: str) -> dict:
        """区域销售分布"""
        result = self.con.execute(f"""
            SELECT region,
                   COUNT(*) as txn_count,
                   SUM(revenue) as revenue,
                   AVG(unit_price) as avg_price
            FROM sales 
            WHERE store_id = '{store_id}'
            GROUP BY region
        """).fetchall()
        return {r[0]: {"txns": r[1], "revenue": r[2], "avg_price": r[3]} for r in result}

from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
import uvicorn

app = FastAPI(title="电商销售分析 API")
engine = SalesAnalyticsEngine()

@app.middleware("http")
async def auth_middleware(request, call_next):
    """简单 API Key 鉴权"""
    api_key = request.headers.get("X-API-Key")
    if api_key != "your-premium-key":
        raise HTTPException(status_code=401)
    return await call_next(request)

@app.get("/api/v1/{store_id}/daily")
async def get_daily(store_id: str, days: int = 30):
    return {"store_id": store_id, "data": engine.daily_revenue(store_id, days)}

@app.get("/api/v1/{store_id}/top-products")
async def get_top_products(store_id: str, n: int = 10):
    return {"store_id": store_id, "data": engine.top_products(store_id, n)}

@app.get("/api/v1/{store_id}/regional")
async def get_regional(store_id: str):
    return {"store_id": store_id, "data": engine.regional_breakdown(store_id)}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

import json, redis

r = redis.Redis(host='localhost', port=6379, db=0)

def get_with_cache(store_id: str, query_type: str, ttl: int = 300):
    key = f"sales:{store_id}:{query_type}"
    cached = r.get(key)
    if cached:
        return json.loads(cached)
    
    if query_type == "daily":
        data = engine.daily_revenue(store_id)
    elif query_type == "top_products":
        data = engine.top_products(store_id)
    else:
        data = engine.regional_breakdown(store_id)
    
    r.setex(key, ttl, json.dumps(data))
    return data