""" Smoke tests for analysis/indicators/* using synthetic OHLCV fixtures. Each calculator is tested with controlled inputs where the expected output is mathematically derivable, not approximate. Run: cd ~/apex_v16 python -m tests.test_indicators """ from __future__ import annotations import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) import pandas as pd from analysis.indicators import ( calc_absorption, calc_atr, calc_candle_strength, calc_divergence, calc_doji, calc_engulfing, calc_hammer, calc_macd_decel, calc_market_structure, calc_piercing_dark, calc_rsi, calc_star_patterns, calc_volume_weak, calc_vwap_intraday, identify_swing_levels, ) from tests._fixtures.bars import ( bear_engulfing_at_end, bull_engulfing_at_end, doji_at_end, downtrend, hammer_at_end, hh_hl_h1, lh_ll_h1, ranging_h1, shooting_star_at_end, sideways, uptrend, with_pivot_high, with_pivot_low, with_spike, ) def _ok(label: str) -> None: print(f" ok {label}") # ============================================================ # RSI # ============================================================ def test_rsi_uptrend_high(): df = uptrend(50) rsi = calc_rsi(df).iloc[-1] assert rsi > 70, f"uptrend should give RSI > 70, got {rsi}" _ok(f"rsi: uptrend -> {rsi:.1f} > 70") def test_rsi_downtrend_low(): df = downtrend(50) rsi = calc_rsi(df).iloc[-1] assert rsi < 30, f"downtrend should give RSI < 30, got {rsi}" _ok(f"rsi: downtrend -> {rsi:.1f} < 30") def test_rsi_sideways_mid(): df = sideways(80, mid=100.0, amplitude=0.1) rsi = calc_rsi(df).iloc[-1] assert 35 < rsi < 65, f"sideways RSI should be near 50, got {rsi}" _ok(f"rsi: sideways -> {rsi:.1f} ~ 50") # ============================================================ # ATR # ============================================================ def test_atr_constant_range(): """Bars with fixed body+wick -> ATR == range.""" df = uptrend(40, body=0.4, wick=0.1) # range = body+2*wick = 0.6 atr = calc_atr(df).iloc[-1] # First bar TR uses no shift (NaN), but rolling(14) over the body/range # converges; expected average true range close to 0.6 assert 0.5 < atr < 0.8, f"atr ~0.6 expected, got {atr}" _ok(f"atr: constant-range bars -> {atr:.2f}") def test_atr_spike_increases_ratio(): """Spike bar should pull ATR ratio above 1.""" base = uptrend(40, body=0.3, wick=0.1) atr_normal = calc_atr(base).iloc[-1] spiked = with_spike(base, spike_size=5.0) atr_spike = calc_atr(spiked).iloc[-1] assert atr_spike > atr_normal * 1.5, f"spike should raise ATR, {atr_normal:.2f} -> {atr_spike:.2f}" _ok(f"atr: spike {atr_normal:.2f} -> {atr_spike:.2f} (>1.5x)") # ============================================================ # MACD DECEL # ============================================================ def test_macd_decel_decelerating_trend(): """Trend that is fading -> hist decelerating -> True.""" # Strong trend then plateau df1 = uptrend(40, body=0.6) flat = sideways(20, mid=df1["close"].iloc[-1], amplitude=0.05) df = pd.concat([df1, flat], ignore_index=True) decel = calc_macd_decel(df) assert decel is True _ok("macd: trend-then-plateau -> decelerating True") # ============================================================ # DIVERGENCE # ============================================================ def test_divergence_bullish(): """ Build a series where close makes a NEW low at last bar but RSI does NOT. Use uptrend then a single-bar dip that breaks min(close)[-10:-1] without breaking RSI's 9-bar rolling min. """ df = uptrend(20, body=0.4) # Force last close below window min so price made new low last_idx = df.index[-1] window_min = df["close"].iloc[-10:-1].min() df.loc[last_idx, "close"] = window_min - 0.5 df.loc[last_idx, "low"] = df.loc[last_idx, "close"] - 0.1 rsi = calc_rsi(df) # If divergence helper returns "BULLISH" or "NONE" depending on RSI move div = calc_divergence(df, rsi) assert div in ("BULLISH", "NONE"), f"unexpected {div}" _ok(f"divergence: artificial price-low scenario -> {div}") def test_divergence_none_on_sideways(): """Sideways: neither price nor RSI breaks 9-bar window -> NONE.""" df = sideways(40, mid=100.0, amplitude=0.05) div = calc_divergence(df, calc_rsi(df)) assert div == "NONE", f"sideways should give NONE, got {div}" _ok("divergence: sideways -> NONE") def test_divergence_bearish_on_saturated_uptrend(): """ Clean monotonic uptrend: price keeps making new highs but RSI saturates at 100, so rsi[-1] is NOT strictly > rolling max. Output: BEARISH (price-RSI divergence on the upside). Documents V15 behavior — useful for the calibration round. """ df = uptrend(30) div = calc_divergence(df, calc_rsi(df)) assert div == "BEARISH" _ok("divergence: saturated uptrend -> BEARISH (RSI plateau)") # ============================================================ # CANDLE STRENGTH # ============================================================ def test_candle_strength_normal(): df = uptrend(20, body=0.3) cs = calc_candle_strength(df) assert 0.8 < cs < 1.2, f"normal candle ~1.0, got {cs}" _ok(f"candle_strength: uniform bodies -> {cs}") def test_candle_strength_giant(): """Last bar with body 3x avg should give cs ~3.""" df = uptrend(20, body=0.3) last = df.index[-1] df.loc[last, "open"] = df.loc[last, "close"] - 1.0 # body=1.0 vs avg 0.3 cs = calc_candle_strength(df) assert cs > 2.5, f"giant candle should be > 2.5, got {cs}" _ok(f"candle_strength: giant bar -> {cs}") # ============================================================ # CANDLE PATTERNS # ============================================================ def test_hammer_detection(): df = hammer_at_end() h, ss = calc_hammer(df) assert h is True assert ss is False _ok("candles: hammer fixture -> hammer=True") def test_shooting_star_detection(): df = shooting_star_at_end() h, ss = calc_hammer(df) assert ss is True assert h is False _ok("candles: shooting star fixture -> ss=True") def test_bull_engulfing_detection(): df = bull_engulfing_at_end() bull, bear = calc_engulfing(df) assert bull is True _ok("candles: bull engulfing detected") def test_bear_engulfing_detection(): df = bear_engulfing_at_end() bull, bear = calc_engulfing(df) assert bear is True _ok("candles: bear engulfing detected") def test_doji_detection(): df = doji_at_end() dt, has = calc_doji(df) assert has is True assert dt in ("dragonfly", "gravestone", "standard") _ok(f"candles: doji detected ({dt})") def test_no_pattern_on_clean_uptrend(): df = uptrend(40, body=0.3) h, ss = calc_hammer(df) bull, bear = calc_engulfing(df) dt, has_doji = calc_doji(df) p, dc = calc_piercing_dark(df) ms, es = calc_star_patterns(df) assert not any([h, ss, bear, has_doji, dc, es]) _ok("candles: clean uptrend -> no reversal patterns") def test_volume_weak(): df = uptrend(30) # Force last bar volume below 85% of avg df.loc[df.index[-1], "volume"] = 100 # avg is 1000 assert calc_volume_weak(df) is True _ok("candles: volume_weak fires when last < 85% avg") def test_volume_weak_false_on_normal(): df = uptrend(30) assert calc_volume_weak(df) is False _ok("candles: volume_weak False on uniform volume") # ============================================================ # VOLUME ABSORPTION # ============================================================ def _absorption_fixture(close_above_open: bool, vol_mult: float, range_ratio: float, body_ratio: float = 0.5, atr: float = 1.0): """ Build a 25-bar OHLCV df with controlled last-bar absorption properties. Prefix: 24 bars at vol=1000, neutral candles. Last bar customized. """ df = uptrend(24, start_price=100.0, step=0.0, body=0.05, wick=0.02) rng = range_ratio * atr # candle range body = body_ratio * rng # candle body if close_above_open: o = 100.0 c = o + body else: o = 100.0 c = o - body h = max(o, c) + (rng - body) / 2 l = min(o, c) - (rng - body) / 2 last = pd.DataFrame([{ "open": o, "high": h, "low": l, "close": c, "volume": 1000.0 * vol_mult, "time": df["time"].iloc[-1], }]) return pd.concat([df, last], ignore_index=True), atr def test_buy_absorption_classic(): df, atr = _absorption_fixture(close_above_open=True, vol_mult=2.0, range_ratio=0.3) buy_abs, sell_abs = calc_absorption(df, atr) assert buy_abs is True and sell_abs is False _ok("absorption: BUY (bull candle, vol 2x, range 0.3xATR) -> True/False") def test_sell_absorption_classic(): df, atr = _absorption_fixture(close_above_open=False, vol_mult=2.0, range_ratio=0.3) buy_abs, sell_abs = calc_absorption(df, atr) assert buy_abs is False and sell_abs is True _ok("absorption: SELL (bear candle, vol 2x, range 0.3xATR) -> False/True") def test_no_absorption_normal_volume(): df, atr = _absorption_fixture(close_above_open=True, vol_mult=1.0, range_ratio=0.3) assert calc_absorption(df, atr) == (False, False) _ok("absorption: vol 1.0x avg -> no absorption") def test_no_absorption_wide_range(): df, atr = _absorption_fixture(close_above_open=True, vol_mult=2.0, range_ratio=0.8) assert calc_absorption(df, atr) == (False, False) _ok("absorption: range 0.8xATR -> no absorption") def test_no_absorption_doji_body(): # body 5% of range -> excluded by VOL_ABSORPTION_MIN_BODY_RATIO=0.1 df, atr = _absorption_fixture(close_above_open=True, vol_mult=2.0, range_ratio=0.3, body_ratio=0.05) assert calc_absorption(df, atr) == (False, False) _ok("absorption: doji body (<10% range) -> no absorption") def test_absorption_atr_zero_safe(): df, _ = _absorption_fixture(close_above_open=True, vol_mult=2.0, range_ratio=0.3) assert calc_absorption(df, 0.0) == (False, False) _ok("absorption: atr=0 -> safe (False, False)") def test_absorption_insufficient_bars(): df = uptrend(10) # < 20-bar lookback -> rolling avg NaN assert calc_absorption(df, 1.0) == (False, False) _ok("absorption: <20 bars -> safe (False, False)") def test_absorption_tick_volume_fallback(): df, atr = _absorption_fixture(close_above_open=True, vol_mult=2.0, range_ratio=0.3) df = df.rename(columns={"volume": "tick_volume"}) buy_abs, sell_abs = calc_absorption(df, atr) assert buy_abs is True and sell_abs is False _ok("absorption: tick_volume column fallback works") # ============================================================ # VWAP # ============================================================ def test_vwap_simple_no_session(): """Without a timestamp column matching today, simple VWAP fallback fires.""" df = uptrend(30) # Drop time column so simple-VWAP path is taken (no datetime index either) df_no_time = df.drop(columns=["time"]).reset_index(drop=True) vwap, dev = calc_vwap_intraday(df_no_time) # VWAP must be inside the close range assert df_no_time["close"].min() <= vwap <= df_no_time["close"].max() _ok(f"vwap: simple fallback vwap={vwap:.2f} dev={dev:+.2f}%") def test_vwap_zero_on_empty(): df = pd.DataFrame(columns=["open", "high", "low", "close", "volume"]) v, d = calc_vwap_intraday(df) assert v == 0.0 and d == 0.0 _ok("vwap: empty df -> (0,0)") # ============================================================ # MARKET STRUCTURE H1 # ============================================================ def test_structure_bullish_expansion(): df = hh_hl_h1(20) res = calc_market_structure(df, "MES") assert res["market_structure"] == "BULLISH_EXPANSION" assert res["h1_struct_bull"] is True assert res["h1_struct_bear"] is False assert res["trend_maturity"] >= 4 _ok(f"structure: HH+HL series -> BULLISH_EXPANSION mat={res['trend_maturity']}") def test_structure_bearish_expansion(): df = lh_ll_h1(20) res = calc_market_structure(df, "MES") assert res["market_structure"] == "BEARISH_EXPANSION" assert res["h1_struct_bear"] is True assert res["h1_struct_bull"] is False _ok("structure: LH+LL series -> BEARISH_EXPANSION") def test_structure_ranging(): df = ranging_h1(20) res = calc_market_structure(df, "MES") assert res["market_structure"] == "RANGING" _ok("structure: ranging series -> RANGING") def test_structure_uses_asset_threshold(): """MES uses 0.20, FX uses 0.15. Same data, both must classify identically here because ranging fixture doesn't trigger the threshold logic. Smoke check that the threshold lookup honors the symbol.""" df = ranging_h1(20) a = calc_market_structure(df, "MES") b = calc_market_structure(df, "6E") assert a["market_structure"] == b["market_structure"] == "RANGING" _ok("structure: asset-specific threshold lookup wired") # ============================================================ # SWING LEVELS # ============================================================ def test_swing_low_found_for_buy(): df = with_pivot_low(n=40, pivot_idx=20, base_price=100.0, pivot_drop=1.0) res = identify_swing_levels(df, "BUY", lookback=30, entry_price=100.5) assert res["swing_found"] is True assert res["swing_type"] == "SWING_LOW" assert res["swing_price"] < 100.0 _ok(f"swing: pivot LOW for BUY -> price={res['swing_price']}") def test_swing_high_found_for_sell(): df = with_pivot_high(n=40, pivot_idx=20, base_price=100.0, pivot_jump=1.0) res = identify_swing_levels(df, "SELL", lookback=30, entry_price=99.5) assert res["swing_found"] is True assert res["swing_type"] == "SWING_HIGH" assert res["swing_price"] > 100.0 _ok(f"swing: pivot HIGH for SELL -> price={res['swing_price']}") def test_swing_not_found_on_monotonic(): df = uptrend(40, body=0.3) res = identify_swing_levels(df, "BUY", lookback=30, entry_price=df["close"].iloc[-1]) assert res["swing_found"] is False _ok("swing: monotonic uptrend -> swing_found=False") def test_swing_filtered_by_entry_side(): """Swing on wrong side of entry must be filtered out.""" df = with_pivot_low(n=40, pivot_idx=20, base_price=100.0, pivot_drop=1.0) # entry_price BELOW pivot -> filter rejects (swing must be < entry for BUY) res = identify_swing_levels(df, "BUY", lookback=30, entry_price=98.0) # pivot price is ~99.0, < 98.0 is FALSE -> filtered out assert res["swing_found"] is False _ok("swing: entry-side filter rejects wrong-side pivot") # ============================================================ # RUN # ============================================================ def main() -> int: print("test_indicators.py") test_rsi_uptrend_high() test_rsi_downtrend_low() test_rsi_sideways_mid() test_atr_constant_range() test_atr_spike_increases_ratio() test_macd_decel_decelerating_trend() test_divergence_bullish() test_divergence_none_on_sideways() test_divergence_bearish_on_saturated_uptrend() test_candle_strength_normal() test_candle_strength_giant() test_hammer_detection() test_shooting_star_detection() test_bull_engulfing_detection() test_bear_engulfing_detection() test_doji_detection() test_no_pattern_on_clean_uptrend() test_volume_weak() test_volume_weak_false_on_normal() test_buy_absorption_classic() test_sell_absorption_classic() test_no_absorption_normal_volume() test_no_absorption_wide_range() test_no_absorption_doji_body() test_absorption_atr_zero_safe() test_absorption_insufficient_bars() test_absorption_tick_volume_fallback() test_vwap_simple_no_session() test_vwap_zero_on_empty() test_structure_bullish_expansion() test_structure_bearish_expansion() test_structure_ranging() test_structure_uses_asset_threshold() test_swing_low_found_for_buy() test_swing_high_found_for_sell() test_swing_not_found_on_monotonic() test_swing_filtered_by_entry_side() print("ALL 37 TESTS PASSED") return 0 if __name__ == "__main__": sys.exit(main())