Blame for analytics.py

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