bbstrader.trading

Live execution and strategy orchestration: manages live sessions, coordinates signals from strategies and risk from models, and dispatches execution via bbstrader.metatrader.

trading

Overview

The Trading Module is responsible for the execution of trading strategies. It provides a structured framework for implementing and managing trading strategies, from signal generation to order execution. This module is designed to be flexible and extensible, allowing for the customization of trading logic and integration with various execution handlers.

Features

• Strategy Execution Framework: Defines a clear structure for creating and executing trading strategies.
• Signal Generation: Supports the generation of trading signals based on market data and strategy logic.
• Order Management: Manages the creation and execution of orders based on generated signals.
• Extensibility: Allows for the implementation of custom strategies and execution handlers.

Components

• Execution: Handles the execution of trades, with a base class for creating custom execution handlers.
• Strategy: Defines the core logic of the trading strategy, including signal generation and order creation.
• Utils: Provides utility functions to support the trading process.

Notes

This module can be used in both backtesting and live trading environments by swapping out the execution handler.

Mt5ExecutionEngine

Mt5ExecutionEngine(symbol_list: List[str], trades_instances: Dict[str, Trade], strategy_cls: Strategy | LiveStrategy, /, mm: bool = True, auto_trade: bool = True, prompt_callback: Callable = None, multithread: bool = False, shutdown_event: Event = None, optimizer: str = 'equal', trail: bool = True, stop_trail: Optional[int] = None, trail_after_points: int | str = None, be_plus_points: Optional[int] = None, show_positions_orders: bool = False, iter_time: int | float = 5, use_trade_time: bool = True, period: Literal['24/7', 'day', 'week', 'month'] = 'month', period_end_action: Literal['break', 'sleep'] = 'sleep', closing_pnl: Optional[float] = None, trading_days: Optional[List[str]] = None, comment: Optional[str] = None, **kwargs)

The Mt5ExecutionEngine class serves as the central hub for executing your trading strategies within the bbstrader framework. It orchestrates the entire trading process, ensuring seamless interaction between your strategies, market data, and your chosen trading platform.

Key Features

• Strategy Execution: The Mt5ExecutionEngine is responsible for running your strategy, retrieving signals, and executing trades based on those signals.
• Time Management: You can define a specific time frame for your trades and set the frequency with which the engine checks for signals and manages trades.
• Trade Period Control: Define whether your strategy runs for a day, a week, or a month, allowing for flexible trading durations.
• Money Management: The engine supports optional money management features, allowing you to control risk and optimize your trading performance.
• Trading Day Configuration: You can customize the days of the week your strategy will execute, providing granular control over your trading schedule.
• Platform Integration: The Mt5ExecutionEngine is currently designed to work with MT5.

Examples

from bbstrader.metatrader import create_trade_instance from bbstrader.trading.execution import Mt5ExecutionEngine from examples.strategies import StockIndexSTBOTrading from bbstrader.config import config_logger

if name == 'main': logger = config_logger(index_trade.log, console_log=True) # Define symbols ndx = '[NQ100]' spx = '[SP500]' dji = '[DJI30]' dax = 'GERMANY40'

symbol_list = [spx, dax, dji,  ndx]

trade_kwargs = {

... 'expert_id': 5134, ... 'version': 2.0, ... 'time_frame': '15m', ... 'var_level': 0.99, ... 'start_time': '8:30', ... 'finishing_time': '19:30', ... 'ending_time': '21:30', ... 'max_risk': 5.0, ... 'daily_risk': 0.10, ... 'pchange_sl': 1.5, ... 'rr': 3.0, ... 'logger': logger ... } strategy_kwargs = { ... 'max_trades': {ndx: 3, spx: 3, dji: 3, dax: 3}, ... 'expected_returns': {ndx: 1.5, spx: 1.5, dji: 1.0, dax: 1.0}, ... 'strategy_name': 'SISTBO', ... 'logger': logger, ... 'expert_id': 5134 ... } trades_instances = create_trade_instance( ... symbol_list, trade_kwargs, ... logger=logger, ... )

engine = Mt5ExecutionEngine(

... symbol_list, ... trades_instances, ... StockIndexCFDTrading, ... time_frame='15m', ... iter_time=5, ... mm=True, ... period='week', ... comment='bbs_SISTBO_@2.0', ... **strategy_kwargs ... ) engine.run()

Parameters:

Name	Type	Description	Default
symbol_list		List of symbols to trade	required
trades_instances		Dictionary of Trade instances	required
strategy_cls		Strategy class to use for trading	required
mm		Enable Money Management. Defaults to True.	required
optimizer		Risk management optimizer. Defaults to 'equal'. See bbstrader.models.optimization module for more information.	required
auto_trade		If set to true, when signal are generated by the strategy class, the Execution engine will automaticaly open position in other whise it will prompt the user for confimation.	required
prompt_callback		Callback function to prompt the user for confirmation. This is useful when integrating with GUI applications. multithread : If True, use a thread pool to process signals in parallel. If False, process them sequentially. Set this to True only if the engine is running in a separate process. Default to False.	required
shutdown_event		Use to terminate the copy process when runs in a custum environment like web App or GUI.	required
show_positions_orders		Print open positions and orders. Defaults to False.	required
iter_time		Interval to check for signals and mm. Defaults to 5.	required
use_trade_time		Open trades after the time is completed. Defaults to True.	required
period		Period to trade ("24/7", "day", "week", "month"). Defaults to 'week'.	required
period_end_action		Action to take at the end of the period ("break", "sleep"). Defaults to 'break', this only applies when period is 'day', 'week'.	required
closing_pnl		Minimum profit in percentage of target profit to close positions. Defaults to -0.001.	required
trading_days		Trading days in a week. Defaults to monday to friday.	required
comment	Optional[str]	Comment for trades. Defaults to None.	None
**kwargs		Additional keyword arguments _ time_frame : Time frame to trade. Defaults to '15m'. - strategy_name (Optional[str]): Strategy name. Defaults to None. - max_trades (Dict[str, int]): Maximum trades per symbol. Defaults to None. - notify (bool): Enable notifications. Defaults to False. - telegram (bool): Enable telegram notifications. Defaults to False. - bot_token (str): Telegram bot token. Defaults to None. - chat_id (Union[int, str, List] ): Telegram chat id. Defaults to None. - MT5 connection arguments.	{}

Note

1. For trail , stop_trail , trail_after_points , be_plus_points see bbstrader.metatrader.trade.Trade.break_even() .

2. All Strategies must inherit from bbstrader.btengine.strategy.MT5Strategy class and have a calculate_signals method that returns a List of bbstrader.metatrader.trade.TradingSignal.

3. All strategies must have the following arguments in their __init__ method:

   • bars (DataHandler): DataHandler instance default to None
   • events (Queue): Queue instance default to None
   • symbol_list (List[str]): List of symbols to trade can be none for backtesting
   • mode (str): Mode of the strategy. Must be either 'live' or 'backtest'
   • **kwargs: Additional keyword arguments The keyword arguments are all the additional arguments passed to the Mt5ExecutionEngine class, the Strategy class, the DataHandler class, the Portfolio class and the ExecutionHandler class.
   • The bars and events arguments are used for backtesting only.

4. All strategies must generate signals for backtesting and live trading. See the bbstrader.trading.strategies module for more information on how to create custom strategies. See bbstrader.metatrader.account.check_mt5_connection() for more details on how to connect to MT5 terminal.

Source code in src/bbstrader/trading/execution.py

