# ============================================================

# MULTI-STRATEGY + MULTI-BENCHMARK INTERACTIVE REPORT (PLOTLY)

# + Portfolio Optimization (Strategies only)
# - baseline: Risk Parity (robust)
# - aggressive: Markowitz (Max Sharpe)
# - estimation uses LOG-returns + covariance shrinkage + weight cap
#
# FIXES:
# 1) Extend each strategy equity to "today" by forward-filling last equity
# 2) Portfolio uses DAILY arithmetic returns (0 on no-trade days) and continues
until today
# 3) Optimization uses a common date range (max start among strategies -> report
end)
# 4) Max DD table consistent with plotted DD (same extended daily equity)
# 5) Timezone helper fixed (no tz_localize on tz-aware Timestamp)
#
# CHANGES (requested):
# A) Risk Parity optimizer replaced by robust RC-based + multi-start
# B) Added debug: optimizer success/message + risk contributions + vols
# C) Fixed tail section: define returns_df before extra correlation analyses
# ============================================================

from pathlib import Path

from typing import Optional, Dict, List, Tuple
import sys, subprocess
import numpy as np
import pandas as pd

# --- SciPy
from [Link] import skew, kurtosis
from [Link] import minimize

# ============================================================
# Install helper (works in script + notebook)
# ============================================================
def _pip_install(pkg: str):
subprocess.check_call([[Link], "-m", "pip", "-q", "install", "--
upgrade", pkg])

# --- Plotly
try:
import plotly # noqa: F401
import plotly.graph_objects as go
import [Link] as pio
except Exception:
_pip_install("plotly")
import plotly.graph_objects as go
import [Link] as pio

# --- yfinance
try:
import yfinance as yf
except Exception:
_pip_install("yfinance")
import yfinance as yf

# --- openpyxl (para ler .xlsx)

try:
import openpyxl # noqa: F401
except Exception:
_pip_install("openpyxl")
import openpyxl # noqa: F401

from [Link] import display

# Renderer (Colab/Jupyter)
try:
[Link] = "colab"
except Exception:
pass

# ============================================================
# CONFIG (EDITE AQUI)
# ============================================================

STRATEGIES = [
{
"name": "Strategy 1",
"path": Path("Path 1"),
"start_balance": None,
},
{
"name": "Strategy 2",
"path": Path("Path 2"),
"start_balance": None,
},
{
"name": "Strategy N",
"path": Path("Path N"),
"start_balance": None,
},
]

BENCHMARKS = [
{"name": "ETH Buy&Hold", "ticker": "ETH-USD"},
{"name": "BTC Buy&Hold", "ticker": "BTC-USD"},
]

# Normalização (comparação justa)

NORMALIZE_ALL_TO_BASE = True
BASE_VALUE = 100.0

RF_ANNUAL = 0.00
ANN_DAYS_MAIN = 365.25 # cripto
ANN_DAYS_ALT = 252.0

# esticar a equity até hoje (para plots, métricas e portfolio)

EXTEND_EQUITY_TO_TODAY_FOR_PLOTS = True
LOCAL_TZ = "America/Sao_Paulo"

# PORTFOLIO
PORTFOLIO_ENABLE = True
PORTFOLIO_NAME = "Portfolio"
PORTFOLIO_LONG_ONLY = True
PORTFOLIO_MAX_WEIGHT = 0.40
PORTFOLIO_REG_EPS = 1e-10

# ============================================================
# PDF EXPORT CONFIG (NOVO)
# ============================================================
SAVE_PDF = True # True para gerar PDF
PDF_OUT_PATH = Path("./report_full.pdf")

PDF_INCLUDE_PER_SERIES = True
PDF_INCLUDE_SIMULATIONS = True
PDF_TABLE_MAX_ROWS = 35

# Otimizador usado para CONSTRUIR o portfolio:

# - "risk_parity" = baseline robusto
# - "markowitz" = agressivo (max Sharpe)
PORTFOLIO_OPT_MODE = "risk_parity"

# Shrinkage (0.10–0.30 costuma ser bom)

PORTFOLIO_SHRINKAGE = 0.15

# Mostrar Markowitz (mesmo se usar risk parity para construir)

PORTFOLIO_SHOW_AGGRESSIVE_MARKOWITZ = True

# Simulações (somente estratégias + portfolio)

N_MC = 1500
N_BOOT = 1500
PERTURB_SIGMA = 0.35
SLIPPAGE_BPS = 5
rng = [Link].default_rng(42)

# Tema (Aqua/Purple)
AQUA = "#20D3D8"
PURPLE = "#6C4BFF"
DARK = "#111827"

# Top-3 DD: mesma paleta do heatmap

RANK_COLORS = {1: PURPLE, 2: DARK, 3: AQUA}

SAVE_HTML = False
HTML_OUT_DIR = Path("./report_interactive_html")
HTML_OUT_DIR.mkdir(parents=True, exist_ok=True)

# ============================================================
# HELPERS: cores
# ============================================================

def _hex_to_rgb(h: str) -> Tuple[int, int, int]:

h = [Link]("#")
return tuple(int(h[i:i+2], 16) for i in (0, 2, 4))

def _rgb_to_hex(rgb: Tuple[int, int, int]) -> str:

return "#{:02x}{:02x}{:02x}".format(*rgb)

def gradient_colors(n: int, c1: str = PURPLE, c2: str = AQUA) -> List[str]:
if n <= 1:
return [c1]
r1, g1, b1 = _hex_to_rgb(c1)
r2, g2, b2 = _hex_to_rgb(c2)
out = []
for i in range(n):
t = i / (n - 1)
r = int(round(r1 + (r2 - r1) * t))
 g = int(round(g1 + (g2 - g1) * t))
b = int(round(b1 + (b2 - b1) * t))
[Link](_rgb_to_hex((r, g, b)))
return out

def hex_to_rgba(hex_color: str, alpha: float) -> str:

r, g, b = _hex_to_rgb(hex_color)
a = float(alpha)
a = 0.0 if a < 0 else (1.0 if a > 1 else a)
return f"rgba({r},{g},{b},{a})"

# ============================================================
# DATE / TIME HELPERS
# ============================================================

def local_today_date(tz_str: str = "America/Sao_Paulo") -> [Link]:

"""
Return today's date (00:00) in local timezone, as tz-naive pandas Timestamp.
Fixed to avoid tz_localize() on tz-aware timestamps.
"""
now_utc = [Link](tz="UTC")
try:
now_local = now_utc.tz_convert(tz_str)
except Exception:
now_local = now_utc
return now_local.normalize().tz_localize(None)

# ============================================================
# UTIL – robustez de colunas / leitura
# ============================================================

def ensure_series(x):
if isinstance(x, [Link]):
return x
if isinstance(x, [Link]):
if [Link][1] == 1:
return [Link][:, 0]
raise ValueError(f"Esperava DataFrame 1 coluna, recebi {[Link][1]}")
return [Link]([Link](x).ravel())

def smart_col(df: [Link], include=None, exclude=None) -> Optional[str]:

include = include or []
exclude = exclude or []
cols = list([Link])
low = {c: str(c).strip().lower() for c in cols}

def ok(c):
s = low[c]
return all(p in s for p in include) and not any(p in s for p in exclude)

hits = [c for c in cols if ok(c)]

return hits[0] if hits else None

def guess_trade_sheet(xls: [Link]) -> str:

for name in xls.sheet_names:
low = [Link]()
if "list" in low and "trade" in low:
return name
return xls.sheet_names[0]
def read_initial_capital(xls: [Link]) -> Optional[float]:
try:
if "Performance" not in xls.sheet_names:
return None
perf = pd.read_excel(xls, sheet_name="Performance")
label_col = [Link][0]
all_usdt_col = next((c for c in [Link] if str(c).strip().lower() ==
"all usdt"), None) or [Link][1]
row =
[Link][perf[label_col].astype(str).[Link]().[Link]().eq("initial capital")]
if not [Link]:
v = [Link][0][all_usdt_col]
v = pd.to_numeric(str(v).replace(",", "."), errors="coerce")
return float(v) if [Link](v) else None
except Exception:
return None
return None

def normalize_percent(col: [Link], reference_decimal: Optional[[Link]] =

None) -> [Link]:
s = ([Link](str)
.[Link]("%", "", regex=False)
.[Link]()
.[Link](",", ".", regex=False)
.[Link]("−", "-", regex=False)
.[Link](r"[^\d\.\-]", "", regex=True))
n = pd.to_numeric(s, errors="coerce").astype(float)

if reference_decimal is not None:

ref = pd.to_numeric(reference_decimal, errors="coerce").astype(float)
mask = [Link]() & [Link]()
if [Link]():
err1 = [Link]([Link](n[mask] - ref[mask]))
err2 = [Link]([Link]((n[mask] / 100.0) - ref[mask]))
if [Link](err2) and (not [Link](err1) or err2 < err1):
return n / 100.0
return n

if [Link]().abs().max() > 1.0:

n = n / 100.0
return n

def extract_exit_trades(df: [Link], initial_capital: Optional[float]) ->

[Link]:
cols = {str(c).strip().lower(): c for c in [Link]}

type_col = [Link]("type") or smart_col(df, include=["type"])

dt_col = [Link]("date/time") or smart_col(df, include=["date", "time"]) or
smart_col(df, include=["date"])
cum_pct_col = [Link]("cumulative p&l %") or smart_col(df,
include=["cumulative", "%"])
cum_usdt_col = [Link]("cumulative p&l usdt") or smart_col(df,
include=["cumulative", "usdt"])

if not type_col or not dt_col:

raise KeyError("Não encontrei colunas Type e Date/Time no arquivo.")

exits = [Link][df[type_col].astype(str).[Link]().[Link]("exit",
na=False)].copy()
[Link](columns={dt_col: "Date/Time"}, inplace=True)
exits["Date/Time"] = pd.to_datetime(exits["Date/Time"], errors="coerce")

if cum_usdt_col:
exits["CumPnlUSDT"] = pd.to_numeric(exits[cum_usdt_col], errors="coerce")

ref_cum = None
if ("CumPnlUSDT" in [Link]) and initial_capital:
ref_cum = exits["CumPnlUSDT"] / float(initial_capital)

if cum_pct_col:
exits["CumRet"] = normalize_percent(exits[cum_pct_col],
reference_decimal=ref_cum)

if "CumRet" not in [Link] and ("CumPnlUSDT" in [Link]) and

initial_capital:
exits["CumRet"] = exits["CumPnlUSDT"] / float(initial_capital)

if "CumRet" not in [Link] and "CumPnlUSDT" not in [Link]:

raise KeyError("Não achei 'Cumulative P&L %' nem 'Cumulative P&L USDT' para
reconstruir a equity.")

keep = ["Date/Time"]
if "CumRet" in [Link]:
[Link]("CumRet")
if "CumPnlUSDT" in [Link]:
[Link]("CumPnlUSDT")

exits = exits[keep].dropna(subset=["Date/Time"])
exits = exits.sort_values("Date/Time", ignore_index=True)
return exits

# ============================================================
# DRAWDOWN
# ============================================================

def drawdown_series(equity: [Link]) -> [Link]:

eq = ensure_series(equity).dropna()
return eq / [Link]() - 1.0

def drawdown_days_series(equity_daily: [Link]) -> [Link]:

eqd = ensure_series(equity_daily).dropna()
peaks = [Link]()
in_dd = eqd < peaks
dur, c = [], 0
for flag in in_dd.values:
c = c + 1 if flag else 0
[Link](c)
return [Link](dur, index=[Link])

# ============================================================
# METRICS
# ============================================================

def sharpe_sortino_from_returns(ret: [Link], rf_annual=0.0, ann_days=365.25) ->

Tuple[float, float, float]:
ret = ensure_series(ret).dropna()
if len(ret) < 2:
 return [Link], [Link], [Link]

rf_step = (1.0 + rf_annual) ** (1.0 / ann_days) - 1.0

mu = float([Link]())
sd = float([Link](ddof=1))

sharpe = ((mu - rf_step) / sd) * [Link](ann_days) if ([Link](sd) and sd >

0) else [Link]

neg = ret[ret < 0]

dsd = float([Link](ddof=1)) if len(neg) >= 2 else [Link]
sortino = ((mu - rf_step) / dsd) * [Link](ann_days) if ([Link](dsd) and
dsd > 0) else [Link]

vol_ann = sd * [Link](ann_days) if [Link](sd) else [Link]

return sharpe, sortino, vol_ann

def sharpe_event_time_from_equity(equity_ts: [Link], rf_annual=0.0):

eq = ensure_series(equity_ts).dropna().sort_index()
if len(eq) < 3:
return [Link], [Link], [Link]

logret = [Link](eq / [Link](1)).dropna()

dt_years = ([Link].to_series().diff().dt.total_seconds().dropna()
/ (365.25 * 24 * 3600))
logret = [Link][dt_years.index]
T = float(dt_years.sum())
if not [Link](T) or T <= 0:
return [Link], [Link], [Link]

mu_ann = float([Link]() / T)
rf_log = float([Link](1.0 + rf_annual)) if rf_annual != 0 else 0.0

resid = logret - (mu_ann * dt_years)

vol_ann = float([Link]((resid**2).sum() / T))

sharpe = (mu_ann - rf_log) / vol_ann if ([Link](vol_ann) and vol_ann > 0)

else [Link]
return sharpe, mu_ann, vol_ann

def metrics_from_daily_equity(equity_daily: [Link], start_balance: float,

rf_annual=0.0) -> Dict[str, float]:
eq = ensure_series(equity_daily).dropna()
if [Link]:
return {}

start_dt, end_dt = [Link][0], [Link][-1]

years = max((end_dt - start_dt).days, 1) / 365.25

final_bal = float([Link][-1])
cagr = (final_bal / float(start_balance)) ** (1.0 / years) - 1.0

ret_d = eq.pct_change().dropna()

sh_365, so_365, vol_365 = sharpe_sortino_from_returns(ret_d, rf_annual,

ANN_DAYS_MAIN)
sh_252, so_252, vol_252 = sharpe_sortino_from_returns(ret_d, rf_annual,
ANN_DAYS_ALT)
 dd = eq / [Link]() - 1.0
max_dd = float([Link]()) if len(dd) else 0.0
mar = (cagr / abs(max_dd)) if ([Link](max_dd) and max_dd != 0) else [Link]

skewness = float(skew(ret_d, bias=False)) if len(ret_d) > 2 else [Link]

kurt_excess = float(kurtosis(ret_d, fisher=True, bias=False)) if len(ret_d) > 3
else [Link]

return {
"Start Balance": float(start_balance),
"Final Balance": final_bal,
"Net Profit": final_bal - float(start_balance),
"Net % Gain": final_bal / float(start_balance) - 1.0,
"CAGR": cagr,

"Sharpe (365)": sh_365,

"Sortino (365)": so_365,
"Volatility (365)": vol_365,

"Sharpe (252)": sh_252,

"Sortino (252)": so_252,
"Volatility (252)": vol_252,

"Max Drawdown": max_dd,

"Max Drawdown %": max_dd * 100.0,
"MAR (Calmar)": mar,

"Skewness": skewness,
"Kurtosis (excess)": kurt_excess,
"Days": int(len(ret_d)),
}

# ============================================================
# HEATMAP MENSAL (compound)
# ============================================================

def monthly_returns_compound(equity_daily: [Link]) -> [Link]:

rd = ensure_series(equity_daily).pct_change().dropna()
df = [Link]({"date": [Link], "ret": [Link]})
df["ym"] = df["date"].dt.to_period("M")
monthly = [Link]("ym")["ret"].apply(lambda x: (1.0 + x).prod() -
1.0).to_timestamp()

m = monthly.to_frame("ret")
m["Year"] = [Link]
m["Month"] = [Link]
heat = [Link](index="Year", columns="Month", values="ret").fillna(0.0)

for mm in range(1, 13):

if mm not in [Link]:
heat[mm] = 0.0
return heat[sorted([Link])]

def heatmap_figure(heat: [Link], title: str):

z = ([Link] * 100.0)
text = [Link](lambda v: f"{v:+.1f}%")(z)

fig = [Link](
data=[Link](
 z=z,
x=[str(m) for m in [Link]],
y=[str(y) for y in [Link]],
colorscale=[[0.0, PURPLE],[0.5, DARK],[1.0, AQUA]],
zmid=0.0,
text=text,
texttemplate="%{text}",
hovertemplate="Year=%{y}<br>Month=%{x}<br>Return=%{z:.2f}%<extra></
extra>"
)
)
fig.update_layout(
title=title,
xaxis_title="Month",
yaxis_title="Year",
template="plotly_white",
height=420,
)
return fig

# ============================================================
# BENCHMARK FETCH (yfinance)
# ============================================================

def fetch_close(ticker: str, start_date: [Link], end_date: [Link]) ->

Optional[[Link]]:
try:
data = [Link](
ticker,
start=start_date,
end=end_date + [Link](days=1),
progress=False,
auto_adjust=False
)
if data is None or [Link] or "Close" not in data:
return None
close = data["Close"].dropna()
if close is None or len(close) == 0:
return None
close = [Link]("D").ffill()
return close
except Exception:
return None

def build_benchmark_equity(ticker: str, start_date: [Link], end_date:

[Link], start_balance: float) -> Optional[[Link]]:
close = fetch_close(ticker, start_date, end_date)
if close is None or len(close) == 0:
return None
ret = close.pct_change().dropna()
eq = (1.0 + ret).cumprod() * float(start_balance)
[Link] = [Link]
return eq

# ============================================================
# SIMULAÇÕES
# ============================================================

def mc_perturbed_trades(returns: [Link], n_sims: int, start: float,

sigma_scale=0.35, slippage_bps=5):
returns = [Link](returns, dtype=float)
if len(returns) == 0:
return [Link]([]), [Link]([])
finals, maxdds = [], []
sd = [Link](returns) if len(returns) else 0.0
slip = slippage_bps / 10000.0
for _ in range(n_sims):
noise = [Link](0, sigma_scale * sd, size=len(returns)) if sd > 0 else
0.0
tr = (returns + noise - slip).clip(-0.999, None)
eq = start * [Link](1.0 + tr)
s = [Link](eq)
[Link]([Link][-1])
[Link]((s / [Link]() - 1.0).min())
return [Link](finals), [Link](maxdds)

def bootstrap_resample(returns: [Link], n_sims: int, start: float):

returns = [Link](returns, dtype=float)
if len(returns) == 0:
return [Link]([]), [Link]([])
finals, maxdds = [], []
for _ in range(n_sims):
idx = [Link](0, len(returns), size=len(returns))
tr = returns[idx]
eq = start * [Link](1.0 + tr)
s = [Link](eq)
[Link]([Link][-1])
[Link]((s / [Link]() - 1.0).min())
return [Link](finals), [Link](maxdds)

def hist_fig(data: [Link], title: str, color: str, x_title: str):

data = [Link](data, dtype=float)
fig = [Link]()
fig.add_trace([Link](
x=data,
nbinsx=50,
marker_color=color,
opacity=0.85,
name=title
))
fig.update_layout(
title=title,
xaxis_title=x_title,
yaxis_title="Frequency",
template="plotly_white",
height=420,
bargap=0.05
)
return fig

def overlay_hist_fig(a: [Link], b: [Link], title: str, name_a: str, name_b:

str, color_a: str, color_b: str, x_title: str):
a = [Link](a, dtype=float)
b = [Link](b, dtype=float)
fig = [Link]()
fig.add_trace([Link](x=a, nbinsx=50, marker_color=color_a, opacity=0.65,
name=name_a))
fig.add_trace([Link](x=b, nbinsx=50, marker_color=color_b, opacity=0.65,
name=name_b))
fig.update_layout(
title=title,
xaxis_title=x_title,
yaxis_title="Frequency",
barmode="overlay",
template="plotly_white",
height=420,
bargap=0.05
)
return fig

# ============================================================
# BUILD STRATEGY FROM EXCEL
# ============================================================

def build_strategy_from_excel(file_path: Path, start_balance_override:

Optional[float] = None) -> dict:
if not file_path.exists():
raise FileNotFoundError(f"Arquivo não encontrado: {file_path}")

xls = [Link](file_path)
initial_cap_file = read_initial_capital(xls)

start_balance = (
start_balance_override if start_balance_override is not None
else (initial_cap_file if initial_cap_file is not None else 100.0)
)

sheet = guess_trade_sheet(xls)
raw = pd.read_excel(xls, sheet_name=sheet)
exits = extract_exit_trades(raw, initial_cap_file)

if ("CumPnlUSDT" in [Link]) and (initial_cap_file is not None) and

[Link](initial_cap_file):
scale = float(start_balance) / float(initial_cap_file)
equity_exit = float(start_balance) + exits["CumPnlUSDT"].astype(float) *
scale
equity_exit = [Link](equity_exit.values,
index=exits["Date/Time"]).sort_index().groupby(level=0).last()
else:
if "CumRet" not in [Link]:
raise KeyError("Sem CumPnlUSDT e sem CumRet: impossível reconstruir
equity.")
equity_exit = float(start_balance) * (1.0 + exits["CumRet"].astype(float))
equity_exit = [Link](equity_exit.values,
index=exits["Date/Time"]).sort_index().groupby(level=0).last()

# retornos por trade (event-time) para simulação das estratégias

trade_ret = equity_exit.pct_change()
if len(equity_exit) > 0:
trade_ret.iloc[0] = (equity_exit.iloc[0] / float(start_balance)) - 1.0
trade_ret_arr = trade_ret.dropna().values

equity_daily = equity_exit.resample("D").ffill()

metrics = metrics_from_daily_equity(equity_daily, float(start_balance),

rf_annual=RF_ANNUAL)
 try:
dd_event = float(drawdown_series(equity_exit).min())
except Exception:
dd_event = [Link]
metrics["Max Drawdown (event/exits)"] = dd_event
metrics["Max Drawdown (event/exits) %"] = dd_event * 100.0 if
[Link](dd_event) else [Link]

sh_ev, mu_ev, vol_ev = sharpe_event_time_from_equity(equity_exit,

rf_annual=RF_ANNUAL)
metrics["Sharpe (event-time)"] = sh_ev
metrics["Return rate (event-time, log)"] = mu_ev
metrics["Volatility (event-time, log)"] = vol_ev
metrics["Trades (EXIT)"] = int(len(equity_exit))

return dict(
sheet=sheet,
start_balance=float(start_balance),
initial_cap_file=initial_cap_file,
exits=exits,
equity_exit=equity_exit,
equity_daily=equity_daily,
trade_ret=trade_ret_arr,
metrics=metrics,
start_date=pd.to_datetime(equity_daily.[Link]()).normalize(),
end_date=pd.to_datetime(equity_daily.[Link]()).normalize(),
)

# ============================================================
# NORMALIZAÇÃO + EXTENSÃO
# ============================================================

def rebase_to(series: [Link], base: float) -> [Link]:

s = ensure_series(series).dropna()
if [Link]:
return s
first = float([Link][0])
if not [Link](first) or first == 0:
return s
return s * (base / first)

def extend_daily_to(series: [Link], end_date: [Link]) -> [Link]:

"""
Extend a (daily or timestamp) equity series to end_date, carrying last value
forward.
"""
s = ensure_series(series).dropna()
if [Link]:
return s

s = s.sort_index()

# força daily
try:
s = [Link]("D").ffill()
except Exception:
[Link] = pd.to_datetime([Link]).normalize()
s = [Link](level=0).last().asfreq("D").ffill()
 end_date = pd.to_datetime(end_date).normalize()
start_date = pd.to_datetime([Link]()).normalize()

full_idx = pd.date_range(start_date, end_date, freq="D")

s = [Link](full_idx, method="ffill")
return s

# ============================================================
# FIGURES: equity/drawdown/days-in-dd
# ============================================================

def equity_figure(series_map: Dict[str, [Link]], colors_map: Dict[str, str],

title: str):
fig = [Link]()
for name, s in series_map.items():
s = ensure_series(s).dropna()
if [Link]:
continue
fig.add_trace([Link](
x=[Link], y=[Link],
mode="lines",
name=name,
line=dict(color=colors_map.get(name, PURPLE), width=2.5),

hovertemplate="%{x|%Y-%m-%d}<br>Equity=%{y:,.2f}<extra>"+name+"</extra>"
))
fig.update_layout(
title=title,
xaxis_title="Date",
yaxis_title="Equity",
template="plotly_white",
hovermode="x unified",
height=520,
legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="left",
x=0.0),
)
return fig

def drawdown_figure(series_map: Dict[str, [Link]], colors_map: Dict[str, str],

title: str):
fig = [Link]()
for name, eq in series_map.items():
eq = ensure_series(eq).dropna()
if [Link]:
continue
dd = drawdown_series(eq) * 100.0
col = colors_map.get(name, PURPLE)
fig.add_trace([Link](
x=[Link], y=[Link],
mode="lines",
name=name,
line=dict(color=col, width=2.0),
fill="tozeroy",
fillcolor=hex_to_rgba(col, 0.28),
hovertemplate="%{x|%Y-%m-%d}<br>DD=%{y:.2f}%<extra>"+name+"</extra>"
))
fig.update_layout(
title=title,
xaxis_title="Date",
 yaxis_title="Drawdown (%)",
template="plotly_white",
hovermode="x unified",
height=480,
legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="left",
x=0.0),
)
return fig

def days_in_dd_figure(series_map: Dict[str, [Link]], colors_map: Dict[str, str],

title: str):
fig = [Link]()
for name, eq in series_map.items():
eq = ensure_series(eq).dropna()
if [Link]:
continue
days = drawdown_days_series(eq)
col = colors_map.get(name, PURPLE)
fig.add_trace([Link](
x=[Link], y=[Link],
mode="lines",
name=name,
line=dict(color=col, width=2.0),
fill="tozeroy",
fillcolor=hex_to_rgba(col, 0.22),
hovertemplate="%{x|%Y-%m-%d}<br>Dias em DD=%{y}<extra>"+name+"</extra>"
))
fig.update_layout(
title=title,
xaxis_title="Date",
yaxis_title="Days in Drawdown",
template="plotly_white",
hovermode="x unified",
height=480,
legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="left",
x=0.0),
)
return fig

# ============================================================
# CORR HEATMAP (strategies)
# ============================================================

def corr_heatmap_figure(corr: [Link], title: str):

z = [Link]
text = [Link](lambda v: f"{v:+.2f}")(z)

fig = [Link]([Link](
z=z,
x=[Link](),
y=[Link](),
zmin=-1.0,
zmax=1.0,
zmid=0.0,
colorscale=[[0.0, PURPLE],[0.5, DARK],[1.0, AQUA]],
text=text,
texttemplate="%{text}",
hovertemplate="A=%{y}<br>B=%{x}<br>Corr=%{z:.3f}<extra></extra>"
))
 fig.update_layout(
title=title,
template="plotly_white",
height=max(420, 90 + 28 * len(corr)),
xaxis_title="Strategy",
yaxis_title="Strategy",
)
return fig

# ============================================================
# OPTIMIZERS (Log-returns + Shrinkage + Risk Parity + Markowitz)
# ============================================================

def shrink_covariance(cov: [Link], shrink: float = 0.15) -> [Link]:

"""
Simple shrinkage toward diagonal (variance).
cov_shrunk = (1-a)*cov + a*diag(cov)
"""
cov = [Link](cov, dtype=float)
a = float([Link](shrink, 0.0, 1.0))
diag = [Link]([Link](cov))
return (1.0 - a) * cov + a * diag

# ---------- NEW: robust risk parity + helper ----------

def _risk_contributions(w: [Link], cov: [Link]):
w = [Link](w, dtype=float)
cov = [Link](cov, dtype=float)
port_var = float(w @ cov @ w)
mrc = cov @ w
rc = (w * mrc) / (port_var + 1e-18)
return rc, port_var

def risk_parity_weights(
cov_ann: [Link],
long_only: bool = True,
max_weight: float = 1.0,
reg_eps: float = 1e-10,
n_random_starts: int = 8,
seed: int = 42,
):
"""
Risk Parity robusto:
- minimiza (RC - target)^2
- multi-start (equal, inverse-vol, random)
- bounds + sum(w)=1
"""
cov = [Link](cov_ann, dtype=float)
n = [Link][0]
cov = cov + [Link](n) * float(reg_eps)

target = [Link](n) / n

if long_only:
bounds = [(0.0, float(max_weight))] * n
w_eq = [Link](n) / n
else:
bounds = [(-float(max_weight), float(max_weight))] * n
w_eq = [Link](n) / n
cons = [{"type": "eq", "fun": lambda w: [Link](w) - 1.0}]

def obj(w):
w = [Link](w, dtype=float)
if long_only:
w = [Link](w, 0.0, float(max_weight))
else:
w = [Link](w, -float(max_weight), float(max_weight))
s = [Link]()
if not [Link](s) or abs(s) < 1e-18:
w = w_eq
else:
w = w / s

rc, _ = _risk_contributions(w, cov)

return float([Link]((rc - target) ** 2))

vols = [Link]([Link]([Link](cov), 1e-18))

w_iv = 1.0 / [Link](vols, 1e-18)
w_iv = w_iv / w_iv.sum()

starts = [w_iv, w_eq]

rng_local = [Link].default_rng(seed)
if long_only:
for _ in range(n_random_starts):
w0 = rng_local.random(n)
w0 = w0 / [Link]()
w0 = [Link](w0, 0.0, float(max_weight))
w0 = w0 / [Link]()
[Link](w0)
else:
for _ in range(n_random_starts):
w0 = rng_local.normal(0, 1, size=n)
w0 = [Link](w0, -float(max_weight), float(max_weight))
s = [Link]()
w0 = (w0 / s) if abs(s) > 1e-18 else w_eq
[Link](w0)

best_w, best_res, best_val = None, None, None

for w0 in starts:
res = minimize(
obj,
x0=w0,
method="SLSQP",
bounds=bounds,
constraints=cons,
options={"maxiter": 2000, "ftol": 1e-14}
)
w = [Link](res.x, dtype=float)

if long_only:
w = [Link](w, 0.0, float(max_weight))
else:
w = [Link](w, -float(max_weight), float(max_weight))

s = [Link]()
w = (w / s) if ([Link](s) and abs(s) > 1e-18) else w_eq
 val = obj(w)
if (best_val is None) or (val < best_val):
best_val, best_w, best_res = val, w, res

return best_w, best_res

def markowitz_max_sharpe(
returns_df_log: [Link],
rf_annual: float,
ann_days: float,
long_only: bool = True,
max_weight: float = 1.0,
reg_eps: float = 1e-10,
shrinkage: float = 0.15,
):
"""
Max Sharpe using LOG-RETURNS for estimation + shrinkage covariance.
returns_df_log: daily log-returns (columns=strategies), aligned, no NaN
"""
R = returns_df_log.copy().dropna(how="any")
n = [Link][1]
if n == 0:
raise ValueError("returns_df_log vazio.")

mu_ann = ([Link]() * ann_days).values

cov_ann = ([Link]() * ann_days).values
cov_ann = shrink_covariance(cov_ann, shrink=float(shrinkage))
cov_ann = cov_ann + [Link](n) * float(reg_eps)

rf_log = float([Link](1.0 + rf_annual)) if rf_annual != 0 else 0.0

if n == 1:
w = [Link]([1.0], dtype=float)
return w, mu_ann, cov_ann, None

def neg_sharpe(w):
w = [Link](w, dtype=float)
pret = float([Link](w, mu_ann))
pvol = float([Link](w @ cov_ann @ w))
if not [Link](pvol) or pvol <= 0:
return 1e9
return - (pret - rf_log) / pvol

cons = [{"type": "eq", "fun": lambda w: [Link](w) - 1.0}]

if long_only:
bounds = [(0.0, float(max_weight))] * n
x0 = [Link](n) / n
else:
bounds = [(-float(max_weight), float(max_weight))] * n
x0 = [Link](n) / n

res = minimize(neg_sharpe, x0=x0, method="SLSQP", bounds=bounds,

constraints=cons)

w = [Link](res.x, dtype=float)
if long_only:
w = [Link](w, 0.0, float(max_weight))
s = [Link]()
 w = w / s if s > 0 else [Link](n) / n

return w, mu_ann, cov_ann, res

# ============================================================
# TOP-3 DRAWDOWNS
# ============================================================

def top_n_drawdowns_from_equity(equity: [Link], n: int = 3, tol: float = 1e-12)

-> List[dict]:
eq = ensure_series(equity).dropna()
if [Link]:
return []
peaks = [Link]()
dd = eq / peaks - 1.0

episodes = []
in_dd = False
start_i = trough_i = None
trough_val = 0.0

for t in [Link]:
val = float([Link][t])
at_peak = float([Link][t] - [Link][t]) >= -tol

if (not in_dd) and (val < 0):

in_dd = True
start_i = t
trough_i = t
trough_val = val

elif in_dd:
if val < trough_val:
trough_val, trough_i = val, t
if at_peak:
[Link]({"depth": trough_val, "start": start_i, "trough":
trough_i, "end": t})
in_dd = False
start_i = trough_i = None
trough_val = 0.0

if in_dd:
[Link]({"depth": trough_val, "start": start_i, "trough": trough_i,
"end": [Link][-1]})

episodes = sorted(episodes, key=lambda d: d["depth"])[:n]

out = []
for k, e in enumerate(episodes, 1):
dur = int((pd.to_datetime(e["end"]) - pd.to_datetime(e["start"])).days)
[Link]({
"rank": k,
"depth_pct": float(e["depth"] * 100.0),
"start": pd.to_datetime(e["start"]).date(),
"trough": pd.to_datetime(e["trough"]).date(),
"end": pd.to_datetime(e["end"]).date(),
"duration_days": dur
})
return out
def top3_timeline_all(df_top3dd: [Link]) -> Optional[[Link]]:
df = df_top3dd.copy().dropna(subset=["start", "end", "trough", "depth_pct"])
if [Link]:
return None

df["start"] = pd.to_datetime(df["start"])
df["end"] = pd.to_datetime(df["end"])
df["trough"] = pd.to_datetime(df["trough"])
df["rank"] = df["rank"].astype(int)

series_order = sorted(df["Series"].unique().tolist())
y_map = {s: i for i, s in enumerate(series_order)}

fig = [Link]()
shown_rank_legend = set()

for _, r in [Link]():
series = r["Series"]
y = y_map[series]
rank = int(r["rank"])
col = RANK_COLORS.get(rank, PURPLE)

showleg = rank not in shown_rank_legend

if showleg:
shown_rank_legend.add(rank)

fig.add_trace([Link](
x=[r["start"], r["end"]],
y=[y, y],
mode="lines",
line=dict(color=col, width=10),
name=f"Rank {rank}",
legendgroup=f"rank{rank}",
showlegend=showleg,
hovertemplate=(
f"<b>{series}</b><br>"
f"Rank={rank}<br>"
f"Depth={r['depth_pct']:.2f}%<br>"
f"Start=%{{x|%Y-%m-%d}}<br>"
f"End=%{{x|%Y-%m-%d}}<extra></extra>"
)
))

fig.add_trace([Link](
x=df["trough"],
y=df["Series"].map(y_map),
mode="markers",
marker=dict(size=10, symbol="x", color=DARK),
name="Trough",
hovertemplate="<b>%{text}</b><br>Trough=%{x|%Y-%m-%d}<extra></extra>",
text=df["Series"]
))

fig.update_layout(
title="Top-3 Drawdowns — episódios (todas as séries)",
template="plotly_white",
height=max(520, 120 + 35 * len(series_order)),
xaxis_title="Date",
 yaxis=dict(
title="Series",
tickmode="array",
tickvals=list(y_map.values()),
ticktext=list(y_map.keys()),
autorange="reversed"
),
legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="left",
x=0.0),
)
return fig

