diff --git a/.github/workflows/dbt_ci.yml b/.github/workflows/dbt_ci.yml new file mode 100644 index 0000000..6f5abf4 --- /dev/null +++ b/.github/workflows/dbt_ci.yml @@ -0,0 +1,45 @@ +name: dbt CI + +on: + push: + branches: [dev] + pull_request: + branches: [main] + +jobs: + dbt-build: + name: Generate data → load → dbt build + runs-on: ubuntu-latest + + steps: + - name: Checkout repo + uses: actions/checkout@v4 + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: "3.12" + cache: "pip" + + - name: Install dependencies + run: pip install dbt-duckdb faker plotly streamlit + + - name: Generate source CSVs + run: | + python generator/generate.py + python scripts/generate_changes.py + + - name: Load raw tables into DuckDB + run: python scripts/load_raw.py + + - name: Run dbt build (models + tests) + working-directory: ecom + run: dbt build --profiles-dir . + + - name: Upload DuckDB artifact + uses: actions/upload-artifact@v4 + if: always() + with: + name: ecom-duckdb + path: data/ecom.duckdb + retention-days: 3 diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..e515a19 --- /dev/null +++ b/.gitignore @@ -0,0 +1,11 @@ +data/raw/ +data/processed/ +*.duckdb +__pycache__/ +.env +# dbt artifacts +ecom/target/ +ecom/dbt_packages/ +ecom/logs/ +ecom/.user.ymllogs/ +logs/ \ No newline at end of file diff --git a/README.md b/README.md index 5f28b19..a62f838 100644 --- a/README.md +++ b/README.md @@ -1 +1,227 @@ -# e-commerce-analytics \ No newline at end of file +# E-Commerce Analytics Engineering Project + +![dbt CI](https://github.com/b10000-coder/e-commerce-analytics/actions/workflows/dbt_ci.yml/badge.svg?branch=dev) +![Python](https://img.shields.io/badge/python-3.12-blue) +![dbt](https://img.shields.io/badge/dbt-1.11-orange) +![DuckDB](https://img.shields.io/badge/DuckDB-1.10-yellow) + +A end-to-end analytics engineering project built on a **Trendyol-style e-commerce domain** — from raw synthetic data through a fully tested dbt star schema to a live Streamlit dashboard. Demonstrates modern data warehousing practices including SCD Type 2, CI/CD, and metric-layer design. + +--- + +## Architecture + +``` +generate.py generate_changes.py + │ │ + ▼ ▼ +data/raw/ ──load_raw.py──▶ DuckDB (raw schema) + │ + dbt build + │ + ┌─────────────┼─────────────┐ + ▼ ▼ ▼ + stg_* views dim_* tables fact_orders + (staging) (marts) (marts) + │ + Streamlit + dashboard +``` + +--- + +## Tech Stack + +| Layer | Tool | +|---|---| +| Data generation | Python · Faker | +| Database | DuckDB | +| Transformation | dbt-duckdb | +| Testing | dbt schema tests | +| Dashboard | Streamlit · Plotly | +| CI/CD | GitHub Actions | + +--- + +## Data Model + +### Star schema (`main_marts`) + +``` + dim_date + (182 rows) + │ + ┌───────────────┼───────────────┐ + │ │ │ + dim_customer fact_orders dim_product + dim_customer (4,531 rows) (400 rows) + _scd2 │ + (SCD Type 2) │ + dim_seller + (80 rows) +``` + +**Fact table grain:** one row per order line item — the most granular level, allowing roll-up to order, customer, product, category, or day. + +### Staging layer (`main_staging`) + +| Model | Source | Key transformations | +|---|---|---| +| `stg_customers` | `raw.customers` | Deduplicates re-registrations, removes soft-deletes | +| `stg_orders` | `raw.orders` | Parses two date formats (`ISO` + `dd/mm/yyyy`) via `TRY_STRPTIME` | +| `stg_order_items` | `raw.order_items` | Casts types, derives `line_revenue` and `is_flash_sale` | +| `stg_products` | `raw.products` | Derives `gross_margin_pct` | +| `stg_sellers` | `raw.sellers` | Type casting | +| `stg_sessions` | `raw.sessions` | Preserves anonymous sessions (null `customer_id`) | +| `stg_customer_changes` | `raw.customer_changes` | Cleans change events for SCD2 | + +### SCD Type 2 — `dim_customer_scd2` + +Tracks historical changes to customer `city` and `customer_tier`. Each version of a customer gets a separate row stamped with `valid_from` / `valid_to` dates, enabling point-in-time joins to the fact table. + +```sql +-- Get the customer's attributes at the time of their order +JOIN dim_customer_scd2 scd + ON fact.customer_id = scd.customer_id + AND fact.order_date >= scd.valid_from + AND (fact.order_date < scd.valid_to OR scd.valid_to IS NULL) +``` + +--- + +## Project Structure + +``` +e-commerce-analytics/ +├── .github/ +│ └── workflows/ +│ └── dbt_ci.yml CI: full pipeline on every push + PR +├── generator/ +│ └── generate.py Synthetic e-commerce data (Faker, seed=42) +├── scripts/ +│ ├── load_raw.py Loads CSVs into DuckDB raw schema +│ └── generate_changes.py Customer attribute change events (SCD2 source) +├── ecom/ dbt project +│ ├── dbt_project.yml +│ ├── profiles.yml +│ └── models/ +│ ├── staging/ 7 stg_* views — clean, typed, tested +│ └── marts/ Star schema — 5 tables (4 dims + 1 fact) +│ ├── dim_date.sql +│ ├── dim_customer.sql +│ ├── dim_customer_scd2.sql +│ ├── dim_product.sql +│ ├── dim_seller.sql +│ └── fact_orders.sql +├── dashboard/ +│ └── app.py Streamlit BI dashboard +├── data/ +│ └── raw/ Generated CSVs (git-ignored) +├── requirements.txt +└── .gitignore +``` + +--- + +## Quick Start + +### 1. Clone and install + +```bash +git clone https://github.com/b10000-coder/e-commerce-analytics.git +cd e-commerce-analytics +python -m venv .venv && source .venv/bin/activate +pip install -r requirements.txt +``` + +### 2. Generate data + +```bash +python generator/generate.py # creates 6 CSVs in data/raw/ +python scripts/generate_changes.py # creates customer_changes.csv +``` + +### 3. Load into DuckDB + +```bash +python scripts/load_raw.py +``` + +Output: +``` +OK raw.customers 618 rows +OK raw.sellers 80 rows +OK raw.products 400 rows +OK raw.orders 2,500 rows +OK raw.order_items 4,881 rows +OK raw.sessions 4,000 rows +OK raw.customer_changes 142 rows +``` + +### 4. Run dbt (models + tests) + +```bash +cd ecom +dbt build --profiles-dir . +``` + +Expected output: +``` +Done. PASS=78 WARN=0 ERROR=0 SKIP=0 NO-OP=0 TOTAL=78 +``` + +### 5. Launch the dashboard + +```bash +cd .. +streamlit run dashboard/app.py +``` + +Opens at `http://localhost:8501` + +--- + +## dbt Test Coverage + +78 tests across all models covering: + +- `unique` and `not_null` on all primary and surrogate keys +- `accepted_values` on status, tier, platform, and device_type columns +- `relationships` tests enforcing referential integrity across all foreign keys +- Point-in-time join integrity for SCD Type 2 + +--- + +## CI/CD + +Every push to `dev` and every PR to `main` triggers the full pipeline on a clean GitHub Actions runner: + +``` +checkout → install deps → generate data → load DuckDB → dbt build (78 tests) +``` + +Total runtime: ~46 seconds. The compiled `ecom.duckdb` is uploaded as a workflow artifact. + +--- + +## Domain + +Trendyol-style Turkish e-commerce — 6 months of data (Oct 2024 → Mar 2025): + +- **8 product categories** — Electronics, Fashion, Home & Life, Beauty, Sports, Books, Toys +- **80 sellers** across major Turkish cities +- **~600 customers** with bronze / silver / gold tiers +- **2,500 orders** / **4,881 line items** across iOS, Android, and web +- **4,000 sessions** including 20% anonymous (null customer_id) +- **Intentional data quality issues** — mixed date formats, duplicate customers, orphaned foreign keys — resolved in the staging layer + +--- + +## Key Concepts Demonstrated + +- **Layered architecture** — raw → staging → marts with clear separation of concerns +- **Star schema** — fact + dimension tables with surrogate keys +- **SCD Type 2** — full customer attribute history via `LEAD()` window function +- **dbt best practices** — sources, refs, schema tests, documented models +- **CI/CD for data** — reproducible pipeline from raw data to tested marts in one command +- **Metric layer** — GMV, AOV, gross margin, flash sale rate defined once in SQL diff --git a/dashboard/app.py b/dashboard/app.py new file mode 100644 index 0000000..3551064 --- /dev/null +++ b/dashboard/app.py @@ -0,0 +1,286 @@ +""" +E-Commerce Analytics Dashboard +------------------------------- +Runs on top of the dbt marts layer in ecom.duckdb. +Every chart is a direct SQL query against main_marts — no pandas wrangling. + +Usage (from repo root): + streamlit run dashboard/app.py +""" + +import os +import duckdb +import streamlit as st +import plotly.express as px + +# ── connection ─────────────────────────────────────────────────────────────── +DB_PATH = os.path.join(os.path.dirname(__file__), "..", "data", "ecom.duckdb") + +@st.cache_resource +def get_con(): + return duckdb.connect(DB_PATH, read_only=True) + +con = get_con() + +def q(sql): + """Run a SQL query and return a pandas DataFrame.""" + return con.execute(sql).df() + + +# ── page config ────────────────────────────────────────────────────────────── +st.set_page_config( + page_title="E-Commerce Analytics", + layout="wide", + initial_sidebar_state="collapsed", +) + +st.title("E-Commerce Analytics") +st.caption("dbt · DuckDB · Star schema — Oct 2024 → Mar 2025") + +# ── sidebar filters ────────────────────────────────────────────────────────── +with st.sidebar: + st.header("Filters") + status_filter = st.multiselect( + "Order status", + options=["delivered", "shipped", "processing", "pending", "cancelled", "returned"], + default=["delivered"], + ) + platform_filter = st.multiselect( + "Platform", + options=["ios", "android", "web"], + default=["ios", "android", "web"], + ) + +status_in = "(" + ",".join(f"'{s}'" for s in status_filter) + ")" if status_filter else "('delivered')" +platform_in = "(" + ",".join(f"'{p}'" for p in platform_filter) + ")" if platform_filter else "('ios','android','web')" + + +# ── KPI row ────────────────────────────────────────────────────────────────── +kpis = q(f""" + SELECT + ROUND(SUM(line_revenue), 0) AS gmv, + COUNT(DISTINCT order_id) AS orders, + ROUND(SUM(line_revenue) / COUNT(DISTINCT order_id), 2) AS aov, + COUNT(DISTINCT customer_key) AS customers, + ROUND(AVG(CASE WHEN is_flash_sale THEN 1.0 ELSE 0.0 END) * 100, 1) AS flash_pct + FROM main_marts.fact_orders + WHERE order_status IN {status_in} + AND platform IN {platform_in} +""") + +c1, c2, c3, c4, c5 = st.columns(5) +c1.metric("GMV", f"₺{int(kpis['gmv'][0]):,}") +c2.metric("Orders", f"{int(kpis['orders'][0]):,}") +c3.metric("AOV", f"₺{kpis['aov'][0]:,.0f}") +c4.metric("Customers", f"{int(kpis['customers'][0]):,}") +c5.metric("Flash sale rate", f"{kpis['flash_pct'][0]}%") + +st.divider() + +# ── row 1: monthly GMV ──────────────────────────────────────────────────────── +st.subheader("Monthly GMV") +monthly = q(f""" + SELECT + d.year, + d.month_number, + d.month_name, + ROUND(SUM(f.line_revenue), 0) AS gmv, + COUNT(DISTINCT f.order_id) AS orders + FROM main_marts.fact_orders f + JOIN main_marts.dim_date d ON f.date_key = d.date_key + WHERE f.order_status IN {status_in} + AND f.platform IN {platform_in} + GROUP BY d.year, d.month_number, d.month_name + ORDER BY d.year, d.month_number +""") +monthly["label"] = monthly["month_name"] + " " + monthly["year"].astype(str) + +fig = px.bar( + monthly, x="label", y="gmv", + labels={"label": "", "gmv": "GMV (₺)"}, + color_discrete_sequence=["#1D9E75"], + text_auto=".3s", +) +fig.update_traces(textposition="outside") +fig.update_layout(plot_bgcolor="rgba(0,0,0,0)", yaxis_gridcolor="rgba(0,0,0,0.06)") +st.plotly_chart(fig, use_container_width=True) + +st.divider() + +# ── row 2: category + order status ─────────────────────────────────────────── +col1, col2 = st.columns(2) + +with col1: + st.subheader("Revenue by category") + cat = q(f""" + SELECT p.category, ROUND(SUM(f.line_revenue), 0) AS revenue + FROM main_marts.fact_orders f + JOIN main_marts.dim_product p ON f.product_key = p.product_key + WHERE f.order_status IN {status_in} + AND f.platform IN {platform_in} + GROUP BY p.category + ORDER BY revenue DESC + """) + fig = px.bar( + cat, x="revenue", y="category", orientation="h", + labels={"revenue": "Revenue (₺)", "category": ""}, + color_discrete_sequence=["#534AB7"], + ) + fig.update_layout( + yaxis={"categoryorder": "total ascending"}, + plot_bgcolor="rgba(0,0,0,0)", + ) + st.plotly_chart(fig, use_container_width=True) + +with col2: + st.subheader("Orders by status") + status_df = q(""" + SELECT order_status, COUNT(DISTINCT order_id) AS orders + FROM main_marts.fact_orders + GROUP BY order_status ORDER BY orders DESC + """) + fig = px.pie( + status_df, values="orders", names="order_status", hole=0.45, + color_discrete_sequence=px.colors.qualitative.Set2, + ) + fig.update_traces(textposition="inside", textinfo="percent+label") + st.plotly_chart(fig, use_container_width=True) + +st.divider() + +# ── row 3: platform + customer tier ────────────────────────────────────────── +col1, col2 = st.columns(2) + +with col1: + st.subheader("Revenue by platform") + plat = q(f""" + SELECT platform, ROUND(SUM(line_revenue), 0) AS revenue + FROM main_marts.fact_orders + WHERE order_status IN {status_in} + GROUP BY platform ORDER BY revenue DESC + """) + fig = px.bar( + plat, x="platform", y="revenue", + labels={"platform": "", "revenue": "Revenue (₺)"}, + color_discrete_sequence=["#D85A30"], + text_auto=".3s", + ) + fig.update_layout(plot_bgcolor="rgba(0,0,0,0)") + st.plotly_chart(fig, use_container_width=True) + +with col2: + st.subheader("Customer tier distribution") + tier = q(f""" + SELECT c.customer_tier, COUNT(DISTINCT f.customer_key) AS customers + FROM main_marts.fact_orders f + JOIN main_marts.dim_customer c ON f.customer_key = c.customer_key + WHERE f.order_status IN {status_in} + GROUP BY c.customer_tier + """) + fig = px.pie( + tier, values="customers", names="customer_tier", hole=0.45, + color_discrete_sequence=["#EF9F27", "#1D9E75", "#7F77DD"], + ) + fig.update_traces(textposition="inside", textinfo="percent+label") + st.plotly_chart(fig, use_container_width=True) + +st.divider() + +# ── row 4: flash sale vs regular ───────────────────────────────────────────── +col1, col2 = st.columns(2) + +with col1: + st.subheader("Flash sale vs regular revenue") + flash = q(f""" + SELECT + CASE WHEN is_flash_sale THEN 'Flash sale' ELSE 'Regular price' END AS type, + ROUND(SUM(line_revenue), 0) AS revenue, + COUNT(*) AS items + FROM main_marts.fact_orders + WHERE order_status IN {status_in} + GROUP BY is_flash_sale + """) + fig = px.bar( + flash, x="type", y="revenue", color="type", + labels={"type": "", "revenue": "Revenue (₺)"}, + color_discrete_sequence=["#D85A30", "#1D9E75"], + text_auto=".3s", + ) + fig.update_layout(showlegend=False, plot_bgcolor="rgba(0,0,0,0)") + st.plotly_chart(fig, use_container_width=True) + +with col2: + st.subheader("Gross margin by category") + margin = q(""" + SELECT p.category, + ROUND(AVG(p.gross_margin_pct) * 100, 1) AS avg_margin_pct + FROM main_marts.dim_product p + WHERE p.is_active = true + GROUP BY p.category + ORDER BY avg_margin_pct DESC + """) + fig = px.bar( + margin, x="avg_margin_pct", y="category", orientation="h", + labels={"avg_margin_pct": "Avg gross margin %", "category": ""}, + color_discrete_sequence=["#0F6E56"], + ) + fig.update_layout( + yaxis={"categoryorder": "total ascending"}, + plot_bgcolor="rgba(0,0,0,0)", + ) + st.plotly_chart(fig, use_container_width=True) + +st.divider() + +# ── SCD2 spotlight ──────────────────────────────────────────────────────────── +st.subheader("SCD Type 2 — customer history") +st.caption("Customers with more than one version (city or tier change during the period)") + +scd = q(""" + SELECT + customer_id, + full_name, + version_number AS version, + city, + customer_tier AS tier, + valid_from, + valid_to, + is_current + FROM main_marts.dim_customer_scd2 + WHERE customer_id IN ( + SELECT customer_id FROM main_marts.dim_customer_scd2 + GROUP BY customer_id HAVING COUNT(*) > 1 + LIMIT 8 + ) + ORDER BY customer_id, version_number +""") +scd["valid_to"] = scd["valid_to"].fillna("active") +st.dataframe( + scd, + use_container_width=True, + column_config={ + "is_current": st.column_config.CheckboxColumn("current?"), + "valid_from": "valid from", + "valid_to": "valid to", + }, + hide_index=True, +) + +st.divider() + +# ── top 10 sellers ──────────────────────────────────────────────────────────── +st.subheader("Top 10 sellers by revenue") +sellers = q(f""" + SELECT + s.seller_name, + s.category, + ROUND(SUM(f.line_revenue), 0) AS revenue, + COUNT(DISTINCT f.order_id) AS orders, + ROUND(AVG(s.rating), 1) AS rating + FROM main_marts.fact_orders f + JOIN main_marts.dim_seller s ON f.seller_key = s.seller_key + WHERE f.order_status IN {status_in} + GROUP BY s.seller_name, s.category, s.rating + ORDER BY revenue DESC LIMIT 10 +""") +st.dataframe(sellers, use_container_width=True, hide_index=True) diff --git a/ecom/.user.yml b/ecom/.user.yml new file mode 100644 index 0000000..21c1fd0 --- /dev/null +++ b/ecom/.user.yml @@ -0,0 +1 @@ +id: ffdb83f1-7624-4929-a57f-3814645ccb07 diff --git a/ecom/dbt_project.yml b/ecom/dbt_project.yml new file mode 100644 index 0000000..b0a753c --- /dev/null +++ b/ecom/dbt_project.yml @@ -0,0 +1,22 @@ +name: ecom +version: '1.0.0' +config-version: 2 + +profile: ecom + +model-paths: ["models"] +test-paths: ["tests"] +macro-paths: ["macros"] +snapshot-paths: ["snapshots"] + +target-path: "target" +clean-targets: ["target", "dbt_packages"] + +models: + ecom: + staging: + +materialized: view + +schema: staging + marts: + +materialized: table + +schema: marts diff --git a/ecom/models/marts/_schema.yml b/ecom/models/marts/_schema.yml new file mode 100644 index 0000000..2067c45 --- /dev/null +++ b/ecom/models/marts/_schema.yml @@ -0,0 +1,102 @@ +version: 2 + +models: + - name: dim_date + description: "Generated date dimension. date_key (YYYYMMDD integer) is the FK used in fact_orders." + columns: + - name: date_key + data_tests: + - unique + - not_null + - name: date + data_tests: + - unique + - not_null + + - name: dim_seller + description: "One row per seller with surrogate key." + columns: + - name: seller_key + data_tests: + - unique + - not_null + - name: seller_id + data_tests: + - unique + - not_null + + - name: dim_product + description: "One row per product with surrogate key and gross margin." + columns: + - name: product_key + data_tests: + - unique + - not_null + - name: product_id + data_tests: + - unique + - not_null + + - name: dim_customer + description: "Type 1 customer dimension (current values only). Type 2 SCD added Day 4." + columns: + - name: customer_key + data_tests: + - unique + - not_null + - name: customer_id + data_tests: + - unique + - not_null + + - name: fact_orders + description: "Fact table at order line-item grain. Join to dim_* tables via *_key columns." + columns: + - name: order_item_id + data_tests: + - unique + - not_null + - name: date_key + data_tests: + - not_null + - relationships: + to: ref('dim_date') + field: date_key + - name: customer_key + data_tests: + - not_null + - relationships: + to: ref('dim_customer') + field: customer_key + - name: product_key + data_tests: + - not_null + - relationships: + to: ref('dim_product') + field: product_key + - name: seller_key + data_tests: + - not_null + - relationships: + to: ref('dim_seller') + field: seller_key + + - name: dim_customer_scd2 + description: > + SCD Type 2 customer dimension. One row per customer per time period. + Use valid_from/valid_to to join to fact_orders at the correct point in time. + is_current = TRUE for the version active today. + columns: + - name: customer_scd_key + data_tests: + - unique + - not_null + - name: customer_id + data_tests: + - not_null + - name: valid_from + data_tests: + - not_null + - name: version_number + data_tests: + - not_null diff --git a/ecom/models/marts/dim_customer.sql b/ecom/models/marts/dim_customer.sql new file mode 100644 index 0000000..ef2342b --- /dev/null +++ b/ecom/models/marts/dim_customer.sql @@ -0,0 +1,31 @@ +/* + dim_customer + ------------ + One row per customer — this is the Type 1 (overwrite) version. + Day 4 will replace this with a Type 2 (history-tracking) snapshot + that records changes to city and customer_tier over time. + + Type 1 = always shows the current/latest value, no history. + Type 2 = keeps every historical version with valid_from / valid_to dates. +*/ + +with source as ( + select * from {{ ref('stg_customers') }} +), + +final as ( + select + row_number() over (order by customer_id) as customer_key, -- surrogate key + customer_id, -- natural key + first_name, + last_name, + first_name || ' ' || last_name as full_name, + email, + city, + country, + customer_tier, + signup_date + from source +) + +select * from final diff --git a/ecom/models/marts/dim_customer_scd2.sql b/ecom/models/marts/dim_customer_scd2.sql new file mode 100644 index 0000000..bf01c61 --- /dev/null +++ b/ecom/models/marts/dim_customer_scd2.sql @@ -0,0 +1,149 @@ +/* + dim_customer_scd2 — Slowly Changing Dimension Type 2 + ------------------------------------------------------- + + WHY SCD TYPE 2? + --------------- + A customer's city and tier change over time. If we just overwrite the + current value (Type 1), we lose history. That means a report asking + "what was this customer's tier when they placed this order?" becomes + unanswerable — we only know what they are *now*. + + Type 2 solves this by keeping every historical version of a customer + as a separate row, each stamped with valid_from and valid_to dates. + + GRAIN: one row per customer per time period (version). + + KEY COLUMNS: + customer_scd_key surrogate key — unique per VERSION (not per customer) + customer_id natural key — same across all versions of one customer + valid_from date this version became active + valid_to date this version was superseded (NULL = still active) + is_current TRUE for the version active right now + version_number 1 = original record, 2 = first change, etc. + + HOW IT WORKS (step by step): + 1. Take the original signup record for every customer. + 2. Union with all change events from stg_customer_changes. + 3. Use LEAD() window function to look ahead to the next event date + for the same customer — that becomes the current row's valid_to. + 4. The last version per customer has no next event, so valid_to = NULL + and is_current = TRUE. + + FACT TABLE USAGE: + To get the customer dimension that was active when an order was placed: + JOIN dim_customer_scd2 scd + ON fact.customer_id = scd.customer_id + AND fact.order_date >= scd.valid_from + AND (fact.order_date < scd.valid_to OR scd.valid_to IS NULL) +*/ + +with customers as ( + select + customer_id, + city, + customer_tier, + signup_date as effective_date, + first_name, + last_name, + email + from {{ ref('stg_customers') }} +), + +changes as ( + select + customer_id, + city, + customer_tier, + effective_date, + null as first_name, + null as last_name, + null as email + from {{ ref('stg_customer_changes') }} +), + +-- combine original records and change events into one timeline +all_versions as ( + select * from customers + union all + select * from changes +), + +-- carry forward name/email from the original record +-- (change events don't re-supply these) +with_context as ( + select + customer_id, + effective_date, + -- coalesce fills nulls in change rows with the original values + coalesce( + city, + first_value(city) over ( + partition by customer_id + order by effective_date + rows between unbounded preceding and current row + ) + ) as city, + coalesce( + customer_tier, + first_value(customer_tier) over ( + partition by customer_id + order by effective_date + rows between unbounded preceding and current row + ) + ) as customer_tier, + first_value(first_name) over ( + partition by customer_id order by effective_date + ) as first_name, + first_value(last_name) over ( + partition by customer_id order by effective_date + ) as last_name, + first_value(email) over ( + partition by customer_id order by effective_date + ) as email, + row_number() over ( + partition by customer_id + order by effective_date + ) as version_number + from all_versions +), + +-- apply LEAD() to compute valid_to from the next version's start date +with_validity as ( + select + customer_id, + first_name, + last_name, + email, + city, + customer_tier, + effective_date as valid_from, + lead(effective_date) over ( + partition by customer_id + order by effective_date + ) as valid_to, -- NULL on the current version + version_number + from with_context +), + +final as ( + select + -- surrogate key: unique per customer VERSION + row_number() over ( + order by customer_id, valid_from + ) as customer_scd_key, + customer_id, + first_name, + last_name, + first_name || ' ' || last_name as full_name, + email, + city, + customer_tier, + valid_from, + valid_to, + valid_to is null as is_current, + version_number + from with_validity +) + +select * from final diff --git a/ecom/models/marts/dim_date.sql b/ecom/models/marts/dim_date.sql new file mode 100644 index 0000000..2d6b636 --- /dev/null +++ b/ecom/models/marts/dim_date.sql @@ -0,0 +1,39 @@ +/* + dim_date + -------- + A generated date dimension covering the full data range (Oct 2024 – Mar 2025). + No source table needed — built purely from DuckDB's generate_series(). + + date_key is an integer in YYYYMMDD format (e.g. 20241015). + This is the foreign key pattern used in fact_orders. + + Why a date dimension? + - Lets analysts filter/group by year, month, quarter, weekday without + writing date functions in every query. + - A single tested source of date logic = single source of truth. +*/ + +with date_spine as ( + select + unnest( + generate_series(date '2024-10-01', date '2025-03-31', interval '1 day') + )::date as date +), + +final as ( + select + cast(strftime(date, '%Y%m%d') as integer) as date_key, + date, + year(date) as year, + month(date) as month_number, + strftime(date, '%B') as month_name, + quarter(date) as quarter, + 'Q' || quarter(date) as quarter_label, + weekday(date) + 1 as day_of_week, + strftime(date, '%A') as day_name, + weekday(date) >= 5 as is_weekend, + week(date) as week_of_year + from date_spine +) + +select * from final diff --git a/ecom/models/marts/dim_product.sql b/ecom/models/marts/dim_product.sql new file mode 100644 index 0000000..692478e --- /dev/null +++ b/ecom/models/marts/dim_product.sql @@ -0,0 +1,30 @@ +/* + dim_product + ----------- + One row per product with surrogate key. + Includes gross_margin_pct derived in staging — ready for margin analysis. + is_active flag preserved so analysts can filter to current catalogue. +*/ + +with source as ( + select * from {{ ref('stg_products') }} +), + +final as ( + select + row_number() over (order by product_id) as product_key, -- surrogate key + product_id, -- natural key + product_name, + category, + subcategory, + seller_id, + list_price, + cost_price, + gross_margin_pct, + is_active, + stock_qty, + created_at + from source +) + +select * from final diff --git a/ecom/models/marts/dim_seller.sql b/ecom/models/marts/dim_seller.sql new file mode 100644 index 0000000..21c706f --- /dev/null +++ b/ecom/models/marts/dim_seller.sql @@ -0,0 +1,27 @@ +/* + dim_seller + ---------- + One row per seller. Surrogate key (seller_key) added as best practice — + in production you never use natural keys as FK in fact tables because + source systems can recycle or change them. +*/ + +with source as ( + select * from {{ ref('stg_sellers') }} +), + +final as ( + select + row_number() over (order by seller_id) as seller_key, -- surrogate key + seller_id, -- natural key + seller_name, + category, + city, + is_verified, + joined_date, + rating, + total_reviews + from source +) + +select * from final diff --git a/ecom/models/marts/fact_orders.sql b/ecom/models/marts/fact_orders.sql new file mode 100644 index 0000000..de1cfe3 --- /dev/null +++ b/ecom/models/marts/fact_orders.sql @@ -0,0 +1,83 @@ +/* + fact_orders + ----------- + Grain: one row per order LINE ITEM (most granular fact table). + + Why line-item grain and not order grain? + - An order can have multiple products — line-item grain lets you analyse + revenue, margin, and flash-sale rates per product, category, and seller. + - You can always roll up to order level with COUNT(DISTINCT order_id). + + Measures (the numbers): + quantity, unit_price, list_price, line_revenue, discount_pct + + Foreign keys (the context — join to dims for labels): + date_key → dim_date + customer_key → dim_customer + product_key → dim_product + seller_key → dim_seller + + Degenerate dimensions (IDs with no dim table): + order_id, order_item_id — kept directly in the fact table +*/ + +with orders as ( + select * from {{ ref('stg_orders') }} +), + +order_items as ( + select * from {{ ref('stg_order_items') }} +), + +dim_customer as ( + select customer_key, customer_id from {{ ref('dim_customer') }} +), + +dim_product as ( + select product_key, product_id from {{ ref('dim_product') }} +), + +dim_seller as ( + select seller_key, seller_id from {{ ref('dim_seller') }} +), + +joined as ( + select + -- degenerate dimensions (natural keys kept in fact) + oi.order_item_id, + o.order_id, + + -- foreign keys to dimensions + cast(strftime(o.order_date::date, '%Y%m%d') as integer) as date_key, + dc.customer_key, + dp.product_key, + ds.seller_key, + + -- measures + oi.quantity, + oi.unit_price, + oi.list_price, + oi.discount_pct, + oi.line_revenue, + oi.is_flash_sale, + + -- order-level attributes (non-additive context) + o.order_status, + o.payment_method, + o.shipping_city, + o.platform, + o.discount_amount as order_discount_amount, + o.coupon_code + + from order_items oi + inner join orders o + on oi.order_id = o.order_id + left join dim_customer dc + on o.customer_id = dc.customer_id + left join dim_product dp + on oi.product_id = dp.product_id + left join dim_seller ds + on oi.seller_id = ds.seller_id +) + +select * from joined diff --git a/ecom/models/staging/_schema.yml b/ecom/models/staging/_schema.yml new file mode 100644 index 0000000..071a347 --- /dev/null +++ b/ecom/models/staging/_schema.yml @@ -0,0 +1,124 @@ +version: 2 + +models: + - name: stg_customers + description: "Cleaned, deduplicated customer records. One row per unique email." + columns: + - name: customer_id + description: "Primary key" + data_tests: + - unique + - not_null + - name: email + data_tests: + - unique + - not_null + - name: customer_tier + data_tests: + - accepted_values: + values: ['bronze', 'silver', 'gold'] + - name: signup_date + data_tests: + - not_null + + - name: stg_sellers + description: "Cleaned seller accounts." + columns: + - name: seller_id + data_tests: + - unique + - not_null + - name: rating + data_tests: + - not_null + + - name: stg_products + description: "Cleaned product listings with derived margin." + columns: + - name: product_id + data_tests: + - unique + - not_null + - name: seller_id + data_tests: + - not_null + - relationships: + to: ref('stg_sellers') + field: seller_id + + - name: stg_orders + description: > + Cleaned orders. Mixed date formats resolved via TRY_STRPTIME. + Orphaned customer_ids removed. + columns: + - name: order_id + data_tests: + - unique + - not_null + - name: customer_id + data_tests: + - not_null + - relationships: + to: ref('stg_customers') + field: customer_id + - name: order_date + data_tests: + - not_null + - name: order_status + data_tests: + - accepted_values: + values: ['pending', 'processing', 'shipped', 'delivered', 'cancelled', 'returned'] + - name: payment_method + data_tests: + - accepted_values: + values: ['credit_card', 'debit_card', 'bank_transfer', 'cash_on_delivery', 'wallet'] + + - name: stg_order_items + description: "Cleaned order line items with flash-sale flag and line revenue." + columns: + - name: order_item_id + data_tests: + - unique + - not_null + - name: order_id + data_tests: + - not_null + - relationships: + to: ref('stg_orders') + field: order_id + - name: product_id + data_tests: + - not_null + - relationships: + to: ref('stg_products') + field: product_id + + - name: stg_sessions + description: "App/web sessions. customer_id is null for anonymous sessions." + columns: + - name: session_id + data_tests: + - unique + - not_null + - name: platform + data_tests: + - accepted_values: + values: ['ios', 'android', 'web'] + - name: device_type + data_tests: + - accepted_values: + values: ['mobile', 'desktop', 'tablet'] + + - name: stg_customer_changes + description: "Cleaned customer attribute change events." + columns: + - name: change_id + data_tests: + - unique + - not_null + - name: customer_id + data_tests: + - not_null + - name: effective_date + data_tests: + - not_null diff --git a/ecom/models/staging/_sources.yml b/ecom/models/staging/_sources.yml new file mode 100644 index 0000000..bfd2ed9 --- /dev/null +++ b/ecom/models/staging/_sources.yml @@ -0,0 +1,21 @@ +version: 2 + +sources: + - name: raw + description: "Raw CSV files loaded into DuckDB from data/raw/" + schema: raw + tables: + - name: customers + description: "Raw customer registrations — contains duplicates and null cities" + - name: sellers + description: "Raw seller accounts" + - name: products + description: "Raw product listings with cost and list price" + - name: orders + description: "Raw orders — mixed date formats (ISO and dd/mm/yyyy)" + - name: order_items + description: "Raw order line items — unit price may differ from list price" + - name: sessions + description: "Raw app/web sessions — 20% anonymous (null customer_id)" + - name: customer_changes + description: "Customer attribute change events — city moves and tier upgrades" diff --git a/ecom/models/staging/stg_customer_changes.sql b/ecom/models/staging/stg_customer_changes.sql new file mode 100644 index 0000000..191d99f --- /dev/null +++ b/ecom/models/staging/stg_customer_changes.sql @@ -0,0 +1,31 @@ +/* + stg_customer_changes + -------------------- + Cleans the raw customer attribute change events. + Each row = one point in time when a customer's city or tier changed. + + Only keeps changes for customer_ids that exist in stg_customers + (orphaned change events for unknown customers are dropped). +*/ + +with source as ( + select * from {{ source('raw', 'customer_changes') }} +), + +valid_customers as ( + select customer_id from {{ ref('stg_customers') }} +), + +final as ( + select + change_id, + customer_id, + cast(effective_date as date) as effective_date, + nullif(trim(city), '') as city, + lower(trim(customer_tier)) as customer_tier, + lower(trim(change_reason)) as change_reason + from source + where customer_id in (select customer_id from valid_customers) +) + +select * from final diff --git a/ecom/models/staging/stg_customers.sql b/ecom/models/staging/stg_customers.sql new file mode 100644 index 0000000..94d83ba --- /dev/null +++ b/ecom/models/staging/stg_customers.sql @@ -0,0 +1,57 @@ +/* + stg_customers + ------------- + - Renames columns to project convention + - Casts signup_date to DATE + - Deduplicates re-registered customers (keeps the earliest signup per email) + - Replaces empty city strings with NULL + - Excludes soft-deleted records +*/ + +with source as ( + select * from {{ source('raw', 'customers') }} +), + +renamed as ( + select + customer_id, + first_name, + last_name, + lower(trim(email)) as email, + phone, + nullif(trim(city), '') as city, + country, + lower(customer_tier) as customer_tier, + cast(signup_date as date) as signup_date, + cast(is_deleted as boolean) as is_deleted + from source +), + +deduped as ( + -- Keep earliest signup per email (first registration wins) + select *, + row_number() over ( + partition by email + order by signup_date asc + ) as rn + from renamed +), + +final as ( + select + customer_id, + first_name, + last_name, + email, + phone, + city, + country, + customer_tier, + signup_date, + is_deleted + from deduped + where rn = 1 + and is_deleted = false +) + +select * from final diff --git a/ecom/models/staging/stg_order_items.sql b/ecom/models/staging/stg_order_items.sql new file mode 100644 index 0000000..44ef2d5 --- /dev/null +++ b/ecom/models/staging/stg_order_items.sql @@ -0,0 +1,47 @@ +/* + stg_order_items + --------------- + - Casts numeric columns + - Adds derived column: is_flash_sale flag when unit_price < list_price + - Filters items whose order_id no longer exists after stg_orders cleanup +*/ + +with source as ( + select * from {{ source('raw', 'order_items') }} +), + +valid_orders as ( + select order_id from {{ ref('stg_orders') }} +), + +casted as ( + -- cast first in a separate CTE so arithmetic below works on typed values + select + order_item_id, + order_id, + product_id, + seller_id, + cast(quantity as integer) as quantity, + cast(unit_price as decimal(10,2)) as unit_price, + cast(list_price as decimal(10,2)) as list_price, + cast(discount_pct as decimal(6,4)) as discount_pct + from source + where order_id in (select order_id from valid_orders) +), + +renamed as ( + select + order_item_id, + order_id, + product_id, + seller_id, + quantity, + unit_price, + list_price, + discount_pct, + unit_price < list_price as is_flash_sale, + round(unit_price * quantity * (1 - coalesce(discount_pct, 0)), 2) as line_revenue + from casted +) + +select * from renamed diff --git a/ecom/models/staging/stg_orders.sql b/ecom/models/staging/stg_orders.sql new file mode 100644 index 0000000..09e6a70 --- /dev/null +++ b/ecom/models/staging/stg_orders.sql @@ -0,0 +1,55 @@ +/* + stg_orders + ---------- + - Key challenge: order_date arrives in two formats (ISO and dd/mm/yyyy). + We COALESCE two TRY_STRPTIME calls to handle both safely. + - Filters rows where the date cannot be parsed (data quality guard) + - Filters orphaned orders (customer_id not in stg_customers) + - Casts and standardises order_status and payment_method +*/ + +with source as ( + select * from {{ source('raw', 'orders') }} +), + +date_parsed as ( + select + order_id, + customer_id, + -- Try ISO first, then dd/mm/yyyy — whichever parses wins + coalesce( + try_strptime(order_date, '%Y-%m-%d'), + try_strptime(order_date, '%d/%m/%Y') + ) as order_date, + order_status, + payment_method, + shipping_city, + cast(discount_amount as decimal(10,2)) as discount_amount, + nullif(trim(coupon_code), '') as coupon_code, + lower(platform) as platform + from source +), + +valid_customers as ( + select customer_id from {{ ref('stg_customers') }} +), + +final as ( + select + o.order_id, + o.customer_id, + o.order_date, + o.order_status, + o.payment_method, + o.shipping_city, + o.discount_amount, + o.coupon_code, + o.platform + from date_parsed o + -- Drop unparseable dates + where o.order_date is not null + -- Drop orphaned customer_ids (referential integrity fix) + and o.customer_id in (select customer_id from valid_customers) +) + +select * from final diff --git a/ecom/models/staging/stg_products.sql b/ecom/models/staging/stg_products.sql new file mode 100644 index 0000000..dc8a21c --- /dev/null +++ b/ecom/models/staging/stg_products.sql @@ -0,0 +1,33 @@ +/* + stg_products + ------------ + - Casts price columns and boolean + - Derives gross_margin_pct for later mart use + - Filters inactive products are kept (marts can filter by is_active) +*/ + +with source as ( + select * from {{ source('raw', 'products') }} +), + +final as ( + select + product_id, + trim(product_name) as product_name, + trim(category) as category, + trim(subcategory) as subcategory, + seller_id, + cast(list_price as decimal(10,2)) as list_price, + cast(cost_price as decimal(10,2)) as cost_price, + round( + (cast(list_price as decimal(10,2)) - cast(cost_price as decimal(10,2))) + / nullif(cast(list_price as decimal(10,2)), 0), + 4 + ) as gross_margin_pct, + cast(is_active as boolean) as is_active, + cast(stock_qty as integer) as stock_qty, + cast(created_at as date) as created_at + from source +) + +select * from final diff --git a/ecom/models/staging/stg_sellers.sql b/ecom/models/staging/stg_sellers.sql new file mode 100644 index 0000000..c6720ee --- /dev/null +++ b/ecom/models/staging/stg_sellers.sql @@ -0,0 +1,26 @@ +/* + stg_sellers + ----------- + - Casts joined_date to DATE + - Casts is_verified to boolean + - Standardises category casing +*/ + +with source as ( + select * from {{ source('raw', 'sellers') }} +), + +final as ( + select + seller_id, + trim(seller_name) as seller_name, + trim(category) as category, + city, + cast(is_verified as boolean) as is_verified, + cast(joined_date as date) as joined_date, + cast(rating as decimal(3,1)) as rating, + cast(total_reviews as integer) as total_reviews + from source +) + +select * from final diff --git a/ecom/models/staging/stg_sessions.sql b/ecom/models/staging/stg_sessions.sql new file mode 100644 index 0000000..37238c7 --- /dev/null +++ b/ecom/models/staging/stg_sessions.sql @@ -0,0 +1,28 @@ +/* + stg_sessions + ------------ + - Null customer_id = anonymous session (kept, not filtered) + - Derives session_duration_mins for readability + - Casts is_converted to boolean +*/ + +with source as ( + select * from {{ source('raw', 'sessions') }} +), + +final as ( + select + session_id, + nullif(trim(customer_id), '') as customer_id, -- preserve anonymous + cast(session_date as date) as session_date, + lower(platform) as platform, + lower(device_type) as device_type, + cast(page_views as integer) as page_views, + cast(duration_seconds as integer) as duration_seconds, + round(cast(duration_seconds as decimal) / 60, 2) as duration_mins, + cast(is_converted as boolean) as is_converted, + nullif(trim(utm_source), '') as utm_source + from source +) + +select * from final diff --git a/ecom/profiles.yml b/ecom/profiles.yml new file mode 100644 index 0000000..802da62 --- /dev/null +++ b/ecom/profiles.yml @@ -0,0 +1,7 @@ +ecom: + target: dev + outputs: + dev: + type: duckdb + path: "{{ env_var('DBT_DUCKDB_PATH', '../data/ecom.duckdb') }}" + threads: 4 diff --git a/ecommerce_analytics_project_architecture.svg b/ecommerce_analytics_project_architecture.svg new file mode 100644 index 0000000..5227e6c --- /dev/null +++ b/ecommerce_analytics_project_architecture.svg @@ -0,0 +1,84 @@ + + E-commerce Analytics Engineering Project Architecture + Five-layer pipeline: data generator, raw layer, dbt staging, dbt marts with star schema, and BI dashboard + + + + + + + + + + Data generator + Python · Faker · e-commerce domain + + generate + orders · customers + products · sessions + + + + + + + + Raw layer + DuckDB · Parquet · source tables + + ingest + schema-on-read + immutable sources + + + + + + + + dbt staging models + stg_* · clean · typed · tested + + stage + rename · cast · dedupe + dbt tests + docs + + + + + + + + dbt marts — star schema + fact_orders · dims · SCD Type 2 + + model + fact_orders + dim_customer (SCD2) + dim_product · dim_date + + + + + + + + BI dashboard + metrics + Evidence.dev · single source of truth + + serve + GMV · AOV · retention + self-serve SQL queries + + + + + Git · GitHub Actions + CI · dbt build on PR + + + + + + domain: trendyol-style e-commerce · orders · customers · products · sellers + \ No newline at end of file diff --git a/generator/generate.py b/generator/generate.py new file mode 100644 index 0000000..bb0d30c --- /dev/null +++ b/generator/generate.py @@ -0,0 +1,269 @@ +""" +Day 1 — Trendyol-style e-commerce data generator. + +Produces 6 CSV files that land in data/raw/ as immutable source tables: + customers.csv, sellers.csv, products.csv, + orders.csv, order_items.csv, sessions.csv + +Intentional messiness (documented so you can narrate it in the interview): + - ~3% duplicate customer rows (same person, re-registered) + - ~5% of orders reference a customer_id that doesn't exist + - Null city on ~8% of customers + - Two date formats in orders.csv (ISO + dd/mm/yyyy) to force casting in staging + - Unit prices occasionally differ from the product's listed price (promo/flash sales) + - Some sessions have null customer_id (anonymous browsing) +""" + +import csv +import random +import os +from datetime import datetime, timedelta +from faker import Faker + +fake = Faker("tr_TR") # Turkish locale for realistic names/cities +fake_en = Faker("en_US") # English for product names +random.seed(42) + +RAW_DIR = os.path.join(os.path.dirname(__file__), "..", "data", "raw") +os.makedirs(RAW_DIR, exist_ok=True) + +# ── date range: Oct 2024 – Mar 2025 (6 months) ───────────────────────────── +START_DATE = datetime(2024, 10, 1) +END_DATE = datetime(2025, 3, 31) + +def rand_date(start=START_DATE, end=END_DATE): + delta = (end - start).days + return start + timedelta(days=random.randint(0, delta)) + +def iso(dt): return dt.strftime("%Y-%m-%d") +def euro(dt): return dt.strftime("%d/%m/%Y") # intentional bad format +def maybe_null(v, pct=0.08): + return "" if random.random() < pct else v + + +# ── CONFIG ────────────────────────────────────────────────────────────────── +N_CUSTOMERS = 600 +N_SELLERS = 80 +N_PRODUCTS = 400 +N_ORDERS = 2500 +N_SESSIONS = 4000 + +TURKISH_CITIES = [ + "Istanbul", "Ankara", "Izmir", "Bursa", "Antalya", + "Adana", "Konya", "Gaziantep", "Mersin", "Kayseri", + "Eskisehir", "Trabzon", "Diyarbakir", "Samsun", "Denizli", +] + +CATEGORIES = { + "Electronics": ["Smartphone", "Laptop", "Tablet", "Headphones", "Smartwatch", "Charger", "Speaker"], + "Fashion": ["T-Shirt", "Dress", "Jeans", "Jacket", "Sneakers", "Boots", "Bag", "Scarf"], + "Home & Life": ["Pillow", "Duvet", "Cookware Set", "Coffee Maker", "Vacuum Cleaner", "Lamp", "Mirror"], + "Beauty": ["Moisturizer", "Serum", "Lipstick", "Perfume", "Shampoo", "Face Mask"], + "Sports": ["Running Shoes", "Yoga Mat", "Resistance Band", "Protein Powder", "Bicycle Helmet"], + "Books": ["Novel", "Cookbook", "Self-Help Book", "Science Book", "Children's Book"], + "Toys": ["Board Game", "Puzzle", "Action Figure", "Building Blocks", "Doll"], +} + +PAYMENT_METHODS = ["credit_card", "debit_card", "bank_transfer", "cash_on_delivery", "wallet"] +ORDER_STATUSES = ["pending", "processing", "shipped", "delivered", "cancelled", "returned"] +STATUS_WEIGHTS = [0.05, 0.10, 0.15, 0.55, 0.10, 0.05] +PLATFORMS = ["ios", "android", "web"] +DEVICES = ["mobile", "mobile", "mobile", "desktop", "tablet"] # mobile-heavy like Trendyol + + +# ── HELPER ─────────────────────────────────────────────────────────────────── +def write_csv(rows, filename): + if not rows: + return + path = os.path.join(RAW_DIR, filename) + with open(path, "w", newline="", encoding="utf-8") as f: + writer = csv.DictWriter(f, fieldnames=rows[0].keys()) + writer.writeheader() + writer.writerows(rows) + + +# ── 1. SELLERS ─────────────────────────────────────────────────────────────── +sellers = [] +for i in range(1, N_SELLERS + 1): + category = random.choice(list(CATEGORIES.keys())) + sellers.append({ + "seller_id": f"S{i:04d}", + "seller_name": fake.company(), + "category": category, + "city": random.choice(TURKISH_CITIES), + "is_verified": random.choice([True, True, True, False]), # 75% verified + "joined_date": iso(rand_date(datetime(2020, 1, 1), START_DATE)), + "rating": round(random.uniform(3.0, 5.0), 1), + "total_reviews": random.randint(10, 50000), + }) + +write_csv(sellers, "sellers.csv") +print(f" sellers: {len(sellers):,} rows") + + +# ── 2. PRODUCTS ────────────────────────────────────────────────────────────── +products = [] +for i in range(1, N_PRODUCTS + 1): + category = random.choice(list(CATEGORIES.keys())) + subcategory = random.choice(CATEGORIES[category]) + cost = round(random.uniform(10, 800), 2) + price = round(cost * random.uniform(1.15, 2.5), 2) + seller = random.choice(sellers) + products.append({ + "product_id": f"P{i:05d}", + "product_name": f"{fake_en.color_name()} {subcategory}", + "category": category, + "subcategory": subcategory, + "seller_id": seller["seller_id"], + "list_price": price, + "cost_price": cost, + "is_active": random.choice([True, True, True, True, False]), # 80% active + "stock_qty": random.randint(0, 500), + "created_at": iso(rand_date(datetime(2022, 1, 1), START_DATE)), + }) + +write_csv(products, "products.csv") +print(f" products: {len(products):,} rows") + + +# ── 3. CUSTOMERS (with ~3% duplicates) ─────────────────────────────────────── +customers = [] +customer_ids = [] +for i in range(1, N_CUSTOMERS + 1): + signup = rand_date(datetime(2020, 1, 1), END_DATE) + tier = random.choices( + ["bronze", "silver", "gold"], + weights=[0.60, 0.28, 0.12] + )[0] + cid = f"C{i:05d}" + customer_ids.append(cid) + customers.append({ + "customer_id": cid, + "first_name": fake.first_name(), + "last_name": fake.last_name(), + "email": fake_en.email(), + "phone": fake.phone_number(), + "city": maybe_null(random.choice(TURKISH_CITIES), pct=0.08), + "country": "TR", + "customer_tier": tier, + "signup_date": iso(signup), + "is_deleted": random.random() < 0.02, # 2% soft-deleted + }) + +# inject ~3% duplicates (same person re-registered with slightly different email) +n_dupes = int(N_CUSTOMERS * 0.03) +for _ in range(n_dupes): + original = random.choice(customers) + dupe = original.copy() + dupe["customer_id"] = f"C{random.randint(99000, 99999)}" + dupe["email"] = fake_en.email() # different email, same person + customers.append(dupe) + +random.shuffle(customers) +write_csv(customers, "customers.csv") +print(f" customers: {len(customers):,} rows ({n_dupes} dupes injected)") + + +# ── 4. ORDERS (with two date formats + ~5% bad customer_ids) ───────────────── +# build a dict for fast lookup +valid_customer_ids = customer_ids # original IDs only (not dupes) + +orders = [] +order_ids = [] +for i in range(1, N_ORDERS + 1): + oid = f"O{i:06d}" + order_ids.append(oid) + order_date = rand_date() + + # ~5% reference a non-existent customer (referential integrity break) + if random.random() < 0.05: + cust_id = f"C{random.randint(90000, 90999)}" # bogus + else: + cust_id = random.choice(valid_customer_ids) + + # ~40% use European date format (intentional mess) + if random.random() < 0.40: + date_str = euro(order_date) + else: + date_str = iso(order_date) + + discount = round(random.uniform(0, 50), 2) if random.random() < 0.35 else 0.0 + status = random.choices(ORDER_STATUSES, weights=STATUS_WEIGHTS)[0] + + orders.append({ + "order_id": oid, + "customer_id": cust_id, + "order_date": date_str, # mixed formats! + "order_status": status, + "payment_method": random.choice(PAYMENT_METHODS), + "shipping_city": random.choice(TURKISH_CITIES), + "discount_amount": discount, + "coupon_code": maybe_null(f"COUP{random.randint(100,999)}", pct=0.70), + "platform": random.choice(PLATFORMS), + }) + +write_csv(orders, "orders.csv") +print(f" orders: {len(orders):,} rows (40% euro dates, 5% bad customer_ids)") + + +# ── 5. ORDER_ITEMS ─────────────────────────────────────────────────────────── +order_items = [] +item_counter = 1 +for order in orders: + n_items = random.choices([1, 2, 3, 4, 5], weights=[0.45, 0.30, 0.14, 0.07, 0.04])[0] + chosen_products = random.sample(products, min(n_items, len(products))) + for prod in chosen_products: + # unit price occasionally differs (flash sale / negotiated price) + list_price = prod["list_price"] + if random.random() < 0.20: + unit_price = round(list_price * random.uniform(0.60, 0.95), 2) # flash sale + else: + unit_price = list_price + + order_items.append({ + "order_item_id": f"OI{item_counter:07d}", + "order_id": order["order_id"], + "product_id": prod["product_id"], + "seller_id": prod["seller_id"], + "quantity": random.randint(1, 4), + "unit_price": unit_price, + "list_price": list_price, + "discount_pct": round(random.uniform(0, 0.30), 3) if random.random() < 0.25 else 0.0, + }) + item_counter += 1 + +write_csv(order_items, "order_items.csv") +print(f" order_items:{len(order_items):,} rows") + + +# ── 6. SESSIONS ────────────────────────────────────────────────────────────── +sessions = [] +for i in range(1, N_SESSIONS + 1): + session_date = rand_date() + # ~20% anonymous sessions + cust_id = maybe_null(random.choice(valid_customer_ids), pct=0.20) + duration = random.randint(10, 3600) + page_views = max(1, int(duration / random.uniform(20, 120))) + is_converted = random.random() < 0.12 # 12% conversion rate + + sessions.append({ + "session_id": f"SS{i:07d}", + "customer_id": cust_id, + "session_date": iso(session_date), + "platform": random.choice(PLATFORMS), + "device_type": random.choice(DEVICES), + "page_views": page_views, + "duration_seconds": duration, + "is_converted": is_converted, + "utm_source": maybe_null( + random.choice(["google", "facebook", "instagram", "email", "direct", "affiliate"]), + pct=0.30 + ), + }) + +write_csv(sessions, "sessions.csv") +print(f" sessions: {len(sessions):,} rows") + + +print("\nDone. Files written to data/raw/") +print(f"Date range: {iso(START_DATE)} → {iso(END_DATE)}") \ No newline at end of file diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..bbf10be --- /dev/null +++ b/requirements.txt @@ -0,0 +1,4 @@ +dbt-duckdb +faker +plotly +streamlit \ No newline at end of file diff --git a/scripts/generate_changes.py b/scripts/generate_changes.py new file mode 100644 index 0000000..6302005 --- /dev/null +++ b/scripts/generate_changes.py @@ -0,0 +1,114 @@ +""" +Generates customer_changes.csv — simulated attribute change events. + +These represent customers who changed city (e.g. moved) or were +upgraded/downgraded in tier (e.g. reached gold status) during the +Oct 2024 – Mar 2025 period. + +This is the source data that drives the SCD Type 2 dim_customer_scd2 model. +Run this once from the repo root before `dbt build`. + +Usage: + python scripts/generate_changes.py +""" + +import csv +import random +import os +from datetime import datetime, timedelta + +random.seed(99) + +RAW_DIR = os.path.join(os.path.dirname(__file__), "..", "data", "raw") +CUSTOMERS_CSV = os.path.join(RAW_DIR, "customers.csv") +OUTPUT_CSV = os.path.join(RAW_DIR, "customer_changes.csv") + +START_DATE = datetime(2024, 10, 1) +END_DATE = datetime(2025, 3, 31) + +TURKISH_CITIES = [ + "Istanbul", "Ankara", "Izmir", "Bursa", "Antalya", + "Adana", "Konya", "Gaziantep", "Mersin", "Kayseri", + "Eskisehir", "Trabzon", "Diyarbakir", "Samsun", "Denizli", +] + +TIER_UPGRADES = { + "bronze": "silver", + "silver": "gold", + "gold": "gold", # already at top +} + +def rand_date(start=START_DATE, end=END_DATE): + delta = (end - start).days + return start + timedelta(days=random.randint(1, delta)) + +def main(): + # read existing customers + with open(CUSTOMERS_CSV, newline="", encoding="utf-8") as f: + customers = list(csv.DictReader(f)) + + # pick ~18% to have at least one change + changers = random.sample(customers, k=int(len(customers) * 0.18)) + + changes = [] + change_id = 1 + + for customer in changers: + cid = customer["customer_id"] + city = customer["city"] or "Istanbul" + tier = customer["customer_tier"] + + # decide what changes: city, tier, or both + change_type = random.choice(["city", "tier", "city", "both"]) + + # first change event + date1 = rand_date() + new_city1 = city + new_tier1 = tier + + if change_type in ("city", "both"): + candidates = [c for c in TURKISH_CITIES if c != city] + new_city1 = random.choice(candidates) + if change_type in ("tier", "both"): + new_tier1 = TIER_UPGRADES[tier] + + changes.append({ + "change_id": f"CHG{change_id:05d}", + "customer_id": cid, + "effective_date": date1.strftime("%Y-%m-%d"), + "city": new_city1, + "customer_tier": new_tier1, + "change_reason": change_type, + }) + change_id += 1 + + # ~30% get a second change later in the period + second_start = date1 + timedelta(days=14) + if random.random() < 0.30 and second_start < END_DATE: + date2 = rand_date(start=second_start, end=END_DATE) + if date2 > date1: + candidates = [c for c in TURKISH_CITIES if c != new_city1] + changes.append({ + "change_id": f"CHG{change_id:05d}", + "customer_id": cid, + "effective_date": date2.strftime("%Y-%m-%d"), + "city": random.choice(candidates), + "customer_tier": TIER_UPGRADES[new_tier1], + "change_reason": "city", + }) + change_id += 1 + + # write output + with open(OUTPUT_CSV, "w", newline="", encoding="utf-8") as f: + writer = csv.DictWriter(f, fieldnames=[ + "change_id", "customer_id", "effective_date", + "city", "customer_tier", "change_reason" + ]) + writer.writeheader() + writer.writerows(changes) + + print(f" OK customer_changes.csv {len(changes):,} change events " + f"across {len(changers)} customers") + +if __name__ == "__main__": + main() diff --git a/scripts/load_raw.py b/scripts/load_raw.py new file mode 100644 index 0000000..577b776 --- /dev/null +++ b/scripts/load_raw.py @@ -0,0 +1,48 @@ +""" +Loads all CSVs from data/raw/ into DuckDB as tables under the 'raw' schema. +Run this once before `dbt build`. + +Usage (from repo root): + python scripts/load_raw.py +""" + +import duckdb +import os + +RAW_DIR = os.path.join(os.path.dirname(__file__), "..", "data", "raw") +DB_PATH = os.path.join(os.path.dirname(__file__), "..", "data", "ecom.duckdb") + +TABLES = [ + "customers", + "sellers", + "products", + "orders", + "order_items", + "sessions", + "customer_changes", +] + +def main(): + con = duckdb.connect(DB_PATH) + con.execute("CREATE SCHEMA IF NOT EXISTS raw") + + for table in TABLES: + csv_path = os.path.join(RAW_DIR, f"{table}.csv") + if not os.path.exists(csv_path): + print(f" SKIP {table}.csv not found") + continue + + con.execute(f"DROP TABLE IF EXISTS raw.{table}") + con.execute(f""" + CREATE TABLE raw.{table} AS + SELECT * FROM read_csv_auto('{csv_path}', all_varchar=true) + """) + + count = con.execute(f"SELECT COUNT(*) FROM raw.{table}").fetchone()[0] + print(f" OK raw.{table:<15} {count:>6,} rows") + + con.close() + print(f"\nDatabase written to: {DB_PATH}") + +if __name__ == "__main__": + main()