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 selection is
residual cross-sectional momentum: a setup's ticker must rank in the top
``min_momentum_percentile`` of the universe by beta-adjusted 12-1 month momentum.
R:R and confidence remain as floors, and conviction/conflict survive as optional
tighteners (off by default). Qualified setups must also have a primary target
with at least ``MIN_TARGET_PROBABILITY`` reach probability: a primary below the
floor is a lottery target whose distance inflates R:R, so it would otherwise
game the min_rr gate (the model clamps probabilities at 3%, and far targets pin
there while their live R:R stays high forever).
"""

from __future__ import annotations

from typing import Any

HIGH_CONVICTION_ACTIONS = {"LONG_HIGH", "SHORT_HIGH"}

# Floor for the primary target's reach probability, shared with the primary
# target selection in recommendation_service and mirrored in the frontend
# (qualification.ts). Under the two-barrier model a fair-race 1.5:1 target sits
# near ~34% before drift adjustments, so 20% only excludes targets the model
# itself considers long shots.
MIN_TARGET_PROBABILITY = 20.0


def _action_direction(action: str | None) -> str:
    if not action or action == "NEUTRAL":
        return "neutral"
    if action.startswith("LONG"):
        return "long"
    if action.startswith("SHORT"):
        return "short"
    return "neutral"


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 primary/headline target, falling back to best target."""
    targets = getattr(setup, "targets", None) or []
    for target in targets:
        if not isinstance(target, dict) or not target.get("is_primary"):
            continue
        probability = target.get("probability")
        return float(probability) if probability is not None else None
    best = best_target_probability(setup)
    return best if best > 0 else None


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), target probability, confidence
    floor, momentum percentile (the core selection), then optional conviction /
    conflict tighteners. ``min_momentum_percentile`` defaults to 0 (off) for
    callers that pass a legacy config without the key.
    """
    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
    target_probability = primary_target_probability(setup)
    if target_probability is None or target_probability < MIN_TARGET_PROBABILITY:
        return False
    if (setup.confidence_score or 0.0) < config["min_confidence"]:
        return False
    # Residual cross-sectional momentum: the core selection. A setup's ticker
    # must rank in the top ``min_momentum_percentile`` of the universe by
    # beta-adjusted 12-1 momentum. The validated edge is long-only, so while the
    # gate is active shorts (which fight the trend) never qualify. Missing ranks
    # do not qualify because the production edge depends on this cross-sectional
    # selection.
    min_pct = float(config.get("min_momentum_percentile", 0.0))
    if min_pct > 0:
        if (getattr(setup, "direction", "long") or "long") == "short":
            return False
        momentum_percentile = getattr(setup, "momentum_percentile", None)
        if momentum_percentile is None or momentum_percentile < min_pct:
            return False
    # A setup is actionable only when the live ticker action points in the same
    # direction. NEUTRAL means no clear signal; an opposite action means the
    # setup is counter-bias. ``exclude_neutral`` defaults on; callers that omit
    # it keep legacy floor-only behavior.
    if config.get("exclude_neutral"):
        action_direction = _action_direction(getattr(setup, "recommended_action", None))
        setup_direction = (getattr(setup, "direction", "long") or "long").lower()
        if action_direction == "neutral" or action_direction != setup_direction:
            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
    return True