def top3_timeline_per_series(df_top3dd: [Link]) -> Dict[str, [Link]]:

df = df_top3dd.copy().dropna(subset=["start", "end", "trough", "depth_pct"])
figs = {}
if [Link]:
return figs

df["start"] = pd.to_datetime(df["start"])
df["end"] = pd.to_datetime(df["end"])
df["trough"] = pd.to_datetime(df["trough"])
df["rank"] = df["rank"].astype(int)

for series in sorted(df["Series"].unique()):

d = df[df["Series"] == series].sort_values("rank")

fig = [Link]()
shown_rank_legend = set()

for _, r in [Link]():
rank = int(r["rank"])
col = RANK_COLORS.get(rank, PURPLE)

showleg = rank not in shown_rank_legend

if showleg:
shown_rank_legend.add(rank)

fig.add_trace([Link](
x=[r["start"], r["end"]],
y=[rank, rank],
mode="lines",
line=dict(color=col, width=10),
name=f"Rank {rank}",
legendgroup=f"rank{rank}",
showlegend=showleg,
hovertemplate=(
f"<b>{series}</b><br>"
f"Rank={rank}<br>"
f"Depth={r['depth_pct']:.2f}%<br>"
f"Start=%{{x|%Y-%m-%d}}<br>"
f"End=%{{x|%Y-%m-%d}}<extra></extra>"
)
))

