| 1 | from datetime import datetime, timedelta, timezone |
| 2 | from decimal import Decimal |
| 3 | from uuid import UUID |
| 4 | from unittest.mock import Mock |
| 5 | |
| 6 | import pytest |
| 7 | |
| 8 | from app.models import DailyMarketBar |
| 9 | from app.technical_features import TechnicalFeatureEngine |
| 10 | from test_technical_features import history, NOW, STOCK |
| 11 | |
| 12 | |
| 13 | def candles(values, *, volumes=None, **changes): |
| 14 | return [DailyMarketBar(global_instrument_id=STOCK, trading_date=p.observed_at.date(), |
| 15 | open=p.price.quantize(Decimal('.00000001')), high=(p.price+1).quantize(Decimal('.00000001')), |
| 16 | low=(p.price-1).quantize(Decimal('.00000001')), close=p.price.quantize(Decimal('.00000001')), previous_close=Decimal('999'), |
| 17 | volume=100 if volumes is None else volumes[i], currency='INR', provider='NSE', provider_symbol='TEST', |
| 18 | source_mode='REAL', source_url='https://www.nseindia.com/history', retrieved_at=p.retrieved_at |
| 19 | ).model_copy(update=changes) for i,p in enumerate(history(values))] |
| 20 | |
| 21 | |
| 22 | def evaluate(rows, fallback=(), **kwargs): |
| 23 | return TechnicalFeatureEngine().compute(STOCK, fallback, as_of=NOW, currency='INR', daily_bar_history=rows, **kwargs) |
| 24 | |
| 25 | |
| 26 | def test_atr_hand_calculated_gaps_and_wilder_update(): |
| 27 | values=[10]*13+[14,6] |
| 28 | r=evaluate(candles(values)) |
| 29 | # Twelve TR=2, gap up TR=5, gap down TR=9. Provider prevClose=999 is ignored. |
| 30 | assert r.atr14==pytest.approx(float(Decimal(38)/14),abs=1e-8) |
| 31 | assert r.atr_pct==pytest.approx(float(Decimal(38)/14/6*100),abs=1e-8) |
| 32 | r=evaluate(candles(values+[10])) |
| 33 | assert r.atr14==pytest.approx(float((Decimal(38)/14*13+5)/14),abs=1e-8) |
| 34 | |
| 35 | |
| 36 | @pytest.mark.parametrize('count,atr,adx',[(14,None,None),(15,2,None),(27,2,None),(28,2,100),(60,2,100)]) |
| 37 | def test_atr_adx_minimum_and_constant_range(count,atr,adx): |
| 38 | r=evaluate(candles(range(100,100+count))) |
| 39 | assert r.atr14==atr and r.adx14==adx |
| 40 | assert r.feature_readiness['ATR14']==('AVAILABLE' if atr is not None else 'INSUFFICIENT_HISTORY') |
| 41 | assert r.feature_readiness['ADX14']==('AVAILABLE' if adx is not None else 'INSUFFICIENT_HISTORY') |
| 42 | |
| 43 | |
| 44 | @pytest.mark.parametrize('values,expected',[(list(range(100,150)),100),(list(range(150,100,-1)),100),([100]*50,0)]) |
| 45 | def test_adx_direction_is_not_strength(values,expected): |
| 46 | assert evaluate(candles(values)).adx14==expected |
| 47 | |
| 48 | |
| 49 | def test_zero_range_and_directional_movement_ties(): |
| 50 | flat=candles([100]*30,open=Decimal(100),high=Decimal(100),low=Decimal(100)) |
| 51 | r=evaluate(flat) |
| 52 | assert r.atr14==r.atr_pct==r.adx14==0 |
| 53 | # Outside bars with equally expanding high/low give +DM=-DM=0. |
| 54 | tied=[b.model_copy(update={'high':Decimal(101+i),'low':Decimal(99-i)}) for i,b in enumerate(candles([100]*30))] |
| 55 | assert evaluate(tied).adx14==0 |
| 56 | |
| 57 | |
| 58 | def test_adx_independent_closed_form_decimal_reference(): |
| 59 | rows=candles([100+(i*7 % 17) for i in range(65)]) |
| 60 | trs=[]; plus=[]; minus=[] |
| 61 | for previous,current in zip(rows,rows[1:]): |
| 62 | trs.append(max(current.high-current.low,abs(current.high-previous.close),abs(current.low-previous.close))) |
| 63 | up=current.high-previous.high; down=previous.low-current.low |
| 64 | plus.append(max(up,Decimal(0)) if up>down else Decimal(0)) |
| 65 | minus.append(max(down,Decimal(0)) if down>up else Decimal(0)) |
| 66 | def weighted(values,index): |
| 67 | q=Decimal(13)/14 |
| 68 | return sum(values[:14])/14*q**(index-13)+sum(values[k]/14*q**(index-k) for k in range(14,index+1)) |
| 69 | dx=[] |
| 70 | for i in range(13,len(trs)): |
| 71 | tr=weighted(trs,i); positive=100*weighted(plus,i)/tr; negative=100*weighted(minus,i)/tr |
| 72 | dx.append(100*abs(positive-negative)/(positive+negative) if positive+negative else Decimal(0)) |
| 73 | r=evaluate(rows) |
| 74 | assert r.adx14==pytest.approx(float(weighted(dx,len(dx)-1)),abs=1e-8) |
| 75 | assert r.atr14==pytest.approx(float(weighted(trs,len(trs)-1)),abs=1e-8) |
| 76 | assert 0 <= r.adx14 <= 100 |
| 77 | |
| 78 | |
| 79 | def test_missing_ohlc_restarts_warmup_chronology_and_no_blending(): |
| 80 | rows=candles(range(100,140)) |
| 81 | rows[-2]=rows[-2].model_copy(update={'high':None}) |
| 82 | r=evaluate(rows,history(range(1000,1040))) |
| 83 | assert r.atr14 is r.adx14 is None and r.feature_readiness['ATR14']=='MISSING_OHLC' |
| 84 | assert r.latest_price==139 and r.dma20 is not None |
| 85 | assert r==evaluate(list(reversed(rows)),history(range(1000,1040))) |
| 86 | other=[b.model_copy(update={'provider':'YAHOO_FINANCE','high':Decimal(9999)}) for b in rows] |
| 87 | assert evaluate(rows+other).atr14 is None |
| 88 | |
| 89 | |
| 90 | @pytest.mark.parametrize('count,volumes,average,ratio,state',[ |
| 91 | (20,[100]*20,None,None,'UNAVAILABLE'),(21,[100]*21,100,1,'NORMAL'), |
| 92 | (21,[100]*20+[150],100,1.5,'EXPANSION'),(21,[100]*20+[75],100,.75,'CONTRACTION'), |
| 93 | (21,[100]*20+[0],100,0,'CONTRACTION'),(21,[0]*21,0,None,'UNAVAILABLE'), |
| 94 | (21,[100]*20+[None],100,None,'UNAVAILABLE'),(21,[None]+[100]*20,None,None,'UNAVAILABLE')]) |
| 95 | def test_volume_boundaries(count,volumes,average,ratio,state): |
| 96 | r=evaluate(candles([100]*count,volumes=volumes)) |
| 97 | assert r.current_volume==volumes[-1] and r.volume_average20==average and r.volume_ratio20==ratio and r.volume_state==state |
| 98 | |
| 99 | |
| 100 | def test_bigint_current_exact_ratio_decimal(): |
| 101 | maximum=9223372036854775807 |
| 102 | r=evaluate(candles([100]*21,volumes=[maximum]*21)) |
| 103 | assert r.current_volume==maximum and type(r.current_volume) is int and r.volume_ratio20==1 |
| 104 | assert r.model_dump()['current_volume']==maximum |
| 105 | |
| 106 | |
| 107 | @pytest.mark.parametrize('volume,confirmed,breakout',[(200,True,'VOLUME_CONFIRMED'),(100,False,'PRICE_BREAKOUT'),(None,None,'PRICE_BREAKOUT')]) |
| 108 | def test_breakout_confirmation_and_existing_bonus(volume,confirmed,breakout): |
| 109 | values=[100]*60+[104] |
| 110 | r=evaluate(candles(values,volumes=[100]*60+[volume])) |
| 111 | base=TechnicalFeatureEngine().compute(STOCK,history(values),as_of=NOW,currency='INR') |
| 112 | assert r.technical_state=='BREAKOUT' and r.breakout_state==breakout |
| 113 | assert r.breakout_volume_confirmed is confirmed |
| 114 | assert r.technical_score==min(100,base.technical_score+(5 if confirmed else 0)) |
| 115 | |
| 116 | |
| 117 | def test_reversal_confirmation_is_diagnostic_only(): |
| 118 | values=[200-.6*i for i in range(240)]+[56.6+.3*i for i in range(20)] |
| 119 | r=evaluate(candles(values,volumes=[100]*259+[200])) |
| 120 | base=TechnicalFeatureEngine().compute(STOCK,history(values),as_of=NOW,currency='INR') |
| 121 | assert r.technical_state=='REVERSAL_CANDIDATE' and r.reversal_volume_confirmed is True |
| 122 | assert r.technical_score==base.technical_score |
| 123 | |
| 124 | |
| 125 | def test_fallback_and_daily_source_priority(): |
| 126 | fallback=history(range(100,160)) |
| 127 | r=evaluate([],fallback) |
| 128 | assert r==TechnicalFeatureEngine().compute(STOCK,fallback,as_of=NOW,currency='INR') |
| 129 | assert r.technical_input_source=='CLOSE_ONLY_FALLBACK' and r.atr14 is r.adx14 is None |
| 130 | richer=evaluate(candles(range(200,260)),fallback) |
| 131 | assert richer.latest_price==259 and richer.technical_input_source=='DAILY_MARKET_BAR_NSE' |
| 132 | assert richer.history_trading_end==candles(range(200,260))[-1].trading_date |
| 133 | other=evaluate(candles(range(200,260),provider='OTHER'),fallback) |
| 134 | assert other.latest_price==159 and other.technical_input_source=='CLOSE_ONLY_FALLBACK' |
| 135 | |
| 136 | |
| 137 | def test_deterministic_duplicates_corrections_conflicts_and_asof(): |
| 138 | rows=candles(range(100,140)); last=rows[-1] |
| 139 | r=evaluate(rows+[last]); assert r.observation_count==40 and r.duplicate_observation_count==1 |
| 140 | newer=last.model_copy(update={'close':Decimal(140),'retrieved_at':NOW}) |
| 141 | assert evaluate(rows+[newer]).latest_price==140 |
| 142 | conflicting=newer.model_copy(update={'close':Decimal(141)}) |
| 143 | r=evaluate(rows+[newer,conflicting],history(range(200,240))) |
| 144 | assert r.latest_price is r.atr14 is r.adx14 is None and r.technical_score is None |
| 145 | assert 'MIXED_NOT_ALLOWED' in r.source_diagnostics |
| 146 | assert r==evaluate(list(reversed(rows+[newer,conflicting])),history(range(200,240))) |
| 147 | future=last.model_copy(update={'close':Decimal(999),'retrieved_at':NOW+timedelta(seconds=1)}) |
| 148 | assert evaluate(rows+[future]).latest_price==139 |
| 149 | |
| 150 | |
| 151 | @pytest.mark.asyncio |
| 152 | @pytest.mark.parametrize('count,queries',[(0,0),(1,2),(18,2)]) |
| 153 | async def test_stage_b_batch_queries_and_no_network(count,queries,monkeypatch): |
| 154 | from app.global_scanner import GlobalScanner |
| 155 | from app.persistence import SqliteResearchPersistence |
| 156 | from test_global_scanner import instrument,persisted,scan |
| 157 | from app.nse_historical_daily import NseHistoricalDailyProvider |
| 158 | monkeypatch.setattr(NseHistoricalDailyProvider,'fetch',Mock(side_effect=AssertionError('provider called'))) |
| 159 | store=SqliteResearchPersistence(); items=[instrument(n) for n in range(1,count+1)] |
| 160 | for item in items: |
| 161 | persisted(store,item) |
| 162 | key=UUID(item['globalInstrumentId']) |
| 163 | for b in candles(range(100,140)): |
| 164 | store.upsert_daily_market_bar(b.model_copy(update={'global_instrument_id':key})) |
| 165 | initial=await scan(items,store) |
| 166 | before=initial.model_dump() |
| 167 | statements=[]; store._connection.set_trace_callback(statements.append) |
| 168 | result=GlobalScanner(None,store).enrich_candidates(initial) |
| 169 | assert len(statements)==queries and len(result)==count |
| 170 | assert initial.model_dump()==before |
| 171 | if count: |
| 172 | assert sum('global_daily_market_bars' in s for s in statements)==1 |
| 173 | assert all(c.technical_feature_snapshot.technical_input_source=='DAILY_MARKET_BAR_NSE' for c in result) |
| 174 | assert all(c.technical_score==c.stage_b_score for c in result) |
| 175 | assert result==GlobalScanner(None,store).enrich_candidates(initial) |
| 176 | |
| 177 | @pytest.mark.parametrize('change',[{'open':None},{'high':None},{'low':None}]) |
| 178 | def test_missing_latest_ohlc(change): |
| 179 | rows=candles(range(100,140)); rows[-1]=rows[-1].model_copy(update=change) |
| 180 | r=evaluate(rows) |
| 181 | assert r.atr14 is r.adx14 is None and r.feature_readiness['ATR14']=='MISSING_OHLC' |
| 182 | assert r.technical_score is not None |
| 183 | |
| 184 | |
| 185 | def test_conflict_cannot_be_compressed_out_of_ohlc_or_volume(): |
| 186 | rows=candles(range(100,140)) |
| 187 | duplicate=rows[-2].model_copy(update={'close':Decimal(999)}) |
| 188 | r=evaluate(rows+[duplicate]) |
| 189 | assert r.atr14 is r.adx14 is r.volume_ratio20 is None |
| 190 | assert r.feature_readiness['VOLUME20']=='CONFLICTING' |
| 191 | |
| 192 | |
| 193 | def test_recovery_after_missing_candle_and_no_source_identity_guessing(): |
| 194 | rows=candles(range(100,160)) |
| 195 | rows[10]=rows[10].model_copy(update={'low':None}) |
| 196 | r=evaluate(rows) |
| 197 | assert r.atr14==2 and r.adx14==100 |
| 198 | unknown=[b.model_copy(update={'global_instrument_id':UUID(int=2)}) for b in rows] |
| 199 | assert evaluate(unknown).latest_price is None |
| 200 | assert evaluate(rows,trusted_providers=frozenset()).latest_price is None |
| 201 | assert evaluate([b.model_copy(update={'source_mode':'DEMO'}) for b in rows]).latest_price is None |
| 202 | |
| 203 | |
| 204 | def test_exchange_date_is_not_utc_observation_date(): |
| 205 | # At 20:00 UTC it is already the next exchange DATE. Retrieval is known. |
| 206 | asof=datetime(2026,9,13,20,tzinfo=timezone.utc) |
| 207 | row=candles([100])[0].model_copy(update={'trading_date':datetime(2026,9,14).date(),'retrieved_at':asof}) |
| 208 | r=TechnicalFeatureEngine().compute(STOCK,[],as_of=asof,currency='INR',daily_bar_history=[row]) |
| 209 | assert r.history_trading_start==r.history_trading_end==row.trading_date |
| 210 | assert r.observation_count==1 |
| 211 | |
| 212 | |
| 213 | def test_no_network_in_ohlcv_computation(monkeypatch): |
| 214 | import socket |
| 215 | monkeypatch.setattr(socket,'create_connection',Mock(side_effect=AssertionError('network'))) |
| 216 | rows=candles(range(100,160)) |
| 217 | engine=TechnicalFeatureEngine() |
| 218 | before=[b.model_dump() for b in rows] |
| 219 | first=engine.compute(STOCK,[],as_of=NOW,daily_bar_history=rows) |
| 220 | assert first==engine.compute(STOCK,[],as_of=NOW,daily_bar_history=reversed(rows)) |
| 221 | assert before==[b.model_dump() for b in rows] |
| 222 | |
| 223 | |
| 224 | def test_stale_candles_expose_diagnostics_but_do_not_score(): |
| 225 | rows=candles(range(100,140)) |
| 226 | rows=[b.model_copy(update={'trading_date':b.trading_date-timedelta(days=20)}) for b in rows] |
| 227 | r=evaluate(rows) |
| 228 | assert r.atr14==2 and r.adx14==100 and r.feature_readiness['ATR14']=='STALE' |
| 229 | assert r.technical_score is None and r.technical_state=='INSUFFICIENT_DATA' |
| 230 | |
| 231 | |
| 232 | @pytest.mark.asyncio |
| 233 | async def test_stage_b_score_change_is_volume_evidence_only_and_fallback_works(): |
| 234 | from app.global_scanner import GlobalScanner |
| 235 | from app.persistence import SqliteResearchPersistence |
| 236 | from test_global_scanner import instrument,persisted,scan |
| 237 | store=SqliteResearchPersistence(); items=[instrument(1),instrument(2)] |
| 238 | values=[100]*60+[104] |
| 239 | for item in items: |
| 240 | persisted(store,item,price=False) |
| 241 | for p in history(values,UUID(item['globalInstrumentId'])): |
| 242 | store.upsert_market_price_observation(p.model_copy(update={'provider':'YAHOO_FINANCE'})) |
| 243 | initial=await scan(items,store) |
| 244 | scanner=GlobalScanner(None,store) |
| 245 | before={r.global_instrument_id:r for r in scanner.enrich_candidates(initial)} |
| 246 | store.upsert_daily_market_bars(candles(values,volumes=[100]*60+[200])) |
| 247 | after={r.global_instrument_id:r for r in scanner.enrich_candidates(initial)} |
| 248 | assert after[UUID(int=2)]==before[UUID(int=2)] |
| 249 | assert before[STOCK].pre_score==after[STOCK].pre_score |
| 250 | assert after[STOCK].technical_score==min(100,before[STOCK].technical_score+5) |
| 251 | assert after[STOCK].stage_b_score==after[STOCK].technical_score |
| 252 | |
| 253 | @pytest.mark.parametrize('daily_count,age,reason',[(4,0,'NSE_DAILY_HISTORY_BELOW_20'),(40,20,'NSE_DAILY_HISTORY_STALE')]) |
| 254 | def test_immature_or_stale_daily_source_does_not_disable_current_fallback(daily_count,age,reason): |
| 255 | rows=candles(range(200,200+daily_count)) |
| 256 | rows=[b.model_copy(update={'trading_date':b.trading_date-timedelta(days=age)}) for b in rows] |
| 257 | fallback=history(range(100,160)) |
| 258 | before=evaluate([],fallback) |
| 259 | r=evaluate(rows,fallback) |
| 260 | assert r.technical_input_source=='CLOSE_ONLY_FALLBACK' and reason in r.source_diagnostics |
| 261 | assert r.technical_score==before.technical_score and r.latest_price==before.latest_price |
| 262 | assert r.atr14 is r.adx14 is r.current_volume is None |
| 263 | assert r.daily_bar_observation_count==daily_count |
| 264 | |
| 265 | |
| 266 | def test_short_daily_still_used_when_all_history_insufficient(): |
| 267 | r=evaluate(candles([100]*4),history([200]*3)) |
| 268 | assert r.technical_input_source=='DAILY_MARKET_BAR_NSE' and r.observation_count==4 |
| 269 | assert r.technical_state=='INSUFFICIENT_DATA' |