Download tests/test_construct.py from s-ttp/vitals: direct link, hf CLI and curl.
- Browser
- Download file 12.6 kB
-
https://huggingface.co/spaces/s-ttp/vitals/resolve/main/tests/test_construct.py
- Command line
-
hf download hf://spaces/s-ttp/vitals/tests/test_construct.py
-
curl -L -o test_construct.py https://huggingface.co/spaces/s-ttp/vitals/resolve/main/tests/test_construct.py
12.6 kB
| """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 | |