Blame for data_sources.py

# ============================================================
# 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",
]