Mini-proyecto: el primer warehouse analítico de Kiosko

Descripción

Este proyecto cierra el módulo —y la guía completa— integrando las siete piezas anteriores: bronze y silver reconstruidos desde foundations (lección 3), el star con dim_product_scd historizada y el join punto-en-el-tiempo (lección 4), fact_sessions y fact_store_activity (lección 5), mart_daily_sales_obt publicada para BI (lección 6), y el mapa de guías hermanas (lección 7). Lo que falta es reunir todo en un solo script, de principio a fin, en una sola conexión de DuckDB, verificado con assert en cada paso, cerrando con el reporte final que la gerencia de Kiosko pidió desde la lección 2.

El proyecto tiene ocho partes. Primero, bronze: aterrizaje crudo de órdenes y eventos. Segundo, silver: la compuerta de calidad y fact_orders modelado. Tercero, el star: dim_store, dim_date, dim_product_scd historizada con dos corridas de MERGE INTO. Cuarto, revenue histórico correcto: el join punto-en-el-tiempo contrastado contra el roto. Quinto, el funnel y la actividad: fact_sessions y fact_store_activity. Sexto, la OBT: mart_daily_sales_obt, con el join correcto ya resuelto. Séptimo, el contrato: validate_gold_schema() sobre las cuatro tablas gold de la guía. Octavo, la declaración formal: KIOSKO_WAREHOUSE, la estructura que documenta todo el proyecto y el reporte final para la gerencia.

Conexión con el módulo. Este proyecto no introduce ningún concepto nuevo — es la integración final de los ocho módulos completos de esta guía, empaquetada como KIOSKO_WAREHOUSE, el warehouse analítico completo de Kiosko, verificado de punta a punta.

Una analogía: la inauguración del edificio completo, con las siete inspecciones ya aprobadas

Cada módulo de esta guía fue una inspección distinta, aprobada por separado: los cimientos (grano), la estructura completa (star), el sistema eléctrico y de plomería comparados con alternativas (snowflake/OBT), el registro de cambios del edificio (SCD), la certificación de que cada pregunta se responde con la fecha correcta (join punto-en-el-tiempo), dos sistemas adicionales que ningún plano tradicional cubre (funnel, actividad), y el inventario completo con su contrato de calidad (dominio feo, Medallion). Este proyecto es la inauguración: el edificio completo, con las siete inspecciones ya aprobadas, abierto por primera vez para que alguien —la gerencia de Kiosko— entre y lo use de verdad.

El material que necesitas

Necesitas, en la misma carpeta: kiosko.py (con DIM_STORE, DIM_PRODUCT, Order, transform_fact_orders), raw_orders.py (las 40 órdenes fijas) y events.py (los 32 eventos canónicos) — los mismos tres archivos que usaste en cada mini-proyecto de los ocho módulos de esta guía. No necesitas ningún archivo adicional: generate_date_dim(), validate_orders(), validate_gold_schema(), y toda la lógica de MERGE INTO y del cumulative design se definen directamente en el script de este proyecto.

La solución de referencia, verificada

Parte 1 — BRONZE: aterrizaje crudo de orders y events

# kiosko_analytics_warehouse.py -- el primer warehouse analitico de Kiosko, cierre de la guia
from datetime import date, datetime, timedelta

import duckdb

from kiosko import DIM_PRODUCT, DIM_STORE, Order, transform_fact_orders
from raw_orders import RAW_ORDERS
from events import RAW_EVENTS

DAY_NAMES = ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday"]
REQUIRED_COLUMNS = ["order_id", "store_id", "product_id", "quantity", "unit_price", "order_ts"]


def generate_date_dim(start_date: str, end_date: str) -> list[dict]:
    start = date.fromisoformat(start_date)
    end = date.fromisoformat(end_date)
    rows = []
    current = start
    while current <= end:
        weekday_index = current.weekday()
        rows.append({
            "date_key": int(current.strftime("%Y%m%d")), "calendar_date": current,
            "day_of_week": DAY_NAMES[weekday_index], "month": current.month,
            "quarter": (current.month - 1) // 3 + 1, "year": current.year,
            "is_weekend": weekday_index >= 5,
        })
        current += timedelta(days=1)
    return rows


def validate_orders(rows: list[dict]) -> tuple[list[dict], list[dict]]:
    valid: list[dict] = []
    rejected: list[dict] = []
    seen_order_ids: set[str] = set()
    for row in rows:
        reasons: list[str] = []
        missing = [c for c in REQUIRED_COLUMNS if not row.get(c)]
        if missing:
            reasons.append(f"missing or empty fields: {missing}")
            rejected.append({"row": row, "reasons": reasons})
            continue
        try:
            quantity = int(row["quantity"])
        except ValueError:
            reasons.append(f"quantity is not a valid integer: {row['quantity']!r}")
            quantity = None
        try:
            unit_price = float(row["unit_price"])
        except ValueError:
            reasons.append(f"unit_price is not a valid decimal: {row['unit_price']!r}")
            unit_price = None
        if quantity is not None and quantity <= 0:
            reasons.append(f"quantity must be > 0, got {quantity}")
        if unit_price is not None and unit_price < 0:
            reasons.append(f"unit_price must be >= 0, got {unit_price}")
        if row["order_id"] in seen_order_ids:
            reasons.append(f"duplicate order_id: {row['order_id']}")
        if reasons:
            rejected.append({"row": row, "reasons": reasons})
        else:
            seen_order_ids.add(row["order_id"])
            valid.append(row)
    return valid, rejected


