"""Unit tests for services.signal_engine.signals.cup_handle — Cup & Handle evaluator. Requirements: 2.4, 2.6, 2.7 """ from __future__ import annotations from datetime import datetime, timezone from services.signal_engine.models import OHLCVBar, SignalDirection from services.signal_engine.signals.cup_handle import ( DEFAULT_MIN_BARS, CupHandleEvaluator, ) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _bar( close: float, high: float | None = None, low: float | None = None, ) -> OHLCVBar: """Create a minimal OHLCVBar for testing.""" h = high if high is not None else close lo = low if low is not None else close return OHLCVBar( timestamp=datetime(2024, 1, 1, tzinfo=timezone.utc), open=close, high=h, low=lo, close=close, volume=1000.0, ) def _make_cup_handle_bars( n: int = 40, left_rim: float = 100.0, bottom: float = 80.0, right_rim: float = 99.0, handle_low: float = 96.0, ) -> list[OHLCVBar]: """Create a synthetic cup & handle pattern. Generates bars that form: 1. Rise to left_rim in the first third 2. Descent to bottom in the middle 3. Rise to right_rim in the last third 4. Small pullback to handle_low at the end """ bars: list[OHLCVBar] = [] first_third = n // 3 last_third_start = n - (n // 3) handle_start = n - max(2, int(n * 0.15)) for i in range(n): if i < first_third: # Rise to left rim frac = i / max(1, first_third - 1) price = bottom + frac * (left_rim - bottom) h = price + 1.0 lo = price - 1.0 elif i < last_third_start: # Cup: descend to bottom then rise mid = (first_third + last_third_start) / 2.0 if i <= mid: frac = (i - first_third) / max(1, mid - first_third) price = left_rim - frac * (left_rim - bottom) else: frac = (i - mid) / max(1, last_third_start - mid) price = bottom + frac * (right_rim - bottom) h = price + 1.0 lo = price - 1.0 elif i < handle_start: # Rise to right rim frac = (i - last_third_start) / max(1, handle_start - last_third_start - 1) price = right_rim - 2.0 + frac * 2.0 h = price + 1.0 lo = price - 1.0 else: # Handle: small pullback handle_len = n - handle_start frac = (i - handle_start) / max(1, handle_len - 1) price = right_rim - frac * (right_rim - handle_low) h = price + 0.5 lo = price - 0.5 bars.append(_bar(price, high=h, low=lo)) # Ensure the left rim bar has the correct high bars[first_third - 1] = _bar( left_rim - 1.0, high=left_rim, low=left_rim - 2.0, ) # Ensure the right rim bar has the correct high right_rim_idx = last_third_start + (handle_start - last_third_start) // 2 if right_rim_idx < n: bars[right_rim_idx] = _bar( right_rim - 1.0, high=right_rim, low=right_rim - 2.0, ) return bars # --------------------------------------------------------------------------- # Constants # --------------------------------------------------------------------------- def test_default_min_bars() -> None: assert DEFAULT_MIN_BARS == 30 # --------------------------------------------------------------------------- # Insufficient data → None (Requirement 2.6) # --------------------------------------------------------------------------- def test_returns_none_when_insufficient_bars() -> None: """Requirement 2.6: return None when fewer than min_bars.""" evaluator = CupHandleEvaluator() bars = [_bar(100.0) for _ in range(29)] assert evaluator.evaluate(bars, "D") is None def test_returns_none_with_empty_bars() -> None: evaluator = CupHandleEvaluator() assert evaluator.evaluate([], "D") is None def test_returns_none_with_one_bar() -> None: evaluator = CupHandleEvaluator() assert evaluator.evaluate([_bar(100.0)], "D") is None # --------------------------------------------------------------------------- # No pattern detected → None # --------------------------------------------------------------------------- def test_returns_none_for_flat_market() -> None: """Flat prices have no cup formation.""" evaluator = CupHandleEvaluator() bars = [_bar(100.0, high=100.0, low=100.0) for _ in range(40)] assert evaluator.evaluate(bars, "D") is None def test_returns_none_for_monotonic_uptrend() -> None: """A steady uptrend has no cup shape.""" evaluator = CupHandleEvaluator() bars = [_bar(50.0 + i * 1.0, high=51.0 + i * 1.0, low=49.0 + i * 1.0) for i in range(40)] # Cup depth would be too shallow or non-existent result = evaluator.evaluate(bars, "D") # Either None or invalid pattern — the uptrend doesn't form a cup assert result is None def test_returns_none_when_cup_too_shallow() -> None: """Cup depth < 12% should be rejected.""" evaluator = CupHandleEvaluator() # Left rim at 100, bottom at 92 → depth = 8% (too shallow) bars = _make_cup_handle_bars( n=40, left_rim=100.0, bottom=92.0, right_rim=99.0, handle_low=97.0, ) result = evaluator.evaluate(bars, "D") assert result is None def test_returns_none_when_cup_too_deep() -> None: """Cup depth > 33% should be rejected.""" evaluator = CupHandleEvaluator() # Left rim at 100, bottom at 60 → depth = 40% (too deep) bars = _make_cup_handle_bars( n=40, left_rim=100.0, bottom=60.0, right_rim=99.0, handle_low=95.0, ) result = evaluator.evaluate(bars, "D") assert result is None def test_returns_none_when_handle_too_deep() -> None: """Handle retracement > 50% of cup depth should be rejected.""" evaluator = CupHandleEvaluator() # Cup depth = 100 - 80 = 20. Handle depth > 10 (50% of 20) → rejected bars = _make_cup_handle_bars( n=40, left_rim=100.0, bottom=80.0, right_rim=99.0, handle_low=85.0, # handle depth = 99 - 85 = 14 > 10 ) result = evaluator.evaluate(bars, "D") assert result is None # --------------------------------------------------------------------------- # Valid pattern detection # --------------------------------------------------------------------------- def test_detects_valid_cup_and_handle() -> None: """Requirement 2.4: detect cup formation and handle.""" evaluator = CupHandleEvaluator() bars = _make_cup_handle_bars( n=40, left_rim=100.0, bottom=80.0, right_rim=99.0, handle_low=95.0, ) result = evaluator.evaluate(bars, "D") assert result is not None assert result.signal_type == "cup_handle" assert result.direction == SignalDirection.BULLISH def test_always_bullish_direction() -> None: """Cup & Handle is always a bullish continuation pattern.""" evaluator = CupHandleEvaluator() bars = _make_cup_handle_bars( n=40, left_rim=100.0, bottom=80.0, right_rim=98.0, handle_low=95.0, ) result = evaluator.evaluate(bars, "D") assert result is not None assert result.direction == SignalDirection.BULLISH # --------------------------------------------------------------------------- # Completeness scoring # --------------------------------------------------------------------------- def test_strength_in_unit_interval() -> None: """Strength must be in [0, 1].""" evaluator = CupHandleEvaluator() bars = _make_cup_handle_bars( n=40, left_rim=100.0, bottom=80.0, right_rim=99.0, handle_low=96.0, ) result = evaluator.evaluate(bars, "D") assert result is not None assert 0.0 <= result.strength <= 1.0 def test_confidence_in_unit_interval() -> None: """Confidence must be in [0, 1].""" evaluator = CupHandleEvaluator() bars = _make_cup_handle_bars( n=40, left_rim=100.0, bottom=80.0, right_rim=99.0, handle_low=96.0, ) result = evaluator.evaluate(bars, "D") assert result is not None assert 0.0 <= result.confidence <= 1.0 def test_confidence_proportional_to_completeness() -> None: """Requirement 2.4: confidence proportional to pattern completeness.""" evaluator = CupHandleEvaluator() bars = _make_cup_handle_bars( n=40, left_rim=100.0, bottom=80.0, right_rim=99.0, handle_low=96.0, ) result = evaluator.evaluate(bars, "D") assert result is not None # confidence = completeness * 0.90 expected_confidence = result.strength * 0.90 assert abs(result.confidence - expected_confidence) < 1e-9 def test_better_symmetry_yields_higher_completeness() -> None: """More symmetric rims should produce higher completeness.""" evaluator = CupHandleEvaluator() # Good symmetry: right rim very close to left rim bars_good = _make_cup_handle_bars( n=40, left_rim=100.0, bottom=80.0, right_rim=100.0, handle_low=96.0, ) result_good = evaluator.evaluate(bars_good, "D") # Worse symmetry: right rim further from left rim bars_worse = _make_cup_handle_bars( n=40, left_rim=100.0, bottom=80.0, right_rim=88.0, handle_low=85.0, ) result_worse = evaluator.evaluate(bars_worse, "D") if result_good is not None and result_worse is not None: assert result_good.metadata["symmetry_score"] >= result_worse.metadata["symmetry_score"] # --------------------------------------------------------------------------- # Metadata (Requirement 2.7) # --------------------------------------------------------------------------- def test_metadata_contains_required_fields() -> None: """Metadata should include left_rim, right_rim, bottom, handle_depth, completeness.""" evaluator = CupHandleEvaluator() bars = _make_cup_handle_bars( n=40, left_rim=100.0, bottom=80.0, right_rim=99.0, handle_low=96.0, ) result = evaluator.evaluate(bars, "D") assert result is not None meta = result.metadata assert "left_rim" in meta assert "right_rim" in meta assert "bottom" in meta assert "handle_depth" in meta assert "completeness" in meta assert "cup_depth_pct" in meta assert "symmetry_score" in meta assert "handle_score" in meta # --------------------------------------------------------------------------- # Signal result structure (Requirement 2.7) # --------------------------------------------------------------------------- def test_signal_result_structure() -> None: """Requirement 2.7: SignalResult has all required fields.""" evaluator = CupHandleEvaluator() bars = _make_cup_handle_bars( n=40, left_rim=100.0, bottom=80.0, right_rim=99.0, handle_low=96.0, ) result = evaluator.evaluate(bars, "D") assert result is not None assert result.signal_type == "cup_handle" assert result.timeframe == "D" assert 0.0 <= result.strength <= 1.0 assert 0.0 <= result.confidence <= 1.0 assert result.direction == SignalDirection.BULLISH # --------------------------------------------------------------------------- # Timeframe passthrough # --------------------------------------------------------------------------- def test_timeframe_passthrough() -> None: """The timeframe label is passed through to the result.""" evaluator = CupHandleEvaluator() bars = _make_cup_handle_bars( n=40, left_rim=100.0, bottom=80.0, right_rim=99.0, handle_low=96.0, ) for tf in ("M30", "H1", "H4", "D", "W", "M"): result = evaluator.evaluate(bars, tf) assert result is not None assert result.timeframe == tf # --------------------------------------------------------------------------- # Custom min_bars # --------------------------------------------------------------------------- def test_custom_min_bars() -> None: """CupHandleEvaluator with a custom min_bars should use that value.""" evaluator = CupHandleEvaluator(min_bars=50) assert evaluator.min_bars == 50 # 40 bars should be insufficient bars = _make_cup_handle_bars(n=40) assert evaluator.evaluate(bars, "D") is None def test_exactly_min_bars_works() -> None: """Exactly min_bars should be sufficient if pattern is present.""" evaluator = CupHandleEvaluator(min_bars=30) bars = _make_cup_handle_bars( n=30, left_rim=100.0, bottom=80.0, right_rim=99.0, handle_low=96.0, ) result = evaluator.evaluate(bars, "D") # Should produce a result if the pattern is valid # (may be None if the synthetic data doesn't form a clean pattern at 30 bars) # At minimum, it should not crash assert result is None or result.signal_type == "cup_handle"