vitals / tests /test_construct.py
s-ttp's picture
Manager desk: conviction-ranked funding with cash residual, watch tier, min-size exits, tax-lot plan, watchlist with entry conditions, catalysts, post-trade risk, 12m scenarios, rule-of-40 valuation for loss-makers, specific tripwires
6c814fb verified
Raw History Blame Contribute Delete
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