"""Conclave PDF report — typeset with reportlab (crucible's report stack, simplified). Built directly from the result payload; tolerates missing sections (no LLM, no target). """ from __future__ import annotations import io from reportlab.lib import colors from reportlab.lib.pagesizes import A4 from reportlab.lib.styles import ParagraphStyle from reportlab.lib.units import mm from reportlab.platypus import (Paragraph, SimpleDocTemplate, Spacer, Table, TableStyle) INK = colors.HexColor("#131a22") BRASS = colors.HexColor("#0b7fa6") # brand accent (print-safe deep cyan) DIM = colors.HexColor("#5c6a7a") GOOD = colors.HexColor("#1e7a4a") WARN = colors.HexColor("#9a6b12") BAD = colors.HexColor("#a33a40") LINE = colors.HexColor("#ccd6e0") _ACTION_COLOR = {"Buy More": GOOD, "Keep": DIM, "Trim": WARN, "Sell": BAD} _SEV_COLOR = {"good": GOOD, "warn": WARN, "bad": BAD, "OK": GOOD, "WATCH": WARN, "ALERT": BAD} def _sty(name, **kw): base = dict(fontName="Helvetica", fontSize=9.5, leading=13, textColor=INK) base.update(kw) return ParagraphStyle(name, **base) # type: ignore[arg-type] S_TITLE = _sty("t", fontName="Times-Bold", fontSize=24, leading=28) S_TAG = _sty("tag", fontName="Times-Italic", fontSize=11, textColor=BRASS) S_H2 = _sty("h2", fontName="Helvetica-Bold", fontSize=11, leading=15, textColor=BRASS, spaceBefore=14, spaceAfter=5) S_BODY = _sty("b") S_VERDICT = _sty("v", fontName="Times-Roman", fontSize=12.5, leading=17) S_DIM = _sty("d", fontSize=8, textColor=DIM) S_CELL = _sty("c", fontSize=8.5, leading=11) _UNI = {"—": "-", "–": "-", "→": "->", "←": "<-", "≈": "~", "≥": ">=", "≤": "<=", "↑": "up", "↓": "down", "•": "-", "‘": "'", "’": "'", "“": '"', "”": '"', "₹": "Rs ", "…": "...", "⚠": "!", "✓": "OK", "✗": "x"} def _txt(x) -> str: s = str(x if x is not None else "") for u, a in _UNI.items(): s = s.replace(u, a) s = s.replace("&", "&").replace("<", "<").replace(">", ">") return s.encode("latin-1", "replace").decode("latin-1") # Helvetica-safe def _pct(x, signed=True): if not isinstance(x, (int, float)): return "-" return f"{x * 100:+.1f}%" if signed else f"{x * 100:.1f}%" def _table(data, widths, header=True, aligns=None): t = Table(data, colWidths=widths, repeatRows=1 if header else 0) style = [("VALIGN", (0, 0), (-1, -1), "TOP"), ("LINEBELOW", (0, 0), (-1, -2), 0.4, LINE), ("TOPPADDING", (0, 0), (-1, -1), 4), ("BOTTOMPADDING", (0, 0), (-1, -1), 4)] if header: style += [("FONT", (0, 0), (-1, 0), "Helvetica-Bold", 8), ("TEXTCOLOR", (0, 0), (-1, 0), DIM), ("LINEBELOW", (0, 0), (-1, 0), 0.8, BRASS)] for i, a in enumerate(aligns or []): style.append(("ALIGN", (i, 0), (i, -1), a)) t.setStyle(TableStyle(style)) return t def build_pdf(res: dict) -> bytes: buf = io.BytesIO() doc = SimpleDocTemplate(buf, pagesize=A4, leftMargin=16 * mm, rightMargin=16 * mm, topMargin=16 * mm, bottomMargin=16 * mm, title="Vitals — Portfolio Verdict") E = [] E.append(Paragraph("Vitals", S_TITLE)) E.append(Paragraph("Five analysts. One verdict.", S_TAG)) meta = f"Generated {_txt(res.get('generated_at'))}" ctx = res.get("context") or {} if ctx: meta += " · Context: " + ", ".join(f"{k}: {v}" for k, v in ctx.items()) E.append(Paragraph(_txt(meta), S_DIM)) E.append(Spacer(1, 8)) E.append(Paragraph(f"Portfolio health: {res.get('health_score', '?')}/100", _sty("hs", fontName="Helvetica-Bold", fontSize=13))) hb = res.get("health_breakdown") or {} if hb.get("pillars"): E.append(Paragraph(_txt(" · ".join(f"{p['label']} {p['score']:.0f}" for p in hb["pillars"])), S_DIM)) E.append(Spacer(1, 3)) E.append(Spacer(1, 3)) E.append(Paragraph(_txt(res.get("verdict", "")), S_VERDICT)) ac = res.get("action_counts") or {} strip = " · ".join(f"{ac[a]} {a}" for a in ("Buy More", "Keep", "Trim", "Sell") if ac.get(a)) if res.get("total_pl_pct") is not None: strip += f" · Portfolio P/L {_pct(res['total_pl_pct'])}" E.append(Spacer(1, 4)) E.append(Paragraph(strip, _sty("strip", fontName="Helvetica-Bold", fontSize=9.5, textColor=DIM))) ol0 = (res.get("plan") or {}).get("outlook") or {} cur0, pl0 = ol0.get("current") or {}, ol0.get("planned") or {} if cur0 and pl0: # the payoff line belongs on page 1, not buried mid-report E.append(Spacer(1, 3)) E.append(Paragraph( f"If you follow the plan: est. health {cur0.get('est_health')} -> " f"{pl0.get('est_health')}" + (f", model 12-mo edge {_pct(cur0.get('expected_excess_12m'))} -> " f"{_pct(pl0.get('expected_excess_12m'))}" if cur0.get("expected_excess_12m") is not None and pl0.get("expected_excess_12m") is not None else "") + (f", beta {cur0.get('beta')} -> {pl0.get('beta')}" if pl0.get("beta") else ""), S_BODY)) E.append(Paragraph("KEY FINDINGS", S_H2)) for h in res.get("highlights", []): c = _SEV_COLOR.get(str(h.get("severity")), DIM) E.append(Paragraph( f'■  {_txt(h.get("title"))} — ' f'{_txt(h.get("detail"))} ({_txt(h.get("lens", "portfolio"))} lens)', S_BODY)) E.append(Spacer(1, 3)) E.append(Paragraph("RECOMMENDATIONS", S_H2)) rows: list[list] = [["Holding", "Weight", "P/L", "vs intrinsic", "Action", "Why"]] for r in res.get("holdings", []): c = _ACTION_COLOR.get(r.get("action"), DIM) rows.append([ Paragraph(f"{_txt(r['ticker'])}
{_txt(r.get('name'))}", S_CELL), _pct(r.get("weight"), signed=False), _pct(r.get("pl_pct")), _pct(r.get("upside")), Paragraph(f"{_txt(r.get('action'))}" + ("
watch" if r.get("watch") else ""), S_CELL), Paragraph(_txt(r.get("rationale")), S_CELL)]) E.append(_table(rows, [26 * mm, 14 * mm, 18 * mm, 18 * mm, 16 * mm, 84 * mm], aligns=["LEFT", "RIGHT", "RIGHT", "RIGHT", "LEFT", "LEFT"])) E.append(Paragraph("PORTFOLIO REVIEW", S_H2)) st = res.get("strategist") or {} if st.get("strategy_read"): E.append(Paragraph(f"Strategy read: {_txt(st['strategy_read'])}", S_BODY)) if st.get("style_tilt"): E.append(Paragraph(f"Style tilt: {_txt(st['style_tilt'])}", S_BODY)) if st.get("gaps"): E.append(Paragraph("Missing exposures: " + _txt(", ".join(map(str, st["gaps"]))), S_BODY)) E.append(Spacer(1, 5)) for c0 in res.get("risk_scorecard", []): c = _SEV_COLOR.get(str(c0.get("status")), DIM) E.append(Paragraph( f"[{_txt(c0.get('status'))}] " f"{_txt(c0.get('check'))} — {_txt(c0.get('detail'))}", S_BODY)) E.append(Spacer(1, 2)) sa = res.get("sector_allocation") or {} if sa: E.append(Spacer(1, 4)) E.append(Paragraph("Sector mix: " + _txt(" · ".join( f"{s} {w * 100:.0f}%" for s, w in sa.items())), S_BODY)) rk = res.get("risk") or {} if rk: bn = (res.get("benchmark") or {}).get("name") or "benchmark" E.append(Paragraph("RISK ANALYTICS", S_H2)) if rk.get("stress"): rows_s: list[list] = [["Stress replay (today's weights)", "Window", "Portfolio", bn]] for s in rk["stress"]: rows_s.append([_txt(s["name"]), _txt(str(s["window"]).replace("→", "to").replace("->", "to")), _pct(s["portfolio"]), _pct(s["benchmark"]) if s.get("benchmark") is not None else "—"]) E.append(_table(rows_s, [55 * mm, 55 * mm, 25 * mm, 25 * mm], aligns=["LEFT", "LEFT", "RIGHT", "RIGHT"])) E.append(Spacer(1, 4)) bits = [] if rk.get("var_1d_95") is not None: bits.append(f"VaR (95%): 1-day {rk['var_1d_95'] * 100:.1f}%, 1-month {rk['var_1m_95'] * 100:.1f}%" + (f", expected shortfall {rk['cvar_1d_95'] * 100:.1f}%" if rk.get("cvar_1d_95") else "")) if rk.get("tracking_error"): bits.append(f"Tracking error vs {_txt(bn)}: {rk['tracking_error'] * 100:.1f}%") if rk.get("currency_exposure"): bits.append("Currency: " + _txt(", ".join(f"{c} {w * 100:.0f}%" for c, w in rk["currency_exposure"].items()))) lt = rk.get("lookthrough") or {} if lt.get("top_names"): bits.append("Fund look-through (top effective names): " + _txt(", ".join( f"{n} {tot * 100:.1f}%" for n, _, _, tot in lt["top_names"][:6]))) ft = rk.get("factor_tilts") or {} if ft.get("tilts"): bits.append("Style tilts: " + _txt(", ".join(f"{k} {v:+.2f}" for k, v in ft["tilts"].items()))) fb = rk.get("factor_betas") or {} if fb.get("betas"): bits.append(f"Measured factor betas (market beta {fb['market_beta']}): " + _txt(", ".join(f"{k} {v:+.2f}" for k, v in fb["betas"].items()))) br = rk.get("benchmark_relative") or {} if br: bits.append(f"Vs {_txt(bn)}: " + _txt(", ".join(f"{k.replace('_', ' ')} {v}" for k, v in br.items()))) if rk.get("active_sectors"): bits.append("Active sector bets: " + _txt(", ".join( f"{x['sector']} {x['active'] * 100:+.0f}pp" for x in rk["active_sectors"][:6]))) if rk.get("liquidity"): bits.append("Liquidity: " + _txt("; ".join( f"{x['ticker']} ~{x['days_to_liquidate']} days to exit" for x in rk["liquidity"][:4]))) for b in bits: E.append(Paragraph(b, S_BODY)) E.append(Spacer(1, 2)) alts = res.get("alternatives") or [] if alts: E.append(Paragraph("WORTH A LOOK — GAP FILLERS", S_H2)) needs = res.get("portfolio_needs") or {} if needs: E.append(Paragraph("What this portfolio needs: " + _txt( "; ".join(v["label"] for v in needs.values() if v.get("label"))), S_BODY)) E.append(Spacer(1, 3)) for g in (res.get("plan") or {}).get("gap_search") or []: line = (f"{_txt(g['sector'])}: {g['screened']} names screened — " + (f"buyable today: {_txt(', '.join(g['buyable']))}" if g.get("buyable") else "none buyable today") + (f" ({_txt(g['why_not'])})" if g.get("why_not") else "") + (f"; on the bench with entry conditions: {_txt(', '.join(g['bench']))}" if g.get("bench") else "")) E.append(Paragraph(line, S_DIM)); E.append(Spacer(1, 2)) for a in alts: fit = a.get("fit") or {} com = a.get("committee") or {} sig = " · ".join((fit.get("fixes") or []) + [f"CAUTION: {x}" for x in (fit.get("cautions") or [])] + ([f"committee {com['conviction']}/100 on adding"] if com.get("conviction") else []) + (a.get("signals") or [])) E.append(Paragraph( f"{_txt(a.get('ticker'))} — {_txt(a.get('reason'))}" + (f" [{_txt(sig)}]" if sig else ""), S_BODY)) E.append(Spacer(1, 3)) plan = res.get("plan") or {} if plan: E.append(Paragraph("REVIVAL PLAN" if plan.get("mode") == "revival" else "ENHANCEMENT PLAN", S_H2)) E.append(Paragraph(_txt(plan.get("pitch")), S_VERDICT)) E.append(Spacer(1, 5)) for i, m in enumerate(plan.get("moves") or [], 1): E.append(Paragraph(f"{i}. {_txt(m.get('title'))} — {_txt(m.get('detail'))}", S_BODY)) E.append(Spacer(1, 3)) orders = plan.get("orders") or [] if orders: E.append(Spacer(1, 5)) rows2: list[list] = [["", "Holding", "Size", "Approx."]] for o in orders: amt = (f"${o['amount_usd']:,.0f}" + (f" · ~{o['shares']} sh" if o.get("shares") else "") ) if o.get("amount_usd") else "-" if o.get("cost_note"): amt += f" (+{o['cost_note']})" rows2.append([o["side"], o["ticker"] + (" (new)" if o.get("is_new") else ""), f"{abs(o['delta_pct']) * 100:.1f}% of book", amt]) E.append(_table(rows2, [18 * mm, 42 * mm, 40 * mm, 45 * mm], aligns=["LEFT", "LEFT", "RIGHT", "RIGHT"])) if plan.get("cash"): E.append(Spacer(1, 3)) E.append(Paragraph(f"Unallocated cash: {plan['cash'] * 100:.1f}% of the book kept as dry powder for the " f"watchlist — the committee's ideas could not absorb every sale at sensible sizes.", S_BODY)) if plan.get("do_nothing"): E.append(Spacer(1, 4)) E.append(Paragraph("Do nothing. Nothing clears the hurdle for a trade; inactivity is the recommendation until a sell trigger fires.", S_BODY)) desk = [] for x in plan.get("tax_plan") or []: desk.append(f"Tax plan {_txt(x['ticker'])}: {_txt(x['note'])}") ra, rb = plan.get("risk_contributions_after") or {}, plan.get("risk_contributions") or {} if ra.get("top"): desk.append("Where the risk sits after the trade: " + _txt(", ".join(f"{t} {p * 100:.0f}%" for t, p in ra["top"][:5])) + (" (today: " + _txt(", ".join(f"{t} {p * 100:.0f}%" for t, p in rb["top"][:5])) + ")" if rb.get("top") else "")) s12 = plan.get("scenarios_12m") or {} if (s12.get("current") or {}).get("base") is not None: c, pl = s12["current"], s12.get("planned") or {} desk.append(f"Valuation risk/reward (where the book would trade if every holding converged to its bear / base / bull value, probability-weighted, {c['coverage'] * 100:.0f}% of the book): " f"today bear {c['bear'] * 100:+.0f}% / base {c['base'] * 100:+.0f}% / bull {c['bull'] * 100:+.0f}%" + (f"; after the plan bear {pl['bear'] * 100:+.0f}% / base {pl['base'] * 100:+.0f}% / bull {pl['bull'] * 100:+.0f}%" if pl.get("base") is not None else "")) if desk: E.append(Spacer(1, 5)) E.append(Paragraph("Execution — the manager's desk", S_BODY)) for b in desk: E.append(Paragraph(b, S_BODY)); E.append(Spacer(1, 2)) wl = plan.get("watchlist") or [] if wl: E.append(Spacer(1, 4)) E.append(Paragraph("Watchlist — the bench, with entry conditions", S_BODY)) rows_w: list[list] = [["Ticker", "Entry condition", "Why it is on the bench"]] for x in wl: rows_w.append([_txt(x["ticker"]), Paragraph(_txt(x["condition"]), S_CELL), Paragraph(_txt(x.get("why")), S_CELL)]) E.append(_table(rows_w, [22 * mm, 75 * mm, 75 * mm])) cats = plan.get("catalysts") or [] if cats: E.append(Spacer(1, 4)) E.append(Paragraph("Catalysts — next 90 days: " + _txt(", ".join( f"{x['ticker']} {x['event']} {x['date']}" for x in cats)), S_BODY)) trip = [(r["ticker"], (r.get("us") or {}).get("sell_triggers")) for r in res.get("holdings", []) if r.get("action") != "Sell" and (r.get("us") or {}).get("sell_triggers")] if trip: E.append(Spacer(1, 4)) E.append(Paragraph("Tripwires — what would make us sell", S_BODY)) rows_t: list[list] = [["Holding", "Tripwire"]] for t, tr in trip[:14]: rows_t.append([_txt(t), Paragraph(_txt(tr[0]), S_CELL)]) E.append(_table(rows_t, [22 * mm, 150 * mm])) for hp in plan.get("harvest_pairs") or []: E.append(Spacer(1, 3)) E.append(Paragraph(f"Tax-loss harvest: sell {_txt(hp['sell'])}, buy {_txt(hp['buy'])} ({_txt(hp['buy_name'])}, correlation {hp['correlation']}) - {_txt(hp['note'])}", S_BODY)) pmx = res.get("premortem") if pmx and pmx.get("story"): E.append(Spacer(1, 4)) E.append(Paragraph(f"Pre-mortem ({_txt(pmx['ticker'])}): {_txt(pmx['story'])}", S_BODY)) tri = plan.get("triage") or [] if tri: E.append(Spacer(1, 6)) E.append(Paragraph("Losing-position triage", S_BODY)) E.append(Spacer(1, 2)) rows3: list[list] = [["Holding", "Call", "Why"]] for t in tri: rows3.append([_txt(t.get("ticker")), _txt(t.get("disposition")), Paragraph(_txt(t.get("note")), S_CELL)]) E.append(_table(rows3, [22 * mm, 28 * mm, 122 * mm])) target = res.get("target_portfolio") or [] if target: E.append(Paragraph("RECOMMENDED TARGET PORTFOLIO", S_H2)) E.append(Paragraph("Committee conviction sizes positions under risk-profile caps. " "Educational illustration, not advice.", S_DIM)) E.append(Spacer(1, 4)) rows = [["Holding", "Sector", "Current", "Target", "Change"]] for r in target: name = r["ticker"] + (" (new)" if r.get("is_new") else "") rows.append([_txt(name), _txt(r.get("sector")), _pct(r["current_w"], signed=False), _pct(r["target_w"], signed=False), f"{r['delta'] * 100:+.1f}pp"]) E.append(_table(rows, [38 * mm, 55 * mm, 22 * mm, 22 * mm, 22 * mm], aligns=["LEFT", "LEFT", "RIGHT", "RIGHT", "RIGHT"])) if res.get("llm_used"): # one line per holding — the old prose dump ran ~7 pages on a 47-name book and # nobody reads it; the full reasoning stays on screen in the expandable rows E.append(Paragraph("APPENDIX — COMMITTEE SCORES AT A GLANCE", S_H2)) E.append(Paragraph("Conviction 0-100 per lens (50 = neutral). Full reasoning and the " "headlines weighed are in the app under each holding's row.", S_DIM)) E.append(Spacer(1, 4)) lens_order = ("value", "growth", "macro", "technical", "risk") rows4: list[list] = [["Holding", "Action", "Val", "Gro", "Mac", "Tec", "Rsk", "vs intr.", "Committee's top concern"]] for r in res.get("holdings", []): scores = r.get("lens_scores") or {} cells = [f"{scores[k]:.0f}" if k in scores else "-" for k in lens_order] concern = (r.get("risks") or [""])[0] c = _ACTION_COLOR.get(r.get("action"), DIM) rows4.append([ _txt(r["ticker"]), Paragraph(f"{_txt(r.get('action'))}", S_CELL), *cells, _pct(r.get("upside")), Paragraph(_txt(concern)[:140], S_CELL)]) E.append(_table(rows4, [18 * mm, 17 * mm, 9 * mm, 9 * mm, 9 * mm, 9 * mm, 9 * mm, 15 * mm, 77 * mm], aligns=["LEFT", "LEFT", "RIGHT", "RIGHT", "RIGHT", "RIGHT", "RIGHT", "RIGHT", "LEFT"])) ds = res.get("data_sources") or [] if ds: diag = res.get("diagnostics") or {} E.append(Paragraph("DATA SOURCES & METHODOLOGY", S_H2)) E.append(Paragraph(_txt( f"Benchmark {(res.get('benchmark') or {}).get('name', 'S&P 500')} · base currency " f"{res.get('base_currency', 'USD')} · methodology {res.get('methodology_version', '?')} · " f"prompt {diag.get('prompt_version', '?')} · models " + ", ".join(f"{k} {v}" for k, v in (diag.get("models") or {}).items() if v)), S_DIM)) E.append(Spacer(1, 3)) rows_d: list[list] = [["Holding", "Fundamentals", "As of", "Price as of"]] for d in ds: rows_d.append([_txt(d["ticker"]), _txt(d.get("fundamentals") or "?"), _txt(d.get("fund_asof") or "—"), _txt(d.get("price_asof") or "—")]) E.append(_table(rows_d, [30 * mm, 50 * mm, 35 * mm, 45 * mm], aligns=["LEFT", "LEFT", "LEFT", "LEFT"])) E.append(Spacer(1, 10)) E.append(Paragraph(_txt(res.get("disclaimer", "")), S_DIM)) doc.build(E) return buf.getvalue()