| 1 | from datetime import datetime, timedelta, timezone |
| 2 | from decimal import Decimal |
| 3 | from uuid import UUID |
| 4 | |
| 5 | import pytest |
| 6 | |
| 7 | from app.models import MarketPriceObservation |
| 8 | from app.technical_features import TechnicalConfig, TechnicalFeatureEngine, PersistedVolumeObservation |
| 9 | |
| 10 | |
| 11 | NOW = datetime(2026, 9, 13, 18, tzinfo=timezone.utc) |
| 12 | STOCK = UUID(int=1) |
| 13 | |
| 14 | |
| 15 | def history(values, instrument_id=STOCK, currency="INR", end=NOW): |
| 16 | # Observed weekday sessions, with no calendar interpolation by the engine. |
| 17 | dates, day = [], end |
| 18 | while len(dates) < len(values): |
| 19 | if day.weekday() < 5: |
| 20 | dates.append(day) |
| 21 | day -= timedelta(days=1) |
| 22 | dates.reverse() |
| 23 | return [MarketPriceObservation(instrument_id=instrument_id, price=Decimal(str(value)), currency=currency, |
| 24 | observed_at=stamp, retrieved_at=stamp, provider="PERSISTED", source_url="https://evidence.test/history") |
| 25 | for value, stamp in zip(values, dates)] |
| 26 | |
| 27 | |
| 28 | def compute(values, **kwargs): |
| 29 | return TechnicalFeatureEngine().compute(STOCK, history(values), as_of=NOW, currency="INR", **kwargs) |
| 30 | |
| 31 | |
| 32 | @pytest.mark.parametrize("values,state", [ |
| 33 | ([100 + .2*i for i in range(260)], "UPTREND"), |
| 34 | ([200 - .2*i for i in range(260)], "DOWNTREND"), |
| 35 | ([100]*260, "BASE_BUILDING"), |
| 36 | ([100 + 10*(i % 2) for i in range(260)], "RANGE_BOUND"), |
| 37 | ([100]*60 + [104], "BREAKOUT"), |
| 38 | ([100 + .2*i for i in range(255)] + [151, 150, 149, 148, 147], "PULLBACK_IN_UPTREND"), |
| 39 | ([100 + .2*i for i in range(259)] + [200], "OVEREXTENDED"), |
| 40 | ([200 - .6*i for i in range(240)] + [56.6 + .3*i for i in range(20)], "REVERSAL_CANDIDATE"), |
| 41 | ([100]*19, "INSUFFICIENT_DATA"), |
| 42 | ]) |
| 43 | def test_explicit_technical_states(values, state): |
| 44 | result = compute(values) |
| 45 | assert result.technical_state == state |
| 46 | assert (result.technical_score is None) == (state == "INSUFFICIENT_DATA") |
| 47 | |
| 48 | |
| 49 | @pytest.mark.parametrize("count,readiness", [(19, "INSUFFICIENT_HISTORY"), (20, "SHORT_HISTORY"), |
| 50 | (49, "SHORT_HISTORY"), (50, "MEDIUM_HISTORY"), (99, "MEDIUM_HISTORY"), (100, "EXTENDED_HISTORY"), |
| 51 | (199, "EXTENDED_HISTORY"), (200, "FULL_HISTORY")]) |
| 52 | def test_history_boundaries_and_exact_dma_math(count, readiness): |
| 53 | result = compute(list(range(1, count+1))) |
| 54 | assert result.history_readiness == readiness |
| 55 | for window in (20, 50, 100, 200): |
| 56 | value = getattr(result, f"dma{window}") |
| 57 | if count >= window: |
| 58 | expected = (count + count-window+1) / 2 |
| 59 | assert value == expected |
| 60 | assert getattr(result, f"distance_to_dma{window}_pct") == pytest.approx((count/expected-1)*100) |
| 61 | else: |
| 62 | assert value is None |
| 63 | assert result.feature_states[f"dma{window}"] == "MISSING" |
| 64 | |
| 65 | |
| 66 | def test_rsi_wilder_independent_known_worksheet(): |
| 67 | values = [44.34, 44.09, 44.15, 43.61, 44.33, 44.83, 45.10, 45.42, 45.84, 46.08, 45.89, 46.03, 45.61, 46.28, 46.28] |
| 68 | # First 14 changes: total gains 3.34, losses 1.40. No EMA warmup ambiguity. |
| 69 | assert compute(values).rsi14 == pytest.approx(100*3.34/(3.34+1.40), abs=1e-7) |
| 70 | # One Wilder update: gain=(3.34/14*13)/14; loss=(1.40/14*13+.28)/14. |
| 71 | gain, loss = Decimal("3.34")/14*13/14, (Decimal("1.40")/14*13+Decimal(".28"))/14 |
| 72 | assert compute(values + [46.0]).rsi14 == pytest.approx(float(100*gain/(gain+loss)), abs=1e-7) |
| 73 | assert compute(values[:14]).rsi14 is None |
| 74 | assert compute([100]*20).rsi14 == 50 |
| 75 | assert compute(list(range(1, 21))).rsi14 == 100 |
| 76 | assert compute(list(range(21, 1, -1))).rsi14 == 0 |
| 77 | |
| 78 | |
| 79 | def test_macd_linear_series_exact_and_independent_nonlinear_ema(): |
| 80 | result = compute(list(range(1, 61))) |
| 81 | assert result.macd == pytest.approx(7) |
| 82 | assert result.macd_signal == pytest.approx(7) |
| 83 | assert result.macd_histogram == pytest.approx(0) |
| 84 | values = [Decimal(100) + Decimal(i*i % 37) for i in range(70)] |
| 85 | # Independent closed-form weighted sum, not the implementation's recursion. |
| 86 | def ema_at(series, period, index): |
| 87 | alpha = Decimal(2)/Decimal(period+1) |
| 88 | decay = 1-alpha |
| 89 | tail = index-period+1 |
| 90 | seed = sum(series[:period])/period |
| 91 | return seed*decay**tail + sum(alpha*series[k]*decay**(index-k) for k in range(period, index+1)) |
| 92 | macds = [ema_at(values, 12, i)-ema_at(values, 26, i) for i in range(25, len(values))] |
| 93 | expected_signal = ema_at(macds, 9, len(macds)-1) |
| 94 | result = compute(values) |
| 95 | assert result.macd == pytest.approx(float(macds[-1]), abs=1e-7) |
| 96 | assert result.macd_signal == pytest.approx(float(expected_signal), abs=1e-7) |
| 97 | assert result.macd_histogram == pytest.approx(float(macds[-1]-expected_signal), abs=1e-7) |
| 98 | assert compute([100]*25).macd is None |
| 99 | assert compute([100]*26).macd == 0 |
| 100 | assert compute([100]*33).macd_signal is None |
| 101 | assert compute([100]*34).macd_signal == 0 |
| 102 | |
| 103 | |
| 104 | def test_missing_ohlc_volume_and_actual_zero_volume(): |
| 105 | rows = history([100]*30) |
| 106 | engine = TechnicalFeatureEngine() |
| 107 | result = engine.compute(STOCK, rows, as_of=NOW) |
| 108 | assert result.adx14 is result.atr14 is result.atr_pct is None |
| 109 | assert result.volume_average20 is result.volume_ratio20 is None |
| 110 | volumes = [PersistedVolumeObservation(instrument_id=STOCK, observed_at=r.observed_at, |
| 111 | retrieved_at=r.retrieved_at, volume=100 if i < 29 else 0, provider=r.provider, source_url=r.source_url) for i, r in enumerate(rows)] |
| 112 | result = engine.compute(STOCK, rows, as_of=NOW, volume_history=volumes) |
| 113 | assert result.volume_average20 == 100 |
| 114 | assert result.volume_ratio20 == 0 |
| 115 | assert result.feature_states["volumeRatio20"] == "AVAILABLE" |
| 116 | zero = [r.model_copy(update={"volume": Decimal(0)}) for r in volumes] |
| 117 | result = engine.compute(STOCK, rows, as_of=NOW, volume_history=zero) |
| 118 | assert result.volume_average20 == 0 and result.volume_ratio20 is None |
| 119 | assert "NONZERO_VOLUME_BASELINE" in result.missing_inputs |
| 120 | |
| 121 | |
| 122 | def test_latest_conflict_is_not_averaged_or_replaced_by_prior_close(): |
| 123 | rows = history(range(100, 160)) |
| 124 | conflict = rows[-1].model_copy(update={"price": Decimal(999), "provider": "OTHER"}) |
| 125 | engine = TechnicalFeatureEngine() |
| 126 | result = engine.compute(STOCK, rows+[conflict], as_of=NOW) |
| 127 | assert result.latest_price is result.dma20 is result.technical_score is None |
| 128 | assert result.technical_state == "INSUFFICIENT_DATA" and result.confidence == 0 |
| 129 | assert result.feature_states["latestPrice"] == "CONFLICTING" |
| 130 | assert result == engine.compute(STOCK, list(reversed(rows+[conflict])), as_of=NOW) |
| 131 | |
| 132 | |
| 133 | def test_same_date_deduplication_latest_timestamp_provenance_and_no_fill(): |
| 134 | rows = history(range(100, 160)) |
| 135 | engine = TechnicalFeatureEngine() |
| 136 | first = engine.compute(STOCK, rows, as_of=NOW) |
| 137 | duplicate = rows[-1].model_copy(update={"provider": "OTHER"}) |
| 138 | repeated = engine.compute(STOCK, [*reversed(rows), duplicate], as_of=NOW) |
| 139 | assert repeated.observation_count == 60 and repeated.dma50 == first.dma50 |
| 140 | assert repeated.duplicate_observation_count == 1 |
| 141 | later = rows[-1].model_copy(update={"price": Decimal(160), "observed_at": rows[-1].observed_at+timedelta(minutes=1)}) |
| 142 | assert engine.compute(STOCK, rows+[later], as_of=NOW).latest_price == 160 |
| 143 | sparse = [r for i, r in enumerate(rows) if i % 2] |
| 144 | assert engine.compute(STOCK, sparse, as_of=NOW).observation_count == 30 |
| 145 | |
| 146 | |
| 147 | def test_stale_future_currency_and_provider_filtering(): |
| 148 | engine = TechnicalFeatureEngine() |
| 149 | rows = history(range(100, 160), end=NOW-timedelta(days=20)) |
| 150 | result = engine.compute(STOCK, rows, as_of=NOW) |
| 151 | assert result.dma50 is not None and result.feature_states["dma50"] == "STALE" |
| 152 | assert result.technical_state == "INSUFFICIENT_DATA" and result.technical_score is None |
| 153 | assert result.stale_inputs == ["PRICE_HISTORY"] |
| 154 | rows = history(range(100, 160)) |
| 155 | future = rows[-1].model_copy(update={"retrieved_at": NOW+timedelta(days=1), "price": Decimal(999)}) |
| 156 | result = engine.compute(STOCK, rows+[future], as_of=NOW) |
| 157 | assert result.latest_price == 159 and result.rejected_observation_count == 1 |
| 158 | assert engine.compute(STOCK, rows, as_of=NOW, currency="USD").latest_price is None |
| 159 | assert engine.compute(STOCK, rows, as_of=NOW, trusted_providers=frozenset()).latest_price is None |
| 160 | |
| 161 | |
| 162 | def test_return_offsets_support_and_year_readiness(): |
| 163 | result = compute(list(range(1, 254))) |
| 164 | for field, offset in [("return1_w", 5), ("return1_m", 21), ("return3_m", 63), ("return6_m", 126), ("return1_y", 252)]: |
| 165 | assert getattr(result, field) == pytest.approx((253/(253-offset)-1)*100) |
| 166 | assert result.support_level == 233 and result.resistance_level == 252 |
| 167 | assert result.distance_to_support_pct == pytest.approx((253/233-1)*100) |
| 168 | assert result.higher_highs_higher_lows is True and result.lower_highs_lower_lows is False |
| 169 | assert compute([100]*251).distance_from52_week_high_pct is None |
| 170 | assert compute([100]*252).distance_from52_week_high_pct == 0 |
| 171 | assert compute([100]*252).return1_y is None |
| 172 | |
| 173 | |
| 174 | def test_deterministic_pure_execution_and_configuration(monkeypatch): |
| 175 | import socket |
| 176 | monkeypatch.setattr(socket, "create_connection", lambda *a, **k: pytest.fail("Network invoked")) |
| 177 | rows = history([100+.1*i for i in range(260)]) |
| 178 | original = [r.model_dump() for r in rows] |
| 179 | engine = TechnicalFeatureEngine() |
| 180 | first = engine.compute(STOCK, rows, as_of=NOW) |
| 181 | assert first == engine.compute(STOCK, reversed(rows), as_of=NOW) |
| 182 | assert original == [r.model_dump() for r in rows] |
| 183 | assert first.feature_version == "TECHNICAL_FEATURES_V2" |
| 184 | with pytest.raises(ValueError): TechnicalConfig(momentum_full_scale_pct=0) |
| 185 | with pytest.raises(ValueError): TechnicalConfig(score_weights=(float("nan"), 1, 1)) |