"""Portfolio construction — the institutional layer between "good ideas" and "a book". - shrunk_cov: Ledoit-Wolf constant-correlation shrinkage over ~2y of daily returns (sample correlations from one year are noise; shrinkage is standard) - risk_contributions: marginal / percent contribution to portfolio risk per holding — the diagnosis that says WHICH positions to cut, not just which are large - greedy_select: conditional selection — each add is chosen given the adds before it, by variance reduction plus need-fit, so two redundant names never both win - optimize_target: SLSQP sizing under name/sector caps, a turnover budget and a tax penalty; three scenarios (full, tax-light, new-cash-only) - post_trade: the ENTIRE risk block re-run on the recommended book, side by side All deterministic; nothing here touches an LLM. """ from __future__ import annotations import numpy as np import pandas as pd from . import metrics as M TRADING_DAYS = 252 COV_WINDOW = 500 # ~2 years of daily returns MIN_OBS = 120 # ---------------- covariance ---------------- def returns_matrix(hist: pd.DataFrame, cand_hist: pd.DataFrame | None, tickers: list[str]) -> pd.DataFrame: """Aligned daily returns for holdings + candidates over the last COV_WINDOW days. Columns with fewer than MIN_OBS observations are dropped; remaining gaps are zeros.""" frames = [hist] if cand_hist is not None and not cand_hist.empty: frames.append(cand_hist[[c for c in cand_hist.columns if c not in hist.columns]]) px = pd.concat(frames, axis=1) cols = [t for t in tickers if t in px.columns] if not cols: return pd.DataFrame() sub: pd.DataFrame = px.loc[:, cols].iloc[-(COV_WINDOW + 1):] rets = sub.pct_change(fill_method=None).iloc[1:] keep = [c for c in rets.columns if int(rets[c].notna().sum()) >= MIN_OBS] return rets[keep].fillna(0.0) def shrunk_cov(rets: pd.DataFrame) -> tuple[np.ndarray, list[str], float]: """Ledoit-Wolf (2003) shrinkage toward the constant-correlation target. Returns the ANNUALIZED covariance, the ticker order, and the shrinkage intensity used.""" X = rets.to_numpy(dtype=float) T, N = X.shape if T < 30 or N == 0: return np.zeros((N, N)), list(rets.columns), 0.0 X = X - X.mean(axis=0) S = X.T @ X / T var = np.diag(S).copy() sd = np.sqrt(np.maximum(var, 1e-12)) corr = S / np.outer(sd, sd) if N > 1: rbar = (corr.sum() - N) / (N * (N - 1)) else: rbar = 0.0 F = rbar * np.outer(sd, sd) np.fill_diagonal(F, var) # pi: sum of asymptotic variances of the sample covariance entries Y = X ** 2 pi_mat = (Y.T @ Y) / T - S ** 2 pi_hat = pi_mat.sum() # rho: covariance between shrinkage target and sample estimator term = ((X ** 3).T @ X) / T - var[:, None] * S theta = term # theta_{ii,ij} rho_hat = np.diag(pi_mat).sum() if N > 1: r_over = rbar / 2.0 ratio = np.sqrt(np.outer(1.0 / var, var)) # sqrt(var_j / var_i) rho_mat = r_over * (ratio * theta + ratio.T * theta.T) rho_hat += rho_mat.sum() - np.trace(rho_mat) gamma = float(((S - F) ** 2).sum()) kappa = (pi_hat - rho_hat) / gamma if gamma > 0 else 0.0 delta = float(np.clip(kappa / T, 0.0, 1.0)) sigma = delta * F + (1 - delta) * S return sigma * TRADING_DAYS, list(rets.columns), round(delta, 3) # ---------------- risk decomposition ---------------- def risk_contributions(weights: dict[str, float], cov: np.ndarray, tickers: list[str]) -> dict: """{'portfolio_vol', 'contrib': {t: {weight, mctr, pct_risk}}, 'top': [[t, pct], ...]}""" idx = {t: i for i, t in enumerate(tickers)} w = np.array([weights.get(t, 0.0) for t in tickers], dtype=float) if w.sum() <= 0 or cov.size == 0: return {} w = w / w.sum() var = float(w @ cov @ w) if var <= 0: return {} vol = np.sqrt(var) mctr = (cov @ w) / vol contrib = w * mctr out = {t: {"weight": round(float(w[idx[t]]), 4), "mctr": round(float(mctr[idx[t]]), 4), "pct_risk": round(float(contrib[idx[t]] / vol), 4)} for t in tickers if w[idx[t]] > 1e-9} # holdings only, not the candidate pool top = sorted(((t, v["pct_risk"]) for t, v in out.items()), key=lambda kv: -kv[1]) return {"portfolio_vol": round(float(vol), 4), "contrib": out, "top": [[t, p] for t, p in top[:5]]} # ---------------- conditional (greedy) selection ---------------- def greedy_select(cur_w: dict[str, float], candidates: list[dict], cov: np.ndarray, tickers: list[str], max_adds: int = 4, w_add: float | None = None) -> list[dict]: """Pick adds one at a time, each conditional on the previous picks: score = 0.6 × rank of variance reduction (given the book AS IT WOULD BE) + 0.4 × rank of need-fit. Two names that diversify the book but not each other cannot both win.""" idx = {t: i for i, t in enumerate(tickers)} held = [t for t in cur_w if t in idx] if not held or cov.size == 0: return [] n = len(cur_w) w_add = w_add or min(0.08, 1.0 / (n + 1)) w = np.zeros(len(tickers)) for t in held: w[idx[t]] = cur_w[t] w = w / w.sum() pool = [c for c in candidates if c["ticker"] in idx and c["ticker"] not in cur_w] picks: list[dict] = [] for _ in range(max_adds): if not pool: break var0 = float(w @ cov @ w) rows = [] for c in pool: w1 = w * (1 - w_add) w1[idx[c["ticker"]]] += w_add var1 = float(w1 @ cov @ w1) rows.append((c, (var0 - var1) / var0 if var0 > 0 else 0.0)) vr = np.array([r[1] for r in rows]) fs = np.array([float((r[0].get("fit") or {}).get("score", 0.0)) for r in rows]) rank = lambda a: (np.argsort(np.argsort(a)) + 1) / len(a) if len(a) > 1 else np.ones(len(a)) combo = 0.6 * rank(vr) + 0.4 * rank(fs) best = int(np.argmax(combo)) c, red = rows[best] picks.append({"ticker": c["ticker"], "order": len(picks) + 1, "variance_reduction": round(float(red), 4), "vol_before": round(float(np.sqrt(var0)), 4), "vol_after": round(float(np.sqrt(var0 * (1 - red))), 4)}) w = w * (1 - w_add) w[idx[c["ticker"]]] += w_add pool = [p for p in pool if p["ticker"] != c["ticker"]] return picks # ---------------- constrained sizing ---------------- def optimize_target(cur_w: dict[str, float], mu: dict[str, float], cov: np.ndarray, tickers: list[str], bounds: dict[str, tuple[float, float]], sector_of: dict[str, str], sector_cap: float, turnover_max: float, tax_cost: dict[str, float], tax_weight: float = 0.5, risk_aversion: float = 3.0, stress_rets: dict[str, dict[str, float]] | None = None, loss_tol: float | None = None, betas: dict[str, float] | None = None, beta_floor: float | None = None) -> dict[str, float] | None: """min λ·w'Σw − μ'w + τ·Σ tax_cost_i·(w_cur_i − w_i)+ s.t. Σw=1, bounds, sector caps, turnover Σ|w−w_cur|/2 ≤ T, and — when the investor stated a loss tolerance — every crash replay Σ w_i·r_i(window) ≥ −tolerance (stress_rets: {window: {ticker: return}}). If the tolerance cannot be met inside the bounds the constraint is dropped and `tolerance_met` is set False for the caller. Returns weights or None if the solver fails.""" # per-call state read by scenarios(): reset FIRST — a stale list once made a 25-name book's # narrative say "VRTX and UNH are trimmed in stages" when neither was in that book optimize_target.relaxed = [] # noqa optimize_target.tolerance_met = None # noqa optimize_target.cash = 0.0 # noqa — unallocated proceeds (dry powder), read by scenarios try: from scipy.optimize import minimize except Exception: return None U = [t for t in tickers if t in bounds] if not U: return None idx = {t: i for i, t in enumerate(tickers)} C = cov[np.ix_([idx[t] for t in U], [idx[t] for t in U])] w0 = np.array([cur_w.get(t, 0.0) for t in U]) m = np.array([mu.get(t, 0.0) for t in U]) tc = np.array([tax_cost.get(t, 0.0) for t in U]) lo = np.array([bounds[t][0] for t in U]) hi = np.array([bounds[t][1] for t in U]) # feasibility: if lower bounds already exceed 1 (or upper bounds cannot reach 1) relax if lo.sum() > 1.0: lo = lo / lo.sum() * 0.98 if hi.sum() < 1.0: # feasibility must not silently lift the caps on names the verdicts froze (a gated # NVDA once grew from 6.8% to 9.2% this way): only names allowed to grow absorb the # shortfall; the uniform stretch is the last resort growable = hi > w0 + 1e-9 short = 1.02 - hi.sum() if growable.any(): room = hi[growable] - w0[growable] lifted = hi[growable] + short * (room / room.sum() if room.sum() > 0 else 1.0 / growable.sum()) lift_cap = 0.10 if len(U) <= 12 else 0.06 # in a broad book an add is never a 10% bet (five 6% adds once ate 31%) hi[growable] = np.minimum(lifted, np.maximum(hi[growable], lift_cap)) if hi.sum() < 1.0: # still short: the sales are bigger than the book can absorb at starter sizes — # trim the TRIMS less (a staged reduction), never un-sell a Sell, and keep at least # half of each intended cut so the trade stays meaningful shrinking = (hi < w0 - 1e-9) & (hi > 1e-9) if shrinking.any(): room = 0.5 * (w0[shrinking] - hi[shrinking]) add = np.minimum((1.02 - hi.sum()) * room / room.sum(), room) hi[shrinking] = hi[shrinking] + add optimize_target.relaxed = [U[i] for i in np.where(shrinking)[0] if add[list(np.where(shrinking)[0]).index(i)] > 1e-4] if hi.sum() < 1.0: # proceeds exceed what the committee's ideas can absorb: the professional answer is # cash held for the watchlist, not a forced top-up of lukewarm Keeps optimize_target.cash = float(min(0.25, 1.0 - hi.sum())) total = 1.0 - optimize_target.cash def obj(w): return (risk_aversion * float(w @ C @ w) - float(m @ w) + tax_weight * float(tc @ np.maximum(w0 - w, 0.0))) cons = [{"type": "eq", "fun": lambda w: float(w.sum() - total)}, {"type": "ineq", "fun": lambda w: turnover_max - 0.5 * float(np.abs(w - w0).sum())}] for s in set(sector_of.get(t, "?") for t in U): if s in ("Diversified fund", "?"): # funds are internally diversified — the per-name continue # fund cap governs them, not a "sector" cap that members = [i for i, t in enumerate(U) if sector_of.get(t, "?") == s] # once starved BIL cap = max(sector_cap, float(lo[members].sum()) + 1e-6) # never infeasible vs the floors cons.append({"type": "ineq", "fun": (lambda w, mem=members, c=cap: c - float(w[mem].sum()))}) if betas and beta_floor is not None: # keep the book's character: no silent conversion bv = np.array([float(betas.get(t, 1.0)) for t in U]) # of a growth book into utilities cons.append({"type": "ineq", "fun": (lambda w, b=bv, f=beta_floor: float(w @ b) - f)}) stress_cons = [] if stress_rets and loss_tol is not None: for win, rets in stress_rets.items(): r = np.array([float(rets.get(t, 0.0)) for t in U]) stress_cons.append({"type": "ineq", "fun": (lambda w, rv=r, tol=loss_tol: tol + float(w @ rv))}) start = np.clip(w0, lo, hi) start = start / start.sum() * total if start.sum() > 0 else np.full(len(U), total / len(U)) def _fill_to_one(w): """Σw must be 1 without breaking a bound: the residual goes into names with room (a plain rescale once lifted 10% starters to 12%).""" for _ in range(4): resid = total - float(w.sum()) if abs(resid) < 1e-6: break room = (hi - w) if resid > 0 else (w - lo) room = np.maximum(room, 0.0) if room.sum() <= 1e-12: return w / w.sum() * total w = np.clip(w + resid * room / room.sum(), lo, hi) return w / w.sum() * total if abs(total - float(w.sum())) > 1e-4 else w def _solve(extra): res = minimize(obj, start, method="SLSQP", bounds=list(zip(lo, hi)), constraints=cons + extra, options={"maxiter": 300, "ftol": 1e-9}) w = np.clip(res.x, lo, hi) w = _fill_to_one(w) if not np.all(np.isfinite(w)): return None, False ok = bool(res.success) and all(c["fun"](w) >= -1e-4 for c in extra) return {t: float(w[i]) for i, t in enumerate(U)}, ok try: if stress_cons: w, ok = _solve(stress_cons) if w and ok: optimize_target.tolerance_met = True # noqa — read by scenarios() return w optimize_target.tolerance_met = False if stress_cons else None w, _ = _solve([]) return w except Exception: return None def _grow(w: float, cap: float) -> float: """An existing position may at most double, or gain 3 points of the book, whichever is larger — a 0.8% satellite is never turned into a 14% core by the solver.""" return min(cap, max(w * 2.0, w + 0.03)) def build_bounds(holdings: list[dict], decisions: dict, picks: list[str], caps: dict[str, float], mode: str = "full", new_cash: float = 0.0) -> dict[str, tuple[float, float]]: """Action-consistent bounds: Sell → 0; Trim → at most current (and below cap); Keep → not gutted (≥50% of current) and not grown past its cap; Buy More → at least current, up to cap; new picks → 0…cap. 'new_cash' mode forbids sells: existing weights can only shrink by dilution from the new money.""" b: dict[str, tuple[float, float]] = {} for h in holdings: t, w = h["symbol"], float(h["weight"]) cap = caps.get(t, M.MAX_POS_W) act = decisions.get(t, {}).get("action", "Keep") if mode == "new_cash": floor = w / (1 + new_cash) b[t] = (floor, max(floor, min(cap, w)) if act != "Buy More" else max(floor, cap)) continue if decisions.get(t, {}).get("ballast"): # protected bond/cash cushion: never sold down b[t] = (min(w, cap), max(w, cap)) continue if mode == "de_risk": # honouring a loss tolerance may mean cutting b[t] = (0.0, max(min(w, cap), 0.0) if act in ("Sell", "Trim") else cap) # Keeps well below half if act == "Sell": b[t] = (0.0, 0.0) continue if act == "Sell": b[t] = (0.0, 0.0) elif act == "Trim": band = min(0.02, max(0.005, 0.5 / max(len(holdings), 1))) # same band the order plan uses note = str(decisions.get(t, {}).get("size_note") or "") if "size itself is the risk" in note and w > cap: # a size trim on a backed winner brings it back TOWARD the cap (META 24% → 12-15%), # never down to 40% of the position as a damage trim would b[t] = (min(w, 0.8 * cap), min(w, cap)) else: hi_t = min(0.8 * w, max(w - band, 0.5 * w), cap) # a Trim is a real trade, not a 0.1% shave b[t] = (min(0.4 * w, hi_t), hi_t) # …and never a full exit (that is a Sell verdict) elif act == "Buy More": min_size = min(0.02, max(0.01, 0.5 / max(len(holdings), 1))) lo_b = max(min(w, cap), min(min_size, cap)) # a Buy More is funded at least to a real position b[t] = (lo_b, max(lo_b, min(max(w, cap), _grow(w, cap)))) else: d = decisions.get(t, {}) min_size = min(0.02, max(0.01, 0.5 / max(len(holdings), 1))) conv = float(d.get("conviction") or 50) lo_k = (0.25 if d.get("watch") else 0.8 if d.get("protect") else 0.5) * w # watch names are sources of funds if d.get("no_add") or conv < 58 or w >= min_size: hi_k = min(w, cap) # fresh capital goes to Buy More verdicts and new ideas, never to a plain Keep else: hi_k = min(cap, min_size) # a backed Keep below the trading band is rounded up to a real position b[t] = (min(lo_k, cap), max(hi_k, 0.0) if w > cap else hi_k) for t in picks: b[t] = (0.0, caps.get(t, M.MAX_POS_W)) return b def tax_cost_per_weight(holdings: list[dict]) -> dict[str, float]: """Tax cost per unit of weight SOLD ≈ gain fraction of proceeds × rate (long-term 20%, short-term/unknown 30%). Losses are free (a small negative → harvesting is rewarded).""" out = {} for h in holdings: px, entry = float(h["price"]), float(h["entry"]) gain_frac = (px - entry) / px if px > 0 else 0.0 rate = 0.30 b = h.get("bought") if b: try: from datetime import date y, mo, d = (int(x) for x in str(b)[:10].split("-")) if (date.today() - date(y, mo, d)).days >= 365: rate = 0.20 except Exception: pass out[h["symbol"]] = round(max(gain_frac, -0.2) * rate, 4) return out # ---------------- post-trade risk (the whole block, on the recommended book) ---------------- def post_trade(weights: dict[str, float], holdings: list[dict], cand_data: dict[str, dict], hist: pd.DataFrame, cand_hist: pd.DataFrame | None, bench: str, fund_lt: dict[str, dict] | None = None, decisions: dict | None = None, context: dict | None = None) -> dict: """Re-run metrics/stress/VaR/look-through on the target book. Candidates become pseudo holdings using their live data.""" by = {h["symbol"]: h for h in holdings} pseudo = [] for t, w in weights.items(): if w <= 1e-6: continue if t in by: h = dict(by[t], weight=w) else: d = cand_data.get(t) or {"ticker": t} h = {"symbol": t, "weight": w, "data": d, "entry": d.get("price") or 1.0, "price": d.get("price") or 1.0, "pl_pct": 0.0, "market_value": None, "cost_usd": None, "equal_weighted": False, "dcf": {}} pseudo.append(h) if not pseudo: return {} frames = [hist] if cand_hist is not None and not cand_hist.empty: extra = [c for c in cand_hist.columns if c not in hist.columns] if extra: frames.append(cand_hist[extra]) full = pd.concat(frames, axis=1) pm = M.portfolio_metrics(pseudo, full, bench) cards = M.risk_scorecard(pm) stress = M.stress_tests(pseudo, full, bench) lt = M.lookthrough(pseudo, fund_lt or {}) if fund_lt else {} return {"vol": pm.get("portfolio_vol"), "effective_n": pm.get("effective_n"), "beta": pm.get("realized_beta") or pm.get("weighted_beta"), "var_1d_95": pm.get("var_1d_95"), "var_1m_95": pm.get("var_1m_95"), "top_sector": max(pm["sector_allocation"].items(), key=lambda kv: kv[1]) if pm.get("sector_allocation") else None, "sector_allocation": pm.get("sector_allocation"), "corr_clusters": len(pm.get("corr_clusters") or []), "stress": stress, "lookthrough_top": (lt.get("top_names") or [])[:3], "est_health": M.health_score(cards, pm, pseudo, decisions, context, {"stress": stress}), "cards": [(c["check"], c["status"]) for c in cards]} def scenarios(holdings: list[dict], decisions: dict, picks: list[str], mu: dict[str, float], cov: np.ndarray, tickers: list[str], caps: dict[str, float], sector_of: dict[str, str], sector_cap: float, book_usd: float | None, stress_rets: dict[str, dict[str, float]] | None = None, loss_tol: float | None = None, risk_aversion: float = 3.0, betas: dict[str, float] | None = None, beta_floor: float | None = None) -> dict[str, dict]: """Three ways to get there. Each: weights, turnover, est. tax cost, expected edge, and — when a loss tolerance was stated — whether every crash replay stays inside it.""" cur = {h["symbol"]: float(h["weight"]) for h in holdings} tc = tax_cost_per_weight(holdings) specs = {"full": {"turnover": 0.40, "tax_w": 0.5, "mode": "full"}, "tax_light": {"turnover": 0.15, "tax_w": 2.5, "mode": "full"}, "new_cash": {"turnover": 0.12, "tax_w": 0.0, "mode": "new_cash"}} if loss_tol is not None: # what it actually takes to honour the stated tolerance specs["de_risk"] = {"turnover": 0.80, "tax_w": 0.5, "mode": "de_risk"} out: dict[str, dict] = {} for name, sp in specs.items(): bounds = build_bounds(holdings, decisions, picks, caps, sp["mode"], new_cash=0.10) w = optimize_target(cur, mu, cov, tickers, bounds, sector_of, sector_cap, sp["turnover"], tc, tax_weight=sp["tax_w"], risk_aversion=risk_aversion, stress_rets=stress_rets, loss_tol=loss_tol, betas=betas, beta_floor=(None if name == "de_risk" else beta_floor)) if not w: continue tol_met = getattr(optimize_target, "tolerance_met", None) staged = list(getattr(optimize_target, "relaxed", []) or []) cash = round(float(getattr(optimize_target, "cash", 0.0) or 0.0), 4) turnover = 0.5 * sum(abs(w.get(t, 0.0) - cur.get(t, 0.0)) for t in set(w) | set(cur)) sold = {t: max(cur.get(t, 0.0) - w.get(t, 0.0), 0.0) for t in cur} tax_frac = sum(tc.get(t, 0.0) * s for t, s in sold.items()) edge_before = sum(cur[t] * mu.get(t, 0.0) for t in cur) edge_after = sum(w[t] * mu.get(t, 0.0) for t in w) out[name] = {"weights": {t: round(v, 4) for t, v in w.items() if v > 1e-4}, "turnover": round(turnover, 4), "est_tax_usd": round(tax_frac * book_usd) if book_usd else None, "est_tax_pct_of_book": round(tax_frac, 4), "edge_before": round(edge_before, 4), "edge_after": round(edge_after, 4), "tolerance_met": tol_met, "staged_trims": staged, "cash": cash, "label": {"full": "Full rebalance", "tax_light": "Tax-light rebalance", "new_cash": "New cash only (≈10% added, no sells)", "de_risk": "Meet your loss tolerance (bigger trade)"}[name]} return out