vitals / src /pdf_report.py
s-ttp's picture
Sector-gap sourcing six deep, gap-driven second batch, gap-covering shortlist, per-sector search record, narrator share rules
06f8709 verified
Raw History Blame Contribute Delete
21.1 kB
"""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("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
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: <b>{res.get('health_score', '?')}/100</b>",
_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"<b>If you follow the plan:</b> est. health {cur0.get('est_health')} -> "
f"<b>{pl0.get('est_health')}</b>"
+ (f", model 12-mo edge {_pct(cur0.get('expected_excess_12m'))} -> "
f"<b>{_pct(pl0.get('expected_excess_12m'))}</b>"
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'<font color="#{c.hexval()[2:]}">&#9632;</font>&nbsp; <b>{_txt(h.get("title"))}</b> — '
f'{_txt(h.get("detail"))} <font size="7" color="#{DIM.hexval()[2:]}">({_txt(h.get("lens", "portfolio"))} lens)</font>',
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"<b>{_txt(r['ticker'])}</b><br/><font size='7' color='#{DIM.hexval()[2:]}'>{_txt(r.get('name'))}</font>", S_CELL),
_pct(r.get("weight"), signed=False), _pct(r.get("pl_pct")), _pct(r.get("upside")),
Paragraph(f"<b><font color='#{c.hexval()[2:]}'>{_txt(r.get('action'))}</font></b>"
+ ("<br/><font size='6'>watch</font>" 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"<b>Strategy read:</b> {_txt(st['strategy_read'])}", S_BODY))
if st.get("style_tilt"):
E.append(Paragraph(f"<b>Style tilt:</b> {_txt(st['style_tilt'])}", S_BODY))
if st.get("gaps"):
E.append(Paragraph("<b>Missing exposures:</b> " + _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"<font color='#{c.hexval()[2:]}'><b>[{_txt(c0.get('status'))}]</b></font> "
f"<b>{_txt(c0.get('check'))}</b> — {_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("<b>Sector mix:</b> " + _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"<b>VaR (95%):</b> 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"<b>Tracking error vs {_txt(bn)}:</b> {rk['tracking_error'] * 100:.1f}%")
if rk.get("currency_exposure"):
bits.append("<b>Currency:</b> " + _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("<b>Fund look-through (top effective names):</b> " + _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("<b>Style tilts:</b> " + _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"<b>Measured factor betas (market beta {fb['market_beta']}):</b> "
+ _txt(", ".join(f"{k} {v:+.2f}" for k, v in fb["betas"].items())))
br = rk.get("benchmark_relative") or {}
if br:
bits.append(f"<b>Vs {_txt(bn)}:</b> " + _txt(", ".join(f"{k.replace('_', ' ')} {v}" for k, v in br.items())))
if rk.get("active_sectors"):
bits.append("<b>Active sector bets:</b> " + _txt(", ".join(
f"{x['sector']} {x['active'] * 100:+.0f}pp" for x in rk["active_sectors"][:6])))
if rk.get("liquidity"):
bits.append("<b>Liquidity:</b> " + _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("<b>What this portfolio needs:</b> " + _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"<b>{_txt(g['sector'])}:</b> {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"<b>{_txt(a.get('ticker'))}</b> — {_txt(a.get('reason'))}"
+ (f" <font size='7' color='#{DIM.hexval()[2:]}'>[{_txt(sig)}]</font>" 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"<b>{i}. {_txt(m.get('title'))}</b> — {_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"<b>Unallocated cash:</b> {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("<b>Do nothing.</b> 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"<b>Tax plan {_txt(x['ticker'])}:</b> {_txt(x['note'])}")
ra, rb = plan.get("risk_contributions_after") or {}, plan.get("risk_contributions") or {}
if ra.get("top"):
desk.append("<b>Where the risk sits after the trade:</b> " + _txt(", ".join(f"{t} {p * 100:.0f}%" for t, p in ra["top"][:5]))
+ (" <font size='7'>(today: " + _txt(", ".join(f"{t} {p * 100:.0f}%" for t, p in rb["top"][:5])) + ")</font>" 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"<b>Valuation risk/reward</b> (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("<b>Execution — the manager's desk</b>", 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("<b>Watchlist — the bench, with entry conditions</b>", 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("<b>Catalysts — next 90 days:</b> " + _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("<b>Tripwires — what would make us sell</b>", 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"<b>Tax-loss harvest:</b> 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"<b>Pre-mortem ({_txt(pmx['ticker'])}):</b> {_txt(pmx['story'])}", S_BODY))
tri = plan.get("triage") or []
if tri:
E.append(Spacer(1, 6))
E.append(Paragraph("<b>Losing-position triage</b>", 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"<font color='#{c.hexval()[2:]}'><b>{_txt(r.get('action'))}</b></font>", 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()