#!/usr/bin/env python3 """ Daily observability report de Leonex. Cada dia (idealmente a las 22:00 UTC, post-cierre US) genera un resumen del estado del sistema y lo escribe a: logs/daily_summary_YYYY-MM-DD.md Lee los reports JSON ya generados por el pipeline (paper_report, drift_report, bridge_promoted_report, meta_report, cpcv_report) — no llama a Alpaca ni recalcula nada. Pensado para ser invocado por n8n cron diario o manual. USO: python tools/daily_observability.py python tools/daily_observability.py --root /ruta/al/proyecto python tools/daily_observability.py --print # stdout en vez de archivo EL RESUMEN INCLUYE: - Equity actual y delta vs ayer - Fires del dia (cuantas ordenes, de que carril, en que tickers) - Estado del Drift Monitor (overall + sticky pending si aplica) - PBO + Meta AUC primary/comparison + delta - Lock status - Carrier Health verdict - Top-3 promovidas mas cerca de fire (highest prob_win sin live) """ from __future__ import annotations import argparse import json import sys from datetime import datetime, timedelta, timezone from pathlib import Path def _load_json(path: Path) -> dict: try: return json.loads(path.read_text(encoding="utf-8")) except Exception: return {} def _fmt_money(v) -> str: try: return f"${float(v):,.2f}" except Exception: return str(v) def _fmt_pct(v, places: int = 2) -> str: try: return f"{float(v) * 100:.{places}f}%" except Exception: return "—" def build_report(root: Path) -> str: data_dir = root / "dashboard" / "data" logs_dir = root / "logs" now = datetime.now(timezone.utc) today_str = now.strftime("%Y-%m-%d") paper = _load_json(data_dir / "paper_report.json") drift = _load_json(data_dir / "drift_report.json") bridge = _load_json(data_dir / "bridge_promoted_report.json") meta = _load_json(data_dir / "meta_report.json") cpcv = _load_json(data_dir / "cpcv_report.json") lines: list[str] = [] lines.append(f"# Leonex — Daily Observability · {today_str}") lines.append("") lines.append(f"*Generado: {now.strftime('%Y-%m-%d %H:%M UTC')}*") lines.append("") # ─── Equity ──────────────────────────────────────────────────────────── lines.append("## Equity") equity = paper.get("equity") if equity is not None: lines.append(f"- **Equity actual**: {_fmt_money(equity)}") # Buscar equity de ayer en historial yesterday_str = (now - timedelta(days=1)).strftime("%Y-%m-%d") yest_path = logs_dir / f"daily_summary_{yesterday_str}.md" if yest_path.exists(): try: yest_text = yest_path.read_text(encoding="utf-8") for line in yest_text.splitlines(): if line.startswith("- **Equity actual**:"): prev = line.split("$", 1)[-1].replace(",", "").strip() delta = float(equity) - float(prev) sign = "+" if delta >= 0 else "" lines.append( f"- **Delta vs ayer**: {sign}{_fmt_money(delta)}") break except Exception: pass else: lines.append("- Equity no disponible (paper_report.json vacio).") lines.append("") # ─── Fires de hoy ────────────────────────────────────────────────────── lines.append("## Fires del dia") actions = paper.get("actions", []) or [] today_actions = [a for a in actions if str(a.get("timestamp", "")).startswith(today_str)] if not today_actions: lines.append("- Sin acciones registradas hoy.") else: for a in today_actions[:20]: ts = a.get("timestamp", "")[11:16] ticker = a.get("ticker", "?") side = a.get("side", "?").upper() status = a.get("status", "?") note = (a.get("notes") or "")[:90] lines.append(f"- `{ts}` **{ticker:<6}** {side:<5} → {status:<12} {note}") lines.append("") # ─── Drift Monitor ──────────────────────────────────────────────────── lines.append("## Drift Monitor (carril legacy)") overall = drift.get("overall_status") or drift.get("overall") or "—" paused = drift.get("pause_new_trades") note = (drift.get("summary_note") or "")[:200] lines.append(f"- **Estado**: {overall}") lines.append(f"- **PAUSE_NEW_TRADES**: {paused}") if note: lines.append(f"- **Nota**: {note}") lines.append("") # ─── Carrier Health (Meta + CPCV + Drift) ───────────────────────────── lines.append("## Carrier Health") pbo = cpcv.get("pbo") or cpcv.get("pbo_score") auc_p = meta.get("auc_oof_primary") or meta.get("auc_oof") auc_c = meta.get("auc_oof_comparison") delta = meta.get("delta_auc") cmp_mode = meta.get("comparison_mode") if pbo is not None: verdict = "RED" if pbo >= 0.50 else ("AMBER" if pbo >= 0.30 else "GREEN") lines.append(f"- **PBO**: {pbo:.2%} ({verdict})") if auc_p is not None: v = "RED" if auc_p < 0.55 else ("AMBER" if auc_p < 0.60 else "GREEN") lines.append(f"- **Meta AUC primary**: {auc_p:.3f} ({v})") if auc_c is not None: lines.append(f"- **Meta AUC comparison ({cmp_mode})**: {auc_c:.3f}") if delta is not None: sign = "+" if delta >= 0 else "" signal = ("primary gana" if delta > 0.01 else ("comparison gana" if delta < -0.01 else "empate practico")) lines.append(f"- **Delta AUC**: {sign}{delta:.3f} ({signal})") lines.append("") # ─── Bridge promoted ────────────────────────────────────────────────── lines.append("## Bridge promoted") evaluated = bridge.get("evaluated", []) or [] live = [e for e in evaluated if e.get("live")] n_promoted = len(evaluated) lines.append(f"- **Promovidas evaluadas**: {n_promoted}") lines.append(f"- **Live hoy**: {len(live)}") if live: for e in live[:8]: t = e.get("ticker", "?") s = (e.get("strategy") or "")[:42] pw = e.get("prob_win", 0) lines.append(f" - **{t:<6}** {s:<42} prob_win={pw:.2%}") # Top-3 mas cerca de fire (no live pero con prob_win alta) not_live = [e for e in evaluated if not e.get("live") and e.get("prob_win", 0) > 0] not_live.sort(key=lambda e: -e.get("prob_win", 0)) if not_live[:3]: lines.append("- **Top-3 mas cerca de fire** (alto prob_win, trigger pendiente):") for e in not_live[:3]: t = e.get("ticker", "?") s = (e.get("strategy") or "")[:42] pw = e.get("prob_win", 0) note_e = (e.get("note") or "")[:60] lines.append(f" - **{t:<6}** {s:<42} prob_win={pw:.2%} ({note_e})") lines.append("") # ─── Open positions ──────────────────────────────────────────────────── lines.append("## Open positions ahora mismo") open_pos = paper.get("open_positions", []) or [] if not open_pos: lines.append("- Sin posiciones abiertas.") else: for pos in open_pos[:20]: t = pos.get("ticker", "?") qty = pos.get("qty", 0) avg = pos.get("avg_price", 0) cur = pos.get("current_price", 0) pnl = pos.get("unrealized_pl", 0) lines.append(f"- **{t:<6}** qty={qty} avg=${avg:.2f} cur=${cur:.2f} " f"P/L={_fmt_money(pnl)}") lines.append("") # ─── Summary line ───────────────────────────────────────────────────── lines.append("## TL;DR") if today_actions: n_fires_today = sum(1 for a in today_actions if "sent" in str(a.get("status", "")).lower() or "filled" in str(a.get("status", "")).lower()) lines.append(f"Hoy hubo **{n_fires_today} fire(s)** vivos. " f"{len(open_pos)} posiciones abiertas. " f"Drift {overall}.") else: lines.append(f"Hoy **sin fires nuevos**. {len(open_pos)} posiciones " f"abiertas. Drift {overall}.") return "\n".join(lines) def main() -> int: parser = argparse.ArgumentParser(description="Leonex daily observability") parser.add_argument("--root", type=str, default=str(Path(__file__).resolve().parents[1]), help="Raíz del proyecto Leonex") parser.add_argument("--print", dest="print_only", action="store_true", help="Imprime a stdout en vez de escribir a logs/") args = parser.parse_args() root = Path(args.root) if not root.exists(): print(f"[ERROR] root no existe: {root}", file=sys.stderr) return 2 report = build_report(root) if args.print_only: print(report) return 0 logs_dir = root / "logs" logs_dir.mkdir(parents=True, exist_ok=True) today = datetime.now(timezone.utc).strftime("%Y-%m-%d") out_path = logs_dir / f"daily_summary_{today}.md" out_path.write_text(report, encoding="utf-8") print(f"[OK] Daily summary escrito a {out_path}") return 0 if __name__ == "__main__": sys.exit(main())