""" Known-case tests for peer-group construction (StockLab final engineering pass, Part B2). Pure, dependency-free — genuinely executed. Dual-mode: pytest, or `PYTHONPATH=. python3 tests/test_peer_groups.py`. """ from __future__ import annotations from app.engines.scoring.peer_groups import ( TIER_INDUSTRY, TIER_INDUSTRY_BUCKET, TIER_SECTOR, PeerMetricRow, build_peer_metric_values, industry_median_multiples, ) from app.engines.scoring.percentile import percentile_rank, peer_values_with_fallback def _row(sec, key, val, ind="SOFT", sec_id="TECH", bucket="LARGE"): return PeerMetricRow(security_id=sec, metric_key=key, value=val, industry_id=ind, sector_id=sec_id, market_cap_bucket=bucket) def _universe(): rows = [] # 3 large-cap software, 2 small-cap software, 2 large-cap semis (same TECH sector) for i, v in enumerate([10.0, 20.0, 30.0]): rows.append(_row(f"soft-large-{i}", "roic", v)) for i, v in enumerate([40.0, 50.0]): rows.append(_row(f"soft-small-{i}", "roic", v, bucket="SMALL")) for i, v in enumerate([60.0, 70.0]): rows.append(_row(f"semi-{i}", "roic", v, ind="SEMI")) # one energy company, different sector entirely rows.append(_row("energy-0", "roic", 99.0, ind="OILGAS", sec_id="ENERGY")) return rows # --- tiering --- def test_industry_bucket_tier_holds_only_same_industry_same_bucket_peers(): lookup = build_peer_metric_values(_universe(), "target", "SOFT", "TECH", "LARGE") assert sorted(lookup.by_metric["roic"][TIER_INDUSTRY_BUCKET]) == [10.0, 20.0, 30.0] def test_industry_tier_is_a_superset_of_the_bucket_tier(): lookup = build_peer_metric_values(_universe(), "target", "SOFT", "TECH", "LARGE") tiers = lookup.by_metric["roic"] assert sorted(tiers[TIER_INDUSTRY]) == [10.0, 20.0, 30.0, 40.0, 50.0] assert set(tiers[TIER_INDUSTRY_BUCKET]).issubset(set(tiers[TIER_INDUSTRY])) def test_sector_tier_is_a_superset_of_the_industry_tier(): lookup = build_peer_metric_values(_universe(), "target", "SOFT", "TECH", "LARGE") tiers = lookup.by_metric["roic"] assert sorted(tiers[TIER_SECTOR]) == [10.0, 20.0, 30.0, 40.0, 50.0, 60.0, 70.0] assert 99.0 not in tiers[TIER_SECTOR] # the ENERGY company is in neither def test_widening_grows_the_sample_rather_than_swapping_it(): """If the tiers were disjoint, peer_values_with_fallback()'s widening would replace one small group with a different small group instead of enlarging the sample.""" tiers = build_peer_metric_values(_universe(), "target", "SOFT", "TECH", "LARGE").by_metric["roic"] assert len(tiers[TIER_INDUSTRY_BUCKET]) < len(tiers[TIER_INDUSTRY]) < len(tiers[TIER_SECTOR]) # --- exclusions --- def test_the_target_security_is_never_its_own_peer(): rows = _universe() + [_row("target", "roic", 1000.0)] lookup = build_peer_metric_values(rows, "target", "SOFT", "TECH", "LARGE") for tier in (TIER_INDUSTRY_BUCKET, TIER_INDUSTRY, TIER_SECTOR): assert 1000.0 not in lookup.by_metric["roic"][tier] def test_none_values_are_dropped_not_coerced_to_zero(): """A peer group padded with zeros would push every real company's percentile up.""" rows = _universe() + [PeerMetricRow("other", "roic", None, "SOFT", "TECH", "LARGE")] lookup = build_peer_metric_values(rows, "target", "SOFT", "TECH", "LARGE") assert 0.0 not in lookup.by_metric["roic"][TIER_INDUSTRY] assert len(lookup.by_metric["roic"][TIER_INDUSTRY]) == 5 def test_target_with_no_industry_still_gets_a_sector_tier(): lookup = build_peer_metric_values(_universe(), "target", None, "TECH", "LARGE") tiers = lookup.by_metric["roic"] assert tiers[TIER_INDUSTRY] == [] and tiers[TIER_INDUSTRY_BUCKET] == [] assert len(tiers[TIER_SECTOR]) == 7 def test_target_with_no_classification_at_all_gets_empty_tiers(): lookup = build_peer_metric_values(_universe(), "target", None, None, None) assert lookup.is_empty("roic") def test_is_empty_is_true_for_a_metric_nobody_reports(): lookup = build_peer_metric_values(_universe(), "target", "SOFT", "TECH", "LARGE") assert lookup.is_empty("some_metric_no_peer_has") assert not lookup.is_empty("roic") # --- the bug this part fixes, demonstrated end to end --- def test_empty_peer_set_makes_percentile_rank_return_a_neutral_fifty(): """The defect: with peer_metric_values never supplied, every metric's peer list was empty and percentile_rank returned 50.0 -- so every security scored exactly 50 on every pillar. This test pins the underlying behaviour that made the omission invisible.""" assert percentile_rank(123.0, []) == 50.0 def test_a_real_peer_group_produces_a_real_percentile(): lookup = build_peer_metric_values(_universe(), "target", "SOFT", "TECH", "LARGE") peers, tier = peer_values_with_fallback( lookup.by_metric["roic"][TIER_INDUSTRY_BUCKET], lookup.by_metric["roic"][TIER_INDUSTRY], lookup.by_metric["roic"][TIER_SECTOR], ) # 7 sector peers: 10,20,30,40,50,60,70. A company at 65 beats 6 of them. assert tier == "insufficient_peer_group" # 7 < MIN_PEER_GROUP_SIZE of 8, honestly reported assert abs(percentile_rank(65.0, peers) - 600.0 / 7) < 1e-9 def test_lower_is_better_metric_inverts_the_percentile(): peers = [10.0, 20.0, 30.0, 40.0] assert percentile_rank(5.0, peers, lower_is_better=True) == 100.0 assert percentile_rank(50.0, peers, lower_is_better=True) == 0.0 # --- industry median multiples --- def _multiple_rows(): rows = [] for i, v in enumerate([15.0, 16.0, 17.0, 18.0, 19.0]): rows.append(_row(f"soft-{i}", "pe", v)) for i, v in enumerate([8.0, 9.0]): # only 2 -> below min_group_size rows.append(_row(f"semi-{i}", "pe", v, ind="SEMI")) return rows def test_industry_median_multiples_known_median(): out = industry_median_multiples(_multiple_rows(), ["pe"]) assert out["SOFT"]["pe"] == 17.0 def test_industry_below_min_group_size_is_absent_not_published(): """A 'median P/E' from two companies is not an industry reference.""" out = industry_median_multiples(_multiple_rows(), ["pe"]) assert "SEMI" not in out def test_min_group_size_is_configurable(): out = industry_median_multiples(_multiple_rows(), ["pe"], min_group_size=2) assert out["SEMI"]["pe"] == 8.5 def test_negative_and_zero_multiples_are_excluded(): """A negative P/E is a company with negative earnings, not a cheap company. Including it would drag the median toward zero and produce a negative 'fair value'.""" rows = _multiple_rows() + [ _row("loss-maker", "pe", -30.0), _row("zero", "pe", 0.0), ] out = industry_median_multiples(rows, ["pe"]) assert out["SOFT"]["pe"] == 17.0 # unchanged by the two junk rows def test_rows_without_an_industry_are_ignored(): rows = _multiple_rows() + [_row("no-ind", "pe", 999.0, ind=None)] out = industry_median_multiples(rows, ["pe"]) assert None not in out assert out["SOFT"]["pe"] == 17.0 def test_unrequested_metrics_are_not_computed(): rows = _multiple_rows() + [_row(f"x{i}", "roic", 5.0) for i in range(6)] out = industry_median_multiples(rows, ["pe"]) assert "roic" not in out["SOFT"] def test_empty_universe_returns_empty_dict(): assert industry_median_multiples([], ["pe"]) == {} ALL_TESTS = [v for k, v in sorted(globals().items()) if k.startswith("test_")] if __name__ == "__main__": passed = failed = 0 for t in ALL_TESTS: try: t() print(f"PASS {t.__name__}") passed += 1 except Exception as exc: # noqa: BLE001 print(f"FAIL {t.__name__}: {exc}") failed += 1 print(f"\n{passed}/{passed + failed} passed") raise SystemExit(1 if failed else 0)