# ============================================================ # analytics.py — Cálculo de correlaciones entre activos y BTC # ------------------------------------------------------------ # NOTA PARA AUDITORÍA: # - Este módulo expone la función `evolucion_correlacion`, encargada de # calcular la correlación rolling entre el precio de cierre de un # activo (empresa/ETF) y BTC-USD. # - La función: # 1) Obtiene (o descarga si falta) los históricos con yfinance. # 2) Alinea por fecha y calcula rendimientos diarios (% change). # 3) Aplica correlación móvil (rolling window) de tamaño # `window_months * 30` días. # 4) Devuelve una Serie con la evolución o, si step=0, el último # valor de correlación como serie de un solo punto. # - Dependencias externas: yfinance, pandas, datetime. # - Caché en memoria: `all_historical_data` (dict[str, DataFrame]). # Este diccionario debe existir en tiempo de ejecución. Por defecto, # se intenta importar desde `data_sources` (propuesta de modularización) # o, si no existe, desde `processes` (compatibilidad con tu estructura # actual). Si no se encuentra, se crea un dict vacío. # - Recomendación futura: sustituir `print` por `logging`. # ============================================================ from __future__ import annotations from datetime import datetime, timedelta from typing import Optional import pandas as pd 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 ) # ------------------------------------------------------------ # Importación del caché global de históricos # - Primero intenta el módulo recomendado `data_sources` # - Si no existe, recurre al módulo histórico `processes` # - Como último recurso, define un dict vacío (seguirá funcionando, # pero descargará siempre con yfinance). # ------------------------------------------------------------ try: # Proyecto modularizado (recomendado) from data_sources import all_historical_data # type: ignore except Exception: try: # Compatibilidad con el proyecto actual from processes import all_historical_data # type: ignore except Exception: # Fallback mínimo all_historical_data = {} # type: ignore print( "[analytics] Advertencia: no se encontró 'all_historical_data' en " "'data_sources' ni en 'processes'. Se usará un diccionario vacío." ) __all__ = ["evolucion_correlacion"] def evolucion_correlacion( ticker_company: str, step: int, ticker_btc: str = "BTC-USD", window_months: int = 12, ) -> Optional[pd.Series]: """ Calcula la evolución de la correlación (rolling) entre un activo (empresa/ETF) y BTC-USD, usando precios de CIERRE diarios. Parámetros ---------- ticker_company : str Ticker del activo a analizar (p. ej., 'MSTR', 'TSLA', 'COIN'). step : int Salto en días para muestrear la serie de correlación resultante. - Si `step == 0`: devuelve SOLO el último valor de correlación (una Serie con un único punto indexado en `end_date`). - Si `step > 0`: devuelve la serie completa de correlación rolling muestreada cada `step` filas (conservando el offset original). ticker_btc : str, opcional Ticker de BTC (por defecto: "BTC-USD"). Debe existir o poder descargarse con yfinance. window_months : int, opcional Tamaño de la ventana móvil en meses (por defecto: 12). Internamente se convierte a días: `window_months * 30`. Devolución ---------- pandas.Series | None Serie temporal con la evolución de la correlación rolling. Devuelve `None` si faltan datos o no se pudieron descargar. Notas de implementación ----------------------- - Lógica conservada tal cual respecto a tu implementación original: * Descarga condicional con yfinance si el ticker no está en `all_historical_data`. * Cálculo de rendimientos con `pct_change()` y `dropna()`. * Ventana rolling expresada como texto con sufijo 'D' (días). * Cuando `step > 0`, se devuelve `dynamic_corr.iloc[9::step]`, manteniendo el mismo offset de tu código. - Recomendación: sustituir prints por `logging` y añadir control de reintentos. """ # Rango temporal de trabajo: ~10 años atrás hasta hoy (misma lógica) end_date = datetime.now() start_date = end_date - timedelta(days=10 * 365) # ------------------------------------------------------------ # Descarga condicional para el ticker de la empresa # ------------------------------------------------------------ if ticker_company not in all_historical_data: print(f"Descargando datos para {ticker_company}...") data_company = yf.download(ticker_company, start=start_date, end=end_date) if not data_company.empty: all_historical_data[ticker_company] = data_company print(f"Datos de {ticker_company} descargados con éxito.") else: print(f"Error al descargar datos para {ticker_company}.") return None # ------------------------------------------------------------ # Descarga condicional para BTC-USD (o el ticker_btc especificado) # ------------------------------------------------------------ if ticker_btc not in all_historical_data: print(f"Descargando datos para {ticker_btc}...") data_btc = yf.download(ticker_btc, start=start_date, end=end_date) if not data_btc.empty: all_historical_data[ticker_btc] = data_btc print(f"Datos de {ticker_btc} descargados con éxito.") else: print(f"Error al descargar datos para {ticker_btc}.") return None # ------------------------------------------------------------ # Extracción de cierres y alineación por fechas # ------------------------------------------------------------ company_data: pd.DataFrame = all_historical_data.get(ticker_company, pd.DataFrame()) btc_data: pd.DataFrame = all_historical_data.get(ticker_btc, pd.DataFrame()) if company_data.empty or btc_data.empty: print( f"No podemos calcular la correlación dinámica: " f"No hay datos disponibles para {ticker_company} o {ticker_btc}." ) return None company_prices = company_data["Close"] btc_prices = btc_data["Close"] # Alinear por índice (fecha) y eliminar faltantes combined_prices = pd.concat([company_prices, btc_prices], axis=1).dropna() combined_prices.columns = [ticker_company, ticker_btc] if combined_prices.empty: print( f"No hay datos históricos en común entre {ticker_company} y {ticker_btc}." ) return None # ------------------------------------------------------------ # Cálculo de rendimientos y correlación rolling # ------------------------------------------------------------ company_returns = combined_prices[ticker_company].pct_change().dropna() btc_returns = combined_prices[ticker_btc].pct_change().dropna() window_days = window_months * 30 # Mantener equivalencia con el código original if step == 0: # Correlación en la última fecha disponible (Serie 1x1) dynamic_corr_last = ( company_returns.rolling(window=f"{window_days}D").corr(btc_returns).iloc[-1] ) # Devolvemos una Serie con índice en la fecha actual (misma firma de retorno) return pd.Series([dynamic_corr_last], index=[end_date]) # Serie completa de correlación rolling, muestreada cada `step` elementos dynamic_corr = company_returns.rolling(window=f"{window_days}D").corr(btc_returns).dropna() # Conserva el muestreo con desplazamiento como en tu implementación (iloc[9::step]) return dynamic_corr.iloc[9::step]