"""Construction layer — shrinkage covariance, risk contributions, greedy selection, optimizer.""" import numpy as np import pandas as pd from src import construct as CX def _rets(n=600, seed=0): rng = np.random.default_rng(seed) f = rng.normal(0, 0.01, n) A = f + rng.normal(0, 0.006, n) B = 0.9 * f + rng.normal(0, 0.006, n) # near-duplicate of A C = -0.2 * f + rng.normal(0, 0.008, n) # diversifier D = 0.05 * f + rng.normal(0, 0.003, n) # low-vol ballast idx = pd.bdate_range(end="2026-09-01", periods=n) return pd.DataFrame({"A": A, "B": B, "C": C, "D": D}, index=idx) def test_shrunk_cov_is_psd_and_shrinks(): cov, tick, delta = CX.shrunk_cov(_rets()) assert tick == ["A", "B", "C", "D"] and 0.0 <= delta <= 1.0 assert np.all(np.linalg.eigvalsh(cov) > -1e-9) assert cov[0, 1] > 0 and cov[0, 2] < 0 def test_risk_contributions_sum_to_one_and_find_the_culprit(): cov, tick, _ = CX.shrunk_cov(_rets()) rc = CX.risk_contributions({"A": 0.4, "B": 0.4, "C": 0.1, "D": 0.1}, cov, tick) tot = sum(v["pct_risk"] for v in rc["contrib"].values()) assert abs(tot - 1.0) < 0.02 assert rc["top"][0][0] in ("A", "B") and rc["contrib"]["D"]["pct_risk"] < 0.1 def test_greedy_select_prefers_diversifier_and_is_conditional(): cov, tick, _ = CX.shrunk_cov(_rets()) cands = [{"ticker": "B", "fit": {"score": 3.0}}, {"ticker": "C", "fit": {"score": 1.0}}, {"ticker": "D", "fit": {"score": 1.0}}] picks = CX.greedy_select({"A": 1.0}, cands, cov, tick, max_adds=2) assert [p["ticker"] for p in picks][0] in ("C", "D") # not the near-duplicate B assert picks[0]["variance_reduction"] > 0 and picks[0]["vol_after"] < picks[0]["vol_before"] def test_optimizer_respects_bounds_turnover_and_sums_to_one(): cov, tick, _ = CX.shrunk_cov(_rets()) cur = {"A": 0.6, "B": 0.4} bounds = {"A": (0.0, 0.3), "B": (0.2, 0.4), "C": (0.0, 0.3), "D": (0.0, 0.3)} w = CX.optimize_target(cur, {"C": 0.05, "D": 0.02}, cov, tick, bounds, {"A": "T", "B": "T", "C": "U", "D": "F"}, 0.6, 0.5, {"A": 0.05, "B": 0.0}) assert w is not None and abs(sum(w.values()) - 1) < 1e-6 for t, (lo, hi) in bounds.items(): assert lo - 1e-6 <= w[t] <= hi + 1e-6 assert 0.5 * sum(abs(w.get(t, 0) - cur.get(t, 0)) for t in set(w) | set(cur)) <= 0.5 + 1e-6 def test_build_bounds_action_consistency(): hs = [{"symbol": "S", "weight": 0.2}, {"symbol": "T", "weight": 0.3}, {"symbol": "K", "weight": 0.1}, {"symbol": "M", "weight": 0.4}] dec = {"S": {"action": "Sell"}, "T": {"action": "Trim"}, "K": {"action": "Keep"}, "M": {"action": "Buy More"}} caps = {t: 0.25 for t in "STKM"} | {"N": 0.25} b = CX.build_bounds(hs, dec, ["N"], caps) assert b["S"] == (0.0, 0.0) and b["T"][1] <= 0.25 and b["K"][0] == 0.05 and b["M"][0] == 0.25 b2 = CX.build_bounds(hs, dec, ["N"], caps, mode="new_cash", new_cash=0.10) assert b2["S"][0] > 0 # no sells in new-cash mode def test_tax_cost_per_weight_rates(): hs = [{"symbol": "G", "price": 200.0, "entry": 100.0, "bought": "2020-01-01"}, {"symbol": "S", "price": 200.0, "entry": 100.0, "bought": "2026-08-01"}, {"symbol": "L", "price": 50.0, "entry": 100.0}] tc = CX.tax_cost_per_weight(hs) assert tc["G"] == 0.1 and tc["S"] == 0.15 and tc["L"] < 0 def test_loss_tolerance_constraint_binds(): import numpy as np from src import construct as CX U = ["A", "B", "CASH"] cov = np.array([[0.09, 0.05, 0.0], [0.05, 0.08, 0.0], [0.0, 0.0, 0.0001]]) cur = {"A": 0.5, "B": 0.5, "CASH": 0.0} mu = {"A": 0.40, "B": 0.35, "CASH": 0.02} bounds = {"A": (0.0, 0.6), "B": (0.0, 0.6), "CASH": (0.0, 1.0)} sec = {"A": "Tech", "B": "Tech", "CASH": "Diversified fund"} stress = {"COVID": {"A": -0.40, "B": -0.35, "CASH": 0.0}} w_free = CX.optimize_target(cur, mu, cov, U, bounds, sec, 0.9, 1.0, {}, tax_weight=0.0) w_tol = CX.optimize_target(cur, mu, cov, U, bounds, sec, 0.9, 1.0, {}, tax_weight=0.0, stress_rets=stress, loss_tol=0.15) replay = lambda w: sum(w[t] * stress["COVID"][t] for t in U) assert replay(w_tol) >= -0.15 - 1e-3 and w_tol["CASH"] > w_free["CASH"] assert CX.optimize_target.tolerance_met is True def test_bounds_growth_discipline_and_beta_floor(): import numpy as np from src import construct as CX hs = [{"symbol": "A", "weight": 0.008}, {"symbol": "B", "weight": 0.10}] dec = {"A": {"action": "Keep"}, "B": {"action": "Buy More"}} b = CX.build_bounds(hs, dec, ["N"], {"A": 0.15, "B": 0.15, "N": 0.03}) assert b["A"][1] == 0.008 and b["B"][1] == 0.15 and b["N"] == (0.0, 0.03) # plain Keep: no add / Buy More may double / starter backed = CX.build_bounds(hs, {"A": {"action": "Keep", "conviction": 60}, "B": {"action": "Buy More"}}, ["N"], {"A": 0.15, "B": 0.15, "N": 0.03}) assert abs(backed["A"][1] - 0.02) < 1e-9 # a backed Keep below the band is rounded up to a real position, no more U = ["G", "U", "N"] cov = np.array([[0.10, 0.01, 0.02], [0.01, 0.02, 0.005], [0.02, 0.005, 0.06]]) cur = {"G": 0.7, "U": 0.0, "N": 0.3} mu = {"G": 0.05, "U": 0.03, "N": 0.05} bounds = {"G": (0.2, 0.8), "U": (0.0, 0.6), "N": (0.1, 0.5)} sec = {"G": "Technology", "U": "Utilities", "N": "Technology"} betas = {"G": 1.4, "U": 0.4, "N": 1.2} free = CX.optimize_target(cur, mu, cov, U, bounds, sec, 0.9, 1.0, {}, tax_weight=0.0) kept = CX.optimize_target(cur, mu, cov, U, bounds, sec, 0.9, 1.0, {}, tax_weight=0.0, betas=betas, beta_floor=1.0) beta = lambda w: sum(w[t] * betas[t] for t in U) assert beta(free) < 1.0 < beta(kept) + 1e-3 def test_bounds_follow_verdicts(): from src import construct as CX hs = [{"symbol": "A", "weight": 0.05}, {"symbol": "B", "weight": 0.05}, {"symbol": "C", "weight": 0.05}] dec = {"A": {"action": "Keep", "no_add": True, "protect": True}, # lukewarm/flagged: no adds, not churned "B": {"action": "Keep", "no_add": False, "protect": True}, # backed Keep: may grow "C": {"action": "Keep", "no_add": False, "protect": False}} # in the overweight sector: may be cut in half b = CX.build_bounds(hs, dec, [], {"A": 0.15, "B": 0.15, "C": 0.15}) assert abs(b["A"][0] - 0.04) < 1e-9 and abs(b["A"][1] - 0.05) < 1e-9 assert abs(b["B"][1] - 0.05) < 1e-9 and abs(b["C"][0] - 0.025) < 1e-9 # capital goes to Buy More verdicts, never to a plain Keep bm = CX.build_bounds(hs, {"A": {"action": "Buy More"}, "B": {"action": "Keep", "watch": True}, "C": {"action": "Keep"}}, [], {"A": 0.15, "B": 0.15, "C": 0.15}) assert bm["A"][1] > 0.05 and abs(bm["B"][0] - 0.0125) < 1e-9 # a watch name can be cut to a quarter def test_trim_bound_forces_a_reduction(): from src import construct as CX b = CX.build_bounds([{"symbol": "T", "weight": 0.05}], {"T": {"action": "Trim"}}, [], {"T": 0.15}) assert abs(b["T"][1] - 0.03) < 1e-9 # min(0.8w, max(w − 2% band, 0.5w)) on a 5% position def test_frozen_caps_are_not_lifted_by_feasibility(): import numpy as np from src import construct as CX U = ["A", "B", "N"] cov = np.array([[0.05, 0.01, 0.0], [0.01, 0.05, 0.0], [0.0, 0.0, 0.04]]) cur = {"A": 0.6, "B": 0.4, "N": 0.0} mu = {"A": 0.05, "B": 0.05, "N": 0.05} bounds = {"A": (0.3, 0.6), "B": (0.2, 0.4), "N": (0.0, 0.07)} # uppers sum to 1.07 but A/B frozen at current w = CX.optimize_target(cur, mu, cov, U, bounds, {"A": "X", "B": "Y", "N": "Z"}, 0.9, 0.5, {}, tax_weight=0.0) assert w["A"] <= 0.6 + 1e-6 and w["B"] <= 0.4 + 1e-6 bounds2 = {"A": (0.3, 0.5), "B": (0.2, 0.3), "N": (0.0, 0.07)} # uppers sum to 0.87: only N may grow w2 = CX.optimize_target(cur, mu, cov, U, bounds2, {"A": "X", "B": "Y", "N": "Z"}, 0.9, 0.9, {}, tax_weight=0.0) assert w2["A"] <= 0.6 + 1e-6 and w2["B"] <= 0.4 + 1e-6 and w2["N"] > 0.07 def test_shortfall_slows_the_trim_instead_of_inflating_starters(): import numpy as np from src import construct as CX U = ["BIG", "K1", "K2", "N1", "N2"] cov = np.eye(5) * 0.05 cur = {"BIG": 0.40, "K1": 0.30, "K2": 0.30, "N1": 0.0, "N2": 0.0} mu = {t: 0.05 for t in U} bounds = {"BIG": (0.0, 0.15), "K1": (0.24, 0.30), "K2": (0.24, 0.30), "N1": (0.0, 0.07), "N2": (0.0, 0.07)} sec = {t: t for t in U} w = CX.optimize_target(cur, mu, cov, U, bounds, sec, 0.9, 0.9, {}, tax_weight=0.0) assert w["N1"] <= 0.10 + 1e-6 and w["N2"] <= 0.10 + 1e-6 # starters capped assert w["K1"] <= 0.30 + 1e-6 and w["K2"] <= 0.30 + 1e-6 # frozen Keeps not lifted assert 0.15 < w["BIG"] <= 0.275 + 1e-6 # the Trim is staged but keeps ≥ half its cut assert "BIG" in CX.optimize_target.relaxed bounds_s = dict(bounds, BIG=(0.0, 0.0)) # a SELL is never un-sold by feasibility ws = CX.optimize_target(cur, mu, cov, U, bounds_s, sec, 0.9, 0.9, {}, tax_weight=0.0) assert ws["BIG"] < 1e-6 def test_solution_respects_upper_bounds_after_normalisation(): import numpy as np from src import construct as CX U = ["A", "B", "N"] cov = np.eye(3) * 0.05 cur = {"A": 0.5, "B": 0.5, "N": 0.0} mu = {"A": 0.02, "B": 0.02, "N": 0.30} bounds = {"A": (0.3, 0.5), "B": (0.3, 0.5), "N": (0.0, 0.10)} w = CX.optimize_target(cur, mu, cov, U, bounds, {"A": "X", "B": "Y", "N": "Z"}, 0.9, 0.9, {}, tax_weight=0.0) assert w["N"] <= 0.10 + 1e-6 and abs(sum(w.values()) - 1) < 1e-6 def test_staged_trim_state_never_leaks_into_the_next_book(): """A stale `relaxed` list once made an unrelated 25-name book's narrative say two names from the previous run were 'trimmed in stages'.""" U = ["BIG", "K1", "K2", "N1", "N2"] cov = np.eye(5) * 0.04 cur = {"BIG": 0.40, "K1": 0.30, "K2": 0.30, "N1": 0.0, "N2": 0.0} mu = {t: 0.05 for t in U} bounds = {"BIG": (0.0, 0.15), "K1": (0.24, 0.30), "K2": (0.24, 0.30), "N1": (0.0, 0.07), "N2": (0.0, 0.07)} sec = {t: t for t in U} CX.optimize_target(cur, mu, cov, U, bounds, sec, 0.9, 0.9, {}, tax_weight=0.0) assert "BIG" in CX.optimize_target.relaxed easy = {"BIG": (0.0, 0.40), "K1": (0.0, 0.40), "K2": (0.0, 0.40), "N1": (0.0, 0.2), "N2": (0.0, 0.2)} CX.optimize_target(cur, mu, cov, U, easy, sec, 0.9, 0.9, {}, tax_weight=0.0) assert CX.optimize_target.relaxed == [] def test_a_trim_never_becomes_a_full_exit(): hs = [{"symbol": "A", "weight": 0.0208, "data": {}}, {"symbol": "B", "weight": 0.9792, "data": {}}] dec = {"A": {"action": "Trim"}, "B": {"action": "Keep"}} b = CX.build_bounds(hs, dec, [], {"A": 0.15, "B": 0.6}, "full") lo, hi = b["A"] assert 0.008 <= lo <= hi <= 0.0105 def test_size_trim_lands_at_the_cap_not_at_forty_percent(): hs = [{"symbol": "META", "weight": 0.242, "data": {}}, {"symbol": "B", "weight": 0.758, "data": {}}] dec = {"META": {"action": "Trim", "size_note": "at 24% of the book against a 15% cap, size itself is the risk — trim back toward the cap"}, "B": {"action": "Keep"}} b = CX.build_bounds(hs, dec, [], {"META": 0.15, "B": 0.6}, "full") assert abs(b["META"][0] - 0.12) < 1e-9 and abs(b["META"][1] - 0.15) < 1e-9 def test_unabsorbed_proceeds_become_cash_not_forced_top_ups(): U = ["S", "K1", "K2", "B"] cov = np.eye(4) * 0.04 cur = {"S": 0.30, "K1": 0.35, "K2": 0.30, "B": 0.05} mu = {t: 0.05 for t in U} bounds = {"S": (0.0, 0.0), "K1": (0.30, 0.35), "K2": (0.25, 0.30), "B": (0.05, 0.10)} # only B may grow w = CX.optimize_target(cur, mu, cov, U, bounds, {t: t for t in U}, 0.9, 0.9, {}, tax_weight=0.0) assert CX.optimize_target.cash > 0.15 # ≈ 30% sold, ~10% absorbable (lifted to 10%) → cash assert abs(sum(w.values()) + CX.optimize_target.cash - 1.0) < 1e-3 assert w["K1"] <= 0.35 + 1e-6 and w["K2"] <= 0.30 + 1e-6 # Keeps were not stretched def test_ladder_lift_is_six_percent_in_a_broad_book(): U = [f"K{i}" for i in range(18)] + ["S", "B"] cov = np.eye(20) * 0.04 cur = {**{f"K{i}": 0.04 for i in range(18)}, "S": 0.26, "B": 0.02} mu = {t: 0.05 for t in U} bounds = {**{f"K{i}": (0.03, 0.04) for i in range(18)}, "S": (0.0, 0.0), "B": (0.02, 0.05)} w = CX.optimize_target(cur, mu, cov, U, bounds, {t: t for t in U}, 0.9, 0.9, {}, tax_weight=0.0) assert w["B"] <= 0.06 + 1e-6 and CX.optimize_target.cash > 0.15 # the add stops at 6%; the rest is cash