| """Deterministic portfolio math — no LLM anywhere in this module. |
| |
| Reused: slipstream/book.py (_name_cap 15% justification, sector-cap bands, _regime_on), |
| marketfm_space _curve (Sharpe/drawdown), follow_the_money_backtest (beta via covariance), |
| slipstream/services/supervisor.py (OK/WATCH/ALERT scorecard pattern), |
| crucible/engine/damodaran_portfolio.py (2-stage DCF math + trap flags — pure functions only). |
| New here (existed nowhere): correlation clusters, HHI/effective-N, weighted beta, health score. |
| """ |
| from __future__ import annotations |
|
|
| import numpy as np |
| import pandas as pd |
|
|
| MAX_POS_W = 0.15 |
| FUND_CAP = 0.60 |
| FUND_TYPES = ("ETF", "MUTUALFUND", "INDEX", "FUND") |
|
|
|
|
| def effective_name_cap(n: int) -> float: |
| """Per-name cap a book of n names can actually satisfy. A 15% cap needs ≥7 names — |
| judging a 6-stock book against it produced "trim everything" (every position is |
| ≥16.7% by construction). Small books are judged on a proportional cap, and the |
| breadth check carries the real message: add names.""" |
| return min(0.30, max(MAX_POS_W, 1.25 / max(n, 1))) |
| SECTOR_CAP_BASE, SECTOR_CAP_LO, SECTOR_CAP_HI = 0.30, 0.15, 0.45 |
| CORR_ALERT = 0.80 |
|
|
| |
| RF, ERP, G_STABLE = 0.045, 0.045, 0.025 |
| SECTOR_BETA = {"Technology": 1.15, "Consumer Cyclical": 1.10, "Financial Services": 1.10, |
| "Industrials": 1.05, "Healthcare": 0.95, "Energy": 1.10, "Basic Materials": 1.05, |
| "Consumer Defensive": 0.75, "Communication Services": 1.00, "Utilities": 0.55, |
| "Real Estate": 0.90} |
|
|
|
|
| |
|
|
| def dcf_from_assumptions(eps_base: float | None, price: float | None, A: dict | None): |
| """Deterministic 2-stage DCF driven by LLM-Damodaran assumptions (auditable math). |
| Returns (intrinsic, upside, ke, lifecycle) or (nan, nan, nan, None).""" |
| if not A or not eps_base or eps_base <= 0 or not price or price <= 0: |
| return float("nan"), float("nan"), float("nan"), None |
| ke = float(np.clip(A.get("cost_of_equity", 0.09), 0.06, 0.16)) |
| g1 = float(np.clip(A.get("high_growth_rate", 0.05), -0.30, 0.60)) |
| N = int(np.clip(A.get("high_growth_years", 5), 1, 12)) |
| gt = float(np.clip(A.get("terminal_growth", 0.025), 0.0, 0.045)) |
| roic = float(np.clip(A.get("reinvestment_roic", 0.15), 0.05, 0.40)) |
| if g1 > 0: |
| |
| |
| |
| roic = max(roic, min(g1 / 0.8, 0.60)) |
| eps = eps_base * float(np.clip(A.get("normalized_eps_factor", 1.0), 0.5, 1.5)) |
| if ke <= gt: |
| return float("nan"), float("nan"), ke, A.get("lifecycle_stage") |
| pv = 0.0 |
| for t in range(1, N + 1): |
| g = g1 + (gt - g1) * (t - 1) / max(N - 1, 1) |
| eps *= (1 + g) |
| po = float(np.clip(1 - g / roic, 0.0, 1.0)) |
| pv += (eps * po) / (1 + ke) ** t |
| po_t = float(np.clip(1 - gt / roic, 0.30, 1.0)) |
| pv += (eps * (1 + gt) * po_t) / (ke - gt) / (1 + ke) ** N |
| up = float(np.clip(pv / price - 1, -0.90, 3.0)) |
| return pv, up, ke, A.get("lifecycle_stage") |
|
|
|
|
| def growth_stage_value(d: dict, price) -> dict | None: |
| """Loss-making growth names have no EPS to discount, so the DCF column stayed blank for |
| SNOW, COIN, SNAP and friends. Justified EV/Sales from the rule of 40: multiple = 0.2 × (revenue |
| growth % + FCF margin %), floored at 1× and capped at 15× — a stated, auditable yardstick, |
| shown with a wide range.""" |
| ps, g = d.get("ps"), d.get("rev_growth") |
| if not isinstance(ps, (int, float)) or ps <= 0 or not isinstance(g, (int, float)) or not price: |
| return None |
| econ = ((d.get("us") or {}).get("econ") or {}) |
| score = econ.get("rule_of_40") |
| if not isinstance(score, (int, float)): |
| return None |
| if any("distress" in str(f).lower() for f in ((d.get("us") or {}).get("flags") or [])): |
| return None |
| |
| mult = float(np.clip(0.2 * score if score >= 20 else 0.1 * max(score, 5.0), 0.5, 15.0)) |
| sales_ps = float(price) / float(ps) |
| iv = sales_ps * mult |
| return {"intrinsic": round(iv, 2), "upside": round(float(np.clip(iv / price - 1, -0.9, 3.0)), 4), |
| "intrinsic_low": round(iv * 0.65, 2), "intrinsic_high": round(iv * 1.35, 2), |
| "implied_growth": None, "model": "Growth-stage EV/Sales (rule of 40)", |
| "assumptions": {"rule_of_40_score": round(score, 1), "justified_ev_sales": round(mult, 2), |
| "current_ev_sales": round(float(ps), 2)}, |
| "ke": None, "lifecycle": "high-growth", "src": "mechanical"} |
|
|
|
|
| def mechanical_assumptions(d: dict) -> dict: |
| """Crucible's fixed-DCF path as data-driven default assumptions — used when the |
| LLM-Damodaran call fails or is unavailable, so a positive-EPS name always gets an |
| intrinsic value. Growth from reported EPS/revenue growth (faded), ke from sector beta.""" |
| def num(x, default): |
| return float(x) if isinstance(x, (int, float)) and x == x else default |
| g1 = num(d.get("eps_growth"), num(d.get("rev_growth"), 0.05)) |
| roe = num(d.get("roe"), 0.15) |
| return {"lifecycle_stage": "mechanical", |
| "high_growth_rate": float(np.clip(g1, -0.30, 0.40)), |
| "high_growth_years": 5, |
| "terminal_growth": G_STABLE, |
| "cost_of_equity": RF + SECTOR_BETA.get(str(d.get("sector")), 1.0) * ERP, |
| "reinvestment_roic": float(np.clip(roe, 0.05, 0.40)), |
| "normalized_eps_factor": 1.0} |
|
|
|
|
| def waterfill(raw: dict, sect: dict, name_cap: float, sec_cap: float, |
| no_topup: frozenset = frozenset()) -> dict: |
| """Normalize raw>=0 into weights summing to 1 under joint per-name and per-sector caps |
| (crucible damodaran_portfolio._waterfill, pure port). Caps relax if infeasible for the |
| counts so the book is always fully invested. Names in `no_topup` keep their pro-rata |
| share but never receive overflow redistribution (e.g. committee-Trim names must not |
| grow just because everything else is capped).""" |
| tickers = list(raw) |
| groups: dict[str, list] = {} |
| for t in tickers: |
| groups.setdefault(sect.get(t, "?"), []).append(t) |
| name_cap = max(name_cap, 1.0 / max(len(tickers), 1) + 1e-9) |
| sec_cap = max(sec_cap, 1.0 / max(len(groups), 1) + 1e-9) |
| cap_of = lambda nc, sc: sum(min(sc, len(mem) * nc) for mem in groups.values()) |
| while cap_of(name_cap, sec_cap) < 1.0 - 1e-9 and name_cap < 1.0: |
| name_cap = min(1.0, name_cap + 0.02) |
| sec_cap = min(1.0, sec_cap + 0.02) |
| tot = sum(raw.values()) |
| if tot <= 0: |
| return {t: 0.0 for t in tickers} |
| w = {t: raw[t] / tot for t in tickers} |
| for _ in range(500): |
| excess = 0.0 |
| for t in tickers: |
| if w[t] > name_cap + 1e-12: |
| excess += w[t] - name_cap |
| w[t] = name_cap |
| for mem in groups.values(): |
| ssum = sum(w[t] for t in mem) |
| if ssum > sec_cap + 1e-12: |
| sc = sec_cap / ssum |
| for t in mem: |
| excess += w[t] * (1 - sc) |
| w[t] *= sc |
| if excess < 1e-9: |
| break |
| room = {} |
| for mem in groups.values(): |
| sec_room = max(sec_cap - sum(w[t] for t in mem), 0.0) |
| for t in mem: |
| if t not in no_topup and min(name_cap - w[t], sec_room) > 1e-12: |
| room[t] = max(raw[t], 1e-9) |
| if not room: |
| break |
| rs = sum(room.values()) |
| for t in room: |
| w[t] += excess * room[t] / rs |
| s = sum(w.values()) |
| return {t: w[t] / s for t in w} if s > 0 else w |
|
|
|
|
| def trap_flags(d: dict) -> list[str]: |
| """Value-trap guards (crucible): deterioration, inorganic EPS, leverage-inflated ROE.""" |
| flags = [] |
|
|
| def num(x) -> float | None: |
| return float(x) if isinstance(x, (int, float)) and x == x else None |
|
|
| rev_g, eps_g = num(d.get("rev_growth")), num(d.get("eps_growth")) |
| roe, de = num(d.get("roe")), num(d.get("de")) |
| |
| |
| if rev_g is not None and eps_g is not None: |
| if rev_g < 0 and eps_g < 0: |
| flags.append("VALUE_TRAP(rev↓+eps↓)") |
| if eps_g > 0.20 and rev_g < 0: |
| flags.append("INORGANIC(eps↑/rev↓)") |
| if de is not None and roe is not None and de > 3.0 and roe > 0.20: |
| flags.append("LEVERAGE_INFLATED_ROE") |
| return flags |
|
|
|
|
| |
|
|
| def _regime_on(hist: pd.DataFrame, bench: str = "SPY") -> bool | None: |
| """Benchmark at/above its 200-day MA now AND ~1 month ago (slipstream confirmation rule).""" |
| col = bench if bench in hist.columns else "SPY" |
| if col not in hist.columns or len(hist) < 221: |
| return None |
| spx = hist[col].ffill() |
| now = spx.iloc[-1] >= spx.iloc[-200:].mean() |
| prev = spx.iloc[-22] >= spx.iloc[-221:-21].mean() |
| return bool(now and prev) |
|
|
|
|
| def sector_etf_sector(symbol: str) -> str | None: |
| """XLF is a Financials bet, not a 'diversified fund': the sector it tracks, else None.""" |
| return _SECTOR_ETFS.get(str(symbol or "").upper()) |
|
|
|
|
| def _load_sector_etfs() -> dict[str, str]: |
| try: |
| import json |
| from pathlib import Path |
| rows = json.loads((Path(__file__).resolve().parent.parent / "data" / "instruments.json").read_text()) |
| return {str(i["ticker"]).upper(): str(i["sector"]) for i in rows |
| if i.get("asset_class") == "sector_etf" and i.get("sector")} |
| except Exception: |
| return {} |
|
|
|
|
| _SECTOR_ETFS = _load_sector_etfs() |
|
|
|
|
| def portfolio_metrics(holdings: list[dict], hist: pd.DataFrame, bench: str = "SPY") -> dict: |
| """All the deterministic numbers: allocation, concentration, risk, correlation, benchmark. |
| `hist` may span several years (for stress tests); 1y stats use the last ~252 rows.""" |
| bcol = bench if bench in hist.columns else "SPY" |
| if len(hist) > 262: |
| hist = hist.iloc[-262:] |
| w = {h["symbol"]: h["weight"] for h in holdings} |
| |
| |
| fund_syms = {h["symbol"] for h in holdings |
| if str(h["data"].get("quote_type") or "").upper() in FUND_TYPES |
| and not sector_etf_sector(h["symbol"])} |
| sectors: dict[str, float] = {} |
| for h in holdings: |
| s = "Diversified fund" if h["symbol"] in fund_syms \ |
| else (sector_etf_sector(h["symbol"]) or h["data"].get("sector") or "Unknown") |
| sectors[s] = sectors.get(s, 0.0) + h["weight"] |
| hhi = sum(x * x for x in w.values()) |
| m: dict = { |
| "n_holdings": len(holdings), |
| "equal_weighted": bool(holdings and holdings[0].get("equal_weighted")), |
| "sector_allocation": dict(sorted(sectors.items(), key=lambda kv: -kv[1])), |
| "hhi": round(hhi, 4), |
| "effective_n": round(1.0 / hhi, 1) if hhi > 0 else None, |
| "top_position": max(w.items(), key=lambda kv: kv[1]) if w else None, |
| "top_is_fund": (max(w.items(), key=lambda kv: kv[1])[0] in fund_syms) if w else False, |
| "weighted_beta": None, "portfolio_vol": None, "max_drawdown_1y": None, |
| "return_1y": None, "spy_return_1y": None, "sharpe_1y": None, |
| "corr_clusters": [], "regime_risk_on": _regime_on(hist, bcol), |
| "total_pl_pct": None, "benchmark": bcol, |
| } |
| betas = [(h["weight"], h["data"].get("beta")) for h in holdings |
| if isinstance(h["data"].get("beta"), (int, float))] |
| if betas: |
| m["weighted_beta"] = round(sum(wt * b for wt, b in betas) / sum(wt for wt, _ in betas), 2) |
| if all(h.get("market_value") for h in holdings): |
| cost = sum(h["cost_usd"] for h in holdings) |
| m["total_pl_pct"] = round(sum(h["market_value"] for h in holdings) / cost - 1, 4) if cost else None |
| else: |
| m["total_pl_pct"] = round(float(np.mean([h["pl_pct"] for h in holdings])), 4) |
|
|
| cols = [t for t in w if t in hist.columns] |
| if len(hist) > 60 and cols: |
| sub: pd.DataFrame = hist.loc[:, cols] |
| rets = sub.pct_change(fill_method=None).dropna(how="all") |
| wv = np.array([w[t] for t in cols]) |
| wv = wv / wv.sum() |
| port_ret = pd.Series(rets.fillna(0.0).to_numpy() @ wv, index=rets.index) |
| if len(port_ret) > 40: |
| m["portfolio_vol"] = round(float(port_ret.std() * np.sqrt(252)), 4) |
| m["sharpe_1y"] = round(float(port_ret.mean() / port_ret.std() * np.sqrt(252)), 2) \ |
| if port_ret.std() > 0 else None |
| curve = (1 + port_ret).cumprod() |
| m["return_1y"] = round(float(curve.iloc[-1] - 1), 4) |
| m["max_drawdown_1y"] = round(float((curve / curve.cummax() - 1).min()), 4) |
| m["equity_curve"] = [ |
| {"date": str(d.date()), "portfolio": round(float(v), 4)} |
| for d, v in curve.iloc[::max(1, len(curve) // 120)].items()] |
| if bcol in hist.columns: |
| spy = hist[bcol].pct_change(fill_method=None).reindex(port_ret.index).fillna(0.0) |
| if len(spy) > 40: |
| |
| |
| m["spy_return_1y"] = round(float((1 + spy).prod() - 1), 4) |
| cov = float(np.cov(port_ret, spy)[0, 1]) |
| var = float(spy.var()) |
| if var > 0: |
| m["realized_beta"] = round(cov / var, 2) |
| resid = port_ret - (cov / var) * spy |
| m["tracking_error"] = round(float(resid.std() * np.sqrt(252)), 4) |
| m.update(var_cvar(port_ret)) |
| |
| if len(cols) >= 2: |
| corr = rets.corr() |
| clusters = [] |
| for i, a in enumerate(cols): |
| for b in cols[i + 1:]: |
| c = corr.loc[a, b] |
| if isinstance(c, (int, float)) and c == c and c >= CORR_ALERT: |
| clusters.append({"pair": [a, b], "corr": round(float(c), 2)}) |
| m["corr_clusters"] = sorted(clusters, key=lambda x: -x["corr"])[:5] |
| return m |
|
|
|
|
| |
|
|
| def risk_scorecard(m: dict) -> list[dict]: |
| """Never depends on an LLM call succeeding. Each check: {check, status, detail}.""" |
| cards = [] |
| |
| top = m.get("top_position") |
| if top: |
| t, wt = top |
| n = int(m.get("n_holdings") or 0) |
| cap = FUND_CAP if m.get("top_is_fund") else effective_name_cap(n) |
| status = "ALERT" if wt > cap + 0.07 else "WATCH" if wt > cap + 0.03 else "OK" |
| why = (" for a broad, internally diversified fund" if m.get("top_is_fund") |
| else f" in a {n}-stock book" if cap > MAX_POS_W else "") |
| cards.append({"check": "Position concentration", "status": status, |
| "detail": f"Largest position {t} is {wt * 100:.0f}% of the portfolio " |
| f"(healthy cap ~{cap * 100:.0f}%{why})."}) |
| |
| sa = {s: v for s, v in m.get("sector_allocation", {}).items() if s != "Diversified fund"} |
| if sa: |
| s, wt = max(sa.items(), key=lambda kv: kv[1]) |
| status = "ALERT" if wt > SECTOR_CAP_HI else "WATCH" if wt > SECTOR_CAP_BASE else "OK" |
| cards.append({"check": "Sector concentration", "status": status, |
| "detail": f"{s} is {wt * 100:.0f}% of the portfolio " |
| f"(healthy band {SECTOR_CAP_LO * 100:.0f}–{SECTOR_CAP_HI * 100:.0f}%, base {SECTOR_CAP_BASE * 100:.0f}%)."}) |
| |
| n, eff = m.get("n_holdings", 0), m.get("effective_n") |
| if eff: |
| want = max(7.0, 0.4 * n) if n >= 12 else 7.0 |
| status = "ALERT" if eff < 4 else "WATCH" if eff < want else "OK" |
| cards.append({"check": "Diversification breadth", "status": status, |
| "detail": f"{n} holdings behave like ~{eff:.0f} independent positions " |
| f"(effective N from concentration)."}) |
| |
| beta = m.get("realized_beta") or m.get("weighted_beta") |
| if beta is not None: |
| status = "ALERT" if beta > 1.5 else "WATCH" if beta > 1.2 else "OK" |
| cards.append({"check": "Market sensitivity", "status": status, |
| "detail": f"Portfolio beta ≈ {beta:.2f} vs the S&P 500 — " |
| + ("amplifies market swings." if beta > 1.1 else |
| "roughly market-like." if beta > 0.9 else "defensive tilt.")}) |
| |
| cl = m.get("corr_clusters", []) |
| if cl: |
| worst = cl[0] |
| status = "ALERT" if worst["corr"] >= 0.9 else "WATCH" |
| cards.append({"check": "Correlated positions", "status": status, |
| "detail": f"{' & '.join(worst['pair'])} move together (1y correlation " |
| f"{worst['corr']:.2f}) — they act like one bet." |
| + (f" {len(cl)} correlated pairs total." if len(cl) > 1 else "")}) |
| else: |
| cards.append({"check": "Correlated positions", "status": "OK", |
| "detail": "No highly correlated pairs (≥0.80) among your holdings."}) |
| |
| reg = m.get("regime_risk_on") |
| if reg is not None: |
| cards.append({"check": "Market regime", "status": "OK" if reg else "WATCH", |
| "detail": "S&P 500 above its 200-day average — trend supportive." if reg |
| else "S&P 500 below its 200-day average — trend regime is risk-off."}) |
| return cards |
|
|
|
|
| |
|
|
| HEALTH_WEIGHTS = {"prospects": 0.25, "concentration": 0.20, "diversification": 0.15, |
| "quality": 0.15, "drawdown": 0.15, "valuation": 0.10} |
|
|
|
|
| def _profile_weights(context: dict | None) -> dict[str, float]: |
| """Drawdown weight follows the stated profile: a growth investor accepts a bigger ride.""" |
| w = dict(HEALTH_WEIGHTS) |
| risk = str((context or {}).get("risk") or "").lower() |
| if risk == "conservative": |
| w["drawdown"] = 0.25 |
| elif risk == "aggressive": |
| w["drawdown"] = 0.08 |
| tot = sum(w.values()) |
| return {k: v / tot for k, v in w.items()} |
|
|
|
|
| def _pillar_prospects(holdings, decisions) -> tuple[float, str]: |
| """Forward return and growth quality, weight-averaged: committee conviction anchored, then |
| the payoff asymmetry where the DCF is trusted, the quant model's 12m excess where covered, |
| growth durability (revenue/EPS CAGR, incremental ROIC, Rule of 40) and estimate momentum. |
| Growth-stage names are judged on durability and momentum, never on a mature-company DCF gap.""" |
| tot, acc, notes = 0.0, 0.0, [] |
| for h in holdings: |
| w = float(h.get("weight") or 0) |
| if w <= 0: |
| continue |
| t = h["symbol"] |
| d = h.get("data") or {} |
| conv = ((decisions or {}).get(t) or {}).get("conviction") |
| sc = float(conv) if isinstance(conv, (int, float)) else 50.0 |
| a = h.get("asymmetry") or {} |
| if a and a.get("gated", True) and isinstance(a.get("expected_return"), (int, float)): |
| sc += float(np.clip(a["expected_return"] * 100, -25, 25)) * 0.5 |
| pred = (d.get("panel") or {}).get("pred_excess_12m") |
| if isinstance(pred, (int, float)): |
| sc += float(np.clip(pred * 200, -20, 20)) |
| e = ((d.get("us") or {}).get("econ") or {}) |
| g = e.get("rev_cagr_3y"); ge = e.get("eps_cagr_3y") |
| best_g = max([x for x in (g, ge) if isinstance(x, (int, float))], default=None) |
| if best_g is not None: |
| sc += 8 if best_g >= 0.15 else 4 if best_g >= 0.08 else -8 if best_g <= 0 else 0 |
| if isinstance(e.get("incremental_roic_5y"), (int, float)): |
| sc += 4 if e["incremental_roic_5y"] >= 0.15 else -4 if e["incremental_roic_5y"] < 0.05 else 0 |
| if isinstance(e.get("rule_of_40"), (int, float)): |
| sc += 4 if e["rule_of_40"] >= 40 else -4 |
| st = (d.get("us") or {}).get("street") or {} |
| if isinstance(st.get("implied_return"), (int, float)) and (st.get("n_analysts") or 0) >= 5: |
| sc += float(np.clip(st["implied_return"] * 50, -6, 6)) |
| er = (d.get("us") or {}).get("earnings") or {} |
| if isinstance(er.get("eps_revision_90d_pct"), (int, float)): |
| sc += float(np.clip(er["eps_revision_90d_pct"], -10, 10)) * 0.8 |
| if er.get("n_q"): |
| sc += 4 if er["beats_8q"] >= 7 else -4 if er["beats_8q"] <= 3 else 0 |
| sc = float(np.clip(sc, 0, 100)) |
| acc += w * sc |
| tot += w |
| if not tot: |
| return 50.0, "no holdings" |
| val = acc / tot |
| return round(val, 1), f"weighted forward-return read {val:.0f}/100 across the book" |
|
|
|
|
| def _pillar_concentration(m: dict, holdings, risk) -> tuple[float, str]: |
| pen, notes = 0.0, [] |
| top = m.get("top_position") |
| n = int(m.get("n_holdings") or len(holdings) or 1) |
| if top: |
| cap = FUND_CAP if m.get("top_is_fund") else effective_name_cap(n) |
| x = max(0.0, float(top[1]) - cap) |
| if x > 0: |
| pen += min(60.0, 350 * x) |
| if float(top[1]) >= 0.20 and not m.get("top_is_fund"): |
| pen += 10 |
| notes.append(f"{top[0]} {float(top[1]) * 100:.0f}% vs {cap * 100:.0f}% cap") |
| ws = sorted((float(h.get("weight") or 0) for h in holdings), reverse=True) |
| top3 = sum(ws[:3]) |
| if n >= 6 and top3 > 0.5: |
| pen += min(20.0, 100 * (top3 - 0.5)) |
| notes.append(f"top three names {top3 * 100:.0f}%") |
| sa = {s: v for s, v in (m.get("sector_allocation") or {}).items() if s != "Diversified fund"} |
| if sa: |
| s_name, sw = max(sa.items(), key=lambda kv: kv[1]) |
| if sw > SECTOR_CAP_BASE: |
| pen += min(40.0, 120 * (sw - SECTOR_CAP_BASE)) |
| notes.append(f"{s_name} {sw * 100:.0f}% of the book") |
| act = (risk or {}).get("active_sectors") or [] |
| if act: |
| mx = max((float(a.get("active") or 0) for a in act), default=0.0) |
| if mx > 0.15: |
| pen += min(20.0, 60 * (mx - 0.15)) |
| notes.append(f"largest active sector bet +{mx * 100:.0f}pp vs benchmark") |
| return round(float(np.clip(100 - pen, 0, 100)), 1), ("; ".join(notes) or "no oversized bet") |
|
|
|
|
| def _pillar_diversification(m: dict, holdings, context) -> tuple[float, str]: |
| n = int(m.get("n_holdings") or len(holdings) or 1) |
| eff = m.get("effective_n") |
| target = min(n, max(8.0, 0.4 * n)) |
| base = float(np.clip(100 * float(eff) / max(target, 1), 20, 100)) if eff else 60.0 |
| notes = [f"~{eff} effective positions of {n}"] if eff else [] |
| k = len(m.get("corr_clusters") or []) |
| if k: |
| base -= 8 * min(k, 3) |
| notes.append(f"{k} correlated pair{'s' if k > 1 else ''}") |
| cautious = (str((context or {}).get("risk") or "").lower() == "conservative" |
| or "under 3" in str((context or {}).get("horizon") or "").lower()) |
| if cautious: |
| ballast = sum(float(h.get("weight") or 0) for h in holdings if is_ballast_fund(h)) |
| if ballast < 0.10: |
| base -= 20 |
| notes.append("no bond/cash ballast for a cautious profile") |
| return round(float(np.clip(base, 0, 100)), 1), ("; ".join(notes) or "balanced") |
|
|
|
|
| def _pillar_quality(holdings) -> tuple[float, str]: |
| tot, exposure, f_acc, f_w, worst = 0.0, 0.0, 0.0, 0.0, [] |
| for h in holdings: |
| w = float(h.get("weight") or 0) |
| if w <= 0: |
| continue |
| tot += w |
| d = h.get("data") or {} |
| us = d.get("us") or {} |
| sev = 0.0 |
| for fl in us.get("flags") or []: |
| sev += (1.0 if ("distress" in fl or "earnings-quality" in fl) |
| else 0.25 if ("insiders" in fl or "estimates cut" in fl) |
| else 0.5) |
| sev += 0.5 * len((h.get("dcf") or {}).get("flags") or []) |
| if str((h.get("dcf") or {}).get("note") or "").startswith("loss-making"): |
| sev += 0.5 |
| sev = min(sev, 1.0) |
| exposure += w * sev |
| if sev >= 1.0: |
| worst.append(h["symbol"]) |
| pz = (us.get("piotroski") or {}).get("score") |
| if isinstance(pz, (int, float)): |
| f_acc += w * pz |
| f_w += w |
| if not tot: |
| return 60.0, "no data" |
| score = 100 - 90 * (exposure / tot) |
| if f_w: |
| score += (f_acc / f_w - 5) * 4 |
| note = (f"{exposure / tot * 100:.0f}% of the book carries forensic flags" + (f" ({', '.join(worst[:4])} serious)" if worst else "")) |
| return round(float(np.clip(score, 0, 100)), 1), note |
|
|
|
|
| def _pillar_drawdown(m: dict, risk) -> tuple[float, str]: |
| """Drawdown EFFICIENCY: did the book fall further than its beta explains? A high-beta |
| growth book that fell in line with beta is efficient; one that fell more is not.""" |
| beta = m.get("realized_beta") or m.get("weighted_beta") |
| stress = (risk or {}).get("stress") or [] |
| pairs = [(float(s["portfolio"]), float(s["benchmark"])) for s in stress |
| if isinstance(s.get("portfolio"), (int, float)) and isinstance(s.get("benchmark"), (int, float))] |
| if not pairs: |
| return 65.0, "no stress history" |
| |
| |
| wb = m.get("weighted_beta") |
| b = float(wb) if isinstance(wb, (int, float)) and wb > 0 else (float(beta) if isinstance(beta, (int, float)) and beta > 0 else 1.0) |
| b = max(b, 0.5) |
| excess = [p - b * q for p, q in pairs] |
| e = sum(excess) / len(excess) |
| eff = float(np.clip(70 + 300 * e, 0, 100)) |
| ratios = [p / q for p, q in pairs if q < 0] |
| ratio = sum(ratios) / len(ratios) if ratios else 1.0 |
| absolute = float(np.clip(100 - 65 * ratio + 30, 0, 100)) |
| score = 0.5 * eff + 0.5 * absolute |
| v = m.get("var_1m_95") |
| if isinstance(v, (int, float)) and v > 0.15: |
| score -= 8 |
| note = (f"crash replays average {ratio:.2f}× the index's fall, " |
| f"{'better' if e >= 0 else 'worse'} than beta {b:.2f} explains by {abs(e) * 100:.0f}pp") |
| return round(float(np.clip(score, 0, 100)), 1), note |
|
|
|
|
| def _pillar_valuation(holdings) -> tuple[float, str]: |
| ups, pct_acc, pct_w = [], 0.0, 0.0 |
| for h in holdings: |
| w = float(h.get("weight") or 0) |
| dcf = h.get("dcf") or {} |
| a = h.get("asymmetry") or {} |
| if isinstance(dcf.get("upside"), (int, float)) and a.get("gated", True) and not dcf.get("note"): |
| ups.append((w, float(dcf["upside"]))) |
| pe = (((h.get("data") or {}).get("us") or {}).get("valuation_hist") or {}).get("pe") |
| if isinstance(pe, dict) and isinstance(pe.get("pctile"), (int, float)): |
| pct_acc += w * pe["pctile"] |
| pct_w += w |
| if ups: |
| cov_w = sum(w for w, _ in ups) |
| gap = sum(w * u for w, u in ups) / max(cov_w, 1e-9) |
| |
| |
| share = float(np.clip(cov_w / max(sum(float(h.get("weight") or 0) for h in holdings), 1e-9), 0.0, 1.0)) |
| score = share * (60 + 100 * float(np.clip(gap, -0.5, 0.35))) + (1 - share) * 60 |
| note = (f"book trades {abs(gap) * 100:.0f}% {'below' if gap > 0 else 'above'} DCF value where the model applies" |
| + (f" ({share:.0%} of the book)" if share < 0.85 else "")) |
| else: |
| score, note = 60.0, "DCF not applicable to most of the book" |
| if pct_w: |
| pc = pct_acc / pct_w |
| score += -10 if pc > 90 else -5 if pc > 75 else 5 if pc < 25 else 0 |
| note += f"; P/E at the {pc:.0f}th percentile of own history" |
| return round(float(np.clip(score, 0, 100)), 1), note |
|
|
|
|
| def health_breakdown(cards: list[dict], m: dict, holdings: list[dict], decisions: dict | None = None, |
| context: dict | None = None, risk: dict | None = None) -> dict: |
| """Six pillars, each 0-100, weighted by the investor's profile → one health number with |
| the reasons visible. Past performance is a tie-breaker (±3), never a pillar; a stated loss |
| tolerance the book breaches is a flat charge.""" |
| weights = _profile_weights(context) |
| pillars = { |
| "prospects": _pillar_prospects(holdings, decisions), |
| "concentration": _pillar_concentration(m, holdings, risk), |
| "diversification": _pillar_diversification(m, holdings, context), |
| "quality": _pillar_quality(holdings), |
| "drawdown": _pillar_drawdown(m, risk), |
| "valuation": _pillar_valuation(holdings), |
| } |
| score = sum(weights[k] * v[0] for k, v in pillars.items()) |
| r1, rs = m.get("return_1y"), m.get("spy_return_1y") |
| tie = float(np.clip((r1 - rs) * 10, -3, 3)) if isinstance(r1, (int, float)) and isinstance(rs, (int, float)) else 0.0 |
| breach = -8.0 if any(c.get("check") == "Loss tolerance" and c.get("status") == "ALERT" for c in cards) else 0.0 |
| |
| |
| weakest = min(v[0] for v in pillars.values()) |
| score = min(score, 45 + 0.6 * weakest) |
| total = int(np.clip(round(score + tie + breach), 5, 97)) |
| labels = {"prospects": "Prospects", "concentration": "Concentration", "diversification": "Diversification", |
| "quality": "Quality", "drawdown": "Drawdown efficiency", "valuation": "Valuation"} |
| return {"score": total, |
| "pillars": [{"key": k, "label": labels[k], "score": v[0], "weight": round(weights[k], 3), "note": v[1]} |
| for k, v in pillars.items()], |
| "tiebreak": round(tie, 1), "tolerance_penalty": breach} |
|
|
|
|
| def health_score(cards: list[dict], m: dict, holdings: list[dict], decisions: dict | None = None, |
| context: dict | None = None, risk: dict | None = None) -> int: |
| return health_breakdown(cards, m, holdings, decisions, context, risk)["score"] |
|
|
|
|
| |
|
|
| |
|
|
| def dcf_range(eps_base, price, A: dict | None) -> tuple[float, float]: |
| """Bear/bull intrinsic values from the same DCF with stressed assumptions — a point |
| estimate invites false precision. Bear: growth ×0.6, cost of equity +1.5pp, terminal |
| growth −0.5pp. Bull: growth ×1.25, ke −1pp. Returns (low, high); NaNs on failure.""" |
| if not A: |
| return float("nan"), float("nan") |
| bear = dict(A, high_growth_rate=float(A.get("high_growth_rate", 0.05)) * 0.6, |
| cost_of_equity=float(A.get("cost_of_equity", 0.09)) + 0.015, |
| terminal_growth=max(0.0, float(A.get("terminal_growth", 0.025)) - 0.005)) |
| bull = dict(A, high_growth_rate=float(A.get("high_growth_rate", 0.05)) * 1.25, |
| cost_of_equity=max(0.06, float(A.get("cost_of_equity", 0.09)) - 0.01)) |
| lo, _, _, _ = dcf_from_assumptions(eps_base, price, bear) |
| hi, _, _, _ = dcf_from_assumptions(eps_base, price, bull) |
| if lo == lo and hi == hi and lo > hi: |
| lo, hi = hi, lo |
| return lo, hi |
|
|
|
|
| def reverse_dcf(eps_base, price, A: dict | None) -> float | None: |
| """The growth rate the CURRENT PRICE implies under the same DCF (bisection on the |
| high-growth rate, everything else held). 'The price assumes 28%/yr for 7 years — |
| do you believe that?' is the most honest valuation statement the tool can make.""" |
| if not A or not eps_base or eps_base <= 0 or not price or price <= 0: |
| return None |
| lo, hi = -0.30, 0.60 |
| f = lambda g: dcf_from_assumptions(eps_base, price, dict(A, high_growth_rate=g))[0] - price |
| try: |
| flo, fhi = f(lo), f(hi) |
| if flo != flo or fhi != fhi: |
| return None |
| if flo > 0: |
| return lo |
| if fhi < 0: |
| return hi |
| for _ in range(50): |
| mid = (lo + hi) / 2 |
| fm = f(mid) |
| if fm != fm: |
| return None |
| if flo * fm <= 0: |
| hi, fhi = mid, fm |
| else: |
| lo, flo = mid, fm |
| return round((lo + hi) / 2, 4) |
| except Exception: |
| return None |
|
|
|
|
| def bank_value(bvps, roe, sector: str | None, g: float = G_STABLE) -> tuple[float, float]: |
| """Justified P/B for financials: P/B* = (ROE − g) / (ke − g); intrinsic = BVPS × P/B*. |
| An EPS-DCF misreads banks and insurers (leverage IS the business); this is the |
| standard excess-returns shortcut. Returns (intrinsic, ke); NaN if inputs unusable.""" |
| if not isinstance(bvps, (int, float)) or bvps <= 0 or not isinstance(roe, (int, float)): |
| return float("nan"), float("nan") |
| ke = RF + SECTOR_BETA.get(str(sector), 1.10) * ERP |
| roe = float(np.clip(roe, -0.2, 0.6)) |
| if ke <= g: |
| return float("nan"), ke |
| pb = max(0.2, (roe - g) / (ke - g)) |
| return bvps * pb, ke |
|
|
|
|
| |
|
|
| STRESS_WINDOWS = [ |
| ("COVID crash", "2020-02-19", "2020-03-23"), |
| ("2022 rate shock", "2022-01-03", "2022-10-12"), |
| ("Q4-2018 selloff", "2018-09-20", "2018-12-24"), |
| ("2025 tariff shock", "2025-02-19", "2025-04-08"), |
| ] |
|
|
|
|
| def stress_tests(holdings: list[dict], hist: pd.DataFrame, bench: str = "SPY") -> list[dict]: |
| """Historical shock replay: what would TODAY's weights have returned through each |
| window, vs the benchmark. Coverage = share of the book with prices in the window.""" |
| out = [] |
| if hist is None or hist.empty: |
| return out |
| bcol = bench if bench in hist.columns else "SPY" |
| idx = pd.to_datetime(hist.index) |
| for name, a, b in STRESS_WINDOWS: |
| sub = hist[(idx >= a) & (idx <= b)] |
| if len(sub) < 5: |
| continue |
| first, last = sub.ffill().bfill().iloc[0], sub.ffill().iloc[-1] |
| tot_w, ret = 0.0, 0.0 |
| for h in holdings: |
| t, w = h["symbol"], float(h.get("weight") or 0) |
| if t in sub.columns and int(pd.Series(sub[t]).notna().sum()) >= 3: |
| p0, p1 = float(first[t]), float(last[t]) |
| if p0 > 0 and p1 == p1: |
| ret += w * (p1 / p0 - 1) |
| tot_w += w |
| if tot_w < 0.5: |
| continue |
| b_ret = None |
| if bcol in sub.columns and float(first[bcol]) > 0: |
| b_ret = round(float(last[bcol] / first[bcol] - 1), 4) |
| out.append({"name": name, "window": f"{a} → {b}", "portfolio": round(ret / tot_w, 4), |
| "benchmark": b_ret, "coverage": round(tot_w, 2)}) |
| return out |
|
|
|
|
| def var_cvar(port_ret: pd.Series, level: float = 0.95) -> dict: |
| """Historical 1-day VaR/CVaR at 95% from the last year of daily portfolio returns, |
| plus a √21-scaled one-month VaR. Loss expressed as a positive fraction.""" |
| r = pd.Series(port_ret).dropna() |
| if len(r) < 60: |
| return {} |
| q = float(np.quantile(r, 1 - level)) |
| tail = r[r <= q] |
| return {"var_1d_95": round(-q, 4), |
| "cvar_1d_95": round(-float(tail.mean()), 4) if len(tail) else None, |
| "var_1m_95": round(-q * float(np.sqrt(21)), 4)} |
|
|
|
|
| def currency_exposure(holdings: list[dict]) -> dict[str, float]: |
| out: dict[str, float] = {} |
| for h in holdings: |
| c = str(h["data"].get("currency") or "USD") |
| c = "GBP" if c == "GBp" else c |
| out[c] = out.get(c, 0.0) + float(h.get("weight") or 0) |
| return dict(sorted(((k, round(v, 4)) for k, v in out.items()), key=lambda kv: -kv[1])) |
|
|
|
|
| def liquidity(holdings: list[dict]) -> list[dict]: |
| """Days to liquidate each position at 20% of average daily dollar volume — only when |
| quantities are known. Flags anything over one trading day.""" |
| out = [] |
| for h in holdings: |
| mv, d = h.get("market_value"), h["data"] |
| vol, px, fx = d.get("avg_volume"), d.get("price"), h.get("fx_usd") or 1.0 |
| if not (mv and isinstance(vol, (int, float)) and vol > 0 and px): |
| continue |
| adv_usd = vol * px * fx |
| days = mv / (0.2 * adv_usd) |
| if days > 1.0: |
| out.append({"ticker": h["symbol"], "days_to_liquidate": round(days, 1), |
| "adv_usd": round(adv_usd)}) |
| return sorted(out, key=lambda x: -x["days_to_liquidate"]) |
|
|
|
|
| def lookthrough(holdings: list[dict], fund_lt: dict[str, dict]) -> dict: |
| """Effective single-name and sector exposure with funds decomposed into their |
| disclosed constituents. {'top_names': [[sym, direct, via_funds, total]], 'sectors': {...}, |
| 'coverage': share of fund weight with disclosed holdings}.""" |
| direct: dict[str, float] = {} |
| via: dict[str, float] = {} |
| sectors: dict[str, float] = {} |
| fund_w, disclosed_w = 0.0, 0.0 |
| for h in holdings: |
| t, w = h["symbol"], float(h.get("weight") or 0) |
| lt = fund_lt.get(t) |
| if lt: |
| fund_w += w |
| tops = lt.get("top") or [] |
| if tops: |
| disclosed_w += w * sum(p for _, p in tops) |
| for sym, p in tops: |
| via[sym] = via.get(sym, 0.0) + w * p |
| for s, p in (lt.get("sectors") or {}).items(): |
| sectors[s] = sectors.get(s, 0.0) + w * p |
| else: |
| direct[t] = direct.get(t, 0.0) + w |
| s = str(h["data"].get("sector") or "Unknown") |
| sectors[s] = sectors.get(s, 0.0) + w |
| if not fund_w: |
| return {} |
| names = sorted(set(direct) | set(via)) |
| rows = [[n, round(direct.get(n, 0), 4), round(via.get(n, 0), 4), |
| round(direct.get(n, 0) + via.get(n, 0), 4)] for n in names] |
| rows.sort(key=lambda r: -r[3]) |
| return {"top_names": rows[:10], |
| "sectors": dict(sorted(((k, round(v, 4)) for k, v in sectors.items()), key=lambda kv: -kv[1])), |
| "fund_weight": round(fund_w, 4), |
| "coverage": round(disclosed_w / fund_w, 2) if fund_w else 0.0} |
|
|
|
|
| _FACTOR_GROUPS = { |
| "value": [("n_earnings_yield", 1), ("n_fcf_yield", 1), ("n_book_to_price", 1), ("n_sales_yield", 1)], |
| "quality": [("n_roe", 1), ("n_roa", 1), ("n_netmargin", 1), ("n_grossmargin", 1), ("n_de", -1)], |
| "momentum": [("n_mom_12_1", 1), ("n_ret_26w", 1), ("n_ret_13w", 1)], |
| "low_vol": [("n_vol_13w", -1)], |
| } |
|
|
|
|
| def factor_tilts(holdings: list[dict]) -> dict: |
| """Weighted style exposures (−1…+1) from the MarketFM normalized factor scores, plus |
| the share of the book covered (US names in the panel only).""" |
| tot, cov = 0.0, 0.0 |
| acc = {g: 0.0 for g in _FACTOR_GROUPS} |
| for h in holdings: |
| w = float(h.get("weight") or 0) |
| tot += w |
| fac = (h["data"].get("panel") or {}).get("factors") or {} |
| if not fac: |
| continue |
| cov += w |
| for g, feats in _FACTOR_GROUPS.items(): |
| vals = [sign * fac[f] for f, sign in feats if isinstance(fac.get(f), (int, float))] |
| if vals: |
| acc[g] += w * sum(vals) / len(vals) |
| if not cov or cov / max(tot, 1e-9) < 0.4: |
| return {} |
| return {"coverage": round(cov / tot, 2), |
| "tilts": {g: round(v / cov, 3) for g, v in acc.items()}} |
|
|
|
|
| _BALLAST_WORDS = ("bond", "treasury", "aggregate", "gilt", "t-bill", "bill etf", "money market", "fixed income", |
| "short-term", "ultra short", "floating rate", "tips", "inflation-protected", "corporate bond", "muni") |
| _BALLAST_TICKERS = {"BND", "AGG", "SHY", "IEF", "TLT", "BIL", "SGOV", "GOVT", "LQD", "VCIT", "VCSH", "BSV", "BIV", "TIP", |
| "VTIP", "SCHZ", "SCHO", "SCHR", "IUSB", "BNDX", "VGIT", "VGSH", "SHV", "MINT", "JPST", "ICSH", "IGLT.L", |
| "IGLS.L", "LIQUIDBEES.NS"} |
|
|
|
|
| def is_ballast_fund(h: dict) -> bool: |
| """Bond / cash funds are ballast, not equity theses — equity lenses must not eject them.""" |
| d = h.get("data") or {} |
| if str(d.get("quote_type") or "").upper() not in FUND_TYPES: |
| return False |
| if h["symbol"].upper() in _BALLAST_TICKERS: |
| return True |
| name = str(d.get("name") or "").lower() |
| return any(w in name for w in _BALLAST_WORDS) |
|
|
|
|
| def tax_note(h: dict, action: str) -> str | None: |
| """Every Trim/Sell carries its tax cost: estimated gain (in local currency) and the |
| holding period when a purchase date was given. Advice that ignores tax is not advice.""" |
| if action not in ("Trim", "Sell"): |
| return None |
| gain_ps = float(h["price"]) - float(h["entry"]) |
| qty = h.get("qty") |
| ccy = str(h["data"].get("currency") or "USD") |
| parts = [] |
| if qty: |
| g = gain_ps * float(qty) |
| parts.append(f"{'gain' if g >= 0 else 'loss'} ≈ {abs(g):,.0f} {ccy} on the full position") |
| else: |
| parts.append(f"{'gain' if gain_ps >= 0 else 'loss'} of {abs(h['pl_pct']) * 100:.0f}% since entry") |
| bought = h.get("bought") |
| if "." in str(h.get("symbol") or ""): |
| parts.append("capital-gains treatment depends on your jurisdiction (e.g. UK CGT annual " |
| "exemption, no holding-period distinction) — check your local rules") |
| bought = None |
| lots = [l for l in (h.get("lots") or []) if l.get("side") != "sell" and l.get("bought")] |
| if bought and len(lots) > 1: |
| try: |
| from datetime import date |
| def _days(b): |
| y, mo, d = (int(x) for x in str(b)[:10].split("-")) |
| return (date.today() - date(y, mo, d)).days |
| st = [l for l in lots if _days(l["bought"]) < 365] |
| if st: |
| parts.append(f"{len(lots)} lots — {len(st)} SHORT-TERM (bought " |
| + ", ".join(str(l["bought"]) for l in st) + "); sell the long-term lots first") |
| else: |
| parts.append(f"{len(lots)} lots, all long-term") |
| bought = None |
| except Exception: |
| pass |
| if bought: |
| try: |
| from datetime import date |
| y, mo, d = (int(x) for x in str(bought)[:10].split("-")) |
| days = (date.today() - date(y, mo, d)).days |
| parts.append(f"held {days // 30} months — " |
| + ("long-term" if days >= 365 else "SHORT-TERM (higher tax rate)")) |
| except Exception: |
| pass |
| else: |
| parts.append("add a purchase date to see short- vs long-term treatment") |
| if gain_ps > 0: |
| parts.append("realizing it is taxable — weigh the tax cost against the risk you remove") |
| return "; ".join(parts) |
|
|
|
|
| |
| |
| |
| |
|
|
| ALL_SECTORS = list(SECTOR_BETA) |
|
|
|
|
| def _tol_from_context(context: dict | None) -> float | None: |
| tol = str((context or {}).get("loss_tolerance") or "") |
| try: |
| return float(tol.replace("%", "").split()[0]) / 100 if tol else None |
| except Exception: |
| return None |
|
|
|
|
| def diagnose_needs(pm: dict, risk: dict, holdings: list[dict], context: dict | None) -> dict: |
| """{need: {weight 0-1, label, ...}} — what would most improve this book.""" |
| context = context or {} |
| needs: dict = {} |
| if (str(context.get("risk") or "").lower() == "conservative" |
| or "under 3" in str(context.get("horizon") or "").lower()): |
| needs["_cautious"] = {"weight": 0.0, "label": None} |
| sa = {s: w for s, w in (pm.get("sector_allocation") or {}).items() if s != "Diversified fund"} |
| lt = ((risk or {}).get("lookthrough") or {}).get("sectors") or {} |
| eff = lt if lt else sa |
| gaps = [s for s in ALL_SECTORS if eff.get(s, 0.0) < 0.03] |
| if gaps: |
| cautious = bool(needs.get("_cautious")) |
| |
| needs["sector_gaps"] = {"weight": (0.3 if cautious else 1.0) * min(1.0, 0.3 + 0.08 * len(gaps)), "sectors": gaps, |
| "label": f"no exposure to {', '.join(gaps[:4])}" + (" and more" if len(gaps) > 4 else "")} |
| if sa: |
| s, w = max(sa.items(), key=lambda kv: kv[1]) |
| if w > SECTOR_CAP_BASE: |
| needs["sector_overweight"] = {"weight": min(1.0, (w - SECTOR_CAP_BASE) / 0.3), "sector": s, |
| "label": f"{s} is {w * 100:.0f}% of the book"} |
| effn = float(pm.get("effective_n") or 0) |
| if effn and effn < 7: |
| needs["diversifier"] = {"weight": min(1.0, (7 - effn) / 5), "label": f"only ~{effn:.1f} effective positions"} |
| beta = pm.get("realized_beta") or pm.get("weighted_beta") |
| stress = (risk or {}).get("stress") or [] |
| worst = min((s["portfolio"] for s in stress), default=None) |
| tol = _tol_from_context(context) |
| d_w = 0.0 |
| if isinstance(beta, (int, float)) and beta > 1.25: |
| d_w = max(d_w, min(1.0, (beta - 1.25) / 0.5)) |
| if tol is not None and worst is not None and worst < -tol: |
| d_w = 1.0 |
| if str(context.get("risk") or "").lower() == "conservative": |
| d_w = max(d_w, 0.6) |
| if d_w: |
| lab = (f"crash replay {worst * 100:.0f}% breaches your {tol * 100:.0f}% tolerance" if tol is not None and worst is not None and worst < -tol |
| else f"portfolio beta {beta:.2f}" if isinstance(beta, (int, float)) and beta > 1.25 else "conservative profile") |
| needs["defensive"] = {"weight": d_w, "label": lab} |
| if "income" in str(context.get("objective") or "").lower(): |
| needs["income"] = {"weight": 1.0, "label": "income objective"} |
| tilts = ((risk or {}).get("factor_tilts") or {}).get("tilts") or {} |
| ups = [(h["weight"], h["dcf"]["upside"]) for h in holdings |
| if isinstance((h.get("dcf") or {}).get("upside"), (int, float))] |
| wavg = sum(w * u for w, u in ups) / max(sum(w for w, _ in ups), 1e-9) if ups else None |
| v_w = 0.0 |
| if tilts.get("value", 0) < -0.15: |
| v_w = min(1.0, -tilts["value"]) |
| if wavg is not None and wavg < -0.2: |
| v_w = max(v_w, min(1.0, -wavg)) |
| if v_w: |
| needs["value"] = {"weight": v_w, "label": (f"holdings sit {abs(wavg) * 100:.0f}% above DCF value on average" |
| if wavg is not None and wavg < -0.2 else "book tilts expensive")} |
| if tilts.get("quality", 0) < -0.1: |
| needs["quality"] = {"weight": min(1.0, -tilts["quality"] * 2), "label": "quality tilt is negative"} |
| if not needs: |
| needs["quality"] = {"weight": 0.5, "label": "no structural gaps — favour quality compounders"} |
| return needs |
|
|
|
|
| def portfolio_returns(holdings: list[dict], hist: pd.DataFrame, window: int = 262) -> pd.Series | None: |
| cols = [h["symbol"] for h in holdings if h["symbol"] in hist.columns] |
| if len(hist) < 60 or not cols: |
| return None |
| sub: pd.DataFrame = hist.loc[:, cols].iloc[-window:] |
| rets = sub.pct_change(fill_method=None).dropna(how="all") |
| wv = np.array([float(h["weight"]) for h in holdings if h["symbol"] in hist.columns]) |
| wv = wv / wv.sum() |
| return pd.Series(rets.fillna(0.0).to_numpy() @ wv, index=rets.index) |
|
|
|
|
| def _window_return(series: pd.Series, a: str, b: str) -> float | None: |
| idx = pd.to_datetime(series.index) |
| s = series[(idx >= a) & (idx <= b)].dropna() |
| if len(s) < 5 or float(s.iloc[0]) <= 0: |
| return None |
| return float(s.iloc[-1] / s.iloc[0] - 1) |
|
|
|
|
| def candidate_fit(cand: dict, holdings: list[dict], pm: dict, risk: dict, needs: dict, |
| port_ret: pd.Series | None, cand_hist: pd.DataFrame | None) -> dict: |
| """Score one live-verified candidate on the MEASURED improvement of adding it at a |
| starter weight: correlation to the book, effective-N, sector balance, beta, worst |
| crash replay, yield, valuation — each weighted by the diagnosed needs. Returns |
| {score, fixes[], corr_to_book, w_add, deltas{}} — the evidence the CIO must cite.""" |
| t = cand["ticker"] |
| live = cand.get("live") |
| if not isinstance(live, dict): |
| live = {} |
| cand["live"] = live |
| n = len(holdings) |
| w_add = min(0.08, 1.0 / (n + 1)) |
| score, fixes = 0.0, [] |
| deltas: dict = {} |
| |
| rp = cand.get("rank_pct") |
| if isinstance(rp, (int, float)): |
| score += rp / 100 * 0.4 |
| if any("smart-money" in s for s in cand.get("signals") or []): |
| score += 0.2 |
| up = cand.get("upside") |
| if isinstance(up, (int, float)): |
| score += float(np.clip(up, -0.5, 0.5)) * 0.4 |
| |
| corr = None |
| if port_ret is not None and cand_hist is not None and t in cand_hist.columns: |
| cr = pd.Series(cand_hist[t]).pct_change(fill_method=None).dropna() |
| joined = pd.DataFrame({"p": port_ret, "c": cr}).dropna() |
| p_arr, c_arr = joined["p"].to_numpy(dtype=float), joined["c"].to_numpy(dtype=float) |
| if len(joined) > 60 and c_arr.std() > 0 and p_arr.std() > 0: |
| corr = float(np.corrcoef(p_arr, c_arr)[0, 1]) |
| if corr is not None: |
| w_div = max(0.3, float(needs.get("diversifier", {}).get("weight", 0.0))) |
| score += w_div * (1.0 - corr) |
| if corr < 0.5: |
| fixes.append(f"low correlation to your holdings ({corr:.2f})") |
| |
| sec = str(cand.get("sector") or "?") |
| gap = needs.get("sector_gaps") |
| if gap and sec in gap["sectors"]: |
| score += gap["weight"] |
| fixes.append(f"adds {sec} (currently ~0% of the book)") |
| ow = needs.get("sector_overweight") |
| if ow and sec == ow["sector"]: |
| score -= 0.8 * ow["weight"] |
| |
| effn = float(pm.get("effective_n") or 0) |
| if effn: |
| w_old = [float(h["weight"]) * (1 - w_add) for h in holdings] + [w_add] |
| new_effn = 1.0 / sum(x * x for x in w_old) |
| deltas["effective_n"] = [round(effn, 1), round(new_effn, 1)] |
| if new_effn - effn > 0.3 and "diversifier" in needs: |
| score += needs["diversifier"]["weight"] * min(1.0, (new_effn - effn) / 2) |
| fixes.append(f"lifts effective positions {effn:.1f}→{new_effn:.1f}") |
| |
| d_need = needs.get("defensive") |
| beta_book = pm.get("realized_beta") or pm.get("weighted_beta") |
| beta_c = live.get("beta") |
| if not isinstance(beta_c, (int, float)) and cand_hist is not None and t in cand_hist.columns \ |
| and "SPY" in cand_hist.columns: |
| j = cand_hist[[t, "SPY"]].pct_change(fill_method=None).dropna().iloc[-252:] |
| if len(j) > 60 and float(j["SPY"].var()) > 0: |
| beta_c = float(np.cov(j[t].to_numpy(dtype=float), j["SPY"].to_numpy(dtype=float))[0, 1] / j["SPY"].var()) |
| live["beta"] = round(beta_c, 2) |
| if cand.get("instrument") and cand.get("need_target") in needs: |
| |
| |
| nt = cand["need_target"] |
| score += needs[nt]["weight"] * 0.9 |
| fixes.append({"defensive": "asset-class ballast for the defensive need", |
| "income": "asset-class income fix", "diversifier": "broad diversified exposure", |
| "sector_gaps": f"diversified {sec} exposure without single-name risk"}.get(nt, "asset-class fix")) |
| if isinstance(beta_book, (int, float)) and isinstance(beta_c, (int, float)): |
| new_beta = beta_book * (1 - w_add) + beta_c * w_add |
| deltas["beta"] = [round(beta_book, 2), round(new_beta, 2)] |
| if d_need: |
| score += d_need["weight"] * max(0.0, (beta_book - beta_c) / max(beta_book, 0.5)) |
| if beta_c < beta_book - 0.25: |
| fixes.append(f"lowers portfolio beta {beta_book:.2f}→{new_beta:.2f}") |
| elif beta_c > beta_book + 0.4: |
| score -= d_need["weight"] * min(1.2, (beta_c - beta_book) / 0.8) |
| stress = (risk or {}).get("stress") or [] |
| if stress and cand_hist is not None and t in cand_hist.columns: |
| worst = min(stress, key=lambda s: s["portfolio"]) |
| a, b = worst["window"].split(" → ") |
| cr_w = _window_return(pd.Series(cand_hist[t]), a, b) |
| if cr_w is not None: |
| new_loss = worst["portfolio"] * (1 - w_add) + cr_w * w_add |
| deltas["worst_stress"] = [round(worst["portfolio"], 3), round(new_loss, 3), worst["name"]] |
| gain = new_loss - worst["portfolio"] |
| if d_need and gain > 0.004: |
| score += d_need["weight"] * min(1.0, gain * 15) |
| fixes.append(f"cuts the {worst['name']} replay {worst['portfolio'] * 100:.0f}%→{new_loss * 100:.0f}%") |
| |
| dy = live.get("div_yield") |
| if needs.get("income"): |
| if isinstance(dy, (int, float)) and dy >= 1.5: |
| score += needs["income"]["weight"] * min(1.0, dy / 5.0) |
| if dy >= 2.5: |
| fixes.append(f"pays {dy:.1f}% (income goal)") |
| else: |
| score -= needs["income"]["weight"] * 1.0 |
| |
| if needs.get("value") and isinstance(up, (int, float)) and up > 0.05: |
| score += needs["value"]["weight"] * min(1.0, up / 0.5) |
| fixes.append(f"trades {up * 100:.0f}% below DCF value") |
| |
| |
| |
| cautions = [] |
| eps_g, sma, px = live.get("eps_growth"), live.get("sma200"), live.get("price") |
| ext = (px / sma - 1) if isinstance(px, (int, float)) and isinstance(sma, (int, float)) and sma else None |
| if (sec in CYCLICAL_SECTORS and isinstance(eps_g, (int, float)) and eps_g > 0.40 |
| and ext is not None and ext > 0.12): |
| cautions.append(f"peak-cycle risk: EPS +{eps_g * 100:.0f}% in a cyclical business with the " |
| f"price {ext * 100:+.0f}% above its 200-day") |
| score -= 0.6 |
| elif ext is not None and ext > 0.25: |
| cautions.append(f"extended: {ext * 100:+.0f}% above its 200-day average") |
| score -= 0.3 |
| if cand.get("smart_money_neg"): |
| cautions.append("informed money is not accumulating (13F/insider score negative)") |
| score -= 0.35 |
| if cand.get("flags"): |
| score -= 1.0 |
| return {"score": round(score, 3), "corr_to_book": round(corr, 2) if corr is not None else None, |
| "w_add": round(w_add, 3), "fixes": fixes[:4], "cautions": cautions, "deltas": deltas} |
|
|
|
|
| CYCLICAL_SECTORS = {"Industrials", "Energy", "Basic Materials", "Consumer Cyclical"} |
| |
| _CYCLICAL_INDUSTRY = ("semiconductor", "memory", "auto", "steel", "chemical", "oil", "gas", "mining", "metal", |
| "homebuild", "airline", "shipping", "freight", "truck", "machinery", "construction", |
| "lumber", "paper", "copper", "aluminum", "coal", "drilling", "equipment") |
|
|
|
|
| def is_cyclical(sector, industry) -> bool: |
| sec, ind = str(sector or "").title(), str(industry or "").lower() |
| return sec in CYCLICAL_SECTORS or any(k in ind for k in _CYCLICAL_INDUSTRY) |
|
|
|
|
| |
|
|
| def asymmetry(dcf: dict | None, conviction: float | None = None) -> dict | None: |
| """Klarman/Marks payoff test from the DCF range: bear/base/bull outcomes with probabilities |
| tilted by committee conviction → expected 12-18m return and upside:downside ratio. |
| A 'Buy More' that cannot show ≥ 2:1 up/down and a positive expected return is not one.""" |
| if not dcf or dcf.get("note") or not isinstance(dcf.get("intrinsic"), (int, float)): |
| return None |
| up = dcf.get("upside") |
| lo, hi = dcf.get("intrinsic_low"), dcf.get("intrinsic_high") |
| price_ratio_base = 1 + float(up) if isinstance(up, (int, float)) else None |
| if price_ratio_base is None or not isinstance(lo, (int, float)) or not isinstance(hi, (int, float)): |
| return None |
| iv = float(dcf["intrinsic"]) |
| if iv <= 0: |
| return None |
| price = iv / price_ratio_base |
| bear, base, bull = lo / price - 1, up, hi / price - 1 |
| c = float(conviction) if isinstance(conviction, (int, float)) else 50.0 |
| tilt = max(-0.15, min(0.15, (c - 50) / 100)) |
| p_bull, p_bear = 0.25 + tilt, 0.25 - tilt |
| p_base = 1 - p_bull - p_bear |
| exp = p_bear * bear + p_base * base + p_bull * bull |
| upside = max(bull, 0.0) |
| downside = abs(min(bear, 0.0)) |
| ratio = (upside / downside) if downside > 0.01 else (9.9 if upside > 0 else 0.0) |
| return {"bear": round(bear, 3), "base": round(float(base), 3), "bull": round(bull, 3), |
| "p": {"bear": round(p_bear, 2), "base": round(p_base, 2), "bull": round(p_bull, 2)}, |
| "expected_return": round(exp, 3), "up_down_ratio": round(min(ratio, 9.9), 2), |
| "clears_hurdle": bool(exp >= 0.08 and ratio >= 2.0)} |
|
|
|
|
| def kelly_size(expected_return: float | None, vol: float | None, cap: float) -> dict | None: |
| """Half-Kelly guardrail: f* = μ/σ² on the 12m view, halved, capped at the name cap. |
| Not a target — the ceiling conviction alone can justify (Buffett concentration, Thorp maths).""" |
| if not isinstance(expected_return, (int, float)) or not isinstance(vol, (int, float)) or vol <= 0: |
| return None |
| f = expected_return / (vol ** 2) |
| half = max(0.0, min(cap, 0.5 * f)) |
| return {"full_kelly": round(max(0.0, f), 3), "half_kelly": round(half, 3), "cap": cap} |
|
|
|
|
| def factor_betas(port_ret: pd.Series | None, hist: pd.DataFrame, bench: str = "SPY") -> dict: |
| """Measured style exposures: multivariate OLS of the book's daily excess returns on |
| long-only factor ETF excess returns (momentum MTUM, quality QUAL, value VLUE, |
| min-vol USMV, size IWM), each vs the market. Complements the panel-score tilts.""" |
| if port_ret is None or hist is None or hist.empty or bench not in hist.columns: |
| return {} |
| facs = [f for f in ("MTUM", "QUAL", "VLUE", "USMV", "IWM") if f in hist.columns] |
| if len(facs) < 3: |
| return {} |
| rets = hist[[bench] + facs].pct_change() |
| df = pd.concat([port_ret.rename("p"), rets], axis=1).dropna() |
| if len(df) < 120: |
| return {} |
| y = (df["p"] - df[bench]).values |
| X = np.column_stack([np.ones(len(df))] + [(df[f] - df[bench]).values for f in facs]) |
| try: |
| beta, *_ = np.linalg.lstsq(X, y, rcond=None) |
| except Exception: |
| return {} |
| mkt_beta = float(np.polyfit(df[bench].values, df["p"].values, 1)[0]) |
| names = {"MTUM": "momentum", "QUAL": "quality", "VLUE": "value", "USMV": "low_vol", "IWM": "small_size"} |
| return {"market_beta": round(mkt_beta, 2), "n_days": int(len(df)), |
| "betas": {names[f]: round(float(b), 2) for f, b in zip(facs, beta[1:])}} |
|
|
|
|
| def benchmark_relative(port_ret: pd.Series | None, hist: pd.DataFrame, bench: str = "SPY") -> dict: |
| """Upside/downside capture, Sortino, Calmar and the annualized active return vs the benchmark.""" |
| if port_ret is None or hist is None or bench not in hist.columns: |
| return {} |
| b = hist[bench].pct_change() |
| df = pd.concat([port_ret.rename("p"), b.rename("b")], axis=1).dropna() |
| if len(df) < 120: |
| return {} |
| up, dn = df[df["b"] > 0], df[df["b"] < 0] |
| out = {} |
| if len(up) > 20 and up["b"].mean() > 0: |
| out["upside_capture"] = round(float(up["p"].mean() / up["b"].mean()), 2) |
| if len(dn) > 20 and dn["b"].mean() < 0: |
| out["downside_capture"] = round(float(dn["p"].mean() / dn["b"].mean()), 2) |
| ann = float(df["p"].mean() * 252) |
| downside = float(df["p"][df["p"] < 0].std() * np.sqrt(252)) if (df["p"] < 0).any() else None |
| if downside: |
| out["sortino"] = round((ann - RF) / downside, 2) |
| curve = (1 + df["p"]).cumprod() |
| mdd = float((curve / curve.cummax() - 1).min()) |
| if mdd < 0: |
| out["calmar"] = round(ann / abs(mdd), 2) |
| out["active_return_ann"] = round(ann - float(df["b"].mean() * 252), 4) |
| return out |
|
|
|
|
| |
| |
| SPY_SECTORS_FALLBACK = {"Technology": 0.33, "Financial Services": 0.13, "Consumer Cyclical": 0.105, |
| "Communication Services": 0.095, "Healthcare": 0.095, "Industrials": 0.085, |
| "Consumer Defensive": 0.055, "Energy": 0.03, "Utilities": 0.025, |
| "Real Estate": 0.02, "Basic Materials": 0.02} |
|
|
|
|
| def active_sector_weights(pm_sectors: dict, bench_sectors: dict) -> list[dict]: |
| """Book vs benchmark sector weights — the active bets, largest first.""" |
| if not bench_sectors: |
| bench_sectors = SPY_SECTORS_FALLBACK |
| if not pm_sectors: |
| return [] |
| tot = sum(float(v) for v in bench_sectors.values()) or 1.0 |
| bench = {str(k).title(): float(v) / tot for k, v in bench_sectors.items()} |
| rows = [] |
| for s in set(pm_sectors) | set(bench): |
| if s in ("Diversified Fund", "Diversified fund", "Unknown", "?"): |
| continue |
| w, bw = float(pm_sectors.get(s, 0.0)), float(bench.get(s, 0.0)) |
| rows.append({"sector": s, "book": round(w, 4), "benchmark": round(bw, 4), "active": round(w - bw, 4)}) |
| rows.sort(key=lambda r: -abs(r["active"])) |
| return rows[:8] |
|
|
