| 1 | from datetime import datetime, timezone |
| 2 | from uuid import UUID |
| 3 | |
| 4 | import pytest |
| 5 | |
| 6 | from app.global_opportunity_ranker import GlobalOpportunityRanker, WEIGHTS, RANKER_VERSION |
| 7 | from app.global_scanner import StageBCandidate |
| 8 | from app.sector_relative_strength import SectorRelativeStrengthSnapshot |
| 9 | from app.technical_features import TechnicalFeatureSnapshot |
| 10 | from app.stock_rule_engine import StockRuleEngineResult, AreaScoreResult, RiskOverrideResult, STOCK_RULE_ENGINE_AREA_WEIGHTS |
| 11 | |
| 12 | NOW = datetime(2026, 9, 13, tzinfo=timezone.utc) |
| 13 | |
| 14 | |
| 15 | def inputs(n=1, core=80, tech=60, sector=40, confidence=90): |
| 16 | key = UUID(int=n) |
| 17 | technical = TechnicalFeatureSnapshot(global_instrument_id=key, as_of=NOW, configuration={}, |
| 18 | observation_count=200, history_readiness='FULL', technical_score=tech, confidence=80, |
| 19 | technical_state='UPTREND') |
| 20 | relative = SectorRelativeStrengthSnapshot(global_instrument_id=key, as_of=NOW, configuration={}, |
| 21 | relative_strength_score=sector, confidence=70, |
| 22 | sector_state='NEUTRAL' if sector is not None else 'INSUFFICIENT_DATA') |
| 23 | candidate = StageBCandidate(global_instrument_id=key, symbol='PROVENANCE', pre_score=99, |
| 24 | technical_feature_snapshot=technical, sector_relative_strength_snapshot=relative, |
| 25 | technical_score=tech, sector_score=sector, stage_b_score=99, confidence=99, |
| 26 | score_coverage=100, score_weights={}) |
| 27 | rule = StockRuleEngineResult(global_instrument_id=key, calculated_at=NOW, input_fingerprint='fixture', |
| 28 | overall_score=75, quality_score=99, opportunity_score=99, risk_score=20, |
| 29 | confidence_score=confidence, confidence='HIGH', decision_signal='HOLD', partial=False, |
| 30 | eligibility=dict(full_analysis_allowed=True, partial_analysis_allowed=True, reason='READY'), |
| 31 | area_scores=[AreaScoreResult(area=name, weight=weight, raw_score=core, applicable=True, |
| 32 | status='READY_FRESH') for name, weight in STOCK_RULE_ENGINE_AREA_WEIGHTS.items()]) |
| 33 | return candidate, rule |
| 34 | |
| 35 | |
| 36 | def area(rule, name): |
| 37 | return next(a for a in rule.area_scores if a.area == name) |
| 38 | |
| 39 | |
| 40 | def test_full_evidence_hand_calculation_and_no_double_counting(): |
| 41 | c, r = inputs() |
| 42 | result = GlobalOpportunityRanker().score(c, r) |
| 43 | assert sum(WEIGHTS.values()) == 100 |
| 44 | assert dict(WEIGHTS) == {str(k):v for k,v in STOCK_RULE_ENGINE_AREA_WEIGHTS.items() |
| 45 | if k not in {'PRICE_TECHNICAL', 'SECTOR_MACRO'}} | {'TECHNICAL':7, 'SECTOR':3} |
| 46 | assert result.opportunity_score == 77.4 # (90*80 + 7*60 + 3*40)/100 |
| 47 | assert result.opportunity_confidence == 88.7 # (90*90 + 7*80 + 3*70)/100 |
| 48 | assert result.score_coverage == 100 and result.rank_eligible |
| 49 | assert result.rule_engine_score == 75 and result.ranker_version == RANKER_VERSION |
| 50 | # These aggregate scores and replaced V1 slots cannot alter the composition. |
| 51 | c.pre_score = c.stage_b_score = 0 |
| 52 | r.quality_score = r.opportunity_score = 0 |
| 53 | area(r, 'PRICE_TECHNICAL').raw_score = area(r, 'SECTOR_MACRO').raw_score = 0 |
| 54 | assert GlobalOpportunityRanker().score(c, r) == result |
| 55 | payload = result.model_dump(by_alias=True) |
| 56 | assert 'opportunityScore' in payload and 'rankEligible' in payload |
| 57 | assert 'decisionSignal' not in payload |
| 58 | |
| 59 | |
| 60 | def test_missing_sector_renormalization(): |
| 61 | c, r = inputs(sector=None) |
| 62 | result = GlobalOpportunityRanker().score(c, r) |
| 63 | assert result.opportunity_score == pytest.approx(7620/97) |
| 64 | assert result.score_coverage == 97 |
| 65 | assert result.opportunity_confidence == 86.6 |
| 66 | assert result.rank_eligible and 'UNAVAILABLE:SECTOR' in result.top_negative_reasons |
| 67 | |
| 68 | |
| 69 | def test_zero_is_not_missing(): |
| 70 | c, r = inputs(core=0, tech=0, sector=0) |
| 71 | zero = GlobalOpportunityRanker().score(c, r) |
| 72 | assert zero.opportunity_score == 0 and zero.score_coverage == 100 and zero.rank_eligible |
| 73 | area(r, 'SHAREHOLDING').raw_score = None |
| 74 | missing = GlobalOpportunityRanker().score(c, r) |
| 75 | assert missing.opportunity_score == 0 and missing.score_coverage == 96 |
| 76 | |
| 77 | |
| 78 | def test_non_applicable_shareholding_removed_from_denominator(): |
| 79 | c, r = inputs() |
| 80 | area(r, 'SHAREHOLDING').status = 'NOT_APPLICABLE' |
| 81 | area(r, 'SHAREHOLDING').applicable = False |
| 82 | area(r, 'SHAREHOLDING').raw_score = None |
| 83 | result = GlobalOpportunityRanker().score(c, r) |
| 84 | assert result.score_coverage == 100 |
| 85 | assert result.opportunity_score == pytest.approx(7420/96) |
| 86 | assert result.opportunity_confidence == pytest.approx(8510/96) |
| 87 | assert 'MISSING:SHAREHOLDING' not in result.top_negative_reasons |
| 88 | |
| 89 | |
| 90 | @pytest.mark.parametrize('severity', ['HIGH', 'CRITICAL']) |
| 91 | def test_existing_risk_override_suppresses_eligibility_not_diagnostic_score(severity): |
| 92 | c, r = inputs() |
| 93 | r.risk_overrides = [RiskOverrideResult(code='EXTREME_BALANCE_SHEET_STRESS', severity=severity)] |
| 94 | result = GlobalOpportunityRanker().score(c, r) |
| 95 | assert not result.rank_eligible and result.opportunity_score == 77.4 |
| 96 | assert result.top_negative_reasons[0] == 'V1_RISK_OVERRIDE:EXTREME_BALANCE_SHEET_STRESS' |
| 97 | |
| 98 | |
| 99 | @pytest.mark.parametrize('status', ['READY_STALE', 'CONFLICTING']) |
| 100 | def test_critical_evidence_gate(status): |
| 101 | c, r = inputs() |
| 102 | area(r, 'BALANCE_SHEET').status = status |
| 103 | result = GlobalOpportunityRanker().score(c, r) |
| 104 | assert not result.rank_eligible and result.score_coverage == 91 |
| 105 | assert any(code.startswith('CRITICAL_') for code in result.eligibility_reasons) |
| 106 | |
| 107 | |
| 108 | @pytest.mark.parametrize('kind', ['stale', 'conflict']) |
| 109 | def test_current_price_gate(kind): |
| 110 | c, r = inputs() |
| 111 | if kind == 'stale': c.technical_feature_snapshot.stale_inputs = ['PRICE_HISTORY'] |
| 112 | else: c.technical_feature_snapshot.feature_states = {'latestPrice':'CONFLICTING'} |
| 113 | result = GlobalOpportunityRanker().score(c, r) |
| 114 | assert not result.rank_eligible and result.score_coverage == 90 |
| 115 | |
| 116 | |
| 117 | def test_optional_stale_sector_is_excluded_not_zero_or_gate(): |
| 118 | c, r = inputs() |
| 119 | c.sector_relative_strength_snapshot.sector_state = 'INSUFFICIENT_DATA' |
| 120 | c.sector_relative_strength_snapshot.stale_inputs = ['SECTOR_HISTORY'] |
| 121 | result = GlobalOpportunityRanker().score(c, r) |
| 122 | assert result.rank_eligible and result.score_coverage == 97 |
| 123 | assert result.opportunity_score == pytest.approx(7620/97) |
| 124 | |
| 125 | |
| 126 | def test_partial_and_missing_v1_fail_closed(): |
| 127 | c, r = inputs() |
| 128 | r.partial = True |
| 129 | assert not GlobalOpportunityRanker().score(c, r).rank_eligible |
| 130 | missing = GlobalOpportunityRanker().score(c, None) |
| 131 | assert not missing.rank_eligible and missing.rule_engine_score is None |
| 132 | assert missing.eligibility_reasons == ['V1_RESULT_MISSING'] |
| 133 | |
| 134 | |
| 135 | def test_deterministic_reasons_and_repeat_without_mutation_or_network(monkeypatch): |
| 136 | import socket |
| 137 | monkeypatch.setattr(socket, 'create_connection', lambda *a, **k: pytest.fail('network')) |
| 138 | c, r = inputs() |
| 139 | before = c.model_dump(), r.model_dump() |
| 140 | ranker = GlobalOpportunityRanker() |
| 141 | first = ranker.score(c, r) |
| 142 | assert first == ranker.score(c, r) |
| 143 | assert before == (c.model_dump(), r.model_dump()) |
| 144 | r.area_scores.reverse() |
| 145 | assert first == ranker.score(c, r) |
| 146 | assert first.top_positive_reasons == ['SUPPORT:BALANCE_SHEET', 'SUPPORT:FUNDAMENTAL_BUSINESS_QUALITY', |
| 147 | 'SUPPORT:GROWTH', 'SUPPORT:MANAGEMENT_GOVERNANCE', 'SUPPORT:NEWS_GEOPOLITICAL_EVENTS', |
| 148 | 'SUPPORT:ORDER_BOOK_CAPACITY_CATALYSTS'] |
| 149 | assert first.top_negative_reasons == ['WEAK:SECTOR'] |
| 150 | |
| 151 | |
| 152 | def test_sort_score_confidence_coverage_rule_score_and_uuid(): |
| 153 | pairs = [inputs(n, core=50, tech=50, sector=50, confidence=100) for n in range(1,7)] |
| 154 | # Equal opportunity 50 throughout except candidate 6 at 80. |
| 155 | pairs[0][1].overall_score = 80 |
| 156 | pairs[1][1].overall_score = 80 # UUID 1 beats UUID 2. |
| 157 | pairs[2][1].overall_score = 70 |
| 158 | # Same confidence as full rows (88.7), but coverage 97 vs 100. |
| 159 | c, r = pairs[3] |
| 160 | c.sector_score = c.sector_relative_strength_snapshot.relative_strength_score = None |
| 161 | c.sector_relative_strength_snapshot.sector_state = 'INSUFFICIENT_DATA' |
| 162 | c.technical_feature_snapshot.confidence = 100 |
| 163 | for i in range(3): pairs[i][1].confidence_score = 90 |
| 164 | r.confidence_score = (8870-700)/90 |
| 165 | pairs[4][1].confidence_score = 20 |
| 166 | c, r = inputs(6, core=80, tech=80, sector=80, confidence=0) |
| 167 | pairs[5] = c, r |
| 168 | ranker = GlobalOpportunityRanker() |
| 169 | rules = {r.global_instrument_id:r for _,r in pairs} |
| 170 | rows = ranker.rank([c for c,_ in reversed(pairs)], rules) |
| 171 | assert [r.global_instrument_id.int for r in rows] == [6,1,2,3,4,5] |
| 172 | assert rows == ranker.rank([c for c,_ in pairs], rules) |
| 173 | |
| 174 | |
| 175 | def test_identity_duplicates_and_versions_rejected(): |
| 176 | c, r = inputs() |
| 177 | ranker = GlobalOpportunityRanker() |
| 178 | with pytest.raises(ValueError, match='DUPLICATE_RANKER_INSTRUMENT'): |
| 179 | ranker.rank([c,c], {r.global_instrument_id:r}) |
| 180 | r.global_instrument_id = UUID(int=2) |
| 181 | with pytest.raises(ValueError, match='IDENTITY_MISMATCH'): ranker.score(c,r) |
| 182 | r.global_instrument_id = c.global_instrument_id |
| 183 | r.rule_engine_version = 'FUTURE' |
| 184 | with pytest.raises(ValueError, match='UNSUPPORTED_RULE_ENGINE_VERSION'): ranker.score(c,r) |
| 185 | assert ranker.rank([], {}) == [] |