MyStrategy

"""QuantGist macro-event Protection plugin for freqtrade.

Implements ``IProtection`` — freqtrade's plugin interface for blocking
trade entries/exits around risk events.

How it works
------------
1. On each ``global_stop`` / ``stop_per_pair`` call, the plugin fetches
   upcoming high-impact events from the QuantGist API (cached for
   ``cache_ttl_seconds``, default 300 s / 5 min).
2. If any event falls within [now - pause_minutes_after,
   now + pause_minutes_before], it returns ``ProtectionReturn(True, ...)``.
3. For ``stop_per_pair``, only events whose currency matches the pair's
   affected currencies (via ``symbol_map``) trigger the stop.
4. If the QuantGist API is unreachable, the plugin **fails open** —
   trading continues and a warning is logged.

Config keys (inside the protection object in config.json):
- ``api_key``              (required) — your ``qg_live_...`` key
- ``pause_minutes_before`` (default 10) — minutes before event to block
- ``pause_minutes_after``  (default  5) — minutes after event to block
- ``impact``               (default "high") — minimum impact level to react to
- ``cache_ttl_seconds``    (default 300) — how long to cache API results
"""

from __future__ import annotations

import logging
import os
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any

logger = logging.getLogger(__name__)

# ---------------------------------------------------------------------------
# freqtrade imports — wrapped so the module can be imported without freqtrade
# installed (useful for unit tests and type-checking outside the bot).
# ---------------------------------------------------------------------------
try:
    from freqtrade.constants import Config
    from freqtrade.persistence import Trade
    from freqtrade.plugins.protections import IProtection, ProtectionReturn

    _FT_AVAILABLE = True
except ImportError:  # pragma: no cover
    _FT_AVAILABLE = False

    # Minimal stubs so the module is importable for testing/type-checking
    class IProtection:  # type: ignore[no-redef]
        def __init__(self, config: Any, protection_config: dict[str, Any]) -> None:
            pass

    @dataclass
    class ProtectionReturn:  # type: ignore[no-redef]
        stop: bool = False
        reason: str = ""
        until: datetime | None = None

    Trade = Any  # type: ignore[assignment, misc]
    Config = dict  # type: ignore[assignment, misc]

# QuantGist SDK
try:
    from quantgist import QuantGistClient
    from quantgist.exceptions import QuantGistError
    from quantgist.models import Event

    _SDK_AVAILABLE = True
except ImportError as exc:  # pragma: no cover
    logger.error("quantgist SDK not installed: %s — pip install quantgist", exc)
    _SDK_AVAILABLE = False
    QuantGistClient = None  # type: ignore[assignment, misc]
    QuantGistError = Exception  # type: ignore[assignment, misc]
    Event = Any  # type: ignore[assignment, misc]

from qg_freqtrade.symbol_map import get_affected_currencies


# ---------------------------------------------------------------------------
# Simple TTL cache — avoids repeated API calls within the same minute
# ---------------------------------------------------------------------------
@dataclass
class _EventCache:
    events: list[Any] = field(default_factory=list)
    fetched_at: datetime = field(default_factory=lambda: datetime.min.replace(tzinfo=timezone.utc))
    ttl_seconds: int = 300

    def is_fresh(self, now: datetime) -> bool:
        age = (now - self.fetched_at).total_seconds()
        return age < self.ttl_seconds

    def update(self, events: list[Any], now: datetime) -> None:
        self.events = events
        self.fetched_at = now


