PulseStrategy

"""PulseStrategy — Freqtrade base strategy that speaks PULSE.

This module provides PulseStrategy: a Freqtrade IStrategy subclass that:
  1. Emits PULSE messages when trades open/close (for other agents to observe)
  2. Can receive PULSE signals from external AI agents to influence entry/exit
  3. Logs all trading events in PULSE format for audit trails

Usage in Freqtrade config:
    strategy = "MyPulseStrategy"  # your class that extends PulseStrategy

Example:
    >>> class MyStrategy(PulseStrategy):
    ...     timeframe = "1h"
    ...
    ...     def populate_entry_trend(self, dataframe, metadata):
    ...         # Standard Freqtrade logic
    ...         dataframe.loc[dataframe["rsi"] < 30, "enter_long"] = 1
    ...         return dataframe
    ...
    ...     def populate_exit_trend(self, dataframe, metadata):
    ...         dataframe.loc[dataframe["rsi"] > 70, "exit_long"] = 1
    ...         return dataframe

Note:
    This module gracefully degrades if freqtrade is not installed —
    useful for testing and standalone PULSE usage.
"""

import logging
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional

from pulse.message import PulseMessage

logger = logging.getLogger(__name__)


# Try to import Freqtrade — optional dependency
try:
    from freqtrade.strategy import IStrategy
    import pandas as pd
    FREQTRADE_AVAILABLE = True
except ImportError:
    FREQTRADE_AVAILABLE = False
    IStrategy = object  # fallback base class for type checking
    logger.debug("freqtrade not installed — PulseStrategy running in standalone mode.")


class PulseSignalBus:
    """Simple in-process signal bus for PULSE ↔ Freqtrade communication.

    External AI agents push signals here; PulseStrategy reads them
    on the next candle close.

    Example:
        >>> bus = PulseSignalBus()
        >>> # AI agent pushes a signal
        >>> bus.push(PulseMessage(
        ...     action="ACT.RECOMMEND.ACTION",
        ...     parameters={"pair": "BTC/USDT", "direction": "long"}
        ... ))
        >>> # Strategy reads on next tick
        >>> signals = bus.pop_all()
    """

    def __init__(self) -> None:
        self._queue: List[PulseMessage] = []
        self._listeners: List[Any] = []

    def push(self, message: PulseMessage) -> None:
        """Push a PULSE signal into the bus.

        Args:
            message: PULSE message with trading signal
        """
        self._queue.append(message)
        logger.info(
            f"[PulseSignalBus] Signal received: "
            f"{message.content.get('action')} "
            f"| pair={message.content.get('parameters', {}).get('pair', 'N/A')}"
        )

    def pop_all(self) -> List[PulseMessage]:
        """Consume and return all pending signals.

        Returns:
            List of pending PULSE messages (clears the queue)
        """
        signals = self._queue.copy()
        self._queue.clear()
        return signals

    def pending_count(self) -> int:
        """Number of unprocessed signals in queue."""
        return len(self._queue)

    def subscribe(self, callback: Any) -> None:
        """Register a callback for immediate signal notification.

        Args:
            callback: Callable that receives a PulseMessage
        """
        self._listeners.append(callback)


# Global signal bus — shared between strategy and external agents
signal_bus = PulseSignalBus()


