| """Result payload assembly (what the summary screen renders) + the full downloadable |
| markdown report (where the per-holding persona reasoning lives — it is never shown on screen). |
| Plain-English framing throughout (deepquant jargon→English discipline). |
| """ |
| from __future__ import annotations |
|
|
| from datetime import datetime, timezone |
|
|
| from . import metrics as M |
| from .personas import PERSONAS |
|
|
| DISCLAIMER = ("Educational analysis tool — not investment advice. AI-generated content can be " |
| "wrong; verify independently and consult a licensed advisor before acting. " |
| "The intrinsic-value approach is inspired by Prof. Aswath Damodaran's published " |
| "valuation framework; he is not affiliated with and has not endorsed this tool.") |
|
|
|
|
| def _fallback_verdict(pm: dict, cards: list[dict]) -> str: |
| worst = next((c for c in cards if c["status"] == "ALERT"), |
| next((c for c in cards if c["status"] == "WATCH"), None)) |
| base = f"{pm['n_holdings']}-stock portfolio" |
| return f"{base} — biggest issue: {worst['detail']}" if worst else f"{base} — no major structural issues detected." |
|
|
|
|
| def _fallback_highlights(cards: list[dict]) -> list[dict]: |
| sev = {"OK": "good", "WATCH": "warn", "ALERT": "bad"} |
| ranked = sorted(cards, key=lambda c: {"ALERT": 0, "WATCH": 1, "OK": 2}[c["status"]]) |
| return [{"severity": sev[c["status"]], "title": c["check"], "detail": c["detail"], |
| "lens": "portfolio"} for c in ranked[:4]] |
|
|
|
|
| def assemble(holdings, pm, cards, decisions, panel, strat, cio, errors, news=None, |
| extra: dict | None = None) -> dict: |
| news = news or {} |
| cio = cio or {} |
| rationale = cio.get("rationale") or {} |
| hb = (extra or {}).get("health_breakdown") or {} |
| score = hb.get("score") if isinstance(hb.get("score"), int) else M.health_score(cards, pm, holdings, decisions) |
| action_counts: dict[str, int] = {} |
| rows = [] |
| for h in holdings: |
| t = h["symbol"] |
| dec = decisions[t] |
| action_counts[dec["action"]] = action_counts.get(dec["action"], 0) + 1 |
| dcf = h.get("dcf") or {} |
| lens_points = {k: v[t] for k, v in panel.items() if t in v} |
| memos = (cio or {}).get("memos") or {} |
| rows.append({ |
| "ticker": t, "name": h["name"], |
| "sector": ("Diversified fund" if str(h["data"].get("quote_type") or "").upper() in M.FUND_TYPES |
| else h["data"].get("sector")), |
| "weight": round(h["weight"], 4), "entry": h["entry"], "price": h["price"], |
| "pl_pct": round(h["pl_pct"], 4), "qty": h.get("qty"), |
| "n_lots": h.get("n_lots") or 1, "lots": h.get("lots"), "lot_note": h.get("lot_note"), |
| "entry_note": h.get("entry_note"), |
| "action": dec["action"], "conviction": dec["conviction"], "watch": bool(dec.get("watch")), |
| "dispersion": dec["dispersion"], "n_lenses": dec["n_lenses"], |
| "rationale": str(rationale.get(t, ""))[:300] or _row_fallback(dec, dcf), |
| "intrinsic": dcf.get("intrinsic"), "upside": dcf.get("upside"), |
| "intrinsic_low": dcf.get("intrinsic_low"), "intrinsic_high": dcf.get("intrinsic_high"), |
| "implied_growth": dcf.get("implied_growth"), "valuation_model": dcf.get("model"), |
| "dcf_note": dcf.get("note"), |
| "tax_note": dec.get("tax_note"), "size_note": dec.get("size_note"), |
| "peer": {k: round((v + 1) / 2 * 100) for k, v in |
| ((h["data"].get("panel") or {}).get("factors") or {}).items() |
| if k in ("n_earnings_yield", "n_evebitda", "n_roe", "n_netmargin", "n_mom_12_1")}, |
| "peer_universe": (h["data"].get("panel") or {}).get("universe") or "US", |
| "us": h["data"].get("us") or None, "asymmetry": h.get("asymmetry"), |
| "memo": (memos or {}).get(t), "sizing": h.get("sizing"), |
| "data_source": h["data"].get("fund_source"), "fund_asof": h["data"].get("fund_asof"), |
| "price_asof": h["data"].get("price_asof"), "data_flags": h["data"].get("data_flags") or [], |
| "lifecycle": dcf.get("lifecycle"), "flags": dcf.get("flags") or [], |
| "strengths": [v["key_point"] for v in lens_points.values() if v.get("key_point")][:3], |
| "risks": [v["key_risk"] for v in lens_points.values() if v.get("key_risk")][:3], |
| "lens_scores": dec["scores"], |
| "news": [{"headline": n.get("headline", ""), "sentiment": n.get("sentiment", "")} |
| for n in news.get(t, [])[:4]], |
| }) |
| rows.sort(key=lambda r: -r["weight"]) |
| alternatives = [] |
| for a in (cio.get("alternatives") or [])[:4]: |
| if isinstance(a, dict) and a.get("ticker"): |
| alternatives.append({"ticker": str(a["ticker"]).upper()[:12], |
| "reason": str(a.get("reason", ""))[:220]}) |
| highlights = cio.get("highlights") if isinstance(cio.get("highlights"), list) else None |
| result = { |
| "generated_at": datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC"), |
| "health_score": score, |
| "verdict": str(cio.get("verdict", ""))[:300] or _fallback_verdict(pm, cards), |
| "action_counts": action_counts, |
| "total_pl_pct": pm.get("total_pl_pct"), |
| "highlights": (highlights or _fallback_highlights(cards))[:5], |
| "holdings": rows, |
| "sector_allocation": pm.get("sector_allocation", {}), |
| "sector_band": [M.SECTOR_CAP_LO, M.SECTOR_CAP_HI], |
| "name_cap": M.MAX_POS_W, |
| "risk_scorecard": cards, |
| "metrics": {k: pm.get(k) for k in |
| ("effective_n", "weighted_beta", "realized_beta", "portfolio_vol", |
| "sharpe_1y", "return_1y", "spy_return_1y", "max_drawdown_1y", |
| "equal_weighted", "regime_risk_on")}, |
| "strategist": strat or {}, |
| "alternatives": alternatives, |
| "input_errors": errors, |
| "llm_used": bool(panel), |
| "disclaimer": DISCLAIMER, |
| } |
| result.update(extra or {}) |
| result["report_md"] = _markdown(result, panel) |
| return result |
|
|
|
|
| def _row_fallback(dec, dcf) -> str: |
| if not dec["scores"]: |
| up = dcf.get("upside") |
| if isinstance(up, (int, float)): |
| side = "less" if up > 0 else "more" |
| return (f"The market is paying {abs(up) * 100:.0f}% {side} than our estimate of " |
| f"what the business is worth.") |
| return "Quant view only — no AI panel available for this run." |
| c = dec["conviction"] |
| mood = ("the analysts see serious problems here" if c < 35 else |
| "the analysts found more concerns than positives" if c < 45 else |
| "the analysts' views are balanced on this one" if c <= 62 else |
| "the analysts see clear positives outweighing the risks") |
| return f"On balance, {mood} — open the row for the detail." |
|
|
|
|
| def _pct(x): |
| return f"{x * 100:+.1f}%" if isinstance(x, (int, float)) else "n/a" |
|
|
|
|
| def _markdown(res: dict, panel: dict) -> str: |
| L = ["# Vitals — Portfolio Verdict", "*Five analysts. One verdict.*", |
| f"*Generated {res['generated_at']}*", "", |
| f"**Portfolio health: {res['health_score']}/100** — {res['verdict']}", ""] |
| deg = ((res.get("diagnostics") or {}).get("degraded") or []) |
| if deg: |
| L += ["*Note: " + "; ".join(deg) + ".*"] |
| hb = res.get("health_breakdown") or {} |
| if hb.get("pillars"): |
| L += ["**Health pillars:** " + " · ".join(f"{p['label']} {p['score']:.0f} ({p['weight'] * 100:.0f}%)" for p in hb["pillars"])] |
| for p in hb["pillars"]: |
| L.append(f"- {p['label']}: {p['note']}") |
| L += ["", |
| "## Key findings"] |
| for hl in res["highlights"]: |
| icon = {"good": "✅", "warn": "⚠️", "bad": "🔴"}.get(hl.get("severity"), "•") |
| L.append(f"- {icon} **{hl.get('title', '')}** — {hl.get('detail', '')} " |
| f"*({hl.get('lens', 'portfolio')} lens)*") |
| L += ["", "## Recommendations", "", |
| "| Holding | Weight | P/L | Intrinsic vs price | Action | Why |", |
| "|---|---|---|---|---|---|"] |
| for r in res["holdings"]: |
| L.append(f"| **{r['ticker']}** {r['name']} | {r['weight'] * 100:.0f}% | {_pct(r['pl_pct'])} " |
| f"| {_pct(r['upside'])} | **{r['action']}**{' · watch' if r.get('watch') else ''} | {r['rationale']} |") |
| L += ["", "## Portfolio review"] |
| st = res.get("strategist") or {} |
| if st.get("strategy_read"): |
| L.append(f"**Strategy read:** {st['strategy_read']}") |
| if st.get("style_tilt"): |
| L.append(f"**Style tilt:** {st['style_tilt']}") |
| if st.get("gaps"): |
| L.append("**Missing exposures:** " + ", ".join(map(str, st["gaps"]))) |
| L.append("") |
| L.append("### Risk scorecard") |
| for c in res["risk_scorecard"]: |
| icon = {"OK": "✅", "WATCH": "⚠️", "ALERT": "🔴"}[c["status"]] |
| L.append(f"- {icon} **{c['check']}** — {c['detail']}") |
| L += ["", "### Sector allocation"] |
| for s, w in res["sector_allocation"].items(): |
| L.append(f"- {s}: {w * 100:.0f}%") |
| rk = res.get("risk") or {} |
| if rk: |
| bn = (res.get("benchmark") or {}).get("name") or "benchmark" |
| L += ["", "## Risk analytics"] |
| if rk.get("stress"): |
| L += [f"**Stress replay (today's weights, vs {bn}):**"] |
| for s in rk["stress"]: |
| L.append(f"- {s['name']} ({s['window']}): portfolio {_pct(s['portfolio'])}" |
| + (f" vs {bn} {_pct(s['benchmark'])}" if s.get("benchmark") is not None else "") |
| + (f" (coverage {s['coverage'] * 100:.0f}%)" if s.get("coverage", 1) < 1 else "")) |
| if rk.get("var_1d_95") is not None: |
| L.append(f"**Value at risk (95%, historical):** 1-day {rk['var_1d_95'] * 100:.1f}% · " |
| f"1-month {rk['var_1m_95'] * 100:.1f}%" |
| + (f" · expected shortfall (1-day) {rk['cvar_1d_95'] * 100:.1f}%" if rk.get("cvar_1d_95") else "") |
| + (f" · tracking error vs {bn} {rk['tracking_error'] * 100:.1f}%" if rk.get("tracking_error") else "")) |
| if rk.get("currency_exposure"): |
| L.append("**Currency exposure:** " + ", ".join(f"{c} {w * 100:.0f}%" for c, w in rk["currency_exposure"].items())) |
| lt = rk.get("lookthrough") or {} |
| if lt.get("top_names"): |
| L.append(f"**Fund look-through (constituents disclosed for {lt['coverage'] * 100:.0f}% of fund weight):** " |
| + ", ".join(f"{n} {tot * 100:.1f}%" + (f" (direct {d * 100:.1f}% + via funds {v * 100:.1f}%)" if d and v else "") |
| for n, d, v, tot in lt["top_names"][:6])) |
| ft = rk.get("factor_tilts") or {} |
| if ft.get("tilts"): |
| L.append(f"**Style tilts (−1…+1, {ft['coverage'] * 100:.0f}% of book covered):** " |
| + ", ".join(f"{k} {v:+.2f}" for k, v in ft["tilts"].items())) |
| fb = rk.get("factor_betas") or {} |
| if fb.get("betas"): |
| L.append(f"**Measured factor betas (vs {bn}, market beta {fb['market_beta']}):** " |
| + ", ".join(f"{k} {v:+.2f}" for k, v in fb["betas"].items())) |
| br = rk.get("benchmark_relative") or {} |
| if br: |
| L.append(f"**Vs {bn}:** " + ", ".join(f"{k.replace('_', ' ')} {v}" for k, v in br.items())) |
| if rk.get("active_sectors"): |
| L.append(f"**Active sector bets vs {bn}:** " + ", ".join( |
| f"{x['sector']} {x['active'] * 100:+.0f}pp ({x['book'] * 100:.0f}% vs {x['benchmark'] * 100:.0f}%)" for x in rk["active_sectors"][:6])) |
| if rk.get("liquidity"): |
| L.append("**Liquidity:** " + "; ".join(f"{x['ticker']} ≈ {x['days_to_liquidate']} days to exit at 20% of volume" |
| for x in rk["liquidity"][:5])) |
| usage = ((res.get("diagnostics") or {}).get("usage") or {}) |
| if usage.get("by_model"): |
| L += ["", "**Model usage this run:** " + "; ".join( |
| f"{m} {v['calls']} calls, {(v['input'] + v['cache_read'] + v['cache_write']) / 1000:.0f}k in / {v['output'] / 1000:.1f}k out (${v['cost_usd']:.2f})" |
| for m, v in usage["by_model"].items()) + f" — total ≈ ${usage.get('cost_usd', 0):.2f}"] |
| plan = res.get("plan") or {} |
| if plan: |
| L += ["", "## Execution — the manager's desk"] |
| if plan.get("cash"): |
| L.append(f"**Unallocated cash:** {plan['cash'] * 100:.1f}% of the book held as dry powder for the watchlist " |
| f"(the committee's ideas could not absorb every sale at sensible sizes).") |
| for x in plan.get("tax_plan") or []: |
| L.append(f"- **Tax plan {x['ticker']}:** {x['note']}") |
| for x in plan.get("watchlist") or []: |
| L.append(f"- **Watchlist {x['ticker']}** ({x.get('name')}): {x['condition']} — {x.get('why')}") |
| for x in plan.get("catalysts") or []: |
| L.append(f"- **Catalyst:** {x['ticker']} {x['event']} on {x['date']} (in {x['days']} days)") |
| ra = plan.get("risk_contributions_after") or {} |
| rb = plan.get("risk_contributions") or {} |
| if ra.get("top"): |
| L.append("**Where the risk sits after the trade:** " + ", ".join(f"{t} {p * 100:.0f}%" for t, p in ra["top"][:5]) |
| + (f" (today: " + ", ".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 {} |
| L.append(f"**Valuation risk/reward (where the book would trade if every holding converged to its bear / base / bull value, probability-weighted):** today " |
| f"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 "") |
| + f" ({c['coverage'] * 100:.0f}% of the book covered)") |
| trip = [(r["ticker"], (r.get("us") or {}).get("sell_triggers")) for r in res["holdings"] |
| if r.get("action") != "Sell" and (r.get("us") or {}).get("sell_triggers")] |
| if trip: |
| L.append("**Tripwires (what would make us sell):** " + "; ".join(f"{t}: {tr[0]}" for t, tr in trip[:12])) |
| if plan.get("harvest_pairs") or plan.get("do_nothing") or res.get("premortem"): |
| L += ["", "## Discipline"] |
| if plan.get("do_nothing"): |
| L.append("**Do nothing.** Nothing clears the hurdle for a trade; inactivity is the recommendation until a sell trigger fires.") |
| for h in plan.get("harvest_pairs") or []: |
| L.append(f"- **Tax-loss harvest:** sell {h['sell']}, buy {h['buy']} ({h['buy_name']}, correlation {h['correlation']}) — {h['note']}") |
| pm = res.get("premortem") |
| if pm and pm.get("story"): |
| L.append(f"**Pre-mortem ({pm['ticker']}):** {pm['story']}") |
| if res["alternatives"]: |
| L += ["", "## Suggested alternatives"] |
| needs = res.get("portfolio_needs") or {} |
| if needs: |
| L.append("*What this portfolio needs:* " + "; ".join(v["label"] for v in needs.values() if v.get("label"))) |
| for g in (res.get("plan") or {}).get("gap_search") or []: |
| L.append(f"- *{g['sector']}:* {g['screened']} names screened — " |
| + (f"buyable today: {', '.join(g['buyable'])}" if g.get("buyable") else "none buyable today") |
| + (f" ({g['why_not']})" if g.get("why_not") else "") |
| + (f"; on the bench: {', '.join(g['bench'])}" if g.get("bench") else "")) |
| for a in res["alternatives"]: |
| fit = a.get("fit") or {} |
| com = a.get("committee") or {} |
| L.append(f"- **{a['ticker']}** — {a['reason']}" |
| + (f" *(fixes: {'; '.join(fit['fixes'])})*" if fit.get("fixes") else "") |
| + (f" ⚠ {'; '.join(fit['cautions'])}" if fit.get("cautions") else "") |
| + (f" · committee {com['conviction']}/100 on adding" if com.get("conviction") else "")) |
| if panel: |
| L += ["", "## Appendix — analyst committee detail (per holding)"] |
| for r in res["holdings"]: |
| t = r["ticker"] |
| L += ["", f"### {t} — {r['name']} · {r['action']} (conviction {r['conviction']}/100" |
| + (f", dispersion {r['dispersion']}" if r["dispersion"] else "") + ")"] |
| if r.get("intrinsic"): |
| L.append(f"Intrinsic value {r['intrinsic']} ({_pct(r['upside'])} vs price, " |
| f"{r.get('valuation_model') or 'DCF'}, lifecycle: {r.get('lifecycle') or '?'})" |
| + (f" · bear–bull {r['intrinsic_low']}–{r['intrinsic_high']}" if r.get("intrinsic_low") else "") |
| + (f" · price implies {r['implied_growth'] * 100:+.0f}%/yr growth" if isinstance(r.get("implied_growth"), (int, float)) else "") |
| + (f" · guards: {', '.join(r['flags'])}" if r["flags"] else "")) |
| if r.get("memo"): |
| m = r["memo"] |
| L.append(f"Memo — thesis: {m.get('thesis')} · debate: {m.get('debate')} · would change our mind: {m.get('change_mind')}") |
| if r.get("asymmetry"): |
| a = r["asymmetry"] |
| L.append(f"Payoff: bear {_pct(a['bear'])} / base {_pct(a['base'])} / bull {_pct(a['bull'])} → expected {_pct(a['expected_return'])}, " |
| f"upside:downside {a['up_down_ratio']}:1 ({'clears' if a['clears_hurdle'] else 'below'} the 2:1 / +8% hurdle)" |
| + (f" · half-Kelly size {r['sizing']['half_kelly'] * 100:.0f}% vs {r['sizing']['current'] * 100:.0f}% held" if r.get("sizing") else "")) |
| if r.get("us"): |
| from .fundamentals import us_lines |
| for line in us_lines(r["us"]): |
| L.append(line) |
| if r["us"].get("sell_triggers"): |
| L.append("Sell triggers: " + "; ".join(r["us"]["sell_triggers"])) |
| if r.get("tax_note"): |
| L.append(f"Tax: {r['tax_note']}") |
| if r.get("data_flags"): |
| L.append(f"Data quality: {'; '.join(r['data_flags'])}") |
| for k, scores in panel.items(): |
| if t in scores: |
| p = PERSONAS[k] |
| v = scores[t] |
| L.append(f"- **{p['label']}** ({v['conviction']:.0f}/100): {v.get('key_point', '')}" |
| + (f" — *risk: {v['key_risk']}*" if v.get("key_risk") else "")) |
| ds = res.get("data_sources") or [] |
| if ds: |
| L += ["", "## Data sources & methodology", |
| f"Benchmark: {(res.get('benchmark') or {}).get('name', 'S&P 500')} · " |
| f"Base currency: {res.get('base_currency', 'USD')} · " |
| f"Methodology version {res.get('methodology_version', '?')} · " |
| f"Prompt version {(res.get('diagnostics') or {}).get('prompt_version', '?')} · " |
| f"Models: {', '.join(f'{k} {v}' for k, v in ((res.get('diagnostics') or {}).get('models') or {}).items() if v)}", |
| "", "| Holding | Fundamentals source | As of | Price as of |", "|---|---|---|---|"] |
| for d in ds: |
| L.append(f"| {d['ticker']} | {d.get('fundamentals') or '?'} | {d.get('fund_asof') or '—'} | {d.get('price_asof') or '—'} |") |
| L += ["", "---", f"*{res['disclaimer']}*"] |
| return "\n".join(L) |
|
|
|
|
| _CTX_LABELS = {"horizon": "Investment horizon", "risk": "Risk appetite", "objective": "Objective"} |
|
|
|
|
| def append_target_md(md: str, target: list[dict], context: dict) -> str: |
| L = [md, "", "## Recommended target portfolio", |
| "*Based on your context: " + "; ".join( |
| f"{_CTX_LABELS.get(k, k)}: {v}" for k, v in context.items() if v) + ".*", |
| "*Deterministic construction: committee conviction sizes positions under " |
| "risk-profile name/sector caps. Educational illustration, not advice.*", "", |
| "| Holding | Current | Target | Change |", "|---|---|---|---|"] |
| for r in target: |
| tag = " *(new)*" if r["is_new"] else "" |
| L.append(f"| {r['ticker']}{tag} | {r['current_w'] * 100:.1f}% | {r['target_w'] * 100:.1f}% " |
| f"| {r['delta'] * 100:+.1f}pp |") |
| return "\n".join(L) |
|
|
