app.py

import gradio as gr
import pandas as pd
import yfinance as yf
import numpy as np
import matplotlib.pyplot as plt
import plotly.express as px
import plotly.graph_objects as go
import seaborn as sns
from scipy import stats, optimize
import warnings
from io import BytesIO
import base64
from datetime import datetime

warnings.filterwarnings("ignore")

# Default parameters from the notebook
default_tickers = ['BBAS3.SA', 'ITSA4.SA', 'TAEE11.SA', 'TTEN3.SA', 'BPAC11.SA', '^BVSP']
default_start_date = '2012-01-01'
default_end_date = '2024-07-31'
num_ports = 50000
initial_investment = 35000

# Fetch data
def fetch_data(tickers, start_date, end_date):
    acoes_df = pd.DataFrame()
    for acao in tickers:
        try:
            acoes_df[acao] = yf.download(acao, start=start_date, end=end_date, progress=False)['Close']
        except Exception as e:
            print(f"Error downloading {acao}: {e}")
    acoes_df.index = acoes_df.index.strftime('%Y-%m-%d')
    acoes_df.reset_index(inplace=True)
    acoes_df.rename(columns={'index': 'Date'}, inplace=True)
    return acoes_df

# Historical Prices Plot
def plot_historical_prices(acoes_df):
    acoes_viz = acoes_df.copy()
    figura = px.line(title='Histórico do Preço das Ações')
    for i in acoes_viz.columns[1:]:
        figura.add_scatter(x=acoes_viz["Date"], y=acoes_viz[i], name=i)
    figura.update_layout(xaxis_title="Data", yaxis_title="Preço (R$)", hovermode='x unified')
    return figura

# Returns Calculation and Plot
def calculate_and_plot_returns(acoes_df):
    dataset = acoes_df.copy()
    dataset.drop(labels=['Date'], axis=1, inplace=True)
    taxas_retorno = np.log(dataset / dataset.shift(1))
    date = acoes_df.filter(["Date"])
    taxas_retorno = pd.concat([date, taxas_retorno], axis=1)
    taxas_retorno = taxas_retorno.dropna()

    figura = px.line(title='Histórico de Retorno das Ações')
    for i in taxas_retorno.columns[1:]:
        figura.add_scatter(x=taxas_retorno["Date"], y=taxas_retorno[i], name=i)
    figura.update_layout(xaxis_title="Data", yaxis_title="Retorno Logarítmico", hovermode='x unified')

    stats = taxas_retorno.describe().to_html()
    return figura, stats

# Correlation Matrix
def plot_correlation_matrix(taxas_retorno):
    correlacao_cols = taxas_retorno.select_dtypes(include=[np.number]).columns
    correlacao = taxas_retorno[correlacao_cols].corr()
    correlacao = np.round(correlacao, 2)
    custom_colorscale = [[0.0, 'green'], [0.5, 'blue'], [1.0, 'red']]
    fig = px.imshow(correlacao, text_auto=True, aspect="auto", color_continuous_scale=custom_colorscale,
                    labels=dict(color="Correlações"), zmin=-1, zmax=1, title="Matriz de Correlação dos Retornos")
    return fig

# Efficient Frontier
def simulate_efficient_frontier(acoes_df, num_ports):
    acoes_port = acoes_df.copy()
    acoes_port.drop(labels=['^BVSP'], axis=1, inplace=True)
    log_ret = acoes_port.copy()
    log_ret.drop(labels=["Date"], axis=1, inplace=True)
    log_ret = np.log(log_ret / log_ret.shift(1))
    log_ret = log_ret.dropna()

    np.random.seed(42)
    all_weights = np.zeros((num_ports, len(acoes_port.columns[1:])))
    ret_arr = np.zeros(num_ports)
    vol_arr = np.zeros(num_ports)
    sharpe_arr = np.zeros(num_ports)

    for x in range(num_ports):
        weights = np.array(np.random.random(5))
        weights = weights / np.sum(weights)
        all_weights[x, :] = weights
        ret_arr[x] = np.sum((log_ret.mean() * weights))
        vol_arr[x] = np.sqrt(np.dot(weights.T, np.dot(log_ret.cov(), weights)))
        sharpe_arr[x] = ret_arr[x] / vol_arr[x]

    melhores_pesos = all_weights[sharpe_arr.argmax(), :]
    max_sr_vol = vol_arr[sharpe_arr.argmax()]
    max_sr_ret = ret_arr[sharpe_arr.argmax()]

    fig, ax = plt.subplots(figsize=(12, 8))
    ax.scatter(vol_arr, ret_arr, c=sharpe_arr, cmap='viridis')
    ax.scatter(max_sr_vol, max_sr_ret, c='red', s=200, marker='*', label='Carteira Ótima')
    ax.set_xlabel('Volatilidade')
    ax.set_ylabel('Retorno')
    ax.set_title('Fronteira Eficiente - Simulação Monte Carlo')
    ax.legend()
    ax.grid(True, alpha=0.3)
    buf = BytesIO()
    fig.savefig(buf, format="png")
    buf.seek(0)
    img_base64 = base64.b64encode(buf.read()).decode('utf-8')
    plt.close(fig)
    return f'<img src="data:image/png;base64,{img_base64}">', melhores_pesos

