Create analytics.py

analytics.py

@@ -0,0 +1,177 @@
+# ============================================================
+# 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
+
+# ------------------------------------------------------------
+# 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]