| """Portfolio-aware recommendation engine — needs diagnosis and marginal-fit scoring.""" |
| import numpy as np |
| import pandas as pd |
|
|
| from src import metrics as M |
|
|
|
|
| def _pm(**kw): |
| m = {"sector_allocation": {"Technology": 0.7, "Consumer Cyclical": 0.3}, "effective_n": 2.5, |
| "realized_beta": 1.6, "n_holdings": 3} |
| m.update(kw) |
| return m |
|
|
|
|
| def test_diagnose_needs_reads_the_book(): |
| risk = {"stress": [{"name": "COVID crash", "window": "2020-02-19 → 2020-03-23", "portfolio": -0.40}], |
| "factor_tilts": {"tilts": {"value": -0.3, "quality": 0.2}}} |
| hs = [{"weight": 0.5, "dcf": {"upside": -0.4}}, {"weight": 0.5, "dcf": {"upside": -0.3}}] |
| needs = M.diagnose_needs(_pm(), risk, hs, {"loss_tolerance": "20%", "objective": "income"}) |
| assert "Healthcare" in needs["sector_gaps"]["sectors"] |
| assert needs["sector_overweight"]["sector"] == "Technology" |
| assert needs["diversifier"]["weight"] > 0.5 |
| assert needs["defensive"]["weight"] == 1.0 and "breaches" in needs["defensive"]["label"] |
| assert "income" in needs and "value" in needs |
|
|
|
|
| def test_diagnose_needs_healthy_book_falls_back_to_quality(): |
| pm = _pm(sector_allocation={s: 1 / 11 for s in M.ALL_SECTORS}, effective_n=12, realized_beta=1.0) |
| needs = M.diagnose_needs(pm, {}, [], None) |
| assert list(needs) == ["quality"] |
|
|
|
|
| def _hist_pair(corr_target: float, n=700, seed=3): |
| rng = np.random.default_rng(seed) |
| idx = pd.bdate_range(end="2026-09-01", periods=n) |
| base = rng.normal(0, 0.012, n) |
| other = corr_target * base + np.sqrt(1 - corr_target ** 2) * rng.normal(0, 0.012, n) |
| px = lambda r: 100 * np.cumprod(1 + r) |
| return pd.DataFrame({"AAA": px(base), "BBB": px(base * 0.9), "CAND": px(other)}, index=idx) |
|
|
|
|
| def test_candidate_fit_prefers_diversifier_that_fills_gap(): |
| hist = _hist_pair(0.1) |
| hs = [{"symbol": "AAA", "weight": 0.6, "data": {}}, {"symbol": "BBB", "weight": 0.4, "data": {}}] |
| pm = _pm(effective_n=1.9, realized_beta=1.5, sector_allocation={"Technology": 1.0}) |
| needs = M.diagnose_needs(pm, {}, [], {"loss_tolerance": "20%"}) |
| port = M.portfolio_returns(hs, hist) |
| good = {"ticker": "CAND", "sector": "Healthcare", "rank_pct": 80, "signals": [], |
| "live": {"beta": 0.7, "div_yield": 1.0}, "upside": 0.1} |
| bad = {"ticker": "CAND", "sector": "Technology", "rank_pct": 95, "signals": [], |
| "live": {"beta": 1.8, "div_yield": 0.0}, "upside": 0.1} |
| fg = M.candidate_fit(good, hs, pm, {}, needs, port, hist) |
| fb = M.candidate_fit(bad, hs, pm, {}, needs, port, hist) |
| assert fg["score"] > fb["score"] |
| assert any("adds Healthcare" in f for f in fg["fixes"]) |
| assert any("correlation" in f for f in fg["fixes"]) and fg["corr_to_book"] < 0.5 |
| assert fg["deltas"]["effective_n"][1] > fg["deltas"]["effective_n"][0] |
| assert any("beta" in f for f in fg["fixes"]) |
|
|
|
|
| def test_conservative_income_prefers_payer_over_high_beta_diversifier(): |
| hist = _hist_pair(-0.1) |
| hs = [{"symbol": "AAA", "weight": 0.5, "data": {}}, {"symbol": "BBB", "weight": 0.5, "data": {}}] |
| pm = _pm(effective_n=2.0, realized_beta=0.1, sector_allocation={"Real Estate": 1.0}) |
| needs = M.diagnose_needs(pm, {}, [], {"risk": "conservative", "objective": "income"}) |
| port = M.portfolio_returns(hs, hist) |
| tech = {"ticker": "CAND", "sector": "Technology", "rank_pct": 80, "signals": [], |
| "live": {"beta": 1.3, "div_yield": 1.0}, "upside": 0.1} |
| payer = {"ticker": "CAND", "sector": "Consumer Defensive", "rank_pct": 80, "signals": [], |
| "live": {"beta": 0.5, "div_yield": 6.0}, "upside": 0.1} |
| assert M.candidate_fit(payer, hs, pm, {}, needs, port, hist)["score"] > \ |
| M.candidate_fit(tech, hs, pm, {}, needs, port, hist)["score"] |
|
|
|
|
| def test_switch_pairs_avoid_overweight_sector(): |
| from src.revival import switch_pairs |
| hs = [{"symbol": "PLTR", "weight": 0.4, "data": {"sector": "Technology", "price": 1.0}}] |
| dec = {"PLTR": {"action": "Trim", "conviction": 38}} |
| cands = [{"ticker": "TSM", "sector": "Technology", "trend_ok": True, "live": {}}, |
| {"ticker": "ED", "sector": "Utilities", "trend_ok": True, "live": {}}] |
| assert switch_pairs(hs, dec, cands)[0]["in"] == "TSM" |
| assert switch_pairs(hs, dec, cands, {"sector_overweight": {"sector": "Technology"}})[0]["in"] == "ED" |
|
|
|
|
| def test_peak_cycle_and_smart_money_guards(): |
| hs = [{"symbol": "AAA", "weight": 1.0, "data": {}}] |
| pm = _pm(sector_allocation={"Technology": 1.0}, effective_n=1.0) |
| needs = M.diagnose_needs(pm, {}, [], None) |
| clean = {"ticker": "X", "sector": "Industrials", "rank_pct": 94, "signals": [], |
| "live": {"eps_growth": 0.08, "sma200": 100.0, "price": 104.0, "beta": 1.0}, "upside": 0.0} |
| expd = {"ticker": "X", "sector": "Industrials", "rank_pct": 94, "signals": [], "smart_money_neg": True, |
| "live": {"eps_growth": 0.51, "sma200": 158.0, "price": 186.0, "beta": 1.0}, "upside": 0.0} |
| fc = M.candidate_fit(clean, hs, pm, {}, needs, None, None) |
| fe = M.candidate_fit(expd, hs, pm, {}, needs, None, None) |
| assert fe["score"] < fc["score"] - 0.9 |
| assert any("peak-cycle" in c for c in fe["cautions"]) and any("informed money" in c for c in fe["cautions"]) |
| assert fc["cautions"] == [] |
|
|
|
|
| def test_strong_ok_requires_clean_committee_verdict(): |
| from src.revival import strong_ok |
| assert strong_ok({"committee": {"conviction": 64}, "fit": {"cautions": ["peak-cycle"]}}) |
| assert not strong_ok({"committee": {"conviction": 60}, "fit": {"cautions": ["peak-cycle"]}}) |
| assert strong_ok({"committee": {"conviction": 56}, "fit": {"cautions": []}}) |
| assert not strong_ok({"committee": {"conviction": 52}, "fit": {"cautions": []}}) |
| assert not strong_ok({"committee": {"conviction": 45}, "fit": {}}) |
| assert not strong_ok({"committee": {"conviction": 70}, "fit": {}, "trend_ok": False}) |
| assert strong_ok({"fit": {"cautions": []}}) and not strong_ok({"fit": {"cautions": ["extended"]}}) |
|
|
|
|
| def test_candidate_fit_flags_penalized(): |
| hs = [{"symbol": "AAA", "weight": 1.0, "data": {}}] |
| c = {"ticker": "X", "sector": "Energy", "rank_pct": 99, "signals": [], "live": {}, "flags": ["VALUE_TRAP"]} |
| assert M.candidate_fit(c, hs, _pm(), {}, {"quality": {"weight": 0.5}}, None, None)["score"] < 0 |
|
|
|
|
| def test_exclusions_short_tokens_match_whole_words(monkeypatch): |
| from src import orchestrator as O |
| monkeypatch.setattr(O, "fetch_holding", lambda t: {"name": {"CL": "Colgate-Palmolive", "MO": "Altria Group"}.get(t, t), "industry": ""}) |
| cands = [{"ticker": "CL", "sector": "Consumer Defensive"}, {"ticker": "MO", "sector": "Consumer Defensive"}, |
| {"ticker": "PM", "sector": "Consumer Defensive"}] |
| kept, rej = O._apply_exclusions(cands, {"exclusions": "tobacco, MO"}) |
| assert [c["ticker"] for c in kept] == ["CL", "PM"] and rej[0]["ticker"] == "MO" |
|
|
|
|
| def test_instrument_gets_need_credit_and_measured_beta(): |
| hist = _hist_pair(0.0) |
| hist["SPY"] = hist["AAA"] |
| hs = [{"symbol": "AAA", "weight": 1.0, "data": {}}] |
| pm = _pm(effective_n=1.0, realized_beta=1.5, sector_allocation={"Technology": 1.0}) |
| needs = M.diagnose_needs(pm, {}, [], {"loss_tolerance": "20%"}) |
| port = M.portfolio_returns(hs, hist) |
| shy = {"ticker": "CAND", "sector": "Diversified fund", "instrument": True, "need_target": "defensive", |
| "signals": [], "live": {}} |
| f = M.candidate_fit(shy, hs, pm, {}, needs, port, hist) |
| assert any("ballast" in x for x in f["fixes"]) and isinstance(shy["live"].get("beta"), float) |
| assert f["score"] > 0.5 |
|
|
|
|
| def test_european_benchmark_and_tax_note(): |
| from src.market_data import benchmark_for |
| assert benchmark_for([{"symbol": "VOD.L", "weight": .4}, {"symbol": "SAP.DE", "weight": .3}, {"symbol": "MC.PA", "weight": .3}]) == "^STOXX" |
| assert benchmark_for([{"symbol": "VOD.L", "weight": .7}, {"symbol": "SAP.DE", "weight": .3}]) == "^FTSE" |
| assert benchmark_for([{"symbol": "AAPL", "weight": .5}, {"symbol": "SAP.DE", "weight": .5}]) == "ACWI" |
| h = {"symbol": "VOD.L", "price": 130, "entry": 100, "qty": 100, "pl_pct": 0.3, "data": {"currency": "GBp"}, "bought": "2026-08-01"} |
| note = M.tax_note(h, "Trim") or "" |
| assert "jurisdiction" in note and "SHORT-TERM" not in note |
|
|
|
|
| def test_europanel_scoring_is_cross_sectional(): |
| import pandas as pd |
| from src import europanel as EP |
| df = pd.DataFrame({"pe": [10, 20, 40], "pb": [1, 2, 4], "ev_ebitda": [6, 10, 20], "div_yield": [5, 3, 1], |
| "roe": [0.2, 0.1, 0.05], "roa": [0.1, 0.05, 0.02], "op_margin": [0.3, 0.2, 0.1], "de": [0.2, 0.5, 1.0], |
| "mom_12_1": [0.1, 0.2, 0.3], "ret_26w": [0.05, 0.1, 0.2], "ret_13w": [0.02, 0.05, 0.1], |
| "vol_13w": [0.15, 0.25, 0.4], "dist_52w_high": [-0.05, -0.1, -0.2]}, index=["CHEAP.L", "MID.DE", "DEAR.PA"]) |
| sc = EP._score(df) |
| assert sc.loc["CHEAP.L", "n_earnings_yield"] > sc.loc["DEAR.PA", "n_earnings_yield"] |
| assert sc.loc["CHEAP.L", "pctile"] > sc.loc["DEAR.PA", "pctile"] |
| assert -1 <= sc["n_mom_12_1"].min() and sc["n_mom_12_1"].max() <= 1 |
|
|
|
|
| def test_uk_stamp_duty_in_order_plan(): |
| from src import revival as R |
| hs = [{"symbol": "VOD.L", "weight": 1.0, "market_value": 10000.0, "price": 1.0, "fx_usd": 1.0, |
| "data": {"avg_volume": 1e9, "quote_type": "EQUITY"}}] |
| target = [{"ticker": "VOD.L", "current_w": 1.0, "target_w": 0.6, "delta": -0.4, "is_new": False}, |
| {"ticker": "AZN.L", "current_w": 0.0, "target_w": 0.4, "delta": 0.4, "is_new": True}] |
| R.order_plan.quote_types = {"VOD.L": "EQUITY", "AZN.L": "EQUITY"} |
| orders = {o["ticker"]: o for o in R.order_plan(hs, target)} |
| assert orders["AZN.L"]["cost_note"] == "UK stamp duty 0.5%" and orders["AZN.L"]["est_cost_usd"] == 4 + 20 |
| assert "cost_note" not in orders["VOD.L"] |
|
|
|
|
| def test_europanel_resolves_home_market_primary_lines(monkeypatch): |
| from src import europanel as EP |
| monkeypatch.setattr(EP, "av_symbol_search", lambda q: [ |
| {"av": "SAP", "yahoo": "SAP", "name": "Sap SE", "region": "United States", "currency": "USD", "type": "Equity"}, |
| {"av": "SAP.FRK", "yahoo": "SAP.F", "name": "SAP SE", "region": "Frankfurt", "currency": "EUR", "type": "Equity"}, |
| {"av": "0NW4.LON", "yahoo": "0NW4.L", "name": "SAP SE", "region": "United Kingdom", "currency": "EUR", "type": "Equity"}, |
| {"av": "SAP.DEX", "yahoo": "SAP.DE", "name": "SAP SE", "region": "XETRA", "currency": "EUR", "type": "Equity"}]) |
| assert EP._resolve_name("SAP SE")["yahoo"] == "SAP.DE" |
| monkeypatch.setattr(EP, "av_symbol_search", lambda q: [ |
| {"av": "YASM.LON", "yahoo": "YASM.L", "name": "ASML (ASML) OPTIONS ETP", "region": "United Kingdom", "currency": "GBX", "type": "ETF"}, |
| {"av": "ASML.AMS", "yahoo": "ASML.AS", "name": "ASML Holding N.V.", "region": "Amsterdam", "currency": "EUR", "type": "Equity"}]) |
| assert EP._resolve_name("ASML Holding NV")["yahoo"] == "ASML.AS" |
|
|
|
|
| def test_europanel_universe_dedupes_vendor_spellings(monkeypatch): |
| from src import europanel as EP |
| profiles = {"VGK": [{"description": "UNILEVER PLC", "weight": "0.009"}], |
| "SPEU": [{"description": "Unilever plc", "weight": "0.0085"}, {"description": "RELX PLC", "weight": "0.004"}], |
| "DBEU": [], "FEZ": []} |
| monkeypatch.setattr(EP, "_av_call", lambda params, raw=False: {"holdings": profiles[params["symbol"]]}) |
| u = EP._universe(10) |
| assert [r["name"] for r in u] == ["UNILEVER PLC", "RELX PLC"] and u[0]["weight"] == 0.009 |
|
|
|
|
| def test_legal_form_prefers_home_venue(): |
| from src import europanel as EP |
| assert EP._home_sfx("HSBC HOLDINGS PLC") == {"L"} |
| assert "DE" in EP._home_sfx("Novo Nordisk AS Class B") and "F" in EP._home_sfx("Novo Nordisk AS Class B") |
| assert EP._home_sfx("Some Company") == set() |
|
|
|
|
| def test_fundamentals_scores_on_synthetic_statements(): |
| from src import fundamentals as FU |
| y1 = {"assets": 1000, "liabilities": 400, "netinc": 120, "ncfo": 150, "debt": 100, "currentratio": 1.8, "sharesbas": 100, |
| "grossmargin": 0.45, "revenue": 900, "workingcapital": 200, "retearn": 300, "ebit": 160, "marketcap": 2400, |
| "receivables": 90, "gp": 405, "assetsc": 400, "ppnenet": 300, "depamor": 40, "sgna": 150, "capex": -50, "fcf": 100, |
| "sbcomp": 10, "dps": 1.0, "roic": 0.18, "invcap": 700, "taxexp": 30, "ebt": 150, "epsdil": 1.2, "bvps": 6.0, "pe": 20} |
| y0 = {**y1, "netinc": 100, "ncfo": 120, "debt": 120, "currentratio": 1.6, "sharesbas": 102, "grossmargin": 0.43, |
| "revenue": 820, "receivables": 80, "gp": 353, "epsdil": 0.98, "dps": 0.9} |
| older = [{**y0, "revenue": 700, "epsdil": 0.9, "dps": 0.8}, {**y0, "revenue": 650, "epsdil": 0.85, "dps": 0.7}, |
| {**y0, "revenue": 600, "epsdil": 0.8, "dps": 0.6}, {**y0, "revenue": 560, "epsdil": 0.7, "dps": 0.5, "invcap": 500, "ebit": 100}] |
| a = [y1, y0] + older |
| f = FU.piotroski(a) |
| assert f["score"] == 9 |
| assert FU.altman_z(y1, 2400) > 2.99 |
| assert FU.beneish_m(a) < -1.78 |
| e = FU.economics(a, 24.0, 2400, "Technology", "Software - Application") |
| assert e["cash_conversion"] == round(100 / 120, 2) and e["dividend_streak_years"] == 5 |
| assert abs(e["buyback_yield_1y"] - (1 - 100 / 102)) < 1e-3 and e["fcf_payout"] == 1.0 |
| assert e.get("greenblatt_ey") is None |
| assert e["lynch_category"] in ("stalwart", "fast grower") |
| us = {"piotroski": f, "altman_z": 5.0, "beneish_m": -2.5, "econ": e, "insiders": {}, "earnings": {}} |
| assert FU.forensic_flags(us) == [] and FU.sell_triggers({}, us) |
|
|
|
|
| def test_harvest_pairs_pick_correlated_non_identical_substitute(): |
| import numpy as np, pandas as pd |
| from src import revival as R |
| idx = pd.bdate_range("2024-01-01", periods=300) |
| rng = np.random.default_rng(3) |
| base = rng.normal(0, 0.01, 300) |
| px = lambda r: pd.Series(100 * np.cumprod(1 + r), index=idx) |
| port_hist = pd.DataFrame({"PFE": px(base), "SPY": px(rng.normal(0, 0.01, 300))}) |
| cand_hist = pd.DataFrame({"XLV": px(0.9 * base + rng.normal(0, 0.003, 300)), |
| "XLE": px(rng.normal(0, 0.01, 300))}) |
| hs = [{"symbol": "PFE", "entry": 42.0, "price": 25.0, "qty": 100, "pl_pct": -0.4, "fx_usd": 1.0, |
| "data": {"sector": "Healthcare"}}] |
| dec = {"PFE": {"action": "Keep"}} |
| cands = [{"ticker": "XLV", "sector": "Healthcare", "instrument": True, "live": {"name": "Health Care SPDR"}}, |
| {"ticker": "XLE", "sector": "Energy", "instrument": True, "live": {"name": "Energy SPDR"}}] |
| out = R.harvest_pairs(hs, dec, cands, cand_hist, port_hist) |
| assert len(out) == 1 and out[0]["buy"] == "XLV" and out[0]["loss_usd"] == 1700 and "31 days" in out[0]["note"] |
| assert R.harvest_pairs(hs, {"PFE": {"action": "Trim"}}, cands, cand_hist, port_hist) == [] |
|
|
|
|
| def test_cautious_book_marks_needs_and_mutes_gaps(): |
| from src import metrics as M |
| pm = {"sector_allocation": {"Consumer Defensive": 0.6, "Communication Services": 0.4}, "effective_n": 6.5} |
| n = M.diagnose_needs(pm, {}, [], {"risk": "conservative", "horizon": "under 3 years"}) |
| assert "_cautious" in n and n["_cautious"]["label"] is None |
| n2 = M.diagnose_needs(pm, {}, [], {"risk": "aggressive"}) |
| assert "_cautious" not in n2 and n["sector_gaps"]["weight"] < n2["sector_gaps"]["weight"] |
|
|
|
|
| def test_ballast_floor_in_bounds(): |
| from src import construct as CX |
| hs = [{"symbol": "BND", "weight": 0.14}, {"symbol": "JNJ", "weight": 0.14}] |
| dec = {"BND": {"action": "Keep", "ballast": True}, "JNJ": {"action": "Trim"}} |
| b = CX.build_bounds(hs, dec, [], {"BND": 0.3, "JNJ": 0.3}) |
| assert b["BND"][0] == 0.14 and abs(b["JNJ"][0] - 0.056) < 1e-9 |
|
|
|
|
| def test_importer_reads_plain_tables_without_llm(): |
| from src.importer import parse_table |
| csv = "Symbol,Description,Quantity,Avg Cost,Current Price,Market Value\nAAPL,Apple,10,150.25,320,3200\nNFLX,Netflix,5,\"1,210.50\",78,390\nTOTAL,,,,,3590\n" |
| rows = parse_table(csv) |
| assert rows == [{"query": "AAPL", "entry": 150.25, "qty": 10.0}, {"query": "NFLX", "entry": 1210.5, "qty": 5.0}] |
| assert parse_table("Date,Action,Symbol,Price,Quantity\n2024-01-01,BUY,AAPL,150,10\n") is None |
| assert parse_table("just some text\nwithout a table") is None |
|
|
|
|
| def test_presplit_entry_is_a_question_never_a_silent_rewrite(monkeypatch): |
| from src import portfolio as PF |
| monkeypatch.setattr(PF, "resolve_symbol", lambda q: {"symbol": "NFLX", "name": "Netflix", "price": 78.0, "data": {"currency": "USD"}}) |
| monkeypatch.setattr(PF, "fx_to_usd", lambda c: 1.0) |
| monkeypatch.setattr(PF, "validate_row", lambda *a, **k: [{"code": "presplit", "message": "NFLX had a 10-for-1 split on 2025-11-17 — x", |
| "suggest": {"field": "entry", "value": 17.1}}]) |
| hs, errs = PF.build_holdings([{"query": "NFLX", "entry": 171, "qty": 5}]) |
| assert not hs and errs[0]["code"] == "needs_decision" and "keep as entered" in errs[0]["error"] |
| hs, _ = PF.build_holdings([{"query": "NFLX", "entry": 171, "qty": 5, "split": "both"}]) |
| assert hs[0]["entry"] == 17.1 and hs[0]["qty"] == 50 and "split-adjusted" in hs[0]["entry_note"] |
| hs, _ = PF.build_holdings([{"query": "NFLX", "entry": 171, "qty": 5, "split": "price"}]) |
| assert hs[0]["entry"] == 17.1 and hs[0]["qty"] == 5 |
| hs, _ = PF.build_holdings([{"query": "NFLX", "entry": 171, "qty": 5, "split": "keep"}]) |
| assert hs[0]["entry"] == 171 and hs[0]["qty"] == 5 and "kept as entered" in hs[0]["entry_note"] |
|
|
|
|
| def test_strong_ok_raises_the_bar_for_flagged_names_in_a_cautious_book(): |
| from src.revival import strong_ok |
| flagged = {"committee": {"conviction": 64}, "fit": {"cautions": ["peak-cycle"]}} |
| strong_ok.cautious = True |
| try: |
| assert not strong_ok(flagged) |
| assert strong_ok({"committee": {"conviction": 67}, "fit": {"cautions": ["peak-cycle"]}}) |
| assert strong_ok({"committee": {"conviction": 56}, "fit": {"cautions": []}}) |
| finally: |
| strong_ok.cautious = False |
| assert strong_ok(flagged) |
|
|
|
|
| def test_harvest_substitutes_a_broad_fund_with_a_broad_fund_only(): |
| from src.revival import harvest_pairs |
| rng = np.random.default_rng(1) |
| base = rng.normal(0, 0.01, 400) |
| px = lambda r: pd.Series(100 * np.exp(np.cumsum(r)), index=pd.date_range("2024-01-01", periods=400)) |
| port = pd.DataFrame({"VOO": px(base)}) |
| cands = pd.DataFrame({"VTI": px(base + rng.normal(0, 0.002, 400)), |
| "XLY": px(base + rng.normal(0, 0.006, 400))}) |
| hs = [{"symbol": "VOO", "entry": 600.0, "price": 400.0, "qty": 10, "pl_pct": -0.33, |
| "data": {"quote_type": "ETF", "sector": None}}] |
| dec = {"VOO": {"action": "Keep"}} |
| only_sector = [{"ticker": "XLY", "sector": "Consumer Cyclical", "instrument": True, "live": {"name": "XLY"}}] |
| assert harvest_pairs(hs, dec, only_sector, cands, port) == [] |
| both = only_sector + [{"ticker": "VTI", "sector": "Diversified fund", "instrument": True, "broad": True, "live": {"name": "VTI"}}] |
| out = harvest_pairs(hs, dec, both, cands, port) |
| assert out and out[0]["buy"] == "VTI" and out[0]["correlation"] >= 0.9 |
|
|
|
|
| def test_distress_flag_disqualifies_a_candidate(): |
| from src.revival import strong_ok |
| assert not strong_ok({"committee": {"conviction": 64}, "fit": {"cautions": ["forensic: distress zone (Altman Z 1.45)"]}}) |
| assert strong_ok({"committee": {"conviction": 64}, "fit": {"cautions": ["peak-cycle risk"]}}) |
|
|
|
|
| def test_tax_plan_and_triage_consistency(): |
| from src.revival import tax_plan, triage |
| from datetime import date, timedelta |
| recent = (date.today() - timedelta(days=300)).isoformat() |
| hs = [{"symbol": "META", "weight": 0.24, "price": 700.0, "entry": 135.0, "qty": 95, "fx_usd": 1.0, "market_value": 66500.0, |
| "bought": None, "pl_pct": 4.2, "lots": [{"entry": 100.0, "qty": 50, "bought": "2021-01-05"}, {"entry": 170.0, "qty": 45, "bought": recent}], "data": {}}, |
| {"symbol": "LULU", "weight": 0.003, "price": 98.0, "entry": 276.0, "qty": 3, "fx_usd": 1.0, "market_value": 294.0, "bought": None, "pl_pct": -0.64, "data": {}, "dcf": {}}] |
| orders = [{"side": "SELL", "ticker": "META", "delta_pct": -0.10}] |
| tp = tax_plan(hs, {}, orders, 277000.0) |
| note = tp[0]["note"] |
| assert "170.00 first" in note and "turns long-term" in note and "two tax years" in note |
| dec = {"META": {"action": "Trim", "conviction": 58}, "LULU": {"action": "Keep", "conviction": 38}} |
| t = {x["ticker"]: x["disposition"] for x in triage(hs, dec, bought=set(), sold={"META"})} |
| assert t["LULU"] != "exit to fund" |
|
|
|
|
| def test_shortlist_covers_the_missing_sectors_first(): |
| from src import orchestrator as O |
| meta = lambda t, s, conv: {"ticker": t, "sector": s, "committee": {"conviction": conv}, "fit": {"fixes": [f"adds {s}"], "cautions": []}, |
| "trend_ok": True, "live": {"name": t}} |
| eligible = [meta("TSM", "Technology", 68), meta("NOC", "Industrials", 60), meta("SHW", "Basic Materials", 59), meta("EXC", "Utilities", 57)] |
| holdings = [{"symbol": f"H{i}", "data": {}} for i in range(14)] |
| O._ensure_single_names.needs = {"sector_gaps": {"sectors": ["Industrials", "Basic Materials", "Utilities"]}} |
| try: |
| out = O._ensure_single_names([{"ticker": "TSM", "reason": "x"}], eligible, holdings, set()) |
| finally: |
| O._ensure_single_names.needs = None |
| tickers = [a["ticker"] for a in out] |
| assert tickers[:3] == ["TSM", "NOC", "SHW"] and "EXC" in tickers |
|
|
