José Manuel Gómez
ff3aafe82a
- Compute monthly_verticals in run.py and send_slack_report.py using per-vertical CPL targets - Display vertical breakdown (spend, leads, CPL, target, margin) inside the monthly table block - Add alarmas vertical (CPL target 8€) to Baserow Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
423 lines
18 KiB
Python
423 lines
18 KiB
Python
"""Meta Optimizer — main entry point."""
|
|
import sys
|
|
import io
|
|
import os
|
|
import time
|
|
|
|
from datetime import datetime
|
|
|
|
import config
|
|
from meta_ads_client import MetaAdsClient
|
|
from agent import decide, analyze_creative, analyze_unit
|
|
from baserow_client import BaserowClient
|
|
import slack_notifier
|
|
|
|
|
|
_ACTION_MAP = {
|
|
"PAUSE": "PAUSE",
|
|
"REDUCE_BUDGET": "REDUCE_BUDGET",
|
|
"INCREASE_BUDGET": "INCREASE_BUDGET",
|
|
"MAINTAIN": "MAINTAIN",
|
|
"REVIEW_CREATIVES": "REVIEW_CREATIVES",
|
|
# legacy Spanish names
|
|
"PAUSAR": "PAUSE",
|
|
"REDUCIR_PRESUPUESTO": "REDUCE_BUDGET",
|
|
"AUMENTAR_PRESUPUESTO": "INCREASE_BUDGET",
|
|
"MANTENER": "MAINTAIN",
|
|
"REVISAR_CREATIVIDADES":"REVIEW_CREATIVES",
|
|
}
|
|
|
|
|
|
def _extract_vertical(name: str) -> str:
|
|
"""VIVIFUL_13_telefonia_leadads → telefonia"""
|
|
prefix = config.META_CAMPAIGN_PREFIX
|
|
rest = name[len(prefix):].lstrip("_")
|
|
parts = rest.split("_")
|
|
start = 1 if parts and parts[0].isdigit() else 0
|
|
return parts[start].lower() if start < len(parts) else "otros"
|
|
|
|
|
|
class Tee:
|
|
def __init__(self, filepath: str):
|
|
os.makedirs(os.path.dirname(filepath), exist_ok=True)
|
|
self._file = open(filepath, "w", encoding="utf-8")
|
|
self._stdout = sys.stdout
|
|
|
|
def write(self, data):
|
|
self._stdout.write(data)
|
|
self._file.write(data)
|
|
|
|
def flush(self):
|
|
self._stdout.flush()
|
|
if not self._file.closed:
|
|
self._file.flush()
|
|
|
|
def close(self):
|
|
self._file.close()
|
|
|
|
|
|
def _execute_action(meta: MetaAdsClient, action: dict):
|
|
"""Apply an approved action via Meta API."""
|
|
action_type = action.get("action_type", "")
|
|
cid = action.get("campaign_id", "")
|
|
parameter = float(action.get("parameter") or 1.0)
|
|
|
|
if action_type == "PAUSE":
|
|
if cid.startswith("ad:"):
|
|
meta.pause_ad(cid[3:])
|
|
else:
|
|
meta.pause_campaign(cid)
|
|
|
|
elif action_type in ("REDUCE_BUDGET", "INCREASE_BUDGET"):
|
|
try:
|
|
bid_cfg = meta.get_campaign_bid_config(cid)
|
|
current_budget = bid_cfg.get("daily_budget_eur")
|
|
if current_budget:
|
|
meta.set_campaign_budget(cid, int(current_budget * parameter * 100))
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def run():
|
|
start_ts = time.time()
|
|
now = datetime.now()
|
|
|
|
print(f"\n{'='*55}")
|
|
print(f" META OPTIMIZER — {now.strftime('%d/%m/%Y %H:%M')}")
|
|
print(f" Prefix: {config.META_CAMPAIGN_PREFIX} | Target CPL: {config.META_TARGET_CPL}€")
|
|
print(f" Mode: {'DRY RUN (no changes)' if config.DRY_RUN else 'PRODUCTION'}")
|
|
print(f"{'='*55}\n")
|
|
|
|
meta = MetaAdsClient()
|
|
baserow = BaserowClient()
|
|
|
|
# ── Fetch all vertical CPL targets upfront ────────────────────────────────
|
|
vertical_cpls: dict = {}
|
|
try:
|
|
for v in baserow.get_all_verticals():
|
|
name = (v.get("Nombre") or "").strip().lower()
|
|
cpl = float(v.get("target_cpl") or 0)
|
|
if name and cpl:
|
|
vertical_cpls[name] = cpl
|
|
except Exception:
|
|
pass
|
|
|
|
# ── Execute previously approved actions ───────────────────────────────────
|
|
actions_executed = 0
|
|
if not config.DRY_RUN:
|
|
approved = baserow.get_approved_actions()
|
|
print(f"→ Executing {len(approved)} approved actions...\n")
|
|
for action in approved:
|
|
try:
|
|
_execute_action(meta, action)
|
|
baserow.update_action_status(action["id"], "executed")
|
|
actions_executed += 1
|
|
print(f" ✓ {action.get('campaign_name')} — {action.get('action_type')}")
|
|
except Exception as e:
|
|
print(f" ✗ Error on action {action['id']}: {e}")
|
|
|
|
# ── Monthly daily totals (per-campaign rows → aggregate with margins) ─────
|
|
print(f"→ Fetching monthly daily totals for {config.META_CAMPAIGN_PREFIX}...")
|
|
daily_rows = meta.get_monthly_daily_totals()
|
|
_daily: dict = {}
|
|
monthly_verticals: dict = {}
|
|
for row in daily_rows:
|
|
v = _extract_vertical(row["campaign_name"])
|
|
target = vertical_cpls.get(v, config.META_TARGET_CPL)
|
|
margin = round((target - row["spend"] / row["leads"]) * row["leads"], 2) if row["leads"] > 0 else round(-row["spend"], 2)
|
|
d = _daily.setdefault(row["date"], {"spend": 0.0, "leads": 0, "margin": 0.0})
|
|
d["spend"] += row["spend"]
|
|
d["leads"] += row["leads"]
|
|
d["margin"] += margin
|
|
mv = monthly_verticals.setdefault(v, {"spend": 0.0, "leads": 0, "margin": 0.0, "target_cpl": target})
|
|
mv["spend"] += row["spend"]
|
|
mv["leads"] += row["leads"]
|
|
mv["margin"] += margin
|
|
daily_totals = [
|
|
{
|
|
"date": date,
|
|
"spend": round(d["spend"], 2),
|
|
"leads": int(d["leads"]),
|
|
"cpl": round(d["spend"] / d["leads"], 2) if d["leads"] > 0 else 0.0,
|
|
"margin": round(d["margin"], 2),
|
|
}
|
|
for date, d in sorted(_daily.items())
|
|
]
|
|
print(f" ✓ {len(daily_totals)} days with data.\n")
|
|
|
|
# ── Yesterday metrics ─────────────────────────────────────────────────────
|
|
print(f"→ Fetching yesterday metrics ({config.META_CAMPAIGN_PREFIX} only, spend > 0)...")
|
|
metrics_all = meta.get_yesterday_metrics()
|
|
print(f" ✓ {len(metrics_all)} campaigns active yesterday.\n")
|
|
|
|
# ── 3-day and 7-day metrics for trend analysis ────────────────────────────
|
|
print(f"→ Fetching 3-day and 7-day metrics...")
|
|
metrics_3d = meta.get_period_campaign_metrics(days=3)
|
|
metrics_7d = meta.get_period_campaign_metrics(days=7)
|
|
print(f" ✓ Multi-window data ready.\n")
|
|
|
|
# ── Analyze campaigns & propose actions ───────────────────────────────────
|
|
actions_proposed_list = []
|
|
campaign_details = {} # {cid: {vertical, margin, adsets, ads}}
|
|
verticals = {} # {vertical: {spend, leads, margin}}
|
|
errors = []
|
|
|
|
for cid, metrics in metrics_all.items():
|
|
vertical = _extract_vertical(metrics["name"])
|
|
max_cpl = vertical_cpls.get(vertical, config.META_TARGET_CPL) or config.META_TARGET_CPL
|
|
margin = round((max_cpl - metrics["cpl"]) * metrics["leads"], 2) if metrics["leads"] > 0 else round(-metrics["spend"], 2)
|
|
|
|
m3 = metrics_3d.get(cid, {})
|
|
m7 = metrics_7d.get(cid, {})
|
|
analysis = {
|
|
"campaign_id": cid,
|
|
"name": metrics["name"],
|
|
"status": metrics["status"],
|
|
# Yesterday (1d) — contexto inmediato
|
|
"spend": metrics["spend"],
|
|
"leads": metrics["leads"],
|
|
"cpl": metrics["cpl"],
|
|
"max_cpl": max_cpl,
|
|
"ctr": metrics["ctr"],
|
|
"cpm": metrics["cpm"],
|
|
"impressions": metrics["impressions"],
|
|
"clicks": metrics["clicks"],
|
|
# 3 días — ventana principal para decisiones
|
|
"spend_3d": m3.get("spend", 0.0),
|
|
"leads_3d": m3.get("leads", 0),
|
|
"cpl_3d": m3.get("cpl", 0.0),
|
|
# 7 días — tendencia de fondo
|
|
"spend_7d": m7.get("spend", 0.0),
|
|
"leads_7d": m7.get("leads", 0),
|
|
"cpl_7d": m7.get("cpl", 0.0),
|
|
}
|
|
|
|
try:
|
|
decision = decide(analysis)
|
|
except Exception as e:
|
|
errors.append(f"{metrics['name']}: {e}")
|
|
continue
|
|
|
|
action_type = _ACTION_MAP.get(
|
|
decision.get("action") or decision.get("accion", "MAINTAIN"),
|
|
"MAINTAIN",
|
|
)
|
|
|
|
# ── Bid config (campaña + adsets) ─────────────────────────────────
|
|
campaign_bid = {}
|
|
try:
|
|
campaign_bid = meta.get_campaign_bid_config(cid)
|
|
except Exception as e:
|
|
errors.append(f"Bid config {metrics['name']}: {e}")
|
|
|
|
adset_bids = {}
|
|
try:
|
|
adset_bids = meta.get_adset_bid_configs(cid)
|
|
except Exception as e:
|
|
errors.append(f"Adset bids {metrics['name']}: {e}")
|
|
|
|
# ABO campaigns (presupuesto solo a nivel de conjunto): omitir ajustes de campaña
|
|
is_cbo = campaign_bid.get("daily_budget_eur") is not None
|
|
if action_type in ("INCREASE_BUDGET", "REDUCE_BUDGET") and not is_cbo:
|
|
action_type = "MAINTAIN"
|
|
|
|
print(f" {metrics['name'][:52]}")
|
|
print(f" CPL: 1d={metrics['cpl']}€ / 3d={m3.get('cpl',0.0):.2f}€ / 7d={m7.get('cpl',0.0):.2f}€ MaxCPL: {max_cpl}€ Margen: {margin:+.2f}€")
|
|
print(f" Leads: 1d={metrics['leads']} / 3d={m3.get('leads',0)} / 7d={m7.get('leads',0)} Spend 1d: {metrics['spend']}€")
|
|
print(f" Vertical: {vertical} Decision: {action_type} — {(decision.get('justification') or '')[:70]}")
|
|
if decision.get("alert"):
|
|
print(f" ALERT: {decision['alert']}")
|
|
print()
|
|
|
|
if action_type != "MAINTAIN":
|
|
try:
|
|
row = baserow.save_action({
|
|
"campaign_id": cid,
|
|
"campaign_name": metrics["name"],
|
|
"action_type": action_type,
|
|
"parameter": decision.get("parameter") or 1.0,
|
|
"justification": decision.get("justification") or "",
|
|
"advice": decision.get("advice") or "",
|
|
"alert": decision.get("alert") or "",
|
|
"confidence": decision.get("confidence") or 0.0,
|
|
})
|
|
actions_proposed_list.append({
|
|
"campaign_name": metrics["name"],
|
|
"action_type": action_type,
|
|
"parameter": decision.get("parameter") or 1.0,
|
|
"justification": decision.get("justification") or "",
|
|
"advice": decision.get("advice") or "",
|
|
"alert": decision.get("alert") or "",
|
|
"confidence": decision.get("confidence") or 0.0,
|
|
"cpl": metrics["cpl"],
|
|
"max_cpl": max_cpl,
|
|
"row_id": row["id"],
|
|
})
|
|
except Exception as e:
|
|
errors.append(f"Save action {metrics['name']}: {e}")
|
|
|
|
# ── Ad set analysis ────────────────────────────────────────────────
|
|
adsets_detail = []
|
|
try:
|
|
for as_m in meta.get_period_adset_metrics(cid, days=3)[:5]:
|
|
bid = adset_bids.get(as_m["id"], {})
|
|
as_m["bid_strategy"] = bid.get("bid_strategy", "")
|
|
as_m["cost_cap_eur"] = bid.get("cost_cap_eur")
|
|
result = analyze_unit(as_m, "adset")
|
|
adsets_detail.append({**as_m, **result})
|
|
print(f" [Adset] {as_m['name'][:45]} — {result.get('evaluacion','')[:60]}")
|
|
except Exception as e:
|
|
errors.append(f"Adsets {metrics['name']}: {e}")
|
|
|
|
# ── Ad analysis ────────────────────────────────────────────────────
|
|
ads_detail = []
|
|
try:
|
|
for ad_m in meta.get_period_ad_metrics(cid, days=3)[:5]:
|
|
ad_m["max_cpl"] = max_cpl
|
|
result = analyze_unit(ad_m, "ad")
|
|
ad_entry = {**ad_m, **result}
|
|
# Propose ad pause if Claude recommends it
|
|
if result.get("accion") == "PAUSE":
|
|
try:
|
|
ad_row = baserow.save_action({
|
|
"campaign_id": f"ad:{ad_m['id']}",
|
|
"campaign_name": ad_m["name"],
|
|
"action_type": "PAUSE",
|
|
"parameter": 1.0,
|
|
"justification": result.get("recomendacion", ""),
|
|
"advice": result.get("evaluacion", ""),
|
|
"confidence": 0.8,
|
|
})
|
|
ad_entry["row_id"] = ad_row["id"]
|
|
except Exception as e:
|
|
errors.append(f"Ad action {ad_m['name']}: {e}")
|
|
ads_detail.append(ad_entry)
|
|
action_tag = " ⛔PAUSE" if result.get("accion") == "PAUSE" else ""
|
|
print(f" [Ad] {ad_m['name'][:45]} — {result.get('evaluacion','')[:60]}{action_tag}")
|
|
except Exception as e:
|
|
errors.append(f"Ads {metrics['name']}: {e}")
|
|
|
|
campaign_details[cid] = {
|
|
"name": metrics["name"],
|
|
"vertical": vertical,
|
|
"margin": margin,
|
|
"adsets": adsets_detail,
|
|
"ads": ads_detail,
|
|
"bid_config": campaign_bid,
|
|
}
|
|
|
|
# ── Daily snapshot (persists analysis to Baserow for dashboard) ───────
|
|
try:
|
|
action_info = next(
|
|
(a for a in actions_proposed_list if a["campaign_name"] == metrics["name"]),
|
|
None,
|
|
)
|
|
baserow.save_daily_snapshot({
|
|
"run_date": now.strftime("%Y-%m-%d"),
|
|
"campaign_id": cid,
|
|
"campaign_name": metrics["name"],
|
|
"vertical": vertical,
|
|
"spend": metrics["spend"],
|
|
"leads": metrics["leads"],
|
|
"cpl": metrics["cpl"],
|
|
"margin": margin,
|
|
"action_type": action_info["action_type"] if action_info else "MAINTAIN",
|
|
"justification": action_info["justification"] if action_info else "",
|
|
"adsets": adsets_detail,
|
|
"ads": ads_detail,
|
|
})
|
|
except Exception as e:
|
|
errors.append(f"Snapshot {metrics['name']}: {e}")
|
|
|
|
# ── Vertical aggregation ───────────────────────────────────────────
|
|
if vertical not in verticals:
|
|
verticals[vertical] = {"spend": 0.0, "leads": 0, "margin": 0.0, "target_cpl": max_cpl}
|
|
verticals[vertical]["spend"] += metrics["spend"]
|
|
verticals[vertical]["leads"] += metrics["leads"]
|
|
verticals[vertical]["margin"] += margin
|
|
|
|
# ── Creative visual analysis (Baserow storage) ─────────────────────
|
|
try:
|
|
for ad in meta.get_ads_with_creatives(cid):
|
|
if not ad["thumbnail_url"]:
|
|
continue
|
|
result = analyze_creative(ad["thumbnail_url"], ad["ad_name"])
|
|
baserow.save_creative_analysis({
|
|
"ad_id": ad["ad_id"],
|
|
"ad_name": ad["ad_name"],
|
|
"campaign_id": cid,
|
|
"image_url": ad["thumbnail_url"],
|
|
"analysis": result.get("analysis", ""),
|
|
"score": result.get("score", 0),
|
|
"recommendations": result.get("recommendations", ""),
|
|
})
|
|
except Exception as e:
|
|
errors.append(f"Creatives {metrics['name']}: {e}")
|
|
|
|
# ── Top 10 best and worst ─────────────────────────────────────────────────
|
|
with_leads = [m for m in metrics_all.values() if m["leads"] > 0]
|
|
best_10 = sorted(with_leads, key=lambda x: x["cpl"])[:10]
|
|
|
|
all_active = list(metrics_all.values())
|
|
worst_10 = sorted(
|
|
all_active,
|
|
key=lambda x: (x["leads"] > 0, -x["cpl"] if x["leads"] > 0 else 0),
|
|
)[:10]
|
|
|
|
# ── Send consolidated Slack report ────────────────────────────────────────
|
|
duration = round(time.time() - start_ts, 1)
|
|
|
|
try:
|
|
slack_notifier.send_daily_report(
|
|
daily_totals=daily_totals,
|
|
best_campaigns=best_10,
|
|
worst_campaigns=worst_10,
|
|
actions=actions_proposed_list,
|
|
target_cpl=config.META_TARGET_CPL,
|
|
campaigns_analyzed=len(metrics_all),
|
|
mode="DRY_RUN" if config.DRY_RUN else "PRODUCTION",
|
|
verticals=verticals,
|
|
campaign_details=campaign_details,
|
|
monthly_verticals=monthly_verticals,
|
|
)
|
|
except Exception as e:
|
|
print(f" Warning: Slack notification failed: {e}")
|
|
|
|
# ── Execution log ─────────────────────────────────────────────────────────
|
|
summary = (
|
|
f"{len(metrics_all)} campaigns analyzed, "
|
|
f"{len(actions_proposed_list)} actions proposed, "
|
|
f"{actions_executed} executed."
|
|
)
|
|
try:
|
|
baserow.save_execution_log({
|
|
"mode": "DRY_RUN" if config.DRY_RUN else "PRODUCTION",
|
|
"campaigns_analyzed": len(metrics_all),
|
|
"actions_proposed": len(actions_proposed_list),
|
|
"actions_executed": actions_executed,
|
|
"errors": "\n".join(errors),
|
|
"summary": summary,
|
|
"duration_seconds": duration,
|
|
})
|
|
except Exception as e:
|
|
print(f" Warning: could not save execution log: {e}")
|
|
|
|
print(f"{'='*55}")
|
|
print(f" Done in {duration}s. {summary}")
|
|
if errors:
|
|
print(f" Errors ({len(errors)}): {'; '.join(errors[:3])}")
|
|
print(f"{'='*55}\n")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", line_buffering=True)
|
|
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
log_path = os.path.join("logs", f"{timestamp}.log")
|
|
tee = Tee(log_path)
|
|
sys.stdout = tee
|
|
try:
|
|
run()
|
|
finally:
|
|
tee.close()
|
|
sys.stdout = tee._stdout
|