src/streamlit_app.py

import numpy as np
import pandas as pd
import streamlit as st
import datetime as dt
from typing import List, Tuple

# Dependências opcionais (se instaladas) para gráficos e dados
try:
    import yfinance as yf
    YF_AVAILABLE = True
except Exception:
    YF_AVAILABLE = False

try:
    import altair as alt
    ALTAIR_AVAILABLE = True
except Exception:
    ALTAIR_AVAILABLE = False


# Utilidades
def carregar_precos(
    tickers: List[str],
    start: str,
    end: str,
    progress: bool = False
) -> pd.DataFrame:
    """
    Baixa preços via yfinance. Prioriza 'Adj Close'; se não houver, usa 'Close'.
    Retorna DataFrame com colunas = tickers e índice = datas.
    """
    if not YF_AVAILABLE:
        raise ImportError(
            "yfinance não está disponível neste ambiente. "
            "Instale com: pip install yfinance"
        )

    df = yf.download(
        tickers,
        start=start,
        end=end,
        progress=progress,
        auto_adjust=False,
        group_by='ticker'
    )

    # Normalizar para um formato (colunas simples por ticker)
    if isinstance(df.columns, pd.MultiIndex):
        # tentar pegar Adj Close > Close
        if ("Adj Close" in df.columns.get_level_values(1)):
            prices = df.xs("Adj Close", axis=1, level=1, drop_level=True)
        elif ("Close" in df.columns.get_level_values(1)):
            prices = df.xs("Close", axis=1, level=1, drop_level=True)
        else:
            raise KeyError("Nem 'Adj Close' nem 'Close' encontrados no retorno do yfinance.")
    else:
        # dataset simples (um único ticker)
        if "Adj Close" in df.columns:
            prices = df[["Adj Close"]].rename(columns={"Adj Close": tickers[0]})
        elif "Close" in df.columns:
            prices = df[["Close"]].rename(columns={"Close": tickers[0]})
        else:
            raise KeyError("Nem 'Adj Close' nem 'Close' encontrados no retorno do yfinance.")

    # manter apenas os tickers solicitados (quando baixar lista)
    missing = [t for t in tickers if t not in prices.columns]
    if missing:
        # Alguns tickers podem vir faltando; vamos remover os inexistentes
        st.warning(f"Atenção: sem dados para: {', '.join(missing)}. Eles serão ignorados.")
        tickers = [t for t in tickers if t in prices.columns]
        prices = prices[tickers]

    prices = prices.dropna(how="all")
    prices = prices.dropna(axis=0)  # remove datas com NaN
    return prices


def log_retorno(precos: pd.DataFrame) -> pd.DataFrame:
    """Retorno logarítmico diário."""
    return np.log(precos / precos.shift(1)).dropna()


def max_sharpe_random(mu: pd.Series, cov: pd.DataFrame, n_sims: int = 50_000, rf: float = 0.0, seed: int = 42) -> np.ndarray:
    """
    Busca pesos de Máximo Sharpe por amostragem aleatória (Dirichlet).
    mu: média dos retornos (por período)
    cov: matriz de covariância (por período)
    rf: taxa livre de risco no mesmo período (ex.: diária)
    Retorna vetor de pesos ótimos.
    """
    rng = np.random.default_rng(seed)
    k = len(mu)
    W = rng.dirichlet(np.ones(k), size=n_sims)
    # retorno e risco da carteira
    ret = W @ mu.values
    # risco: sqrt(w' Σ w) eficiente com einsum
    risco = np.sqrt(np.einsum('ij,jk,ik->i', W, cov.values, W))
    # Sharpe
    excesso = ret - rf
    sharpe = np.divide(excesso, risco, out=np.full_like(excesso, -np.inf), where=risco > 0)
    idx = int(np.nanargmax(sharpe))
    return W[idx]


def simular_mc_multivariado(
    mu: pd.Series,
    cov: pd.DataFrame,
    pesos: np.ndarray,
    n_sim: int = 50_000,
    horizonte_dias: int = 1,
    seed: int = 42
) -> np.ndarray:
    """
    Monte Carlo multivariado (Normal) para retornos da carteira.
    - Gera amostras multivariadas ~ N(mu, cov)
    - Para horizonte > 1, soma retornos log (aprox. soma normal) ou compõe geometricamente.
    Retorna array de retornos (log) simulados da carteira no horizonte.
    """
    rng = np.random.default_rng(seed)
    k = len(mu)
    # Cholesky para gerar multivariada
    L = np.linalg.cholesky(cov.values)

    # Se horizonte > 1, simulamos soma de passos (log-returns ~ soma)
    # Gera (n_sim, horizonte, k)
    Z = rng.standard_normal(size=(n_sim, horizonte_dias, k))
    # passo diário: mu + L @ z
    # broadcasting: (n_sim, horizonte, k) -> (n_sim, horizonte, k)
    passos = mu.values + (Z @ L.T)
    # soma ao longo do horizonte (log-returns)
    ret_log_multi = passos.sum(axis=1)  # (n_sim, k)

    # retorno log da carteira = soma_i (w_i * ret_i) em log-approx -> carteira em log via combinação linear
    # (n_sim, k) @ (k,) -> (n_sim,)
    ret_log_port = ret_log_multi @ pesos
    return ret_log_port