def __init__(
    self,
    symbol_list: List[str],
    trades_instances: Dict[str, Trade],
    strategy_cls: Strategy | LiveStrategy,
    /,
    mm: bool = True,
    auto_trade: bool = True,
    prompt_callback: Callable = None,
    multithread: bool = False,
    shutdown_event: Event = None,
    optimizer: str = "equal",
    trail: bool = True,
    stop_trail: Optional[int] = None,
    trail_after_points: int | str = None,
    be_plus_points: Optional[int] = None,
    show_positions_orders: bool = False,
    iter_time: int | float = 5,
    use_trade_time: bool = True,
    period: Literal["24/7", "day", "week", "month"] = "month",
    period_end_action: Literal["break", "sleep"] = "sleep",
    closing_pnl: Optional[float] = None,
    trading_days: Optional[List[str]] = None,
    comment: Optional[str] = None,
    **kwargs,
):
    """
    Args:
        symbol_list : List of symbols to trade
        trades_instances : Dictionary of Trade instances
        strategy_cls : Strategy class to use for trading
        mm : Enable Money Management. Defaults to True.
        optimizer : Risk management optimizer. Defaults to 'equal'.
            See `bbstrader.models.optimization` module for more information.
        auto_trade :  If set to true, when signal are generated by the strategy class,
            the Execution engine will automaticaly open position in other whise it will prompt
            the user for confimation.
        prompt_callback : Callback function to prompt the user for confirmation.
            This is useful when integrating with GUI applications.
         multithread : If True, use a thread pool to process signals in parallel.
            If False, process them sequentially. Set this to True only if the engine
            is running in a separate process. Default to False.
        shutdown_event : Use to terminate the copy process when runs in a custum environment like web App or GUI.
        show_positions_orders : Print open positions and orders. Defaults to False.
        iter_time : Interval to check for signals and `mm`. Defaults to 5.
        use_trade_time : Open trades after the time is completed. Defaults to True.
        period : Period to trade ("24/7", "day", "week", "month"). Defaults to 'week'.
        period_end_action : Action to take at the end of the period ("break", "sleep"). Defaults to 'break',
            this only applies when period is 'day', 'week'.
        closing_pnl : Minimum profit in percentage of target profit to close positions. Defaults to -0.001.
        trading_days : Trading days in a week. Defaults to monday to friday.
        comment: Comment for trades. Defaults to None.
        **kwargs: Additional keyword arguments
            _ time_frame : Time frame to trade. Defaults to '15m'.
            - strategy_name (Optional[str]): Strategy name. Defaults to None.
            - max_trades (Dict[str, int]): Maximum trades per symbol. Defaults to None.
            - notify (bool): Enable notifications. Defaults to False.
            - telegram (bool): Enable telegram notifications. Defaults to False.
            - bot_token (str): Telegram bot token. Defaults to None.
            - chat_id (Union[int, str, List] ): Telegram chat id. Defaults to None.
            - MT5 connection arguments.

    Note:
        1. For `trail` , `stop_trail` , `trail_after_points` , `be_plus_points` see `bbstrader.metatrader.trade.Trade.break_even()` .
        2. All Strategies must inherit from `bbstrader.btengine.strategy.MT5Strategy` class
        and have a `calculate_signals` method that returns a List of ``bbstrader.metatrader.trade.TradingSignal``.

        3. All strategies must have the following arguments in their `__init__` method:
            - bars (DataHandler): DataHandler instance default to None
            - events (Queue): Queue instance default to None
            - symbol_list (List[str]): List of symbols to trade can be none for backtesting
            - mode (str): Mode of the strategy. Must be either 'live' or 'backtest'
            - **kwargs: Additional keyword arguments
                The keyword arguments are all the additional arguments passed to the `Mt5ExecutionEngine` class,
                the `Strategy` class, the `DataHandler` class, the `Portfolio` class and the `ExecutionHandler` class.
            - The `bars` and `events` arguments are used for backtesting only.

        4. All strategies must generate signals for backtesting and live trading.
        See the `bbstrader.trading.strategies` module for more information on how to create custom strategies.
        See `bbstrader.metatrader.account.check_mt5_connection()` for more details on how to connect to MT5 terminal.
    """
    self.symbols = symbol_list.copy()
    self.trades_instances = trades_instances
    self.strategy_cls = strategy_cls
    self.mm = mm
    self.auto_trade = auto_trade
    self.prompt_callback = prompt_callback
    self.multithread = multithread
    self.optimizer = optimizer
    self.trail = trail
    self.stop_trail = stop_trail
    self.trail_after_points = trail_after_points
    self.be_plus_points = be_plus_points
    self.show_positions_orders = show_positions_orders
    self.iter_time = iter_time
    self.use_trade_time = use_trade_time
    self.period = period.strip()
    self.period_end_action = period_end_action
    self.closing_pnl = closing_pnl
    self.comment = comment
    self.kwargs = kwargs

    self.time_intervals = 0
    self.time_frame = kwargs.get("time_frame", "15m")
    self.trade_time = _TF_MAPPING[self.time_frame]

    self.long_market = {symbol: False for symbol in self.symbols}
    self.short_market = {symbol: False for symbol in self.symbols}

    self._initialize_engine(**kwargs)
    self.strategy = self._init_strategy(**kwargs)
    self.shutdown_event = (
        shutdown_event if shutdown_event is not None else mp.Event()
    )
    self._running = True

stop

stop()

Stops the execution engine.

Source code in src/bbstrader/trading/execution.py

def stop(self):
    """Stops the execution engine."""
    if self._running:
        logger.info(
            f"Stopping Execution Engine for {self.STRATEGY} STRATEGY on {self.ACCOUNT} Account"
        )
        self._running = False
        self.shutdown_event.set()
    logger.info("Execution Engine stopped successfully.")

LiveStrategy

LiveStrategy(symbol_list: List[str], **kwargs: Any)

Bases: BaseStrategy

Strategy implementation for Live Trading. Relies on the Account class for state (orders, positions, cash) and Rates for data.

Initialize the LiveStrategy object.

Parameters:

Name	Type	Description	Default
symbol_list		The list of symbols for the strategy.	required
**kwargs		Additional keyword arguments for other classes (e.g, Portfolio, ExecutionHandler). - max_trades : The maximum number of trades allowed per symbol. - time_frame : The time frame for the strategy. - logger : The logger object for the strategy.	required

Source code in src/bbstrader/trading/strategy.py

def __init__(
    self,
    symbol_list: List[str],
    **kwargs: Any,
) -> None:
    """
    Initialize the `LiveStrategy` object.

    Args:
        symbol_list : The list of symbols for the strategy.
        **kwargs : Additional keyword arguments for other classes (e.g, Portfolio, ExecutionHandler).
            - max_trades : The maximum number of trades allowed per symbol.
            - time_frame : The time frame for the strategy.
            - logger : The logger object for the strategy.
    """
    super().__init__(symbol_list, **kwargs)
    self.mode = TradingMode.LIVE

account property

account: Account

Create or access the MT5 Account.

orders property

orders: List[TradeOrder]

Returns active orders from the Broker.

positions property

positions: List[Any]

Returns open positions from the Broker.

signal

signal(signal: int, symbol: str, sl: float = None, tp: float = None) -> TradeSignal

Generate a TradeSignal object based on the signal value.

Parameters

signal : int An integer value representing the signal type: * 0: BUY * 1: SELL * 2: EXIT_LONG * 3: EXIT_SHORT * 4: EXIT_ALL_POSITIONS * 5: EXIT_ALL_ORDERS * 6: EXIT_STOP * 7: EXIT_LIMIT symbol : str The symbol for the trade.