# Portfolio Simulation
def simulate_portfolio(acoes_df, initial_investment, melhores_pesos):
    dados_sim = acoes_df.copy()
    if '^BVSP' in dados_sim.columns:
        dados_sim = dados_sim.drop(columns=['^BVSP'])
    datas = dados_sim['Date'].copy()
    precos = dados_sim.drop(columns=['Date']).copy()

    for col in precos.columns:
        precos[col] = pd.to_numeric(precos[col], errors='coerce')
    precos = precos.dropna()
    datas = datas.iloc[:len(precos)].reset_index(drop=True)
    precos = precos.reset_index(drop=True)

    precos_norm = precos.copy()
    for col in precos_norm.columns:
        precos_norm[col] = precos_norm[col] / precos_norm[col].iloc[0]

    portfolio_valores = pd.DataFrame()
    portfolio_valores['Data'] = datas
    nomes_ativos = ['BBAS3.SA', 'ITSA4.SA', 'TAEE11.SA', 'TTEN3.SA', 'BPAC11.SA']

    for i, ativo in enumerate(nomes_ativos):
        if ativo in precos_norm.columns:
            portfolio_valores[ativo] = precos_norm[ativo] * melhores_pesos[i] * initial_investment

    colunas_ativos = [col for col in portfolio_valores.columns if col != 'Data']
    portfolio_valores['Valor_Total'] = portfolio_valores[colunas_ativos].sum(axis=1)
    portfolio_valores['Retorno_Diario'] = 0.0

    for i in range(1, len(portfolio_valores)):
        valor_hoje = portfolio_valores['Valor_Total'].iloc[i]
        valor_ontem = portfolio_valores['Valor_Total'].iloc[i-1]
        if valor_ontem > 0:
            portfolio_valores['Retorno_Diario'].iloc[i] = np.log(valor_hoje / valor_ontem) * 100

    # Portfolio Evolution Plot
    fig_evo = px.line(x=portfolio_valores['Data'], y=portfolio_valores['Valor_Total'],
                      title='Evolução do Patrimônio da Carteira',
                      labels={'x': 'Data', 'y': 'Valor (R$)'})
    fig_evo.add_hline(y=portfolio_valores['Valor_Total'].mean(), line_color="green", line_dash="dot",
                      annotation_text="Valor Médio")
    fig_evo.add_hline(y=initial_investment, line_color="red", line_dash="dot",
                      annotation_text=f"Investimento Inicial (R$ {initial_investment:,.0f})")

    # Daily Returns Plot
    fig_ret = px.line(x=portfolio_valores['Data'], y=portfolio_valores['Retorno_Diario'],
                      title='Retorno Diário do Portfólio',
                      labels={'x': 'Data', 'y': 'Retorno (%)'})
    media_retorno = portfolio_valores['Retorno_Diario'].mean()
    fig_ret.add_hline(y=media_retorno, line_color="red", line_dash="dot",
                      annotation_text=f"Retorno Médio: {media_retorno:.3f}%")

    valor_final = portfolio_valores['Valor_Total'].iloc[-1]
    return fig_evo, fig_ret, portfolio_valores, valor_final