fig.add_trace([Link](
x=d["trough"],
y=d["rank"],
mode="markers",
marker=dict(size=10, symbol="x", color=DARK),
 name="Trough",
hovertemplate="<b>Trough</b><br>%{x|%Y-%m-%d}<extra></extra>",
))

fig.update_layout(
title=f"Top-3 Drawdowns — {series}",
template="plotly_white",
height=420,
xaxis_title="Date",
yaxis=dict(title="Rank (1 = pior)", autorange="reversed",
tickmode="array", tickvals=[1, 2, 3]),
legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="left",
x=0.0),
)
figs[series] = fig

return figs
# ============================================================
# PDF EXPORT HELPERS (Plotly -> PNG -> ReportLab PDF)
# ============================================================

from io import BytesIO

from datetime import datetime

from [Link] import canvas

from [Link] import A4, landscape
from [Link] import cm
from [Link] import ImageReader

import [Link] as plt

def _fig_to_png_bytes(fig, width=1400, height=800, scale=2):

"""Plotly figure -> PNG bytes (via kaleido)."""
img_bytes = pio.to_image(fig, format="png", width=width, height=height,
scale=scale)
return BytesIO(img_bytes)
def _df_to_png_bytes(df: [Link], title: str = "", max_rows: int = 35,
font_size: int = 8):
"""DataFrame -> PNG bytes (matplotlib table). Splits handled outside."""
d = [Link]()

