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." )