""" Diagnostico del universo de Leonex. Imprime informacion clara sobre: - Si la tabla active_universe existe y cuantos tickers tiene. - Si la tabla esta vacia o el agente datos esta usando el fallback legacy. - Que tickers hay en market_snapshot.json (lo que el dashboard pinta hoy). - Cuando fue la ultima ejecucion del Universe / Datos. - Comando exacto a ejecutar para arreglarlo si esta mal. Uso: python tools/diagnose_universe.py """ from __future__ import annotations import json import sqlite3 import sys from datetime import datetime from pathlib import Path ROOT = Path(__file__).resolve().parents[1] DB_PATH = ROOT / "data" / "Leonex.sqlite" SNAPSHOT_PATH = ROOT / "dashboard" / "data" / "market_snapshot.json" UNIVERSE_REPORT = ROOT / "dashboard" / "data" / "universe_report.json" LEGACY_FIXED = { "SPY", "QQQ", "IWM", "AAPL", "MSFT", "NVDA", "AMZN", "META", "GOOGL", "EURUSD=X", "GBPUSD=X", "USDJPY=X", "BTC-USD", "ETH-USD", "SOL-USD", } def banner(title: str) -> None: print() print("=" * 60) print(f" {title}") print("=" * 60) def main() -> int: banner("LEONEX - DIAGNOSTICO UNIVERSO") print(f" DB esperada : {DB_PATH}") print(f" DB existe? : {DB_PATH.exists()}") if not DB_PATH.exists(): print() print(" [X] No hay base de datos. Ejecuta primero:") print(" python .\\agents\\agente_datos.py") return 1 with sqlite3.connect(DB_PATH) as conn: # 1. Tabla active_universe banner("1) Tabla active_universe") try: rows = conn.execute( "SELECT ticker, score, rank, reason, asset_class, selected_at " "FROM active_universe ORDER BY rank ASC NULLS LAST, ticker" ).fetchall() except sqlite3.OperationalError as exc: print(f" [X] No existe la tabla: {exc}") print() print(" La tabla se crea automaticamente al ejecutar agente_universe.") print(" Causa probable: aun no has corrido el agente.") rows = None if rows is not None: print(f" Filas: {len(rows)}") if not rows: print(" [!] Tabla VACIA. agente_datos.py esta usando el universo legacy de 15 tickers.") print() print(" SOLUCION:") print(" python .\\agents\\agente_universe.py --force") else: ts = rows[0][5] if rows else "—" print(f" Ultima seleccion: {ts}") print() print(" Primeros 10 tickers (rank, score, ticker, reason):") for r in rows[:10]: score = (r[1] * 100) if r[1] else 0.0 print(f" #{r[2]:>3} {score:>+7.2f}% {r[0]:<10} {r[3]:<14}") if len(rows) > 10: print(f" ... y {len(rows) - 10} mas") legacy_count = sum(1 for r in rows if r[0] in LEGACY_FIXED) top_count = sum(1 for r in rows if (r[3] or "") == "top_momentum") print() print(f" Activos por categoria:") print(f" top_momentum : {top_count}") print(f" open_position : {sum(1 for r in rows if r[3] == 'open_position')}") print(f" legacy : {sum(1 for r in rows if r[3] == 'legacy')}") # 2. Tabla prices banner("2) Tabla prices (lo que el agente_datos ha descargado)") try: tickers_in_prices = conn.execute( "SELECT DISTINCT ticker, MAX(date) FROM prices GROUP BY ticker ORDER BY ticker" ).fetchall() print(f" Tickers distintos en prices: {len(tickers_in_prices)}") if tickers_in_prices: last_dates = [t[1] for t in tickers_in_prices if t[1]] if last_dates: print(f" Fecha mas reciente registrada: {max(last_dates)}") if len(tickers_in_prices) <= 16: # Probablemente seguimos en universo legacy print(" [!] Tienes <=16 tickers en prices: el universo dinamico aun NO se aplica.") else: print(" [OK] Tienes mas de 16 tickers: universo dinamico operativo.") print() print(" Lista completa:") for tk, last in tickers_in_prices[:50]: legacy = "L" if tk in LEGACY_FIXED else " " print(f" [{legacy}] {tk:<12} last={last}") if len(tickers_in_prices) > 50: print(f" ... y {len(tickers_in_prices) - 50} mas") except sqlite3.OperationalError as exc: print(f" [X] No existe la tabla prices: {exc}") # 3. market_snapshot.json (lo que Vision pinta) banner("3) dashboard/data/market_snapshot.json (lo que VES en Vision)") if not SNAPSHOT_PATH.exists(): print(" [!] No existe market_snapshot.json. Ejecuta agente_datos.py") else: snap = json.loads(SNAPSHOT_PATH.read_text(encoding="utf-8")) assets = snap.get("assets", []) print(f" Generado : {snap.get('generated_at', '—')}") print(f" Fuente : {snap.get('source', '—')}") print(f" Activos : {len(assets)}") if assets: print() print(" Lista de tickers (lo que Vision pinta):") for a in assets: print(f" {a.get('ticker', '?'):<10}") if len(assets) <= 16: print() print(" [DIAGNOSTICO] Tienes <=16 tickers en el snapshot.") print(" Esto significa que cuando agente_datos.py corrio, la tabla") print(" active_universe estaba VACIA y uso el universo legacy.") # 4. universe_report.json banner("4) dashboard/data/universe_report.json (output del Universe)") if not UNIVERSE_REPORT.exists(): print(" [!] No existe universe_report.json — el agente_universe NUNCA se ha ejecutado.") print() print(" SOLUCION:") print(" python .\\agents\\agente_universe.py --force") else: u = json.loads(UNIVERSE_REPORT.read_text(encoding="utf-8")) print(f" Generado : {u.get('generated_at', '—')}") print(f" Rebalanceado: {u.get('rebalanced', '?')}") print(f" Tamano : {u.get('universe_size', 0)}") print(f" S&P 500 : {u.get('sp500_size', 0)} tickers escaneados") print(f" Razon : {u.get('rebalance_reason', '—')}") # 5. Recomendacion final banner("5) PROXIMOS PASOS") print(""" Si las tablas estan vacias o ves los mismos 15 activos: 1. Forzar el universo dinamico: python .\\agents\\agente_universe.py --force Tardara 2-4 minutos en la PRIMERA ejecucion (descarga 500 tickers). 2. Re-ejecutar el pipeline completo: .\\actualizar_datos.bat 3. Refrescar el dashboard (F5 en el navegador). """) return 0 if __name__ == "__main__": if sys.stdout.encoding and sys.stdout.encoding.lower() != "utf-8": import io sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace") raise SystemExit(main())