DuckDB Multi-Platform E-commerce Data Integration System: Build a Automated Analytics Product from Scratch

import duckdb
import pandas as pd
from pathlib import Path
from datetime import datetime, timedelta
import random

# Generate simulated multi-platform order data
def generate_mock_data(output_dir: str = "data"):
 Path(output_dir).mkdir(exist_ok=True)
 
 # Taobao orders (CSV format)
 tb_data = []
 for i in range(200):
 days_ago = random.randint(0, 30)
 date = (datetime.now() - timedelta(days=days_ago)).strftime("%Y-%m-%d")
 tb_data.append({
 "order_id": f"tb_{random.randint(100000, 999999)}",
 "order_time": date,
 "product_name": random.choice(["Phone Case", "Data Cable", "Earphones", "Power Bank"]),
 "quantity": random.randint(1, 5),
 "amount": round(random.uniform(29, 299), 2),
 "platform": "taobao",
 "buyer_note": random.choice(["", "Ship ASAP", "Quality issue refund"])
 })
 pd.DataFrame(tb_data).to_csv(f"{output_dir}/taobao_orders.csv", index=False, encoding="utf-8-sig")
 
 # JD.com orders (Excel format, slightly different fields)
 jd_data = []
 for i in range(150):
 days_ago = random.randint(0, 30)
 date = (datetime.now() - timedelta(days=days_ago)).strftime("%Y-%m-%d")
 jd_data.append({
 "订单编号": f"JD{random.randint(10000000, 99999999)}",
 "下单时间": date,
 "商品名称": random.choice(["Phone Case", "Data Cable", "Earphones", "Power Bank", "Charger"]),
 "数量": random.randint(1, 3),
 "实付金额": round(random.uniform(39, 399), 2),
 "平台": "jd"
 })
 pd.DataFrame(jd_data).to_excel(f"{output_dir}/jd_orders.xlsx", index=False)
 
 # Pinduoduo orders (CSV, missing some fields)
 pdd_data = []
 for i in range(180):
 days_ago = random.randint(0, 30)
 date = (datetime.now() - timedelta(days=days_ago)).strftime("%Y-%m-%d")
 pdd_data.append({
 "订单号": f"PDD{random.randint(1000000, 9999999)}",
 "订单时间": date,
 "商品名称": random.choice(["Phone Case", "Data Cable", "Earphones", "Power Bank"]),
 "购买数量": random.randint(1, 10),
 "订单金额": round(random.uniform(19, 199), 2)
 })
 pd.DataFrame(pdd_data).to_csv(f"{output_dir}/pdd_orders.csv", index=False, encoding="utf-8-sig")
 
 print(f"✅ Data generated in {output_dir}/")

generate_mock_data()

import duckdb
from pathlib import Path

