Spaces:

Quant0
/

AFI

Sleeping

AFI / data_sources.py

Update data_sources.py

b05eae8 verified over 1 year ago

10.4 kB

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

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