Returns

TradeSignal A TradeSignal object representing the trade signal.

Raises

ValueError If the signal value is not between 0 and 7.

Notes

This generates only common signals. For more complex signals, use generate_signal directly.

Source code in src/bbstrader/trading/strategy.py

def signal(
    self, signal: int, symbol: str, sl: float = None, tp: float = None
) -> TradeSignal:
    """
    Generate a ``TradeSignal`` object based on the signal value.

    Parameters
    ----------
    signal : int
        An integer value representing the signal type:
        * 0: BUY
        * 1: SELL
        * 2: EXIT_LONG
        * 3: EXIT_SHORT
        * 4: EXIT_ALL_POSITIONS
        * 5: EXIT_ALL_ORDERS
        * 6: EXIT_STOP
        * 7: EXIT_LIMIT
    symbol : str
        The symbol for the trade.

    Returns
    -------
    TradeSignal
        A ``TradeSignal`` object representing the trade signal.

    Raises
    ------
    ValueError
        If the signal value is not between 0 and 7.

    Notes
    -----
    This generates only common signals. For more complex signals, use
    ``generate_signal`` directly.
    """
    signal_id = getattr(self, "id", getattr(self, "ID", None))
    if signal_id is None:
        raise ValueError("Strategy ID not set")

    action_map = {
        SignalType.BUY: TradeAction.BUY,
        SignalType.SELL: TradeAction.SELL,
        SignalType.EXIT_LONG: TradeAction.EXIT_LONG,
        SignalType.EXIT_SHORT: TradeAction.EXIT_SHORT,
        SignalType.EXIT_ALL_POSITIONS: TradeAction.EXIT_ALL_POSITIONS,
        SignalType.EXIT_ALL_ORDERS: TradeAction.EXIT_ALL_ORDERS,
        SignalType.EXIT_STOP: TradeAction.EXIT_STOP,
        SignalType.EXIT_LIMIT: TradeAction.EXIT_LIMIT,
    }

    try:
        action = action_map[SignalType(signal)]
    except (ValueError, KeyError):
        raise ValueError(f"Invalid signal value: {signal}")
    kwargs = (
        {"sl": sl, "tp": tp}
        if action in (TradeAction.BUY, TradeAction.SELL)
        else {}
    )

    return generate_signal(signal_id, symbol, action, **kwargs)

ispositions

ispositions(symbol: str, strategy_id: int, position: int, max_trades: int, one_true: bool = False) -> bool

This function is use for live trading to check if there are open positions for a given symbol and strategy. It is used to prevent opening more trades than the maximum allowed trades per symbol.

Parameters:

Name	Type	Description	Default
symbol		The symbol for the trade.	required
strategy_id		The unique identifier for the strategy.	required
position		The position type (1: short, 0: long).	required
max_trades		The maximum number of trades allowed per symbol.	required
one_true		If True, return True if there is at least one open position.	required
account		The bbstrader.metatrader.Account object for the strategy.	required

Returns:

Name	Type	Description
bool	bool	True if there are open positions, False otherwise

Source code in src/bbstrader/trading/strategy.py

def ispositions(
    self,
    symbol: str,
    strategy_id: int,
    position: int,
    max_trades: int,
    one_true: bool = False,
) -> bool:
    """
    This function is use for live trading to check if there are open positions
    for a given symbol and strategy. It is used to prevent opening more trades
    than the maximum allowed trades per symbol.

    Args:
        symbol : The symbol for the trade.
        strategy_id : The unique identifier for the strategy.
        position : The position type (1: short, 0: long).
        max_trades : The maximum number of trades allowed per symbol.
        one_true : If True, return True if there is at least one open position.
        account : The `bbstrader.metatrader.Account` object for the strategy.

    Returns:
        bool : True if there are open positions, False otherwise
    """
    positions = self.account.get_positions(symbol=symbol)
    if positions is not None:
        open_positions = [
            pos.ticket
            for pos in positions
            if pos.type == position and pos.magic == strategy_id
        ]
        if one_true:
            return len(open_positions) in range(1, max_trades + 1)
        return len(open_positions) >= max_trades
    return False

get_positions_prices

get_positions_prices(symbol: str, strategy_id: int, position: int) -> np.ndarray

Get the buy or sell prices for open positions of a given symbol and strategy.

Parameters:

Name	Type	Description	Default
symbol		The symbol for the trade.	required
strategy_id		The unique identifier for the strategy.	required
position		The position type (1: short, 0: long).	required
account		The bbstrader.metatrader.Account object for the strategy.	required

Returns:

Name	Type	Description
prices	ndarray	numpy array of buy or sell prices for open positions if any or an empty array.

Source code in src/bbstrader/trading/strategy.py

def get_positions_prices(
    self,
    symbol: str,
    strategy_id: int,
    position: int,
) -> np.ndarray:
    """
    Get the buy or sell prices for open positions of a given symbol and strategy.

    Args:
        symbol : The symbol for the trade.
        strategy_id : The unique identifier for the strategy.
        position : The position type (1: short, 0: long).
        account : The `bbstrader.metatrader.Account` object for the strategy.

    Returns:
        prices : numpy array of buy or sell prices for open positions if any or an empty array.
    """
    positions = self.account.get_positions(symbol=symbol)
    if positions is not None:
        prices = np.array(
            [
                pos.price_open
                for pos in positions
                if pos.type == position and pos.magic == strategy_id
            ]
        )
        return prices
    return np.array([])

get_active_orders

get_active_orders(symbol: str, strategy_id: int, order_type: Optional[int] = None) -> List[TradeOrder]

Get the active orders for a given symbol and strategy.

Parameters:

Name	Type	Description	Default
symbol		The symbol for the trade.	required
strategy_id		The unique identifier for the strategy.	required
order_type		The type of order to filter by (optional): "BUY_LIMIT": 2 "SELL_LIMIT": 3 "BUY_STOP": 4 "SELL_STOP": 5 "BUY_STOP_LIMIT": 6 "SELL_STOP_LIMIT": 7	required

Returns:

Type	Description
List[TradeOrder]	List[TradeOrder] : A list of active orders for the given symbol and strategy.

Source code in src/bbstrader/trading/strategy.py

def get_active_orders(
    self, symbol: str, strategy_id: int, order_type: Optional[int] = None
) -> List[TradeOrder]:
    """
    Get the active orders for a given symbol and strategy.

    Args:
        symbol : The symbol for the trade.
        strategy_id : The unique identifier for the strategy.
        order_type : The type of order to filter by (optional):
                "BUY_LIMIT": 2
                "SELL_LIMIT": 3
                "BUY_STOP": 4
                "SELL_STOP": 5
                "BUY_STOP_LIMIT": 6
                "SELL_STOP_LIMIT": 7

    Returns:
        List[TradeOrder] : A list of active orders for the given symbol and strategy.
    """
    all_orders = self.orders
    orders = [
        o
        for o in all_orders
        if isinstance(o, TradeOrder)
        and o.symbol == symbol
        and o.magic == strategy_id
    ]
    if order_type is not None and len(orders) > 0:
        orders = [o for o in orders if o.type == order_type]
    return orders

exit_positions

exit_positions(position: int, prices: NDArray, asset: str, th: float = 0.01) -> bool

Logic to determine if positions should be exited based on threshold.

Source code in src/bbstrader/trading/strategy.py

