Building an E-commerce Competitor Monitoring Service with DuckDB: From Data to Revenue

store_name,product_name,price,sales_volume,date,review_count,negative_keywords
"StoreA","Pet Snack Gift Pack",49.9,1200,2026-07-20,,praise
"StoreB","Pet Nutritional Paste",89.0,800,2026-07-20,15,slow shipping|damaged packaging

import duckdb

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

# Read CSV directly with error tolerance
con.execute("""
    CREATE TABLE competitor_data AS
    SELECT * FROM read_csv_auto('competitors/*.csv',
        RETURN_NULL_ON_ERROR=true,
        header=true,
        sep=','
    )
""")

# Check how many rows were imported, how many had dirty data
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"Total rows: {result[0]}, Valid prices: {result[1]}, Missing prices: {result[3]}")

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

-- Unify price format, remove currency symbols, convert to numeric
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;

-- Check cleaned data distribution
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;

-- Use DuckDB's fuzzy extension for name similarity calculation
CREATE EXTENSION fuzzy;

-- Find similar product names (to merge different representations of the same product)
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;

-- Split comma-separated negative keywords into individual rows and count
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;

-- Calculate daily average price changes per store
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;

-- Calculate week-over-week change rate
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;

-- Identify products that appeared for the first time (likely new arrivals)
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';  -- New this month

-- Count new product frequency per store
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;

-- Detect stores with sudden sales spikes or drops
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
ORDER BY ABS(z_score) DESC;

-- Generate competitive landscape summary table
COPY (
    SELECT 
        store_name AS "Store",
        COUNT(DISTINCT product_name) AS "Products",
        ROUND(AVG(price_cny), 2) AS "Avg Price",
        SUM(COALESCE(sales_volume, 0)) AS "Total Sales",
        COUNT(DISTINCT CASE WHEN date >= DATE '2026-07-20' THEN product_name END) AS "New This Week",
        STRING_AGG(DISTINCT keyword, ', ' ORDER BY mention_count DESC) AS "Top Complaints"
    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 (
            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
"""
E-commerce Competitor Monitoring Service - One-click Script
Usage: python3 monitor.py --category pet-supplies --days 7
"""

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

def main(category="pet_supplies", days=7):
    con = duckdb.connect(f"monitor_{category}.db")
    
    # 1. Import raw data
    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. Clean data
    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. Generate weekly competitor report
    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. Export report
    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"Report generated: reports/{category}_report_{datetime.now():%Y%m%d}.json")
    con.close()

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