用 DuckDB 构建电商竞品监控服务：从数据抓取到变现的全流程

store_name,product_name,price,sales_volume,date,review_count,negative_keywords
"店铺A","宠物零食大礼包",49.9,1200,2026-07-20,,好评如潮
"店铺B","宠物营养膏",89.0,800,2026-07-20,15,物流慢|包装破损

import duckdb

con = duckdb.connect("monitor.db")

# 直接读取远程或本地 CSV，启用错误容错
con.execute("""
    CREATE TABLE competitor_data AS
    SELECT * FROM read_csv_auto('competitors/*.csv',
        RETURN_NULL_ON_ERROR=true,
        header=true,
        sep=','
    )
""")

# 查看导入了多少行，多少行有脏数据
result = con.execute("""
    SELECT 
        COUNT(*) AS total_rows,
        COUNT(price) AS valid_price_rows,
        COUNT(sales_volume) AS valid_sales_rows,
        COUNT(*) - COUNT(price) AS null_price_count
    FROM competitor_data
""").fetchone()

print(f"总行数: {result[0]}, 有效价格: {result[1]}, 缺失价格: {result[3]}")

CREATE TABLE remote_data AS
SELECT * FROM read_csv_auto(
    'https://example.com/data/competitors-july.csv',
    RETURN_NULL_ON_ERROR=true
);

-- 统一价格格式，去除货币符号，转为数值
CREATE TABLE clean_prices AS
SELECT 
    store_name,
    product_name,
    TRY_CAST(regexp_replace(price::VARCHAR, '[^0-9.]', '', 'g') AS DOUBLE) AS price_cny,
    sales_volume,
    date
FROM competitor_data;

-- 查看清洗后的数据分布
SELECT 
    percentile_cont(0.25) WITHIN GROUP (ORDER BY price_cny) AS p25,
    percentile_cont(0.50) WITHIN GROUP (ORDER BY price_cny) AS median,
    percentile_cont(0.75) WITHIN GROUP (ORDER BY price_cny) AS p75,
    AVG(price_cny) AS avg_price,
    MIN(price_cny) AS min_price,
    MAX(price_cny) AS max_price
FROM clean_prices
WHERE price_cny IS NOT NULL;

-- 使用 DuckDB 的 fuzzy 扩展进行商品名相似度计算
CREATE EXTENSION fuzzy;

-- 找出相似的商品名（用于合并同一商品的不同表述）
SELECT 
    a.product_name AS name_a,
    b.product_name AS name_b,
    similarity(a.product_name, b.product_name) AS sim_score
FROM clean_prices a
CROSS JOIN clean_prices b
ON a.rowid < b.rowid
WHERE similarity(a.product_name, b.product_name) > 0.7
LIMIT 20;

-- 将逗号分隔的差评关键词拆分为独立行并统计
WITH keyword_expanded AS (
    SELECT 
        store_name,
        product_name,
        UNNEST(string_split(negative_keywords, '|')) AS keyword
    FROM competitor_data
    WHERE negative_keywords IS NOT NULL 
      AND negative_keywords != ''
)
SELECT 
    keyword,
    COUNT(*) AS mention_count,
    COUNT(DISTINCT store_name) AS affected_stores
FROM keyword_expanded
GROUP BY keyword
ORDER BY mention_count DESC
LIMIT 15;

-- 按店铺统计每日平均价格变化
CREATE VIEW price_trend AS
SELECT 
    store_name,
    date,
    AVG(price_cny) AS avg_price,
    COUNT(DISTINCT product_name) AS product_count,
    MIN(price_cny) AS lowest_price,
    MAX(price_cny) AS highest_price
FROM clean_prices
GROUP BY store_name, date
ORDER BY store_name, date;

-- 计算周环比变化率
SELECT 
    store_name,
    date,
    avg_price,
    LAG(avg_price) OVER (PARTITION BY store_name ORDER BY date) AS prev_week_price,
    ROUND(
        (avg_price - LAG(avg_price) OVER (PARTITION BY store_name ORDER BY date)) 
        / NULLIF(LAG(avg_price) OVER (PARTITION BY store_name ORDER BY date), 0) 
        * 100, 2
    ) AS week_over_week_change_pct
FROM price_trend;

-- 识别首次出现的产品（可能是新品）
CREATE VIEW new_products AS
SELECT 
    store_name,
    product_name,
    MIN(date) AS first_seen_date,
    COUNT(DISTINCT date) AS days_active,
    AVG(price_cny) AS avg_price
FROM clean_prices
GROUP BY store_name, product_name
HAVING MIN(date) >= DATE '2026-07-01';  -- 本月新增

-- 按店铺统计上新频率
SELECT 
    store_name,
    COUNT(*) AS new_product_count,
    ROUND(AVG(avg_price), 2) AS avg_new_product_price,
    FIRST_VALUE(product_name) OVER (
        PARTITION BY store_name ORDER BY first_seen_date
    ) AS earliest_new_product
FROM new_products
GROUP BY store_name
ORDER BY new_product_count DESC;