def var_percentual(simulacoes_ret_log: np.ndarray, alpha: float = 0.95) -> float:
    """
    VaR (como perda positiva) a partir da distribuição de retornos (log).
    Retorna percentual (ex.: 0.042 = 4,2%).
    Por convenção: VaR_α = - quantil_{1-α}(retorno)  => perda mínima no pior (1-α)
    """
    # converter log-return -> return simples aproximado (e^r - 1)
    ret_simple = np.expm1(simulacoes_ret_log)
    # quantil no tail esquerdo (ex.: 5% para 95%)
    q = np.quantile(ret_simple, 1 - alpha)
    return -q  # perda positiva


def format_percent(x: float) -> str:
    return f"{x*100:.2f}%"


# Interface Streamlit
st.set_page_config(page_title="VaR Monte Carlo - Carteira (B3)", page_icon="=>", layout="wide")

st.title("VaR da Carteira via Simulação de Monte Carlo")
st.caption( 
    "Baseado na carteira de ações (ITUB4.SA, MGLU3.SA, COGN3.SA, PETR4.SA, ABEV3.SA)."
)

with st.expander("Funcionamento deste app"):
    st.write("""
- Download dos preços diários dos tickers selecionados (por padrão, cinco ativos).
- Calculo dos retornos logarítmicos diários, suas médias e covariâncias.
- Opcionalmente se busca pesos de **Máximo Sharpe** (amostragem) ou você define pesos manualmente.
- Roda-se uma simulação **Monte Carlo multivariada Normal** dos retornos no horizonte escolhido.
- Calculo do **VaR** (percentual e em R$) para os níveis de confiança (ex.: 95% e 99%).
""")

# Parâmetros
colA, colB = st.columns([2, 1])

default_tickers = ["ITUB4.SA", "MGLU3.SA", "COGN3.SA", "PETR4.SA", "ABEV3.SA"]  
with colA:
    tickers = st.multiselect(
        "Tickers (B3, sufixo .SA)",
        options=default_tickers,
        default=default_tickers,
        help="Por padrão, os 5 ativos."
    )

with colB:
    hoje = dt.date.today()
    start = st.date_input("Início", value=dt.date(2015, 1, 1))
    end = st.date_input("Fim", value=hoje)

col1, col2, col3, col4 = st.columns([1, 1, 1, 1])

with col1:
    aporte = st.number_input("Investimento inicial (R$)", min_value=1000.0, value=50_000.0, step=1000.0, format="%.2f")

with col2:
    horizonte = st.number_input("Horizonte (dias úteis)", min_value=1, value=1, step=1)

with col3:
    n_sim = st.number_input("Nº simulações Monte Carlo", min_value=1000, value=50_000, step=1000, format="%d")

with col4:
    seed = st.number_input("Seed aleatória", min_value=0, value=42, step=1, format="%d")

col5, col6, col7 = st.columns([1, 1, 1])
with col5:
    nivel_1 = st.slider("Confiança 1 (VaR α)", min_value=0.80, max_value=0.999, value=0.95, step=0.01)
with col6:
    nivel_2 = st.slider("Confiança 2 (opcional)", min_value=0.80, max_value=0.999, value=0.99, step=0.01)
with col7:
    usar_max_sharpe = st.toggle("Usar pesos de Máx. Sharpe", value=True, help="Se desligado, você define os pesos abaixo.")

st.divider()

# ------------------------
# Dados
# ------------------------
if not tickers:
    st.error("Escolha pelo menos um ticker.")
    st.stop()

try:
    precos = carregar_precos([t.strip() for t in tickers], start.isoformat(), end.isoformat(), progress=False)
except Exception as e:
    st.error(f"Falha ao carregar dados: {e}")
    st.stop()

ret = log_retorno(precos)
mu = ret.mean()                 # média diária
cov = ret.cov()                 # covariância diária
cols = list(ret.columns)

st.subheader("Série de preços (últimas 10 linhas)")
st.dataframe(precos.tail(10))

st.subheader("Estatísticas diárias dos retornos")
stats_df = pd.DataFrame({
    "média": mu,
    "desvio": ret.std(),
    "assimetria": ret.skew(),
    "curtose (excesso)": ret.kurt()
})
st.dataframe(stats_df.style.format({"média": "{:.5f}", "desvio": "{:.5f}", "assimetria": "{:.2f}", "curtose (excesso)": "{:.2f}"}))

