Spaces:

s-ttp
/

vitals

Sleeping

vitals / tests /test_validator.py

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 about 2 months ago

Raw History Blame Contribute Delete

17.4 kB

"""Deterministic decision layer and LLM-output coercion — the code that must never let a
model's mistake reach the user. No network."""
from src import personas as P
from src.market_data import sanitize
from src.revival import _for_narrative, plan_mode


def _panel(**convs):
    return {k: {"X": {"conviction": v, "key_point": "", "key_risk": ""}} for k, v in convs.items()}


def test_actions_from_conviction_bands():
    assert P.decide_actions(_panel(a=70, b=66), ["X"])["X"]["action"] == "Buy More"
    assert P.decide_actions(_panel(a=50, b=52), ["X"])["X"]["action"] == "Keep"
    assert P.decide_actions(_panel(a=40, b=38), ["X"])["X"]["action"] == "Trim"
    assert P.decide_actions(_panel(a=20, b=25), ["X"])["X"]["action"] == "Sell"


def test_ejection_rule_three_low_votes():
    d = P.decide_actions(_panel(a=25, b=28, c=29, d=60, e=65), ["X"])["X"]
    assert d["action"] == "Sell" and d["votes_low"] == 3


def test_weight_aware_no_buy_more_when_oversized():
    d = P.decide_actions(_panel(a=70, b=68), ["X"] + [f"H{i}" for i in range(19)], {"X": 0.35})["X"]
    assert d["action"] == "Keep" and "cap" in d["size_note"]


def test_fund_cap_relaxed():
    d = P.decide_actions(_panel(a=70, b=68), ["X", "Y"], {"X": 0.44}, funds={"X"})["X"]
    assert d["action"] == "Buy More"


def test_small_book_not_trimmed_mechanically():
    tickers = ["X", "A", "B", "C", "D", "E"]
    d = P.decide_actions(_panel(a=48, b=50), tickers, {"X": 0.16})["X"]
    assert d["action"] == "Keep"
    d2 = P.decide_actions(_panel(a=48, b=50), tickers, {"X": 0.33})["X"]
    assert d2["action"] == "Trim"


def test_sample_merge_uses_median():
    s = [{"X": {"conviction": 40, "key_point": "a", "key_risk": ""}},
         {"X": {"conviction": 60, "key_point": "b", "key_risk": ""}},
         {"X": {"conviction": 65, "key_point": "c", "key_risk": ""}}]
    m = P._merge_samples(s)
    assert m["X"]["conviction"] == 60 and m["X"]["n_samples"] == 3 and m["X"]["key_point"] == "a"


def test_coerce_cio_tolerates_garbage():
    assert P.coerce_cio("not json") is None
    out = P.coerce_cio({"verdict": 12, "highlights": [{"severity": "BAD!", "title": "x"}, "junk"],
                        "rationale": {"aapl": 5}, "alternatives": [{"ticker": "msft"}, {"nope": 1}]})
    assert out is not None
    assert out["verdict"] == "12" and out["highlights"][0]["severity"] == "warn"
    assert out["rationale"] == {"AAPL": "5"} and out["alternatives"] == [{"ticker": "MSFT", "reason": ""}]


def test_coerce_strategist_shapes():
    out = P.coerce_strategist({"strategy_read": "x", "gaps": "not a list", "findings": None})
    assert out is not None
    assert out["gaps"] == [] and out["findings"] == [] and out["style_tilt"] == ""


def test_sanitize_drops_unit_errors():
    d = {"price": 100.0, "pe": 36.0, "de": 954.0, "div_yield": 86.0, "roe": 1.4, "hi52": 20000.0,
         "eps_ttm": 0.01}
    sanitize(d)
    assert d["de"] is None and d["div_yield"] is None and d["hi52"] is None and d["eps_ttm"] is None
    assert d["pe"] == 36.0 and d["roe"] == 1.4 and len(d["data_flags"]) == 4


def test_plan_mode_never_revival_for_winners():
    alerts = [{"status": "ALERT"}, {"status": "ALERT"}]
    assert plan_mode(30, 0.5, alerts) == "enhancement"
    assert plan_mode(30, -0.2, alerts) == "revival"
    assert plan_mode(30, 0.02, alerts) == "revival"


def test_narrative_payload_is_percent_safe():
    p = _for_narrative({"outlook": {"current": {"expected_excess_12m": 0.0708, "top_sector": ["T", 0.47]},
                                    "stress": [{"portfolio": -0.31, "benchmark": -0.2}]},
                        "orders": [{"delta_pct": -0.05}]})
    assert p["outlook"]["current"]["expected_excess_12m"] == "7.1%"
    assert p["orders"][0]["delta_pct"].endswith("of the book")


def test_isin_resolution_prefers_primary_listing(monkeypatch):
    from src import portfolio as PF
    import json
    rows = [{"ticker": "BSP", "exchCode": "GR", "marketSector": "Equity"},
            {"ticker": "BA/", "exchCode": "LN", "marketSector": "Equity"},
            {"ticker": "BAESY", "exchCode": "US", "marketSector": "Equity"}]

    class _R:
        def __init__(self, data): self.data = json.dumps([{"data": data}]).encode()
        def read(self): return self.data
        def __enter__(self): return self
        def __exit__(self, *a): return False
    monkeypatch.setattr(PF, "_quote_exists", lambda s: {"name": s, "price": 1.0} if s == "BA.L" else None)
    import urllib.request
    monkeypatch.setattr(urllib.request, "urlopen", lambda req, timeout=20, context=None: _R(rows))
    assert PF.resolve_isin("GB0002634946")[0] == "BA.L"
    assert PF.resolve_symbol("GB0002634946")["symbol"] == "BA.L"