-- 检测销量突增或突降的店铺
WITH daily_stats AS (
    SELECT 
        store_name,
        date,
        COALESCE(SUM(sales_volume), 0) AS daily_sales,
        AVG(SUM(sales_volume)) OVER (PARTITION BY store_name) AS moving_avg,
        STDDEV_SAMP(SUM(sales_volume)) OVER (PARTITION BY store_name) AS moving_std
    FROM competitor_data
    GROUP BY store_name, date
)
SELECT 
    store_name,
    date,
    daily_sales,
    ROUND(moving_avg, 0) AS baseline,
    ROUND((daily_sales - moving_avg) / NULLIF(moving_std, 0), 2) AS z_score
FROM daily_stats
WHERE ABS((daily_sales - moving_avg) / NULLIF(moving_std, 0)) > 2  -- Z-score > 2 为异常
ORDER BY ABS(z_score) DESC;

-- 生成竞争态势概览表
COPY (
    SELECT 
        store_name AS "店铺",
        COUNT(DISTINCT product_name) AS "在售商品数",
        ROUND(AVG(price_cny), 2) AS "平均价格",
        SUM(COALESCE(sales_volume, 0)) AS "总销量",
        COUNT(DISTINCT CASE WHEN date >= DATE '2026-07-20' THEN product_name END) AS "本周上新",
        STRING_AGG(DISTINCT keyword, ', ' ORDER BY mention_count DESC) AS "主要差评点"
    FROM (
        SELECT cp.*, ke.keyword, ke.mention_count
        FROM clean_prices cp
        LEFT JOIN (
            SELECT store_name, keyword, mention_count,
                   ROW_NUMBER() OVER (PARTITION BY store_name ORDER BY mention_count DESC) AS rn
            FROM (
                SELECT store_name, 
                       UNNEST(string_split(negative_keywords, '|')) AS keyword,
                       COUNT(*) OVER (PARTITION BY store_name, UNNEST(string_split(negative_keywords, '|'))) AS mention_count
                FROM competitor_data
                WHERE negative_keywords IS NOT NULL AND negative_keywords != ''
            ) sub
        ) ke ON cp.store_name = ke.store_name AND ke.rn <= 3
    ) combined
    GROUP BY store_name
) TO 'reports/weekly_competitor_summary.csv' (HEADER);

COPY (
    SELECT json_object_agg(store_name, json_build_object(
        'avg_price', avg_price,
        'new_products', new_product_count,
        'z_score_max', max_z_score
    ))
    FROM (
        SELECT store_name,
               ROUND(AVG(avg_price)::numeric, 2) AS avg_price,
               COUNT(*) FILTER (WHERE week_over_week_change_pct < -5) AS new_product_count,
               MAX(ABS(z_score)) AS max_z_score
        FROM (
            -- 上面定义的 price_trend 和 anomaly views
            SELECT * FROM price_trend
        ) pt
        LEFT JOIN (
            SELECT * FROM daily_stats WHERE ABS(z_score) > 2
        ) ds ON pt.store_name = ds.store_name
        GROUP BY store_name
    ) summary
) TO 'reports/summary.json';

#!/usr/bin/env python3
"""
电商竞品监控服务 - 一键脚本
用法: python3 monitor.py --category 宠物用品 --days 7
"""

import duckdb
import json
import sys
from datetime import datetime, timedelta

def main(category="宠物用品", days=7):
    con = duckdb.connect(f"monitor_{category}.db")
    
    # 1. 导入原始数据
    con.execute("""
        CREATE TABLE IF NOT EXISTS raw_data AS
        SELECT * FROM read_csv_auto(
            f'data/{category}/*.csv',
            RETURN_NULL_ON_ERROR=true
        )
    """)
    
    # 2. 清洗数据
    con.execute("""
        CREATE OR REPLACE TABLE clean_data AS
        SELECT 
            store_name,
            product_name,
            TRY_CAST(regexp_replace(price::VARCHAR, '[^0-9.]', '', 'g') AS DOUBLE) AS price,
            TRY_CAST(sales_volume AS BIGINT) AS sales,
            date
        FROM raw_data
    """)
    
    # 3. 生成竞品周报
    report = con.execute("""
        SELECT 
            store_name,
            COUNT(DISTINCT product_name) AS products,
            ROUND(AVG(price), 2) AS avg_price,
            SUM(COALESCE(sales, 0)) AS total_sales,
            COUNT(*) FILTER (WHERE date >= CURRENT_DATE - INTERVAL '{days}' DAY) AS recent_activity
        FROM clean_data
        GROUP BY store_name
        ORDER BY total_sales DESC
    """.format(days=days)).fetchall()
    
    # 4. 导出报告
    with open(f'reports/{category}_report_{datetime.now():%Y%m%d}.json', 'w') as f:
        json.dump(report, f, ensure_ascii=False, indent=2)
    
    print(f"✅ 报告已生成: reports/{category}_report_{datetime.now():%Y%m%d}.json")
    con.close()

if __name__ == "__main__":
    category = sys.argv[1] if len(sys.argv) > 1 else "宠物用品"
    days = int(sys.argv[2]) if len(sys.argv) > 2 else 7
    main(category, days)