class QuantGistProtection(IProtection):
    """freqtrade IProtection that blocks trading around high-impact macro events.

    Add to your ``config.json``::

        "protections": [
            {
                "method": "QuantGistProtection",
                "api_key": "qg_live_...",
                "pause_minutes_before": 10,
                "pause_minutes_after": 5,
                "impact": "high"
            }
        ]

    Or reference the class directly in your strategy::

        from qg_freqtrade import QuantGistProtection

        class MyStrategy(IStrategy):
            @property
            def protections(self):
                return [
                    {
                        "method": "QuantGistProtection",
                        "api_key": os.environ["QUANTGIST_API_KEY"],
                        "pause_minutes_before": 10,
                        "pause_minutes_after": 5,
                    }
                ]
    """

    has_global_stop: bool = True
    has_local_stop: bool = True

    def __init__(self, config: Any, protection_config: dict[str, Any]) -> None:
        super().__init__(config, protection_config)

        # Config resolution: protection_config → env var → error
        self._api_key: str = (
            protection_config.get("api_key")
            or os.environ.get("QUANTGIST_API_KEY", "")
        )
        if not self._api_key:
            raise ValueError(
                "QuantGistProtection: 'api_key' must be set in protection config "
                "or QUANTGIST_API_KEY environment variable."
            )

        self._pause_before: int = int(protection_config.get("pause_minutes_before", 10))
        self._pause_after: int = int(protection_config.get("pause_minutes_after", 5))
        self._impact: str = protection_config.get("impact", "high")
        cache_ttl: int = int(protection_config.get("cache_ttl_seconds", 300))

        self._cache = _EventCache(ttl_seconds=cache_ttl)
        self._client: Any = None  # lazy-initialised on first use

    # ------------------------------------------------------------------
    # IProtection interface
    # ------------------------------------------------------------------

    def short_desc(self) -> str:
        return (
            f"QuantGistProtection — pause {self._pause_before} min before / "
            f"{self._pause_after} min after {self._impact}-impact events"
        )

    def global_stop(
        self,
        current_time: datetime,
        side: str = "*",
    ) -> ProtectionReturn:
        """Block ALL pairs if any high-impact event is within the window."""
        events = self._get_upcoming_events(current_time)
        for event in events:
            in_window, reason = self._event_in_window(event, current_time)
            if in_window:
                logger.info(
                    "QuantGistProtection: global stop triggered — %s (%s)",
                    event.title,
                    reason,
                )
                until = self._window_end(event, current_time)
                return ProtectionReturn(stop=True, reason=reason, until=until)

        return ProtectionReturn(stop=False, reason="", until=None)

    def stop_per_pair(
        self,
        pair: str,
        current_time: datetime,
        side: str = "*",
    ) -> ProtectionReturn:
        """Block a specific pair if an event affects its currencies."""
        affected = get_affected_currencies(pair)
        if not affected:
            # No currency mapping — skip rather than block everything
            return ProtectionReturn(stop=False, reason="", until=None)

        events = self._get_upcoming_events(current_time)
        for event in events:
            # Match on the event's currency field
            event_currency = getattr(event, "currency", None)
            if event_currency and event_currency.upper() in [c.upper() for c in affected]:
                in_window, reason = self._event_in_window(event, current_time)
                if in_window:
                    logger.info(
                        "QuantGistProtection: stop for %s — %s (%s)",
                        pair,
                        event.title,
                        reason,
                    )
                    until = self._window_end(event, current_time)
                    return ProtectionReturn(stop=True, reason=reason, until=until)

        return ProtectionReturn(stop=False, reason="", until=None)

    # ------------------------------------------------------------------
    # Internal helpers
    # ------------------------------------------------------------------

    def _get_client(self) -> Any:
        """Lazy-initialise the QuantGist SDK client."""
        if self._client is None:
            if not _SDK_AVAILABLE:
                raise RuntimeError("quantgist SDK not installed. Run: pip install quantgist")
            self._client = QuantGistClient(api_key=self._api_key)
        return self._client

    def _get_upcoming_events(self, current_time: datetime) -> list[Any]:
        """Return upcoming events, using cache when fresh.

        Fails open: returns an empty list and logs a warning if the API
        is unreachable or returns an error.
        """
        now = _ensure_utc(current_time)

        if self._cache.is_fresh(now):
            return self._cache.events

        try:
            client = self._get_client()
            # Look ahead far enough to cover the pause window (+1 h buffer)
            lookahead_hours = max(2, self._pause_before // 60 + 2)
            response = client.get_events(
                from_date=now.isoformat(),
                to_date=None,
                impact=self._impact,  # type: ignore[arg-type]
                limit=100,
            )
            events = response.data
            self._cache.update(events, now)
            logger.debug(
                "QuantGistProtection: fetched %d %s-impact events",
                len(events),
                self._impact,
            )
            return events

        except QuantGistError as exc:
            logger.warning(
                "QuantGistProtection: API error (%s) — failing open, trading continues.",
                exc,
            )
            # Return stale cache if we have it, otherwise empty list
            return self._cache.events

        except Exception as exc:  # noqa: BLE001
            logger.warning(
                "QuantGistProtection: unexpected error (%s) — failing open.",
                exc,
            )
            return self._cache.events

    def _event_in_window(
        self, event: Any, current_time: datetime
    ) -> tuple[bool, str]:
        """Return (True, reason_string) if the event falls within the pause window."""
        now = _ensure_utc(current_time)
        release = _ensure_utc(event.release_time)

        seconds_to_event = (release - now).total_seconds()
        seconds_since_event = (now - release).total_seconds()

        if 0 <= seconds_to_event <= self._pause_before * 60:
            mins = int(seconds_to_event // 60)
            return True, (
                f"{event.title} ({event.currency}) releases in {mins} min — "
                f"blocking {self._pause_before} min before event"
            )

        if 0 <= seconds_since_event <= self._pause_after * 60:
            mins = int(seconds_since_event // 60)
            return True, (
                f"{event.title} ({event.currency}) released {mins} min ago — "
                f"blocking {self._pause_after} min after event"
            )

        return False, ""

    def _window_end(self, event: Any, current_time: datetime) -> datetime:
        """Calculate when the protection window lifts for an event."""
        release = _ensure_utc(event.release_time)
        now = _ensure_utc(current_time)

        from datetime import timedelta

        if release > now:
            # Event is in the future — block until pause_after minutes after release
            return release + timedelta(minutes=self._pause_after)
        else:
            # Event already happened — block until pause_after window expires
            return release + timedelta(minutes=self._pause_after)


# ---------------------------------------------------------------------------
# Utilities
# ---------------------------------------------------------------------------

def _ensure_utc(dt: datetime) -> datetime:
    """Ensure a datetime is timezone-aware (UTC).  Naive datetimes are assumed UTC."""
    if dt.tzinfo is None:
        return dt.replace(tzinfo=timezone.utc)
    return dt.astimezone(timezone.utc)