# ✅ Inclui índice (muito importante pra matrizes/weights)

if not isinstance([Link], [Link]) or ([Link] is not None):
d = d.reset_index()

# string formatting leve

for c in [Link]:
if [Link].is_float_dtype(d[c]) or
[Link].is_integer_dtype(d[c]):
d[c] = d[c].map(lambda x: "" if [Link](x) else f"{x:,.4g}" if
isinstance(x, float) else f"{x:,}")
else:
d[c] = d[c].astype(str)

if len(d) > max_rows:

d = [Link][:max_rows].copy()

# tamanho dinâmico
 nrows, ncols = [Link]
fig_w = min(18, 3 + 1.1 * ncols)
fig_h = min(10, 1.5 + 0.35 * nrows)

fig, ax = [Link](figsize=(fig_w, fig_h))

[Link]("off")

tbl = [Link](
cellText=[Link],
colLabels=[Link](),
loc="center",
cellLoc="center",
)

tbl.auto_set_font_size(False)
tbl.set_fontsize(font_size)
[Link](1.0, 1.15)

if title:
ax.set_title(title, fontsize=12, pad=12)

buf = BytesIO()
plt.tight_layout()
[Link](buf, format="png", dpi=200, bbox_inches="tight")
[Link](fig)
[Link](0)
return buf

