signal-platform/app/services/qualification.py

"""Shared definition of a 'qualified' (actionable) trade setup.

A single predicate, driven by the admin activation config, used by the
performance stats (server) and mirrored on the frontend. The core gate is
expected value (in R): a setup must promise positive, probability-weighted
asymmetry, not just a fat-but-improbable target or a likely-but-thin one. R:R
and confidence remain as floors, and conviction/conflict/target-probability
survive as optional tighteners (off by default).
"""

from __future__ import annotations

from typing import Any

HIGH_CONVICTION_ACTIONS = {"LONG_HIGH", "SHORT_HIGH"}


def best_target_probability(setup: Any) -> float:
    """Highest probability among a setup's targets, 0 if none."""
    targets = getattr(setup, "targets", None) or []
    probs = [float(t.get("probability", 0.0)) for t in targets if isinstance(t, dict)]
    return max(probs, default=0.0)


def primary_target_probability(setup: Any) -> float | None:
    """Probability of the starred primary target (the one the headline R:R refers
    to). Falls back to the best target's probability when none is flagged primary,
    and None when there are no targets at all (probability unknowable).
    """
    targets = getattr(setup, "targets", None) or []
    primary = next(
        (t for t in targets if isinstance(t, dict) and t.get("is_primary")), None
    )
    if primary is not None:
        return float(primary.get("probability", 0.0))
    probs = [float(t.get("probability", 0.0)) for t in targets if isinstance(t, dict)]
    return max(probs) if probs else None


def expected_value_r(setup: Any) -> float | None:
    """Expected value per unit of risk, in R: ``p·(R:R) − (1 − p)``.

    ``p`` is the primary target's hit probability. This single number captures
    "is this worth taking": it rewards both a good payoff ratio and a likely
    target, so a fat-but-improbable target can't outrank a solid, probable one —
    and a high R:R no longer fights a high probability the way the old separate
    gates did. Returns None when no target probability is known.
    """
    p = primary_target_probability(setup)
    if p is None:
        return None
    p = p / 100.0
    return p * setup.rr_ratio - (1.0 - p)


def live_risk_reward(setup: Any, current_price: float) -> float | None:
    """R:R recomputed from the CURRENT price, not the (possibly stale) entry.

    Returns None / a low value when the setup is no longer actionable: price
    already at/past the target (no reward left) or through the stop. This is how
    over-progressed setups get filtered without a separate 'max progress' knob.
    """
    if setup.direction == "long":
        reward = setup.target - current_price
        risk = current_price - setup.stop_loss
    else:
        reward = current_price - setup.target
        risk = setup.stop_loss - current_price
    if reward <= 0 or risk <= 0:
        return 0.0
    return reward / risk


def setup_qualifies(setup: Any, config: dict) -> bool:
    """Whether a setup clears the activation gate.

    ``setup`` is duck-typed: any object exposing rr_ratio, confidence_score,
    recommended_action, risk_level and a ``targets`` list of dicts.

    Gate order: R:R floor → freshness (live R:R) → confidence floor → expected
    value (the core test) → optional conviction / conflict / target-probability
    tighteners. ``min_expected_value`` defaults to -inf for callers that pass a
    legacy config without the key, so they behave exactly as before.
    """
    if setup.rr_ratio < config["min_rr"]:
        return False
    # Live R:R from the current price: drops setups whose price has already run
    # toward the target (reward consumed) or through the stop. Only applied when
    # a current price is attached (live list); skipped for historical setups.
    current_price = getattr(setup, "current_price", None)
    if current_price is not None:
        live_rr = live_risk_reward(setup, float(current_price))
        if live_rr is not None and live_rr < config["min_rr"]:
            return False
    if (setup.confidence_score or 0.0) < config["min_confidence"]:
        return False
    # Expected value (R): the core gate. Only enforced when computable — setups
    # without target probabilities (e.g. legacy historical rows) defer to the
    # R:R + confidence floors above rather than being silently dropped.
    min_ev = float(config.get("min_expected_value", float("-inf")))
    ev = expected_value_r(setup)
    if ev is not None and ev < min_ev:
        return False
    if config.get("require_high_conviction"):
        if (setup.recommended_action or "") not in HIGH_CONVICTION_ACTIONS:
            return False
    if config.get("exclude_conflicts"):
        if (setup.risk_level or "") != "Low":
            return False
    min_tp = float(config.get("min_target_probability", 0.0))
    if min_tp > 0 and best_target_probability(setup) < min_tp:
        return False
    return True