""" Agente Strategy Cleanup de Leonex. Escanea `asset_strategies` (las promovidas que el bridge opera) y, para cada una, busca el tier ACTUAL en el último report del Lab correspondiente: - swing → dashboard/data/strategy_lab_report.json - scalping → dashboard/data/strategy_lab_scalping_report.json - custom → dashboard/data/custom_strategies_report.json (per_asset) Si una promovida ha dejado de ser GOLD/SILVER (porque el tier degradó a BRONZE/REJECTED en runs más recientes), el Lab está diciendo que su edge ya no es robusto. Este agente: 1. La reporta como `deprecated` con la razón del downgrade. 2. (Opcional con --execute) la elimina de `asset_strategies` y guarda copia en la nueva tabla `deprecated_promoted` para historial. 3. Genera `dashboard/data/deprecated_promoted_report.json` para visibilidad. Pensado para correr mensualmente. Idempotente. Uso: python agents/agente_strategy_cleanup.py # dry-run (default) python agents/agente_strategy_cleanup.py --execute # aplica borrados """ from __future__ import annotations import argparse import json import logging import sqlite3 import sys from dataclasses import asdict, dataclass, field from datetime import datetime try: from datetime import UTC except ImportError: from datetime import timezone UTC = timezone.utc from pathlib import Path from typing import Optional PROJECT_ROOT = Path(__file__).resolve().parents[1] DATA_DIR = PROJECT_ROOT / "data" LOGS_DIR = PROJECT_ROOT / "logs" DASHBOARD_DATA_DIR = PROJECT_ROOT / "dashboard" / "data" DB_PATH = DATA_DIR / "Leonex.sqlite" SWING_REPORT_PATH = DASHBOARD_DATA_DIR / "strategy_lab_report.json" SCALPING_REPORT_PATH = DASHBOARD_DATA_DIR / "strategy_lab_scalping_report.json" CUSTOM_REPORT_PATH = DASHBOARD_DATA_DIR / "custom_strategies_report.json" OUTPUT_REPORT_PATH = DASHBOARD_DATA_DIR / "deprecated_promoted_report.json" OPERABLE_TIERS = ("GOLD", "SILVER") DEPRECATED_TIERS = ("BRONZE", "REJECTED") @dataclass class CleanupRecord: ticker: str strategy: str timeframe: str promoted_at: str current_tier: str found_in: str # "swing" | "scalping" | "custom" | "not_found" decision: str # "keep" | "deprecate" | "missing_in_lab" | "removed" reason: str @dataclass class CleanupReport: generated_at: str mode: str # "dry_run" | "execute" total_promoted: int = 0 n_kept: int = 0 n_deprecated: int = 0 n_missing: int = 0 n_removed: int = 0 records: list[CleanupRecord] = field(default_factory=list) summary_note: str = "" def _build_logger() -> logging.Logger: LOGS_DIR.mkdir(parents=True, exist_ok=True) logger = logging.getLogger("agente_strategy_cleanup") logger.setLevel(logging.INFO) logger.handlers.clear() fmt = logging.Formatter("%(asctime)s | %(levelname)s | %(name)s | %(message)s") fh = logging.FileHandler(LOGS_DIR / "agente_strategy_cleanup.log", encoding="utf-8") fh.setFormatter(fmt) sh = logging.StreamHandler() sh.setFormatter(fmt) logger.addHandler(fh) logger.addHandler(sh) return logger def _safe_load_json(path: Path, logger: logging.Logger) -> dict: if not path.exists(): logger.warning("Report no existe: %s", path.name) return {} try: return json.loads(path.read_text(encoding="utf-8")) except Exception as exc: logger.warning("No pude leer %s: %s", path.name, exc) return {} def find_current_tier(ticker: str, strategy: str, swing_report: dict, scalping_report: dict, custom_report: dict) -> tuple[str, str]: """Devuelve (tier_actual, source_report) buscando el (ticker, strategy) en los 3 reports. Si no se encuentra, devuelve ("UNKNOWN", "not_found").""" # 1) Swing (assets[].results[]) for a in swing_report.get("assets", []): if a.get("ticker") != ticker: continue for r in a.get("results", []): if r.get("strategy") == strategy: return (r.get("tier", "UNKNOWN"), "swing") # 2) Scalping (misma estructura) for a in scalping_report.get("assets", []): if a.get("ticker") != ticker: continue for r in a.get("results", []): if r.get("strategy") == strategy: return (r.get("tier", "UNKNOWN"), "scalping") # 3) Custom (strategies[].per_asset[].ticker/tier) for s in custom_report.get("strategies", []): if s.get("name") != strategy: continue for pa in s.get("per_asset", []): if pa.get("ticker") == ticker: return (pa.get("tier", "UNKNOWN"), "custom") return ("UNKNOWN", "not_found") def ensure_deprecated_schema(db_path: Path = DB_PATH) -> None: """Crea tabla `deprecated_promoted` si no existe — historial de cleanups.""" with sqlite3.connect(db_path) as conn: conn.execute( """ CREATE TABLE IF NOT EXISTS deprecated_promoted ( id INTEGER PRIMARY KEY AUTOINCREMENT, ticker TEXT NOT NULL, strategy TEXT NOT NULL, timeframe TEXT, promoted_at TEXT, deprecated_at TEXT NOT NULL, last_tier TEXT, source_report TEXT, reason TEXT ) """ ) conn.commit() def scan_and_classify(db_path: Path, swing_report: dict, scalping_report: dict, custom_report: dict, logger: logging.Logger) -> list[CleanupRecord]: """Itera asset_strategies y clasifica cada promovida como keep / deprecate / missing_in_lab. NO toca BD — solo devuelve los records para inspección.""" records: list[CleanupRecord] = [] with sqlite3.connect(db_path) as conn: # Check si la tabla existe row = conn.execute( "SELECT name FROM sqlite_master WHERE type='table' " "AND name='asset_strategies'").fetchone() if not row: logger.warning("Tabla asset_strategies no existe — nada que limpiar.") return records promoted = conn.execute( "SELECT ticker, strategy, timeframe, promoted_at " "FROM asset_strategies ORDER BY ticker ASC" ).fetchall() for ticker, strategy, timeframe, promoted_at in promoted: tier_now, source = find_current_tier( ticker, strategy, swing_report, scalping_report, custom_report) if source == "not_found": decision = "missing_in_lab" reason = (f"No se encontró ({ticker}, {strategy}) en ningún " f"report. Quizá Lab no se corrió tras un cambio de " f"catálogo. Mantener por ahora y reintentar tras " f"el siguiente run del pipeline.") elif tier_now in OPERABLE_TIERS: decision = "keep" reason = (f"Tier {tier_now} en {source} — sigue operable.") elif tier_now in DEPRECATED_TIERS: decision = "deprecate" reason = (f"Tier degradado a {tier_now} en {source} — el Lab " f"dice que ya no tiene edge robusto. Candidata a " f"eliminar.") else: decision = "missing_in_lab" reason = f"Tier desconocido '{tier_now}' en {source}." records.append(CleanupRecord( ticker=ticker, strategy=strategy, timeframe=timeframe or "1d", promoted_at=promoted_at or "", current_tier=tier_now, found_in=source, decision=decision, reason=reason, )) return records def execute_removals(records: list[CleanupRecord], db_path: Path, logger: logging.Logger) -> int: """Para cada record con decision=deprecate, lo borra de asset_strategies y lo registra en deprecated_promoted. Devuelve cuántos se eliminaron.""" to_drop = [r for r in records if r.decision == "deprecate"] if not to_drop: return 0 now_iso = datetime.now(UTC).isoformat() ensure_deprecated_schema(db_path) n_removed = 0 with sqlite3.connect(db_path) as conn: for r in to_drop: try: conn.execute( """INSERT INTO deprecated_promoted (ticker, strategy, timeframe, promoted_at, deprecated_at, last_tier, source_report, reason) VALUES (?, ?, ?, ?, ?, ?, ?, ?)""", (r.ticker, r.strategy, r.timeframe, r.promoted_at, now_iso, r.current_tier, r.found_in, r.reason), ) conn.execute( "DELETE FROM asset_strategies WHERE ticker = ? " "AND strategy = ?", (r.ticker, r.strategy), ) r.decision = "removed" n_removed += 1 logger.info("Removed %s/%s (was %s in %s)", r.ticker, r.strategy, r.current_tier, r.found_in) except Exception as exc: logger.error("Failed to remove %s/%s: %s", r.ticker, r.strategy, exc) conn.commit() return n_removed def export_report(report: CleanupReport, path: Path, logger: logging.Logger) -> None: DASHBOARD_DATA_DIR.mkdir(parents=True, exist_ok=True) payload = asdict(report) payload["records"] = [asdict(r) for r in report.records] path.write_text( json.dumps(payload, indent=2, ensure_ascii=False, default=str), encoding="utf-8", ) logger.info("Report exportado → %s", path) def main() -> int: parser = argparse.ArgumentParser( description="Cleanup de promovidas obsoletas en Leonex") parser.add_argument("--execute", action="store_true", help="Aplica borrados. Sin esta bandera = dry-run.") parser.add_argument("--db", type=str, default=str(DB_PATH)) args = parser.parse_args() logger = _build_logger() db_path = Path(args.db) if not db_path.exists(): logger.error("DB no existe: %s", db_path) return 2 mode = "execute" if args.execute else "dry_run" logger.info("Iniciando cleanup en modo %s sobre %s", mode, db_path.name) swing_report = _safe_load_json(SWING_REPORT_PATH, logger) scalping_report = _safe_load_json(SCALPING_REPORT_PATH, logger) custom_report = _safe_load_json(CUSTOM_REPORT_PATH, logger) records = scan_and_classify(db_path, swing_report, scalping_report, custom_report, logger) n_total = len(records) n_keep = sum(1 for r in records if r.decision == "keep") n_dep = sum(1 for r in records if r.decision == "deprecate") n_missing = sum(1 for r in records if r.decision == "missing_in_lab") n_removed = 0 if args.execute and n_dep: n_removed = execute_removals(records, db_path, logger) note_parts = [ f"{n_total} promovidas escaneadas", f"{n_keep} keep", f"{n_dep} deprecate", f"{n_missing} missing_in_lab", ] if args.execute: note_parts.append(f"{n_removed} eliminadas") note = " · ".join(note_parts) + "." report = CleanupReport( generated_at=datetime.now(UTC).isoformat(), mode=mode, total_promoted=n_total, n_kept=n_keep, n_deprecated=n_dep, n_missing=n_missing, n_removed=n_removed, records=records, summary_note=note, ) export_report(report, OUTPUT_REPORT_PATH, logger) print() print("=" * 72) print(f"Strategy Cleanup — mode={mode}") print("=" * 72) print(note) print() if n_dep: print("Candidatas a deprecar (tier degradado a BRONZE/REJECTED):") for r in records: if r.decision in ("deprecate", "removed"): tag = "[REMOVED]" if r.decision == "removed" else "[DEPRECATE]" print(f" {tag} {r.ticker:<6} {r.strategy:<60} " f"({r.current_tier} en {r.found_in})") if n_missing: print() print("Sin tier actual (Lab no las encontró):") for r in records: if r.decision == "missing_in_lab": print(f" [MISSING] {r.ticker:<6} {r.strategy:<60}") if not args.execute and n_dep: print() print(f"DRY-RUN: ninguna eliminada. Re-ejecuta con --execute para borrar.") return 0 if __name__ == "__main__": sys.exit(main())