diff --git a/app/services/recommendation_service.py b/app/services/recommendation_service.py index 4b9b95b..b171e48 100644 --- a/app/services/recommendation_service.py +++ b/app/services/recommendation_service.py @@ -2,6 +2,7 @@ from __future__ import annotations import json import logging +import math from typing import Any from sqlalchemy import select @@ -27,10 +28,30 @@ DEFAULT_RECOMMENDATION_CONFIG: dict[str, float] = { } +# Horizon (trading days) over which a target's reach-probability is estimated. +# Kept in step with the outcome evaluator's window so probability predicts the +# metric we actually measure. +_TARGET_HORIZON_DAYS = 30.0 + + def _clamp(value: float, low: float, high: float) -> float: return max(low, min(high, value)) +def _norm_cdf(x: float) -> float: + """Standard normal CDF via erf (no SciPy dependency).""" + return 0.5 * (1.0 + math.erf(x / math.sqrt(2.0))) + + +def _classify_by_probability(probability: float) -> str: + """Label a target by how likely it is to be reached (derived, not assumed).""" + if probability >= 60.0: + return "Conservative" + if probability >= 40.0: + return "Moderate" + return "Aggressive" + + def _sentiment_value(sentiment_classification: str | None) -> str | None: if sentiment_classification is None: return None @@ -238,20 +259,9 @@ class TargetGenerator: selected = candidates[:5] selected.sort(key=lambda row: row["distance_from_entry"]) - if not selected: - return [] - - n = len(selected) - for idx, target in enumerate(selected): - if n <= 2: - target["classification"] = "Conservative" if idx == 0 else "Aggressive" - elif idx <= 1: - target["classification"] = "Conservative" - elif idx >= n - 2: - target["classification"] = "Aggressive" - else: - target["classification"] = "Moderate" - + # Classification is assigned later from the computed reach-probability + # (see _classify_by_probability), not from distance rank. + for target in selected: target.pop("quality", None) return selected @@ -266,67 +276,47 @@ class ProbabilityEstimator: direction: str, config: dict[str, float], ) -> float: - classification = str(target.get("classification", "Moderate")) + """Probability the target is reached within the outcome horizon. + + Base = probability of price *touching* a level at the target's distance + within the evaluation window, under a driftless random walk (reflection + principle): 2·(1 − Φ(d / (ATR·√T))). Distance is in ATR multiples and T + is the horizon in trading days, so a far target is inherently unlikely — + no more 90% on a +39% move. Strength and signal alignment (drift toward + the target) then modulate it modestly. + """ strength = float(target.get("sr_strength", 50.0)) atr_multiple = float(target.get("distance_atr_multiple", 1.0)) - if classification == "Conservative": - base_prob = 70.0 - elif classification == "Aggressive": - base_prob = 40.0 - else: - base_prob = 55.0 - - if strength >= 80: - strength_adj = 15.0 - elif strength >= 60: - strength_adj = 10.0 - elif strength >= 40: - strength_adj = 5.0 - else: - strength_adj = -10.0 + expected_move_atr = math.sqrt(_TARGET_HORIZON_DAYS) # ≈ 5.48 ATR over 30d + z = atr_multiple / expected_move_atr if expected_move_atr > 0 else 99.0 + touch_prob = 2.0 * (1.0 - _norm_cdf(z)) # 0..1 + probability = touch_prob * 100.0 technical = float(dimension_scores.get("technical", 50.0)) momentum = float(dimension_scores.get("momentum", 50.0)) sentiment = _sentiment_value(sentiment_classification) - alignment_adj = 0.0 + # Drift toward the target raises touch probability; against it lowers. if direction == "long": - if technical > 60 and (sentiment == "bullish" or momentum > 60): - alignment_adj = 15.0 - elif technical < 40 or (sentiment == "bearish" and momentum < 40): - alignment_adj = -15.0 + aligned = technical > 60 and (sentiment == "bullish" or momentum > 60) + opposed = technical < 40 or (sentiment == "bearish" and momentum < 40) else: - if technical < 40 and (sentiment == "bearish" or momentum < 40): - alignment_adj = 15.0 - elif technical > 60 or (sentiment == "bullish" and momentum > 60): - alignment_adj = -15.0 + aligned = technical < 40 and (sentiment == "bearish" or momentum < 40) + opposed = technical > 60 or (sentiment == "bullish" and momentum > 60) - volatility_adj = 0.0 - if atr_multiple > 5: - volatility_adj = 5.0 - elif atr_multiple < 2: - volatility_adj = 5.0 - - signal_weight = float(config.get("recommendation_signal_alignment_weight", 0.15)) sr_weight = float(config.get("recommendation_sr_strength_weight", 0.20)) - distance_penalty = float(config.get("recommendation_distance_penalty_factor", 0.10)) + signal_weight = float(config.get("recommendation_signal_alignment_weight", 0.15)) - scaled_alignment_adj = alignment_adj * (signal_weight / 0.15) - scaled_strength_adj = strength_adj * (sr_weight / 0.20) - distance_adj = -distance_penalty * max(atr_multiple - 1.0, 0.0) * 2.0 + # Strength magnet: ±(sr_weight·50) at the extremes (±10 by default) + probability += ((strength - 50.0) / 50.0) * (sr_weight * 50.0) + # Alignment: ±(signal_weight·100) (±15 by default) + if aligned: + probability += signal_weight * 100.0 + elif opposed: + probability -= signal_weight * 100.0 - probability = base_prob + scaled_strength_adj + scaled_alignment_adj + volatility_adj + distance_adj - probability = _clamp(probability, 10.0, 90.0) - - if classification == "Conservative": - probability = max(probability, 61.0) - elif classification == "Moderate": - probability = _clamp(probability, 40.0, 70.0) - elif classification == "Aggressive": - probability = min(probability, 49.0) - - return round(probability, 2) + return round(_clamp(probability, 3.0, 95.0), 2) signal_conflict_detector = SignalConflictDetector() @@ -516,6 +506,8 @@ async def enhance_trade_setup( direction=direction, config=config, ) + # Label follows from the reach-probability: high prob = Conservative. + target["classification"] = _classify_by_probability(target["probability"]) # Primary target = most-likely target with real asymmetry (see # _select_primary_target), not the old quality-score pick that ignored diff --git a/tests/unit/test_recommendation_service.py b/tests/unit/test_recommendation_service.py index 71033bb..6b5ebf9 100644 --- a/tests/unit/test_recommendation_service.py +++ b/tests/unit/test_recommendation_service.py @@ -159,48 +159,44 @@ def test_generate_targets_respects_direction_and_order(): assert distances == sorted(distances) -def test_probability_ranges_by_classification(): +def test_probability_decreases_with_distance(): + """A far target must be far less likely than a near one — no 90% at +39%.""" config = { "recommendation_signal_alignment_weight": 0.15, "recommendation_sr_strength_weight": 0.20, - "recommendation_distance_penalty_factor": 0.10, } - dimension_scores = {"technical": 70.0, "momentum": 70.0} + dims = {"technical": 50.0, "momentum": 50.0} # neutral, isolate the distance term - conservative = probability_estimator.estimate_probability( - { - "classification": "Conservative", - "sr_strength": 80, - "distance_atr_multiple": 1.5, - }, - dimension_scores, - "bullish", - "long", - config, - ) - moderate = probability_estimator.estimate_probability( - { - "classification": "Moderate", - "sr_strength": 60, - "distance_atr_multiple": 3.0, - }, - dimension_scores, - "bullish", - "long", - config, - ) - aggressive = probability_estimator.estimate_probability( - { - "classification": "Aggressive", - "sr_strength": 40, - "distance_atr_multiple": 6.0, - }, - dimension_scores, - "bullish", - "long", - config, - ) + def prob(atr_multiple: float, strength: float = 50.0) -> float: + return probability_estimator.estimate_probability( + {"sr_strength": strength, "distance_atr_multiple": atr_multiple}, + dims, None, "long", config, + ) - assert conservative > 60 - assert 40 <= moderate <= 70 - assert aggressive < 50 + near = prob(1.5) + mid = prob(4.0) + far = prob(10.0) + + # Monotonic decay with distance + assert near > mid > far + # Near target is genuinely likely; a 10-ATR target is a long shot + assert near > 60 + assert far < 25 + + +def test_far_target_not_high_probability_even_with_strong_level(): + """The AJG case: a far target stays low-probability even at max strength.""" + config = {"recommendation_sr_strength_weight": 0.20, "recommendation_signal_alignment_weight": 0.15} + # ~10 ATR away, strongest possible level, fully aligned bullish + p = probability_estimator.estimate_probability( + {"sr_strength": 100, "distance_atr_multiple": 10.0}, + {"technical": 80.0, "momentum": 80.0}, "bullish", "long", config, + ) + assert p < 40 # nowhere near 90 + + +def test_classify_by_probability_thresholds(): + from app.services.recommendation_service import _classify_by_probability + assert _classify_by_probability(75) == "Conservative" + assert _classify_by_probability(50) == "Moderate" + assert _classify_by_probability(20) == "Aggressive"