Blame for processes.py

import pandas as pd
import yfinance as yf
from datetime import datetime, timedelta
from data import all_historical_data, top_holding
from config import tickers_to_ids
import plotly.graph_objects as go

#     -----------------------------------------------
#      Función de Correlación
#     -----------------------------------------------

def evolucion_correlacion(ticker_company, step, ticker_btc="BTC-USD", window_months=12):
    """
    Calcula la evolución de la correlación entre el precio de las acciones de una empresa y Bitcoin.
    Args:
        ticker_company (str): El símbolo de cotización de la empresa.
        step (int): El número de días para el desplazamiento de la correlación.
        ticker_btc (str): El símbolo de cotización de Bitcoin (por defecto: "BTC-USD").
        window_months (int): El tamaño de la ventana móvil en meses.
    outputs:
        pandas.Series: Una serie con la evolución de la correlación, o None si faltan datos.
    """
    end_date = datetime.now()
    start_date = end_date - timedelta(days=10*365)

    if ticker_company not in all_historical_data:
        print(f"Descargando datos para {ticker_company}...")
        data = yf.download(ticker_company, start=start_date, end=end_date)
        if not data.empty:
            all_historical_data[ticker_company] = data
            print(f"Datos de {ticker_company} descargados con éxito.")
        else:
            print(f"Error al descargar datos para {ticker_company}.")
            return None

    if ticker_btc not in all_historical_data:
        print(f"Descargando datos para {ticker_btc}...")
        data = yf.download(ticker_btc, start=start_date, end=end_date)
        if not data.empty:
            all_historical_data[ticker_btc] = data
            print(f"Datos de {ticker_btc} descargados con éxito.")
        else:
            print(f"Error al descargar datos para {ticker_btc}.")
            return None

    company_data = all_historical_data.get(ticker_company)
    btc_data = all_historical_data.get(ticker_btc)

    if company_data.empty or btc_data.empty:
        print(f"No podemos calcular la correlación dinámica: No hay datos disponibles para {ticker_company} o {ticker_btc}.")
        return None

    company_prices = company_data['Close']
    btc_prices = btc_data['Close']

    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

    company_returns = combined_prices[ticker_company].pct_change().dropna()
    btc_returns = combined_prices[ticker_btc].pct_change().dropna()

    window_days = window_months * 30
    if step == 0:
        dynamic_corr = company_returns.rolling(window=f"{window_days}D").corr(btc_returns).iloc[-1]
        return pd.Series([dynamic_corr], index=[end_date])
    else:
        dynamic_corr = company_returns.rolling(window=f"{window_days}D").corr(btc_returns).dropna()
        return dynamic_corr.iloc[9::step]




#     -----------------------------------------------
#                PLOT DE LA CORRELACIÓN
#     -----------------------------------------------



def plot_evolucion_correlacion_gradio(ticker_company, step, window_months):

    """
    Genera el gráfico de la evolución de la correlación.
    """
    ticker_company = ticker_company.upper()
    serie = evolucion_correlacion(ticker_company, step, window_months=12)

    if serie is None or serie.empty:
        return None, f"No hay datos de correlación disponibles para {ticker_company}."

    # --- Create the Plotly graph ---
    fig = go.Figure()

    # Add the correlation series
    fig.add_trace(go.Scatter(
        x=serie.index,
        y=serie.values,
        mode='lines',
        name='Correlación',
        line=dict(color='#2263eb', width=2), # electric blue color for the line
        fill='tozeroy', # Fill the area down to Y=0 axis
        fillcolor='rgba(34,99,235,0.1)', # Very faint elctric blue shadow (10% opacity)
        hovertemplate='Fecha: %{x|%Y-%m-%d}<br>Correlación: %{y:.4f}<extra></extra>'
    ))

    # --- Configure axes and range selector ---
    fig.update_layout(
        title_text=f'<b>Evolución de la correlación entre {ticker_company} y BTC-USD<b>',
        title_x=0, # Align title to the left
        height=500, # Graph height
        template="plotly_white", # Graph theme (clean and modern)
        margin=dict(l=20, r=20, t=60, b=20), # Graph margins
        showlegend=False, # No legend needed if only one main series

        xaxis=go.layout.XAxis(
            rangeslider=dict(
                visible=True,
                thickness=0.08, # Rangeslider thickness
                bgcolor="rgba(50, 0, 128, 0.1)", # Very transparent purple for rangeslider
                bordercolor="rgba(50, 0, 128, 0.2)", # Slightly more visible border
                borderwidth=1,
                range=[serie.index.min(), serie.index.max()] # Initial slider range
            ),
            rangeselector=dict(
                buttons=list([
                    dict(count=1, label="1m", step="month", stepmode="backward"),
                    dict(count=3, label="3m", step="month", stepmode="backward"),
                    dict(count=6, label="6m", step="month", stepmode="backward"),
                    dict(step="year", count=1, label="YTD", stepmode="todate"), # YTD
                    dict(count=1, label="1y", step="year", stepmode="backward"),
                    dict(step="all", label="All")
                ]),
                bgcolor="rgba(247,247,247,255)", # Transparent background for buttons
                bordercolor="rgba(0,0,0,0)", # Transparent border
                borderwidth=0,
                activecolor="rgba(230,233,255,255)", # Active button color (slightly darker purple)
                font=dict(color="#000000", size=10), # Button font style
                x=0.9, y=1.1, xanchor='right', yanchor='top' # Button position
            ),
            type="date",
            domain=[0, 1], # Occupy full graph width
            rangeslider_yaxis_rangemode="auto", # Rangeslider Y-axis range adjusts automatically
            showgrid=False,
            gridcolor='#e0e0e0',
            zeroline=False
        ),
        yaxis=go.layout.YAxis(
            title_text='Correlación',
            side='left',
            showgrid=True,
            gridcolor='#e0e0e0',
            zeroline=False,
            tickformat=".2f", # 2 decimal places format
            range=[-1.0, 1.0] # Fixed range for correlation
        ),
        hovermode="x unified", # Unify hover to show data for the series
        paper_bgcolor='rgba(255,255,255,1)', # Transparent background for notebook integration
        plot_bgcolor='rgba(255,255,255,1)'
    )

    # --- Add annotation for max value (similar to Matplotlib) ---
    max_value = serie.max()
    max_index = serie.idxmax()
    fig.add_annotation(
        x=max_index,
        y=max_value,
        text=f'Corr_max: {max_value:.4f}<br>Fecha: {max_index.date()}',
        showarrow=True,
        arrowhead=2,
        ax=0,
        ay=-40, # Adjust vertical position of the annotation
        bgcolor="rgba(255, 255, 255, 0.8)", # Semi-transparent white background
        bordercolor="#c7c7c7",
        borderwidth=1,
        borderpad=4,
        font=dict(size=10, color="#333"),
        # You can adjust text position with xanchor/yanchor if needed
    )

    return fig, "Gráfico de correlación generado."



