Día 2: El pipeline de unificacion

1Ivan Romero [09:05 - pizarra]:
2
3Opcion A: día + campana_original + fuente
4  PRO: No pierdes nada. Puedes comparar Google vs Meta por día.
5  CON: Para calcular ROAS unificado necesitas agregar después.
6
7Opcion B: día + campaign_group (unificado)
8  PRO: Ya tienes la vision unificada lista.
9  CON: Pierdes la trazabilidad a la fuente original.
10       Si algo no cuadra, no sabes de donde viene el error.
11
12Opcion C: semana + campaign_group
13  PRO: Nivel de email, todo comparable.
14  CON: Pierdes detalle diario de Google y Meta.
15       Carmen no va a aceptar perder esa granularidad.
16
17Mi recomendación: OPCION A.
18Siempre normaliza al nivel más granular posible.
19Agregar después es trivial. Desagregar es imposible.

Ivan explica por que normalizar al nivel más granular -- una decisión de diseño clave

1-- Tabla destino: unified_campaigns
2-- Granularidad: 1 fila por día + campaña original + fuente
3CREATE TABLE unified_campaigns (
4  id SERIAL PRIMARY KEY,
5  date DATE NOT NULL,
6  source VARCHAR(20) NOT NULL,            -- 'google', 'meta', 'email'
7  campaign_group VARCHAR(100),            -- nombre unificado (de mapping)
8  original_campaign_name VARCHAR(200),    -- nombre original de la plataforma
9  product_category VARCHAR(50),           -- categoría de producto (de mapping)
10  impressions BIGINT DEFAULT 0,
11  clicks INTEGER DEFAULT 0,
12  conversions INTEGER DEFAULT 0,
13  spend_eur NUMERIC(12,2) DEFAULT 0,
14  revenue_eur NUMERIC(12,2) DEFAULT 0,   -- solo email tiene revenue directo
15  is_mapped BOOLEAN DEFAULT FALSE,        -- tiene mapping o es huerfana?
16  loaded_at TIMESTAMP DEFAULT NOW()
17);
18
19-- Indices para queries frecuentes
20CREATE INDEX idx_uc_date_source ON unified_campaigns(date, source);
21CREATE INDEX idx_uc_campaign_group ON unified_campaigns(campaign_group);
22CREATE INDEX idx_uc_unmapped ON unified_campaigns(is_mapped) WHERE is_mapped = FALSE;
23
24-- Constraint: no duplicados (idempotencia)
25CREATE UNIQUE INDEX idx_uc_unique ON unified_campaigns(date, source, original_campaign_name);

Esquema final: cada fila es un día-campaña-fuente, con flag de mapping

1#canal-pharmalife [10:15]
2
3Raul Merino: Oye, sobre lo de distribuir el email a diario.
4En realidad los envios van los martes y jueves (2 envios
5por semana para cada campaña). Los martes el open rate es
6un 15% más alto que los jueves.
7
8Si quieres ser más preciso, podrias distribuir:
9- Martes: 35% del total semanal
10- Miercoles: 25% (aperturas rezagadas del martes)
11- Jueves: 25% (segundo envio)
12- Viernes: 10% (rezagadas del jueves)
13- Resto: 5%
14
15Pero para el MVP, 1/7 por día está bien. Ya refinamos después.
16
17Ivan Romero: @Raul gracias por el dato. @Tu anota esto
18como mejora para v2 en el README. Por ahora 1/7 uniforme.

Contexto de negocio de Raul: la realidad del email no es uniforme

1import pandas as pd
2from datetime import timedelta
3
4def distribute_weekly_to_daily(df_email: pd.DataFrame) -> pd.DataFrame:
5    """
6    Distribuye métricas semanales en 7 filas diarias.
7    Cada día recibe 1/7 de las métricas.
8    """
9    daily_rows = []
10    for _, row in df_email.iterrows():
11        week_start = pd.to_datetime(row['week_start'])
12        for day_offset in range(7):
13            day = week_start + timedelta(days=day_offset)
14            daily_rows.append({
15                'date': day.strftime('%Y-%m-%d'),
16                'source': 'email',
17                'original_campaign_name': row['campaign_name'],
18                'impressions': 0,  # email no tiene impresiones
19                'clicks': round(row['clicked'] / 7),
20                'conversions': round(row['converted'] / 7),
21                'spend_eur': round(row['revenue_eur'] * 0.10 / 7, 2),  # coste = 10% revenue
22                'revenue_eur': round(row['revenue_eur'] / 7, 2),
23            })
24    return pd.DataFrame(daily_rows)
25
26# Ejemplo con 1 fila semanal:
27# Weekly_Vitamins_Promo: 89 conversiones, 8900 EUR revenue
28# -> 7 filas diarias: ~13 conversiones/día, ~1271 EUR/día

Distribucion de email semanal a diario: cada día recibe 1/7

