"""Reglas canónicas de decisión para mantener consistencia entre UI, LLM, gráficos y PDF. Este módulo no reemplaza los modelos. Su función es integrar sus salidas para que el dashboard y el reporte ejecutivo no se contradigan. """ from __future__ import annotations from typing import Any, Dict def _sf(value: Any, default: float = 0.0) -> float: try: if value in (None, ""): return default return float(value) except Exception: return default def _norm_text(value: Any) -> str: return str(value or "").strip().lower() def probability(result: Dict[str, Any] | None = None, final_rec: Dict[str, Any] | None = None, paid_xgb: Dict[str, Any] | None = None) -> float: result = result or {} final_rec = final_rec or {} paid_xgb = paid_xgb or {} return max(0.0, min(1.0, _sf( result.get("probabilidad_rendimiento", final_rec.get("probabilidad_rendimiento", paid_xgb.get("logistic_probability", 0.0))) ))) def visual_score_0_100(visual: Dict[str, Any] | None = None, result: Dict[str, Any] | None = None) -> float | None: visual = visual or (result or {}).get("analisis_visual", {}) or {} raw = visual.get("composition_score", visual.get("visual_score", visual.get("score", None))) if raw in (None, ""): return None score = _sf(raw, 0.0) return score * 100.0 if score <= 1.5 else score def paid_score_0_100(paid_xgb: Dict[str, Any] | None = None, result: Dict[str, Any] | None = None) -> float | None: paid_xgb = paid_xgb or (result or {}).get("xgboost_pauta", {}) or ((result or {}).get("metricas", {}) or {}).get("xgboost_pauta", {}) or {} raw = paid_xgb.get("predicted_paid_performance_score", paid_xgb.get("rules_paid_performance_score", paid_xgb.get("raw_xgboost_paid_performance_score", None))) if raw in (None, ""): return None score = _sf(raw, 0.0) return score * 100.0 if score <= 1.5 else score def visual_interpretation(score: float | None) -> Dict[str, str]: if score is None: return { "label": "sin video", "summary": "No hay score visual suficiente; el análisis se apoya en métricas, texto y políticas.", "severity": "neutral", } if score <= 59: return { "label": "crítico", "summary": "Score visual por debajo de 60: no se recomienda invertir; requiere atención inmediata en composición, legibilidad o montaje.", "severity": "danger", } if score <= 70: return { "label": "regular", "summary": "Score visual entre 60 y 70: es viable como prueba controlada, pero requiere ajustes de posproducción y revisión humana.", "severity": "warning", } if score <= 80: return { "label": "bueno", "summary": "Score visual entre 71 y 80: composición buena; puede mejorar retención y comprensión del mensaje.", "severity": "success", } return { "label": "muy bien compuesto", "summary": "Score visual entre 81 y 100: pieza muy bien compuesta y consistente para pauta controlada.", "severity": "success", } def classify_investment_decision( *, result: Dict[str, Any] | None = None, final_rec: Dict[str, Any] | None = None, paid_xgb: Dict[str, Any] | None = None, policy: Dict[str, Any] | None = None, visual: Dict[str, Any] | None = None, ) -> Dict[str, Any]: """Devuelve una decisión única para UI/PDF/LLM. Estados canónicos: - invest: invertir/pautar - adjust: ajustar antes de invertir - no: no invertir - review: revisión humana inmediata """ result = result or {} final_rec = final_rec or result paid_xgb = paid_xgb or result.get("xgboost_pauta") or ((result.get("metricas", {}) or {}).get("xgboost_pauta")) or {} policy = policy or result.get("analisis_politicas") or {} visual = visual or result.get("analisis_visual") or {} prob = probability(result, final_rec, paid_xgb) vscore = visual_score_0_100(visual, result) pscore = paid_score_0_100(paid_xgb, result) policy_level = _norm_text(result.get("policy_risk_level") or final_rec.get("policy_risk_level") or policy.get("policy_risk_level") or "bajo") policy_cap = policy.get("probability_cap") sensitive = bool(policy.get("policy_forced_probability")) or policy_level in {"alto", "revisión humana", "revision humana", "high"} gate_passed = bool(paid_xgb.get("gate_passed", paid_xgb.get("eligible_for_paid_xgboost", False))) eligible = bool(paid_xgb.get("eligible_for_paid_xgboost", False)) cpm = _sf(paid_xgb.get("predicted_cpm") or result.get("cpm_estimado") or 0.0) # 1) Riesgos sensibles: políticas duras o probabilidad capada a <=20%. if policy_level in {"revisión humana", "revision humana"} or (sensitive and prob <= 0.20) or (_sf(policy_cap, 1.0) <= 0.20): return { "key": "review", "label": "REVISIÓN HUMANA INMEDIATA", "headline": "RECOMENDACIÓN: REVISIÓN HUMANA INMEDIATA", "ui_label": "🟣 Revisión humana inmediata: contenido sensible o riesgo de políticas.", "pdf_label": "RECOMENDACIÓN: REVISIÓN HUMANA", "action_final": "REVISIÓN HUMANA", "color": "violet", "reason": "El contenido activa reglas sensibles de políticas o la probabilidad fue limitada a 20% o menos.", "probability": prob, "visual_score": vscore, "paid_score": pscore, "cpm": cpm, "gate_passed": gate_passed, "visual_interpretation": visual_interpretation(vscore), } # 2) No invertir por debilidad dura de modelo, visual o score de pauta. if prob < 0.50 or (vscore is not None and vscore < 50) or (pscore is not None and pscore < 50): return { "key": "no", "label": "NO INVIERTAS", "headline": "RECOMENDACIÓN: NO INVERTIR", "ui_label": "🔴 No inviertas: las señales no justifican pauta.", "pdf_label": "RECOMENDACIÓN: NO PAUTAR", "action_final": "NO IMPULSAR", "color": "red", "reason": "La probabilidad publicitaria, el score de pauta o el score visual están por debajo del umbral mínimo.", "probability": prob, "visual_score": vscore, "paid_score": pscore, "cpm": cpm, "gate_passed": gate_passed, "visual_interpretation": visual_interpretation(vscore), } # 3) Ajustar cuando pasa el umbral, pero hay regularidad visual/CPM/score de pauta. regular_visual = vscore is not None and 60 <= vscore <= 70 regular_paid = pscore is not None and 50 <= pscore < 65 regular_cpm = cpm >= 7.0 if (not eligible) or regular_visual or regular_paid or regular_cpm: return { "key": "adjust", "label": "REALIZA AJUSTES ANTES DE INVERTIR", "headline": "RECOMENDACIÓN: AJUSTAR ANTES DE INVERTIR", "ui_label": "🟡 Realiza ajustes antes de invertir: hay potencial, pero no conviene escalar todavía.", "pdf_label": "RECOMENDACIÓN: AJUSTAR ANTES DE PAUTAR", "action_final": "AJUSTAR ANTES DE IMPULSAR", "color": "yellow", "reason": "El video muestra señales parciales, pero requiere optimización creativa, visual o de eficiencia antes de escalar presupuesto.", "probability": prob, "visual_score": vscore, "paid_score": pscore, "cpm": cpm, "gate_passed": gate_passed, "visual_interpretation": visual_interpretation(vscore), } # 4) Invertir solo si el gate pasa y no hay bloqueos. if prob >= 0.51 and eligible and (vscore is None or vscore >= 71) and (pscore is None or pscore >= 65): return { "key": "invest", "label": "INVIERTE", "headline": "RECOMENDACIÓN: INVERTIR", "ui_label": "🟢 Invierte: el video cumple el umbral de aptitud publicitaria.", "pdf_label": "RECOMENDACIÓN: PAUTAR", "action_final": "IMPULSAR", "color": "green", "reason": "El video supera el umbral de regresión logística y no presenta bloqueos visuales o de política.", "probability": prob, "visual_score": vscore, "paid_score": pscore, "cpm": cpm, "gate_passed": gate_passed, "visual_interpretation": visual_interpretation(vscore), } return { "key": "adjust", "label": "REALIZA AJUSTES ANTES DE INVERTIR", "headline": "RECOMENDACIÓN: AJUSTAR ANTES DE INVERTIR", "ui_label": "🟡 Realiza ajustes antes de invertir: señales mixtas.", "pdf_label": "RECOMENDACIÓN: AJUSTAR ANTES DE PAUTAR", "action_final": "AJUSTAR ANTES DE IMPULSAR", "color": "yellow", "reason": "Las señales son mixtas; conviene optimizar y volver a evaluar.", "probability": prob, "visual_score": vscore, "paid_score": pscore, "cpm": cpm, "gate_passed": gate_passed, "visual_interpretation": visual_interpretation(vscore), } def integrated_markdown(decision: Dict[str, Any]) -> str: vscore = decision.get("visual_score") pscore = decision.get("paid_score") prob = _sf(decision.get("probability"), 0.0) * 100 cpm = _sf(decision.get("cpm"), 0.0) visual_text = decision.get("visual_interpretation", {}).get("summary", "") return ( "### Visión integradora de decisión\n\n" f"**{decision.get('ui_label', decision.get('label'))}**\n\n" f"- Probabilidad de pauta del modelo principal: **{prob:.1f}%**.\n" f"- Score de pauta XGBoost/calibrado: **{pscore:.1f}/100**.\n" if pscore is not None else "### Visión integradora de decisión\n\n" f"**{decision.get('ui_label', decision.get('label'))}**\n\n" f"- Probabilidad de pauta del modelo principal: **{prob:.1f}%**.\n" ) + ( f"- Score visual/composición: **{vscore:.0f}/100**. {visual_text}\n" if vscore is not None else "- Score visual/composición: **sin video o sin frames suficientes**.\n" ) + ( f"- CPM estimado: **${cpm:.2f}**.\n" f"- Razón integrada: {decision.get('reason', 'Se integraron modelos, políticas y señales multimodales.')}\n" )