# ------------------------
# Pesos
# ------------------------
if usar_max_sharpe:
    pesos = max_sharpe_random(mu, cov, n_sims=50_000, rf=0.0, seed=seed)
    fonte_pesos = "Máximo Sharpe (amostragem aleatória)"
else:
    st.markdown("#### Pesos manuais (devem somar 100%)")
    pesos_inputs = []
    colpesos = st.columns(len(cols))
    for i, c in enumerate(cols):
        with colpesos[i]:
            w = st.number_input(f"{c}", min_value=0.0, max_value=100.0, value=100.0/len(cols), step=1.0, format="%.2f")
            pesos_inputs.append(w)
    soma = sum(pesos_inputs)
    if soma <= 0:
        st.error("A soma dos pesos deve ser > 0.")
        st.stop()
    pesos = np.array(pesos_inputs) / soma
    fonte_pesos = "Definidos manualmente pelo usuário"

pesos_df = pd.DataFrame({"Ativo": cols, "Peso": pesos})
st.subheader("Pesos da carteira")
st.dataframe(pesos_df.style.format({"Peso": "{:.2%}"}))
st.caption(f"Origem dos pesos: **{fonte_pesos}**.")

# Simulação Monte Carlo
rng = np.random.default_rng(seed)
sim_ret_log = simular_mc_multivariado(
    mu=mu,
    cov=cov,
    pesos=pesos,
    n_sim=n_sim,
    horizonte_dias=horizonte,
    seed=seed
)

# VaR em percentual e em moeda
var1_pct = var_percentual(sim_ret_log, alpha=nivel_1)
var2_pct = var_percentual(sim_ret_log, alpha=nivel_2)

# patrimônio atual simulado (a partir do aporte e do retorno da carteira no horizonte)
# retorno simples da carteira (e^r - 1)
sim_ret_simple = np.expm1(sim_ret_log)
valor_futuro = aporte * (1.0 + sim_ret_simple)
# perda em R$ no quantil
var1_rs = aporte * var1_pct
var2_rs = aporte * var2_pct

colM1, colM2, colM3 = st.columns(3)
with colM1:
    st.metric(f"VaR {int(nivel_1*100)}% (percentual)", format_percent(var1_pct))
with colM2:
    st.metric(f"VaR {int(nivel_1*100)}% (R$)", f"R$ {var1_rs:,.2f}".replace(",", "X").replace(".", ",").replace("X", "."))
with colM3:
    st.metric(f"VaR {int(nivel_2*100)}% (R$)", f"R$ {var2_rs:,.2f}".replace(",", "X").replace(".", ",").replace("X", "."))

st.caption(
    f"Interpretação: com {int(nivel_1*100)}% de confiança, a **perda mínima** esperada em {horizonte} dia(s) é de **{format_percent(var1_pct)}** (≈ R$ {var1_rs:,.2f}). "
    "Ou seja, em ~1 a cada {round(1/(1-nivel_1))} dia(s), a perda pode ser **pior** do que esse valor."
)

# ------------------------
# Visualização da distribuição
# ------------------------
st.subheader("Distribuição das simulações de retorno da carteira")

hist_df = pd.DataFrame({
    "retorno_simples": sim_ret_simple
})
hist_df["retorno_%"] = 100 * hist_df["retorno_simples"]

if ALTAIR_AVAILABLE:
    base = alt.Chart(hist_df).mark_bar().encode(
        alt.X("retorno_%:Q", bin=alt.Bin(maxbins=60), title="Retorno (%)"),
        alt.Y("count():Q", title="Frequência")
    ).properties(height=320)

    rule1 = alt.Chart(pd.DataFrame({"x": [(-var1_pct)*100]})).mark_rule(color="red", strokeDash=[6,3]).encode(x="x:Q")
    rule2 = alt.Chart(pd.DataFrame({"x": [(-var2_pct)*100]})).mark_rule(color="orange", strokeDash=[6,3]).encode(x="x:Q")

    st.altair_chart(base + rule1 + rule2, use_container_width=True)
else:
    st.line_chart(hist_df["retorno_%"].value_counts().sort_index())

# ------------------------
# Série acumulada (cenários)
# ------------------------
st.subheader("Cenários simulados (valor do aporte após o horizonte)")
cen_df = pd.DataFrame({"Valor final (R$)": valor_futuro})
st.dataframe(cen_df.describe().T.style.format({"mean": "R$ {:,.2f}".format}).format_index(lambda s: s.replace(",", "X").replace(".", ",").replace("X", ".")))

st.success("Simulação concluída!")

st.divider()
st.caption(
    "Carteira base (ITUB4.SA, MGLU3.SA, COGN3.SA, PETR4.SA, ABEV3.SA)  — "
    "parametrizável acima. O método de Monte Carlo segue a abordagem Normal multivariada com média e covariância empíricas diárias."
)