def test_av_venue_map_and_reverse():
    from src import market_data as MD
    import os
    os.environ.setdefault("ALPHAVANTAGE_API_KEY", "x"); MD.AV_KEY = MD.AV_KEY or "x"
    assert MD.av_symbol("RELIANCE.NS") == "RELIANCE.BSE" and MD.av_symbol("600519.SS") == "600519.SHH"
    assert MD.av_symbol("BMW.F") == "BMW.FRK" and MD.av_symbol("GALP.LS") == "GALP.LIS" and MD.av_symbol("BRK-B") == "BRK.B"
    assert MD.av_symbol("7203.T") is None and MD.av_symbol("^NSEI") is None
    assert MD.AV_TO_YAHOO["SHZ"] == ".SZ" and MD.AV_TO_YAHOO["BSE"] == ".BO"


def test_resolver_prefers_home_market_over_otc_adr(monkeypatch):
    from src import portfolio as PF
    monkeypatch.setattr(PF, "av_symbol_search", lambda q: [
        {"av": "RLNIY", "yahoo": "RLNIY", "name": "Reliance Industries ADR", "region": "United States", "currency": "USD", "type": "Equity"},
        {"av": "RELIANCE.BSE", "yahoo": "RELIANCE.BO", "name": "Reliance Industries", "region": "India/Bombay", "currency": "INR", "type": "Equity"}])
    monkeypatch.setattr(PF, "_quote_exists", lambda s: {"name": s, "price": 1300.0} if s == "RELIANCE.NS" else None)
    monkeypatch.setattr(PF, "resolve_isin", lambda i: [])
    assert PF.resolve_symbol("Reliance Industries")["symbol"] == "RELIANCE.NS"


def test_resolver_demotes_secondary_venues(monkeypatch):
    from src import portfolio as PF
    monkeypatch.setattr(PF, "av_symbol_search", lambda q: [
        {"av": "SQ3.FRK", "yahoo": "SQ3.F", "name": "Block Inc", "region": "Frankfurt", "currency": "EUR", "type": "Equity"}])
    monkeypatch.setattr(PF, "_quote_exists", lambda s: {"name": s, "price": 1.0} if s in ("SQ3.F", "XYZ") else None)
    monkeypatch.setattr(PF, "resolve_isin", lambda i: [])
    monkeypatch.setattr(PF, "call_llm", lambda *a, **k: '{"ticker": "XYZ", "name": "Block"}')
    assert PF.resolve_symbol("SQ")["symbol"] == "XYZ"                 # not the Frankfurt line
    assert PF.resolve_symbol("Block Inc")["symbol"] == "SQ3.F"        # name tokens all present → the only listing found
    assert PF.resolve_symbol("Block payments")["symbol"] == "XYZ"     # 'payments' absent from the name → LLM path


def test_adr_bridge_uses_ratios_and_derives_local_eps(monkeypatch):
    from src import market_data as MD
    monkeypatch.setattr(MD, "NDL_KEY", "x")
    monkeypatch.setattr(MD, "av_symbol_search", lambda q: [
        {"av": "HDFCBANK.BSE", "yahoo": "HDFCBANK.BO", "name": "HDFC Bank Ltd", "region": "India/Bombay", "currency": "INR", "type": "Equity"},
        {"av": "HDB", "yahoo": "HDB", "name": "HDFC Bank Ltd", "region": "United States", "currency": "USD", "type": "Equity"}])
    monkeypatch.setattr(MD, "_sharadar_ticker_meta", lambda t: {"category": "ADR Common Stock", "location": "India"} if t == "HDB" else {})
    monkeypatch.setattr(MD, "fetch_sharadar", lambda t: {"pe": 16.5, "pb": 1.42, "roe": 0.089, "de": 5.3, "eps_ttm": 137.8, "bvps": 1593.0, "fund_asof": "2026-07-29"})
    d = {"ticker": "HDFCBANK.NS", "name": "HDFC BANK LTD", "price": 950.0, "quote_type": "EQUITY", "eps_ttm": 999.0}
    MD.bridge_adr_fundamentals(d)
    assert d["fund_source"].startswith("sharadar (via HDB") and d["roe"] == 0.089
    assert abs(d["eps_ttm"] - 950 / 16.5) < 1e-3 and abs(d["bvps"] - 950 / 1.42) < 1e-3   # derived, not copied


