| 1 | from datetime import timedelta |
| 2 | from uuid import UUID |
| 3 | |
| 4 | import pytest |
| 5 | |
| 6 | from app.sector_relative_strength import ( |
| 7 | BenchmarkReference, SectorContext, SectorRelativeStrengthConfig, SectorRelativeStrengthEngine, |
| 8 | ) |
| 9 | from test_technical_features import NOW, STOCK, history |
| 10 | |
| 11 | |
| 12 | SECTOR, MARKET = UUID(int=2), UUID(int=3) |
| 13 | |
| 14 | |
| 15 | def context(currency="INR", region="INDIA"): |
| 16 | return SectorContext("Financial Services", "CANONICAL_UNIVERSE", NOW-timedelta(days=1), region, |
| 17 | BenchmarkReference(SECTOR, currency), BenchmarkReference(MARKET, currency)) |
| 18 | |
| 19 | |
| 20 | def rate_series(rate, count=160): |
| 21 | return [100*(1+rate)**i for i in range(count)] |
| 22 | |
| 23 | |
| 24 | def evaluate(stock, sector=None, market=None, mapping=None, currency="INR"): |
| 25 | histories = {} |
| 26 | if sector is not None: histories[SECTOR] = history(sector, SECTOR, currency) |
| 27 | if market is not None: histories[MARKET] = history(market, MARKET, currency) |
| 28 | return SectorRelativeStrengthEngine().compute(STOCK, history(stock, currency=currency), as_of=NOW, |
| 29 | currency=currency, context=mapping or context(currency), benchmark_histories=histories) |
| 30 | |
| 31 | |
| 32 | @pytest.mark.parametrize("rates,state", [((.002, .001, .0005), "LEADING"), ((-.002, 0, .001), "LAGGING"), ((0, 0, 0), "NEUTRAL")]) |
| 33 | def test_outperforming_lagging_and_flat_relative_series(rates, state): |
| 34 | result = evaluate(*(rate_series(rate) for rate in rates)) |
| 35 | assert result.sector_state == state |
| 36 | assert result.confidence == 100 |
| 37 | assert result.sector == "Financials" |
| 38 | assert result.relative_vs_sector1_m == pytest.approx(result.stock_return1_m-result.sector_return1_m) |
| 39 | assert result.relative_vs_market3_m == pytest.approx(result.stock_return3_m-result.market_return3_m) |
| 40 | assert (result.relative_strength_score > 50 if state == "LEADING" else |
| 41 | result.relative_strength_score < 50 if state == "LAGGING" else result.relative_strength_score == 50) |
| 42 | |
| 43 | |
| 44 | def test_beating_sector_but_lagging_market_keeps_separate_legs(): |
| 45 | result = evaluate(rate_series(.001), rate_series(0), rate_series(.002)) |
| 46 | assert result.relative_vs_sector1_m > 0 |
| 47 | assert result.relative_vs_market1_m < 0 |
| 48 | assert result.sector_return1_m != result.market_return1_m |
| 49 | |
| 50 | |
| 51 | @pytest.mark.parametrize("anchors,state", [((99, 98, 105, 110), "IMPROVING"), ((101, 102, 95, 90), "WEAKENING")]) |
| 52 | def test_relative_trend_changes_use_horizon_normalization(anchors, state): |
| 53 | stock = [100]*160 |
| 54 | for offset, value in zip((5, 21, 63, 126), anchors): stock[-offset-1] = value |
| 55 | result = evaluate(stock, [100]*160, [100]*160) |
| 56 | assert result.sector_state == state |
| 57 | |
| 58 | |
| 59 | def test_missing_sector_history_does_not_substitute_market(): |
| 60 | result = evaluate(rate_series(.001), market=rate_series(.0005)) |
| 61 | assert result.sector_return1_m is result.relative_vs_sector1_m is None |
| 62 | assert "SECTOR_HISTORY" in result.missing_inputs |
| 63 | assert result.market_return1_m is not None |
| 64 | assert result.sector_state == "INSUFFICIENT_DATA" |
| 65 | assert result.confidence == 50 |
| 66 | |
| 67 | |
| 68 | def test_missing_market_mapping_and_no_history_do_not_become_zero(): |
| 69 | mapping = SectorContext("Industrials", "CANONICAL", NOW, "INDIA", BenchmarkReference(SECTOR, "INR")) |
| 70 | result = evaluate(rate_series(.001), sector=rate_series(.0005), mapping=mapping) |
| 71 | assert result.market_return1_m is result.relative_vs_market1_m is None |
| 72 | assert "MARKET_BENCHMARK_MAPPING" in result.missing_inputs |
| 73 | missing = evaluate(rate_series(.001), mapping=SectorContext()) |
| 74 | assert missing.relative_strength_score is None and missing.confidence == 0 |
| 75 | assert missing.stock_return1_m is not None |
| 76 | |
| 77 | |
| 78 | @pytest.mark.parametrize("count", [0, 5, 6, 21]) |
| 79 | def test_insufficient_lookback_cannot_claim_multi_period_consistency(count): |
| 80 | result = evaluate([100]*count, [100]*count, [100]*count) |
| 81 | assert result.sector_state == "INSUFFICIENT_DATA" |
| 82 | assert result.relative_strength_score is None |
| 83 | assert result.stock_return1_m is None |
| 84 | |
| 85 | |
| 86 | @pytest.mark.parametrize("currency,region", [("INR", "INDIA"), ("USD", "USA"), ("EUR", "EUROPE")]) |
| 87 | def test_region_neutral_fixture(currency, region): |
| 88 | result = evaluate(rate_series(.002), rate_series(.001), rate_series(.0005), mapping=context(currency, region), currency=currency) |
| 89 | assert result.sector_state == "LEADING" and result.region == region |
| 90 | assert result.market_benchmark_id == MARKET and result.sector_benchmark_id == SECTOR |
| 91 | |
| 92 | |
| 93 | def test_exact_date_alignment_prevents_mismatched_period_comparison(): |
| 94 | stock = history(rate_series(.002)) |
| 95 | sector = history(rate_series(.001), SECTOR) |
| 96 | sector.pop(-22) # Remove precisely the 1M stock reference date, not the latest. |
| 97 | result = SectorRelativeStrengthEngine().compute(STOCK, stock, as_of=NOW, context=context(), |
| 98 | benchmark_histories={SECTOR: sector, MARKET: history(rate_series(.001), MARKET)}) |
| 99 | assert result.relative_vs_sector1_m is None |
| 100 | assert "SECTOR_ALIGNED_DATES_1M" in result.missing_inputs |
| 101 | assert result.relative_vs_sector3_m is not None |
| 102 | assert result.comparison_windows["1M"] == (stock[-22].observed_at.date(), stock[-1].observed_at.date()) |
| 103 | |
| 104 | |
| 105 | def test_stale_and_conflicting_benchmarks_are_explicit(): |
| 106 | stock = history(rate_series(.002)) |
| 107 | sector = history(rate_series(.001), SECTOR, end=NOW-timedelta(days=20)) |
| 108 | result = SectorRelativeStrengthEngine().compute(STOCK, stock, as_of=NOW, context=context(), benchmark_histories={SECTOR: sector}) |
| 109 | assert result.relative_vs_sector1_m is None |
| 110 | assert result.stale_inputs == ["SECTOR_HISTORY"] |
| 111 | sector = history(rate_series(.001), SECTOR) |
| 112 | sector.append(sector[-1].model_copy(update={"price": 999, "provider": "OTHER"})) |
| 113 | result = SectorRelativeStrengthEngine().compute(STOCK, stock, as_of=NOW, context=context(), benchmark_histories={SECTOR: sector}) |
| 114 | assert "CONFLICTING_SECTOR_PRICE" in result.missing_inputs |
| 115 | assert result.relative_strength_score is None |
| 116 | |
| 117 | |
| 118 | def test_determinism_no_network_and_future_classification_rejected(monkeypatch): |
| 119 | import socket |
| 120 | monkeypatch.setattr(socket, "create_connection", lambda *a, **k: pytest.fail("Network invoked")) |
| 121 | engine = SectorRelativeStrengthEngine() |
| 122 | stock, sector, market = history(rate_series(.002)), history(rate_series(.001), SECTOR), history(rate_series(.0005), MARKET) |
| 123 | first = engine.compute(STOCK, stock, as_of=NOW, context=context(), benchmark_histories={SECTOR: sector, MARKET: market}) |
| 124 | second = engine.compute(STOCK, reversed(stock), as_of=NOW, context=context(), benchmark_histories={MARKET: list(reversed(market)), SECTOR: list(reversed(sector))}) |
| 125 | assert first == second |
| 126 | future = SectorContext("Financials", "CANONICAL", NOW+timedelta(days=1), "INDIA", BenchmarkReference(SECTOR, "INR")) |
| 127 | result = engine.compute(STOCK, stock, as_of=NOW, context=future, benchmark_histories={SECTOR: sector}) |
| 128 | assert result.sector is None and result.relative_vs_sector1_m is None |
| 129 | |
| 130 | |
| 131 | def test_configuration_and_self_benchmark_rejected(): |
| 132 | with pytest.raises(ValueError): SectorRelativeStrengthConfig(period_weights=(0, 0, 0, 0)) |
| 133 | with pytest.raises(ValueError): SectorRelativeStrengthConfig(full_scale_edge_per_observation_pct=0) |
| 134 | result = evaluate(rate_series(.001), mapping=SectorContext("Financials", "CANONICAL", NOW, "INDIA", BenchmarkReference(STOCK, "INR"))) |
| 135 | assert "INVALID_SECTOR_BENCHMARK_MAPPING" in result.missing_inputs |
| 136 | |
| 137 | |
| 138 | @pytest.mark.asyncio |
| 139 | async def test_stage_b_batches_only_deep_eligible_and_preserves_phase1(monkeypatch): |
| 140 | from app.global_scanner import GlobalScanner |
| 141 | from app.persistence import SqliteResearchPersistence |
| 142 | from app.stock_rule_engine import StockRuleEngineService |
| 143 | from test_global_scanner import instrument, persisted, scan |
| 144 | def forbidden(*a, **k): pytest.fail("Provider/V1/universe invoked by Stage B") |
| 145 | monkeypatch.setattr(StockRuleEngineService, "analyze", forbidden) |
| 146 | store = SqliteResearchPersistence() |
| 147 | items = [instrument(n) for n in (1, 4, 5)] |
| 148 | for item in items: persisted(store, item) |
| 149 | # Candidate5 lacks critical financial evidence, despite a complete price history. |
| 150 | persisted(store, items[2], {}) |
| 151 | for key in (STOCK, UUID(int=4), UUID(int=5), SECTOR, MARKET): |
| 152 | for row in history(rate_series(.001), key): |
| 153 | store.upsert_market_price_observation(row.model_copy(update={"provider": "YAHOO_FINANCE"})) |
| 154 | initial = await scan(items, store) |
| 155 | before = initial.model_dump() |
| 156 | queries = [] |
| 157 | store._connection.set_trace_callback(queries.append) |
| 158 | class NoUniverse: |
| 159 | active_global_equities = forbidden |
| 160 | scanner = GlobalScanner(NoUniverse(), store) |
| 161 | enriched = scanner.enrich_candidates(initial, sector_contexts={STOCK: context()}) |
| 162 | assert {c.global_instrument_id for c in enriched} == {STOCK, UUID(int=4)} |
| 163 | assert initial.model_dump() == before |
| 164 | assert len(queries) == 2 |
| 165 | daily_query = next(q for q in queries if "global_daily_market_bars" in q) |
| 166 | close_query = next(q for q in queries if "global_market_price_observations" in q) |
| 167 | assert "instrument_id IN" in close_query |
| 168 | assert all(str(UUID(int=5)) not in q for q in queries) |
| 169 | assert str(SECTOR) in close_query and str(MARKET) in close_query |
| 170 | assert str(SECTOR) in daily_query and str(MARKET) in daily_query |
| 171 | assert all(c.technical_feature_snapshot.global_instrument_id == c.global_instrument_id for c in enriched) |
| 172 | # Private inputs are not part of the enrichment contract. |
| 173 | assert enriched == scanner.enrich_candidates(initial, sector_contexts={STOCK: context()}) |
| 174 | |
| 175 | |
| 176 | @pytest.mark.asyncio |
| 177 | async def test_stage_b_missing_sector_renormalizes_and_order_is_deterministic(): |
| 178 | from app.global_scanner import GlobalScanner |
| 179 | from app.persistence import SqliteResearchPersistence |
| 180 | from test_global_scanner import instrument, persisted, scan |
| 181 | store = SqliteResearchPersistence() |
| 182 | items = [instrument(n) for n in (1, 4)] |
| 183 | for item in items: |
| 184 | persisted(store, item) |
| 185 | for row in history(rate_series(.001), UUID(item["globalInstrumentId"])): |
| 186 | store.upsert_market_price_observation(row.model_copy(update={"provider": "YAHOO_FINANCE"})) |
| 187 | result = await scan(items, store) |
| 188 | enriched = GlobalScanner(None, store).enrich_candidates(result) |
| 189 | assert [c.global_instrument_id.int for c in enriched] == [1, 4] |
| 190 | for candidate in enriched: |
| 191 | assert candidate.sector_score is None |
| 192 | assert candidate.stage_b_score == candidate.technical_score |
| 193 | assert candidate.score_coverage == 70 |
| 194 | assert candidate.confidence == pytest.approx(candidate.technical_feature_snapshot.confidence*.7) |