def validate_gold_schema(con, table, expected_columns):
    actual_rows = con.sql(f"DESCRIBE {table}").fetchall()
    actual_columns = {row[0]: row[1] for row in actual_rows}
    discrepancies = []
    for column_name, expected_type in expected_columns.items():
        if column_name not in actual_columns:
            discrepancies.append(f"{table}: falta la columna '{column_name}' (se esperaba tipo {expected_type})")
        elif actual_columns[column_name] != expected_type:
            discrepancies.append(f"{table}: '{column_name}' tiene tipo {actual_columns[column_name]}, se esperaba {expected_type}")
    for column_name in actual_columns:
        if column_name not in expected_columns:
            discrepancies.append(f"{table}: columna inesperada '{column_name}', no declarada en el contrato")
    return discrepancies


print("=== Kiosko: el primer warehouse analitico completo, de punta a punta ===")
print(f"DuckDB version: {duckdb.__version__}\n")

con = duckdb.connect()

bronze_rows = [
    {"order_id": r[0], "store_id": r[1], "product_id": r[2],
     "quantity": str(r[3]), "unit_price": str(r[4]), "order_ts": r[5]}
    for r in RAW_ORDERS
]
con.execute("""
    CREATE TABLE bronze_orders (
        order_id VARCHAR, store_id VARCHAR, product_id VARCHAR,
        quantity VARCHAR, unit_price VARCHAR, order_ts VARCHAR
    )
""")
con.executemany("INSERT INTO bronze_orders VALUES (?, ?, ?, ?, ?, ?)",
    [(r["order_id"], r["store_id"], r["product_id"], r["quantity"], r["unit_price"], r["order_ts"]) for r in bronze_rows])
con.execute("CREATE TABLE bronze_events (event_id VARCHAR, event_type VARCHAR, session_id VARCHAR, event_ts VARCHAR)")
con.executemany("INSERT INTO bronze_events VALUES (?, ?, ?, ?)", RAW_EVENTS)

bronze_orders_count = con.sql("SELECT COUNT(*) FROM bronze_orders").fetchone()[0]
bronze_events_count = con.sql("SELECT COUNT(*) FROM bronze_events").fetchone()[0]
print("Parte 1 -- BRONZE: aterrizaje crudo, sin transformar")
print(f"  bronze_orders   {bronze_orders_count:3} filas")
print(f"  bronze_events   {bronze_events_count:3} filas")

Parte 2 — SILVER: compuerta de calidad + fact_orders modelado

valid_rows, rejected_rows = validate_orders(bronze_rows)

orders = [
    Order(order_id=r["order_id"], store_id=r["store_id"], product_id=r["product_id"],
          quantity=int(r["quantity"]), unit_price=float(r["unit_price"]),
          order_ts=datetime.fromisoformat(r["order_ts"]))
    for r in valid_rows
]
fact_orders_rows = transform_fact_orders(orders, DIM_STORE, DIM_PRODUCT)

con.execute("""
    CREATE TABLE fact_orders (
        order_id VARCHAR, store_id VARCHAR, product_id VARCHAR,
        quantity INTEGER, unit_price DOUBLE, revenue DOUBLE, order_ts TIMESTAMP
    )
""")
con.executemany("INSERT INTO fact_orders VALUES (?, ?, ?, ?, ?, ?, ?)",
    [(r["order_id"], r["store_id"], r["product_id"], r["quantity"],
      r["unit_price"], r["revenue"], r["order_ts"]) for r in fact_orders_rows])
con.execute("CREATE TABLE events (event_id VARCHAR, event_type VARCHAR, session_id VARCHAR, event_ts TIMESTAMP)")
con.executemany("INSERT INTO events VALUES (?, ?, ?, ?)",
    [(r[0], r[1], r[2], datetime.fromisoformat(r[3])) for r in RAW_EVENTS])

total_orders = con.sql("SELECT COUNT(*) FROM fact_orders").fetchone()[0]
total_revenue = con.sql("SELECT ROUND(SUM(revenue), 2) FROM fact_orders").fetchone()[0]
grain_check = con.sql("SELECT COUNT(*), COUNT(DISTINCT order_id || '-' || product_id) FROM fact_orders").fetchone()

print("\nParte 2 -- SILVER: compuerta de calidad + fact_orders modelado")
print(f"  validate_orders(): {len(valid_rows)} validas, {len(rejected_rows)} rechazadas")
print(f"  fact_orders        {total_orders:3} filas, revenue total = {total_revenue}")
print(f"  grano verificado: COUNT(*)={grain_check[0]} == COUNT(DISTINCT order_id-product_id)={grain_check[1]}")
assert len(valid_rows) == 40 and len(rejected_rows) == 0
assert total_orders == 40 and total_revenue == 106.15
assert grain_check[0] == grain_check[1] == 40
print("  Verificacion OK: bronze -> silver -> gold reproduce fact_orders identico a foundations")

Qué esperar (Partes 1 y 2).

=== Kiosko: el primer warehouse analitico completo, de punta a punta ===
DuckDB version: 1.5.5

Parte 1 -- BRONZE: aterrizaje crudo, sin transformar
  bronze_orders    40 filas
  bronze_events    32 filas

Parte 2 -- SILVER: compuerta de calidad + fact_orders modelado
  validate_orders(): 40 validas, 0 rechazadas
  fact_orders         40 filas, revenue total = 106.15
  grano verificado: COUNT(*)=40 == COUNT(DISTINCT order_id-product_id)=40
  Verificacion OK: bronze -> silver -> gold reproduce fact_orders identico a foundations

Parte 3 — EL STAR: dim_store, dim_date, dim_product_scd historizada

con.execute("CREATE TABLE dim_store_natural (store_id VARCHAR, store_name VARCHAR, city VARCHAR)")
con.executemany("INSERT INTO dim_store_natural VALUES (?, ?, ?)",
                 [(s["store_id"], s["store_name"], s["city"]) for s in DIM_STORE])