def test_lots_average_and_net_sales(monkeypatch):
    from src import portfolio as PF
    monkeypatch.setattr(PF, "resolve_symbol", lambda q: {"symbol": "NVDA", "name": "NVIDIA", "price": 180.0,
                                                          "data": {"currency": "USD"}})
    monkeypatch.setattr(PF, "fx_to_usd", lambda c: 1.0)
    rows = [{"query": "NVDA", "entry": 100, "qty": 10, "bought": "2024-01-10"},
            {"query": "nvidia", "entry": 150, "qty": 10, "bought": "2026-06-01"},
            {"query": "NVDA", "entry": 170, "qty": -5}]                       # negative qty = sale
    hs, errs = PF.build_holdings(rows)
    assert not errs and len(hs) == 1
    h = hs[0]
    assert h["entry"] == 125.0 and h["qty"] == 15 and h["n_lots"] == 3 and h["bought"] == "2024-01-10"
    assert h["market_value"] == 15 * 180.0
    # no quantities → plain mean, unsized, with a nudge
    hs, errs = PF.build_holdings([{"query": "NVDA", "entry": 100}, {"query": "NVDA", "entry": 200}])
    assert hs[0]["entry"] == 150.0 and hs[0]["qty"] is None and "quantities" in hs[0]["lot_note"]
    # fully sold → skipped with an error, not a phantom holding
    hs, errs = PF.build_holdings([{"query": "NVDA", "entry": 100, "qty": 10}, {"query": "NVDA", "entry": 120, "qty": -10}])
    assert not hs and "closed" in errs[0]["error"]


def test_tax_note_names_short_term_lots():
    from src.metrics import tax_note
    from datetime import date, timedelta
    recent = (date.today() - timedelta(days=100)).isoformat()
    h = {"symbol": "NVDA", "entry": 125.0, "price": 180.0, "qty": 20, "pl_pct": 0.44, "bought": "2023-01-10",
         "data": {"currency": "USD"},
         "lots": [{"entry": 100, "qty": 10, "bought": "2023-01-10", "side": "buy"},
                  {"entry": 150, "qty": 10, "bought": recent, "side": "buy"}]}
    note = tax_note(h, "Trim")
    assert "2 lots" in note and "1 SHORT-TERM" in note and recent in note


def test_buy_more_blocked_in_overweight_sector():
    panel = {k: {"X": {"conviction": v, "key_point": "", "key_risk": ""}, "Y": {"conviction": v, "key_point": "", "key_risk": ""}}
             for k, v in {"a": 70, "b": 68}.items()}
    d = P.decide_actions(panel, ["X", "Y"], {"X": 0.1, "Y": 0.1}, sector_over={"X": "Technology is 76% of the book"})
    assert d["X"]["action"] == "Keep" and "deepens" in d["X"]["size_note"] and d["Y"]["action"] == "Buy More"


def test_ballast_fund_detection():
    from src.metrics import is_ballast_fund
    assert is_ballast_fund({"symbol": "BND", "data": {"quote_type": "ETF", "name": "Vanguard Total Bond Market"}})
    assert is_ballast_fund({"symbol": "ZZZ", "data": {"quote_type": "ETF", "name": "iShares 1-3 Year Treasury Bond ETF"}})
    assert not is_ballast_fund({"symbol": "VOO", "data": {"quote_type": "ETF", "name": "Vanguard S&P 500 ETF"}})
    assert not is_ballast_fund({"symbol": "BND", "data": {"quote_type": "EQUITY", "name": "Bond Corp"}})


def test_switch_pairs_follow_order_plan():
    from src import revival as R
    hs = [{"symbol": "A", "data": {"sector": "Technology"}}]
    dec = {"A": {"action": "Sell", "conviction": 20}}
    cands = [{"ticker": "P", "sector": "Healthcare", "trend_ok": True, "flags": [], "live": {}, "fit": {}},
             {"ticker": "Q", "sector": "Energy", "trend_ok": True, "flags": [], "live": {}, "fit": {}}]
    assert R.switch_pairs(hs, dec, cands)[0]["in"] == "P"
    assert R.switch_pairs(hs, dec, cands, prefer=["Q"])[0]["in"] == "Q"


def test_tiny_trim_becomes_exit_or_keep():
    tickers = ["X"] + [f"H{i}" for i in range(30)]
    weak = P.decide_actions(_panel(a=36, b=38), tickers, {"X": 0.006})["X"]
    assert weak["action"] == "Sell" and "exit" in weak["size_note"]
    band = P.decide_actions(_panel(a=43, b=44), tickers, {"X": 0.006})["X"]
    assert band["action"] == "Sell"                                   # below-average and tiny: exit, don't shave
    option = P.decide_actions(_panel(a=43, b=44), tickers, {"X": 0.006},
                              asym={"X": {"expected_return": 0.3, "up_down_ratio": 3.0, "clears_hurdle": True}})["X"]
    assert option["action"] == "Keep" and "option" in option["size_note"]   # asymmetric tiny bet is kept
    tilt = P.decide_actions(_panel(a=55, b=61, c=65, d=47, e=45), tickers, {"X": 0.006}, cyclical={"X"})["X"]
    assert tilt["action"] == "Keep"                                   # a backed name stays even if a tilt said Trim


def test_rationale_writer_cannot_contradict_the_action():
    from src.personas import _strip_action
    assert _strip_action("Keep because revenue grew 28%, although the position is large.") == "Revenue grew 28%, although the position is large."
    assert _strip_action("Sell — profits collapsed and the finances are fragile.") == "Profits collapsed and the finances are fragile."
    assert _strip_action("The business keeps growing and the price is fair.") == "The business keeps growing and the price is fair."