class MultiPlatformAnalyzer:
 """Multi-platform e-commerce data integration and analysis"""
 
 def __init__(self, data_dir: str = "data"):
 self.con = duckdb.connect("multi_platform.db")
 self.data_dir = Path(data_dir)
 self._setup_schema()
 
 def _setup_schema(self):
 """Create unified data model"""
 self.con.execute("""
 CREATE TABLE IF NOT EXISTS unified_orders (
 order_id VARCHAR,
 order_date DATE,
 product_name VARCHAR,
 quantity INT,
 amount DECIMAL(10,2),
 platform VARCHAR,
 buyer_note VARCHAR DEFAULT ''
 )
 """)
 
 def ingest_all_platforms(self):
 """Import all platform data at once, automatically handling field mapping"""
 
 # Taobao data: field names already unified, read directly
 self.con.execute("""
 CREATE OR REPLACE TEMP TABLE taobao_raw AS
 SELECT 
 order_id,
 order_time::DATE AS order_date,
 product_name,
 quantity,
 amount,
 'taobao' AS platform,
 COALESCE(buyer_note, '') AS buyer_note
 FROM read_csv_auto('data/taobao_orders.csv', header=true)
 """)
 
 # JD.com data: different field names, need mapping
 self.con.execute("""
 CREATE OR REPLACE TEMP TABLE jd_raw AS
 SELECT 
 "订单编号" AS order_id,
 "下单时间"::DATE AS order_date,
 "商品名称" AS product_name,
 "数量" AS quantity,
 "实付金额" AS amount,
 'jd' AS platform,
 '' AS buyer_note
 FROM read_excel('data/jd_orders.xlsx')
 """)
 
 # Pinduoduo data: missing buyer_note field, automatically fill empty
 self.con.execute("""
 CREATE OR REPLACE TEMP TABLE pdd_raw AS
 SELECT 
 "订单号" AS order_id,
 "订单时间"::DATE AS order_date,
 "商品名称" AS product_name,
 "购买数量" AS quantity,
 "订单金额" AS amount,
 'pdd' AS platform,
 '' AS buyer_note
 FROM read_csv_auto('data/pdd_orders.csv', header=true)
 """)
 
 # Merge all platform data
 self.con.execute("""
 INSERT INTO unified_orders
 SELECT order_id, order_date, product_name, quantity, amount, platform, buyer_note
 FROM taobao_raw
 UNION ALL
 SELECT order_id, order_date, product_name, quantity, amount, platform, buyer_note
 FROM jd_raw
 UNION ALL
 SELECT order_id, order_date, product_name, quantity, amount, platform, buyer_note
 FROM pdd_raw
 """)
 
 print(f"✅ Imported {self.con.execute('SELECT COUNT(*) FROM unified_orders').fetchone()[0]} orders")
 
 def generate_daily_report(self) -> pd.DataFrame:
 """Generate daily sales report"""
 return self.con.execute("""
 SELECT 
 order_date,
 COUNT(*) AS total_orders,
 SUM(quantity) AS total_items,
 ROUND(SUM(amount), 2) AS total_revenue,
 ROUND(AVG(amount), 2) AS avg_order_value,
 COUNT(CASE WHEN platform = 'taobao' THEN 1 END) AS taobao_orders,
 COUNT(CASE WHEN platform = 'jd' THEN 1 END) AS jd_orders,
 COUNT(CASE WHEN platform = 'pdd' THEN 1 END) AS pdd_orders,
 SUM(CASE WHEN platform = 'taobao' THEN amount ELSE 0 END) AS taobao_revenue,
 SUM(CASE WHEN platform = 'jd' THEN amount ELSE 0 END) AS jd_revenue,
 SUM(CASE WHEN platform = 'pdd' THEN amount ELSE 0 END) AS pdd_revenue
 FROM unified_orders
 WHERE order_date >= CURRENT_DATE - INTERVAL '7 days'
 GROUP BY order_date
 ORDER BY order_date DESC
 """).df()
 
 def generate_product_analysis(self) -> pd.DataFrame:
 """Product dimension analysis"""
 return self.con.execute("""
 SELECT 
 product_name,
 COUNT(*) AS order_count,
 SUM(quantity) AS total_sold,
 ROUND(SUM(amount), 2) AS total_revenue,
 ROUND(AVG(amount), 2) AS avg_order_value
 FROM unified_orders
 WHERE order_date >= CURRENT_DATE - INTERVAL '30 days'
 GROUP BY product_name
 ORDER BY total_revenue DESC
 LIMIT 10
 """).df()
 
 def generate_platform_comparison(self) -> pd.DataFrame:
 """Platform comparison analysis"""
 return self.con.execute("""
 SELECT 
 platform,
 COUNT(*) AS order_count,
 ROUND(SUM(amount), 2) AS total_revenue,
 ROUND(AVG(amount), 2) AS avg_order_value,
 ROUND(100.0 * COUNT(*) / SUM(COUNT(*)) OVER (), 2) AS order_share_pct,
 ROUND(100.0 * SUM(amount) / SUM(SUM(amount)) OVER (), 2) AS revenue_share_pct
 FROM unified_orders
 WHERE order_date >= CURRENT_DATE - INTERVAL '30 days'
 GROUP BY platform
 ORDER BY total_revenue DESC
 """).df()
 
 def export_report(self, output_dir: str = "reports"):
 """Export analysis reports"""
 from pathlib import Path
 Path(output_dir).mkdir(exist_ok=True)
 timestamp = datetime.now().strftime("%Y%m%d_%H%M")
 
 # Export daily report
 daily = self.generate_daily_report()
 daily.to_csv(f"{output_dir}/daily_report_{timestamp}.csv", index=False, encoding="utf-8-sig")
 
 # Export product analysis
 product = self.generate_product_analysis()
 product.to_csv(f"{output_dir}/product_analysis_{timestamp}.csv", index=False, encoding="utf-8-sig")
 
 # Export platform comparison
 platform = self.generate_platform_comparison()
 platform.to_csv(f"{output_dir}/platform_comparison_{timestamp}.csv", index=False, encoding="utf-8-sig")
 
 print(f"✅ Reports exported to {output_dir}/")
 return timestamp

