No-op strategy used when the user supplies no override.
Timeframe
1h
Direction
Long Only
Stoploss
-10.0%
Trailing Stop
No
ROI
N/A
Interface Version
N/A
Startup Candles
30
Indicators
2
freqtrade/freqtrade-strategies
Strategy 003 author@: Gerald Lonlas github@: https://github.com/freqtrade/freqtrade-strategies
"""Freqtrade-inspired strategy interface.
This module provides a pluggable strategy layer that lets users customize
the bot's behavior via callback methods, independent of the underlying
Rust engine. Users subclass :class:`IStrategy` and override only the
hooks they need; :class:`StrategyRunner` applies those callbacks to the
order stream produced by the engine.
"""
from __future__ import annotations
import logging
from abc import ABC, abstractmethod
from dataclasses import dataclass, replace
from datetime import datetime
from typing import TYPE_CHECKING, Any, Callable
from combo_bot.types import (
AccountState,
Candle,
ExchangeParams,
Fill,
Order,
OrderSource,
Position,
Side,
SymbolState,
TrendRegime,
TrendSignal,
)
logger = logging.getLogger(__name__)
if TYPE_CHECKING:
# Imported lazily inside methods to avoid hard pandas dependency at import time.
from pandas import DataFrame
# ---------------------------------------------------------------------------
# Context passed to every strategy callback
# ---------------------------------------------------------------------------
@dataclass
class TradeContext:
"""Snapshot of state passed to strategy callbacks.
All fields are read-only references to the bot's current view of the
world. Strategies must not mutate these values; return new values from
callbacks instead.
"""
symbol: str
side: Side
position: Position
account: AccountState
candle: Candle
signal: TrendSignal | None
current_time_ms: int
exchange_params: ExchangeParams
source: OrderSource = OrderSource.GRID
@property
def current_price(self) -> float:
return self.candle.close
@property
def is_in_position(self) -> bool:
return self.position.is_open
# ---------------------------------------------------------------------------
# Strategy interface
# ---------------------------------------------------------------------------
class IStrategy(ABC):
"""Abstract base for user-defined strategies.
The three ``populate_*`` methods are required. All other hooks have
sensible no-op defaults so subclasses can override only what they
need.
"""
# --- metadata -----------------------------------------------------------
timeframe: str = "1h"
stoploss: float = -0.10
trailing_stop: bool = False
process_only_new_candles: bool = True
startup_candle_count: int = 30
# --- required overrides --------------------------------------------------
@abstractmethod
def populate_indicators(
self, dataframe: "DataFrame", metadata: dict[str, Any]
) -> "DataFrame":
"""Compute indicator columns on the dataframe and return it."""
@abstractmethod
def populate_entry_trend(
self, dataframe: "DataFrame", metadata: dict[str, Any]
) -> "DataFrame":
"""Mark entry signals via ``enter_long``, ``enter_short``, ``enter_tag``."""
@abstractmethod
def populate_exit_trend(
self, dataframe: "DataFrame", metadata: dict[str, Any]
) -> "DataFrame":
"""Mark exit signals via ``exit_long``, ``exit_short``, ``exit_tag``."""
# --- lifecycle hooks -----------------------------------------------------
def bot_start(self, **kwargs: Any) -> None:
"""Called once when the bot starts. Default: no-op."""
def bot_loop_start(self, current_time: datetime, **kwargs: Any) -> None:
"""Called at the start of each tick. Default: no-op."""
def informative_pairs(self) -> list[tuple[str, str]]:
"""Additional (pair, timeframe) tuples to load. Default: none."""
return []
def order_filled(
self,
ctx: "TradeContext",
fill: "Fill",
**kwargs: Any,
) -> None:
"""Round-28: Freqtrade-style post-fill callback.
Fires AFTER a fill has been booked into the position bucket
and the realized PnL credited to balance. Strategies use this
for:
* post-fill audit / structured logging
* tightening trailing stops based on filled price
* recording features/labels for an ML pipeline
* sending an external notification (Slack, webhook)
Default: no-op. Wrapped so a misbehaving hook can't kill the
tick. ``ctx.position`` reflects the post-fill state.
"""
# --- entry / exit veto ---------------------------------------------------
def confirm_trade_entry(
self,
ctx: TradeContext,
proposed_qty: float,
proposed_price: float,
) -> bool:
"""Return ``False`` to veto a proposed entry. Default: allow."""
return True
def confirm_trade_exit(
self,
ctx: TradeContext,
proposed_qty: float,
proposed_price: float,
exit_reason: str,
) -> bool:
"""Return ``False`` to veto a proposed exit. Default: allow."""
return True
# --- price / size adjustment --------------------------------------------
def custom_entry_price(self, ctx: TradeContext, proposed_price: float) -> float:
"""Adjust the entry price. Default: passthrough."""
return proposed_price
def custom_exit_price(
self,
ctx: TradeContext,
proposed_price: float,
exit_reason: str,
) -> float:
"""Adjust the exit price. Default: passthrough."""
return proposed_price
def custom_stake_amount(
self,
ctx: TradeContext,
proposed_stake: float,
min_stake: float,
max_stake: float,
) -> float:
"""Override position size in quote currency. Default: passthrough."""
return proposed_stake
# --- adaptive risk ------------------------------------------------------
def custom_stoploss(
self, ctx: TradeContext, current_profit_pct: float
) -> float | None:
"""Return an absolute stoploss price, or ``None`` to skip. Default: ``None``."""
return None
def custom_exit(self, ctx: TradeContext, current_profit_pct: float) -> str | None:
"""Return an exit reason to trigger a close, or ``None`` to skip."""
return None
def adjust_trade_position(
self, ctx: TradeContext, current_profit_pct: float
) -> float | None:
"""Return qty to add (positive) or trim (negative), or ``None``."""
return None
# --- pending order adjustment -------------------------------------------
def adjust_entry_price(
self, ctx: TradeContext, current_order_price: float
) -> float | None:
"""Modify a pending entry price. Default: keep as-is."""
return None
def adjust_exit_price(
self,
ctx: TradeContext,
current_order_price: float,
exit_reason: str,
) -> float | None:
"""Modify a pending exit price. Default: keep as-is."""
return None
# --- leverage / timeouts ------------------------------------------------
def leverage(
self,
ctx: TradeContext,
proposed_leverage: float,
max_leverage: float,
) -> float:
"""Set leverage for a new position. Default: passthrough, capped."""
return min(proposed_leverage, max_leverage)
def check_entry_timeout(self, ctx: TradeContext, order_age_seconds: int) -> bool:
"""Return ``True`` to cancel a stale pending entry. Default: never."""
return False
def check_exit_timeout(self, ctx: TradeContext, order_age_seconds: int) -> bool:
"""Return ``True`` to cancel a stale pending exit. Default: never."""
return False
# ---------------------------------------------------------------------------
# Strategy runner — bridges the callbacks to the engine's order stream
# ---------------------------------------------------------------------------
class StrategyRunner:
"""Applies a strategy's callbacks to the order list produced by the engine."""
def __init__(self, strategy: IStrategy) -> None:
self.strategy = strategy
# --- entry filtering ----------------------------------------------------
def filter_entries(self, orders: list[Order], context: TradeContext) -> list[Order]:
"""Apply entry-side price + sizing hooks to a list of orders.
Round-25 P1 #4: ``confirm_trade_entry`` is NO LONGER called
here. It now runs in a final pass (:meth:`final_confirm`)
AFTER correlation / vol-target / risk-partial-fit have all
had a chance to scale qty down. Otherwise the strategy was
confirming on a qty the bot never actually transacted on.
Remaining processing here mirrors Freqtrade's pre-confirm
sequence:
1. ``custom_entry_price`` / ``adjust_entry_price`` — adjust price.
2. ``custom_stake_amount`` — adjust qty for new entries.
"""
result: list[Order] = []
for order in orders:
if order.reduce_only:
# Reduce-only orders are exits, not entries — pass through.
result.append(order)
continue
ctx = self._ctx_for_order(context, order)
c_mult = (
ctx.exchange_params.c_mult if ctx.exchange_params is not None else 1.0
)
if order.is_market:
# Market entries SKIP price hooks. Live sends
# ``price=None`` and backtest fills at close ± slip,
# so any value ``custom_entry_price`` returns would be
# a lie when fed to ``custom_stake_amount`` /
# ``confirm_trade_entry``. Use the current execution
# reference price (or the order's stated price as
# fallback) so confirm sees the number it actually
# transacts near.
final_price = (
ctx.current_price if ctx.current_price > 0 else order.price
)
else:
new_price = self.strategy.custom_entry_price(ctx, order.price)
adjusted = self.strategy.adjust_entry_price(ctx, new_price)
final_price = adjusted if adjusted is not None else new_price
# Notional INCLUDES contract multiplier. For c_mult != 1
# (index-multiplied / coin-margined), qty * price is just
# the dimensionless count × index — the real USD-equivalent
# stake is qty * price * c_mult.
proposed_stake = order.qty * final_price * c_mult
min_stake = ctx.exchange_params.min_cost
max_stake = max(ctx.account.balance, proposed_stake)
new_stake = self.strategy.custom_stake_amount(
ctx, proposed_stake, min_stake, max_stake
)
denom = final_price * c_mult
new_qty = new_stake / denom if denom > 0 else order.qty
if order.is_market:
# Don't overwrite the order's stated price for market
# entries — see parallel comment in filter_exits.
result.append(replace(order, qty=new_qty))
else:
result.append(replace(order, price=final_price, qty=new_qty))
return result
# --- exit filtering -----------------------------------------------------
def filter_exits(self, orders: list[Order], context: TradeContext) -> list[Order]:
"""Apply exit-side strategy hooks to a list of orders.
Order matches Freqtrade's lifecycle (freqtradebot.py:2106 area):
``custom_exit_price`` and ``adjust_exit_price`` run FIRST so
the strategy sees the final exit rate, then ``confirm_trade_exit``
decides on that final rate.
Market orders SKIP the price-adjustment hooks — live ignores
the price (sends with ``price=None``) and backtest fills at
close ± slippage, so any value the strategy returns is purely
informational and would mislead ``confirm_trade_exit`` about
the real execution price. For market exits the confirm hook
sees the order's stated reference price (typically the current
close) — the value the bot will actually transact near.
"""
result: list[Order] = []
for order in orders:
if not order.reduce_only:
# Non-reduce-only orders are entries — pass through.
result.append(order)
continue
ctx = self._ctx_for_order(context, order)
exit_reason = self._exit_reason_for(order)
if order.is_market:
# No price-adjustment hooks for market exits. The
# ``confirm_trade_exit`` hook sees the CURRENT execution
# reference price (e.g. the last close), not whatever
# threshold the order's ``price`` field encodes. For a
# custom_stoploss-triggered market exit, ``order.price``
# is the SL threshold; live sends ``price=None`` and the
# exchange fills near current — surfacing the threshold
# to confirm would mislead the strategy.
final_price = (
ctx.current_price if ctx.current_price > 0 else order.price
)
else:
new_price = self.strategy.custom_exit_price(
ctx, order.price, exit_reason
)
adjusted = self.strategy.adjust_exit_price(ctx, new_price, exit_reason)
final_price = adjusted if adjusted is not None else new_price
# Round-25 P1 #4: confirm_trade_exit moved to final_confirm
# pass so the strategy sees the FINAL qty after risk
# filtering / trimming, not the pre-filter qty.
if order.is_market:
# Preserve the order's stated price for market orders.
result.append(order)
else:
result.append(replace(order, price=final_price))
return result
# --- final confirm pass (Round-25 P1 #4) -------------------------------
def fire_order_filled(
self,
fills: list[Fill],
ctx_resolver: "Callable[[Fill], TradeContext]",
) -> None:
"""Round-28: dispatch ``strategy.order_filled`` for each
fill. Wrapped so a hook crash on one fill doesn't block the
rest. Should be called AFTER all bucket / balance / PnL
bookkeeping for the batch is done, so the strategy sees the
post-fill state (``ctx.position`` reflects the just-booked
delta).
"""
for fill in fills:
try:
ctx = ctx_resolver(fill)
except Exception:
logger.exception(
"[strategy] order_filled ctx resolution failed for "
"fill %s — skipping callback",
getattr(fill, "trade_id", None),
)
continue
try:
self.strategy.order_filled(ctx, fill)
except Exception:
logger.exception(
"[strategy] order_filled raised for fill %s — ignored",
getattr(fill, "trade_id", None),
)
def final_confirm(
self,
orders: list[Order],
ctx_resolver: "Callable[[Order], TradeContext]",
) -> list[Order]:
"""Run ``confirm_trade_entry`` / ``confirm_trade_exit`` against
the FINAL qty after all upstream sizing / risk passes.
Freqtrade calls the confirm hook as the last gate before
submitting to the exchange. Up to round 24 we ran it inside
``filter_entries`` / ``filter_exits`` — but correlation,
vol-target, and risk partial-fit can ALL shrink qty after
that, so the strategy never saw the number actually
transacted. This pass runs once at the end of the pipeline
per Backtester / LiveTrader to close that gap.
``ctx_resolver`` is a callable that builds a fresh
:class:`TradeContext` for each order — backtest and live each
supply their own resolver so this layer doesn't need to
know about candle caches, signals, or per-symbol state.
Orders are kept in their input order; a vetoed order is
dropped and no attempt is made to "scale to acceptable".
"""
accepted: list[Order] = []
for order in orders:
try:
ctx = ctx_resolver(order)
except Exception:
# Resolver failed (e.g. missing symbol state) — fail
# CLOSED for entries, OPEN for reduce-only so we
# never silently strand exits.
if order.reduce_only:
accepted.append(order)
continue
# Market orders: surface ctx.current_price instead of the
# order's stated price. For market entries the live
# executor sends price=None and the fill lands near the
# current execution reference. For market exits the
# order.price typically encodes a stoploss threshold,
# which would mislead the strategy about the actual exit
# rate. Limit orders use the (post-adjust) order price.
if order.is_market and ctx.current_price > 0:
confirm_price = ctx.current_price
else:
confirm_price = order.price
try:
if order.reduce_only:
exit_reason = self._exit_reason_for(order)
ok = self.strategy.confirm_trade_exit(
ctx, order.qty, confirm_price, exit_reason
)
else:
ok = self.strategy.confirm_trade_entry(
ctx, order.qty, confirm_price
)
except Exception:
# Same fail-closed/open policy on hook crash.
if order.reduce_only:
accepted.append(order)
continue
if ok:
accepted.append(order)
return accepted
# --- adaptive exit checks -----------------------------------------------
def check_custom_exit(self, context: TradeContext) -> Order | None:
"""Run ``custom_exit`` then ``custom_stoploss``. Returns a market close order if triggered."""
if not context.position.is_open:
return None
current_profit_pct = self._profit_pct(context)
reason = self.strategy.custom_exit(context, current_profit_pct)
if reason is not None:
return self._make_close_order(context, context.current_price)
sl_price = self.strategy.custom_stoploss(context, current_profit_pct)
if sl_price is not None and self._stoploss_hit(context, sl_price):
return self._make_close_order(context, sl_price)
return None
def check_position_adjustment(self, context: TradeContext) -> Order | None:
"""Call ``adjust_trade_position`` and turn its result into an order.
The emitted order targets the same source bucket the strategy
inspected in ``ctx.position``. Grid contexts remain GRID; trend
contexts remain TREND so strategy-driven exits/adjustments do not
accidentally hit the grid bucket.
"""
if not context.position.is_open:
return None
current_profit_pct = self._profit_pct(context)
delta_qty = self.strategy.adjust_trade_position(context, current_profit_pct)
if delta_qty is None or abs(delta_qty) < 1e-12:
return None
# Positive delta = add to position; negative = trim.
reduce_only = (context.side == Side.LONG and delta_qty < 0) or (
context.side == Side.SHORT and delta_qty > 0
)
return Order(
symbol=context.symbol,
side=context.side,
price=context.current_price,
qty=abs(delta_qty),
source=context.source,
reduce_only=reduce_only,
)
# --- internal helpers ---------------------------------------------------
def _ctx_for_order(self, base: TradeContext, order: Order) -> TradeContext:
"""Build a per-order context (side may differ from the base context)."""
if (
order.side == base.side
and order.symbol == base.symbol
and order.source == base.source
):
return base
symbol_state = base.account.symbols.get(order.symbol)
position = self._position_for(symbol_state, order.source, order.side)
return TradeContext(
symbol=order.symbol,
side=order.side,
position=position,
account=base.account,
candle=base.candle,
signal=base.signal,
current_time_ms=base.current_time_ms,
exchange_params=base.exchange_params,
source=order.source,
)
@staticmethod
def _position_for(
symbol_state: SymbolState | None, source: OrderSource, side: Side
) -> Position:
if symbol_state is None:
return Position()
return symbol_state.bucket(source, side)
@staticmethod
def _profit_pct(ctx: TradeContext) -> float:
pos = ctx.position
if not pos.is_open or pos.entry_price <= 0:
return 0.0
if ctx.side == Side.LONG:
return (ctx.current_price - pos.entry_price) / pos.entry_price
return (pos.entry_price - ctx.current_price) / pos.entry_price
@staticmethod
def _stoploss_hit(ctx: TradeContext, sl_price: float) -> bool:
if ctx.side == Side.LONG:
return ctx.current_price <= sl_price
return ctx.current_price >= sl_price
@staticmethod
def _exit_reason_for(order: Order) -> str:
if order.source == OrderSource.RISK:
return "risk_exit"
if order.source == OrderSource.TREND:
return "trend_exit"
return "grid_exit"
@staticmethod
def _make_close_order(ctx: TradeContext, price: float) -> Order:
# Strategy-triggered exits are taker market orders so they actually
# fill on the current candle rather than waiting for a limit cross.
source = (
OrderSource.TREND if ctx.source == OrderSource.TREND else OrderSource.RISK
)
return Order(
symbol=ctx.symbol,
side=ctx.side,
price=price,
qty=abs(ctx.position.size),
source=source,
reduce_only=True,
is_market=True,
)
# ---------------------------------------------------------------------------
# Built-in strategies
# ---------------------------------------------------------------------------
class DefaultStrategy(IStrategy):
"""No-op strategy used when the user supplies no override.
All callbacks fall back to the :class:`IStrategy` defaults; the
required populate methods return the dataframe unchanged so no
trend-driven entries or exits are emitted.
"""
def populate_indicators(
self, dataframe: "DataFrame", metadata: dict[str, Any]
) -> "DataFrame":
return dataframe
def populate_entry_trend(
self, dataframe: "DataFrame", metadata: dict[str, Any]
) -> "DataFrame":
return dataframe
def populate_exit_trend(
self, dataframe: "DataFrame", metadata: dict[str, Any]
) -> "DataFrame":
return dataframe
class ExampleTrendStrategy(IStrategy):
"""Minimal but realistic example of the strategy pattern.
- Indicators: RSI(14), EMA(12) fast, EMA(26) slow.
- Long entries when RSI < 30 and the fast EMA is above the slow EMA.
- Long exits when RSI > 70.
- Trailing 5% stoploss via :meth:`custom_stoploss`.
- Vetoes entries when trend signal strength is weak.
"""
timeframe = "1h"
stoploss = -0.10
startup_candle_count = 30
# --- thresholds ---------------------------------------------------------
rsi_period: int = 14
rsi_oversold: float = 30.0
rsi_overbought: float = 70.0
ema_fast_period: int = 12
ema_slow_period: int = 26
trailing_stop_pct: float = 0.05
min_signal_strength: float = 0.5
# --- required overrides --------------------------------------------------
def populate_indicators(
self, dataframe: "DataFrame", metadata: dict[str, Any]
) -> "DataFrame":
import pandas as pd # local import: keep pandas optional at module load
close = dataframe["close"]
# RSI via Wilder's smoothing.
delta = close.diff()
gain = delta.clip(lower=0.0)
loss = (-delta).clip(lower=0.0)
avg_gain = gain.ewm(alpha=1.0 / self.rsi_period, adjust=False).mean()
avg_loss = loss.ewm(alpha=1.0 / self.rsi_period, adjust=False).mean()
rs = avg_gain / avg_loss.replace(0.0, pd.NA)
dataframe["rsi"] = 100.0 - (100.0 / (1.0 + rs))
dataframe["ema_fast"] = close.ewm(
span=self.ema_fast_period, adjust=False
).mean()
dataframe["ema_slow"] = close.ewm(
span=self.ema_slow_period, adjust=False
).mean()
return dataframe
def populate_entry_trend(
self, dataframe: "DataFrame", metadata: dict[str, Any]
) -> "DataFrame":
long_cond = (dataframe["rsi"] < self.rsi_oversold) & (
dataframe["ema_fast"] > dataframe["ema_slow"]
)
dataframe.loc[long_cond, "enter_long"] = 1
dataframe.loc[long_cond, "enter_tag"] = "rsi_oversold_ema_up"
return dataframe
def populate_exit_trend(
self, dataframe: "DataFrame", metadata: dict[str, Any]
) -> "DataFrame":
exit_cond = dataframe["rsi"] > self.rsi_overbought
dataframe.loc[exit_cond, "exit_long"] = 1
dataframe.loc[exit_cond, "exit_tag"] = "rsi_overbought"
return dataframe
# --- optional overrides --------------------------------------------------
def custom_stoploss(
self, ctx: TradeContext, current_profit_pct: float
) -> float | None:
"""High-water-mark trailing stop, armed only after the trade is in profit.
Uses ``Position.best_price`` (updated each tick by the engine) so the
stop ratchets in the favorable direction and never loosens. Matches
freqtrade's behavior of deriving the trailing stop from ``max_rate``.
"""
if current_profit_pct <= 0:
return None
pos = ctx.position
if not pos.is_open or pos.entry_price <= 0:
return None
# best_price is 0 until the engine starts tracking; fall back to current price.
if ctx.side == Side.LONG:
anchor = pos.best_price if pos.best_price > 0 else ctx.current_price
return anchor * (1.0 - self.trailing_stop_pct)
anchor = pos.best_price if pos.best_price > 0 else ctx.current_price
return anchor * (1.0 + self.trailing_stop_pct)
def confirm_trade_entry(
self,
ctx: TradeContext,
proposed_qty: float,
proposed_price: float,
) -> bool:
if ctx.signal is None:
return True
if ctx.signal.strength < self.min_signal_strength:
return False
# In strong counter-trend regimes, avoid adding to the dominant side.
if ctx.side == Side.LONG and ctx.signal.regime == TrendRegime.STRONG_BEAR:
return False
if ctx.side == Side.SHORT and ctx.signal.regime == TrendRegime.STRONG_BULL:
return False
return True
__all__ = [
"TradeContext",
"IStrategy",
"StrategyRunner",
"DefaultStrategy",
"ExampleTrendStrategy",
]