| 1 | """Canonical identity, official captured contract, and persisted exact-date integration.""" |
| 2 | import json |
| 3 | from datetime import date, datetime, timedelta, timezone |
| 4 | from decimal import Decimal |
| 5 | from pathlib import Path |
| 6 | from types import SimpleNamespace |
| 7 | from unittest.mock import AsyncMock |
| 8 | from uuid import UUID |
| 9 | |
| 10 | import httpx |
| 11 | import pytest |
| 12 | |
| 13 | from app.nse_index_history import ENDPOINT, NseIndexHistoryProvider, parse_index_history |
| 14 | from app.nse_historical_daily import BOOTSTRAP, NseHistoricalResult, persist_daily_result |
| 15 | from app.sector_benchmarks import * |
| 16 | from app.sector_relative_strength import SectorRelativeStrengthEngine |
| 17 | from app.persistence import SqliteResearchPersistence |
| 18 | from app.settings import Settings |
| 19 | |
| 20 | START, END = date(2026, 8, 13), date(2026, 9, 11) |
| 21 | NOW = datetime(2026, 9, 13, 20, tzinfo=timezone.utc) |
| 22 | FIXTURES = Path(__file__).parent / 'fixtures' |
| 23 | FILES = dict(zip(BENCHMARKS, ('500', 'it', 'financials', 'healthcare'))) |
| 24 | STOCK = UUID(int=1) |
| 25 | |
| 26 | |
| 27 | def metadata(key=BROAD_KEY): |
| 28 | return dict(globalInstrumentId=str(benchmark_id(key)), assetType='INDEX', status='ACTIVE', |
| 29 | country='IN', primaryExchange='NSE', primarySymbol='NOT_IDENTITY', currency='INR', |
| 30 | providerMappings=[dict(provider='NSE', providerSymbol=BENCHMARKS[key], status='VERIFIED', |
| 31 | resolutionSource=CATALOG_VERSION, currency='INR', exchange='NSE')]) |
| 32 | |
| 33 | |
| 34 | def classification(sector='Technology', key=STOCK): |
| 35 | return dict(globalInstrumentId=str(key), canonicalSector=sector, source='NSE_INDICES_NIFTY500', |
| 36 | retrievedAt=NOW.isoformat(), status='ACTIVE', assetType='EQUITY', country='IN', exchange='NSE') |
| 37 | |
| 38 | |
| 39 | def parsed(key=BROAD_KEY): |
| 40 | result = NseHistoricalResult(benchmark_id(key), START, END, provider_symbol=BENCHMARKS[key], |
| 41 | source_url=ENDPOINT, retrieved_at=NOW) |
| 42 | parse_index_history((FIXTURES / f'nse_index_{FILES[key]}.json').read_bytes(), result, 'INR') |
| 43 | result.status = 'SUCCESS' |
| 44 | return result |
| 45 | |
| 46 | |
| 47 | @pytest.mark.parametrize('sector', list(SECTOR_KEYS)) |
| 48 | def test_exact_mapping(sector): |
| 49 | c = build_sector_contexts([classification(sector)], [metadata(k) for k in BENCHMARKS], {STOCK})[STOCK] |
| 50 | assert c.mapping_version == 'SECTOR_BENCHMARK_MAPPING_V1' |
| 51 | assert c.sector_benchmark.instrument_id == benchmark_id(SECTOR_KEYS[sector]) |
| 52 | assert c.market_benchmark.instrument_id == benchmark_id(BROAD_KEY) |
| 53 | assert c.sector_mapping_status == 'AVAILABLE' |
| 54 | with pytest.raises(ValueError): benchmark_id(BENCHMARKS[BROAD_KEY]) |
| 55 | |
| 56 | |
| 57 | @pytest.mark.parametrize('sector', ['Industrials', 'Materials', 'Consumer Staples', 'Consumer Discretionary', |
| 58 | 'Energy', 'Utilities', 'Real Estate', 'Communication Services', 'FINANCIAL SERVICES', 'Bank']) |
| 59 | def test_unmapped_not_substituted(sector): |
| 60 | c = build_sector_contexts([classification(sector)], [metadata(k) for k in BENCHMARKS], {STOCK})[STOCK] |
| 61 | assert c.sector_benchmark is None and c.sector_mapping_status == 'UNMAPPED_SECTOR_BENCHMARK' |
| 62 | assert c.market_benchmark is not None |
| 63 | |
| 64 | |
| 65 | @pytest.mark.parametrize('rows', [[], [classification(None)], [classification(), classification('Financials')], |
| 66 | [classification() | dict(source='GUESSED')]]) |
| 67 | def test_missing_ambiguous_classification(rows): |
| 68 | c = build_sector_contexts(rows, [], {STOCK})[STOCK] |
| 69 | assert c.sector_mapping_status == 'NO_SECTOR_CLASSIFICATION' |
| 70 | assert c.market_mapping_status == 'BENCHMARK_IDENTITY_UNAVAILABLE' |
| 71 | |
| 72 | |
| 73 | @pytest.mark.parametrize('change', [dict(status='INACTIVE'), dict(assetType='EQUITY'), dict(currency=None), |
| 74 | dict(globalInstrumentId=str(UUID(int=99))), dict(providerMappings=[]), |
| 75 | dict(providerMappings=metadata()['providerMappings'] * 2), |
| 76 | *[dict(providerMappings=[metadata()['providerMappings'][0] | x]) for x in |
| 77 | [dict(active=False), dict(status='UNVERIFIED'), dict(providerSymbol='NIFTY 50'), |
| 78 | dict(resolutionSource='GUESSED'), dict(currency='USD')]]]) |
| 79 | def test_identity_rejected(change): |
| 80 | with pytest.raises(ValueError, match='BENCHMARK_IDENTITY_UNAVAILABLE'): |
| 81 | benchmark_identity(metadata() | change, BROAD_KEY) |
| 82 | |
| 83 | |
| 84 | @pytest.mark.parametrize('key', list(BENCHMARKS)) |
| 85 | def test_observed_official_contract(key): |
| 86 | r = parsed(key) |
| 87 | assert r.rows_accepted == 22 |
| 88 | assert r.first_trading_date == START and r.last_trading_date == END |
| 89 | assert all(b.volume is b.turnover is b.previous_close is None for b in r.bars) |
| 90 | assert all(type(b.close) is Decimal and b.high >= b.low > 0 for b in r.bars) |
| 91 | assert r.bars == sorted(r.bars, key=lambda b: b.trading_date) |
| 92 | # EOD_TIMESTAMP is the trading date; HI_TIMESTAMP is the prior UTC day. |
| 93 | assert r.bars[-1].trading_date == date(2026, 9, 11) |
| 94 | |
| 95 | |
| 96 | @pytest.mark.parametrize('field,value', [('EOD_INDEX_NAME', 'NIFTY 50'), ('EOD_TIMESTAMP', 'bad'), |
| 97 | ('EOD_TIMESTAMP', '12-SEP-2026'), ('EOD_OPEN_INDEX_VAL', 'NaN'), ('EOD_LOW_INDEX_VAL', 0), |
| 98 | ('EOD_HIGH_INDEX_VAL', 1), ('EOD_CLOSE_INDEX_VAL', None)]) |
| 99 | def test_invalid_rows_fail_whole_window(field, value): |
| 100 | payload = json.loads((FIXTURES / 'nse_index_500.json').read_bytes()) |
| 101 | payload['data'][0][field] = value |
| 102 | r = NseHistoricalResult(benchmark_id(BROAD_KEY), START, END, provider_symbol='NIFTY 500') |
| 103 | with pytest.raises(ValueError): parse_index_history(json.dumps(payload).encode(), r, 'INR') |
| 104 | assert r.bars == [] |
| 105 | |
| 106 | |
| 107 | @pytest.mark.parametrize('payload', [b'', b'<html>blocked</html>', b'{}', b'{"data":[]}', b'{"data":[{}]}']) |
| 108 | def test_invalid_contract(payload): |
| 109 | with pytest.raises(ValueError): parse_index_history(payload, parsed(), 'INR') |
| 110 | |
| 111 | |
| 112 | def test_duplicate_and_precision(): |
| 113 | payload = json.loads((FIXTURES / 'nse_index_500.json').read_bytes()) |
| 114 | payload['data'][0]['EOD_OPEN_INDEX_VAL'] = '12345.123456789012' |
| 115 | r = parsed() |
| 116 | parse_index_history(json.dumps(payload).encode(), r, 'INR') |
| 117 | assert r.bars[-1].open == Decimal('12345.123456789012') |
| 118 | payload['data'].append(payload['data'][0]) |
| 119 | with pytest.raises(ValueError): parse_index_history(json.dumps(payload).encode(), parsed(), 'INR') |
| 120 | |
| 121 | |
| 122 | @pytest.mark.asyncio |
| 123 | async def test_session_reuse_query_spacing_no_cookie_logs(caplog): |
| 124 | requests = [] |
| 125 | def handler(request): |
| 126 | requests.append(request) |
| 127 | assert request.headers['user-agent'].startswith('Mozilla/') |
| 128 | if str(request.url) == BOOTSTRAP: |
| 129 | return httpx.Response(200, headers={'set-cookie': 'session=private-test-cookie; Path=/; Secure'}) |
| 130 | assert request.headers['cookie'] == 'session=private-test-cookie' |
| 131 | assert dict(request.url.params) == {'indexType':'NIFTY 500', 'from':'13-08-2026', 'to':'11-09-2026'} |
| 132 | return httpx.Response(200, content=(FIXTURES / 'nse_index_500.json').read_bytes()) |
| 133 | async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: |
| 134 | sleep = AsyncMock() |
| 135 | p = NseIndexHistoryProvider(SimpleNamespace(global_instrument_metadata=AsyncMock(return_value=metadata())), |
| 136 | Settings(), client=client, sleep=sleep) |
| 137 | for _ in range(2): |
| 138 | r = await p.fetch(benchmark_id(BROAD_KEY), start=START, end=END) |
| 139 | assert r.status == 'SUCCESS' and r.rows_accepted == 22 |
| 140 | assert len(requests) == 3 and sleep.await_count >= 2 |
| 141 | await p.aclose() |
| 142 | assert 'private-test-cookie' not in caplog.text |
| 143 | |
| 144 | |
| 145 | @pytest.mark.asyncio |
| 146 | @pytest.mark.parametrize('status,attempts', [(403,1), (404,1), (429,3), (500,3)]) |
| 147 | async def test_http_failures(status, attempts): |
| 148 | requests = [] |
| 149 | def handler(request): |
| 150 | if str(request.url) == BOOTSTRAP: return httpx.Response(200) |
| 151 | requests.append(request) |
| 152 | return httpx.Response(status) |
| 153 | async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: |
| 154 | p = NseIndexHistoryProvider(SimpleNamespace(global_instrument_metadata=AsyncMock(return_value=metadata())), |
| 155 | Settings(), client=client, sleep=AsyncMock()) |
| 156 | r = await p.fetch(benchmark_id(BROAD_KEY), start=START, end=END) |
| 157 | assert r.failure_reason == f'HISTORICAL_HTTP_{status}' and not r.bars |
| 158 | assert len(requests) == attempts |
| 159 | |
| 160 | |
| 161 | @pytest.mark.asyncio |
| 162 | async def test_identity_before_network_and_bounded_dates(): |
| 163 | def forbidden(request): pytest.fail('Network before identity/window gate') |
| 164 | async with httpx.AsyncClient(transport=httpx.MockTransport(forbidden)) as client: |
| 165 | p = NseIndexHistoryProvider(SimpleNamespace(global_instrument_metadata=AsyncMock(return_value=None)), Settings(), client=client) |
| 166 | r = await p.fetch(benchmark_id(BROAD_KEY), start=START, end=END) |
| 167 | assert r.failure_reason == 'BENCHMARK_IDENTITY_UNAVAILABLE' |
| 168 | r = await p.fetch(benchmark_id(BROAD_KEY), start=START, end=END+timedelta(days=1)) |
| 169 | assert r.failure_reason == 'INVALID_REQUEST_WINDOW' |
| 170 | |
| 171 | |
| 172 | @pytest.mark.asyncio |
| 173 | async def test_persist_repeat_correction_other_provider_and_failure(): |
| 174 | store = SqliteResearchPersistence() |
| 175 | repo = SimpleNamespace(upsert_daily_market_bars_async=AsyncMock(side_effect=store.upsert_daily_market_bars)) |
| 176 | r = parsed() |
| 177 | await persist_daily_result(repo, r) |
| 178 | await persist_daily_result(repo, r) |
| 179 | corrected = r.bars[-1].model_copy(update={'close': r.bars[-1].close + Decimal('.01')}) |
| 180 | store.upsert_daily_market_bar(corrected.model_copy(update={'provider':'OTHER'})) |
| 181 | r.bars[-1] = corrected |
| 182 | await persist_daily_result(repo, r) |
| 183 | rows = store.load_daily_market_bars({r.global_instrument_id}, provider='NSE') |
| 184 | assert len(rows) == 22 and rows[-1].close == corrected.close |
| 185 | assert len(store.load_daily_market_bars({r.global_instrument_id})) == 23 |
| 186 | repo.upsert_daily_market_bars_async.side_effect = RuntimeError('do not expose') |
| 187 | await persist_daily_result(repo, r) |
| 188 | assert r.failure_reason == 'DAILY_BAR_PERSISTENCE_UNAVAILABLE' |
| 189 | assert len(store.load_daily_market_bars({r.global_instrument_id}, provider='NSE')) == 22 |
| 190 | |
| 191 | |
| 192 | def test_daily_alignment_reversed_no_filling_and_provenance(monkeypatch): |
| 193 | import socket |
| 194 | monkeypatch.setattr(socket, 'create_connection', lambda *a, **k: pytest.fail('network during compute')) |
| 195 | sector_key = SECTOR_KEYS['Technology'] |
| 196 | sector, market = parsed(sector_key), parsed() |
| 197 | stock = [b.model_copy(update={'global_instrument_id':STOCK}) for b in sector.bars] |
| 198 | contexts = build_sector_contexts([classification()], [metadata(k) for k in BENCHMARKS], {STOCK}) |
| 199 | daily = {STOCK:stock, sector.global_instrument_id:sector.bars, market.global_instrument_id:market.bars} |
| 200 | def compute(rows): |
| 201 | return SectorRelativeStrengthEngine().compute(STOCK, [], as_of=NOW, currency='INR', context=contexts[STOCK], daily_bar_histories=rows) |
| 202 | first = compute(daily) |
| 203 | assert first == compute({k:list(reversed(v)) for k,v in daily.items()}) |
| 204 | assert first.benchmark_mapping_version == MAPPING_VERSION |
| 205 | assert first.relative_vs_sector1_m == 0 and first.sector_state != 'INSUFFICIENT_DATA' |
| 206 | assert first.sector_return1_m == pytest.approx(float((sector.bars[-1].close / sector.bars[0].close - 1)*100)) |
| 207 | assert first.history_sources['sector'] == 'DAILY_MARKET_BAR_NSE' |
| 208 | daily[sector.global_instrument_id] = sector.bars[1:] |
| 209 | second = compute(daily) |
| 210 | assert second.relative_vs_sector1_m is None and second.relative_vs_sector1_w == 0 |
| 211 | daily[sector.global_instrument_id] = [b.model_copy(update={'provider':'OTHER'}) for b in sector.bars] |
| 212 | assert compute(daily).benchmark_states['sector'] == 'BENCHMARK_HISTORY_UNAVAILABLE' |
| 213 | |
| 214 | |
| 215 | @pytest.mark.asyncio |
| 216 | @pytest.mark.parametrize('count,multiple', [(0,False),(1,False),(18,False),(18,True)]) |
| 217 | async def test_stage_b_bounded_queries(count, multiple): |
| 218 | from app.global_scanner import GlobalScanner |
| 219 | from test_global_scanner import instrument, persisted, scan |
| 220 | store = SqliteResearchPersistence() |
| 221 | items = [instrument(n) for n in range(10, 10+count)] |
| 222 | for item in items: persisted(store, item) |
| 223 | initial = await scan(items, store, top_n=max(count, 1)) |
| 224 | ids = {UUID(item['globalInstrumentId']) for item in items} |
| 225 | sectors = list(SECTOR_KEYS) |
| 226 | rows = [classification(sectors[i%3] if multiple else 'Technology', k) for i,k in enumerate(sorted(ids))] |
| 227 | contexts = build_sector_contexts(rows, [metadata(k) for k in BENCHMARKS], ids) |
| 228 | queries = [] |
| 229 | store._connection.set_trace_callback(queries.append) |
| 230 | enriched = GlobalScanner(None, store).enrich_candidates(initial, sector_contexts=contexts) |
| 231 | assert len(enriched) == count |
| 232 | assert len(queries) == (2 if count else 0) |
| 233 | assert all('SELECT' in q for q in queries) |
| 234 | |
| 235 | |
| 236 | @pytest.mark.asyncio |
| 237 | @pytest.mark.parametrize('reason', ['HISTORICAL_HTTP_403', 'INVALID_INDEX_ROW', 'DAILY_BAR_PERSISTENCE_UNAVAILABLE']) |
| 238 | async def test_worker_failure_isolation_session_closed(monkeypatch, reason): |
| 239 | from app.market_data_population import IndiaMarketDataPopulationJobs |
| 240 | import app.nse_index_history as module |
| 241 | keys = sorted([benchmark_id(k) for k in BENCHMARKS], key=str)[:2] |
| 242 | responses = [NseHistoricalResult(keys[0], START, END, failure_reason=reason), parsed()] |
| 243 | provider = SimpleNamespace(fetch=AsyncMock(side_effect=responses), aclose=AsyncMock()) |
| 244 | factory = lambda *a, **k: provider |
| 245 | monkeypatch.setattr(module, 'NseIndexHistoryProvider', factory) |
| 246 | repo = SimpleNamespace(upsert_daily_market_bars_async=AsyncMock(return_value=22)) |
| 247 | jobs = IndiaMarketDataPopulationJobs(repo, None, None, Settings(), sleep=AsyncMock(), clock=lambda:NOW) |
| 248 | results = await jobs.populate_benchmark_history(set(keys), start=START, end=END, identity_headers={}) |
| 249 | assert len(results) == 2 and results[0].failure_reason == reason and results[1].persisted_rows == 22 |
| 250 | assert provider.fetch.await_count == 2 and provider.aclose.await_count == 1 |
| 251 | assert repo.upsert_daily_market_bars_async.await_count == 1 |
| 252 | |
| 253 | |
| 254 | @pytest.mark.asyncio |
| 255 | async def test_worker_throttling_cooldown_and_batch_bound(monkeypatch): |
| 256 | from app.market_data_population import IndiaMarketDataPopulationJobs |
| 257 | import app.nse_index_history as module |
| 258 | keys = {benchmark_id(k) for k in BENCHMARKS} |
| 259 | provider = SimpleNamespace(fetch=AsyncMock(return_value=NseHistoricalResult(next(iter(keys)), START, END, |
| 260 | failure_reason='HISTORICAL_HTTP_429')), aclose=AsyncMock()) |
| 261 | monkeypatch.setattr(module, 'NseIndexHistoryProvider', lambda *a, **k:provider) |
| 262 | jobs = IndiaMarketDataPopulationJobs(None, None, None, Settings(), sleep=AsyncMock(), clock=lambda:NOW) |
| 263 | results = await jobs.populate_benchmark_history(keys, start=START, end=END, identity_headers={}) |
| 264 | assert provider.fetch.await_count == 1 |
| 265 | assert [r.failure_reason for r in results][1:] == ['RETRY_COOLDOWN']*3 |
| 266 | jobs.settings.market_data_population_batch_size = 1 |
| 267 | with pytest.raises(ValueError, match='BENCHMARK_BATCH_LIMIT'): |
| 268 | await jobs.populate_benchmark_history(keys, start=START, end=END, identity_headers={}) |
| 269 | |
| 270 | |
| 271 | def test_hand_calculated_daily_horizons_and_staleness(): |
| 272 | from app.models import DailyMarketBar |
| 273 | keys = [STOCK, benchmark_id(SECTOR_KEYS['Technology']), benchmark_id(BROAD_KEY)] |
| 274 | daily = {} |
| 275 | # Independent linear prices: stock +2, sector +1, market flat per observation. |
| 276 | for key, slope in zip(keys, (2,1,0)): |
| 277 | daily[key] = [DailyMarketBar(global_instrument_id=key, trading_date=(NOW-timedelta(days=159-i)).date(), |
| 278 | open=Decimal(1000+slope*i), high=Decimal(1000+slope*i), low=Decimal(1000+slope*i), |
| 279 | close=Decimal(1000+slope*i), currency='INR', provider='NSE', provider_symbol='provenance', |
| 280 | source_mode='REAL', source_url=ENDPOINT, retrieved_at=NOW) for i in range(160)] |
| 281 | ctx = build_sector_contexts([classification()], [metadata(k) for k in BENCHMARKS], {STOCK})[STOCK] |
| 282 | def compute(rows): return SectorRelativeStrengthEngine().compute(STOCK, [], as_of=NOW, context=ctx, daily_bar_histories=rows) |
| 283 | result = compute(daily) |
| 284 | for suffix, n in zip(('1_w','1_m','3_m','6_m'), (5,21,63,126)): |
| 285 | assert getattr(result,'stock_return'+suffix) == pytest.approx(100*2*n/(1318-2*n)) |
| 286 | assert getattr(result,'sector_return'+suffix) == pytest.approx(100*n/(1159-n)) |
| 287 | assert getattr(result,'market_return'+suffix) == 0 |
| 288 | daily[keys[1]] = daily[keys[1]][:-10] |
| 289 | assert compute(daily).benchmark_states['sector'] == 'STALE_BENCHMARK_HISTORY' |
| 290 | |
| 291 | |
| 292 | @pytest.mark.asyncio |
| 293 | async def test_metadata_adapter_is_batched_read_only(): |
| 294 | from app.portfolio_orchestration import PortfolioResearchOrchestrator |
| 295 | requests = [] |
| 296 | def handler(request): |
| 297 | requests.append(request) |
| 298 | assert request.method == 'GET' and request.url.path == '/api/v1/instruments/benchmarks' |
| 299 | return httpx.Response(200, json=[metadata(k) for k in BENCHMARKS]) |
| 300 | async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: |
| 301 | obj = PortfolioResearchOrchestrator(None, Settings(), client=client) |
| 302 | obj.india_nifty500_universe = AsyncMock(return_value=[classification()]) |
| 303 | assert await obj.sector_benchmark_contexts(set()) == {} |
| 304 | assert not requests and obj.india_nifty500_universe.await_count == 0 |
| 305 | result = await obj.sector_benchmark_contexts({STOCK}) |
| 306 | assert result[STOCK].sector_mapping_status == 'AVAILABLE' |
| 307 | assert len(requests) == obj.india_nifty500_universe.await_count == 1 |
| 308 | |
| 309 | |
| 310 | def test_india_market_benchmark_never_assigned_to_foreign_or_unknown_stock(): |
| 311 | registered = [metadata(k) for k in BENCHMARKS] |
| 312 | for rows in ([], [classification() | dict(country='US', exchange='NASDAQ')]): |
| 313 | result = build_sector_contexts(rows, registered, {STOCK})[STOCK] |
| 314 | assert result.market_benchmark is None and result.sector_benchmark is None |