# Usage example
if __name__ == "__main__":
 analyzer = MultiPlatformAnalyzer()
 analyzer.ingest_all_platforms()
 
 # Generate and export reports
 timestamp = analyzer.export_report()
 
 # View daily report
 print("\n=== Last 7 Days Sales Daily Report ===")
 print(analyzer.generate_daily_report().to_string(index=False))
 
 print("\n=== Platform Sales Comparison ===")
 print(analyzer.generate_platform_comparison().to_string(index=False))
 
 print("\n=== Top 5 Best-Selling Products ===")
 print(analyzer.generate_product_analysis().head().to_string(index=False))

-- Detect anomalous orders using IQR method
SELECT 
 order_id,
 amount,
 platform,
 CASE 
 WHEN amount > q3 + 1.5 * (q3 - q1) THEN 'high'
 WHEN amount < q1 - 1.5 * (q3 - q1) THEN 'low'
 ELSE 'normal'
 END AS anomaly_type
FROM (
 SELECT 
 *,
 QUANTILE_CONT(amount, 0.25) OVER (PARTITION BY platform) AS q1,
 QUANTILE_CONT(amount, 0.75) OVER (PARTITION BY platform) AS q3
 FROM unified_orders
 WHERE order_date >= CURRENT_DATE - INTERVAL '30 days'
)
WHERE anomaly_type != 'normal'

from jinja2 import Template

def generate_html_report(daily_df, platform_df, product_df):
 template = Template("""
 <!DOCTYPE html>
 <html>
 <head>
 <meta charset="utf-8">
 <title>E-commerce Daily Report</title>
 <style>
 body { font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", sans-serif; padding: 20px; }
 h1 { color: #333; }
 h2 { color: #555; margin-top: 30px; }
 table { border-collapse: collapse; width: 100%; margin-top: 10px; }
 th, td { border: 1px solid #ddd; padding: 8px; text-align: center; }
 th { background-color: #4CAF50; color: white; }
 tr:nth-child(even) { background-color: #f9f9f9; }
 </style>
 </head>
 <body>
 <h1>📊 E-commerce Daily Report - {{ today }}</h1>
 
 <h2>Sales Trend (Last 7 Days)</h2>
 <table>
 <tr>
 <th>Date</th><th>Orders</th><th>Revenue</th><th>Avg Order Value</th>
 </tr>
 {% for row in daily %}
 <tr>
 <td>{{ row.order_date }}</td>
 <td>{{ row.total_orders }}</td>
 <td>${{ "%.2f"|format(row.total_revenue) }}</td>
 <td>${{ "%.2f"|format(row.avg_order_value) }}</td>
 </tr>
 {% endfor %}
 </table>
 
 <h2>Platform Sales Comparison</h2>
 <table>
 <tr>
 <th>Platform</th><th>Orders</th><th>Revenue</th><th>Share</th>
 </tr>
 {% for row in platforms %}
 <tr>
 <td>{{ row.platform }}</td>
 <td>{{ row.order_count }}</td>
 <td>${{ "%.2f"|format(row.total_revenue) }}</td>
 <td>{{ "%.1f"|format(row.revenue_share_pct) }}%</td>
 </tr>
 {% endfor %}
 </table>
 
 <h2>Top 10 Best-Selling Products</h2>
 <table>
 <tr>
 <th>Product</th><th>Orders</th><th>Total Sold</th><th>Revenue</th>
 </tr>
 {% for row in products %}
 <tr>
 <td>{{ row.product_name }}</td>
 <td>{{ row.order_count }}</td>
 <td>{{ row.total_sold }}</td>
 <td>${{ "%.2f"|format(row.total_revenue) }}</td>
 </tr>
 {% endfor %}
 </table>
 </body>
 </html>
 """)
 
 html = template.render(
 today=datetime.now().strftime("%Y-%m-%d"),
 daily=daily_df.to_dict('records'),
 platforms=platform_df.to_dict('records'),
 products=product_df.to_dict('records')
 )
 
 return html