con.execute("CREATE TABLE dim_store AS SELECT ROW_NUMBER() OVER (ORDER BY store_id) AS store_key, store_id, store_name, city FROM dim_store_natural")

dim_date_rows = generate_date_dim("2026-08-01", "2026-08-31")
con.execute("""
    CREATE TABLE dim_date (
        date_key INTEGER, calendar_date DATE, day_of_week VARCHAR,
        month INTEGER, quarter INTEGER, year INTEGER, is_weekend BOOLEAN
    )
""")
con.executemany("INSERT INTO dim_date VALUES (?, ?, ?, ?, ?, ?, ?)",
    [(r["date_key"], r["calendar_date"], r["day_of_week"], r["month"],
      r["quarter"], r["year"], r["is_weekend"]) for r in dim_date_rows])

con.execute("CREATE SEQUENCE product_key_seq START 1")
con.execute("""
    CREATE TABLE dim_product_scd (
        product_key INTEGER PRIMARY KEY, product_id VARCHAR NOT NULL, product_name VARCHAR,
        category VARCHAR, unit_cost DOUBLE, valid_from DATE NOT NULL, valid_to DATE,
        is_current BOOLEAN NOT NULL DEFAULT true
    )
""")
con.executemany(
    "INSERT INTO dim_product_scd VALUES (nextval('product_key_seq'), ?, ?, ?, ?, DATE '2026-08-01', NULL, true)",
    [(p["product_id"], p["product_name"], p["category"], p["unit_cost"]) for p in DIM_PRODUCT])

PRODUCTS_V1 = [("P001", "Bottled Water 600ml", "beverages", 0.40), ("P002", "Energy Bar", "snacks", 0.60),
               ("P003", "Instant Coffee Sachet", "beverages", 0.35), ("P004", "Phone Charger Cable", "electronics", 2.10)]
PRODUCTS_V2 = [("P001", "Bottled Water 600ml", "beverages", 0.40), ("P002", "Energy Bar", "health-snacks", 0.68),
               ("P003", "Instant Coffee Sachet", "beverages", 0.35), ("P004", "Phone Charger Cable", "electronics", 2.10)]


def load_staging(products):
    con.execute("DROP TABLE IF EXISTS staging_product")
    con.execute("CREATE TABLE staging_product (product_id VARCHAR, product_name VARCHAR, category VARCHAR, unit_cost DOUBLE)")
    con.executemany("INSERT INTO staging_product VALUES (?, ?, ?, ?)", products)


def merge_scd(change_date):
    result = con.sql(f"""
        MERGE INTO dim_product_scd AS target USING staging_product AS source
        ON target.product_id = source.product_id AND target.is_current = true
        WHEN MATCHED AND (target.unit_cost <> source.unit_cost OR target.category <> source.category)
        THEN UPDATE SET valid_to = DATE '{change_date}' - INTERVAL 1 DAY, is_current = false
        RETURNING merge_action, product_id
    """)
    changed_ids = [row[1] for row in result.fetchall()]
    if changed_ids:
        placeholders = ", ".join("?" for _ in changed_ids)
        con.execute(f"""
            INSERT INTO dim_product_scd (product_key, product_id, product_name, category, unit_cost, valid_from, valid_to, is_current)
            SELECT nextval('product_key_seq'), source.product_id, source.product_name, source.category, source.unit_cost,
                   DATE '{change_date}', NULL, true
            FROM staging_product AS source WHERE source.product_id IN ({placeholders})
        """, changed_ids)
    return len(changed_ids)


load_staging(PRODUCTS_V1)
rows_changed_1 = merge_scd("2026-08-15")
load_staging(PRODUCTS_V2)
rows_changed_2 = merge_scd("2026-08-15")

dim_product_scd_count = con.sql("SELECT COUNT(*) FROM dim_product_scd").fetchone()[0]
p002_versions = con.sql("""
    SELECT COUNT(*), SUM(CASE WHEN is_current THEN 1 ELSE 0 END) FROM dim_product_scd WHERE product_id = 'P002'
""").fetchone()

con.execute("""
    CREATE TABLE fact_orders_star AS
    SELECT f.order_id, f.store_id, f.product_id, s.store_key, d.date_key, f.quantity, f.unit_price, f.revenue
    FROM fact_orders f
    JOIN dim_store s ON f.store_id = s.store_id
    JOIN dim_date  d ON CAST(strftime(f.order_ts, '%Y%m%d') AS INTEGER) = d.date_key
""")
star_rows = con.sql("SELECT COUNT(*) FROM fact_orders_star").fetchone()[0]

print("\nParte 3 -- EL STAR: dim_store, dim_date, dim_product_scd historizada")
print(f"  dim_store         {con.sql('SELECT COUNT(*) FROM dim_store').fetchone()[0]:3} filas")
print(f"  dim_date          {con.sql('SELECT COUNT(*) FROM dim_date').fetchone()[0]:3} filas")
print(f"  dim_product_scd   {dim_product_scd_count:3} filas (MERGE #1 cerro {rows_changed_1}, MERGE #2 cerro {rows_changed_2})")
print(f"  P002: {p002_versions[0]} versiones, {p002_versions[1]} vigente(s)")
print(f"  fact_orders_star  {star_rows:3} filas")
assert dim_product_scd_count == 5 and p002_versions == (2, 1) and star_rows == 40

Qué esperar (Parte 3).

