DuckDB 实战：用数据流水线把公开数据变成可售卖的商品

import duckdb
from pathlib import Path

# 假设你有三个文件夹，里面是公开数据导出的 CSV
poi_dir = Path("data/poi")        # 高德 POI 数据
weather_dir = Path("data/weather") # 天气数据
population_dir = Path("data/population") # 人口数据

# 一键读取某城市的所有 POI CSV
con = duckdb.connect(":memory:")

# DuckDB 支持通配符路径，自动合并多个文件
con.execute("""
    CREATE TABLE pois AS
    SELECT * FROM read_csv_auto('data/poi/*.csv', autoprompt=true)
""")

# 查看数据结构和分布
print(con.execute("DESCRIBE pois").fetchall())
print(con.execute("SELECT COUNT(*) as total FROM pois").fetchone())

# 商圈指标计算：用 CTE + 窗口函数，一条 SQL 搞定
business_index_sql = """
WITH poi_counts AS (
    -- 每个商圈的 POI 总数和分类数
    SELECT 
        district,
        category,
        COUNT(*) as poi_count,
        COUNT(DISTINCT name) as unique_count
    FROM pois
    WHERE city = '北京'
    GROUP BY district, category
),
district_stats AS (
    -- 每个区的综合指标
    SELECT 
        district,
        SUM(poi_count) as total_pois,
        COUNT(DISTINCT category) as category_diversity,
        AVG(poi_count) as avg_poi_per_category,
        MAX(poi_count) as max_poi_per_category,
        -- 竞争强度：头部POI数量占比
        SUM(CASE WHEN poi_count > avg_poi_per_category * 1.5 THEN 1 ELSE 0 END) 
            as high_competition_count
    FROM poi_counts
    GROUP BY district
)
SELECT 
    district,
    total_pois,
    category_diversity,
    ROUND(avg_poi_per_category, 2) as avg_competitiveness,
    ROUND(high_competition_count::float / category_diversity * 100, 1) 
        as competition_intensity_pct,
    -- 商业活力综合得分（加权公式）
    ROUND(
        total_pois * 0.3 
        + category_diversity * 15 
        + (100 - GREATEST(competition_intensity_pct, 0)) * 0.5,
        1
    ) as business_heat_score
FROM district_stats
ORDER BY business_heat_score DESC
"""

result = con.execute(business_index_sql).fetchall()
columns = [desc[0] for desc in con.execute(business_index_sql).description]

print(f"\n{'='*60}")
print(f"🏙️  北京各商圈热力指数 TOP 10")
print(f"{'='*60}")
for row in result[:10]:
    print(f"  {row[0]:<10} | 综合得分: {row[9]:>6} | POI: {row[1]:>5} | 品类: {row[2]:>2} | 竞争: {row[8]:>5}%")

import json
from datetime import datetime

def build_product(city: str, output_path: str = "output"):
    """构建城市商业数据产品"""
    from pathlib import Path
    Path(output_path).mkdir(exist_ok=True)
    
    # 1. 生成综合评分表
    table = con.execute(f"""
        {business_index_sql.replace("北京", city)}
    """).fetchdf()
    
    # 保存为 CSV（数据买家最常用格式）
    csv_path = f"{output_path}/{city}_商业热力指数.csv"
    table.to_csv(csv_path, index=False, encoding='utf-8-sig')
    
    # 2. 保存 JSON 元数据（方便 API 调用）
    metadata = {
        "city": city,
        "generated_at": datetime.now().isoformat(),
        "data_source": ["高德POI", "国家统计局"],
        "metrics": {
            "total_districts": len(table),
            "avg_score": round(table["business_heat_score"].mean(), 2),
            "top_district": table.iloc[0]["district"],
            "top_score": table.iloc[0]["business_heat_score"]
        },
        "fields": {col: str(dtype) for col, dtype in table.dtypes.items()}
    }
    
    with open(f"{output_path}/{city}_metadata.json", "w", encoding="utf-8") as f:
        json.dump(metadata, f, ensure_ascii=False, indent=2)
    
    # 3. 生成数据报告摘要
    summary = f"""
📊 {city}商业热力指数报告
生成时间: {metadata['generated_at'][:10]}
数据来源: {', '.join(metadata['data_source'])}

核心发现:
• 最高分商圈: {table.iloc[0]['district']} ({table.iloc[0]['business_heat_score']}分)
• 平均商业活力: {metadata['metrics']['avg_score']}分
• 涉及商圈数: {metadata['metrics']['total_districts']}个

指标说明:
- 综合得分 = POI密度×0.3 + 品类多样性×15 + (100-竞争强度)×0.5
- 竞争强度越高代表该区域同类商户越密集
    """
    
    with open(f"{output_path}/{city}_report.txt", "w", encoding="utf-8") as f:
        f.write(summary)
    
    print(f"✅ {city} 数据产品已生成 → {output_path}/")
    print(summary)
    return table

# 一键生成全国重点城市数据
for city in ["北京", "上海", "深圳", "杭州", "成都"]:
    df = build_product(city)

# 启用并行读取，利用多核 CPU
con = duckdb.connect(":memory:", config={
    'threads': '4',           # 使用4个线程
    'max_memory': '2GB',      # 限制内存使用
    'temp_directory': '/tmp/duckdb_temp'  # 临时文件目录
})

# 并行读取+处理
con.execute("""
    SET parallel_degree = 4;
    CREATE TABLE pois AS
    SELECT * FROM read_csv_auto('data/poi/*.csv', autoprompt=true, parallel=true)
""")

# 将结果存为 Parquet，压缩比高且查询更快
con.execute("""
    COPY (
        SELECT * FROM business_index_result
    ) TO 'output/business_heat.parquet' (FORMAT PARQUET, COMPRESSION ZSTD)
""")

# 下次查询时直接读 Parquet，速度提升 10 倍以上
parquet_df = con.execute("SELECT * FROM 'output/business_heat.parquet'").fetchdf()

import schedule
import time

def daily_update():
    print(f"开始更新数据: {datetime.now()}")
    for city in CITIES:
        build_product(city)
    print("✅ 全部更新完成")

# 每天凌晨 2 点自动更新
schedule.every().day.at("02:00").do(daily_update)

while True:
    schedule.run_pending()
    time.sleep(60)