QuantCore-FX/currency_strength_matrix.py

from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass
from datetime import datetime, timezone
import statistics
import config


@dataclass
class CurrencyStrength:
    name: str
    avg_z_score: float
    rank: int
    is_overbought: bool
    is_oversold: bool
    direction: str
    session_srv: float = 0.0  # Session Relative Velocity (Task 1.2)


class SessionTracker:
    """Detects the active trading session and tracks session-start snapshots.

    Sessions (UTC):
        Tokyo:    00:00–08:00
        London:   07:00–16:00
        New York: 13:00–22:00
    Overlapping hours are resolved as London (the dominant session).
    """

    TOKYO = "Tokyo"
    LONDON = "London"
    NEWYORK = "New York"

    def __init__(self):
        self.current_session: Optional[str] = None
        self.session_start_prices: Dict[str, float] = {}  # pair -> price at session open
        self.session_open_time: Optional[datetime] = None

    def get_active_session(self, utc_hour: int = None) -> str:
        if utc_hour is None:
            utc_hour = datetime.now(timezone.utc).hour
        if config.SESSION_LONDON_OPEN <= utc_hour < config.SESSION_LONDON_CLOSE:
            return self.LONDON
        if config.SESSION_TOKYO_OPEN <= utc_hour < config.SESSION_TOKYO_CLOSE:
            return self.TOKYO
        if config.SESSION_NEWYORK_OPEN <= utc_hour < config.SESSION_NEWYORK_CLOSE:
            return self.NEWYORK
        return "Off-Hours"

    def check_new_session(self, current_prices: Dict[str, float]) -> Optional[str]:
        """Detect if a new session has started and snapshot prices."""
        now = datetime.now(timezone.utc)
        session = self.get_active_session(now.hour)
        if session != self.current_session and session != "Off-Hours":
            self.current_session = session
            self.session_start_prices = dict(current_prices)
            self.session_open_time = now
            return session
        if self.current_session is None:
            self.current_session = session
            if session != "Off-Hours":
                self.session_start_prices = dict(current_prices)
                self.session_open_time = now
        return None

    def compute_srv(self, pair: str, current_price: float) -> float:
        """Session Relative Velocity: % change from session open to now."""
        start = self.session_start_prices.get(pair)
        if start is None or start == 0:
            return 0.0
        return ((current_price - start) / start) * 100.0


class CurrencyStrengthMatrix:
    """Computes individual currency strength indices from pair Z-scores.

    Includes Session-Based Indexing (Task 1.2):
    - Tracks performance since Tokyo/London/NY session opens
    - Session Relative Velocity (SRV) per currency
    """

    def __init__(self, z_scores: Dict[str, float] = None):
        self.currencies = config.CURRENCIES
        self.threshold = config.SCALP_Z_SCORE_THRESHOLD
        self._raw_scores: Dict[str, List[float]] = {}
        self._strengths: Dict[str, CurrencyStrength] = {}
        self.session_tracker = SessionTracker()
        self._srv_map: Dict[str, float] = {}  # currency -> avg SRV
        if z_scores:
            self.update(z_scores)

    def update(self, z_scores: Dict[str, float], current_prices: Dict[str, float] = None):
        """Recompute all currency strengths from 28 pair Z-scores.

        If current_prices is provided, also updates session tracking
        and computes Session Relative Velocity.
        """
        self._raw_scores = {}
        for ccy in self.currencies:
            scores = []
            for other in self.currencies:
                if other == ccy:
                    continue
                pair = f"{ccy}_{other}"
                z = z_scores.get(pair)
                if z is not None:
                    scores.append(z)
            self._raw_scores[ccy] = scores

        strengths = {}
        for ccy, scores in self._raw_scores.items():
            avg_z = statistics.mean(scores) if scores else 0.0
            strengths[ccy] = CurrencyStrength(
                name=ccy,
                avg_z_score=avg_z,
                rank=0,
                is_overbought=avg_z >= self.threshold,
                is_oversold=avg_z <= -self.threshold,
                direction="OVERBOUGHT" if avg_z >= self.threshold else ("OVERSOLD" if avg_z <= -self.threshold else "NEUTRAL"),
            )

        sorted_ccys = sorted(strengths.keys(), key=lambda c: strengths[c].avg_z_score, reverse=True)
        for rank, ccy in enumerate(sorted_ccys, 1):
            strengths[ccy].rank = rank

        # Session tracking (Task 1.2)
        if current_prices:
            new_session = self.session_tracker.check_new_session(current_prices)
            self._compute_srv(current_prices, strengths)

        self._strengths = strengths

    def _compute_srv(self, current_prices: Dict[str, float],
                     strengths: Dict[str, CurrencyStrength]):
        """Compute average Session Relative Velocity per currency."""
        srv_scores: Dict[str, List[float]] = {c: [] for c in self.currencies}
        for ccy in self.currencies:
            for other in self.currencies:
                if other == ccy:
                    continue
                pair = f"{ccy}_{other}"
                price = current_prices.get(pair)
                if price is not None:
                    srv = self.session_tracker.compute_srv(pair, price)
                    srv_scores[ccy].append(srv)
        for ccy in self.currencies:
            scores = srv_scores[ccy]
            self._srv_map[ccy] = statistics.mean(scores) if scores else 0.0
            if ccy in strengths:
                strengths[ccy].session_srv = self._srv_map[ccy]

    def get_strongest(self) -> Optional[CurrencyStrength]:
        return max(self._strengths.values(), key=lambda s: s.avg_z_score) if self._strengths else None

    def get_weakest(self) -> Optional[CurrencyStrength]:
        return min(self._strengths.values(), key=lambda s: s.avg_z_score) if self._strengths else None

    def get_matrix_cross(self) -> Optional[str]:
        s = self.get_strongest()
        w = self.get_weakest()
        if s and w and s.name != w.name:
            return f"{s.name}_{w.name}"
        return None

    def get_divergence_gap(self) -> float:
        s = self.get_strongest()
        w = self.get_weakest()
        return (s.avg_z_score - w.avg_z_score) if s and w else 0.0

    def has_divergence(self) -> bool:
        s = self.get_strongest()
        w = self.get_weakest()
        return bool(s and w and s.is_overbought and w.is_oversold)

    def get_strong_currencies(self) -> List[str]:
        return [c.name for c in self._strengths.values() if c.is_overbought]

    def get_weak_currencies(self) -> List[str]:
        return [c.name for c in self._strengths.values() if c.is_oversold]

    def get_ranked_list(self) -> List[CurrencyStrength]:
        return sorted(self._strengths.values(), key=lambda s: s.rank)

    def get_active_session(self) -> str:
        return self.session_tracker.get_active_session()

    def get_srv_map(self) -> Dict[str, float]:
        return dict(self._srv_map)

    def get_report(self) -> Dict:
        ranked = self.get_ranked_list()
        s = self.get_strongest()
        w = self.get_weakest()
        return {
            "ranked": [(c.name, round(c.avg_z_score, 2), c.direction, round(c.session_srv, 4)) for c in ranked],
            "strongest": s.name if s else None,
            "strongest_z": round(s.avg_z_score, 2) if s else 0,
            "weakest": w.name if w else None,
            "weakest_z": round(w.avg_z_score, 2) if w else 0,
            "matrix_cross": self.get_matrix_cross(),
            "divergence_gap": round(self.get_divergence_gap(), 2),
            "has_divergence": self.has_divergence(),
            "overbought": self.get_strong_currencies(),
            "oversold": self.get_weak_currencies(),
            "active_session": self.get_active_session(),
        }