class PulseStrategy(IStrategy):
    """Freqtrade strategy base class with PULSE integration.

    Extend this instead of IStrategy to get PULSE superpowers:
      - All trade events emitted as PULSE messages
      - External AI agents can inject signals via signal_bus
      - Full audit trail in PULSE format

    Your strategy still works exactly like a normal Freqtrade strategy —
    just inherit from PulseStrategy instead of IStrategy.

    Example:
        class RSIPulseStrategy(PulseStrategy):
            timeframe = "1h"
            pulse_agent_id = "rsi-strategy-001"

            def populate_entry_trend(self, dataframe, metadata):
                # Check PULSE signals from AI agents first
                pulse_signals = self.get_pulse_signals(metadata["pair"])
                if pulse_signals:
                    signal = pulse_signals[0]
                    direction = signal.content["parameters"].get("direction")
                    if direction == "long":
                        dataframe["enter_long"] = 1
                        return dataframe

                # Fallback: standard RSI logic
                dataframe.loc[dataframe["rsi"] < 30, "enter_long"] = 1
                return dataframe
    """

    # Override in subclass to identify this agent
    pulse_agent_id: str = "freqtrade-strategy"

    # Set to False to disable PULSE event emission
    pulse_emit_events: bool = True

    def __init__(self, config: Dict[str, Any]) -> None:
        if FREQTRADE_AVAILABLE:
            super().__init__(config)
        self._pulse_events: List[PulseMessage] = []
        logger.info(f"[PulseStrategy] Initialized agent_id={self.pulse_agent_id!r}")

    # --- PULSE Signal Integration ---

    def get_pulse_signals(self, pair: Optional[str] = None) -> List[PulseMessage]:
        """Get pending PULSE signals, optionally filtered by pair.

        Call this in populate_entry_trend() to incorporate AI signals.

        Args:
            pair: Filter signals for specific pair (e.g. "BTC/USDT").
                  If None, returns all pending signals.

        Returns:
            List of PULSE signal messages

        Example:
            >>> signals = self.get_pulse_signals("BTC/USDT")
            >>> if signals:
            ...     direction = signals[0].content["parameters"]["direction"]
        """
        all_signals = signal_bus.pop_all()
        if pair is None:
            return all_signals
        return [
            s for s in all_signals
            if s.content.get("parameters", {}).get("pair") == pair
        ]

    # --- PULSE Event Emission ---

    def emit_trade_opened(self, pair: str, amount: float, price: float, trade_id: int) -> None:
        """Emit a PULSE message when a trade is opened.

        Args:
            pair: Trading pair (e.g. "BTC/USDT")
            amount: Amount purchased
            price: Entry price
            trade_id: Freqtrade internal trade ID
        """
        if not self.pulse_emit_events:
            return
        msg = PulseMessage(
            action="ACT.TRANSACT.REQUEST",
            parameters={
                "event": "trade_opened",
                "pair": pair,
                "amount": amount,
                "price": price,
                "trade_id": trade_id,
                "agent_id": self.pulse_agent_id,
                "timestamp": datetime.now(timezone.utc).isoformat(),
            },
            validate=False,
        )
        self._pulse_events.append(msg)
        logger.info(
            f"[PULSE] Trade opened: {pair} | amount={amount} | price={price} | id={trade_id}"
        )

    def emit_trade_closed(
        self,
        pair: str,
        profit_percent: float,
        profit_abs: float,
        trade_id: int,
        reason: str = "roi",
    ) -> None:
        """Emit a PULSE message when a trade is closed.

        Args:
            pair: Trading pair
            profit_percent: Profit/loss in percent
            profit_abs: Absolute profit/loss in stake currency
            trade_id: Freqtrade internal trade ID
            reason: Close reason (roi, stop_loss, trailing_stop, force_sell)
        """
        if not self.pulse_emit_events:
            return
        action = "ACT.TRANSACT.COMPLETE" if profit_percent >= 0 else "ACT.TRANSACT.FAIL"
        msg = PulseMessage(
            action=action,
            parameters={
                "event": "trade_closed",
                "pair": pair,
                "profit_percent": round(profit_percent, 4),
                "profit_abs": round(profit_abs, 8),
                "trade_id": trade_id,
                "reason": reason,
                "agent_id": self.pulse_agent_id,
                "timestamp": datetime.now(timezone.utc).isoformat(),
            },
            validate=False,
        )
        self._pulse_events.append(msg)
        emoji = "✅" if profit_percent >= 0 else "❌"
        logger.info(
            f"[PULSE] {emoji} Trade closed: {pair} | "
            f"profit={profit_percent:.2f}% ({profit_abs:.4f}) | reason={reason}"
        )

    def emit_signal(self, pair: str, direction: str, confidence: float = 1.0) -> None:
        """Emit a PULSE trading signal for other agents to observe.

        Args:
            pair: Trading pair (e.g. "BTC/USDT")
            direction: "long" or "short"
            confidence: Signal confidence 0.0–1.0
        """
        if not self.pulse_emit_events:
            return
        msg = PulseMessage(
            action="ACT.RECOMMEND.ACTION",
            parameters={
                "pair": pair,
                "direction": direction,
                "confidence": confidence,
                "agent_id": self.pulse_agent_id,
                "timestamp": datetime.now(timezone.utc).isoformat(),
            },
            validate=False,
        )
        self._pulse_events.append(msg)
        logger.info(
            f"[PULSE] Signal emitted: {pair} {direction.upper()} | confidence={confidence:.2f}"
        )

    def get_pulse_events(self, clear: bool = True) -> List[PulseMessage]:
        """Get all PULSE events emitted by this strategy.

        Args:
            clear: Whether to clear the event log after reading (default True)

        Returns:
            List of PULSE messages representing all trade events
        """
        events = self._pulse_events.copy()
        if clear:
            self._pulse_events.clear()
        return events

    # --- Freqtrade hook overrides ---

    def confirm_trade_entry(
        self,
        pair: str,
        order_type: str,
        amount: float,
        rate: float,
        time_in_force: str,
        current_time: datetime,
        entry_tag: Optional[str],
        side: str,
        **kwargs: Any,
    ) -> bool:
        """Hook called before trade entry — emit PULSE signal."""
        self.emit_signal(pair, "long" if side == "buy" else "short")
        return True

    def confirm_trade_exit(
        self,
        pair: str,
        trade: Any,
        order_type: str,
        amount: float,
        rate: float,
        time_in_force: str,
        exit_reason: str,
        current_time: datetime,
        **kwargs: Any,
    ) -> bool:
        """Hook called before trade exit — emit PULSE close event."""
        profit_percent = getattr(trade, "calc_profit_ratio", lambda r: 0.0)(rate) * 100
        profit_abs = getattr(trade, "calc_profit", lambda r: 0.0)(rate)
        trade_id = getattr(trade, "id", 0)
        self.emit_trade_closed(pair, profit_percent, profit_abs, trade_id, exit_reason)
        return True