Parte 3 -- EL STAR: dim_store, dim_date, dim_product_scd historizada
  dim_store           3 filas
  dim_date           31 filas
  dim_product_scd     5 filas (MERGE #1 cerro 0, MERGE #2 cerro 1)
  P002: 2 versiones, 1 vigente(s)
  fact_orders_star   40 filas

Parte 4 — REVENUE HISTORICO CORRECTO: join punto-en-el-tiempo vs is_current

broken = con.sql("""
    SELECT d.category, COUNT(*) AS orders, ROUND(SUM(f.revenue), 2) AS revenue,
           ROUND(SUM(f.revenue - f.quantity * d.unit_cost), 2) AS margin
    FROM fact_orders f JOIN dim_product_scd d ON f.product_id = d.product_id AND d.is_current = true
    GROUP BY d.category ORDER BY d.category
""").fetchall()
correct = con.sql("""
    SELECT d.category, COUNT(*) AS orders, ROUND(SUM(f.revenue), 2) AS revenue,
           ROUND(SUM(f.revenue - f.quantity * d.unit_cost), 2) AS margin
    FROM fact_orders f
    JOIN dim_product_scd d ON f.product_id = d.product_id
       AND f.order_ts BETWEEN d.valid_from AND COALESCE(d.valid_to, DATE '9999-12-31')
    GROUP BY d.category ORDER BY d.category
""").fetchall()
total_broken = con.sql("""
    SELECT ROUND(SUM(f.revenue), 2) FROM fact_orders f
    JOIN dim_product_scd d ON f.product_id = d.product_id AND d.is_current = true
""").fetchone()[0]
total_correct = con.sql("""
    SELECT ROUND(SUM(f.revenue), 2) FROM fact_orders f
    JOIN dim_product_scd d ON f.product_id = d.product_id
       AND f.order_ts BETWEEN d.valid_from AND COALESCE(d.valid_to, DATE '9999-12-31')
""").fetchone()[0]

print("\nParte 4 -- revenue historico: JOIN roto vs JOIN correcto")
print("  ROTO (is_current = true):")
for category, orders_count, revenue, margin in broken:
    print(f"    {category:14} {orders_count:2} ordenes  revenue={revenue:7}  margin={margin:6}")
print("  CORRECTO (BETWEEN valid_from AND valid_to):")
for category, orders_count, revenue, margin in correct:
    print(f"    {category:14} {orders_count:2} ordenes  revenue={revenue:7}  margin={margin:6}")
print(f"  Revenue total: identico en ambos casos -> {total_broken} == {total_correct}")

broken_categories = {c for c, *_ in broken}
correct_categories = {c for c, *_ in correct}
assert "health-snacks" in broken_categories and "health-snacks" not in correct_categories
assert "snacks" in correct_categories and "snacks" not in broken_categories
assert total_broken == total_correct == 106.15
print("  Verificacion OK: 'snacks' es la categoria correcta, no 'health-snacks'")

Qué esperar (Parte 4).

Parte 4 -- revenue historico: JOIN roto vs JOIN correcto
  ROTO (is_current = true):
    beverages      23 ordenes  revenue=  44.05  margin= 14.75
    electronics     7 ordenes  revenue=   40.5  margin=  21.6
    health-snacks  10 ordenes  revenue=   21.6  margin=  9.36
  CORRECTO (BETWEEN valid_from AND valid_to):
    beverages      23 ordenes  revenue=  44.05  margin= 14.75
    electronics     7 ordenes  revenue=   40.5  margin=  21.6
    snacks         10 ordenes  revenue=   21.6  margin=  10.8
  Revenue total: identico en ambos casos -> 106.15 == 106.15
  Verificacion OK: 'snacks' es la categoria correcta, no 'health-snacks'

Parte 5 — fact_sessions y fact_store_activity

STORE_ROTATION = ["S01", "S02", "S03"]


def store_for_session(session_id):
    session_number = int(session_id.split("-")[1])
    return STORE_ROTATION[(session_number - 1) % 3]


ALL_SESSIONS = [f"SESS-{n:02d}" for n in range(1, 18)]
con.execute("CREATE TABLE session_store_map (session_id VARCHAR, store_id VARCHAR)")
con.executemany("INSERT INTO session_store_map VALUES (?, ?)", [(sid, store_for_session(sid)) for sid in ALL_SESSIONS])

con.execute("""
    CREATE TABLE fact_sessions AS
    SELECT e.session_id, m.store_id, MIN(CAST(e.event_ts AS DATE)) AS session_date,
        MAX(CASE WHEN e.event_type = 'page_view'   THEN e.event_ts END) AS view_ts,
        MAX(CASE WHEN e.event_type = 'add_to_cart' THEN e.event_ts END) AS add_to_cart_ts,
        MAX(CASE WHEN e.event_type = 'purchase'    THEN e.event_ts END) AS purchase_ts,
        MAX(CASE WHEN e.event_type = 'purchase' THEN true ELSE false END) AS is_converted
    FROM events e JOIN session_store_map m ON e.session_id = m.session_id
    GROUP BY e.session_id, m.store_id
""")
funnel = con.sql("SELECT COUNT(*), COUNT(view_ts), COUNT(add_to_cart_ts), COUNT(purchase_ts) FROM fact_sessions").fetchone()
conversion_pct = con.sql("SELECT ROUND(100.0 * COUNT(purchase_ts) / COUNT(*), 1) FROM fact_sessions").fetchone()[0]

con.execute("""
    CREATE TABLE fact_store_activity (
        store_id VARCHAR, activity_date DATE, daily_revenue DOUBLE,
        revenue_array_7d DOUBLE[], active_days_7d INTEGER,
        revenue_array_30d DOUBLE[], active_days_30d INTEGER
    )
""")
DAYS = ["2026-08-03", "2026-08-04", "2026-08-05", "2026-08-06", "2026-08-07", "2026-08-08", "2026-08-09"]
for day in DAYS:
    for store_id in STORE_ROTATION:
        daily_revenue = con.sql(f"""
            SELECT COALESCE(ROUND(SUM(revenue), 2), 0.0) FROM fact_orders
            WHERE store_id = '{store_id}' AND CAST(order_ts AS DATE) = DATE '{day}'
        """).fetchone()[0]
        prev = con.sql(f"""
            SELECT revenue_array_7d, revenue_array_30d FROM fact_store_activity
            WHERE store_id = '{store_id}' ORDER BY activity_date DESC LIMIT 1
        """).fetchone()
        if prev is None:
            new_7d, new_30d = [daily_revenue], [daily_revenue]
        else:
            new_7d = ([daily_revenue] + list(prev[0]))[:7]
            new_30d = ([daily_revenue] + list(prev[1]))[:30]
        active_7d = sum(1 for v in new_7d if v > 0)
        active_30d = sum(1 for v in new_30d if v > 0)
        con.execute("INSERT INTO fact_store_activity VALUES (?, ?, ?, ?, ?, ?, ?)",
            (store_id, day, daily_revenue, new_7d, active_7d, new_30d, active_30d))

activity_rows = con.sql("SELECT COUNT(*) FROM fact_store_activity").fetchone()[0]
snapshot = con.sql("""
    SELECT store_id, ROUND(list_sum(revenue_array_7d), 2) AS revenue_7d, active_days_7d, active_days_30d
    FROM fact_store_activity WHERE activity_date = DATE '2026-08-09' ORDER BY store_id
""").fetchall()

print("\nParte 5 -- fact_sessions (accumulating snapshot) + fact_store_activity (cumulative)")
print(f"  fact_sessions        {con.sql('SELECT COUNT(*) FROM fact_sessions').fetchone()[0]:3} filas")
print(f"  funnel: total={funnel[0]} vieron={funnel[1]} carrito={funnel[2]} compraron={funnel[3]} conversion={conversion_pct}%")
print(f"  fact_store_activity  {activity_rows:3} filas")
for store_id, revenue_7d, active_7d, active_30d in snapshot:
    print(f"    {store_id}: revenue_7d={revenue_7d:6}  active_days_7d={active_7d}  active_days_30d={active_30d}")
assert funnel == (17, 17, 9, 6) and conversion_pct == 35.3
assert activity_rows == 21

Qué esperar (Parte 5).

Parte 5 -- fact_sessions (accumulating snapshot) + fact_store_activity (cumulative)
  fact_sessions         17 filas
  funnel: total=17 vieron=17 carrito=9 compraron=6 conversion=35.3%
  fact_store_activity   21 filas
    S01: revenue_7d=  38.3  active_days_7d=7  active_days_30d=7
    S02: revenue_7d=  38.8  active_days_7d=7  active_days_30d=7
    S03: revenue_7d= 29.05  active_days_7d=6  active_days_30d=6

Parte 6 — mart_daily_sales_obt: la OBT para el equipo de BI

con.execute("""
    CREATE TABLE mart_daily_sales_obt AS
    SELECT CAST(f.order_ts AS DATE) AS sale_date, dt.day_of_week, dt.is_weekend,
        s.store_id, s.store_name, s.city, p.product_id, p.product_name, p.category, p.unit_cost,
        SUM(f.quantity) AS quantity, ROUND(SUM(f.revenue), 2) AS revenue,
        ROUND(SUM(f.revenue - f.quantity * p.unit_cost), 2) AS margin
    FROM fact_orders f
    JOIN dim_store s ON f.store_id = s.store_id
    JOIN dim_date  dt ON CAST(strftime(f.order_ts, '%Y%m%d') AS INTEGER) = dt.date_key
    JOIN dim_product_scd p ON f.product_id = p.product_id
       AND f.order_ts BETWEEN p.valid_from AND COALESCE(p.valid_to, DATE '9999-12-31')
    GROUP BY 1,2,3,4,5,6,7,8,9,10
""")
obt_rows = con.sql("SELECT COUNT(*) FROM mart_daily_sales_obt").fetchone()[0]
obt_revenue = con.sql("SELECT ROUND(SUM(revenue), 2) FROM mart_daily_sales_obt").fetchone()[0]
obt_categories = con.sql("SELECT DISTINCT category FROM mart_daily_sales_obt ORDER BY category").fetchall()

print("\nParte 6 -- mart_daily_sales_obt: la OBT para el equipo de BI")
print(f"  mart_daily_sales_obt {obt_rows:3} filas, revenue total = {obt_revenue}")
print(f"  categorias presentes: {[c[0] for c in obt_categories]}")
assert obt_revenue == 106.15
assert "snacks" in {c[0] for c in obt_categories} and "health-snacks" not in {c[0] for c in obt_categories}

Qué esperar (Parte 6).

Parte 6 -- mart_daily_sales_obt: la OBT para el equipo de BI
  mart_daily_sales_obt  39 filas, revenue total = 106.15
  categorias presentes: ['beverages', 'electronics', 'snacks']

Parte 7 — validate_gold_schema() sobre las 4 tablas gold de la guía

GOLD_CONTRACTS = {
    "fact_orders": {
        "order_id": "VARCHAR", "store_id": "VARCHAR", "product_id": "VARCHAR",
        "quantity": "INTEGER", "unit_price": "DOUBLE", "revenue": "DOUBLE", "order_ts": "TIMESTAMP",
    },
    "fact_sessions": {
        "session_id": "VARCHAR", "store_id": "VARCHAR", "session_date": "DATE",
        "view_ts": "TIMESTAMP", "add_to_cart_ts": "TIMESTAMP", "purchase_ts": "TIMESTAMP", "is_converted": "BOOLEAN",
    },
    "fact_store_activity": {
        "store_id": "VARCHAR", "activity_date": "DATE", "daily_revenue": "DOUBLE",
        "revenue_array_7d": "DOUBLE[]", "active_days_7d": "INTEGER",
        "revenue_array_30d": "DOUBLE[]", "active_days_30d": "INTEGER",
    },
    "dim_date": {
        "date_key": "INTEGER", "calendar_date": "DATE", "day_of_week": "VARCHAR",
        "month": "INTEGER", "quarter": "INTEGER", "year": "INTEGER", "is_weekend": "BOOLEAN",
    },
}
print("\nParte 7 -- validate_gold_schema() sobre las 4 tablas gold de la guia")
total_discrepancies = 0
for table, expected_columns in GOLD_CONTRACTS.items():
    discrepancies = validate_gold_schema(con, table, expected_columns)
    total_discrepancies += len(discrepancies)
    status = "OK, 0 discrepancias" if not discrepancies else f"{len(discrepancies)} discrepancias"
    print(f"  {table:20} {status}")
assert total_discrepancies == 0
print(f"  TOTAL: {len(GOLD_CONTRACTS)} tablas gold, {total_discrepancies} discrepancias -- OK")

Qué esperar (Parte 7).

Parte 7 -- validate_gold_schema() sobre las 4 tablas gold de la guia
  fact_orders          OK, 0 discrepancias
  fact_sessions        OK, 0 discrepancias
  fact_store_activity  OK, 0 discrepancias
  dim_date             OK, 0 discrepancias
  TOTAL: 4 tablas gold, 0 discrepancias -- OK

Parte 8 — KIOSKO_WAREHOUSE: la declaración formal, y el reporte final

KIOSKO_WAREHOUSE = {
    "bronze": {"bronze_orders": bronze_orders_count, "bronze_events": bronze_events_count},
    "silver": {"valid_rows": len(valid_rows), "rejected_rows": len(rejected_rows), "fact_orders_rows": total_orders},
    "gold_star": {"dim_store": 3, "dim_date": 31, "dim_product_scd": dim_product_scd_count, "fact_orders_star_rows": star_rows},
    "historized_product": {
        "product_id": "P002", "versions": p002_versions[0], "current_versions": p002_versions[1],
        "change_date": "2026-08-15", "category_before": "snacks", "category_after": "health-snacks",
    },
    "point_in_time_revenue": {
        "broken_category_for_p002": "health-snacks", "correct_category_for_p002": "snacks",
        "total_revenue_broken_vs_correct": [total_broken, total_correct],
    },
    "gold_funnel_and_activity": {
        "fact_sessions_rows": 17, "funnel_total_viewed_cart_purchased": list(funnel),
        "funnel_conversion_pct": conversion_pct, "fact_store_activity_rows": activity_rows,
    },
    "gold_obt": {"mart_daily_sales_obt_rows": obt_rows, "revenue": obt_revenue},
    "gold_contract": {"tables_validated": list(GOLD_CONTRACTS.keys()), "discrepancies": total_discrepancies},
}
print("\nParte 8 -- KIOSKO_WAREHOUSE: la declaracion formal que cierra la guia")
for key, value in KIOSKO_WAREHOUSE.items():
    print(f"  {key}: {value}")

print("\n=== Reporte final para la gerencia de Kiosko ===")
print(con.sql("""
    SELECT store_name, ROUND(SUM(revenue), 2) AS revenue, ROUND(SUM(margin), 2) AS margin
    FROM mart_daily_sales_obt GROUP BY store_name ORDER BY store_name
"""))
print(con.sql("""
    SELECT category, ROUND(SUM(revenue), 2) AS revenue, ROUND(SUM(margin), 2) AS margin
    FROM mart_daily_sales_obt GROUP BY category ORDER BY category
"""))

Qué esperar. Al correr python3 kiosko_analytics_warehouse.py completo (las ocho partes juntas), la salida termina exactamente así:

Parte 8 -- KIOSKO_WAREHOUSE: la declaracion formal que cierra la guia
  bronze: {'bronze_orders': 40, 'bronze_events': 32}
  silver: {'valid_rows': 40, 'rejected_rows': 0, 'fact_orders_rows': 40}
  gold_star: {'dim_store': 3, 'dim_date': 31, 'dim_product_scd': 5, 'fact_orders_star_rows': 40}
  historized_product: {'product_id': 'P002', 'versions': 2, 'current_versions': 1, 'change_date': '2026-08-15', 'category_before': 'snacks', 'category_after': 'health-snacks'}
  point_in_time_revenue: {'broken_category_for_p002': 'health-snacks', 'correct_category_for_p002': 'snacks', 'total_revenue_broken_vs_correct': [106.15, 106.15]}
  gold_funnel_and_activity: {'fact_sessions_rows': 17, 'funnel_total_viewed_cart_purchased': [17, 17, 9, 6], 'funnel_conversion_pct': 35.3, 'fact_store_activity_rows': 21}
  gold_obt: {'mart_daily_sales_obt_rows': 39, 'revenue': 106.15}
  gold_contract: {'tables_validated': ['fact_orders', 'fact_sessions', 'fact_store_activity', 'dim_date'], 'discrepancies': 0}

=== Reporte final para la gerencia de Kiosko ===
┌───────────────┬─────────┬────────┐
│  store_name   │ revenue │ margin │
│    varchar    │ double  │ double │
├───────────────┼─────────┼────────┤
│ Kiosko Centro │    38.3 │   17.4 │
│ Kiosko Norte  │   38.8  │  17.65 │
│ Kiosko Sur    │   29.05 │   12.1 │
└───────────────┴─────────┴────────┘

┌─────────────┬─────────┬────────┐
│  category   │ revenue │ margin │
│   varchar   │ double  │ double │
├─────────────┼─────────┼────────┤
│ beverages   │   44.05 │  14.75 │
│ electronics │    40.5 │   21.6 │
│ snacks      │    21.6 │   10.8 │
└─────────────┴─────────┴────────┘

Detente en KIOSKO_WAREHOUSE y en el reporte final juntos, porque son los que resumen toda la guía en una sola imagen. Ocho campos, cada uno con la evidencia numérica de una lección distinta de este módulo —y, en el fondo, de un módulo distinto de los ocho que componen la guía completa—. Y las dos tablas finales son, literalmente, lo que la gerencia de Kiosko pidió en el brief de la lección 2: revenue y margen por tienda, revenue y margen por categoría —con snacks en vez de health-snacks, sin que nadie del lado de BI haya tenido que escribir un solo BETWEEN valid_from AND valid_to.

Diagrama: el warehouse completo, las ocho partes cerradas con evidencia

flowchart TD
    A["P1: BRONZE\nVERIFICADO -- 40+32 filas"] --> B
    B["P2: SILVER\nVERIFICADO -- 40 filas, 106.15, 0 rechazadas"] --> C
    C["P3: STAR + SCD\nVERIFICADO -- P002 en 2 versiones"] --> D
    D["P4: JOIN punto-en-el-tiempo\nVERIFICADO -- snacks correcto, 106.15"] --> E
    E["P5: fact_sessions + fact_store_activity\nVERIFICADO -- 17->9->6, 35.3%, activos 7/30d"] --> F
    F["P6: mart_daily_sales_obt\nVERIFICADO -- 39 filas, join correcto adentro"] --> G
    G["P7: validate_gold_schema()\nVERIFICADO -- 0 discrepancias"] --> H
    H["KIOSKO_WAREHOUSE\nel contrato formal que cierra la guia"]
    H --> I["Reporte final para la gerencia\nCIERRA data-modeling-for-analytics-guide"]

Cerrando el checklist de la lección 2 del módulo 1, la guía completa

Las siete filas del checklist que abrió esta guía —en la lección 2 del módulo 1— quedan resueltas, verificadas dos veces cada una: primero en su propio módulo, ahora otra vez, integradas, en este proyecto final. Ninguna pieza del modelado dimensional de Kimball que se propuso esta guía queda pendiente.

Errores comunes

Entregar KIOSKO_WAREHOUSE sin los assert de las ocho partes. Qué pasa: alguien, apurado por mostrar la estructura final como resultado, la construye directamente después de correr las consultas, sin haber pasado por los assert que confirman cada número. Por qué pasa: la estructura de resumen se ve más presentable como "el entregable", y los assert se sienten como pasos preliminares descartables. Cómo detectarlo: si tu entrega final no incluye ninguna evidencia ejecutada de que fact_orders tiene 40 filas con revenue 106.15, que P002 terminó con dos versiones, que el JOIN correcto da snacks y no health-snacks, que el funnel da 35.3%, y que las 4 tablas gold pasan sin discrepancias, estás documentando un proceso sin haber confirmado que funcionó. Cómo corregirlo: los assert de este proyecto —ocho, uno por parte— no son opcionales, son la garantía que hace confiable todo lo que KIOSKO_WAREHOUSE documenta.

Pensar que este proyecto "termina" la guía en el sentido de que no queda nada más por aprender de modelado dimensional. Qué pasa: alguien, al ver el checklist completo con las siete filas resueltas, concluye que ya domina el modelado dimensional por completo, sin margen de crecimiento. Por qué pasa: un checklist completo, con evidencia ejecutada en cada fila, se siente naturalmente como un final definitivo. Cómo detectarlo: si no puedes nombrar, de memoria, al menos cuatro de las guías hermanas que la lección 7 mapeó, perdiste de vista que este proyecto cierra esta guía, no el aprendizaje completo de ingeniería de datos. Cómo corregirlo: este proyecto integra con excelencia todo lo que Kimball y Zach Wilson enseñan sobre modelado dimensional aplicado — pero, como advirtió la lección 7, todavía falta la infraestructura de producción alrededor de ese modelo: orquestación real, versionado como código, time travel nativo, CDC, gobierno formal. Ese es el trabajo de las guías hermanas, no de este proyecto.

Confundir el warehouse de Kiosko, tal como quedó aquí, con un sistema listo para producción real. Qué pasa: alguien, impresionado por la cantidad de piezas integradas —bronze, silver, star historizado, funnel, actividad, OBT, contrato de esquema—, asume que este código, tal cual, podría desplegarse contra un warehouse de producción real de un retailer con millones de órdenes. Por qué pasa: la cantidad de piezas y su coherencia interna dan una sensación de completitud que puede confundirse con "listo para producción". Cómo detectarlo: si tu plan es copiar este script directamente a un entorno de producción sin pasar por ninguna de las guías hermanas de la lección 7, vas a encontrar los mismos ocho huecos que esa lección nombró explícitamente —sin orquestación, sin time travel, sin CDC, sin gobierno—. Cómo corregirlo: este warehouse es el modelo correcto, verificado con evidencia real — pero "correcto" y "listo para producción a escala" son dos afirmaciones distintas. El patrón que aprendiste aquí es exactamente el que necesitarías a cualquier escala; la infraestructura que lo soporta en producción es el trabajo de las ocho guías hermanas.

Ejercicios

Ejercicio 1 — Verifica que el revenue de la OBT, agrupado por día, suma exactamente 106.15. Usando mart_daily_sales_obt, escribe una consulta que agrupe por sale_date y confirme que la suma de los siete días da el revenue total conocido.

print(con.sql("""
    SELECT sale_date, ROUND(SUM(revenue), 2) AS revenue
    FROM mart_daily_sales_obt GROUP BY sale_date ORDER BY sale_date
"""))

┌────────────┬─────────┐
│ sale_date  │ revenue │
│    date    │ double  │
├────────────┼─────────┤
│ 2026-08-03 │   15.85 │
│ 2026-08-04 │   15.85 │
│ 2026-08-05 │    9.55 │
│ 2026-08-06 │   11.05 │
│ 2026-08-07 │   18.05 │
│ 2026-08-08 │   31.85 │
│ 2026-08-09 │    3.95 │
└────────────┴─────────┘

Ejercicio 2 — Extiende KIOSKO_WAREHOUSE con un campo guide_complete que confirme, con un booleano, que las siete piezas del checklist original quedaron resueltas. Sin usar datetime.now(), agrega un campo que documente esta confirmación final.

CHECKLIST_ITEMS = [
    "grain_declared", "star_with_conformed_dimensions", "snowflake_vs_obt",
    "scd_historization", "point_in_time_join", "accumulating_and_cumulative",
    "junk_dimension_and_schema_contract",
]
KIOSKO_WAREHOUSE["guide_complete"] = {
    "checklist_items_resolved": len(CHECKLIST_ITEMS),
    "checklist_items_total": len(CHECKLIST_ITEMS),
    "all_resolved": True,
    "verified_on": "2026-08-09",
}
print(KIOSKO_WAREHOUSE["guide_complete"])

{'checklist_items_resolved': 7, 'checklist_items_total': 7, 'all_resolved': True, 'verified_on': '2026-08-09'}

Ejercicio 3 — Explica, de memoria, cuál de las ocho partes de este proyecto sería la primera en romperse si Kiosko abriera una cuarta tienda mañana. Sin escribir código, describe en 4-6 frases qué le pasaría a cada una de las ocho partes de este warehouse si se agregara S04 al catálogo de tiendas, y cuál requeriría el cambio más profundo.

Resumen y siguiente paso: el final de la guía completa

Con este mini-proyecto cierras el módulo 8 —y data-modeling-for-analytics-guide completa—. Construiste el primer warehouse analítico real de Kiosko: bronze y silver reconstruidos desde foundations (40 filas, revenue 106.15, 0 rechazadas), el star con dim_product_scd historizada (P002 en dos versiones reales, historizadas con MERGE INTO), el join punto-en-el-tiempo demostrado contra el roto (snacks/10.8 correcto, no health-snacks/9.36), fact_sessions y fact_store_activity (funnel 35.3%, activos 7/30 días por tienda), mart_daily_sales_obt publicada con el join correcto ya resuelto para el equipo de BI, y validate_gold_schema() confirmando cero discrepancias sobre las cuatro tablas gold de la guía.

Empezaste esta guía con un fact_orders plano, con llaves naturales, sin historia, heredado tal cual de foundations. La terminas con un warehouse dimensional completo: grano declarado y verificado, star schema con llaves sustitutas y dimensiones conformadas, una comparación con evidencia entre star, snowflake y tabla ancha, una dimensión historizada de verdad con SCD tipo 2, el join correcto contra esa historia, dos patrones de hecho que ningún modelo transaccional resuelve, y un dominio con más de un hecho y más de un tipo de dimensión, con su contrato de esquema verificado. Cada patrón —grano, star, SCD, join punto-en-el-tiempo, accumulating snapshot, cumulative design, dominio feo, contrato Medallion— es el mismo patrón que usa un warehouse de producción a cualquier escala, tal como esta guía advirtió desde su primera lección.

Hacia dónde sigues. La lección 7 de este módulo ya trazó el mapa completo: siete guías hermanas del ecosistema de Data Engineering —dbt-analytics-engineering-guide, airflow-and-declarative-orchestration-guide, spark-and-distributed-processing-guide, lakehouse-and-iceberg-guide, streaming-with-kafka-and-flink-guide, data-reliability-and-governance-guide, python-for-data-engineering-guide— más advanced-sql-querying-guide, vinculada desde el ecosistema de SQL. Elige la que resuelva el hueco que más te importa, y sigue construyendo sobre el warehouse dimensional que dejaste, correcto y verificado, en esta guía.

Recursos

• Kimball Group — "Star Schema / OLAP Cube" — el vocabulario dimensional completo que este proyecto integra de punta a punta. kimballgroup.com/data-warehouse-business-intelligence-resources/kimball-techniques/dimensional-modeling-techniques/star-schema-olap-cube. En inglés.
• "The Data Warehouse Toolkit", 3ra edición (Kimball & Ross, Wiley) — la referencia canónica que sostuvo, de principio a fin, los ocho módulos de esta guía. wiley.com/en-jp/The+Data+Warehouse+Toolkit. En inglés.
• Databricks — "What is the medallion lakehouse architecture?" — el marco bronze/silver/gold que organizó el warehouse completo de este proyecto. docs.databricks.com/aws/en/lakehouse/medallion. En inglés.
• DataExpert-io — repositorio cumulative-table-design (Zach Wilson) — el patrón que sostiene fact_store_activity, integrado en la Parte 5 de este proyecto. github.com/DataExpert-io/cumulative-table-design. En inglés.
• DuckDB — documentación oficial del cliente Python, la interfaz que ejecutó cada verificación de esta guía completa. duckdb.org/docs/current/clients/python/overview. En inglés.