import pandas as pd import gradio as gr import ccxt import numpy as np import matplotlib.pyplot as plt import yfinance as yf import requests import pandas as pd import time import re import gdown import shutil import os from datetime import datetime, timedelta from config import API_KEY, SYMBOL, CURRENCY, LIMIT_PER_CALL, BASE_URL, YEARS, api_key,tickers_to_ids from utils import limpiar_texto, expand_money, descargar_tabla_web, extract_file_id # ----------------------------------------------- # DESCARGA DE DATOS DE BTC: all_historical_data # ----------------------------------------------- end_time = datetime.now() start_time = end_time - timedelta(days=YEARS*365) # 10 años atrás aproximadamente # Convertir a timestamp de Unix (segundos) to_timestamp = int(end_time.timestamp()) all_data = [] print(f"Fetching BTC/{CURRENCY} hourly data from {start_time.strftime('%Y-%m-%d')} to {end_time.strftime('%Y-%m-%d')}") while True: params = { "fsym": SYMBOL, "tsym": CURRENCY, "limit": LIMIT_PER_CALL, "toTs": to_timestamp, "api_key": API_KEY # Si tienes una clave } try: response = requests.get(BASE_URL, params=params) data = response.json() if data['Response'] == 'Error': print(f"API Error: {data['Message']}") break ohlcv_data = data['Data']['Data'] if not ohlcv_data: break # No more data to fetch all_data.extend(ohlcv_data) # El siguiente `toTs` es el timestamp del primer punto de datos obtenido menos 1 hora (o 1 segundo) # para asegurarnos de no duplicar y retroceder en el tiempo. # Asegúrate de que los datos estén ordenados por tiempo ascendente desde la API. # CryptoCompare devuelve datos desde 'fromTs' hasta 'toTs' en orden ascendente, # pero cuando usamos `toTs`, devuelve los datos *anteriores* a `toTs`. # Por lo tanto, el nuevo `toTs` debe ser el 'time' del primer elemento de la llamada actual. to_timestamp = ohlcv_data[0]['time'] - 1 # Retrocede 1 segundo para la siguiente llamada # Convertir el timestamp del último dato para referencia oldest_data_time = datetime.fromtimestamp(ohlcv_data[0]['time']) print(f"Fetched data up to {oldest_data_time.strftime('%Y-%m-%d %H:%M:%S')}. Total records: {len(all_data)}") if oldest_data_time <= start_time: print(f"Reached desired start time ({start_time.strftime('%Y-%m-%d')}). Stopping.") break time.sleep(1) # Respetar límites de rate de la API except requests.exceptions.RequestException as e: print(f"Network error: {e}") time.sleep(5) # Esperar y reintentar except Exception as e: print(f"An unexpected error occurred: {e}") break # Convertir la lista de diccionarios a un DataFrame de Pandas btc_hour_10y = pd.DataFrame(all_data) btc_hour_10y['time'] = pd.to_datetime(btc_hour_10y['time'], unit='s') btc_hour_10y.set_index('time', inplace=True) btc_hour_10y.sort_index(inplace=True) # Asegurarse de que esté ordenado cronológicamente # Eliminar duplicados si los hubiera (puede pasar si las llamadas se solapan) btc_hour_10y = btc_hour_10y[~btc_hour_10y.index.duplicated(keep='first')] print("\nDataFrame Head:") print(btc_hour_10y.head()) print("\nDataFrame Tail:") print(btc_hour_10y.tail()) print(f"\nTotal records fetched: {len(btc_hour_10y)}") # Guardar a CSV btc_hour_10y.to_csv('btc_hourly_cryptocompare_10_years.csv') print("Data saved to btc_hourly_cryptocompare_10_years.csv") # Ajuste del formato: btc_hour_10y.index = btc_hour_10y.index.tz_localize('UTC') btc_hour_10y.index = btc_hour_10y.index.tz_convert('America/New_York') #Convertir el dataframe de UTC a 'America NY' btc_hour_10y = btc_hour_10y[btc_hour_10y.index.hour == 16] # Filtrar el dataframe para quedarnos la hora de cierre 4:00 PM EST btc_hour_10y.index = btc_hour_10y.index.date #Limpiamos el índice para que sólo muestre las fechas de los días btc_hour_10y.columns = btc_hour_10y.columns.str.title() #Ponemos los headers con la primera letra en Mayuscula btc_hour_10y.index = pd.to_datetime(btc_hour_10y.index) # Para coincidir con el formato de yahoofinance debemos convertir los indices a fechas en formato: dtype='datetime64[ns]' all_historical_data = { 'BTC-USD': btc_hour_10y } # ----------------------------------------------- # TOP TICKERS DE BTC TREASURY # ----------------------------------------------- ### descargamos por orden las empresas mas importantes def generar_tickers(): """ Genera una lista global de tickers a partir de las tablas de stocks y ETFs. """ global tickers tickers = [] # URLs de donde obtener los datos url_stocks = "https://bitcointreasuries.net/" url_etfs = "https://bitcointreasuries.net/etfs-and-exchanges" # Descargar tablas web (se asume que estas funciones existen) df_stocks = descargar_tabla_web(url_stocks, indice_tabla=3) df_etfs = descargar_tabla_web(url_etfs, indice_tabla=3) # Procesar Stocks si existen if df_stocks is not None and not df_stocks.empty: # Asumiendo que limpiar_texto ya está definida df_stocks['Name'] = df_stocks[2].apply(limpiar_texto) tickers.extend(df_stocks['Name'].dropna().tolist()) # Procesar ETFs si existen if df_etfs is not None and not df_etfs.empty: df_etfs['Name'] = df_etfs[2].apply(limpiar_texto) tickers.extend(df_etfs['Name'].dropna().tolist()) generar_tickers() # ----------------------------------------------- # DESCARGA TOP HOLDING BTC TREASURY # ----------------------------------------------- def top_holding(): url_stocks = "https://bitcointreasuries.net/" url_ETFs = "https://bitcointreasuries.net/etfs-and-exchanges" df_stocks = descargar_tabla_web(url_stocks, indice_tabla=3) df_etfs = descargar_tabla_web(url_ETFs, indice_tabla=3) if df_stocks is None and df_etfs is None: return None, "Error: No se pudieron descargar los datos de holding." # Procesar Stocks if df_stocks is not None: df_descargado_stocks = df_stocks[[2,3,12]].copy() df_descargado_stocks.columns = ['Stocks','Bitcoin','% Total BTC 21M'] df_descargado_stocks['Stocks'] = df_descargado_stocks['Stocks'].apply(limpiar_texto) df_descargado_stocks['Bitcoin'] = df_descargado_stocks['Bitcoin'].apply(limpiar_texto) else: df_descargado_stocks = pd.DataFrame() # Procesar ETFs if df_etfs is not None: df_descargado_ETFs = df_etfs[[2,3,5]].copy() df_descargado_ETFs.columns = ['ETFs','Bitcoin','% Total BTC 21M'] df_descargado_ETFs[['ETFs','Ticker']] = df_descargado_ETFs['ETFs'].apply(limpiar_texto).str.rsplit(' ', n=1, expand=True) df_descargado_ETFs['ETFs'].fillna(df_descargado_ETFs['ETFs'], inplace=True) df_descargado_ETFs['Bitcoin'] = df_descargado_ETFs['Bitcoin'].apply(limpiar_texto) df_descargado_ETFs.set_index('Ticker', inplace=True) else: df_descargado_ETFs = pd.DataFrame() return df_descargado_ETFs, df_descargado_stocks # ----------------------------------------------- # DESCARGA HOLDING BTC TREASURY # ----------------------------------------------- def descargar_resumen_holding_excel(template_url): """ Descarga los datos de holding de BTC para stocks y ETFs, los procesa, calcula correlaciones y exporta a un archivo Excel usando una plantilla de Google Drive. """ global df_descargado_stocks, df_descargado_ETFs try: # Extraer el ID del archivo de la URL 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' # Descargar la plantilla de Google Drive 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 = descargar_tabla_web(url_stocks, indice_tabla=3) df_etfs = descargar_tabla_web(url_ETFs, indice_tabla=3) if df_stocks is None and df_etfs is None: return None, "Error: No se pudieron descargar los datos de holding." # Procesar Stocks if df_stocks is not None: df_descargado_stocks = df_stocks[[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'] corrls = [] for stock in df_descargado_stocks['Ticker']: try: corr_value = evolucion_correlacion(stock, step=0, window_months=12).iloc[0] corrls.append(corr_value) except Exception as e: print(f"Error al calcular correlación para {stock}: {e}") corrls.append(np.nan) df_descargado_stocks['Correlación con BTC (12 meses)'] = pd.Series(corrls) df_descargado_stocks.set_index('Ticker', inplace=True) else: df_descargado_stocks = pd.DataFrame() # Procesar ETFs if df_etfs is not None: df_descargado_ETFs = df_etfs[[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() # Añadir la columna de 'Options' como solicitaste tckrs = [] if not df_descargado_stocks.empty: tckrs.extend(df_descargado_stocks.index.tolist()) if not df_descargado_ETFs.empty: tckrs.extend(df_descargado_ETFs.index.tolist()) options_bool = {} for ticker in tckrs: if ticker is not None: try: options_bool[ticker] = False if not yf.Ticker(ticker).options else True except: options_bool[ticker] = 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 # Exportación a Excel if not df_descargado_stocks.empty or not df_descargado_ETFs.empty: nuevo_archivo_path = 'resumen_datos_holding.xlsx' # Copiar la plantilla descargada antes de escribir try: shutil.copyfile(local_template_path, nuevo_archivo_path) except FileNotFoundError: return None, f"Error: No se encontró la plantilla en la ruta: {local_template_path}" # Escribir en el archivo copiado 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) # Eliminar la plantilla temporal descargada para limpiar el entorno os.remove(local_template_path) return nuevo_archivo_path, "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}"