def exit_positions(
    self, position: int, prices: np.typing.NDArray, asset: str, th: float = 0.01
) -> bool:
    """Logic to determine if positions should be exited based on threshold."""
    if len(prices) == 0:
        return False
    tick_info = self.account.get_tick_info(asset)
    if tick_info is None:
        return False
    bid, ask = tick_info.bid, tick_info.ask
    price = None
    if len(prices) == 1:
        price = prices[0]
    elif len(prices) in range(2, self.max_trades[asset] + 1):
        price = np.mean(prices)

    if price is not None:
        if position == 0:  # Long exit check
            return self.calculate_pct_change(ask, price) >= th
        elif position == 1:  # Short exit check
            return self.calculate_pct_change(bid, price) <= -th
    return False

send_trade_report

send_trade_report(perf_analyzer: Callable, **kwargs: Any) -> None

Generates and sends a trade report message containing performance metrics for the current strategy. This method retrieves the trade history for the current account, filters it by the strategy's ID, computes performance metrics using the provided perf_analyzer callable, and formats the results into a message. The message includes account information, strategy details, a timestamp, and performance metrics. The message is then sent via Telegram using the specified bot token and chat ID.

Parameters:

Name	Type	Description	Default
perf_analyzer	Callable	A function or callable object that takes the filtered trade history (as a DataFrame) and additional keyword arguments, and returns a DataFrame of performance metrics.	required
**kwargs	Any	Additional keyword arguments, which may include - Any other param requires by perf_analyzer	{}

Source code in src/bbstrader/trading/strategy.py

def send_trade_report(self, perf_analyzer: Callable, **kwargs: Any) -> None:
    """
    Generates and sends a trade report message containing performance metrics for the current strategy.
    This method retrieves the trade history for the current account, filters it by the strategy's ID,
    computes performance metrics using the provided `perf_analyzer` callable, and formats the results
    into a message. The message includes account information, strategy details, a timestamp, and
    performance metrics. The message is then sent via Telegram using the specified bot token and chat ID.

    Args:
        perf_analyzer (Callable): A function or callable object that takes the filtered trade history
            (as a DataFrame) and additional keyword arguments, and returns a DataFrame of performance metrics.
        **kwargs: Additional keyword arguments, which may include
            - Any other param requires by ``perf_analyzer``
    """
    from bbstrader.trading.utils import send_message

    history = self.account.get_trades_history()
    if history is None or history.empty:
        self.logger.warning("No trades found on this account.")
        return

    ID = getattr(self, "id", None) or getattr(self, "ID")
    history = history[history["magic"] == ID]
    performance = perf_analyzer(history, **kwargs)
    if performance.empty:
        self.logger.warning("No trades found for the current strategy.")
        return

    account_name = self.kwargs.get("account", "MT5 Account")
    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

    header = (
        f"TRADE REPORT\n\n"
        f"ACCOUNT: {account_name}\n"
        f"STRATEGY: {self.NAME}\n"
        f"ID: {ID}\n"
        f"DESCRIPTION: {self.DESCRIPTION}\n"
        f"TIMESTAMP: {timestamp}\n\n"
        f"📊 PERFORMANCE:\n"
    )
    metrics = performance.iloc[0].to_dict()

    lines = []
    for key, value in metrics.items():
        if isinstance(value, float):
            value = round(value, 4)
        lines.append(f"{key:<15}: {value}")

    performance_str = "\n".join(lines)
    message = f"{header}{performance_str}"

    send_message(
        message=message,
        telegram=True,
        token=self.kwargs.get("bot_token"),
        chat_id=self.kwargs.get("chat_id"),
    )

RunMt5Engine

RunMt5Engine(account_id: str, **kwargs)

Start an MT5 execution engine for a given account.

Parameters

account_id : str Account ID to run the execution engine on.

**kwargs : dict Additional keyword arguments. Possible keys include:

* symbol_list : list
    List of symbols to trade.
* trades_instances : dict
    Dictionary of Trade instances.
* strategy_cls : class
    Strategy class to use for trading.

Returns

None Initializes and runs the MT5 execution engine.

Source code in src/bbstrader/trading/execution.py

def RunMt5Engine(account_id: str, **kwargs):
    """
    Start an MT5 execution engine for a given account.

    Parameters
    ----------
    account_id : str
        Account ID to run the execution engine on.

    **kwargs : dict
        Additional keyword arguments. Possible keys include:

        * symbol_list : list
            List of symbols to trade.
        * trades_instances : dict
            Dictionary of Trade instances.
        * strategy_cls : class
            Strategy class to use for trading.

    Returns
    -------
    None
        Initializes and runs the MT5 execution engine.
    """
    log.info(f"Starting execution engine for {account_id}")

    symbol_list = kwargs.pop("symbol_list")
    trades_instances = kwargs.pop("trades_instances")
    strategy_cls = kwargs.pop("strategy_cls")

    if symbol_list is None or trades_instances is None or strategy_cls is None:
        log.error(f"Missing required arguments for account {account_id}")
        raise ValueError(f"Missing required arguments for account {account_id}")

    try:
        engine = Mt5ExecutionEngine(
            symbol_list, trades_instances, strategy_cls, **kwargs
        )
        engine.run()
    except KeyboardInterrupt:
        log.info(f"Execution engine for {account_id} interrupted by user")
        engine.stop()
        sys.exit(0)
    except Exception as e:
        log.exception(f"Error running execution engine for {account_id}: {e}")
    finally:
        log.info(f"Execution for {account_id} completed")

RunMt5Engines

RunMt5Engines(accounts: Dict[str, Dict], start_delay: float = 1.0)

Runs multiple MT5 execution engines in parallel using multiprocessing.

Parameters:

Name	Type	Description	Default
accounts	Dict[str, Dict]	Dictionary of accounts to run the execution engines on. Keys are the account names or IDs and values are the parameters for the execution engine. The parameters are the same as the ones passed to the Mt5ExecutionEngine class.	required
start_delay	float	Delay in seconds between starting the processes. Defaults to 1.0.	1.0

Source code in src/bbstrader/trading/execution.py

def RunMt5Engines(accounts: Dict[str, Dict], start_delay: float = 1.0):
    """Runs multiple MT5 execution engines in parallel using multiprocessing.

    Args:
        accounts: Dictionary of accounts to run the execution engines on.
            Keys are the account names or IDs and values are the parameters for the execution engine.
            The parameters are the same as the ones passed to the `Mt5ExecutionEngine` class.
        start_delay: Delay in seconds between starting the processes. Defaults to 1.0.
    """

    processes = {}

    for account_id, params in accounts.items():
        log.info(f"Starting process for {account_id}")
        params["multithread"] = True
        process = mp.Process(target=RunMt5Engine, args=(account_id,), kwargs=params)
        process.start()
        processes[process] = account_id

        if start_delay:
            time.sleep(start_delay)

    for process, account_id in processes.items():
        process.join()
        log.info(f"Process for {account_id} joined")

execution

Mt5ExecutionEngine

Mt5ExecutionEngine(symbol_list: List[str], trades_instances: Dict[str, Trade], strategy_cls: Strategy | LiveStrategy, /, mm: bool = True, auto_trade: bool = True, prompt_callback: Callable = None, multithread: bool = False, shutdown_event: Event = None, optimizer: str = 'equal', trail: bool = True, stop_trail: Optional[int] = None, trail_after_points: int | str = None, be_plus_points: Optional[int] = None, show_positions_orders: bool = False, iter_time: int | float = 5, use_trade_time: bool = True, period: Literal['24/7', 'day', 'week', 'month'] = 'month', period_end_action: Literal['break', 'sleep'] = 'sleep', closing_pnl: Optional[float] = None, trading_days: Optional[List[str]] = None, comment: Optional[str] = None, **kwargs)

