| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
|
|
| from __future__ import annotations |
|
|
| import os |
| import re |
| import shutil |
| from typing import Optional, Tuple, Dict, List |
|
|
| import numpy as np |
| 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 |
| ) |
|
|
| from analytics import evolucion_correlacion |
| from scraping import descargar_tabla_web, limpiar_texto, expand_money |
|
|
| |
| import gdown |
|
|
|
|
| |
| |
| |
| def extract_file_id(url: str) -> Optional[str]: |
| """ |
| Extrae el ID del archivo de una URL de Google Drive. |
| |
| Acepta formatos habituales: |
| - https://drive.google.com/file/d/<ID>/view?usp=sharing |
| - https://docs.google.com/spreadsheets/d/<ID>/edit |
| - https://drive.google.com/open?id=<ID> |
| |
| Devuelve: |
| - str con el ID si se reconoce; en caso contrario, None. |
| """ |
| if not url: |
| return None |
|
|
| |
| m = re.search(r"/d/([a-zA-Z0-9_-]+)", url) |
| if m: |
| return m.group(1) |
|
|
| |
| m = re.search(r"[?&]id=([a-zA-Z0-9_-]+)", url) |
| if m: |
| return m.group(1) |
|
|
| return None |
|
|
|
|
| |
| |
| |
| def exportar_correlacion( |
| ticker_company: str, step: int, window_months: int |
| ) -> Tuple[Optional[str], str]: |
| """ |
| Exporta a Excel la serie de correlación rolling entre `ticker_company` |
| y BTC-USD (o el valor por defecto que usa `evolucion_correlacion`). |
| |
| Parámetros |
| ---------- |
| ticker_company : str |
| Ticker de la empresa/ETF (p. ej. 'MSTR'). |
| step : int |
| Salto de muestreo en días (si 0, devuelve solo el último valor). |
| window_months : int |
| Ventana rolling en meses (p. ej., 12). |
| |
| Devuelve |
| -------- |
| (str | None, str) |
| Ruta del archivo generado y un mensaje de estado. |
| """ |
| try: |
| serie = evolucion_correlacion(ticker_company, step, window_months=window_months) |
| if serie is not None and not serie.empty: |
| nombre_archivo = f"{ticker_company}_evolucion_correlacion.xlsx" |
| serie.to_excel(nombre_archivo) |
| return nombre_archivo, "Archivo Excel generado con éxito." |
| else: |
| return None, "No se pudieron generar los datos para exportar." |
| except Exception as e: |
| return None, f"Error al exportar: {e}" |
|
|
|
|
| |
| |
| |
| def descargar_resumen_holding_excel(template_url: str) -> Tuple[Optional[str], str]: |
| """ |
| Descarga una plantilla de Google Drive y genera un Excel con: |
| - Hoja 'Stocks' (datos de bitcointreasuries.net, correlación 12m, opciones...) |
| - Hoja 'ETFs' (datos de bitcointreasuries.net, opciones...) |
| |
| Lógica respetando tu implementación original: |
| 1) Descargar tabla de stocks (índice 3) y ETFs (índice 3) con ScraperAPI. |
| 2) Limpiar columnas, extraer Name y Ticker. |
| 3) Convertir 'BTC in USD' y 'Marketcap' a float (expand_money). |
| 4) Calcular ratio Holding/Mcap (stocks). |
| 5) Calcular correlación 12m para cada ticker (stocks). |
| 6) Detectar disponibilidad de opciones en yfinance (ambas hojas). |
| 7) Escribir hojas en un Excel a partir de una plantilla descargada con gdown. |
| |
| Parámetros |
| ---------- |
| template_url : str |
| URL de la plantilla en Google Drive (formato compartible). |
| |
| Devuelve |
| -------- |
| (str | None, str) |
| Ruta del Excel generado y mensaje de estado. |
| """ |
| try: |
| |
| |
| |
| file_id = extract_file_id(template_url) |
| if not file_id: |
| return None, "Error: No se pudo extraer el ID del archivo de la URL de Google Drive." |
|
|
| local_template_path = "plantilla_temporal.xlsx" |
| try: |
| gdown.download(id=file_id, output=local_template_path, quiet=False) |
| except Exception as e: |
| return None, f"Error al descargar la plantilla desde Google Drive: {e}" |
|
|
| |
| |
| |
| url_stocks = "https://bitcointreasuries.net/" |
| url_etfs = "https://bitcointreasuries.net/etfs-and-exchanges" |
|
|
| df_stocks_raw = descargar_tabla_web(url_stocks, indice_tabla=3) |
| df_etfs_raw = descargar_tabla_web(url_etfs, indice_tabla=3) |
|
|
| if df_stocks_raw is None and df_etfs_raw is None: |
| |
| try: |
| os.remove(local_template_path) |
| except Exception: |
| pass |
| return None, "Error: No se pudieron descargar los datos de holding." |
|
|
| |
| |
| |
| if df_stocks_raw is not None: |
| |
| |
| df_descargado_stocks = df_stocks_raw[[2, 3, 4, 5, 12]].copy() |
| df_descargado_stocks.columns = ['Name', 'Bitcoin', 'BTC in USD', 'Marketcap', '% Total BTC 21M'] |
|
|
| |
| df_descargado_stocks[['Name', 'Ticker']] = ( |
| df_descargado_stocks['Name'].apply(limpiar_texto).str.rsplit(' ', n=1, expand=True) |
| ) |
|
|
| |
| df_descargado_stocks['Bitcoin'] = df_descargado_stocks['Bitcoin'].apply(limpiar_texto) |
| df_descargado_stocks['Marketcap'] = df_descargado_stocks['Marketcap'].apply(limpiar_texto) |
|
|
| df_descargado_stocks['BTC in USD'] = df_descargado_stocks['BTC in USD'].apply(expand_money) |
| df_descargado_stocks['Marketcap'] = df_descargado_stocks['Marketcap'].apply(expand_money) |
|
|
| |
| df_descargado_stocks['Ratio Holding/Mcap'] = ( |
| df_descargado_stocks['BTC in USD'] / df_descargado_stocks['Marketcap'] |
| ) |
|
|
| |
| corrs: List[float] = [] |
| for stock in df_descargado_stocks['Ticker']: |
| try: |
| serie = evolucion_correlacion(stock, step=0, window_months=12) |
| corr_value = float(serie.iloc[0]) if (serie is not None and not serie.empty) else np.nan |
| corrs.append(corr_value) |
| except Exception: |
| corrs.append(np.nan) |
|
|
| df_descargado_stocks['Correlación con BTC (12 meses)'] = pd.Series(corrs) |
| df_descargado_stocks.set_index('Ticker', inplace=True) |
| else: |
| df_descargado_stocks = pd.DataFrame() |
|
|
| |
| |
| |
| if df_etfs_raw is not None: |
| |
| |
| df_descargado_ETFs = df_etfs_raw[[2, 3, 4, 5]].copy() |
| df_descargado_ETFs.columns = ['Name', 'Bitcoin', 'BTC in USD', '% Total BTC 21M'] |
|
|
| df_descargado_ETFs[['Name', 'Ticker']] = ( |
| df_descargado_ETFs['Name'].apply(limpiar_texto).str.rsplit(' ', n=1, expand=True) |
| ) |
|
|
| |
| df_descargado_ETFs['Ticker'].fillna(df_descargado_ETFs['Name'], inplace=True) |
|
|
| df_descargado_ETFs['BTC in USD'] = df_descargado_ETFs['BTC in USD'].apply(expand_money) |
| df_descargado_ETFs['Bitcoin'] = df_descargado_ETFs['Bitcoin'].apply(limpiar_texto).apply(expand_money) |
|
|
| df_descargado_ETFs.set_index('Ticker', inplace=True) |
| else: |
| df_descargado_ETFs = pd.DataFrame() |
|
|
| |
| |
| |
| tickers: List[str] = [] |
| if not df_descargado_stocks.empty: |
| tickers.extend(df_descargado_stocks.index.tolist()) |
| if not df_descargado_ETFs.empty: |
| tickers.extend(df_descargado_ETFs.index.tolist()) |
|
|
| options_bool: Dict[str, bool] = {} |
| for t in tickers: |
| if t is None: |
| continue |
| try: |
| options_bool[t] = False if not yf.Ticker(t).options else True |
| except Exception: |
| options_bool[t] = False |
|
|
| options_series = pd.Series(options_bool) |
|
|
| if not df_descargado_stocks.empty: |
| df_descargado_stocks['Options'] = options_series |
| if not df_descargado_ETFs.empty: |
| df_descargado_ETFs['Options'] = options_series |
|
|
| |
| |
| |
| nuevo_archivo_path = "resumen_datos_holding.xlsx" |
| try: |
| shutil.copyfile(local_template_path, nuevo_archivo_path) |
| except FileNotFoundError: |
| |
| try: |
| os.remove(local_template_path) |
| except Exception: |
| pass |
| return None, f"Error: No se encontró la plantilla en la ruta: {local_template_path}" |
|
|
| |
| with pd.ExcelWriter(nuevo_archivo_path, engine="openpyxl", mode="a", if_sheet_exists="replace") as writer: |
| if not df_descargado_stocks.empty: |
| df_descargado_stocks.to_excel(writer, sheet_name="Stocks", index=True) |
| if not df_descargado_ETFs.empty: |
| df_descargado_ETFs.to_excel(writer, sheet_name="ETFs", index=True) |
|
|
| |
| try: |
| os.remove(local_template_path) |
| except Exception: |
| pass |
| return nuevo_archivo_path, "Archivo Excel generado con éxito." |
|
|
| except Exception as e: |
| |
| return None, f"Error al exportar: {e}" |
|
|
|
|
| __all__ = [ |
| "exportar_correlacion", |
| "descargar_resumen_holding_excel", |
| "extract_file_id", |
| ] |
|
|