def _add_image_page(c: [Link], img_buf: BytesIO, header: str, footer: str =

""):
"""Add one full page with header + image."""
page_w, page_h = landscape(A4)
margin = 0.8 * cm
header_h = 1.2 * cm
footer_h = 0.8 * cm

# Header
[Link]("Helvetica-Bold", 13)
[Link](margin, page_h - margin, header)

# Footer
if footer:
[Link]("Helvetica", 9)
[Link](margin, margin * 0.6, footer)

# Image placement
img = ImageReader(img_buf)
iw, ih = [Link]()

max_w = page_w - 2 * margin

max_h = page_h - (margin + header_h) - (margin + footer_h)

s = min(max_w / iw, max_h / ih)

w = iw * s
h = ih * s

x = (page_w - w) / 2
y = (page_h - h) / 2 - 0.2 * cm
 [Link](img, x, y, width=w, height=h, preserveAspectRatio=True,
mask='auto')
[Link]()

def _split_df(df: [Link], rows_per_page: int):

if rows_per_page is None or rows_per_page <= 0 or len(df) <= rows_per_page:
return [df]
out = []
for i in range(0, len(df), rows_per_page):
[Link]([Link][i:i + rows_per_page].copy())
return out

def export_full_report_pdf(
out_path: Path,
report_end: [Link],
strategies: List[str],
benchmarks: List[str],
df_metrics: [Link],
all_equity: Dict[str, [Link]],
colors_map: Dict[str, str],
corr: Optional[[Link]] = None,
portfolio_weights: Optional[object] = None, # ✅ aceita Series OU dict[str,
Series/DataFrame]
df_top3dd: Optional[[Link]] = None,
sim_targets: Optional[List[dict]] = None,
include_per_series: bool = True,
include_simulations: bool = True,
max_table_rows: int = 35,
extra_tables: Optional[List[Tuple[str, [Link]]]] = None, # ✅ NOVO
):

import importlib
import [Link] as pio
import [Link]._kaleido as _kaleido
[Link](_kaleido)
[Link](pio)

# --- Ensure kaleido is active (no hard fail) ---

if [Link] is None:
try:
from [Link] import PlotlyScope
scope = PlotlyScope()
scope.chromium_args = ("--no-sandbox", "--disable-dev-shm-usage")
[Link] = scope
except Exception as e:
raise RuntimeError(f"Kaleido não inicializou no Plotly. Erro: {e}")

out_path = Path(out_path)
out_path.[Link](parents=True, exist_ok=True)

c = [Link](str(out_path), pagesize=landscape(A4))
page_w, page_h = landscape(A4)

# ---- Cover page (texto simples)

margin = 1.2 * cm
[Link]("Helvetica-Bold", 18)
 [Link](margin, page_h - 2.0 * cm, "Multi-Strategy Report (Strategies +
Benchmarks + Portfolio)")