The Mt5ExecutionEngine class serves as the central hub for executing your trading strategies within the bbstrader framework. It orchestrates the entire trading process, ensuring seamless interaction between your strategies, market data, and your chosen trading platform.

Key Features

• Strategy Execution: The Mt5ExecutionEngine is responsible for running your strategy, retrieving signals, and executing trades based on those signals.
• Time Management: You can define a specific time frame for your trades and set the frequency with which the engine checks for signals and manages trades.
• Trade Period Control: Define whether your strategy runs for a day, a week, or a month, allowing for flexible trading durations.
• Money Management: The engine supports optional money management features, allowing you to control risk and optimize your trading performance.
• Trading Day Configuration: You can customize the days of the week your strategy will execute, providing granular control over your trading schedule.
• Platform Integration: The Mt5ExecutionEngine is currently designed to work with MT5.

Examples

from bbstrader.metatrader import create_trade_instance from bbstrader.trading.execution import Mt5ExecutionEngine from examples.strategies import StockIndexSTBOTrading from bbstrader.config import config_logger

if name == 'main': logger = config_logger(index_trade.log, console_log=True) # Define symbols ndx = '[NQ100]' spx = '[SP500]' dji = '[DJI30]' dax = 'GERMANY40'

symbol_list = [spx, dax, dji,  ndx]

trade_kwargs = {

... 'expert_id': 5134, ... 'version': 2.0, ... 'time_frame': '15m', ... 'var_level': 0.99, ... 'start_time': '8:30', ... 'finishing_time': '19:30', ... 'ending_time': '21:30', ... 'max_risk': 5.0, ... 'daily_risk': 0.10, ... 'pchange_sl': 1.5, ... 'rr': 3.0, ... 'logger': logger ... } strategy_kwargs = { ... 'max_trades': {ndx: 3, spx: 3, dji: 3, dax: 3}, ... 'expected_returns': {ndx: 1.5, spx: 1.5, dji: 1.0, dax: 1.0}, ... 'strategy_name': 'SISTBO', ... 'logger': logger, ... 'expert_id': 5134 ... } trades_instances = create_trade_instance( ... symbol_list, trade_kwargs, ... logger=logger, ... )

engine = Mt5ExecutionEngine(

... symbol_list, ... trades_instances, ... StockIndexCFDTrading, ... time_frame='15m', ... iter_time=5, ... mm=True, ... period='week', ... comment='bbs_SISTBO_@2.0', ... **strategy_kwargs ... ) engine.run()

Parameters:

Name	Type	Description	Default
symbol_list		List of symbols to trade	required
trades_instances		Dictionary of Trade instances	required
strategy_cls		Strategy class to use for trading	required
mm		Enable Money Management. Defaults to True.	required
optimizer		Risk management optimizer. Defaults to 'equal'. See bbstrader.models.optimization module for more information.	required
auto_trade		If set to true, when signal are generated by the strategy class, the Execution engine will automaticaly open position in other whise it will prompt the user for confimation.	required
prompt_callback		Callback function to prompt the user for confirmation. This is useful when integrating with GUI applications. multithread : If True, use a thread pool to process signals in parallel. If False, process them sequentially. Set this to True only if the engine is running in a separate process. Default to False.	required
shutdown_event		Use to terminate the copy process when runs in a custum environment like web App or GUI.	required
show_positions_orders		Print open positions and orders. Defaults to False.	required
iter_time		Interval to check for signals and mm. Defaults to 5.	required
use_trade_time		Open trades after the time is completed. Defaults to True.	required
period		Period to trade ("24/7", "day", "week", "month"). Defaults to 'week'.	required
period_end_action		Action to take at the end of the period ("break", "sleep"). Defaults to 'break', this only applies when period is 'day', 'week'.	required
closing_pnl		Minimum profit in percentage of target profit to close positions. Defaults to -0.001.	required
trading_days		Trading days in a week. Defaults to monday to friday.	required
comment	Optional[str]	Comment for trades. Defaults to None.	None
**kwargs		Additional keyword arguments _ time_frame : Time frame to trade. Defaults to '15m'. - strategy_name (Optional[str]): Strategy name. Defaults to None. - max_trades (Dict[str, int]): Maximum trades per symbol. Defaults to None. - notify (bool): Enable notifications. Defaults to False. - telegram (bool): Enable telegram notifications. Defaults to False. - bot_token (str): Telegram bot token. Defaults to None. - chat_id (Union[int, str, List] ): Telegram chat id. Defaults to None. - MT5 connection arguments.	{}

Note

1. For trail , stop_trail , trail_after_points , be_plus_points see bbstrader.metatrader.trade.Trade.break_even() .

2. All Strategies must inherit from bbstrader.btengine.strategy.MT5Strategy class and have a calculate_signals method that returns a List of bbstrader.metatrader.trade.TradingSignal.

3. All strategies must have the following arguments in their __init__ method:

   • bars (DataHandler): DataHandler instance default to None
   • events (Queue): Queue instance default to None
   • symbol_list (List[str]): List of symbols to trade can be none for backtesting
   • mode (str): Mode of the strategy. Must be either 'live' or 'backtest'
   • **kwargs: Additional keyword arguments The keyword arguments are all the additional arguments passed to the Mt5ExecutionEngine class, the Strategy class, the DataHandler class, the Portfolio class and the ExecutionHandler class.
   • The bars and events arguments are used for backtesting only.

4. All strategies must generate signals for backtesting and live trading. See the bbstrader.trading.strategies module for more information on how to create custom strategies. See bbstrader.metatrader.account.check_mt5_connection() for more details on how to connect to MT5 terminal.

Source code in src/bbstrader/trading/execution.py