1import pandas as pd
2import logging
3
4logging.basicConfig(level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s')
5logger = logging.getLogger('adpulse.unify')
6
7def load_mapping(path: str) -> pd.DataFrame:
8    """Carga la mapping table de Raul."""
9    mapping = pd.read_csv(path)
10    logger.info(f"Mapping cargado: {len(mapping)} campaign groups")
11    return mapping
12
13def enrich_with_mapping(df: pd.DataFrame, mapping: pd.DataFrame, source: str) -> pd.DataFrame:
14    """Enriquece un DataFrame con campaign_group y product_category del mapping."""
15    # Determinar la columna de join segun la fuente
16    join_cols = {
17        'google': 'google_name',
18        'meta': 'meta_name',
19        'email': 'email_name',
20    }
21    join_col = join_cols[source]
22
23    # LEFT JOIN con mapping
24    merged = df.merge(
25        mapping[['campaign_group', join_col, 'product_category']],
26        left_on='original_campaign_name',
27        right_on=join_col,
28        how='left'
29    )
30
31    df['campaign_group'] = merged['campaign_group']
32    df['product_category'] = merged['product_category']
33    df['is_mapped'] = df['campaign_group'].notna()
34
35    # Loguear huerfanas
36    orphans = df[~df['is_mapped']]['original_campaign_name'].unique()
37    if len(orphans) > 0:
38        logger.warning(f"[{source}] {len(orphans)} campañas sin mapping: {list(orphans)}")
39
40    return df
41
42def run_pipeline(google_path: str, meta_path: str, email_path: str, mapping_path: str):
43    """Pipeline principal: ingesta + normalizacion + unificacion."""
44    mapping = load_mapping(mapping_path)
45
46    # 1. Google Ads
47    df_google = transform_google(pd.read_csv(google_path))
48    df_google = enrich_with_mapping(df_google, mapping, 'google')
49    logger.info(f"Google: {len(df_google)} filas, {df_google['is_mapped'].sum()} mapeadas")
50
51    # 2. Meta Ads
52    df_meta = transform_meta(load_meta_json(meta_path))
53    df_meta = enrich_with_mapping(df_meta, mapping, 'meta')
54    logger.info(f"Meta: {len(df_meta)} filas, {df_meta['is_mapped'].sum()} mapeadas")
55
56    # 3. Email
57    df_email = transform_email(pd.read_csv(email_path))
58    df_email = enrich_with_mapping(df_email, mapping, 'email')
59    logger.info(f"Email: {len(df_email)} filas, {df_email['is_mapped'].sum()} mapeadas")
60
61    # 4. Concatenar y devolver
62    unified = pd.concat([df_google, df_meta, df_email], ignore_index=True)
63    logger.info(f"TOTAL UNIFICADO: {len(unified)} filas")
64    logger.info(f"  Mapeadas: {unified['is_mapped'].sum()} ({unified['is_mapped'].mean()*100:.1f}%)")
65    logger.info(f"  Huerfanas: {(~unified['is_mapped']).sum()}")
66
67    return unified

Pipeline completo: patron ETL con logging y detección de huerfanas

1PR #247: feat(pharmalife): pipeline de unificacion v1
2
3Files changed: 8  |  Additions: 342  |  Deletions: 0
4
5Ivan Romero reviewed -- APPROVED with comments
6
7Comment 1 (transform_meta.py, line 34):
8> Estas usando .iterrows() para iterar el DataFrame.
9> Para 31 días x 12 campañas (372 filas) no importa, pero es un
10> mal hábito. Usa vectorizacion de Pandas cuando puedas. Si no
11> puedes, al menos documenta por que iterrows es necesario.
12> En este caso SI puedes vectorizar con .apply() o .explode().
13> Aprobado pero deja un TODO para refactorizarlo.
14
15Comment 2 (enrich_with_mapping.py, line 18):
16> Bien el LEFT JOIN. Pregunta: que pasa si la mapping table tiene
17> un google_name duplicado? (por ejemplo si Raul comete un error
18> y mapea 2 campaign_groups al mismo nombre de Google). Tu código
19> crearia filas duplicadas silenciosamente. Anade un check de
20> unicidad ANTES del merge. Algo tipo:
21>   assert mapping['google_name'].dropna().duplicated().sum() == 0
22> Esto es OBLIGATORIO antes de merge.
23
24Comment 3 (run_pipeline.py, line 52):
25> Me gusta que loguees las huerfanas. Pero el log se pierde si
26> nadie lo mira. Para la v2, quiero que las huerfanas se guarden
27> en una tabla separada (orphan_campaigns) y que el DAG envie
28> un Slack automático a Raul. Por ahora está bien como MVP.
29
30Comment 4 (general):
31> Buen naming en las funciones: transform_google, transform_meta,
32> enrich_with_mapping. Sigue así. El único que cambiaria es
33> "run_pipeline" -> "run_pharmalife_weekly" (más específico).
34
35OVERALL: Aprobado con 1 cambio obligatorio (check de unicidad)
36         y 3 sugerencias para v2. Merge cuando arregles el #2.

Code review real: Ivan aprueba pero señala mejoras con ejemplos concretos

1#canal-pharmalife [17:02]
2
3Raul Merino: Hey! Ivan me dice que encontraste 5 campañas sin
4mapping (3 Google, 2 Meta). Me las pasas?
5
6Tu: Si, aquí van:
7  Google:
8    - PL_Search_NewProduct_Launch (nueva de está semana)
9    - PL_Display_Test_Audience_A (test de audiencia)
10    - PL_Shopping_Bundle_Winter (bundle de invierno)
11  Meta:
12    - pharmalife_protein_launch_broad_fb (nueva)
13    - pharmalife_test_creative_v2_ig (test de creative)
14
15Raul Merino: Perfecto. La de "NewProduct_Launch" va a ser
16"Protein Launch" en mi mapping. La de Meta "protein_launch"
17es la misma campaña! Van al mismo campaign_group.
18
19Las de "test" las dejo fuera del mapping -- son tests que no
20deberian contabilizarse para ROAS. Puedes filtrarlas?
21
22Tu: Si, puedo anadir un filtro que excluya campañas con "test"
23o "Test" en el nombre. Te parece?
24
25Raul Merino: Perfecto. Gestion 10/10. Actualizare el sheet
26mañana a primera hora.

Coordinacion con Raul: el data engineer y el analista trabajan juntos