def test_oversized_lukewarm_position_is_trimmed_toward_cap():
    tickers = ["X"] + [f"H{i}" for i in range(46)]        # 47 names → cap 15%
    d = P.decide_actions(_panel(a=55, b=61, c=65, d=47, e=45), tickers, {"X": 0.227})["X"]
    assert d["action"] == "Trim" and "toward the cap" in d["size_note"]
    strong = P.decide_actions(_panel(a=70, b=68, c=66, d=67, e=65), tickers, {"X": 0.227})["X"]
    assert strong["action"] == "Keep"                       # strongly backed: size note only


def test_two_lens_dissent_only_for_cyclicals():
    tickers = ["X"] + [f"H{i}" for i in range(20)]
    panel = _panel(a=35, b=69, c=72, d=83, e=35)
    from src.metrics import is_cyclical
    assert is_cyclical("Technology", "Semiconductor Memory") and is_cyclical("Energy", None)
    assert not is_cyclical("Healthcare", "Drug Manufacturers") and not is_cyclical("Technology", "Software - Application")
    cyc = P.decide_actions(panel, tickers, {"X": 0.079}, cyclical={"X"})["X"]
    assert cyc["action"] == "Trim" and "cyclical" in cyc["size_note"]
    secular = P.decide_actions(panel, tickers, {"X": 0.079}, cyclical=set())["X"]
    assert secular["action"] in ("Keep", "Buy More")                            # a winner is not sold early
    ok = P.decide_actions(_panel(a=45, b=69, c=72, d=83, e=40), tickers, {"X": 0.079}, cyclical={"X"})["X"]
    assert ok["action"] in ("Keep", "Buy More")


def test_gated_add_and_final_tiny_pass():
    tickers = ["X"] + [f"H{i}" for i in range(20)]
    bad = {"X": {"expected_return": 0.02, "up_down_ratio": 1.1, "clears_hurdle": False}}
    d = P.decide_actions(_panel(a=70, b=68), tickers, {"X": 0.03}, asym=bad)["X"]
    assert d["action"] == "Keep" and d["gated_add"] is True
    # a Trim produced by the cyclical dissent tilt on a 0.4% name still resolves to keep-or-exit
    tiny = P.decide_actions(_panel(a=35, b=69, c=72, d=83, e=35), tickers, {"X": 0.004}, cyclical={"X"})["X"]
    assert tiny["action"] in ("Keep", "Sell") and "friction" in tiny["size_note"] or "exit" in tiny["size_note"]


def test_damage_tilts_run_before_add_gates():
    tickers = ["X"] + [f"H{i}" for i in range(20)]
    bad = {"X": {"expected_return": -0.55, "up_down_ratio": 0.0, "clears_hurdle": False}}
    d = P.decide_actions(_panel(a=35, b=76, c=68, d=83, e=35), tickers, {"X": 0.079}, asym=bad, cyclical={"X"})["X"]
    assert d["action"] == "Trim" and "cyclical" in d["size_note"]      # MU pattern: dissent beats the hurdle gate


def test_sub_two_percent_trim_exits_or_holds_never_shaves():
    from src.personas import decide_actions
    panel = {k: {"PG": {"conviction": 44}, "BIG": {"conviction": 44}} for k in ("value", "risk", "growth", "technical", "macro")}
    d = decide_actions(panel, ["PG", "BIG"], {"PG": 0.015, "BIG": 0.05}, set())
    assert d["PG"]["action"] == "Sell" and "exit cleanly" in d["PG"]["size_note"]
    assert d["BIG"]["action"] == "Trim"


def test_tiny_threshold_scales_with_the_book():
    from src.personas import decide_actions
    tick = ["SBUX"] + [f"H{i}" for i in range(46)]                      # 47 names → band 1.06%
    panel = {k: {t: {"conviction": 44} for t in tick} for k in ("value", "risk", "growth", "technical", "macro")}
    d = decide_actions(panel, tick, {"SBUX": 0.019, **{f"H{i}": 0.02 for i in range(46)}}, set())
    assert d["SBUX"]["action"] == "Trim"                                 # 1.9% is above the band in a 47-name book