# Generate HTML report
daily_df = analyzer.generate_daily_report()
platform_df = analyzer.generate_platform_comparison()
product_df = analyzer.generate_product_analysis()

html = generate_html_report(daily_df, platform_df, product_df)

with open("reports/daily_report.html", "w", encoding="utf-8") as f:
 f.write(html)

print("✅ HTML report generated: reports/daily_report.html")

import time

# DuckDB
start = time.time()
result_duckdb = duckdb.query("SELECT * FROM unified_orders").df()
duckdb_time = time.time() - start

# Pandas
start = time.time()
result_pandas = pd.read_csv("unified_orders.csv")
pandas_time = time.time() - start

print(f"DuckDB: {duckdb_time:.3f}s")
print(f"Pandas: {pandas_time:.3f}s")
print(f"Speed improvement: {pandas_time/duckdb_time:.1f}x")

def incremental_update(self, new_csv_file: str):
 """Incremental update: only import new data"""
 # Get the date of the last order
 last_date = self.con.execute("""
 SELECT MAX(order_date) FROM unified_orders
 """).fetchone()[0]
 
 # Only read data from new dates
 self.con.execute(f"""
 CREATE OR REPLACE TEMP TABLE new_orders AS
 SELECT 
 order_id,
 order_time::DATE AS order_date,
 product_name,
 quantity,
 amount,
 'taobao' AS platform,
 COALESCE(buyer_note, '') AS buyer_note
 FROM read_csv_auto('{new_csv_file}', header=true)
 WHERE order_time::DATE > '{last_date}'
 """)
 
 # Incremental insert
 self.con.execute("""
 INSERT INTO unified_orders
 SELECT * FROM new_orders
 ON CONFLICT (order_id) DO NOTHING
 """)
 
 print(f"✅ Incremental import complete, added {self.con.execute('SELECT COUNT(*) FROM new_orders').fetchone()[0]} records")

def create_materialized_views(self):
 """Create materialized tables to accelerate common queries"""
 
 # Daily sales summary view
 self.con.execute("""
 CREATE materialized table IF NOT EXISTS mv_daily_sales AS
 SELECT 
 order_date,
 COUNT(*) AS total_orders,
 SUM(quantity) AS total_items,
 ROUND(SUM(amount), 2) AS total_revenue,
 ROUND(AVG(amount), 2) AS avg_order_value
 FROM unified_orders
 GROUP BY order_date
 """)
 
 # Platform sales summary view
 self.con.execute("""
 CREATE materialized table IF NOT EXISTS mv_platform_sales AS
 SELECT 
 platform,
 COUNT(*) AS order_count,
 ROUND(SUM(amount), 2) AS total_revenue,
 ROUND(AVG(amount), 2) AS avg_order_value
 FROM unified_orders
 GROUP BY platform
 """)
 
 print("✅ materialized tables created")

# Use materialized tables for faster queries
def get_daily_report_fast(self) -> pd.DataFrame:
 return self.con.execute("""
 SELECT * FROM mv_daily_sales
 WHERE order_date >= CURRENT_DATE - INTERVAL '7 days'
 ORDER BY order_date DESC
 """).df()

# Automatically run at 8 AM every day
0 8 * * * cd /path/to/project && python3 ecommerce_analyzer.py >> /var/log/ecommerce.log 2>&1

from duckdb_cron import CronJob

# Create scheduled task
CronJob.create(
 name="daily_ecommerce_report",
 schedule="0 8 * * *",
 command="python3 generate_report.py",
 timezone="Asia/Shanghai"
)

# View all scheduled tasks
CronJob.list_all()

# Remove scheduled task
CronJob.remove("daily_ecommerce_report")