| """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 |
| MIN_OBS = 120 |
|
|
|
|
| |
|
|
| 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) |
| |
| Y = X ** 2 |
| pi_mat = (Y.T @ Y) / T - S ** 2 |
| pi_hat = pi_mat.sum() |
| |
| term = ((X ** 3).T @ X) / T - var[:, None] * S |
| theta = term |
| rho_hat = np.diag(pi_mat).sum() |
| if N > 1: |
| r_over = rbar / 2.0 |
| ratio = np.sqrt(np.outer(1.0 / var, var)) |
| 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) |
|
|
|
|
| |
|
|
| 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} |
| 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]]} |
|
|
|
|
| |
|
|
| 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 |
|
|
|
|
| |
|
|
| 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.""" |
| |
| |
| optimize_target.relaxed = [] |
| optimize_target.tolerance_met = None |
| optimize_target.cash = 0.0 |
| 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]) |
| |
| if lo.sum() > 1.0: |
| lo = lo / lo.sum() * 0.98 |
| if hi.sum() < 1.0: |
| |
| |
| |
| 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 |
| hi[growable] = np.minimum(lifted, np.maximum(hi[growable], lift_cap)) |
| if hi.sum() < 1.0: |
| |
| |
| |
| 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: |
| |
| |
| 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", "?"): |
| continue |
| members = [i for i, t in enumerate(U) if sector_of.get(t, "?") == s] |
| cap = max(sector_cap, float(lo[members].sum()) + 1e-6) |
| cons.append({"type": "ineq", "fun": (lambda w, mem=members, c=cap: c - float(w[mem].sum()))}) |
| if betas and beta_floor is not None: |
| bv = np.array([float(betas.get(t, 1.0)) for t in U]) |
| 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 |
| 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"): |
| b[t] = (min(w, cap), max(w, cap)) |
| continue |
| if mode == "de_risk": |
| b[t] = (0.0, max(min(w, cap), 0.0) if act in ("Sell", "Trim") else cap) |
| 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))) |
| note = str(decisions.get(t, {}).get("size_note") or "") |
| if "size itself is the risk" in note and w > cap: |
| |
| |
| b[t] = (min(w, 0.8 * cap), min(w, cap)) |
| else: |
| hi_t = min(0.8 * w, max(w - band, 0.5 * w), cap) |
| b[t] = (min(0.4 * w, hi_t), hi_t) |
| 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)) |
| 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 |
| if d.get("no_add") or conv < 58 or w >= min_size: |
| hi_k = min(w, cap) |
| else: |
| hi_k = min(cap, min_size) |
| 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 |
|
|
|
|
| |
|
|
| 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: |
| 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 |
|
|
