# ============================================================ # data_sources.py — Origen de datos y caché en memoria # ------------------------------------------------------------ # NOTA PARA AUDITORÍA: # - Este módulo centraliza la obtención y el almacenamiento en caché # de series históricas (p. ej., precios de cierre diarios con yfinance # y BTC a 1h desde CryptoCompare), sin efectos secundarios al importar. # - Expone: # * all_historical_data: dict[str, pandas.DataFrame] # * get_data(ticker, start, end): descarga con yfinance si falta en caché # * load_btc_hourly_cryptocompare_to_cache(...): descarga BTC 1h desde # CryptoCompare y lo transforma a cierre de las 16:00 (NY), como en tu # código original, guardando además CSV local. # * save_cache_snapshot(...): guarda a CSVs el contenido del caché. # - Las API keys se leen de variables de entorno: # * CRYPTOCOMPARE_API_KEY # * TIMEZONE_TARGET (por defecto: 'America/New_York') # ============================================================ from __future__ import annotations import os import time from datetime import datetime, timedelta from typing import Dict, Optional import pandas as pd import requests import yfinance as yf from config import ( SCRAPERAPI_KEY, CRYPTOCOMPARE_API_KEY, TIMEZONE_TARGET, GRADIO_SHARE, CRYPTOCOMPARE_BASE_URL, BITCOINTREASURIES_STOCKS_URL, BITCOINTREASURIES_ETFS_URL, YF_SEARCH_ENDPOINT, TICKERS_TO_IDS ) # ------------------------------------------------------------ # Caché global de históricos (compartido con analytics.py) # ------------------------------------------------------------ # Clave: ticker (str), Valor: DataFrame con índice de fechas. all_historical_data: Dict[str, pd.DataFrame] = {} # ------------------------------------------------------------ # Configuración (entorno) # ------------------------------------------------------------ CRYPTOCOMPARE_API_KEY = os.getenv("CRYPTOCOMPARE_API_KEY", "TU_API_KEY_DE_CRYPTOCOMPARE") # opcional TIMEZONE_TARGET = os.getenv("TIMEZONE_TARGET", "America/New_York") # para conversión de BTC 1h CRYPTOCOMPARE_BASE_URL = "https://min-api.cryptocompare.com/data/v2/histohour" # ============================================================ # Utilidad: descarga condicional con yfinance y guarda en caché # ============================================================ def get_data( ticker: str, start: Optional[datetime] = None, end: Optional[datetime] = None, silent: bool = False, ) -> pd.DataFrame: """ Obtiene datos de un ticker desde el caché; si no existen, descarga con yfinance. Parámetros ---------- ticker : str Símbolo (p. ej., 'MSTR', 'BTC-USD', 'TSLA'). start : datetime, opcional Fecha inicio (por defecto: ahora - 10 años). end : datetime, opcional Fecha fin (por defecto: ahora). silent : bool Si True, no imprime mensajes informativos. Devuelve -------- pandas.DataFrame DataFrame de precios (tal como lo devuelve yfinance, columnas OHLCV). Si la descarga falla, devuelve DataFrame vacío. """ end = end or datetime.now() start = start or (end - timedelta(days=10 * 365)) if ticker in all_historical_data and not all_historical_data[ticker].empty: if not silent: print(f"[data_sources] Usando caché para {ticker} ({len(all_historical_data[ticker])} filas).") return all_historical_data[ticker] if not silent: print(f"[data_sources] Descargando {ticker} con yfinance... ({start.date()} → {end.date()})") df = yf.download(ticker, start=start, end=end) if not df.empty: all_historical_data[ticker] = df if not silent: print(f"[data_sources] {ticker}: {len(df)} filas descargadas y cacheadas.") else: if not silent: print(f"[data_sources] Error al descargar {ticker} con yfinance.") return all_historical_data.get(ticker, pd.DataFrame()) # ============================================================ # Descarga BTC 1h desde CryptoCompare y normaliza a cierre 16:00 # (misma lógica que en tu `procesos 6.py`, empaquetada en función) # ============================================================ def load_btc_hourly_cryptocompare_to_cache( symbol: str = "BTC", currency: str = "USD", limit_per_call: int = 2000, years_back: int = 10, save_csv_path: str = "btc_hourly_cryptocompare_10_years.csv", sleep_seconds: float = 1.0, silent: bool = False, ) -> pd.DataFrame: """ Descarga datos horarios de BTC desde CryptoCompare, retrocediendo ~N años, concatena y normaliza: - Convierte índice a 'America/New_York' (o TIMEZONE_TARGET), - Filtra la hora 16:00 (cierre), - Ajusta nombres de columnas, - Guarda CSV, - Deja en caché bajo la clave 'BTC-USD'. Parámetros ---------- symbol : str Cripto de origen (por defecto: 'BTC'). currency : str Moneda de destino (por defecto: 'USD'). limit_per_call : int Máximo de filas por llamada (CryptoCompare permite 2000). years_back : int Años hacia atrás (aprox.) para descargar. save_csv_path : str Ruta donde guardar el CSV final. sleep_seconds : float Pausa entre llamadas para respetar rate limits. silent : bool Si True, reduce los prints informativos. Devuelve -------- pandas.DataFrame DataFrame final (filtrado a 16:00 en TIMEZONE_TARGET) y cacheado como 'BTC-USD'. """ end_time = datetime.now() start_time = end_time - timedelta(days=years_back * 365) to_timestamp = int(end_time.timestamp()) all_data = [] if not silent: print(f"[data_sources] Fetching {symbol}/{currency} hourly data from " f"{start_time.strftime('%Y-%m-%d')} to {end_time.strftime('%Y-%m-%d')}") while True: params = { "fsym": symbol, "tsym": currency, "limit": limit_per_call, "toTs": to_timestamp, "api_key": CRYPTOCOMPARE_API_KEY, # opcional } try: response = requests.get(CRYPTOCOMPARE_BASE_URL, params=params) data = response.json() if data.get("Response") == "Error": if not silent: print(f"[data_sources] API Error: {data.get('Message')}") break ohlcv_data = data.get("Data", {}).get("Data", []) if not ohlcv_data: break all_data.extend(ohlcv_data) # Paginación hacia atrás: fijar to_timestamp al primer punto - 1s to_timestamp = ohlcv_data[0]["time"] - 1 oldest_dt = datetime.fromtimestamp(ohlcv_data[0]["time"]) if not silent: print(f"[data_sources] Hasta: {oldest_dt.strftime('%Y-%m-%d %H:%M:%S')} " f"(acumuladas: {len(all_data)})") if oldest_dt <= start_time: if not silent: print(f"[data_sources] Alcanzado inicio deseado ({start_time.strftime('%Y-%m-%d')}).") break time.sleep(sleep_seconds) except requests.exceptions.RequestException as e: if not silent: print(f"[data_sources] Network error: {e}") time.sleep(5) except Exception as e: if not silent: print(f"[data_sources] Unexpected error: {e}") break # Construcción del DataFrame y ordenación btc_hour = pd.DataFrame(all_data) if btc_hour.empty: if not silent: print("[data_sources] No se recibieron datos de CryptoCompare.") return pd.DataFrame() btc_hour["time"] = pd.to_datetime(btc_hour["time"], unit="s") btc_hour.set_index("time", inplace=True) btc_hour.sort_index(inplace=True) # Eliminar duplicados por seguridad btc_hour = btc_hour[~btc_hour.index.duplicated(keep="first")] # Guardado del bruto unificado (opcional) try: btc_hour.to_csv(save_csv_path) if not silent: print(f"[data_sources] Datos guardados en {save_csv_path}") except Exception as e: if not silent: print(f"[data_sources] Aviso: no se pudo guardar CSV ({e})") # Ajuste de zona horaria y filtrado a las 16:00 btc_hour.index = btc_hour.index.tz_localize("UTC") btc_hour.index = btc_hour.index.tz_convert(TIMEZONE_TARGET) btc_hour_4pm = btc_hour[btc_hour.index.hour == 16].copy() # Índice solo fecha, columnas con capitalización inicial y dtype datetime64[ns] btc_hour_4pm.index = btc_hour_4pm.index.date btc_hour_4pm.columns = btc_hour_4pm.columns.str.title() btc_hour_4pm.index = pd.to_datetime(btc_hour_4pm.index) # Cachear con clave 'BTC-USD' para compatibilidad con analytics.py all_historical_data["BTC-USD"] = btc_hour_4pm if not silent: print(f"[data_sources] BTC-USD (16:00 {TIMEZONE_TARGET}) en caché: " f"{len(btc_hour_4pm)} filas.") return btc_hour_4pm # ============================================================ # Guardar una instantánea del caché a CSVs (uno por ticker) # ============================================================ def save_cache_snapshot(directory: str = "cache_snapshot", silent: bool = False) -> None: """ Guarda todos los DataFrames del caché en CSVs individuales dentro de `directory`. Parámetros ---------- directory : str Carpeta donde guardar los CSV (se crea si no existe). silent : bool Si True, no imprime mensajes. """ os.makedirs(directory, exist_ok=True) count = 0 for ticker, df in all_historical_data.items(): if isinstance(df, pd.DataFrame) and not df.empty: path = os.path.join(directory, f"{ticker.replace('/', '-')}.csv") try: df.to_csv(path) count += 1 except Exception as e: if not silent: print(f"[data_sources] No se pudo guardar {ticker} → {e}") if not silent: print(f"[data_sources] Instantánea del caché guardada: {count} archivo(s) en '{directory}'.") __all__ = [ "all_historical_data", "get_data", "load_btc_hourly_cryptocompare_to_cache", "save_cache_snapshot", ]