[Link]("Helvetica", 11)
now = [Link]().strftime("%Y-%m-%d %H:%M:%S")
[Link](margin, page_h - 3.0 * cm, f"Generated at: {now}")
[Link](margin, page_h - 3.7 * cm, f"Report end date:
{pd.to_datetime(report_end).date()}")

y = page_h - 5.0 * cm
[Link]("Helvetica-Bold", 12)
[Link](margin, y, "Strategies:")
y -= 0.6 * cm
[Link]("Helvetica", 11)
for s in strategies:
[Link](margin + 0.5 * cm, y, f"• {s}")
y -= 0.55 * cm
if y < 2.0 * cm:
[Link]()
y = page_h - 2.0 * cm

[Link]("Helvetica-Bold", 12)
[Link](margin, y, "Benchmarks:")
y -= 0.6 * cm
[Link]("Helvetica", 11)
for b in benchmarks:
[Link](margin + 0.5 * cm, y, f"• {b}")
y -= 0.55 * cm
if y < 2.0 * cm:
[Link]()
y = page_h - 2.0 * cm

[Link]()

footer = f"End: {pd.to_datetime(report_end).date()}"

# ---- Metrics table

for k, chunk in enumerate(_split_df(df_metrics, max_table_rows), 1):
img = _df_to_png_bytes(chunk, title=f"Metrics Table (page {k})",
max_rows=max_table_rows, font_size=7)
_add_image_page(c, img, header="Metrics", footer=footer)

# ---- Overlay figs (recria aqui para garantir que existe)

fig_eq = equity_figure(all_equity, colors_map, title=f"Equity Curves (Overlay)
— até {pd.to_datetime(report_end).date()}")
_add_image_page(c, _fig_to_png_bytes(fig_eq), header="Equity Overlay",
footer=footer)

fig_dd = drawdown_figure(all_equity, colors_map, title="Drawdown (%)

(Overlay)")
_add_image_page(c, _fig_to_png_bytes(fig_dd), header="Drawdown Overlay",
footer=footer)

fig_days = days_in_dd_figure(all_equity, colors_map, title="Days in Drawdown

(Overlay)")
_add_image_page(c, _fig_to_png_bytes(fig_days), header="Days in Drawdown
Overlay", footer=footer)

# ---- Correlation heatmap (strategies) + weights

 # ---- Correlation heatmap (strategies) + (optional) correlation table
if corr is not None and isinstance(corr, [Link]) and [Link][0] > 0:
fig_corr = corr_heatmap_figure(corr, title="Correlation Heatmap —
Strategies (Daily LOG Returns)")
_add_image_page(c, _fig_to_png_bytes(fig_corr), header="Correlation
(Strategies)", footer=footer)

# ✅ também adiciona a matriz como TABELA (opcional, mas normalmente

desejado)
corr_tbl = [Link]()
for k, chunk in enumerate(_split_df(corr_tbl, max_table_rows), 1):
img = _df_to_png_bytes(chunk, title=f"Correlation Matrix (table) — page
{k}", max_rows=max_table_rows, font_size=7)
_add_image_page(c, img, header="Correlation (table)", footer=footer)

# ---- Portfolio weights (accept Series OR dict[str, Series/DataFrame])

if portfolio_weights is not None:
weights_dict = None

if isinstance(portfolio_weights, [Link]):
weights_dict = {"portfolio": portfolio_weights}
elif isinstance(portfolio_weights, dict):
weights_dict = portfolio_weights

if weights_dict:
for label, w in weights_dict.items():
if w is None:
continue

if isinstance(w, [Link]):
wdf = w.to_frame("weight")
elif isinstance(w, [Link]):
wdf = [Link]()
else:
continue

for k, chunk in enumerate(_split_df(wdf, max_table_rows), 1):

img = _df_to_png_bytes(chunk, title=f"Weights — {label} (page
{k})", max_rows=max_table_rows, font_size=9)
_add_image_page(c, img, header=f"Portfolio Weights — {label}",
footer=footer)

# ---- Monthly heatmaps (por série)

for name, eq in all_equity.items():
heat = monthly_returns_compound(eq)
fig_hm = heatmap_figure(heat, title=f"Monthly Returns Heatmap — {name}")
_add_image_page(c, _fig_to_png_bytes(fig_hm), header=f"Monthly Heatmap —
{name}", footer=footer)

# ---- Top-3 DD table + timeline

if df_top3dd is not None and isinstance(df_top3dd, [Link]) and
len(df_top3dd) > 0:
for k, chunk in enumerate(_split_df(df_top3dd, max_table_rows), 1):
img = _df_to_png_bytes(chunk, title=f"Top-3 Drawdowns Table (page
{k})", max_rows=max_table_rows, font_size=8)
_add_image_page(c, img, header="Top-3 Drawdowns (Table)",
footer=footer)

fig_top_all = top3_timeline_all(df_top3dd)
 if fig_top_all is not None:
_add_image_page(c, _fig_to_png_bytes(fig_top_all, width=1600,
height=900, scale=2), header="Top-3 Drawdowns (Timeline)", footer=footer)

# ---- Per-series pages (equity + dd + days)

if include_per_series:
for name, eq in all_equity.items():
fig1 = equity_figure({name: eq}, colors_map, title=f"Equity — {name}")
_add_image_page(c, _fig_to_png_bytes(fig1), header=f"Equity — {name}",
footer=footer)

fig2 = drawdown_figure({name: eq}, colors_map, title=f"Drawdown (%) —

{name}")
_add_image_page(c, _fig_to_png_bytes(fig2), header=f"Drawdown —
{name}", footer=footer)

fig3 = days_in_dd_figure({name: eq}, colors_map, title=f"Days in

Drawdown — {name}")
_add_image_page(c, _fig_to_png_bytes(fig3), header=f"Days in DD —
{name}", footer=footer)

# ---- Simulações (re-gera figs a partir de sim_targets)

if include_simulations and sim_targets:
for tgt in sim_targets:
nm = tgt["name"]
ret_arr = [Link](tgt["returns"], dtype=float)
start_for_sim = float(tgt["start"])

if len(ret_arr) == 0:
continue

mc_final, mc_dd = mc_perturbed_trades(

ret_arr, n_sims=N_MC, start=float(start_for_sim),
sigma_scale=PERTURB_SIGMA, slippage_bps=SLIPPAGE_BPS
)
bs_final, bs_dd = bootstrap_resample(ret_arr, n_sims=N_BOOT,
start=float(start_for_sim))

fig_final = overlay_hist_fig(
mc_final, bs_final,
title=f"Final Equity (Overlay) — {nm}",
name_a="Monte Carlo (perturbed)", name_b="Bootstrap (resample)",
color_a=PURPLE, color_b=AQUA,
x_title="Final Equity"
)
_add_image_page(c, _fig_to_png_bytes(fig_final), header=f"Simulations —
Final Equity — {nm}", footer=footer)

fig_ddo = overlay_hist_fig(
mc_dd * 100.0, bs_dd * 100.0,
title=f"Max Drawdown (%) (Overlay) — {nm}",
name_a="Monte Carlo (perturbed)", name_b="Bootstrap (resample)",
color_a=PURPLE, color_b=AQUA,
x_title="Max DD (%)"
)
_add_image_page(c, _fig_to_png_bytes(fig_ddo), header=f"Simulations —
Max DD — {nm}", footer=footer)
# ---- EXTRA TABLES (overlap, weekly corr, spearman, etc.)
if extra_tables:
 for title, df in extra_tables:
if df is None or (not isinstance(df, [Link])) or [Link]:
continue
for k, chunk in enumerate(_split_df(df, max_table_rows), 1):
img = _df_to_png_bytes(chunk, title=f"{title} (page {k})",
max_rows=max_table_rows, font_size=7)
_add_image_page(c, img, header=title, footer=footer)

[Link]()

# ============================================================
# MAIN
# ============================================================

# 1) Carregar estratégias
strategy_ctxs = []
for s in STRATEGIES:
ctx = build_strategy_from_excel(s["path"], [Link]("start_balance", None))
ctx["name"] = s["name"]
ctx["path"] = s["path"]
strategy_ctxs.append(ctx)

global_start = min(c["start_date"] for c in strategy_ctxs)

global_end = max(c["end_date"] for c in strategy_ctxs)

# report end date (today in local tz), and extend equities up to it

TODAY = local_today_date(LOCAL_TZ)
REPORT_END_DATE = max(global_end, TODAY) if EXTEND_EQUITY_TO_TODAY_FOR_PLOTS else
global_end

print(f"\n Range original (estratégias): {global_start.date()} →

{global_end.date()}")
print(f" Report end date (plots/portfolio): {global_start.date()} →
{REPORT_END_DATE.date()} (today={[Link]()})")

# 2) Carregar benchmarks (mesmo range do relatório)

benchmark_series = {}
for b in BENCHMARKS:
eq = build_benchmark_equity(
b["ticker"],
global_start,
REPORT_END_DATE,
start_balance=BASE_VALUE if NORMALIZE_ALL_TO_BASE else strategy_ctxs[0]
["start_balance"]
)
if eq is None or len(eq) == 0:
print(f"[WARN] Falha ao baixar benchmark: {b['ticker']} ({b['name']})")
continue
benchmark_series[b["name"]] = eq

# 3) Montar mapa de séries (estratégias + benchmarks) e normalizar + esticar

all_equity: Dict[str, [Link]] = {}
series_meta: Dict[str, dict] = {}

strategy_names = []
for c in strategy_ctxs:
name = c["name"]
eq = c["equity_daily"]
 if NORMALIZE_ALL_TO_BASE:
eq = rebase_to(eq, BASE_VALUE)

eq = extend_daily_to(eq, REPORT_END_DATE)

all_equity[name] = eq
strategy_names.append(name)
series_meta[name] = {
"Type": "Strategy",
"Trades (EXIT)": c["metrics"].get("Trades (EXIT)", [Link]),
"Sharpe (event-time)": c["metrics"].get("Sharpe (event-time)", [Link]),
"Max Drawdown (event/exits)": c["metrics"].get("Max Drawdown
(event/exits)", [Link]),
}

for b in BENCHMARKS:
name = b["name"]
if name in benchmark_series:
eq = benchmark_series[name]
if NORMALIZE_ALL_TO_BASE:
eq = rebase_to(eq, BASE_VALUE)
eq = extend_daily_to(eq, REPORT_END_DATE)
all_equity[name] = eq
series_meta[name] = {
"Type": "Benchmark",
"Ticker": b["ticker"],
"Trades (EXIT)": [Link],
"Sharpe (event-time)": [Link],
"Max Drawdown (event/exits)": [Link],
}

# ============================================================
# PORTFOLIO (Strategies only)
# - estimation uses LOG-returns + shrinkage
# - baseline: Risk Parity
# - aggressive: Markowitz max Sharpe
# - equity built with arithmetic returns (rebalance daily)
# ============================================================
# ============================================================
# PORTFOLIOS (Strategies only) — build BOTH:
# 1) Risk Parity (weight risk)
# 2) Max Sharpe (sharpe puro) [Markowitz]
# ============================================================

portfolio_weights = {} # dict: {"risk_parity": Series, "max_sharpe":

Series}
portfolio_daily_returns = {} # dict: {"risk_parity": Series, "max_sharpe":
Series}
PORTFOLIO_SERIES_NAMES = [] # <- NEW (lista com os 2 nomes)
PORTFOLIO_SERIES_NAME = None # mantém compatibilidade com o resto do script

# will be used later in "extra correlation" section

returns_df = None

if PORTFOLIO_ENABLE and len(strategy_names) >= 1:

port_start = max(all_equity[nm].[Link]() for nm in strategy_names)
port_end = REPORT_END_DATE
port_index = pd.date_range(port_start, port_end, freq="D")
 eq_mat = [Link](
{nm: all_equity[nm].reindex(port_index, method="ffill") for nm in
strategy_names},
index=port_index
)

ret_arith = eq_mat.pct_change().iloc[1:].dropna(how="any")
ret_log = [Link](eq_mat / eq_mat.shift(1)).iloc[1:].dropna(how="any")

# <- FIX for bottom section (so it exists)

returns_df = ret_arith.copy()

if ret_log.shape[0] < 30:

print("[WARN] Poucos dias alinhados para otimização. Pode ficar instável.")

corr = ret_log.corr()
print("\n🔗 Matriz de correlação (estratégias) — LOG-retornos diários (após
extensão até hoje):")
display(corr)

fig_corr = corr_heatmap_figure(corr, title="Correlation Heatmap — Strategies

(Daily LOG Returns)")
fig_corr.show()
if SAVE_HTML:
fig_corr.write_html(str(HTML_OUT_DIR / "corr_heatmap_strategies.html"))

cov_ann_raw = (ret_log.cov().values * ANN_DAYS_MAIN)

cov_ann = shrink_covariance(cov_ann_raw, shrink=PORTFOLIO_SHRINKAGE) +
[Link](ret_log.shape[1]) * float(PORTFOLIO_REG_EPS)

# -------------------------
# 1) RISK PARITY (weight risk)
# -------------------------
w_rp, res_rp = risk_parity_weights(
cov_ann,
long_only=PORTFOLIO_LONG_ONLY,
max_weight=PORTFOLIO_MAX_WEIGHT,
reg_eps=PORTFOLIO_REG_EPS,
n_random_starts=10,
seed=42
)
w_rp_s = [Link](w_rp, index=ret_log.columns,
name="weight_risk_parity").sort_values(ascending=False)
print("\n⚖️ Risk Parity — Pesos (weight risk) [cov shrinkage + cap]:")
display(w_rp_s.to_frame())

# DEBUG: success + message + RC + vols

print("risk parity success:", bool(getattr(res_rp, "success", False)))
print("risk parity message:", getattr(res_rp, "message", ""))

rc_rp, _ = _risk_contributions(w_rp, cov_ann)

rc_rp_s = [Link](rc_rp / [Link](rc_rp), index=ret_log.columns,
name="risk_contrib_pct").sort_values(ascending=False)
print("\n🧩 Risk Contributions (deveriam ficar ~iguais):")
display(rc_rp_s.to_frame())

vols_ann = [Link]([Link]([Link](cov_ann), 1e-18))

print("\n📌 Vol anual implícita (da cov usada no RP):")
display([Link](vols_ann, index=ret_log.columns,
name="vol_ann").sort_values())

# -------------------------
# 2) MAX SHARPE (sharpe puro) — Markowitz
# -------------------------
w_mw, mu_ann, cov_ann_mw, res_mw = markowitz_max_sharpe(
ret_log,
rf_annual=RF_ANNUAL,
ann_days=ANN_DAYS_MAIN,
long_only=PORTFOLIO_LONG_ONLY,
max_weight=PORTFOLIO_MAX_WEIGHT,
reg_eps=PORTFOLIO_REG_EPS,
shrinkage=PORTFOLIO_SHRINKAGE
)
w_mw_s = [Link](w_mw, index=ret_log.columns,
name="weight_max_sharpe").sort_values(ascending=False)

rf_log = float([Link](1.0 + RF_ANNUAL)) if RF_ANNUAL != 0 else 0.0

pret_ann_mw = float([Link](w_mw, mu_ann))
pvol_ann_mw = float([Link](w_mw @ cov_ann_mw @ w_mw))
psharpe_mw = (pret_ann_mw - rf_log) / pvol_ann_mw if ([Link](pvol_ann_mw)
and pvol_ann_mw > 0) else [Link]

print("\n🎯 Max Sharpe — Pesos (sharpe puro) [log-returns + cov shrinkage +

cap]:")
display(w_mw_s.to_frame())

summary_mw = [Link]([{
"Expected Return (ann, log)": pret_ann_mw,
"Expected Vol (ann, log)": pvol_ann_mw,
"Expected Sharpe (ann, log)": psharpe_mw,
"Shrinkage": PORTFOLIO_SHRINKAGE,
"Long-only": bool(PORTFOLIO_LONG_ONLY),
"Max weight": float(PORTFOLIO_MAX_WEIGHT),
"Days used": int(len(ret_log)),
"Portfolio Start": port_start.date(),
"Portfolio End": port_end.date(),
}])
print("\n📌 Resumo (Max Sharpe — estimado via mean/var de LOG-retornos):")
display(summary_mw)

# -------------------------
# Helper: construir equity a partir dos pesos (arithmetic daily rebalance)
# -------------------------
def _build_portfolio_eq(w: [Link], label: str):
w = [Link](w, dtype=float)
pr = ret_arith.dot(w) # Series index=ret_arith.index

pr_full = [Link](0.0, index=port_index)

pr_full.loc[[Link]] = [Link]

eq = (1.0 + pr_full).cumprod() * float(BASE_VALUE)

name = f"{PORTFOLIO_NAME} [{label}]"
[Link] = name
return name, eq, pr

# -------------------------
# Build BOTH portfolios
# -------------------------
 name_rp, eq_rp, pr_rp = _build_portfolio_eq(w_rp, "risk_parity")
name_ms, eq_ms, pr_ms = _build_portfolio_eq(w_mw, "max_sharpe")

# guardar nomes
PORTFOLIO_SERIES_NAMES = [name_rp, name_ms]

# manter compatibilidade: PORTFOLIO_SERIES_NAME = o "principal" (como antes)

mode = str(PORTFOLIO_OPT_MODE).strip().lower()
if mode not in ("risk_parity", "markowitz"):
mode = "risk_parity"
PORTFOLIO_SERIES_NAME = name_rp if mode == "risk_parity" else name_ms

# guardar pesos/returns em dicts

portfolio_weights["risk_parity"] = [Link](w_rp, index=ret_log.columns,
name="weight_risk_parity").sort_values(ascending=False)
portfolio_weights["max_sharpe"] = [Link](w_mw, index=ret_log.columns,
name="weight_max_sharpe").sort_values(ascending=False)

portfolio_daily_returns["risk_parity"] = pr_rp
portfolio_daily_returns["max_sharpe"] = pr_ms

# registrar em all_equity + meta (para entrar em tudo:

plots/tabelas/dd/heatmaps)
all_equity[name_rp] = eq_rp
series_meta[name_rp] = {"Type": "Portfolio", "Ticker": "", "Trades (EXIT)":
[Link], "Sharpe (event-time)": [Link], "Max Drawdown (event/exits)": [Link]}

all_equity[name_ms] = eq_ms
series_meta[name_ms] = {"Type": "Portfolio", "Ticker": "", "Trades (EXIT)":
[Link], "Sharpe (event-time)": [Link], "Max Drawdown (event/exits)": [Link]}

print(f"\n✅ Portfólios criados: {name_rp} e {name_ms}")

print(f"Compat: PORTFOLIO_SERIES_NAME = {PORTFOLIO_SERIES_NAME}")

# ============================================================
# (1) METRICS TABLE (inclui portfolio)
# ============================================================

metrics_rows = []
for name, eq in all_equity.items():
eq = ensure_series(eq).dropna()
if [Link]:
continue

start_bal = float([Link][0])
m = metrics_from_daily_equity(eq, start_bal, rf_annual=RF_ANNUAL)

meta = series_meta.get(name, {})

row = {"Series": name}
[Link]({k: [Link](k, [Link]) for k in ["Type", "Ticker", "Trades (EXIT)",
"Sharpe (event-time)", "Max Drawdown (event/exits)"]})
[Link](m)
metrics_rows.append(row)

df_metrics = [Link](metrics_rows)

cols_order = [
"Series", "Type", "Ticker", "Trades (EXIT)",
"Start Balance", "Final Balance", "Net Profit", "Net % Gain", "CAGR",
 "Sharpe (365)", "Sortino (365)", "Volatility (365)",
"Sharpe (252)", "Sortino (252)", "Volatility (252)",
"Max Drawdown", "Max Drawdown %", "Max Drawdown (event/exits)",
"MAR (Calmar)",
"Skewness", "Kurtosis (excess)", "Days",
"Sharpe (event-time)"
]
df_metrics = df_metrics[[c for c in cols_order if c in df_metrics.columns]]

print("\n📊 Tabela de métricas (estratégias + benchmarks + portfolio) — já esticadas

até o report end:")
display(df_metrics)
# ============================================================
# COLORS MAP (crie DEPOIS do all_equity existir)
# ============================================================
series_list = list(all_equity.keys()) # ordem de inserção
palette = gradient_colors(len(series_list), c1=PURPLE, c2=AQUA)
colors_map = {name: palette[i] for i, name in enumerate(series_list)}

# ============================================================
# (2) EQUITY / DD / DAYS IN DD
# ============================================================

fig_eq_all = equity_figure(
all_equity,
colors_map,
title=f"Equity Curves (Overlay) — {('Rebased to '+str(BASE_VALUE)) if
NORMALIZE_ALL_TO_BASE else 'Raw'} — até {REPORT_END_DATE.date()}"
)
fig_eq_all.show()
if SAVE_HTML:
fig_eq_all.write_html(str(HTML_OUT_DIR / "equity_overlay.html"))

fig_dd_all = drawdown_figure(all_equity, colors_map, title="Drawdown (%) (Overlay)

— Área")
fig_dd_all.show()
if SAVE_HTML:
fig_dd_all.write_html(str(HTML_OUT_DIR / "drawdown_overlay.html"))

fig_days_all = days_in_dd_figure(all_equity, colors_map, title="Days in Drawdown

(Overlay) — Área")
fig_days_all.show()
if SAVE_HTML:
fig_days_all.write_html(str(HTML_OUT_DIR / "days_in_dd_overlay.html"))

# ============================================================
# (3) HEATMAP MENSAL — por série (inclui portfolio)
# ============================================================

for name, eq in all_equity.items():

heat = monthly_returns_compound(eq)
fig = heatmap_figure(heat, title=f"Monthly Returns Heatmap (Compound) —
{name}")
[Link]()
if SAVE_HTML:
safe = "".join(ch if [Link]() or ch in " _-" else "_" for ch in
name).strip().replace(" ", "_")
fig.write_html(str(HTML_OUT_DIR / f"heatmap_{safe}.html"))
# ============================================================
# (4) TOP-3 DRAWDOWNS — (tabela + plots)
# ============================================================

top3_rows = []
for name, eq in all_equity.items():
eps = top_n_drawdowns_from_equity(eq, n=3)
if not eps:
top3_rows.append({"Series": name, "rank": 1, "depth_pct": [Link], "start":
None, "trough": None, "end": None, "duration_days": None})
continue
for e in eps:
top3_rows.append({"Series": name, **e})

df_top3dd = [Link](top3_rows).sort_values(["Series", "rank"])

print("\n📉 Top-3 drawdowns (episódios) por série:")
display(df_top3dd)

fig_top3_all = top3_timeline_all(df_top3dd)
if fig_top3_all is not None:
fig_top3_all.show()
if SAVE_HTML:
fig_top3_all.write_html(str(HTML_OUT_DIR /
"top3_drawdowns_timeline_all.html"))

figs_top3 = top3_timeline_per_series(df_top3dd)
for series, fig in figs_top3.items():
[Link]()
if SAVE_HTML:
safe = "".join(ch if [Link]() or ch in " _-" else "_" for ch in
series).strip().replace(" ", "_")
fig.write_html(str(HTML_OUT_DIR / f"top3_drawdowns_timeline_{safe}.html"))

# ============================================================
# (5) SIMULAÇÕES — estratégias + portfolio
# - estratégias: trade_ret (event returns)
# - portfolio: daily returns (porque não tem "trades")
# ============================================================

sim_targets = []

for c in strategy_ctxs:
sim_targets.append({
"name": c["name"],
"returns": [Link](c["trade_ret"], dtype=float),
"start": float(BASE_VALUE if NORMALIZE_ALL_TO_BASE else
c["start_balance"]),
"kind": "strategy-trade-returns"
})

if (PORTFOLIO_SERIES_NAME in all_equity):
port_ret_full =
all_equity[PORTFOLIO_SERIES_NAME].pct_change().fillna(0.0).values
sim_targets.append({
"name": PORTFOLIO_SERIES_NAME,
"returns": [Link](port_ret_full, dtype=float),
"start": float(BASE_VALUE),
"kind": "portfolio-daily-returns"
})
for tgt in sim_targets:
name = tgt["name"]
ret_arr = tgt["returns"]
start_for_sim = tgt["start"]

if len(ret_arr) == 0:
print(f"[WARN] Sem retornos para simulação: {name}")
continue

mc_final, mc_dd = mc_perturbed_trades(

ret_arr, n_sims=N_MC, start=float(start_for_sim),
sigma_scale=PERTURB_SIGMA, slippage_bps=SLIPPAGE_BPS
)
bs_final, bs_dd = bootstrap_resample(ret_arr, n_sims=N_BOOT,
start=float(start_for_sim))

fig_mc_final = hist_fig(mc_final, f"Monte Carlo — Final Equity — {name}",

color=PURPLE, x_title="Final Equity")
fig_bs_final = hist_fig(bs_final, f"Bootstrap — Final Equity — {name}",
color=AQUA, x_title="Final Equity")
fig_mc_final.show()
fig_bs_final.show()

fig_overlay_final = overlay_hist_fig(
mc_final, bs_final,
title=f"Final Equity (Overlay) — {name}",
name_a="Monte Carlo (perturbed)", name_b="Bootstrap (resample)",
color_a=PURPLE, color_b=AQUA,
x_title="Final Equity"
)
fig_overlay_final.show()

fig_mc_dd = hist_fig(mc_dd * 100.0, f"Monte Carlo — Max Drawdown (%) — {name}",

color=PURPLE, x_title="Max DD (%)")
fig_bs_dd = hist_fig(bs_dd * 100.0, f"Bootstrap — Max Drawdown (%) — {name}",
color=AQUA, x_title="Max DD (%)")
fig_mc_dd.show()
fig_bs_dd.show()

fig_overlay_dd = overlay_hist_fig(
mc_dd * 100.0, bs_dd * 100.0,
title=f"Max Drawdown (%) (Overlay) — {name}",
name_a="Monte Carlo (perturbed)", name_b="Bootstrap (resample)",
color_a=PURPLE, color_b=AQUA,
x_title="Max DD (%)"
)
fig_overlay_dd.show()

# ============================================================
# EXTRA CORRELATION ANALYSES (fixed)
# ============================================================

def pairwise_corr_overlap(R: [Link], eps: float = 1e-12, method: str =

"pearson"):
cols = [Link]
corr = [Link]([Link], index=cols, columns=cols)
nobs = [Link](0, index=cols, columns=cols)
 for i, a in enumerate(cols):
for j, b in enumerate(cols):
if j < i:
continue
mask = (R[a].abs() > eps) & (R[b].abs() > eps)
x = [Link][mask, a]
y = [Link][mask, b]
n = len(x)
[Link][a, b] = [Link][b, a] = n
if n >= 2:
[Link][a, b] = [Link][b, a] = [Link](y, method=method)

np.fill_diagonal([Link], 1.0)
return corr, nobs

if returns_df is not None and isinstance(returns_df, [Link]) and not

returns_df.empty:
R = returns_df.copy().dropna(how="any")

print("Dias no cálculo (daily):", len(R))

# 1) Pearson diário (como está)

corr_daily = [Link]()
print("\nPearson diário:")
display(corr_daily)

# 2) Overlap ativo (ambas mexem no mesmo dia)

corr_overlap, nobs_overlap = pairwise_corr_overlap(R, eps=1e-12,
method="pearson")
print("\nCorrelação (somente dias em que AMBAS têm retorno != 0):")
display(corr_overlap)
print("\nNº de observações usadas por par (overlap ativo):")
display(nobs_overlap)

# 3) Semanal (composto)
R_week = (1 + R).resample("W").prod() - 1
print("\nSemanas no cálculo:", len(R_week))
corr_week = R_week.corr()
print("\nCorrelação semanal (retorno composto):")
display(corr_week)

# 4) Spearman
corr_spear = [Link](method="spearman")
print("\nSpearman diário:")
display(corr_spear)
else:
print("\n[WARN] returns_df não disponível (portfolio pode estar desabilitado ou
sem dados). Pulando correlações extras.")

print("\n✅ Pronto: Risk Parity (baseline robusto) + Markowitz (agressivo) +

shrinkage + cap + equity esticada até hoje.")
print("Range (estratégias):", global_start.date(), "→", global_end.date())
print("Report end:", REPORT_END_DATE.date())
extra_tables_pdf = []

# (A) correlações extras

if returns_df is not None and isinstance(returns_df, [Link]) and not
returns_df.empty:
extra_tables_pdf.append(("Correlation — Pearson daily (arith)", corr_daily))
 extra_tables_pdf.append(("Correlation — Active overlap only (arith)",
corr_overlap))
extra_tables_pdf.append(("N obs — Active overlap", nobs_overlap))
extra_tables_pdf.append(("Correlation — Weekly compounded", corr_week))
extra_tables_pdf.append(("Correlation — Spearman daily (arith)", corr_spear))

# (B) pesos + risk contrib + vol anual + summary MW (se existirem)

if "portfolio_weights" in globals() and isinstance(portfolio_weights, dict):
# já vão entrar pelo bloco de weights dict, mas se quiser também como “extra”:
pass

if "w_rp_s" in globals():
extra_tables_pdf.append(("Weights — Risk Parity", w_rp_s.to_frame()))
if "w_mw_s" in globals():
extra_tables_pdf.append(("Weights — Max Sharpe", w_mw_s.to_frame()))

if "rc_rp_s" in globals():
extra_tables_pdf.append(("Risk Contributions — Risk Parity",
rc_rp_s.to_frame()))

if "vols_ann" in globals() and "ret_log" in globals():

vol_df = [Link](vols_ann, index=ret_log.columns,
name="vol_ann").sort_values().to_frame()
extra_tables_pdf.append(("Vol anual implícita (RP cov)", vol_df))

if "summary_mw" in globals():
extra_tables_pdf.append(("Max Sharpe — Summary", summary_mw))

if PORTFOLIO_SERIES_NAME in all_equity:
pr = all_equity[PORTFOLIO_SERIES_NAME]
print(f"Range (portfolio): {[Link]().date()} → {[Link]().date()}")
if SAVE_HTML:
print("HTMLs salvos em:", HTML_OUT_DIR)
if PORTFOLIO_NAME in all_equity:
pr = all_equity[PORTFOLIO_NAME]
print(f"Range (portfolio): {[Link]().date()} → {[Link]().date()}")
if SAVE_HTML:
print("HTMLs salvos em:", HTML_OUT_DIR)

if SAVE_PDF:
export_full_report_pdf(
out_path=PDF_OUT_PATH,
report_end=REPORT_END_DATE,
strategies=strategy_names,
benchmarks=[b["name"] for b in BENCHMARKS if b["name"] in all_equity],
df_metrics=df_metrics,
all_equity=all_equity,
colors_map=colors_map,
corr=corr if "corr" in globals() else None,
portfolio_weights=portfolio_weights if "portfolio_weights" in globals()
else None,
df_top3dd=df_top3dd if "df_top3dd" in globals() else None,
sim_targets=sim_targets if "sim_targets" in globals() else None,
include_per_series=PDF_INCLUDE_PER_SERIES,
include_simulations=PDF_INCLUDE_SIMULATIONS,
max_table_rows=PDF_TABLE_MAX_ROWS,
extra_tables=extra_tables_pdf if "extra_tables_pdf" in globals() else None,
)
print("✅ PDF salvo em:", PDF_OUT_PATH)