# VaR Analysis
def analyze_var(portfolio_resultado, valor_final):
    retornos_portfolio = portfolio_resultado['Retorno_Diario'].copy()[1:].dropna()

    def calcular_var_historico(retornos, confianca):
        percentil = (1 - confianca) * 100
        return np.percentile(retornos, percentil)

    var_95_historico = calcular_var_historico(retornos_portfolio, 0.95)
    var_99_historico = calcular_var_historico(retornos_portfolio, 0.99)
    var_95_money = valor_final * (np.exp(var_95_historico/100) - 1)
    var_99_money = valor_final * (np.exp(var_99_historico/100) - 1)

    def var_monte_carlo(retornos, num_simulacoes=10000, confianca=0.95):
        np.random.seed(42)
        media = retornos.mean()
        desvio = retornos.std()
        simulacoes = np.random.normal(media, desvio, num_simulacoes)
        percentil = (1 - confianca) * 100
        var_mc = np.percentile(simulacoes, percentil)
        return var_mc, simulacoes

    var_mc_95, simulacoes_95 = var_monte_carlo(retornos_portfolio, confianca=0.95)
    var_mc_99, simulacoes_99 = var_monte_carlo(retornos_portfolio, confianca=0.99)
    var_mc_95_money = valor_final * (np.exp(var_mc_95/100) - 1)
    var_mc_99_money = valor_final * (np.exp(var_mc_99/100) - 1)

    # VaR Plot
    fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(15, 6))
    ax1.hist(retornos_portfolio, bins=50, alpha=0.7, color='skyblue', edgecolor='black', density=True)
    ax1.axvline(var_95_historico, color='orange', linestyle='--', label=f'VaR 95%: {var_95_historico:.2f}%')
    ax1.axvline(var_99_historico, color='red', linestyle='--', label=f'VaR 99%: {var_99_historico:.2f}%')
    ax1.set_xlabel('Retorno Diário (%)')
    ax1.set_ylabel('Densidade')
    ax1.set_title('Distribuição dos Retornos Históricos\ncom VaR Histórico')
    ax1.legend()
    ax1.grid(True, alpha=0.3)

    ax2.hist(simulacoes_95, bins=50, alpha=0.7, color='lightgreen', edgecolor='black', density=True)
    ax2.axvline(var_mc_95, color='orange', linestyle='--', label=f'VaR MC 95%: {var_mc_95:.2f}%')
    ax2.axvline(var_mc_99, color='red', linestyle='--', label=f'VaR MC 99%: {var_mc_99:.2f}%')
    ax2.set_xlabel('Retorno Diário (%)')
    ax2.set_ylabel('Densidade')
    ax2.set_title('Distribuição Simulada (Monte Carlo)\ncom VaR Paramétrico')
    ax2.legend()
    ax2.grid(True, alpha=0.3)
    plt.tight_layout()

    buf = BytesIO()
    fig.savefig(buf, format="png", dpi=300)
    buf.seek(0)
    img_base64 = base64.b64encode(buf.read()).decode('utf-8')
    plt.close(fig)

    # VaR Summary
    var_summary = f"""
    <h3>Resumo do VaR</h3>
    <table border='1'>
        <tr><th>Método</th><th>VaR 95% (1 dia)</th><th>VaR 99% (1 dia)</th></tr>
        <tr><td>Histórico</td><td>{var_95_historico:.2f}% (R$ {abs(var_95_money):,.2f})</td><td>{var_99_historico:.2f}% (R$ {abs(var_99_money):,.2f})</td></tr>
        <tr><td>Monte Carlo</td><td>{var_mc_95:.2f}% (R$ {abs(var_mc_95_money):,.2f})</td><td>{var_mc_99:.2f}% (R$ {abs(var_mc_99_money):,.2f})</td></tr>
    </table>
    """
    return f'<img src="data:image/png;base64,{img_base64}">', var_summary

# Main Analysis Function
def run_analysis(tickers_str, start_date, end_date, num_simulations, investment):
    try:
        tickers = [t.strip() for t in tickers_str.split(',')]
        start_date = datetime.strptime(start_date, '%Y-%m-%d').strftime('%Y-%m-%d')
        end_date = datetime.strptime(end_date, '%Y-%m-%d').strftime('%Y-%m-%d')
        acoes_df = fetch_data(tickers, start_date, end_date)
        if acoes_df.empty:
            return [f"Error: No data fetched for {tickers}"] * 8

        prices_plot = plot_historical_prices(acoes_df)
        returns_plot, stats = calculate_and_plot_returns(acoes_df)
        corr_plot = plot_correlation_matrix(pd.concat([acoes_df['Date'], np.log(acoes_df.drop('Date', axis=1) / acoes_df.drop('Date', axis=1).shift(1))], axis=1).dropna())
        frontier_img, melhores_pesos = simulate_efficient_frontier(acoes_df, int(num_simulations))
        port_evo, port_ret, portfolio_resultado, valor_final = simulate_portfolio(acoes_df, investment, melhores_pesos)
        var_img, var_summary = analyze_var(portfolio_resultado, valor_final)

        return prices_plot, returns_plot, corr_plot, frontier_img, port_evo, port_ret, var_img, stats + var_summary
    except Exception as e:
        return [f"Error: {str(e)}"] * 8

# Gradio Interface
with gr.Blocks(title="Portfolio Analysis Dashboard") as demo:
    gr.Markdown("# Portfolio Theory Visualization Dashboard")
    gr.Markdown("Dynamic visualizations for portfolio analysis. Customize inputs and explore.")

    with gr.Row():
        tickers_input = gr.Textbox(label="Tickers (comma-separated)", value=','.join(default_tickers))
        start_date = gr.Textbox(label="Start Date (YYYY-MM-DD)", value=default_start_date)
        end_date = gr.Textbox(label="End Date (YYYY-MM-DD)", value=default_end_date)
        num_sim = gr.Number(label="Number of Simulations", value=num_ports)
        investment = gr.Number(label="Initial Investment (R$)", value=initial_investment)

    run_btn = gr.Button("Run Analysis")

    with gr.Tab("Historical Prices"):
        prices_output = gr.Plot()

    with gr.Tab("Returns History"):
        returns_output = gr.Plot()

    with gr.Tab("Correlation Matrix"):
        corr_output = gr.Plot()

    with gr.Tab("Efficient Frontier"):
        frontier_output = gr.HTML()

    with gr.Tab("Portfolio Evolution"):
        port_evo_output = gr.Plot()

    with gr.Tab("Daily Returns"):
        port_ret_output = gr.Plot()

    with gr.Tab("VaR Analysis"):
        var_output = gr.HTML()

    with gr.Tab("Statistics"):
        stats_output = gr.HTML()

    run_btn.click(
        run_analysis,
        inputs=[tickers_input, start_date, end_date, num_sim, investment],
        outputs=[prices_output, returns_output, corr_output, frontier_output, port_evo_output, port_ret_output, var_output, stats_output]
    )

if __name__ == "__main__":
    demo.launch()