#     -----------------------------------------------
#             EXPORTACIÓN DE LA CORRELACIÓN
#     -----------------------------------------------


def exportar_correlacion(ticker_company, step, window_months):

    """
    Exporta la serie de correlación a un archivo Excel.
    """
    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}"


#     -----------------------------------------------
#          EMBED PLOT STOCK VS BTC DE BTC TREASURY
#     -----------------------------------------------


def get_iframe_stock_vs_btc(ticker):
    """
    Genera el código HTML para el iframe de Stock vs Bitcoin.
    """
    try:
        entity_id = tickers_to_ids[ticker]
        iframe_code = f"""
            <iframe src="https://bitcointreasuries.net/embed?component=StockVsBitcoinChart&entityId={entity_id}"
            title="Bitcoin Treasuries" credentialless style="width: 100%; height: 660px; border: none;"></iframe>
        """
        return iframe_code
    except KeyError:
        return f"No hay datos de holding de Bitcoin para {ticker}."



#     -----------------------------------------------
#          EMBED PLOT HOLDING BTC DE BTC TREASURY
#     -----------------------------------------------


def get_iframe_holdings(ticker):
    """
    Genera el código HTML para el iframe de Bitcoin Holdings.
    """
    try:
        entity_id = tickers_to_ids[ticker]
        iframe_code = f"""
            <iframe src="https://bitcointreasuries.net/embed?component=BitcoinHoldingsChart&entityId={entity_id}"
            title="Bitcoin Holdings Chart" credentialless style="width: 100%; height: 660px; border: none;"></iframe>
        """
        return iframe_code
    except KeyError:
        return f"No hay datos de holding de Bitcoin para {ticker}."




#     -----------------------------------------------
#          TABLAS TOP HOLDING RESUMEN
#     -----------------------------------------------



df_descargado_ETFs, df_descargado_stocks = top_holding()
df_descargado_ETFs = df_descargado_ETFs.reset_index()

# Crear un DataFrame de ejemplo
top5_ETFs = df_descargado_ETFs.copy().iloc[:,[1,2,3]].head()
top5_ETFs = pd.DataFrame(top5_ETFs)

top5_stocks = df_descargado_stocks.copy().iloc[:,[0,1,2]].head()
top5_stocks = pd.DataFrame(top5_stocks)

# Aplicar estilo con color de marca y bordes redondeados

AZUL_BORDE = "#0085CA"
GRIS_HEADER = "#F5F6F8"
BLANCO = "#FFFFFF"

styles = [
    # Perímetro y comportamiento de bordes
    dict(selector="table", props=[
        ("border", f"1.5px solid {AZUL_BORDE} !important"),
        ("border-radius", "11px"),
        ("border-collapse", "separate !important"),
        ("border-spacing", "0"),
        ("background-color", BLANCO)
    ]),
    # Eliminar TODOS los bordes internos
    dict(selector="th, td", props=[
        ("border", "none !important"),
        ("background-color", BLANCO)
    ]),
    # Header gris muy claro
    dict(selector="thead th", props=[
        ("background-color", GRIS_HEADER)
    ]),
    # Redondear esquinas visibles
    dict(selector="thead tr:first-child th:first-child", props=[
        ("border-top-left-radius", "12px")
    ]),
    dict(selector="thead tr:first-child th:last-child", props=[
        ("border-top-right-radius", "12px")
    ]),
    dict(selector="tbody tr:last-child td:first-child", props=[
        ("border-bottom-left-radius", "12px")
    ]),
    dict(selector="tbody tr:last-child td:last-child", props=[
        ("border-bottom-right-radius", "12px")
    ]),
]


styled_stocks = top5_stocks.style.set_table_styles(styles).hide(axis="index")
styled_stocks = styled_stocks.set_table_attributes(
    f'style="border:1.5px solid {AZUL_BORDE}; border-radius:11px; border-collapse:separate; border-spacing:0; background:#fff"'
)

styled_ETFs = top5_ETFs.style.set_table_styles(styles).hide(axis="index")
styled_ETFs = styled_ETFs.set_table_attributes(
    f'style="border:1.5px solid {AZUL_BORDE}; border-radius:11px; border-collapse:separate; border-spacing:0; background:#fff"'
)