""" Agente de Datos Intraday de Leonex. Descarga barras intradia de yfinance para equities y forex (excluye crypto). Soporta dos bloques: SWING : 1h (nativo) + 4h (agregado de 1h). Tope yfinance 1h = 730 dias. SCALPING: 30m / 15m / 5m (nativos). Tope yfinance = 60 dias. LIMITACION CLAVE del scalping: yfinance solo conserva 60 dias de historico a 30m/15m/5m. Por eso este agente es INCREMENTAL: cada ejecucion descarga la ventana reciente y la UPSERTA, de modo que el historico crece con los meses (lo que yfinance no guarda, Leonex lo va acumulando en prices_intraday). Cada dia que no se ejecute = datos de scalping perdidos para siempre. DECISIONES de diseno: - 4h NO existe nativo en yfinance: se agrega de 4 velas de 1h. - 30m/15m/5m SI son nativos: se descargan directamente. - Timestamps normalizados a UTC. - Crypto excluido (termina en -USD o /USD). - Tabla prices_intraday(ticker, timeframe, ts, ohlcv). Uso: python agents/agente_datos_intraday.py # 1h + 4h python agents/agente_datos_intraday.py --timeframes 30m,15m,5m # scalping python agents/agente_datos_intraday.py --timeframes 1h,4h,30m,15m,5m python agents/agente_datos_intraday.py --limit-tickers 10 # smoke test """ from __future__ import annotations import argparse import logging import sqlite3 import sys import time from dataclasses import dataclass, field from datetime import datetime try: from datetime import UTC except ImportError: from datetime import timezone UTC = timezone.utc from pathlib import Path import numpy as np import pandas as pd PROJECT_ROOT = Path(__file__).resolve().parents[1] DATA_DIR = PROJECT_ROOT / "data" LOGS_DIR = PROJECT_ROOT / "logs" DB_PATH = DATA_DIR / "Leonex.sqlite" # Configuracion DEFAULT_PERIOD_1H = "730d" # tope yfinance para 1h (primera descarga) DEFAULT_INCREMENTAL_PERIOD = "10d" # overlap diario al ya tener historico DEFAULT_TIMEFRAMES = ["1h", "4h"] SCALPING_TIMEFRAMES = ("30m", "15m", "5m") # nativos en yfinance, max 60 dias SCALPING_MAX_PERIOD = "60d" # tope yfinance para 30m/15m/5m VALID_TIMEFRAMES = ("1h", "4h") + SCALPING_TIMEFRAMES MAX_RETRIES_PER_TICKER = 2 SLEEP_BETWEEN_TICKERS_S = 0.4 # rate limit defensivo def is_crypto(ticker: str) -> bool: s = ticker.upper() return s.endswith("-USD") or s.endswith("/USD") # ───────────────────────────────────────────────────────────────────────────── # Schema # ───────────────────────────────────────────────────────────────────────────── def ensure_schema(db_path: Path = DB_PATH) -> None: with sqlite3.connect(db_path) as conn: conn.execute( """ CREATE TABLE IF NOT EXISTS prices_intraday ( ticker TEXT NOT NULL, timeframe TEXT NOT NULL, ts TEXT NOT NULL, open REAL, high REAL, low REAL, close REAL, volume REAL, PRIMARY KEY (ticker, timeframe, ts) ) """ ) conn.execute( "CREATE INDEX IF NOT EXISTS idx_prices_intraday_ticker_tf " "ON prices_intraday(ticker, timeframe)" ) conn.commit() # ───────────────────────────────────────────────────────────────────────────── # Universe # ───────────────────────────────────────────────────────────────────────────── def load_universe(db_path: Path = DB_PATH, exclude_crypto: bool = True) -> list[str]: """Data pool: todos los tickers con datos diarios en la tabla prices (poblada por agente_datos.py con el S&P 500 completo). Asi el intradia sigue automaticamente al mismo conjunto que el diario. Excluye crypto.""" if not db_path.exists(): return [] with sqlite3.connect(db_path) as conn: try: df = pd.read_sql("SELECT DISTINCT ticker FROM prices", conn) tickers = df["ticker"].tolist() except Exception: tickers = [] if not tickers: try: df = pd.read_sql("SELECT DISTINCT ticker FROM active_universe", conn) tickers = df["ticker"].tolist() if not df.empty else [] except Exception: tickers = [] if exclude_crypto: tickers = [t for t in tickers if not is_crypto(t)] return sorted(tickers) # ───────────────────────────────────────────────────────────────────────────── # Descarga intradia # ───────────────────────────────────────────────────────────────────────────── def _get_last_intraday_ts(ticker: str, timeframe: str, db_path: Path = DB_PATH) -> str | None: """Devuelve el ultimo timestamp registrado para (ticker, timeframe), o None.""" try: with sqlite3.connect(db_path) as conn: row = conn.execute( "SELECT MAX(ts) FROM prices_intraday " "WHERE ticker = ? AND timeframe = ?", (ticker, timeframe), ).fetchone() return row[0] if row and row[0] else None except sqlite3.OperationalError: return None def download_intraday(ticker: str, interval: str, period: str, logger: logging.Logger) -> pd.DataFrame: """Descarga barras intradia via yfinance para el intervalo dado (1h / 30m / 15m / 5m). Devuelve DataFrame open/high/low/close/volume indexado por ts (UTC). DataFrame vacio si falla.""" try: import yfinance as yf except ImportError: logger.error("yfinance no instalado. pip install yfinance") return pd.DataFrame() for attempt in range(MAX_RETRIES_PER_TICKER + 1): try: t = yf.Ticker(ticker) df = t.history(period=period, interval=interval, auto_adjust=True, prepost=False) if df is None or df.empty: if attempt < MAX_RETRIES_PER_TICKER: time.sleep(0.8) continue return pd.DataFrame() df = df.rename(columns=str.lower) keep = ["open", "high", "low", "close", "volume"] df = df[[c for c in keep if c in df.columns]] if df.index.tz is None: df.index = df.index.tz_localize("UTC") else: df.index = df.index.tz_convert("UTC") df.index.name = "ts" return df except Exception as exc: if attempt < MAX_RETRIES_PER_TICKER: logger.warning("Retry %s %s (attempt %d): %s", ticker, interval, attempt + 1, exc) time.sleep(0.8) continue logger.warning("Fallo definitivo %s %s: %s", ticker, interval, exc) return pd.DataFrame() return pd.DataFrame() # ───────────────────────────────────────────────────────────────────────────── # Agregacion 1h → 4h # ───────────────────────────────────────────────────────────────────────────── def aggregate_to_4h(df_1h: pd.DataFrame) -> pd.DataFrame: """Agrega barras 1h a 4h. Resample alineado a 00:00 UTC.""" if df_1h.empty: return df_1h agg = df_1h.resample("4h", label="left", closed="left").agg({ "open": "first", "high": "max", "low": "min", "close": "last", "volume": "sum", }) return agg.dropna(subset=["close"]) # ───────────────────────────────────────────────────────────────────────────── # Persistencia # ───────────────────────────────────────────────────────────────────────────── def upsert_bars(ticker: str, timeframe: str, df: pd.DataFrame, db_path: Path = DB_PATH) -> int: """Upserts barras a prices_intraday. Devuelve n filas insertadas/actualizadas.""" if df.empty: return 0 rows = [] for ts, r in df.iterrows(): rows.append(( ticker, timeframe, ts.isoformat(), float(r.get("open", np.nan)) if pd.notna(r.get("open", np.nan)) else None, float(r.get("high", np.nan)) if pd.notna(r.get("high", np.nan)) else None, float(r.get("low", np.nan)) if pd.notna(r.get("low", np.nan)) else None, float(r.get("close", np.nan)) if pd.notna(r.get("close", np.nan)) else None, float(r.get("volume", np.nan)) if pd.notna(r.get("volume", np.nan)) else None, )) with sqlite3.connect(db_path) as conn: conn.executemany( """ INSERT INTO prices_intraday (ticker, timeframe, ts, open, high, low, close, volume) VALUES (?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(ticker, timeframe, ts) DO UPDATE SET open=excluded.open, high=excluded.high, low=excluded.low, close=excluded.close, volume=excluded.volume """, rows, ) conn.commit() return len(rows) # ───────────────────────────────────────────────────────────────────────────── # Main # ───────────────────────────────────────────────────────────────────────────── def main() -> int: parser = argparse.ArgumentParser(description="Agente Datos Intraday de Leonex") parser.add_argument("--timeframes", type=str, default=",".join(DEFAULT_TIMEFRAMES), help="Timeframes a descargar. Soportados: 1h,4h,30m,15m,5m. " "Default '1h,4h'. Scalping: '30m,15m,5m'.") parser.add_argument("--period", type=str, default=DEFAULT_PERIOD_1H, help="Periodo yfinance para 1h. Default '730d' (max). " "Solo se usa en la PRIMERA descarga por ticker.") parser.add_argument("--incremental-period", type=str, default=DEFAULT_INCREMENTAL_PERIOD, help="Periodo a descargar cuando YA hay historico previo. " "Default '10d' (overlap de seguridad).") parser.add_argument("--full", action="store_true", help="Forzar descarga del rango completo aunque haya historico.") parser.add_argument("--limit-tickers", type=int, default=0, help="Si >0 limita tickers procesados (smoke test).") parser.add_argument("--include-crypto", action="store_true", help="Incluye crypto. Default: excluido.") args = parser.parse_args() LOGS_DIR.mkdir(parents=True, exist_ok=True) DATA_DIR.mkdir(parents=True, exist_ok=True) ensure_schema(DB_PATH) logging.basicConfig( level=logging.INFO, format="%(asctime)s | %(levelname)s | %(name)s | %(message)s", handlers=[ logging.FileHandler(LOGS_DIR / "agente_datos_intraday.log", encoding="utf-8"), logging.StreamHandler(), ], ) log = logging.getLogger("agente_datos_intraday") timeframes = [t.strip() for t in args.timeframes.split(",") if t.strip()] timeframes_to_persist = [t for t in timeframes if t in VALID_TIMEFRAMES] if not timeframes_to_persist: log.error("Ningun timeframe valido. Soportados: %s. Recibido: %s", VALID_TIMEFRAMES, timeframes) return 1 need_1h = any(t in ("1h", "4h") for t in timeframes_to_persist) scalping_tfs = [t for t in SCALPING_TIMEFRAMES if t in timeframes_to_persist] universe = load_universe(DB_PATH, exclude_crypto=not args.include_crypto) if args.limit_tickers and len(universe) > args.limit_tickers: universe = universe[: args.limit_tickers] mode_label = "full" if args.full else "incremental" log.info("Procesando %d tickers | timeframes=%s | mode=%s", len(universe), timeframes_to_persist, mode_label) if scalping_tfs: log.info("Bloque SCALPING activo (%s) — yfinance solo da 60 dias; el " "historico se acumula con cada ejecucion.", scalping_tfs) n_ok = 0 n_fail = 0 n_rows_total = 0 n_full_downloads = 0 n_incr_downloads = 0 for i, ticker in enumerate(universe, 1): rows_total = 0 got_any = False detail: list[str] = [] # --- Bloque 1h / 4h (1h nativo, 4h agregado) --- if need_1h: effective_period = args.period is_incremental = False if not args.full and _get_last_intraday_ts(ticker, "1h", DB_PATH): effective_period = args.incremental_period is_incremental = True df_1h = download_intraday(ticker, "1h", effective_period, log) if not df_1h.empty: got_any = True if "1h" in timeframes_to_persist: rows_total += upsert_bars(ticker, "1h", df_1h, DB_PATH) if "4h" in timeframes_to_persist: rows_total += upsert_bars(ticker, "4h", aggregate_to_4h(df_1h), DB_PATH) detail.append(f"1h:{len(df_1h)}") if is_incremental: n_incr_downloads += 1 else: n_full_downloads += 1 time.sleep(SLEEP_BETWEEN_TICKERS_S) # --- Bloque scalping (30m / 15m / 5m, nativos, max 60d) --- for tf in scalping_tfs: inc = (not args.full and _get_last_intraday_ts(ticker, tf, DB_PATH) is not None) period = args.incremental_period if inc else SCALPING_MAX_PERIOD df_s = download_intraday(ticker, tf, period, log) if not df_s.empty: got_any = True rows_total += upsert_bars(ticker, tf, df_s, DB_PATH) detail.append(f"{tf}:{len(df_s)}") if inc: n_incr_downloads += 1 else: n_full_downloads += 1 time.sleep(SLEEP_BETWEEN_TICKERS_S) if got_any: n_ok += 1 n_rows_total += rows_total log.info(" %4d/%d [OK] %s — +%d barras (%s)", i, len(universe), ticker, rows_total, ", ".join(detail) if detail else "-") else: n_fail += 1 log.info(" %4d/%d [FAIL] %s — sin datos", i, len(universe), ticker) sep = "=" * 64 print(f"\n{sep}") print("Leonex -- Datos Intraday") print(sep) print(f"Tickers procesados : {len(universe)}") print(f" OK : {n_ok}") print(f" FAIL : {n_fail}") print(f"Descargas full : {n_full_downloads}") print(f"Descargas increm. : {n_incr_downloads}") print(f"Barras persistidas : {n_rows_total}") print(f"Timeframes : {timeframes_to_persist}") print(f"Mode : {mode_label}") print(sep) return 0 if n_fail < len(universe) else 1 if __name__ == "__main__": raise SystemExit(main())