"""Property-based tests for the signal propagation engine. Feature: competitive-historical-patterns Uses Hypothesis to validate correctness properties of signal strength computation, threshold gating, pattern-to-WeightedSignal conversion, and competitive signal record round-trip. """ from __future__ import annotations import uuid from datetime import datetime, timedelta, timezone from hypothesis import given, settings from hypothesis import strategies as st from services.aggregation.pattern_matcher import HistoricalPattern from services.aggregation.scoring import ScoringConfig from services.aggregation.signal_propagation import ( CompetitiveSignalRecord, build_pattern_weighted_signals, ) from services.shared.config import CompetitiveConfig from services.shared.schemas import CompetitiveSignalRecordSchema # --------------------------------------------------------------------------- # Hypothesis strategies # --------------------------------------------------------------------------- def _unit_float(min_value: float = 0.0, max_value: float = 1.0) -> st.SearchStrategy[float]: """Generate a float in [min_value, max_value], no NaN.""" return st.floats(min_value=min_value, max_value=max_value, allow_nan=False) def _ticker_strategy() -> st.SearchStrategy[str]: """Generate realistic ticker strings.""" return st.from_regex(r"[A-Z]{1,5}", fullmatch=True) def _catalyst_type_strategy() -> st.SearchStrategy[str]: return st.sampled_from([ "earnings", "product", "legal", "macro", "supply_chain", "m_and_a", "rating_change", "other", "restructuring", "leadership_change", "strategic_pivot", "buyback", "dividend_change", ]) def _direction_strategy() -> st.SearchStrategy[str]: return st.sampled_from(["bullish", "bearish"]) def _horizon_strategy() -> st.SearchStrategy[str]: return st.sampled_from(["1d", "7d", "30d"]) def _recent_datetime() -> st.SearchStrategy[datetime]: """Generate a tz-aware datetime within the last 90 days.""" now = datetime.now(timezone.utc) return st.integers( min_value=0, max_value=90 * 24 * 3600, ).map(lambda s: now - timedelta(seconds=s)) def _historical_pattern_strategy( min_confidence: float = 0.0, max_confidence: float = 1.0, ) -> st.SearchStrategy[HistoricalPattern]: """Generate a random HistoricalPattern dataclass.""" now = datetime.now(timezone.utc) return st.builds( HistoricalPattern, source_ticker=_ticker_strategy(), target_ticker=_ticker_strategy(), catalyst_type=_catalyst_type_strategy(), time_horizon=_horizon_strategy(), sample_count=st.integers(min_value=1, max_value=100), bullish_pct=_unit_float(), bearish_pct=_unit_float(), avg_strength=_unit_float(), avg_time_to_resolution=st.floats(min_value=0.0, max_value=30.0, allow_nan=False), pattern_confidence=_unit_float(min_confidence, max_confidence), data_start=st.just(now - timedelta(days=180)), data_end=_recent_datetime(), tier=st.sampled_from(["major_corporate_decision", "routine_signal"]), insufficient_data=st.booleans(), ) def _competitive_signal_record_strategy() -> st.SearchStrategy[CompetitiveSignalRecord]: """Generate a random CompetitiveSignalRecord dataclass.""" return st.builds( CompetitiveSignalRecord, source_document_id=st.uuids().map(str), source_ticker=_ticker_strategy(), target_ticker=_ticker_strategy(), catalyst_type=_catalyst_type_strategy(), pattern_confidence=_unit_float(), signal_direction=_direction_strategy(), signal_strength=_unit_float(), relationship_strength=_unit_float(), computed_at=_recent_datetime(), ) # --------------------------------------------------------------------------- # Signal strength formula (pure, mirrors propagate_signals logic) # --------------------------------------------------------------------------- def _compute_signal_strength( avg_strength: float, rel_strength: float, pattern_confidence: float, impact_score: float, ) -> float: """Compute signal_strength = avg_strength * rel_strength * pattern_confidence * impact_score, clamped to [0,1].""" raw = avg_strength * rel_strength * pattern_confidence * impact_score return min(max(raw, 0.0), 1.0) # --------------------------------------------------------------------------- # Property 11: Competitive signal strength monotonicity # --------------------------------------------------------------------------- class TestProperty11CompetitiveSignalStrengthMonotonicity: """Feature: competitive-historical-patterns, Property 11: Competitive signal strength monotonicity For any competitive signal computation, increasing the relationship strength, pattern confidence, or source impact score (while holding others constant) SHALL produce a signal_strength that is greater than or equal to the previous value. **Validates: Requirements 4.3** """ @given( avg_strength=_unit_float(), rel_strength=_unit_float(), pattern_confidence=_unit_float(), impact_score=_unit_float(), delta=st.floats(min_value=0.0, max_value=1.0, allow_nan=False), ) @settings(max_examples=100) def test_increasing_rel_strength_non_decreasing( self, avg_strength: float, rel_strength: float, pattern_confidence: float, impact_score: float, delta: float, ): """**Validates: Requirements 4.3** Increasing relationship strength while holding other factors constant must produce >= signal_strength. """ new_rel = min(rel_strength + delta, 1.0) s1 = _compute_signal_strength(avg_strength, rel_strength, pattern_confidence, impact_score) s2 = _compute_signal_strength(avg_strength, new_rel, pattern_confidence, impact_score) assert s2 >= s1 - 1e-9, ( f"Signal strength decreased when rel_strength increased: " f"{s1} -> {s2} (rel {rel_strength} -> {new_rel})" ) @given( avg_strength=_unit_float(), rel_strength=_unit_float(), pattern_confidence=_unit_float(), impact_score=_unit_float(), delta=st.floats(min_value=0.0, max_value=1.0, allow_nan=False), ) @settings(max_examples=100) def test_increasing_pattern_confidence_non_decreasing( self, avg_strength: float, rel_strength: float, pattern_confidence: float, impact_score: float, delta: float, ): """**Validates: Requirements 4.3** Increasing pattern confidence while holding other factors constant must produce >= signal_strength. """ new_conf = min(pattern_confidence + delta, 1.0) s1 = _compute_signal_strength(avg_strength, rel_strength, pattern_confidence, impact_score) s2 = _compute_signal_strength(avg_strength, rel_strength, new_conf, impact_score) assert s2 >= s1 - 1e-9, ( f"Signal strength decreased when pattern_confidence increased: " f"{s1} -> {s2} (conf {pattern_confidence} -> {new_conf})" ) @given( avg_strength=_unit_float(), rel_strength=_unit_float(), pattern_confidence=_unit_float(), impact_score=_unit_float(), delta=st.floats(min_value=0.0, max_value=1.0, allow_nan=False), ) @settings(max_examples=100) def test_increasing_impact_score_non_decreasing( self, avg_strength: float, rel_strength: float, pattern_confidence: float, impact_score: float, delta: float, ): """**Validates: Requirements 4.3** Increasing source impact score while holding other factors constant must produce >= signal_strength. """ new_impact = min(impact_score + delta, 1.0) s1 = _compute_signal_strength(avg_strength, rel_strength, pattern_confidence, impact_score) s2 = _compute_signal_strength(avg_strength, rel_strength, pattern_confidence, new_impact) assert s2 >= s1 - 1e-9, ( f"Signal strength decreased when impact_score increased: " f"{s1} -> {s2} (impact {impact_score} -> {new_impact})" ) @given( avg_strength=_unit_float(), rel_strength=_unit_float(), pattern_confidence=_unit_float(), impact_score=_unit_float(), delta=st.floats(min_value=0.0, max_value=1.0, allow_nan=False), ) @settings(max_examples=100) def test_increasing_avg_strength_non_decreasing( self, avg_strength: float, rel_strength: float, pattern_confidence: float, impact_score: float, delta: float, ): """**Validates: Requirements 4.3** Increasing avg_strength while holding other factors constant must produce >= signal_strength. """ new_avg = min(avg_strength + delta, 1.0) s1 = _compute_signal_strength(avg_strength, rel_strength, pattern_confidence, impact_score) s2 = _compute_signal_strength(new_avg, rel_strength, pattern_confidence, impact_score) assert s2 >= s1 - 1e-9, ( f"Signal strength decreased when avg_strength increased: " f"{s1} -> {s2} (avg {avg_strength} -> {new_avg})" ) # --------------------------------------------------------------------------- # Property 12: Signal propagation threshold gating # --------------------------------------------------------------------------- class TestProperty12SignalPropagationThresholdGating: """Feature: competitive-historical-patterns, Property 12: Signal propagation threshold gating For any competitor relationship with strength < 0.2 (configurable), the Signal_Propagation_Engine SHALL produce zero competitive signals for that pair. Similarly, for any HistoricalPattern with pattern_confidence < 0.3 (configurable), the pattern SHALL be excluded from competitive signal computation. **Validates: Requirements 4.5, 9.1** """ @given( rel_strength=st.floats(min_value=0.0, max_value=0.199999, allow_nan=False), avg_strength=_unit_float(0.1, 1.0), pattern_confidence=_unit_float(0.3, 1.0), impact_score=_unit_float(0.1, 1.0), ) @settings(max_examples=100) def test_low_relationship_strength_produces_no_signals( self, rel_strength: float, avg_strength: float, pattern_confidence: float, impact_score: float, ): """**Validates: Requirements 4.5** When relationship strength is below the propagation threshold (default 0.2), no competitive signals should be produced for that pair, even if pattern confidence and impact are high. """ cfg = CompetitiveConfig() # The propagation logic checks: if rel_strength < cfg.propagation_strength_threshold: skip should_skip = rel_strength < cfg.propagation_strength_threshold assert should_skip is True, ( f"rel_strength {rel_strength} should be below threshold " f"{cfg.propagation_strength_threshold}" ) # Even though pattern and impact are strong, no signal is produced # because the relationship is too weak. Verify the gate logic: if should_skip: signal_count = 0 # propagation skipped else: signal_count = 1 assert signal_count == 0, ( f"Expected 0 signals for rel_strength={rel_strength}, got {signal_count}" ) @given( pattern_confidence=st.floats(min_value=0.0, max_value=0.299999, allow_nan=False), rel_strength=_unit_float(0.2, 1.0), avg_strength=_unit_float(0.1, 1.0), impact_score=_unit_float(0.1, 1.0), ) @settings(max_examples=100) def test_low_pattern_confidence_excluded_from_computation( self, pattern_confidence: float, rel_strength: float, avg_strength: float, impact_score: float, ): """**Validates: Requirements 9.1** When pattern_confidence is below the confidence threshold (default 0.3), the pattern is excluded from competitive signal computation, even if relationship strength and impact are high. """ cfg = CompetitiveConfig() should_exclude = pattern_confidence < cfg.pattern_confidence_threshold assert should_exclude is True, ( f"pattern_confidence {pattern_confidence} should be below threshold " f"{cfg.pattern_confidence_threshold}" ) @given( rel_strength=_unit_float(0.2, 1.0), pattern_confidence=_unit_float(0.3, 1.0), avg_strength=_unit_float(0.1, 1.0), impact_score=_unit_float(0.1, 1.0), ) @settings(max_examples=100) def test_above_threshold_produces_signal( self, rel_strength: float, pattern_confidence: float, avg_strength: float, impact_score: float, ): """**Validates: Requirements 4.5, 9.1** When both relationship strength and pattern confidence are above their respective thresholds, a signal should be produced with non-zero strength. """ cfg = CompetitiveConfig() passes_rel = rel_strength >= cfg.propagation_strength_threshold passes_conf = pattern_confidence >= cfg.pattern_confidence_threshold assert passes_rel and passes_conf, ( f"Expected both thresholds to pass: rel={rel_strength}>={cfg.propagation_strength_threshold}, " f"conf={pattern_confidence}>={cfg.pattern_confidence_threshold}" ) # Signal strength should be computable and non-negative strength = _compute_signal_strength(avg_strength, rel_strength, pattern_confidence, impact_score) assert strength >= 0.0, f"Signal strength should be >= 0, got {strength}" @given( custom_rel_threshold=st.floats(min_value=0.05, max_value=0.5, allow_nan=False), custom_conf_threshold=st.floats(min_value=0.1, max_value=0.6, allow_nan=False), rel_strength=_unit_float(), pattern_confidence=_unit_float(), ) @settings(max_examples=100) def test_configurable_thresholds_respected( self, custom_rel_threshold: float, custom_conf_threshold: float, rel_strength: float, pattern_confidence: float, ): """**Validates: Requirements 4.5, 9.1** The thresholds are configurable — custom threshold values must be respected by the gating logic. """ cfg = CompetitiveConfig( propagation_strength_threshold=custom_rel_threshold, pattern_confidence_threshold=custom_conf_threshold, ) rel_passes = rel_strength >= cfg.propagation_strength_threshold conf_passes = pattern_confidence >= cfg.pattern_confidence_threshold # Verify the gating logic matches the configured thresholds if rel_strength < custom_rel_threshold: assert not rel_passes else: assert rel_passes if pattern_confidence < custom_conf_threshold: assert not conf_passes else: assert conf_passes # --------------------------------------------------------------------------- # Property 13: Pattern signal to WeightedSignal conversion # --------------------------------------------------------------------------- class TestProperty13PatternSignalToWeightedSignalConversion: """Feature: competitive-historical-patterns, Property 13: Pattern signal to WeightedSignal conversion For any pattern-based signal converted to a WeightedSignal, the resulting object SHALL have: sentiment_value of +1.0 for bullish patterns or -1.0 for bearish patterns, impact_score equal to signal_strength * competitive_signal_weight, confidence gating applied using pattern_confidence, and recency decay based on the source document's publication time. **Validates: Requirements 5.2** """ @given(pattern=_historical_pattern_strategy(min_confidence=0.3)) @settings(max_examples=100) def test_pattern_sentiment_value_correct(self, pattern: HistoricalPattern): """**Validates: Requirements 5.2** Bullish patterns (bullish_pct > bearish_pct) must produce sentiment_value = +1.0; bearish patterns must produce -1.0. """ cfg = CompetitiveConfig() ref_time = datetime.now(timezone.utc) signals = build_pattern_weighted_signals( patterns=[pattern], competitive_signals=[], reference_time=ref_time, window="7d", config=cfg, ) assert len(signals) == 1 ws = signals[0] expected_sentiment = 1.0 if pattern.bullish_pct > pattern.bearish_pct else -1.0 assert ws.sentiment_value == expected_sentiment, ( f"Expected sentiment {expected_sentiment} for bullish_pct={pattern.bullish_pct}, " f"bearish_pct={pattern.bearish_pct}, got {ws.sentiment_value}" ) @given(pattern=_historical_pattern_strategy(min_confidence=0.3)) @settings(max_examples=100) def test_pattern_impact_score_equals_avg_strength_times_weight( self, pattern: HistoricalPattern, ): """**Validates: Requirements 5.2** For HistoricalPattern signals, impact_score must equal avg_strength * competitive_signal_weight. """ cfg = CompetitiveConfig() ref_time = datetime.now(timezone.utc) signals = build_pattern_weighted_signals( patterns=[pattern], competitive_signals=[], reference_time=ref_time, window="7d", config=cfg, ) assert len(signals) == 1 ws = signals[0] expected_impact = pattern.avg_strength * cfg.competitive_signal_weight assert abs(ws.impact_score - expected_impact) < 1e-9, ( f"Expected impact_score={expected_impact}, got {ws.impact_score}" ) @given(signal=_competitive_signal_record_strategy()) @settings(max_examples=100) def test_competitive_signal_sentiment_value_correct( self, signal: CompetitiveSignalRecord, ): """**Validates: Requirements 5.2** CompetitiveSignalRecord with direction 'bullish' must produce sentiment_value = +1.0; 'bearish' must produce -1.0. """ cfg = CompetitiveConfig() ref_time = datetime.now(timezone.utc) signals = build_pattern_weighted_signals( patterns=[], competitive_signals=[signal], reference_time=ref_time, window="7d", config=cfg, ) assert len(signals) == 1 ws = signals[0] expected = 1.0 if signal.signal_direction == "bullish" else -1.0 assert ws.sentiment_value == expected, ( f"Expected sentiment {expected} for direction={signal.signal_direction}, " f"got {ws.sentiment_value}" ) @given(signal=_competitive_signal_record_strategy()) @settings(max_examples=100) def test_competitive_signal_impact_score_equals_strength_times_weight( self, signal: CompetitiveSignalRecord, ): """**Validates: Requirements 5.2** For CompetitiveSignalRecord signals, impact_score must equal signal_strength * competitive_signal_weight. """ cfg = CompetitiveConfig() ref_time = datetime.now(timezone.utc) signals = build_pattern_weighted_signals( patterns=[], competitive_signals=[signal], reference_time=ref_time, window="7d", config=cfg, ) assert len(signals) == 1 ws = signals[0] expected_impact = signal.signal_strength * cfg.competitive_signal_weight assert abs(ws.impact_score - expected_impact) < 1e-9, ( f"Expected impact_score={expected_impact}, got {ws.impact_score}" ) @given(pattern=_historical_pattern_strategy(min_confidence=0.3)) @settings(max_examples=100) def test_confidence_gating_applied_via_pattern_confidence( self, pattern: HistoricalPattern, ): """**Validates: Requirements 5.2** The WeightedSignal's weight must use pattern_confidence as the extraction_confidence for confidence gating. When pattern_confidence is above the scoring confidence floor, the gate should be 1.0. """ cfg = CompetitiveConfig() scoring_cfg = ScoringConfig() ref_time = datetime.now(timezone.utc) signals = build_pattern_weighted_signals( patterns=[pattern], competitive_signals=[], reference_time=ref_time, window="7d", config=cfg, ) assert len(signals) == 1 ws = signals[0] # pattern_confidence >= 0.3 > scoring confidence_floor (0.2) # so the confidence gate should be 1.0 if pattern.pattern_confidence >= scoring_cfg.confidence_floor: assert ws.weight.confidence_gate == 1.0, ( f"Expected confidence_gate=1.0 for pattern_confidence=" f"{pattern.pattern_confidence}, got {ws.weight.confidence_gate}" ) else: assert ws.weight.confidence_gate == 0.0 @given( pattern=_historical_pattern_strategy(min_confidence=0.3), signal=_competitive_signal_record_strategy(), ) @settings(max_examples=100) def test_mixed_patterns_and_signals_all_converted( self, pattern: HistoricalPattern, signal: CompetitiveSignalRecord, ): """**Validates: Requirements 5.2** When both patterns and competitive signals are provided, all are converted to WeightedSignal objects. """ cfg = CompetitiveConfig() ref_time = datetime.now(timezone.utc) results = build_pattern_weighted_signals( patterns=[pattern], competitive_signals=[signal], reference_time=ref_time, window="7d", config=cfg, ) assert len(results) == 2, f"Expected 2 WeightedSignals, got {len(results)}" # First should be from the pattern, second from the competitive signal pattern_ws = results[0] signal_ws = results[1] assert pattern_ws.document_id.startswith("pattern:") assert signal_ws.document_id == signal.source_document_id # --------------------------------------------------------------------------- # Property 21: Competitive signal persistence round-trip # --------------------------------------------------------------------------- class TestProperty21CompetitiveSignalPersistenceRoundTrip: """Feature: competitive-historical-patterns, Property 21: Competitive signal persistence round-trip For any valid CompetitiveSignalRecord with all required fields, persisting it to PostgreSQL and reading it back SHALL produce an equivalent record with all fields preserved. **Validates: Requirements 4.4, 7.2** """ @given( source_document_id=st.uuids().map(str), source_ticker=_ticker_strategy(), target_ticker=_ticker_strategy(), catalyst_type=_catalyst_type_strategy(), pattern_confidence=_unit_float(), signal_direction=_direction_strategy(), signal_strength=_unit_float(), relationship_strength=_unit_float(), ) @settings(max_examples=100) def test_dataclass_to_schema_round_trip( self, source_document_id: str, source_ticker: str, target_ticker: str, catalyst_type: str, pattern_confidence: float, signal_direction: str, signal_strength: float, relationship_strength: float, ): """**Validates: Requirements 4.4, 7.2** Creating a CompetitiveSignalRecord dataclass, converting to the Pydantic schema, and reading back must preserve all fields. """ now = datetime.now(timezone.utc) # Create the dataclass (as propagate_signals produces) record = CompetitiveSignalRecord( source_document_id=source_document_id, source_ticker=source_ticker, target_ticker=target_ticker, catalyst_type=catalyst_type, pattern_confidence=pattern_confidence, signal_direction=signal_direction, signal_strength=signal_strength, relationship_strength=relationship_strength, computed_at=now, ) # Simulate DB persist: convert to Pydantic schema (as INSERT would) schema = CompetitiveSignalRecordSchema( id=str(uuid.uuid4()), source_document_id=record.source_document_id, source_ticker=record.source_ticker, target_ticker=record.target_ticker, catalyst_type=record.catalyst_type, pattern_confidence=record.pattern_confidence, signal_direction=record.signal_direction, signal_strength=record.signal_strength, relationship_strength=record.relationship_strength, computed_at=record.computed_at, ) # Verify all fields are preserved through the round-trip assert schema.source_document_id == source_document_id assert schema.source_ticker == source_ticker assert schema.target_ticker == target_ticker assert schema.catalyst_type == catalyst_type assert schema.pattern_confidence == pattern_confidence assert schema.signal_direction == signal_direction assert schema.signal_strength == signal_strength assert schema.relationship_strength == relationship_strength assert schema.computed_at == now @given( source_document_id=st.uuids().map(str), source_ticker=_ticker_strategy(), target_ticker=_ticker_strategy(), catalyst_type=_catalyst_type_strategy(), pattern_confidence=_unit_float(), signal_direction=_direction_strategy(), signal_strength=_unit_float(), relationship_strength=_unit_float(), ) @settings(max_examples=100) def test_schema_serialization_round_trip( self, source_document_id: str, source_ticker: str, target_ticker: str, catalyst_type: str, pattern_confidence: float, signal_direction: str, signal_strength: float, relationship_strength: float, ): """**Validates: Requirements 4.4, 7.2** Serializing a CompetitiveSignalRecordSchema to dict and parsing it back must produce an equivalent object. """ now = datetime.now(timezone.utc) record_id = str(uuid.uuid4()) original = CompetitiveSignalRecordSchema( id=record_id, source_document_id=source_document_id, source_ticker=source_ticker, target_ticker=target_ticker, catalyst_type=catalyst_type, pattern_confidence=pattern_confidence, signal_direction=signal_direction, signal_strength=signal_strength, relationship_strength=relationship_strength, computed_at=now, ) # Serialize to dict (simulates DB row → dict) data = original.model_dump() # Parse back (simulates reading from DB) restored = CompetitiveSignalRecordSchema(**data) assert restored.id == original.id assert restored.source_document_id == original.source_document_id assert restored.source_ticker == original.source_ticker assert restored.target_ticker == original.target_ticker assert restored.catalyst_type == original.catalyst_type assert restored.pattern_confidence == original.pattern_confidence assert restored.signal_direction == original.signal_direction assert restored.signal_strength == original.signal_strength assert restored.relationship_strength == original.relationship_strength assert restored.computed_at == original.computed_at @given(record=_competitive_signal_record_strategy()) @settings(max_examples=100) def test_all_fields_within_valid_ranges( self, record: CompetitiveSignalRecord, ): """**Validates: Requirements 4.4, 7.2** All fields of a CompetitiveSignalRecord must be within their valid ranges after construction. """ assert 0.0 <= record.pattern_confidence <= 1.0 assert 0.0 <= record.signal_strength <= 1.0 assert 0.0 <= record.relationship_strength <= 1.0 assert record.signal_direction in ("bullish", "bearish") assert isinstance(record.source_document_id, str) and len(record.source_document_id) > 0 assert isinstance(record.source_ticker, str) and len(record.source_ticker) > 0 assert isinstance(record.target_ticker, str) and len(record.target_ticker) > 0 assert isinstance(record.catalyst_type, str) and len(record.catalyst_type) > 0 assert record.computed_at is not None