Freqtrade strategy base class with PULSE integration.
Timeframe
1h
Direction
Long Only
Stoploss
N/A
Trailing Stop
No
ROI
N/A
Interface Version
N/A
Startup Candles
N/A
Indicators
1
freqtrade/freqtrade-strategies
freqtrade/freqtrade-strategies
this is an example class, implementing a PSAR based trailing stop loss you are supposed to take the `custom_stoploss()` and `populate_indicators()` parts and adapt it to your own strategy
freqtrade/freqtrade-strategies
Strategy 003 author@: Gerald Lonlas github@: https://github.com/freqtrade/freqtrade-strategies
"""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