Celes Renata
790 lines
29 KiB
Python
790 lines
29 KiB
Python
"""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 typing import Any
|
|
|
|
import pytest
|
|
from hypothesis import assume, given, settings
|
|
from hypothesis import strategies as st
|
|
|
|
from services.aggregation.pattern_matcher import HistoricalPattern
|
|
from services.aggregation.scoring import ScoringConfig, WeightedSignal
|
|
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
|