Spaces:

Quant0
/

AFI

Sleeping

AFI / analytics.py

Update analytics.py

fb1db80 verified over 1 year ago

8.12 kB

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

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