Create data.py

data.py

@@ -0,0 +1,320 @@
+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, api_url, tickers_to_ids
+from utils import limpiar_texto, expand_money, descargar_tabla_web
+
+
+#     -----------------------------------------------
+#      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 = ['Name','Bitcoin','% Total BTC 21M']
+            df_descargado_stocks['Name'] = df_descargado_stocks['Name'].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 = ['Name','Bitcoin','% 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['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()
+
+
+        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}"
+
+
+