def __init__(
    self,
    symbol_list: List[str],
    trades_instances: Dict[str, Trade],
    strategy_cls: Strategy | LiveStrategy,
    /,
    mm: bool = True,
    auto_trade: bool = True,
    prompt_callback: Callable = None,
    multithread: bool = False,
    shutdown_event: Event = None,
    optimizer: str = "equal",
    trail: bool = True,
    stop_trail: Optional[int] = None,
    trail_after_points: int | str = None,
    be_plus_points: Optional[int] = None,
    show_positions_orders: bool = False,
    iter_time: int | float = 5,
    use_trade_time: bool = True,
    period: Literal["24/7", "day", "week", "month"] = "month",
    period_end_action: Literal["break", "sleep"] = "sleep",
    closing_pnl: Optional[float] = None,
    trading_days: Optional[List[str]] = None,
    comment: Optional[str] = None,
    **kwargs,
):
    """
    Args:
        symbol_list : List of symbols to trade
        trades_instances : Dictionary of Trade instances
        strategy_cls : Strategy class to use for trading
        mm : Enable Money Management. Defaults to True.
        optimizer : Risk management optimizer. Defaults to 'equal'.
            See `bbstrader.models.optimization` module for more information.
        auto_trade :  If set to true, when signal are generated by the strategy class,
            the Execution engine will automaticaly open position in other whise it will prompt
            the user for confimation.
        prompt_callback : Callback function to prompt the user for confirmation.
            This is useful when integrating with GUI applications.
         multithread : If True, use a thread pool to process signals in parallel.
            If False, process them sequentially. Set this to True only if the engine
            is running in a separate process. Default to False.
        shutdown_event : Use to terminate the copy process when runs in a custum environment like web App or GUI.
        show_positions_orders : Print open positions and orders. Defaults to False.
        iter_time : Interval to check for signals and `mm`. Defaults to 5.
        use_trade_time : Open trades after the time is completed. Defaults to True.
        period : Period to trade ("24/7", "day", "week", "month"). Defaults to 'week'.
        period_end_action : Action to take at the end of the period ("break", "sleep"). Defaults to 'break',
            this only applies when period is 'day', 'week'.
        closing_pnl : Minimum profit in percentage of target profit to close positions. Defaults to -0.001.
        trading_days : Trading days in a week. Defaults to monday to friday.
        comment: Comment for trades. Defaults to None.
        **kwargs: Additional keyword arguments
            _ time_frame : Time frame to trade. Defaults to '15m'.
            - strategy_name (Optional[str]): Strategy name. Defaults to None.
            - max_trades (Dict[str, int]): Maximum trades per symbol. Defaults to None.
            - notify (bool): Enable notifications. Defaults to False.
            - telegram (bool): Enable telegram notifications. Defaults to False.
            - bot_token (str): Telegram bot token. Defaults to None.
            - chat_id (Union[int, str, List] ): Telegram chat id. Defaults to None.
            - MT5 connection arguments.

    Note:
        1. For `trail` , `stop_trail` , `trail_after_points` , `be_plus_points` see `bbstrader.metatrader.trade.Trade.break_even()` .
        2. All Strategies must inherit from `bbstrader.btengine.strategy.MT5Strategy` class
        and have a `calculate_signals` method that returns a List of ``bbstrader.metatrader.trade.TradingSignal``.

        3. All strategies must have the following arguments in their `__init__` method:
            - bars (DataHandler): DataHandler instance default to None
            - events (Queue): Queue instance default to None
            - symbol_list (List[str]): List of symbols to trade can be none for backtesting
            - mode (str): Mode of the strategy. Must be either 'live' or 'backtest'
            - **kwargs: Additional keyword arguments
                The keyword arguments are all the additional arguments passed to the `Mt5ExecutionEngine` class,
                the `Strategy` class, the `DataHandler` class, the `Portfolio` class and the `ExecutionHandler` class.
            - The `bars` and `events` arguments are used for backtesting only.

        4. All strategies must generate signals for backtesting and live trading.
        See the `bbstrader.trading.strategies` module for more information on how to create custom strategies.
        See `bbstrader.metatrader.account.check_mt5_connection()` for more details on how to connect to MT5 terminal.
    """
    self.symbols = symbol_list.copy()
    self.trades_instances = trades_instances
    self.strategy_cls = strategy_cls
    self.mm = mm
    self.auto_trade = auto_trade
    self.prompt_callback = prompt_callback
    self.multithread = multithread
    self.optimizer = optimizer
    self.trail = trail
    self.stop_trail = stop_trail
    self.trail_after_points = trail_after_points
    self.be_plus_points = be_plus_points
    self.show_positions_orders = show_positions_orders
    self.iter_time = iter_time
    self.use_trade_time = use_trade_time
    self.period = period.strip()
    self.period_end_action = period_end_action
    self.closing_pnl = closing_pnl
    self.comment = comment
    self.kwargs = kwargs

    self.time_intervals = 0
    self.time_frame = kwargs.get("time_frame", "15m")
    self.trade_time = _TF_MAPPING[self.time_frame]

    self.long_market = {symbol: False for symbol in self.symbols}
    self.short_market = {symbol: False for symbol in self.symbols}

    self._initialize_engine(**kwargs)
    self.strategy = self._init_strategy(**kwargs)
    self.shutdown_event = (
        shutdown_event if shutdown_event is not None else mp.Event()
    )
    self._running = True

stop

stop()

Stops the execution engine.

Source code in src/bbstrader/trading/execution.py

def stop(self):
    """Stops the execution engine."""
    if self._running:
        logger.info(
            f"Stopping Execution Engine for {self.STRATEGY} STRATEGY on {self.ACCOUNT} Account"
        )
        self._running = False
        self.shutdown_event.set()
    logger.info("Execution Engine stopped successfully.")

RunMt5Engine

RunMt5Engine(account_id: str, **kwargs)

Start an MT5 execution engine for a given account.

Parameters

account_id : str Account ID to run the execution engine on.

**kwargs : dict Additional keyword arguments. Possible keys include:

* symbol_list : list
    List of symbols to trade.
* trades_instances : dict
    Dictionary of Trade instances.
* strategy_cls : class
    Strategy class to use for trading.

Returns

None Initializes and runs the MT5 execution engine.

Source code in src/bbstrader/trading/execution.py

def RunMt5Engine(account_id: str, **kwargs):
    """
    Start an MT5 execution engine for a given account.

    Parameters
    ----------
    account_id : str
        Account ID to run the execution engine on.

    **kwargs : dict
        Additional keyword arguments. Possible keys include:

        * symbol_list : list
            List of symbols to trade.
        * trades_instances : dict
            Dictionary of Trade instances.
        * strategy_cls : class
            Strategy class to use for trading.

    Returns
    -------
    None
        Initializes and runs the MT5 execution engine.
    """
    log.info(f"Starting execution engine for {account_id}")

    symbol_list = kwargs.pop("symbol_list")
    trades_instances = kwargs.pop("trades_instances")
    strategy_cls = kwargs.pop("strategy_cls")

    if symbol_list is None or trades_instances is None or strategy_cls is None:
        log.error(f"Missing required arguments for account {account_id}")
        raise ValueError(f"Missing required arguments for account {account_id}")

    try:
        engine = Mt5ExecutionEngine(
            symbol_list, trades_instances, strategy_cls, **kwargs
        )
        engine.run()
    except KeyboardInterrupt:
        log.info(f"Execution engine for {account_id} interrupted by user")
        engine.stop()
        sys.exit(0)
    except Exception as e:
        log.exception(f"Error running execution engine for {account_id}: {e}")
    finally:
        log.info(f"Execution for {account_id} completed")

RunMt5Engines

RunMt5Engines(accounts: Dict[str, Dict], start_delay: float = 1.0)

Runs multiple MT5 execution engines in parallel using multiprocessing.

Parameters:

Name	Type	Description	Default
accounts	Dict[str, Dict]	Dictionary of accounts to run the execution engines on. Keys are the account names or IDs and values are the parameters for the execution engine. The parameters are the same as the ones passed to the Mt5ExecutionEngine class.	required
start_delay	float	Delay in seconds between starting the processes. Defaults to 1.0.	1.0

Source code in src/bbstrader/trading/execution.py

def RunMt5Engines(accounts: Dict[str, Dict], start_delay: float = 1.0):
    """Runs multiple MT5 execution engines in parallel using multiprocessing.

    Args:
        accounts: Dictionary of accounts to run the execution engines on.
            Keys are the account names or IDs and values are the parameters for the execution engine.
            The parameters are the same as the ones passed to the `Mt5ExecutionEngine` class.
        start_delay: Delay in seconds between starting the processes. Defaults to 1.0.
    """

    processes = {}

    for account_id, params in accounts.items():
        log.info(f"Starting process for {account_id}")
        params["multithread"] = True
        process = mp.Process(target=RunMt5Engine, args=(account_id,), kwargs=params)
        process.start()
        processes[process] = account_id

        if start_delay:
            time.sleep(start_delay)

    for process, account_id in processes.items():
        process.join()
        log.info(f"Process for {account_id} joined")

strategy

LiveStrategy

LiveStrategy(symbol_list: List[str], **kwargs: Any)

Bases: BaseStrategy

Strategy implementation for Live Trading. Relies on the Account class for state (orders, positions, cash) and Rates for data.