def test_sub_band_keep_with_weak_backing_exits_and_watch_tier_marks_sources_of_funds():
    from src.personas import decide_actions
    tick = ["XYZ", "SPOT"] + [f"H{i}" for i in range(45)]                 # 47 names → band 1.06%
    conv = {"XYZ": 49, "SPOT": 45, **{f"H{i}": 55 for i in range(45)}}
    panel = {k: {t: {"conviction": c} for t, c in conv.items()} for k in ("value", "risk", "growth", "technical", "macro")}
    w = {"XYZ": 0.008, "SPOT": 0.049, **{f"H{i}": 0.02 for i in range(45)}}
    asym = {"SPOT": {"expected_return": -0.41, "clears_hurdle": False}, "XYZ": {"expected_return": -0.6, "clears_hurdle": False}}
    d = decide_actions(panel, tick, w, set(), asym)
    assert d["XYZ"]["action"] == "Sell" and "dead money" in d["XYZ"]["size_note"]      # sub-band, weak, no option value
    assert d["SPOT"]["action"] == "Keep" and d["SPOT"]["watch"] and "source of funds" in d["SPOT"]["size_note"]
    assert d["H0"]["action"] == "Keep" and not d["H0"]["watch"]

	"""Deterministic decision layer and LLM-output coercion — the code that must never let a
	model's mistake reach the user. No network."""
	from src import personas as P
	from src.market_data import sanitize
	from src.revival import _for_narrative, plan_mode


	def _panel(**convs):
	return {k: {"X": {"conviction": v, "key_point": "", "key_risk": ""}} for k, v in convs.items()}


	def test_actions_from_conviction_bands():
	assert P.decide_actions(_panel(a=70, b=66), ["X"])["X"]["action"] == "Buy More"
	assert P.decide_actions(_panel(a=50, b=52), ["X"])["X"]["action"] == "Keep"
	assert P.decide_actions(_panel(a=40, b=38), ["X"])["X"]["action"] == "Trim"
	assert P.decide_actions(_panel(a=20, b=25), ["X"])["X"]["action"] == "Sell"


	def test_ejection_rule_three_low_votes():
	d = P.decide_actions(_panel(a=25, b=28, c=29, d=60, e=65), ["X"])["X"]
	assert d["action"] == "Sell" and d["votes_low"] == 3


	def test_weight_aware_no_buy_more_when_oversized():
	d = P.decide_actions(_panel(a=70, b=68), ["X"] + [f"H{i}" for i in range(19)], {"X": 0.35})["X"]
	assert d["action"] == "Keep" and "cap" in d["size_note"]


	def test_fund_cap_relaxed():
	d = P.decide_actions(_panel(a=70, b=68), ["X", "Y"], {"X": 0.44}, funds={"X"})["X"]
	assert d["action"] == "Buy More"


	def test_small_book_not_trimmed_mechanically():
	tickers = ["X", "A", "B", "C", "D", "E"]
	d = P.decide_actions(_panel(a=48, b=50), tickers, {"X": 0.16})["X"]
	assert d["action"] == "Keep"
	d2 = P.decide_actions(_panel(a=48, b=50), tickers, {"X": 0.33})["X"]
	assert d2["action"] == "Trim"


	def test_sample_merge_uses_median():
	s = [{"X": {"conviction": 40, "key_point": "a", "key_risk": ""}},
	{"X": {"conviction": 60, "key_point": "b", "key_risk": ""}},
	{"X": {"conviction": 65, "key_point": "c", "key_risk": ""}}]
	m = P._merge_samples(s)
	assert m["X"]["conviction"] == 60 and m["X"]["n_samples"] == 3 and m["X"]["key_point"] == "a"


	def test_coerce_cio_tolerates_garbage():
	assert P.coerce_cio("not json") is None
	out = P.coerce_cio({"verdict": 12, "highlights": [{"severity": "BAD!", "title": "x"}, "junk"],
	"rationale": {"aapl": 5}, "alternatives": [{"ticker": "msft"}, {"nope": 1}]})
	assert out is not None
	assert out["verdict"] == "12" and out["highlights"][0]["severity"] == "warn"
	assert out["rationale"] == {"AAPL": "5"} and out["alternatives"] == [{"ticker": "MSFT", "reason": ""}]


	def test_coerce_strategist_shapes():
	out = P.coerce_strategist({"strategy_read": "x", "gaps": "not a list", "findings": None})
	assert out is not None
	assert out["gaps"] == [] and out["findings"] == [] and out["style_tilt"] == ""


	def test_sanitize_drops_unit_errors():
	d = {"price": 100.0, "pe": 36.0, "de": 954.0, "div_yield": 86.0, "roe": 1.4, "hi52": 20000.0,
	"eps_ttm": 0.01}
	sanitize(d)
	assert d["de"] is None and d["div_yield"] is None and d["hi52"] is None and d["eps_ttm"] is None
	assert d["pe"] == 36.0 and d["roe"] == 1.4 and len(d["data_flags"]) == 4


	def test_plan_mode_never_revival_for_winners():
	alerts = [{"status": "ALERT"}, {"status": "ALERT"}]
	assert plan_mode(30, 0.5, alerts) == "enhancement"
	assert plan_mode(30, -0.2, alerts) == "revival"
	assert plan_mode(30, 0.02, alerts) == "revival"


	def test_narrative_payload_is_percent_safe():
	p = _for_narrative({"outlook": {"current": {"expected_excess_12m": 0.0708, "top_sector": ["T", 0.47]},
	"stress": [{"portfolio": -0.31, "benchmark": -0.2}]},
	"orders": [{"delta_pct": -0.05}]})
	assert p["outlook"]["current"]["expected_excess_12m"] == "7.1%"
	assert p["orders"][0]["delta_pct"].endswith("of the book")


	def test_isin_resolution_prefers_primary_listing(monkeypatch):
	from src import portfolio as PF
	import json
	rows = [{"ticker": "BSP", "exchCode": "GR", "marketSector": "Equity"},
	{"ticker": "BA/", "exchCode": "LN", "marketSector": "Equity"},
	{"ticker": "BAESY", "exchCode": "US", "marketSector": "Equity"}]

	class _R:
	def __init__(self, data): self.data = json.dumps([{"data": data}]).encode()
	def read(self): return self.data
	def __enter__(self): return self
	def __exit__(self, *a): return False
	monkeypatch.setattr(PF, "_quote_exists", lambda s: {"name": s, "price": 1.0} if s == "BA.L" else None)
	import urllib.request
	monkeypatch.setattr(urllib.request, "urlopen", lambda req, timeout=20, context=None: _R(rows))
	assert PF.resolve_isin("GB0002634946")[0] == "BA.L"
	assert PF.resolve_symbol("GB0002634946")["symbol"] == "BA.L"


	def test_av_venue_map_and_reverse():
	from src import market_data as MD
	import os
	os.environ.setdefault("ALPHAVANTAGE_API_KEY", "x"); MD.AV_KEY = MD.AV_KEY or "x"
	assert MD.av_symbol("RELIANCE.NS") == "RELIANCE.BSE" and MD.av_symbol("600519.SS") == "600519.SHH"
	assert MD.av_symbol("BMW.F") == "BMW.FRK" and MD.av_symbol("GALP.LS") == "GALP.LIS" and MD.av_symbol("BRK-B") == "BRK.B"
	assert MD.av_symbol("7203.T") is None and MD.av_symbol("^NSEI") is None
	assert MD.AV_TO_YAHOO["SHZ"] == ".SZ" and MD.AV_TO_YAHOO["BSE"] == ".BO"


	def test_resolver_prefers_home_market_over_otc_adr(monkeypatch):
	from src import portfolio as PF
	monkeypatch.setattr(PF, "av_symbol_search", lambda q: [
	{"av": "RLNIY", "yahoo": "RLNIY", "name": "Reliance Industries ADR", "region": "United States", "currency": "USD", "type": "Equity"},
	{"av": "RELIANCE.BSE", "yahoo": "RELIANCE.BO", "name": "Reliance Industries", "region": "India/Bombay", "currency": "INR", "type": "Equity"}])
	monkeypatch.setattr(PF, "_quote_exists", lambda s: {"name": s, "price": 1300.0} if s == "RELIANCE.NS" else None)
	monkeypatch.setattr(PF, "resolve_isin", lambda i: [])
	assert PF.resolve_symbol("Reliance Industries")["symbol"] == "RELIANCE.NS"


	def test_resolver_demotes_secondary_venues(monkeypatch):
	from src import portfolio as PF
	monkeypatch.setattr(PF, "av_symbol_search", lambda q: [
	{"av": "SQ3.FRK", "yahoo": "SQ3.F", "name": "Block Inc", "region": "Frankfurt", "currency": "EUR", "type": "Equity"}])
	monkeypatch.setattr(PF, "_quote_exists", lambda s: {"name": s, "price": 1.0} if s in ("SQ3.F", "XYZ") else None)
	monkeypatch.setattr(PF, "resolve_isin", lambda i: [])
	monkeypatch.setattr(PF, "call_llm", lambda a, *k: '{"ticker": "XYZ", "name": "Block"}')
	assert PF.resolve_symbol("SQ")["symbol"] == "XYZ" # not the Frankfurt line
	assert PF.resolve_symbol("Block Inc")["symbol"] == "SQ3.F" # name tokens all present → the only listing found
	assert PF.resolve_symbol("Block payments")["symbol"] == "XYZ" # 'payments' absent from the name → LLM path


	def test_adr_bridge_uses_ratios_and_derives_local_eps(monkeypatch):
	from src import market_data as MD
	monkeypatch.setattr(MD, "NDL_KEY", "x")
	monkeypatch.setattr(MD, "av_symbol_search", lambda q: [
	{"av": "HDFCBANK.BSE", "yahoo": "HDFCBANK.BO", "name": "HDFC Bank Ltd", "region": "India/Bombay", "currency": "INR", "type": "Equity"},
	{"av": "HDB", "yahoo": "HDB", "name": "HDFC Bank Ltd", "region": "United States", "currency": "USD", "type": "Equity"}])
	monkeypatch.setattr(MD, "_sharadar_ticker_meta", lambda t: {"category": "ADR Common Stock", "location": "India"} if t == "HDB" else {})
	monkeypatch.setattr(MD, "fetch_sharadar", lambda t: {"pe": 16.5, "pb": 1.42, "roe": 0.089, "de": 5.3, "eps_ttm": 137.8, "bvps": 1593.0, "fund_asof": "2026-07-29"})
	d = {"ticker": "HDFCBANK.NS", "name": "HDFC BANK LTD", "price": 950.0, "quote_type": "EQUITY", "eps_ttm": 999.0}
	MD.bridge_adr_fundamentals(d)
	assert d["fund_source"].startswith("sharadar (via HDB") and d["roe"] == 0.089
	assert abs(d["eps_ttm"] - 950 / 16.5) < 1e-3 and abs(d["bvps"] - 950 / 1.42) < 1e-3 # derived, not copied


	def test_lots_average_and_net_sales(monkeypatch):
	from src import portfolio as PF
	monkeypatch.setattr(PF, "resolve_symbol", lambda q: {"symbol": "NVDA", "name": "NVIDIA", "price": 180.0,
	"data": {"currency": "USD"}})
	monkeypatch.setattr(PF, "fx_to_usd", lambda c: 1.0)
	rows = [{"query": "NVDA", "entry": 100, "qty": 10, "bought": "2024-01-10"},
	{"query": "nvidia", "entry": 150, "qty": 10, "bought": "2026-06-01"},
	{"query": "NVDA", "entry": 170, "qty": -5}] # negative qty = sale
	hs, errs = PF.build_holdings(rows)
	assert not errs and len(hs) == 1
	h = hs[0]
	assert h["entry"] == 125.0 and h["qty"] == 15 and h["n_lots"] == 3 and h["bought"] == "2024-01-10"
	assert h["market_value"] == 15 * 180.0
	# no quantities → plain mean, unsized, with a nudge
	hs, errs = PF.build_holdings([{"query": "NVDA", "entry": 100}, {"query": "NVDA", "entry": 200}])
	assert hs[0]["entry"] == 150.0 and hs[0]["qty"] is None and "quantities" in hs[0]["lot_note"]
	# fully sold → skipped with an error, not a phantom holding
	hs, errs = PF.build_holdings([{"query": "NVDA", "entry": 100, "qty": 10}, {"query": "NVDA", "entry": 120, "qty": -10}])
	assert not hs and "closed" in errs[0]["error"]


	def test_tax_note_names_short_term_lots():
	from src.metrics import tax_note
	from datetime import date, timedelta
	recent = (date.today() - timedelta(days=100)).isoformat()
	h = {"symbol": "NVDA", "entry": 125.0, "price": 180.0, "qty": 20, "pl_pct": 0.44, "bought": "2023-01-10",
	"data": {"currency": "USD"},
	"lots": [{"entry": 100, "qty": 10, "bought": "2023-01-10", "side": "buy"},
	{"entry": 150, "qty": 10, "bought": recent, "side": "buy"}]}
	note = tax_note(h, "Trim")
	assert "2 lots" in note and "1 SHORT-TERM" in note and recent in note


	def test_buy_more_blocked_in_overweight_sector():
	panel = {k: {"X": {"conviction": v, "key_point": "", "key_risk": ""}, "Y": {"conviction": v, "key_point": "", "key_risk": ""}}
	for k, v in {"a": 70, "b": 68}.items()}
	d = P.decide_actions(panel, ["X", "Y"], {"X": 0.1, "Y": 0.1}, sector_over={"X": "Technology is 76% of the book"})
	assert d["X"]["action"] == "Keep" and "deepens" in d["X"]["size_note"] and d["Y"]["action"] == "Buy More"


	def test_ballast_fund_detection():
	from src.metrics import is_ballast_fund
	assert is_ballast_fund({"symbol": "BND", "data": {"quote_type": "ETF", "name": "Vanguard Total Bond Market"}})
	assert is_ballast_fund({"symbol": "ZZZ", "data": {"quote_type": "ETF", "name": "iShares 1-3 Year Treasury Bond ETF"}})
	assert not is_ballast_fund({"symbol": "VOO", "data": {"quote_type": "ETF", "name": "Vanguard S&P 500 ETF"}})
	assert not is_ballast_fund({"symbol": "BND", "data": {"quote_type": "EQUITY", "name": "Bond Corp"}})


	def test_switch_pairs_follow_order_plan():
	from src import revival as R
	hs = [{"symbol": "A", "data": {"sector": "Technology"}}]
	dec = {"A": {"action": "Sell", "conviction": 20}}
	cands = [{"ticker": "P", "sector": "Healthcare", "trend_ok": True, "flags": [], "live": {}, "fit": {}},
	{"ticker": "Q", "sector": "Energy", "trend_ok": True, "flags": [], "live": {}, "fit": {}}]
	assert R.switch_pairs(hs, dec, cands)[0]["in"] == "P"
	assert R.switch_pairs(hs, dec, cands, prefer=["Q"])[0]["in"] == "Q"


	def test_tiny_trim_becomes_exit_or_keep():
	tickers = ["X"] + [f"H{i}" for i in range(30)]
	weak = P.decide_actions(_panel(a=36, b=38), tickers, {"X": 0.006})["X"]
	assert weak["action"] == "Sell" and "exit" in weak["size_note"]
	band = P.decide_actions(_panel(a=43, b=44), tickers, {"X": 0.006})["X"]
	assert band["action"] == "Sell" # below-average and tiny: exit, don't shave
	option = P.decide_actions(_panel(a=43, b=44), tickers, {"X": 0.006},
	asym={"X": {"expected_return": 0.3, "up_down_ratio": 3.0, "clears_hurdle": True}})["X"]
	assert option["action"] == "Keep" and "option" in option["size_note"] # asymmetric tiny bet is kept
	tilt = P.decide_actions(_panel(a=55, b=61, c=65, d=47, e=45), tickers, {"X": 0.006}, cyclical={"X"})["X"]
	assert tilt["action"] == "Keep" # a backed name stays even if a tilt said Trim


	def test_rationale_writer_cannot_contradict_the_action():
	from src.personas import _strip_action
	assert _strip_action("Keep because revenue grew 28%, although the position is large.") == "Revenue grew 28%, although the position is large."
	assert _strip_action("Sell — profits collapsed and the finances are fragile.") == "Profits collapsed and the finances are fragile."
	assert _strip_action("The business keeps growing and the price is fair.") == "The business keeps growing and the price is fair."


	def test_oversized_lukewarm_position_is_trimmed_toward_cap():
	tickers = ["X"] + [f"H{i}" for i in range(46)] # 47 names → cap 15%
	d = P.decide_actions(_panel(a=55, b=61, c=65, d=47, e=45), tickers, {"X": 0.227})["X"]
	assert d["action"] == "Trim" and "toward the cap" in d["size_note"]
	strong = P.decide_actions(_panel(a=70, b=68, c=66, d=67, e=65), tickers, {"X": 0.227})["X"]
	assert strong["action"] == "Keep" # strongly backed: size note only


	def test_two_lens_dissent_only_for_cyclicals():
	tickers = ["X"] + [f"H{i}" for i in range(20)]
	panel = _panel(a=35, b=69, c=72, d=83, e=35)
	from src.metrics import is_cyclical
	assert is_cyclical("Technology", "Semiconductor Memory") and is_cyclical("Energy", None)
	assert not is_cyclical("Healthcare", "Drug Manufacturers") and not is_cyclical("Technology", "Software - Application")
	cyc = P.decide_actions(panel, tickers, {"X": 0.079}, cyclical={"X"})["X"]
	assert cyc["action"] == "Trim" and "cyclical" in cyc["size_note"]
	secular = P.decide_actions(panel, tickers, {"X": 0.079}, cyclical=set())["X"]
	assert secular["action"] in ("Keep", "Buy More") # a winner is not sold early
	ok = P.decide_actions(_panel(a=45, b=69, c=72, d=83, e=40), tickers, {"X": 0.079}, cyclical={"X"})["X"]
	assert ok["action"] in ("Keep", "Buy More")


	def test_gated_add_and_final_tiny_pass():
	tickers = ["X"] + [f"H{i}" for i in range(20)]
	bad = {"X": {"expected_return": 0.02, "up_down_ratio": 1.1, "clears_hurdle": False}}
	d = P.decide_actions(_panel(a=70, b=68), tickers, {"X": 0.03}, asym=bad)["X"]
	assert d["action"] == "Keep" and d["gated_add"] is True
	# a Trim produced by the cyclical dissent tilt on a 0.4% name still resolves to keep-or-exit
	tiny = P.decide_actions(_panel(a=35, b=69, c=72, d=83, e=35), tickers, {"X": 0.004}, cyclical={"X"})["X"]
	assert tiny["action"] in ("Keep", "Sell") and "friction" in tiny["size_note"] or "exit" in tiny["size_note"]


	def test_damage_tilts_run_before_add_gates():
	tickers = ["X"] + [f"H{i}" for i in range(20)]
	bad = {"X": {"expected_return": -0.55, "up_down_ratio": 0.0, "clears_hurdle": False}}
	d = P.decide_actions(_panel(a=35, b=76, c=68, d=83, e=35), tickers, {"X": 0.079}, asym=bad, cyclical={"X"})["X"]
	assert d["action"] == "Trim" and "cyclical" in d["size_note"] # MU pattern: dissent beats the hurdle gate


	def test_sub_two_percent_trim_exits_or_holds_never_shaves():
	from src.personas import decide_actions
	panel = {k: {"PG": {"conviction": 44}, "BIG": {"conviction": 44}} for k in ("value", "risk", "growth", "technical", "macro")}
	d = decide_actions(panel, ["PG", "BIG"], {"PG": 0.015, "BIG": 0.05}, set())
	assert d["PG"]["action"] == "Sell" and "exit cleanly" in d["PG"]["size_note"]
	assert d["BIG"]["action"] == "Trim"


	def test_tiny_threshold_scales_with_the_book():
	from src.personas import decide_actions
	tick = ["SBUX"] + [f"H{i}" for i in range(46)] # 47 names → band 1.06%
	panel = {k: {t: {"conviction": 44} for t in tick} for k in ("value", "risk", "growth", "technical", "macro")}
	d = decide_actions(panel, tick, {"SBUX": 0.019, **{f"H{i}": 0.02 for i in range(46)}}, set())
	assert d["SBUX"]["action"] == "Trim" # 1.9% is above the band in a 47-name book


	def test_sub_band_keep_with_weak_backing_exits_and_watch_tier_marks_sources_of_funds():
	from src.personas import decide_actions
	tick = ["XYZ", "SPOT"] + [f"H{i}" for i in range(45)] # 47 names → band 1.06%
	conv = {"XYZ": 49, "SPOT": 45, **{f"H{i}": 55 for i in range(45)}}
	panel = {k: {t: {"conviction": c} for t, c in conv.items()} for k in ("value", "risk", "growth", "technical", "macro")}
	w = {"XYZ": 0.008, "SPOT": 0.049, **{f"H{i}": 0.02 for i in range(45)}}
	asym = {"SPOT": {"expected_return": -0.41, "clears_hurdle": False}, "XYZ": {"expected_return": -0.6, "clears_hurdle": False}}
	d = decide_actions(panel, tick, w, set(), asym)
	assert d["XYZ"]["action"] == "Sell" and "dead money" in d["XYZ"]["size_note"] # sub-band, weak, no option value
	assert d["SPOT"]["action"] == "Keep" and d["SPOT"]["watch"] and "source of funds" in d["SPOT"]["size_note"]
	assert d["H0"]["action"] == "Keep" and not d["H0"]["watch"]