Pipeline de Analytics Portable Multi-Backend

# nutrimetrics_analytics.py
# Pipeline Ibis portable para NutriMetrics SaaS
# -----------------------------------------------
import ibis
from ibis import _
import os

def get_connection() -> ibis.BaseBackend:
    """Devuelve la conexión correcta según ENV=dev|staging|prod."""
    env = os.getenv("NUTRIMETRICS_ENV", "dev")

    if env == "dev":
        # DuckDB local con Parquet — sin levantar ninguna DB
        con = ibis.duckdb.connect("data/nutrimetrics_dev.duckdb")
        con.read_parquet("data/orders.parquet",      table_name="orders")
        con.read_parquet("data/customers.parquet",   table_name="customers")
        con.read_parquet("data/events.parquet",      table_name="events")

    elif env == "staging":
        # PostgreSQL de staging
        con = ibis.postgres.connect(
            url=os.getenv("POSTGRES_URL")
        )

    else:  # prod
        # BigQuery de producción — mismo código, distinto backend
        con = ibis.bigquery.connect(
            project_id="nutrimetrics-prod",
            dataset_id="analytics",
        )

    return con

def monthly_revenue_by_plan(con: ibis.BaseBackend) -> ibis.Table:
    """Revenue mensual por plan (Starter / Pro / Clinic).
    Portable: DuckDB en dev, BigQuery en prod.
    """
    orders = con.table("orders")

    return (
        orders
        .filter(_.status == "completed")
        .group_by(
            month=_.created_at.truncate("M"),
            plan=_.plan,
        )
        .agg(
            revenue_eur    = _.amount.sum(),
            order_count    = _.count(),
            unique_customers = _.customer_id.nunique(),
            avg_order_value = _.amount.mean(),
            mrr_contribution = _.amount.sum() / 1000,  # normalizado en k€
        )
        .order_by([_.month.desc(), _.revenue_eur.desc()])
    )

# Para depurar el SQL generado:
# ibis.to_sql(monthly_revenue_by_plan(con))

def customer_segments(con: ibis.BaseBackend) -> ibis.Table:
    """Segmenta clientes por gasto total con ranking y percentil."""
    orders    = con.table("orders")
    customers = con.table("customers")

    aggregated = (
        orders
        .filter(_.status == "completed")
        .group_by(_.customer_id)
        .agg(
            total_spent   = _.amount.sum(),
            order_count   = _.count(),
            first_order   = _.created_at.min(),
            last_order    = _.created_at.max(),
        )
        .mutate(
            # Ranking global por revenue
            revenue_rank = ibis.rank().over(
                order_by=ibis.desc(_.total_spent)
            ),
            # Segmento por umbral de gasto
            segment = (
                ibis.case()
                    .when(_.total_spent >= 5000, "whale")
                    .when(_.total_spent >= 500,  "regular")
                    .else_("casual")
                    .end()
            ),
            # Días de inactividad
            days_inactive = (
                (ibis.now() - _.last_order).cast("int32") // 86400
            ),
        )
    )

    return (
        aggregated
        .join(customers, _.customer_id == customers.id)
        .select(
            _.customer_id, customers.name,
            customers.plan, _.total_spent,
            _.segment, _.revenue_rank, _.days_inactive,
        )
        .order_by(_.revenue_rank)
        .limit(1000)
    )

def active_users_by_plan(con: ibis.BaseBackend) -> ibis.Table:
    """DAU / WAU / MAU por plan de suscripción.
    Calcula los últimos 30 días sin levantar todo a memoria.
    """
    events = con.table("events")

    dau = (
        events
        .select(_.user_id, _.event_date, _.plan)
        .distinct()                          # 1 entrada por user/día
        .filter(_.event_date >= ibis.date("2026-05-17"))
        .group_by(event_date=_.event_date, plan=_.plan)
        .agg(dau=_.user_id.nunique())
    )

    mau_window = ibis.trailing_range_window(
        preceding=ibis.interval(days=30),
        order_by=_.event_date,
        group_by=_.plan,
    )

    return (
        dau
        .mutate(mau_30d=_.dau.sum().over(mau_window))
        .mutate(stickiness=(_.dau / _.mau_30d).round(3))
        .order_by([_.event_date.desc(), _.dau.desc()])
    )

def cohort_retention(con: ibis.BaseBackend) -> ibis.Table:
    """Análisis de cohortes: % usuarios activos por mes desde el alta."""
    orders = con.table("orders")

    with_cohort = (
        orders
        .filter(_.status == "completed")
        .group_by(_.customer_id)
        .mutate(
            # Mes de alta del cliente
            cohort_month=_.created_at.min().truncate("M"),
        )
        .mutate(
            # Meses desde el alta en cada transacción
            months_since=(
                (_.created_at.truncate("M") - _.cohort_month)
                .cast("int32") // (30 * 86400)
            ),
        )
    )

    cohort_sizes = (
        with_cohort
        .filter(_.months_since == 0)  # Usuarios del primer mes
        .group_by(_.cohort_month)
        .agg(cohort_size=_.customer_id.nunique())
    )

    activity = (
        with_cohort
        .group_by(_.cohort_month, _.months_since)
        .agg(active_users=_.customer_id.nunique())
    )

    return (
        activity
        .join(cohort_sizes, "cohort_month")
        .mutate(retention_pct=(
            _.active_users * 100.0 / _.cohort_size
        ).round(1))
        .order_by([_.cohort_month, _.months_since])
    )

#!/usr/bin/env python3
# Ejecutar: python run_analytics.py
# Con BigQuery: NUTRIMETRICS_ENV=prod python run_analytics.py

from nutrimetrics_analytics import (
    get_connection,
    monthly_revenue_by_plan,
    customer_segments,
    active_users_by_plan,
    cohort_retention,
)
import ibis

def main():
    con = get_connection()
    print(f"Backend: {con.name}")  # "duckdb" o "bigquery"

    # 1. Revenue mensual
    rev = monthly_revenue_by_plan(con).execute()
    rev.to_parquet("output/revenue.parquet")

    # 2. Segmentos de clientes
    segs = customer_segments(con).execute()
    segs.to_csv("output/segments.csv", index=False)

    # 3. Active users
    dau = active_users_by_plan(con).execute()

    # 4. Cohortes
    cohorts = cohort_retention(con).execute()

    print("Revenue (últimas 3 filas):")
    print(rev.tail(3).to_string())
    print(f"Segmentos procesados: {len(segs):,}")
    print(f"Filas de cohorte: {len(cohorts):,}")

if __name__ == "__main__":
    main()