Initialize the LiveStrategy object.

Parameters:

Name	Type	Description	Default
symbol_list		The list of symbols for the strategy.	required
**kwargs		Additional keyword arguments for other classes (e.g, Portfolio, ExecutionHandler). - max_trades : The maximum number of trades allowed per symbol. - time_frame : The time frame for the strategy. - logger : The logger object for the strategy.	required

Source code in src/bbstrader/trading/strategy.py

def __init__(
    self,
    symbol_list: List[str],
    **kwargs: Any,
) -> None:
    """
    Initialize the `LiveStrategy` object.

    Args:
        symbol_list : The list of symbols for the strategy.
        **kwargs : Additional keyword arguments for other classes (e.g, Portfolio, ExecutionHandler).
            - max_trades : The maximum number of trades allowed per symbol.
            - time_frame : The time frame for the strategy.
            - logger : The logger object for the strategy.
    """
    super().__init__(symbol_list, **kwargs)
    self.mode = TradingMode.LIVE

account property

account: Account

Create or access the MT5 Account.

orders property

orders: List[TradeOrder]

Returns active orders from the Broker.

positions property

positions: List[Any]

Returns open positions from the Broker.

signal

signal(signal: int, symbol: str, sl: float = None, tp: float = None) -> TradeSignal

Generate a TradeSignal object based on the signal value.

Parameters

signal : int An integer value representing the signal type: * 0: BUY * 1: SELL * 2: EXIT_LONG * 3: EXIT_SHORT * 4: EXIT_ALL_POSITIONS * 5: EXIT_ALL_ORDERS * 6: EXIT_STOP * 7: EXIT_LIMIT symbol : str The symbol for the trade.

Returns

TradeSignal A TradeSignal object representing the trade signal.

Raises

ValueError If the signal value is not between 0 and 7.

Notes

This generates only common signals. For more complex signals, use generate_signal directly.

Source code in src/bbstrader/trading/strategy.py

def signal(
    self, signal: int, symbol: str, sl: float = None, tp: float = None
) -> TradeSignal:
    """
    Generate a ``TradeSignal`` object based on the signal value.

    Parameters
    ----------
    signal : int
        An integer value representing the signal type:
        * 0: BUY
        * 1: SELL
        * 2: EXIT_LONG
        * 3: EXIT_SHORT
        * 4: EXIT_ALL_POSITIONS
        * 5: EXIT_ALL_ORDERS
        * 6: EXIT_STOP
        * 7: EXIT_LIMIT
    symbol : str
        The symbol for the trade.

    Returns
    -------
    TradeSignal
        A ``TradeSignal`` object representing the trade signal.

    Raises
    ------
    ValueError
        If the signal value is not between 0 and 7.

    Notes
    -----
    This generates only common signals. For more complex signals, use
    ``generate_signal`` directly.
    """
    signal_id = getattr(self, "id", getattr(self, "ID", None))
    if signal_id is None:
        raise ValueError("Strategy ID not set")

    action_map = {
        SignalType.BUY: TradeAction.BUY,
        SignalType.SELL: TradeAction.SELL,
        SignalType.EXIT_LONG: TradeAction.EXIT_LONG,
        SignalType.EXIT_SHORT: TradeAction.EXIT_SHORT,
        SignalType.EXIT_ALL_POSITIONS: TradeAction.EXIT_ALL_POSITIONS,
        SignalType.EXIT_ALL_ORDERS: TradeAction.EXIT_ALL_ORDERS,
        SignalType.EXIT_STOP: TradeAction.EXIT_STOP,
        SignalType.EXIT_LIMIT: TradeAction.EXIT_LIMIT,
    }

    try:
        action = action_map[SignalType(signal)]
    except (ValueError, KeyError):
        raise ValueError(f"Invalid signal value: {signal}")
    kwargs = (
        {"sl": sl, "tp": tp}
        if action in (TradeAction.BUY, TradeAction.SELL)
        else {}
    )

    return generate_signal(signal_id, symbol, action, **kwargs)

ispositions

ispositions(symbol: str, strategy_id: int, position: int, max_trades: int, one_true: bool = False) -> bool

This function is use for live trading to check if there are open positions for a given symbol and strategy. It is used to prevent opening more trades than the maximum allowed trades per symbol.

Parameters:

Name	Type	Description	Default
symbol		The symbol for the trade.	required
strategy_id		The unique identifier for the strategy.	required
position		The position type (1: short, 0: long).	required
max_trades		The maximum number of trades allowed per symbol.	required
one_true		If True, return True if there is at least one open position.	required
account		The bbstrader.metatrader.Account object for the strategy.	required

Returns:

Name	Type	Description
bool	bool	True if there are open positions, False otherwise

Source code in src/bbstrader/trading/strategy.py

def ispositions(
    self,
    symbol: str,
    strategy_id: int,
    position: int,
    max_trades: int,
    one_true: bool = False,
) -> bool:
    """
    This function is use for live trading to check if there are open positions
    for a given symbol and strategy. It is used to prevent opening more trades
    than the maximum allowed trades per symbol.

    Args:
        symbol : The symbol for the trade.
        strategy_id : The unique identifier for the strategy.
        position : The position type (1: short, 0: long).
        max_trades : The maximum number of trades allowed per symbol.
        one_true : If True, return True if there is at least one open position.
        account : The `bbstrader.metatrader.Account` object for the strategy.

    Returns:
        bool : True if there are open positions, False otherwise
    """
    positions = self.account.get_positions(symbol=symbol)
    if positions is not None:
        open_positions = [
            pos.ticket
            for pos in positions
            if pos.type == position and pos.magic == strategy_id
        ]
        if one_true:
            return len(open_positions) in range(1, max_trades + 1)
        return len(open_positions) >= max_trades
    return False

get_positions_prices

get_positions_prices(symbol: str, strategy_id: int, position: int) -> np.ndarray

Get the buy or sell prices for open positions of a given symbol and strategy.

Parameters:

Name	Type	Description	Default
symbol		The symbol for the trade.	required
strategy_id		The unique identifier for the strategy.	required
position		The position type (1: short, 0: long).	required
account		The bbstrader.metatrader.Account object for the strategy.	required

Returns:

Name	Type	Description
prices	ndarray	numpy array of buy or sell prices for open positions if any or an empty array.

Source code in src/bbstrader/trading/strategy.py

def get_positions_prices(
    self,
    symbol: str,
    strategy_id: int,
    position: int,
) -> np.ndarray:
    """
    Get the buy or sell prices for open positions of a given symbol and strategy.

    Args:
        symbol : The symbol for the trade.
        strategy_id : The unique identifier for the strategy.
        position : The position type (1: short, 0: long).
        account : The `bbstrader.metatrader.Account` object for the strategy.

    Returns:
        prices : numpy array of buy or sell prices for open positions if any or an empty array.
    """
    positions = self.account.get_positions(symbol=symbol)
    if positions is not None:
        prices = np.array(
            [
                pos.price_open
                for pos in positions
                if pos.type == position and pos.magic == strategy_id
            ]
        )
        return prices
    return np.array([])

get_active_orders

get_active_orders(symbol: str, strategy_id: int, order_type: Optional[int] = None) -> List[TradeOrder]

Get the active orders for a given symbol and strategy.

Parameters:

Name	Type	Description	Default
symbol		The symbol for the trade.	required
strategy_id		The unique identifier for the strategy.	required
order_type		The type of order to filter by (optional): "BUY_LIMIT": 2 "SELL_LIMIT": 3 "BUY_STOP": 4 "SELL_STOP": 5 "BUY_STOP_LIMIT": 6 "SELL_STOP_LIMIT": 7	required

