pip install duckdb dbt-duckdb openpyxl pandas

# Verify dbt installation
dbt --version
# Core: 1.11.x, Plugin: duckdb 1.10.x

#!/usr/bin/env python3
"""Generate sample e-commerce data for dbt + DuckDB demo."""
import csv, random, os
from datetime import datetime, timedelta

random.seed(42)
OUTPUT_DIR = os.path.dirname(os.path.abspath(__file__))

NUM_CUSTOMERS = 200
NUM_PRODUCTS = 50
NUM_ORDERS = 2000
NUM_REVIEWS = 1500
START_DATE = datetime(2025, 1, 1)
END_DATE = datetime(2026, 5, 1)

def gen_customers():
    cities = ["Beijing", "Shanghai", "Guangzhou", "Shenzhen", "Hangzhou",
              "Chengdu", "Wuhan", "Nanjing", "Chongqing", "Xi'an"]
    levels = ["Regular", "Silver", "Gold", "Diamond"]
    channels = ["Direct", "Search Engine", "Social Media", "Email", "Ads"]
    rows = []
    for i in range(1, NUM_CUSTOMERS + 1):
        reg_date = START_DATE + timedelta(days=random.randint(0, 400))
        rows.append({"customer_id": i, "name": f"User_{i:04d}",
            "city": random.choice(cities),
            "level": random.choices(levels, weights=[50,30,15,5])[0],
            "channel": random.choice(channels),
            "registration_date": reg_date.strftime("%Y-%m-%d"),
            "is_active": 1 if random.random() > 0.15 else 0})
    return rows

def gen_products():
    categories = ["Electronics", "Clothing", "Food", "Home", "Beauty", "Books"]
    suppliers = ["Supplier_A", "Supplier_B", "Supplier_C", "Supplier_D", "Supplier_E"]
    rows = []
    for i in range(1, NUM_PRODUCTS + 1):
        cost = round(random.uniform(10, 500), 2)
        price = round(cost * random.uniform(1.3, 3.0), 2)
        rows.append({"product_id": i, "product_name": f"Product_{i:04d}",
            "category": random.choice(categories), "supplier": random.choice(suppliers),
            "cost": cost, "price": price, "stock": random.randint(0, 1000),
            "shelf_date": (START_DATE + timedelta(days=random.randint(0, 480))).strftime("%Y-%m-%d")})
    return rows

def gen_orders():
    statuses = ["Completed", "Shipped", "Cancelled", "Refunding"]
    payments = ["WeChat Pay", "Alipay", "Bank Card", "COD"]
    rows = []
    for i in range(1, NUM_ORDERS + 1):
        order_date = START_DATE + timedelta(days=random.randint(0, (END_DATE - START_DATE).days - 1))
        quantity = random.randint(1, 5)
        unit_price = round(random.uniform(20, 800), 2)
        rows.append({"order_id": i, "customer_id": random.randint(1, NUM_CUSTOMERS),
            "product_id": random.randint(1, NUM_PRODUCTS),
            "order_date": order_date.strftime("%Y-%m-%d %H:%M:%S"),
            "quantity": quantity, "unit_price": unit_price,
            "total_amount": round(unit_price * quantity, 2),
            "status": random.choices(statuses, weights=[60,20,15,5])[0],
            "payment_method": random.choice(payments)})
    return rows

def gen_reviews():
    rows = []
    for i in range(1, NUM_REVIEWS + 1):
        review_date = START_DATE + timedelta(days=random.randint(0, (END_DATE - START_DATE).days - 1))
        rows.append({"review_id": i, "order_id": random.randint(1, NUM_ORDERS),
            "product_id": random.randint(1, NUM_PRODUCTS),
            "customer_id": random.randint(1, NUM_CUSTOMERS),
            "rating": random.choices([5,4,3,2,1], weights=[40,30,15,10,5])[0],
            "review_date": review_date.strftime("%Y-%m-%d"),
            "is_verified_purchase": 1 if random.random() > 0.3 else 0})
    return rows

