Día 4: Atribución multi-touch y pipeline completo

1#canal-pharmalife [08:35]
2
3Iván Romero: Buenos días. Antes de la call con Ana quiero
4revisar tu pipeline completo. Sube el PR con todo:
51. Script de ingesta (3 fuentes)
62. Unificacion con mapping
73. Reconciliacion proporcional
84. Atribucion lineal (como análisis adicional)
95. Query del report semanal
10
11Quiero ver el PR listo para merge antes de las 9:30.
12Si algo no está claro, me preguntas ahora.
13
14Carmen Reyes: @Iván @Tu - la call con Ana es a las 10.
15Repasamos a las 9:30 en la sala ROAS. Traed el resumen
16ejecutivo impreso (si, en papel, Ana es old school).

La mañana del viernes: pipeline completo + presentación al cliente

1pipelines/pharmalife/
2  README.md                    <- cómo ejecutar, dependencias, contacto
3  config.py                    <- constantes (paths, factor, thresholds)
4  01_ingest_google.py          <- carga CSV de S3 a stg_google
5  02_ingest_meta.py            <- pull API Meta, aplana JSON, carga stg_meta
6  03_ingest_email.py           <- carga CSV email, distribuye a diario
7  04_unify.py                  <- JOIN con mapping, carga unified_campaigns
8  05_reconcile.py              <- calcula factor, genera tabla reconciliada
9  06_attribution_linear.py     <- atribución lineal desde CRM
10  07_report_weekly.py          <- genera informe semanal
11  08_detect_orphans.py         <- detecta campañas sin mapping, alerta Slack
12  sql/
13    create_tables.sql          <- DDL de todas las tablas
14    quality_checks.sql         <- queries de validación post-carga
15    weekly_report.sql          <- query del report semanal
16  tests/
17    test_transform_meta.py     <- test unitario de la transformacion Meta
18    test_mapping_uniqueness.py <- verifica que mapping no tiene duplicados
19    test_factor_range.py       <- factor debe estar entre 0.5 y 0.9

Estructura del pipeline: un script por paso, SQL separado, tests basicos

1from typing import Dict, List, Tuple
2
3def time_decay_attribution(
4    amount: float,
5    touchpoints: List[Tuple[str, str]],
6    decay_factor: float = 0.5
7) -> Dict[str, float]:
8    """
9    Atribucion time-decay: los touchpoints más recientes reciben más credito.
10    
11    El último touchpoint recibe peso 1.0, el penultimo 0.5 (decay_factor),
12    el anterior 0.25 (decay_factor^2), etc.
13    
14    Ejemplo con decay_factor=0.5 y 3 touchpoints:
15      Pesos brutos: [0.25, 0.5, 1.0] (del primero al último)
16      Pesos normalizados: [0.143, 0.286, 0.571] (suman 1.0)
17      Revenue: [amount*0.143, amount*0.286, amount*0.571]
18    """
19    n = len(touchpoints)
20    if n == 0:
21        return {}
22    
23    # Calcular pesos (último = 1.0, penultimo = decay, etc.)
24    weights = [decay_factor ** (n - 1 - i) for i in range(n)]
25    total_weight = sum(weights)
26    
27    # Normalizar y distribuir
28    result: Dict[str, float] = {}
29    for i, (source, _) in enumerate(touchpoints):
30        normalized_weight = weights[i] / total_weight
31        revenue_share = amount * normalized_weight
32        result[source] = result.get(source, 0) + revenue_share
33    
34    return result
35
36# Ejemplo: venta de 90 EUR, touchpoints: meta -> email -> google
37tp = [('meta', 'pharmalife_Z'), ('email', 'Weekly_Y'), ('google', 'PL_Search_X')]
38result = time_decay_attribution(90.0, tp, decay_factor=0.5)
39# meta:   90 * 0.25/1.75 = 12.86 EUR (14.3%)
40# email:  90 * 0.50/1.75 = 25.71 EUR (28.6%)
41# google: 90 * 1.00/1.75 = 51.43 EUR (57.1%)
42print(result)  # {'meta': 12.86, 'email': 25.71, 'google': 51.43}

Time-decay: el último canal recibe más credito — favorece canales de cierre

1PR #251: feat(pharmalife): pipeline completo v1 + atribución
2
3Iván Romero [09:25]:
4> Aprobado. 3 comentarios menores para la v2:
5>
6> 1. config.py: el factor 0.6688 está hardcodeado. Para la v2
7>    deberia calcularse dinamicamente cada semana. Deja un TODO.
8>
9> 2. 06_attribution_linear.py: buena implementación. Me gusta
10>    que separaste parse_touchpoints() como función pura testeable.
11>    Anade un test con un caso edge: venta con un solo touchpoint.
12>
13> 3. 08_detect_orphans.py: perfecto que generes el mensaje de
14>    Slack formateado. Pero falta el pasó de ENVIARLO realmente.
15>    Para el MVP, printea el mensaje y que alguien lo copie.
16>    Para la v2, usa la API de Slack (webhook).
17>
18> GENERAL: Buen trabajo para 4 días. La estructura es limpia,
19> la nomenclatura es consistente y el README explica bien el
20> flujo. Merge aprobado.

PR aprobado: feedback constructivo para iterar después de la presentación

1IDEMPOTENCIA = ejecutar N veces produce el mismo resultado que ejecutar 1 vez.
2
3Patron 1 (el nuestro): ON CONFLICT DO NOTHING
4  - Si la fila ya existe, no hace nada. Simple pero no actualiza cambios.
5
6Patron 2: DELETE + INSERT (truncate and reload)
7  - Borra todo del período y reinsertas. Mas agresivo pero garantiza freshness.
8
9Patron 3: ON CONFLICT DO UPDATE (upsert)
10  - Si existe, actualiza. Si no, inserta. El más completo pero más complejo.
11
12Para el MVP usamos Patron 1. Para producción moveremos a Patron 2
13(DELETE WHERE date BETWEEN start AND end, luego INSERT).
14Es más limpio para re-ejecuciones y permite corregir datos históricos.

Idempotencia: un pipeline que puedes re-ejecutar sin miedo a duplicar