Returns:

Type	Description
List[TradeOrder]	List[TradeOrder] : A list of active orders for the given symbol and strategy.

Source code in src/bbstrader/trading/strategy.py

def get_active_orders(
    self, symbol: str, strategy_id: int, order_type: Optional[int] = None
) -> List[TradeOrder]:
    """
    Get the active orders for a given symbol and strategy.

    Args:
        symbol : The symbol for the trade.
        strategy_id : The unique identifier for the strategy.
        order_type : The type of order to filter by (optional):
                "BUY_LIMIT": 2
                "SELL_LIMIT": 3
                "BUY_STOP": 4
                "SELL_STOP": 5
                "BUY_STOP_LIMIT": 6
                "SELL_STOP_LIMIT": 7

    Returns:
        List[TradeOrder] : A list of active orders for the given symbol and strategy.
    """
    all_orders = self.orders
    orders = [
        o
        for o in all_orders
        if isinstance(o, TradeOrder)
        and o.symbol == symbol
        and o.magic == strategy_id
    ]
    if order_type is not None and len(orders) > 0:
        orders = [o for o in orders if o.type == order_type]
    return orders

exit_positions

exit_positions(position: int, prices: NDArray, asset: str, th: float = 0.01) -> bool

Logic to determine if positions should be exited based on threshold.

Source code in src/bbstrader/trading/strategy.py

def exit_positions(
    self, position: int, prices: np.typing.NDArray, asset: str, th: float = 0.01
) -> bool:
    """Logic to determine if positions should be exited based on threshold."""
    if len(prices) == 0:
        return False
    tick_info = self.account.get_tick_info(asset)
    if tick_info is None:
        return False
    bid, ask = tick_info.bid, tick_info.ask
    price = None
    if len(prices) == 1:
        price = prices[0]
    elif len(prices) in range(2, self.max_trades[asset] + 1):
        price = np.mean(prices)

    if price is not None:
        if position == 0:  # Long exit check
            return self.calculate_pct_change(ask, price) >= th
        elif position == 1:  # Short exit check
            return self.calculate_pct_change(bid, price) <= -th
    return False

send_trade_report

send_trade_report(perf_analyzer: Callable, **kwargs: Any) -> None

Generates and sends a trade report message containing performance metrics for the current strategy. This method retrieves the trade history for the current account, filters it by the strategy's ID, computes performance metrics using the provided perf_analyzer callable, and formats the results into a message. The message includes account information, strategy details, a timestamp, and performance metrics. The message is then sent via Telegram using the specified bot token and chat ID.

Parameters:

Name	Type	Description	Default
perf_analyzer	Callable	A function or callable object that takes the filtered trade history (as a DataFrame) and additional keyword arguments, and returns a DataFrame of performance metrics.	required
**kwargs	Any	Additional keyword arguments, which may include - Any other param requires by perf_analyzer	{}

Source code in src/bbstrader/trading/strategy.py

def send_trade_report(self, perf_analyzer: Callable, **kwargs: Any) -> None:
    """
    Generates and sends a trade report message containing performance metrics for the current strategy.
    This method retrieves the trade history for the current account, filters it by the strategy's ID,
    computes performance metrics using the provided `perf_analyzer` callable, and formats the results
    into a message. The message includes account information, strategy details, a timestamp, and
    performance metrics. The message is then sent via Telegram using the specified bot token and chat ID.

    Args:
        perf_analyzer (Callable): A function or callable object that takes the filtered trade history
            (as a DataFrame) and additional keyword arguments, and returns a DataFrame of performance metrics.
        **kwargs: Additional keyword arguments, which may include
            - Any other param requires by ``perf_analyzer``
    """
    from bbstrader.trading.utils import send_message

    history = self.account.get_trades_history()
    if history is None or history.empty:
        self.logger.warning("No trades found on this account.")
        return

    ID = getattr(self, "id", None) or getattr(self, "ID")
    history = history[history["magic"] == ID]
    performance = perf_analyzer(history, **kwargs)
    if performance.empty:
        self.logger.warning("No trades found for the current strategy.")
        return

    account_name = self.kwargs.get("account", "MT5 Account")
    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

    header = (
        f"TRADE REPORT\n\n"
        f"ACCOUNT: {account_name}\n"
        f"STRATEGY: {self.NAME}\n"
        f"ID: {ID}\n"
        f"DESCRIPTION: {self.DESCRIPTION}\n"
        f"TIMESTAMP: {timestamp}\n\n"
        f"📊 PERFORMANCE:\n"
    )
    metrics = performance.iloc[0].to_dict()

    lines = []
    for key, value in metrics.items():
        if isinstance(value, float):
            value = round(value, 4)
        lines.append(f"{key:<15}: {value}")

    performance_str = "\n".join(lines)
    message = f"{header}{performance_str}"

    send_message(
        message=message,
        telegram=True,
        token=self.kwargs.get("bot_token"),
        chat_id=self.kwargs.get("chat_id"),
    )

utils

send_telegram_message async

send_telegram_message(token, chat_id, text='')

Send a message to a telegram chat

Parameters:

Name	Type	Description	Default
token		str: Telegram bot token	required
chat_id		int or str or list: Chat id or list of chat ids	required
text		str: Message to send	''

Source code in src/bbstrader/trading/utils.py

async def send_telegram_message(token, chat_id, text=""):
    """
    Send a message to a telegram chat

    Args:
        token: str: Telegram bot token
        chat_id: int or str or list: Chat id or list of chat ids
        text: str: Message to send
    """
    try:
        bot = Bot(token=token)
        if isinstance(chat_id, (int, str)):
            chat_id = [chat_id]
        for id in chat_id:
            await bot.send_message(chat_id=id, text=text)
    except TelegramError as e:
        print(f"Error sending message: {e}")

send_notification

send_notification(title, message='')

Send a desktop notification

Parameters:

Name	Type	Description	Default
title		str: Title of the notification	required
message		str: Message of the notification	''

Source code in src/bbstrader/trading/utils.py

def send_notification(title, message=""):
    """
    Send a desktop notification

    Args:
        title: str: Title of the notification
        message: str: Message of the notification
    """
    notification = Notify(default_notification_application_name="bbstrading")
    notification.title = title
    notification.message = message
    notification.send()

send_message

send_message(title='SIGNAL', message='New signal', notify_me=False, telegram=False, token=None, chat_id=None)

Send a message to the user

Parameters:

Name	Type	Description	Default
title		str: Title of the message	'SIGNAL'
message		str: Message of the message	'New signal'
notify_me		bool: Send a desktop notification	False
telegram		bool: Send a telegram message	False
token		str: Telegram bot token	None
chat_id		int or str or list: Chat id or list of chat ids	None

Source code in src/bbstrader/trading/utils.py

def send_message(
    title="SIGNAL",
    message="New signal",
    notify_me=False,
    telegram=False,
    token=None,
    chat_id=None,
):
    """
    Send a message to the user

    Args:
        title: str: Title of the message
        message: str: Message of the message
        notify_me: bool: Send a desktop notification
        telegram: bool: Send a telegram message
        token: str: Telegram bot token
        chat_id: int or str or list: Chat id or list of chat ids
    """
    if notify_me:
        send_notification(title, message=message)
    if telegram:
        if token is None or chat_id is None:
            raise ValueError("Token and chat_id must be provided")
        asyncio.run(send_telegram_message(token, chat_id, text=message))