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)