stonks-oracle/services/signal_engine/confluence.py

"""Multi-Timeframe Confluence Engine.

Evaluates signals across multiple timeframes and computes weighted confluence
scores.  Signals must trigger on at least 2 timeframes **and** include at
least one higher-timeframe anchor (D, W, or M) to pass the confluence filter.

The weighted confluence score is:

    C_confluence = Σ(w_tf · s_tf)

where ``w_tf`` is the timeframe weight and ``s_tf`` is the signal strength on
that timeframe (only summed over timeframes where the signal triggered).

Requirements: 3.1, 3.2, 3.3, 3.4, 3.5, 3.6
"""

from __future__ import annotations

import logging
from collections import Counter

from services.signal_engine.models import (
    ConfluenceSignal,
    SignalDirection,
    SignalResult,
)

logger = logging.getLogger(__name__)

# Higher-timeframe anchors — at least one must be present for a signal to pass.
HIGHER_TIMEFRAME_ANCHORS: frozenset[str] = frozenset({"D", "W", "M"})

# Minimum number of timeframes a signal must trigger on.
MIN_TIMEFRAME_COUNT: int = 2


def _dominant_direction(results: dict[str, SignalResult]) -> SignalDirection:
    """Determine the dominant direction from a set of per-timeframe results.

    Counts bullish vs bearish votes across active timeframes.  Ties resolve
    to NEUTRAL.
    """
    counts: Counter[SignalDirection] = Counter()
    for sr in results.values():
        counts[sr.direction] += 1

    bullish = counts.get(SignalDirection.BULLISH, 0)
    bearish = counts.get(SignalDirection.BEARISH, 0)

    if bullish > bearish:
        return SignalDirection.BULLISH
    if bearish > bullish:
        return SignalDirection.BEARISH
    return SignalDirection.NEUTRAL


def compute_confluence(
    signal_results: dict[str, dict[str, SignalResult]],
    weights: dict[str, float],
) -> list[ConfluenceSignal]:
    """Compute weighted confluence scores across timeframes.

    Args:
        signal_results: ``{signal_type: {timeframe: SignalResult}}``.
            Each inner dict maps timeframe labels (e.g. ``"D"``, ``"H4"``)
            to the :class:`SignalResult` produced by the signal evaluator on
            that timeframe.
        weights: ``{timeframe: weight}`` e.g.
            ``{"M30": 0.03, "H1": 0.07, "H4": 0.15, "D": 0.30, "W": 0.30, "M": 0.15}``.

    Returns:
        List of :class:`ConfluenceSignal` objects that pass **both** filters:

        1. **Minimum confluence threshold** — the signal must trigger on at
           least :data:`MIN_TIMEFRAME_COUNT` (2) timeframes.
        2. **Higher-timeframe anchor** — at least one of D, W, or M must be
           among the active timeframes.

    Requirements: 3.1, 3.2, 3.3, 3.4, 3.5, 3.6
    """
    confluence_signals: list[ConfluenceSignal] = []

    for signal_type, tf_results in signal_results.items():
        active_timeframes = list(tf_results.keys())

        # 3.3 — Minimum confluence threshold: discard if < 2 timeframes
        if len(active_timeframes) < MIN_TIMEFRAME_COUNT:
            logger.debug(
                "Signal %s discarded: only %d timeframe(s) triggered (need >= %d)",
                signal_type,
                len(active_timeframes),
                MIN_TIMEFRAME_COUNT,
            )
            continue

        # 3.4 — Higher-timeframe anchor: discard if none of D, W, M present
        if not HIGHER_TIMEFRAME_ANCHORS.intersection(active_timeframes):
            logger.debug(
                "Signal %s discarded: no higher-timeframe anchor (D/W/M) "
                "among active timeframes %s",
                signal_type,
                active_timeframes,
            )
            continue

        # 3.2 — Compute weighted confluence score
        per_timeframe: dict[str, float] = {}
        confluence_score = 0.0
        for tf, sr in tf_results.items():
            w = weights.get(tf, 0.0)
            per_timeframe[tf] = sr.strength
            confluence_score += w * sr.strength

        # Determine dominant direction across active timeframes
        direction = _dominant_direction(tf_results)

        confluence_signals.append(
            ConfluenceSignal(
                signal_type=signal_type,
                direction=direction,
                confluence_score=confluence_score,
                active_timeframes=active_timeframes,
                per_timeframe=per_timeframe,
            )
        )

        logger.debug(
            "Signal %s passed confluence: score=%.4f, direction=%s, "
            "timeframes=%s",
            signal_type,
            confluence_score,
            direction.value,
            active_timeframes,
        )

    return confluence_signals