Download data.py from Quant0/space: direct link, hf CLI and curl.
- Browser
- Download file 13.2 kB
-
https://huggingface.co/spaces/Quant0/space/resolve/main/data.py
- Command line
-
hf download hf://spaces/Quant0/space/data.py
-
curl -L -o data.py https://huggingface.co/spaces/Quant0/space/resolve/main/data.py
13.2 kB
| 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}" | |