# Write CSV files
os.makedirs(os.path.join(OUTPUT_DIR, "seeds"), exist_ok=True)
for name, gen_fn in [("customers", gen_customers), ("products", gen_products),
                     ("orders", gen_orders), ("reviews", gen_reviews)]:
    rows = gen_fn()
    path = os.path.join(OUTPUT_DIR, "seeds", f"{name}.csv")
    with open(path, "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=rows[0].keys())
        writer.writeheader(); writer.writerows(rows)
    print(f"✅ Generated {path} ({len(rows)} rows)")

name: 'duckdb_shop'
version: '1.0.0'
config-version: 2
profile: 'duckdb_shop'

model-paths: ["models"]
seed-paths: ["seeds"]
test-paths: ["tests"]

models:
  duckdb_shop:
    staging:
      +materialized: view     # Cleaning layer — views save disk space
      +schema: staging
    marts:
      +materialized: table    # Analytics layer — tables for speed
      +schema: marts

seeds:
  duckdb_shop:
    +schema: raw

duckdb_shop:
  target: dev
  outputs:
    dev:
      type: duckdb
      path: duckdb_shop.duckdb
      schema: main
      threads: 4

-- Clean customer data: standardize fields and cast types
with source as (
    select * from {{ ref('customers') }}
),
cleaned as (
    select
        customer_id, name as customer_name, city,
        case
            when level in ('Regular', 'Silver', 'Gold', 'Diamond') then level
            else 'Regular'
        end as customer_level,
        channel as acquisition_channel,
        registration_date::date as registration_date,
        is_active::boolean as is_active,
        current_timestamp as loaded_at
    from source
)
select * from cleaned

-- Clean orders: parse timestamps, add derived fields
with source as (
    select * from {{ ref('orders') }}
),
cleaned as (
    select
        order_id, customer_id, product_id,
        order_date::timestamp as order_timestamp,
        order_date::date as order_date,
        strftime(order_date::timestamp, '%Y') as order_year,
        strftime(order_date::timestamp, '%m') as order_month,
        strftime(order_date::timestamp, '%Y-%m') as order_year_month,
        strftime(order_date::timestamp, '%u') as order_week,
        quantity, unit_price, total_amount,
        status as order_status, payment_method,
        case
            when status in ('Completed', 'Shipped') then 'valid'
            else 'invalid'
        end as is_valid_order,
        current_timestamp as loaded_at
    from source
)
select * from cleaned

-- Clean products: compute gross margin, stock status
with source as (
    select * from {{ ref('products') }}
),
cleaned as (
    select
        product_id, product_name, category, supplier,
        cost, price,
        round((price - cost) / nullif(price, 0) * 100, 2) as gross_margin_pct,
        stock, shelf_date::date as shelf_date,
        case
            when stock = 0 then 'Out of Stock'
            when stock < 50 then 'Low Stock'
            when stock < 200 then 'Normal'
            else 'Well Stocked'
        end as stock_status,
        current_timestamp as loaded_at
    from source
)
select * from cleaned

-- RFM customer segmentation: find high-value customers
with orders as (
    select * from {{ ref('stg_orders') }}
    where is_valid_order = 'valid'
),
customers as (
    select * from {{ ref('stg_customers') }}
    where is_active = true
),
customer_metrics as (
    select
        c.customer_id, c.customer_name, c.city,
        c.customer_level, c.acquisition_channel, c.registration_date,
        count(distinct o.order_id) as total_orders,
        sum(o.total_amount) as total_spent,
        avg(o.total_amount) as avg_order_value,
        max(o.order_date) as last_order_date,
        min(o.order_date) as first_order_date,
        datediff('day', max(o.order_date), current_date) as days_since_last_order,
        count(distinct o.product_id) as unique_products_bought
    from customers c
    left join orders o on c.customer_id = o.customer_id
    group by 1, 2, 3, 4, 5, 6
),
rfm as (
    select *,
        -- Recency (1-5)
        case when days_since_last_order <= 30 then 5
             when days_since_last_order <= 90 then 4
             when days_since_last_order <= 180 then 3
             when days_since_last_order <= 365 then 2
             else 1 end as r_score,
        -- Frequency (1-5)
        case when total_orders >= 10 then 5
             when total_orders >= 6 then 4
             when total_orders >= 3 then 3
             when total_orders >= 1 then 2
             else 1 end as f_score,
        -- Monetary (1-5)
        case when total_spent >= 10000 then 5
             when total_spent >= 5000 then 4
             when total_spent >= 2000 then 3
             when total_spent >= 500 then 2
             else 1 end as m_score
    from customer_metrics
)
select *,
    r_score + f_score + m_score as rfm_total,
    case
        when (r_score >= 4 and f_score >= 4 and m_score >= 4) then '⭐ VIP Customer'
        when (r_score >= 4 and f_score >= 4 and m_score >= 2) then 'Growth Customer'
        when (r_score >= 3 and f_score >= 3) then 'Standard Customer'
        when (r_score >= 1 and total_orders > 0) then 'At-Risk Customer'
        else 'Silent Customer'
    end as customer_segment
from rfm order by rfm_total desc

-- Core KPIs: one query to generate the management dashboard
with daily_sales as (select * from {{ ref('daily_sales_summary') }}),
orders as (select * from {{ ref('stg_orders') }}),
customer_metrics as (select * from {{ ref('customer_analytics') }})

select 'Total Revenue' as metric_name,
    round(sum(total_revenue), 2) as metric_value, 'CNY' as unit
from daily_sales
union all
select 'Total Orders', count(*), 'orders'
from orders where is_valid_order = 'valid'
union all
select 'Avg Order Value', round(avg(total_amount), 2), 'CNY'
from orders where is_valid_order = 'valid'
union all
select 'Active Customers', count(*), 'people'
from customer_metrics where total_orders > 0
union all
select 'VIP Customers', count(*), 'people'
from customer_metrics where customer_segment = '⭐ VIP Customer'
union all
select 'Avg RFM Score', round(avg(rfm_total), 2), 'points'
from customer_metrics
union all
select 'Gross Margin',
    round(sum(total_profit) / nullif(sum(total_revenue), 0) * 100, 2), '%'
from {{ ref('product_performance') }}
order by metric_name

#!/usr/bin/env python3
"""day24_run_all.py — One-click data modeling + report export"""
import subprocess, duckdb, pandas as pd
from pathlib import Path

PROJECT_DIR = Path("day24_dbt_project")
DB_PATH = PROJECT_DIR / "duckdb_shop.duckdb"

# 1. Run dbt seed (import CSVs into DuckDB)
print("📥 Importing CSV data...")
subprocess.run(["dbt", "seed", "--profiles-dir", str(PROJECT_DIR)],
               cwd=PROJECT_DIR, check=True)

# 2. Run dbt run (execute all models)
print("🔨 Running dbt models...")
subprocess.run(["dbt", "run", "--profiles-dir", str(PROJECT_DIR)],
               cwd=PROJECT_DIR, check=True)

# 3. Connect to DuckDB and export reports
conn = duckdb.connect(str(DB_PATH))

# KPI Dashboard
kpi = conn.execute("""
    SELECT metric_name, metric_value, unit
    FROM main_marts.kpi_dashboard
    WHERE metric_name IN ('Total Revenue','Total Orders',
          'Avg Order Value','Active Customers','Gross Margin')
""").fetchdf()
print("\n📈 KPI Dashboard:")
print(kpi.to_string(index=False))

# Top 10 Products
top_products = conn.execute("""
    SELECT product_name, category, units_sold,
           total_revenue, gross_margin_pct
    FROM main_marts.product_performance
    WHERE units_sold > 0
    ORDER BY total_revenue DESC LIMIT 10
""").fetchdf()

# Customer Segments
segments = conn.execute("""
    SELECT customer_segment, count(*) as cnt,
           round(avg(total_spent),2) as avg_spent
    FROM main_marts.customer_analytics
    GROUP BY customer_segment
""").fetchdf()

# Export to Excel
with pd.ExcelWriter("day24_dbt_report.xlsx", engine='openpyxl') as writer:
    kpi.to_excel(writer, sheet_name='KPI Dashboard', index=False)
    top_products.to_excel(writer, sheet_name='Top Products', index=False)
    segments.to_excel(writer, sheet_name='Customer Segments', index=False)

print(f"\n✅ Report exported: day24_dbt_report.xlsx")
conn.close()