Skip to content

bbstrader.core

Shared utilities used across the rest of the package: the venue-neutral Broker execution abstraction (with the in-memory PaperBroker), data structures/logging, and built-in vectorized indicators (SMA, EMA, RSI, ATR, Bollinger Bands, MACD, z-score).

core

Overview

The Core Module provides the fundamental building blocks and abstract base classes for the trading system. It defines the essential components that are extended by other modules to create a complete trading application, ensuring a consistent and modular architecture.

Features

  • Abstract Base Classes: Defines the interfaces for key components like data handlers and strategies, promoting a standardized approach to development.
  • Modularity: Enforces a modular design by providing a clear separation of concerns between data handling, strategy logic, and execution.
  • Extensibility: Designed to be easily extended with concrete implementations, allowing for the creation of custom data sources and trading strategies.

Components

  • Data: Contains the abstract base class DataHandler, which defines the interface for managing market data from various sources.
  • Strategy: Contains the abstract base class Strategy, which provides the framework for developing trading strategies.

This module contains the abstract classes that form the foundation of the trading system. Implementations of these classes can be found in other modules like btengine and trading.

BrokerOrder dataclass 

BrokerOrder(symbol: str, side: OrderSide, quantity: float, order_type: OrderType = OrderType.MARKET, price: Optional[float] = None, id: Optional[int] = None)

A venue-neutral order request and its assigned id once submitted.

BrokerPosition dataclass 

BrokerPosition(symbol: str, quantity: float, avg_price: float)

An open position: signed quantity and volume-weighted average price.

AccountInfo dataclass 

AccountInfo(cash: float, equity: float, currency: str = 'USD')

A snapshot of account cash, mark-to-market equity and currency.

Broker

Bases: ABC

The execution contract every venue adapter implements.

connect abstractmethod 

connect() -> bool

Open the connection to the venue; return True on success.

Source code in src/bbstrader/core/broker.py 
73
74
75
76
@abstractmethod
def connect(self) -> bool:
    """Open the connection to the venue; return True on success."""
    ...

disconnect abstractmethod 

disconnect() -> None

Close the connection to the venue.

Source code in src/bbstrader/core/broker.py 
78
79
80
81
@abstractmethod
def disconnect(self) -> None:
    """Close the connection to the venue."""
    ...

account abstractmethod 

account() -> AccountInfo

Return the current account snapshot (cash, equity, currency).

Source code in src/bbstrader/core/broker.py 
83
84
85
86
@abstractmethod
def account(self) -> AccountInfo:
    """Return the current account snapshot (cash, equity, currency)."""
    ...

get_price abstractmethod 

get_price(symbol: str) -> float

Return the latest market price for symbol.

Parameters:

Name Type Description Default
symbol str

The instrument to price.

 required

Returns:

Name Type Description
float float

The latest price.
Source code in src/bbstrader/core/broker.py 
88
89
90
91
92
93
94
95
96
97
98
@abstractmethod
def get_price(self, symbol: str) -> float:
    """Return the latest market price for ``symbol``.

    Args:
        symbol (str): The instrument to price.

    Returns:
        float: The latest price.
    """
    ...

submit_order abstractmethod 

submit_order(order: BrokerOrder) -> BrokerOrder

Submit order to the venue and return it with venue fields set.

Parameters:

Name Type Description Default
order BrokerOrder

The order to submit.

 required

Returns:

Name Type Description
BrokerOrder BrokerOrder

The submitted order, populated with its assigned id.
Source code in src/bbstrader/core/broker.py 
100
101
102
103
104
105
106
107
108
109
110
@abstractmethod
def submit_order(self, order: BrokerOrder) -> BrokerOrder:
    """Submit ``order`` to the venue and return it with venue fields set.

    Args:
        order (BrokerOrder): The order to submit.

    Returns:
        BrokerOrder: The submitted order, populated with its assigned id.
    """
    ...

positions abstractmethod 

positions() -> List[BrokerPosition]

Return the currently open positions.

Source code in src/bbstrader/core/broker.py 
112
113
114
115
@abstractmethod
def positions(self) -> List[BrokerPosition]:
    """Return the currently open positions."""
    ...

orders abstractmethod 

orders() -> List[BrokerOrder]

Return the currently open (resting) orders.

Source code in src/bbstrader/core/broker.py 
117
118
119
120
@abstractmethod
def orders(self) -> List[BrokerOrder]:
    """Return the currently open (resting) orders."""
    ...

PaperBroker 

PaperBroker(cash: float = 100000.0, currency: str = 'USD')

Bases: Broker

An in-memory simulated broker with immediate market fills.

Maintains cash, positions (volume-weighted average price) and an order log. Prices are set with :meth:set_price; market orders fill at the current price, limit/stop orders rest until :meth:set_price crosses their level.

Initialise the paper broker with starting cash.

Parameters:

Name Type Description Default
cash float

The opening cash balance.

 100000.0
currency str

The account currency code.

 'USD'
Source code in src/bbstrader/core/broker.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
def __init__(self, cash: float = 100000.0, currency: str = "USD") -> None:
    """Initialise the paper broker with starting cash.

    Args:
        cash (float): The opening cash balance.
        currency (str): The account currency code.
    """
    self._cash = float(cash)
    self.currency = currency
    self._positions: Dict[str, BrokerPosition] = {}
    self._orders: List[BrokerOrder] = []
    self._open_orders: List[BrokerOrder] = []
    self._prices: Dict[str, float] = {}
    self._next_id = 1
    self._connected = False
    self.realized_pnl = 0.0

connect 

connect() -> bool

Mark the broker connected; always succeeds for the paper broker.

Source code in src/bbstrader/core/broker.py 
149
150
151
152
def connect(self) -> bool:
    """Mark the broker connected; always succeeds for the paper broker."""
    self._connected = True
    return True

disconnect 

disconnect() -> None

Mark the broker disconnected.

Source code in src/bbstrader/core/broker.py 
154
155
156
def disconnect(self) -> None:
    """Mark the broker disconnected."""
    self._connected = False

set_price 

set_price(symbol: str, price: float) -> None

Update the market price and trigger any resting orders it crosses.

Source code in src/bbstrader/core/broker.py 
159
160
161
162
def set_price(self, symbol: str, price: float) -> None:
    """Update the market price and trigger any resting orders it crosses."""
    self._prices[symbol] = float(price)
    self._check_open_orders(symbol)

get_price 

get_price(symbol: str) -> float

Return the last price set for symbol.

Parameters:

Name Type Description Default
symbol str

The instrument to price.

 required

Returns:

Name Type Description
float float

The most recently set price.

Raises:

Type Description
KeyError

If no price has been set for symbol.
Source code in src/bbstrader/core/broker.py 
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
def get_price(self, symbol: str) -> float:
    """Return the last price set for ``symbol``.

    Args:
        symbol (str): The instrument to price.

    Returns:
        float: The most recently set price.

    Raises:
        KeyError: If no price has been set for ``symbol``.
    """
    if symbol not in self._prices:
        raise KeyError(f"No price set for {symbol}.")
    return self._prices[symbol]

account 

account() -> AccountInfo

Return the account snapshot (cash, mark-to-market equity, currency).

Source code in src/bbstrader/core/broker.py 
181
182
183
184
185
def account(self) -> AccountInfo:
    """Return the account snapshot (cash, mark-to-market equity, currency)."""
    return AccountInfo(
        cash=self._cash, equity=self.equity(), currency=self.currency
    )

equity 

equity() -> float

Return cash plus the mark-to-market value of all open positions.

Source code in src/bbstrader/core/broker.py 
187
188
189
190
191
192
193
def equity(self) -> float:
    """Return cash plus the mark-to-market value of all open positions."""
    market_value = sum(
        pos.quantity * self._prices.get(sym, pos.avg_price)
        for sym, pos in self._positions.items()
    )
    return self._cash + market_value

positions 

positions() -> List[BrokerPosition]

Return the open (non-zero quantity) positions.

Source code in src/bbstrader/core/broker.py 
195
196
197
def positions(self) -> List[BrokerPosition]:
    """Return the open (non-zero quantity) positions."""
    return [p for p in self._positions.values() if p.quantity != 0]

orders 

orders() -> List[BrokerOrder]

Return the resting (not yet filled) orders.

Source code in src/bbstrader/core/broker.py 
199
200
201
def orders(self) -> List[BrokerOrder]:
    """Return the resting (not yet filled) orders."""
    return list(self._open_orders)

submit_order 

submit_order(order: BrokerOrder) -> BrokerOrder

Submit an order, filling market orders immediately at the set price.

Market orders fill at order.price or the current market price; limit/stop orders rest and fill when :meth:set_price crosses them.

Parameters:

Name Type Description Default
order BrokerOrder

The order to submit. Its id is assigned here.

 required

Returns:

Name Type Description
BrokerOrder BrokerOrder

The same order with its assigned id.

Raises:

Type Description
ValueError

If order.quantity is not positive.
Source code in src/bbstrader/core/broker.py 
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
def submit_order(self, order: BrokerOrder) -> BrokerOrder:
    """Submit an order, filling market orders immediately at the set price.

    Market orders fill at ``order.price`` or the current market price;
    limit/stop orders rest and fill when :meth:`set_price` crosses them.

    Args:
        order (BrokerOrder): The order to submit. Its ``id`` is assigned here.

    Returns:
        BrokerOrder: The same order with its assigned ``id``.

    Raises:
        ValueError: If ``order.quantity`` is not positive.
    """
    if order.quantity <= 0:
        raise ValueError("order quantity must be positive.")
    order.id = self._next_id
    self._next_id += 1
    self._orders.append(order)
    if order.order_type is OrderType.MARKET:
        price = order.price or self.get_price(order.symbol)
        self._fill(order, price)
    else:
        self._open_orders.append(order)
        # A resting order may fill immediately if already crossed.
        if order.symbol in self._prices:
            self._check_open_orders(order.symbol)
    return order

FmpNews 

FmpNews(api: str)

Bases: object

FmpNews is responsible for retrieving financial news, press releases, and articles from Financial Modeling Prep (FMP).

FmpNews provides methods to fetch the latest stock, crypto, forex, and general financial news, as well as financial articles and press releases.

Parameters:

Name Type Description Default
api str

The API key for accessing FMP's news data.

 required
Example

fmp_news = FmpNews(api="your_api_key_here")

Source code in src/bbstrader/core/data.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
def __init__(self, api: str) -> None:
    """
    Args:
        api (str): The API key for accessing FMP's news data.

    Example:
        fmp_news = FmpNews(api="your_api_key_here")
    """
    if api is None:
        raise ValueError("API key is required For FmpNews")
    self.__api = api

get_articles 

get_articles(**kwargs: Any) -> List[Dict[str, Any]]

Fetch FMP articles with their HTML content stripped to plain text.

Parameters:

Name Type Description Default
kwargs Any

Optional page/limit paging parameters.

 {}

Returns:

Type Description
List[Dict[str, Any]]

List[Dict[str, Any]]: Records with title, date, content
List[Dict[str, Any]]

(plain text) and tickers.
Source code in src/bbstrader/core/data.py 
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
def get_articles(self, **kwargs: Any) -> List[Dict[str, Any]]:
    """Fetch FMP articles with their HTML content stripped to plain text.

    Args:
        kwargs (Any): Optional ``page``/``limit`` paging parameters.

    Returns:
        List[Dict[str, Any]]: Records with ``title``, ``date``, ``content``
        (plain text) and ``tickers``.
    """

    def html_parser(content: str) -> str:
        """Strip HTML markup from ``content`` to a single line of text.

        Args:
            content (str): The raw HTML article body.

        Returns:
            str: The extracted text with newlines removed.
        """
        soup = BeautifulSoup(content, "html.parser")
        text = soup.get_text(separator="\n")
        return text.replace("\n", "")

    articles = self._load_news("articles", **kwargs)
    df = pd.DataFrame(articles)
    df = df[["title", "date", "content", "tickers"]]
    df["content"] = df["content"].apply(html_parser)
    return df.to_dict(orient="records")  # type: ignore

get_releases 

get_releases(symbol: Optional[str] = None, **kwargs: Any) -> List[Dict[str, Any]]

Fetch the latest FMP press releases, optionally for one symbol.

Parameters:

Name Type Description Default
symbol Optional[str]

Restrict to a single symbol.

 None
kwargs Any

Optional paging/date parameters (see :meth:_load_news).

 {}

Returns:

Type Description
List[Dict[str, Any]]

List[Dict[str, Any]]: The press-release records.
Source code in src/bbstrader/core/data.py 
182
183
184
185
186
187
188
189
190
191
192
193
194
def get_releases(
    self, symbol: Optional[str] = None, **kwargs: Any
) -> List[Dict[str, Any]]:
    """Fetch the latest FMP press releases, optionally for one symbol.

    Args:
        symbol (Optional[str]): Restrict to a single symbol.
        kwargs (Any): Optional paging/date parameters (see :meth:`_load_news`).

    Returns:
        List[Dict[str, Any]]: The press-release records.
    """
    return self._load_news("press-releases", symbol, **kwargs)

get_stock_news 

get_stock_news(symbol: Optional[str] = None, **kwargs: Any) -> List[Dict[str, Any]]

Fetch the latest FMP stock news, optionally for one symbol.

Parameters:

Name Type Description Default
symbol Optional[str]

Restrict to a single symbol.

 None
kwargs Any

Optional paging/date parameters (see :meth:_load_news).

 {}

Returns:

Type Description
List[Dict[str, Any]]

List[Dict[str, Any]]: The stock-news records.
Source code in src/bbstrader/core/data.py 
196
197
198
199
200
201
202
203
204
205
206
207
208
def get_stock_news(
    self, symbol: Optional[str] = None, **kwargs: Any
) -> List[Dict[str, Any]]:
    """Fetch the latest FMP stock news, optionally for one symbol.

    Args:
        symbol (Optional[str]): Restrict to a single symbol.
        kwargs (Any): Optional paging/date parameters (see :meth:`_load_news`).

    Returns:
        List[Dict[str, Any]]: The stock-news records.
    """
    return self._load_news("stock", symbol, **kwargs)

get_crypto_news 

get_crypto_news(symbol: Optional[str] = None, **kwargs: Any) -> List[Dict[str, Any]]

Fetch the latest FMP crypto news, optionally for one symbol.

Parameters:

Name Type Description Default
symbol Optional[str]

Restrict to a single symbol.

 None
kwargs Any

Optional paging/date parameters (see :meth:_load_news).

 {}

Returns:

Type Description
List[Dict[str, Any]]

List[Dict[str, Any]]: The crypto-news records.
Source code in src/bbstrader/core/data.py 
210
211
212
213
214
215
216
217
218
219
220
221
222
def get_crypto_news(
    self, symbol: Optional[str] = None, **kwargs: Any
) -> List[Dict[str, Any]]:
    """Fetch the latest FMP crypto news, optionally for one symbol.

    Args:
        symbol (Optional[str]): Restrict to a single symbol.
        kwargs (Any): Optional paging/date parameters (see :meth:`_load_news`).

    Returns:
        List[Dict[str, Any]]: The crypto-news records.
    """
    return self._load_news("crypto", symbol, **kwargs)

get_forex_news 

get_forex_news(symbol: Optional[str] = None, **kwargs: Any) -> List[Dict[str, Any]]

Fetch the latest FMP forex news, optionally for one symbol.

Parameters:

Name Type Description Default
symbol Optional[str]

Restrict to a single symbol.

 None
kwargs Any

Optional paging/date parameters (see :meth:_load_news).

 {}

Returns:

Type Description
List[Dict[str, Any]]

List[Dict[str, Any]]: The forex-news records.
Source code in src/bbstrader/core/data.py 
224
225
226
227
228
229
230
231
232
233
234
235
236
def get_forex_news(
    self, symbol: Optional[str] = None, **kwargs: Any
) -> List[Dict[str, Any]]:
    """Fetch the latest FMP forex news, optionally for one symbol.

    Args:
        symbol (Optional[str]): Restrict to a single symbol.
        kwargs (Any): Optional paging/date parameters (see :meth:`_load_news`).

    Returns:
        List[Dict[str, Any]]: The forex-news records.
    """
    return self._load_news("forex", symbol, **kwargs)

parse_news 

parse_news(news: List[Dict[str, Any]], symbol: Optional[str] = None, **kwargs: Any) -> List[str]

Flatten and date-filter raw news records into text strings.

Keeps records published within the start/end window and, when a symbol is given, only those mentioning it.

Parameters:

Name Type Description Default
news List[Dict[str, Any]]

The raw records to parse.

 required
symbol Optional[str]

Restrict to records mentioning this symbol.

 None
kwargs Any

Optional start and end date bounds as YYYY-MM-DD HH:MM:SS strings.

 {}

Returns:

Type Description
List[str]

List[str]: One flattened text string per matching record.
Source code in src/bbstrader/core/data.py 
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
def parse_news(
    self, news: List[Dict[str, Any]], symbol: Optional[str] = None, **kwargs: Any
) -> List[str]:
    """Flatten and date-filter raw news records into text strings.

    Keeps records published within the ``start``/``end`` window and, when a
    ``symbol`` is given, only those mentioning it.

    Args:
        news (List[Dict[str, Any]]): The raw records to parse.
        symbol (Optional[str]): Restrict to records mentioning this symbol.
        kwargs (Any): Optional ``start`` and ``end`` date bounds as
            ``YYYY-MM-DD HH:MM:SS`` strings.

    Returns:
        List[str]: One flattened text string per matching record.
    """
    start = kwargs.get("start")
    end = kwargs.get("end")
    end_date = self._last_date(end) if end is not None else datetime.now().date()

    def parse_record(record: Dict[str, Any]) -> str:
        """Flatten a news record's text fields into a single string.

        Args:
            record (Dict[str, Any]): A raw news record.

        Returns:
            str: The symbol, title, text, content and tickers joined by spaces.
        """
        return " ".join(
            [
                record.pop("symbol", ""),
                record.pop("title", ""),
                record.pop("text", ""),
                record.pop("content", ""),
                record.pop("tickers", ""),
            ]
        )

    parsed_news = []
    for record in news:
        date = record.get("publishedDate")
        published_date = self._last_date(record.get("date", date)).date()  # type: ignore
        start_date = (
            self._last_date(start).date() if start is not None else published_date
        )
        if published_date >= start_date and published_date <= end_date:
            if symbol is not None:
                if record.get("symbol", "") == symbol or symbol in record.get(
                    "tickers", ""
                ):
                    parsed_news.append(parse_record(record))
            else:
                parsed_news.append(parse_record(record))
    return parsed_news

get_latest_articles 

get_latest_articles(articles: Optional[List[Dict[str, Any]]] = None, save: bool = False, **kwargs: Any) -> List[Dict[str, Any]]

Return the latest FMP articles, using a local CSV cache when fresh.

Reads latest_fmp_articles.csv if present and recent enough; otherwise downloads fresh articles and, when save is set, refreshes the cache.

Parameters:

Name Type Description Default
articles Optional[List[Dict[str, Any]]]

Pre-fetched articles to use instead of reading the cache or downloading.

 None
save bool

When True, persist the fetched articles to the cache CSV.

 False
kwargs Any

Optional end bound and paging parameters forwarded to :meth:get_articles.

 {}

Returns:

Type Description
List[Dict[str, Any]]

List[Dict[str, Any]]: The latest article records.
Source code in src/bbstrader/core/data.py 
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
def get_latest_articles(
    self,
    articles: Optional[List[Dict[str, Any]]] = None,
    save: bool = False,
    **kwargs: Any,
) -> List[Dict[str, Any]]:
    """Return the latest FMP articles, using a local CSV cache when fresh.

    Reads ``latest_fmp_articles.csv`` if present and recent enough; otherwise
    downloads fresh articles and, when ``save`` is set, refreshes the cache.

    Args:
        articles (Optional[List[Dict[str, Any]]]): Pre-fetched articles to
            use instead of reading the cache or downloading.
        save (bool): When True, persist the fetched articles to the cache CSV.
        kwargs (Any): Optional ``end`` bound and paging parameters forwarded
            to :meth:`get_articles`.

    Returns:
        List[Dict[str, Any]]: The latest article records.
    """
    end = kwargs.get("end")
    now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    end_date = self._last_date(end) if end is not None else self._last_date(now)
    if articles is None:
        try:
            articles = pd.read_csv("latest_fmp_articles.csv")  # type: ignore
            articles = articles.to_dict(orient="records")  # type: ignore
            if self._last_date(articles[0]["date"]).hour < end_date.hour:  # type: ignore
                articles = self.get_articles(**kwargs)
            else:
                return articles  # type: ignore
        except FileNotFoundError:
            articles = self.get_articles(**kwargs)

    if save and len(articles) > 0:
        df = pd.DataFrame(articles)
        df.to_csv("latest_fmp_articles.csv", index=False)
    return articles

get_news 

get_news(query: str, source: str = 'articles', articles: Optional[List[Dict[str, Any]]] = None, symbol: Optional[str] = None, **kwargs: Any) -> List[str]

Retrieves relevant financial news based on the specified source.

Parameters:

Name Type Description Default
query str

The search query or keyword for filtering news, may also be a ticker.

 required
source str

The news source to retrieve from. Defaults to "articles". Available options: "articles", "releases", "stock", "crypto", "forex".

 'articles'
articles list

List of pre-fetched articles to use when source="articles". Defaults to None.

 None
symbol str

The financial asset symbol (e.g., "AAPL" for stocks, "BTC" for crypto). Defaults to None.

 None
**kwargs dict

Additional arguments required for fetching news data. May include: - start (str): The start period for news retrieval (YYY-MM-DD) - end (str): The end period for news retrieval (YYY-MM-DD) - page (int): The number of page to load for each news - limit (int): Maximum Responses per API Call

 {}

Returns:

Type Description
List[str]

list[dict]: A list of filtered news articles relevant to the query. Returns an empty list if no relevant news is found.
Source code in src/bbstrader/core/data.py 
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
def get_news(
    self,
    query: str,
    source: str = "articles",
    articles: Optional[List[Dict[str, Any]]] = None,
    symbol: Optional[str] = None,
    **kwargs: Any,
) -> List[str]:
    """
    Retrieves relevant financial news based on the specified source.

    Args:
        query (str): The search query or keyword for filtering news, may also be a ticker.
        source (str, optional): The news source to retrieve from. Defaults to "articles".
                                Available options: "articles", "releases", "stock", "crypto", "forex".
        articles (list, optional): List of pre-fetched articles to use when source="articles". Defaults to None.
        symbol (str, optional): The financial asset symbol (e.g., "AAPL" for stocks, "BTC" for crypto). Defaults to None.
        **kwargs (dict):
            Additional arguments required for fetching news data. May include:
            - start (str): The start period for news retrieval (YYY-MM-DD)
            - end (str): The end period for news retrieval (YYY-MM-DD)
            - page (int): The number  of page to load  for each news
            - limit (int): Maximum Responses per API Call

    Returns:
        list[dict]: A list of filtered news articles relevant to the query.
                    Returns an empty list if no relevant news is found.
    """
    query = _get_search_query(query)
    if symbol is not None:
        symbol = symbol.replace("-", "").split("=")[
            0
        ]  # if symbol is a yahoo finance ticker
    source_methods = {
        "articles": lambda: self.get_latest_articles(
            articles=articles, save=True, **kwargs
        ),
        "releases": lambda: self.get_releases(symbol=symbol, **kwargs),
        "stock": lambda: self.get_stock_news(symbol=symbol, **kwargs),
        "crypto": lambda: self.get_crypto_news(symbol=symbol, **kwargs),
        "forex": lambda: self.get_forex_news(symbol=symbol, **kwargs),
    }
    news_source = source_methods.get(source, lambda: [])()
    if source == "articles":
        symbol = None  # Articles do not require a symbol filter
    news = self.parse_news(news_source, symbol=symbol, **kwargs)
    return _filter_news(news, query)

FinancialNews

Bases: object

The FinancialNews class provides methods to fetch financial news, articles, and discussions from various sources such as Yahoo Finance, Google Finance, Reddit, Coindesk and Twitter. It also supports retrieving news using Financial Modeling Prep (FMP).

get_yahoo_finance_news 

get_yahoo_finance_news(query: str, asset_type: str = 'stock', n_news: int = 10) -> List[str]

Fetches recent Yahoo Finance news headlines for a given financial asset.

Parameters:

Name Type Description Default
query str

The asset symbol or name (e.g., "AAPL").

 required
asset_type str

The type of asset (e.g., "stock", "etf"). Defaults to "stock", supported types include: - "stock": Stock symbols (e.g., AAPL, MSFT) - "etf": Exchange-traded funds (e.g., SPY, QQQ) - "future": Futures contracts (e.g., CL=F for crude oil) - "forex": Forex pairs (e.g., EURUSD=X, USDJPY=X) - "crypto": Cryptocurrency pairs (e.g., BTC-USD, ETH-USD) - "index": Stock market indices (e.g., ^GSPC for S&P 500)

 'stock'
n_news int

The number of news headlines to return. Defaults to 10.

 10
Note

For commotities and bonds, use the "Future" asset type.

Returns:

Type Description
List[str]

list[str]: A list of Yahoo Finance news headlines relevant to the query.
Source code in src/bbstrader/core/data.py 
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
def get_yahoo_finance_news(
    self, query: str, asset_type: str = "stock", n_news: int = 10
) -> List[str]:
    """
    Fetches recent Yahoo Finance news headlines for a given financial asset.

    Args:
        query (str): The asset symbol or name (e.g., "AAPL").
        asset_type (str, optional): The type of asset (e.g., "stock", "etf"). Defaults to "stock",
            supported types include:
            - "stock": Stock symbols (e.g., AAPL, MSFT)
            - "etf": Exchange-traded funds (e.g., SPY, QQQ)
            - "future": Futures contracts (e.g., CL=F for crude oil)
            - "forex": Forex pairs (e.g., EURUSD=X, USDJPY=X)
            - "crypto": Cryptocurrency pairs (e.g., BTC-USD, ETH-USD)
            - "index": Stock market indices (e.g., ^GSPC for S&P 500)
        n_news (int, optional): The number of news headlines to return. Defaults to 10.

    Note:
        For commotities and bonds, use the "Future" asset type.

    Returns:
        list[str]: A list of Yahoo Finance news headlines relevant to the query.
    """
    if asset_type == "forex" or asset_type == "future":
        assert "=" in query, (
            "Forex query must contain '=' for currency pairs (e.g., EURUSD=X, CL=F)"
        )
    if asset_type == "crypto":
        assert "-" in query, (
            "Crypto query must contain '-' for crypto pairs (e.g., BTC-USD, ETH-USD)"
        )
    if asset_type == "index":
        assert query.startswith("^"), (
            "Index query must start with '^' (e.g., ^GSPC for S&P 500)"
        )
    url = (
        f"https://finance.yahoo.com/quote/{query}/news"
        if asset_type in ["stock", "etf", "index", "future", "forex"]
        else "https://finance.yahoo.com/news"
    )
    return self._fetch_news(url, query, n_news, "h3")

get_google_finance_news 

get_google_finance_news(query: str, asset_type: str = 'stock', n_news: int = 10) -> List[str]

Fetches recent Google Finance news headlines for a given financial asset.

Parameters:

Name Type Description Default
query str

The asset symbol or name (e.g., "AAPL").

 required
asset_type str

The type of asset (e.g., "stock", "crypto"). Defaults to "stock". Supported types include: - "stock": Stock symbols (e.g., AAPL, MSFT) - "etf": Exchange-traded funds (e.g., SPY, QQQ) - "future": Futures contracts (e.g., CL=F or crude oil) - "forex": Forex pairs (e.g., EURUSD, USDJPY) - "crypto": Cryptocurrency pairs (e.g., BTCUSD, ETHUSD)

 'stock'
n_news int

The number of news headlines to return. Defaults to 10.

 10

Returns:

Type Description
List[str]

list[str]: A list of Google Finance news headlines relevant to the query.
Source code in src/bbstrader/core/data.py 
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
def get_google_finance_news(
    self, query: str, asset_type: str = "stock", n_news: int = 10
) -> List[str]:
    """
    Fetches recent Google Finance news headlines for a given financial asset.

    Args:
        query (str): The asset symbol or name (e.g., "AAPL").
        asset_type (str, optional): The type of asset (e.g., "stock", "crypto"). Defaults to "stock".
            Supported types include:
            - "stock": Stock symbols (e.g., AAPL, MSFT)
            - "etf": Exchange-traded funds (e.g., SPY, QQQ)
            - "future": Futures contracts (e.g., CL=F or crude oil)
            - "forex": Forex pairs (e.g., EURUSD, USDJPY)
            - "crypto": Cryptocurrency pairs (e.g., BTCUSD, ETHUSD)
        n_news (int, optional): The number of news headlines to return. Defaults to 10.

    Returns:
        list[str]: A list of Google Finance news headlines relevant to the query.
    """
    search_terms = {
        "stock": f"{query} stock OR {query} shares OR {query} market",
        "etf": f"{query} ETF OR {query} fund OR {query} exchange-traded fund",
        "future": f"{query} futures OR {query} price OR {query} market",
        "forex": f"{query} forex OR {query} exchange rate OR {query} market",
        "crypto": f"{query} cryptocurrency OR {query} price OR {query} market",
        "index": f"{query} index OR {query} stock market OR {query} performance",
    }
    search_query = search_terms.get(asset_type, query)
    url = f"https://news.google.com/search?q={search_query.replace(' ', '+')}"
    return self._fetch_news(url, query, n_news, "a")

get_reddit_posts 

get_reddit_posts(symbol: str, client_id=None, client_secret=None, user_agent=None, asset_class='stock', n_posts=10) -> List[str]

Fetches recent Reddit posts related to a financial asset.

This method queries relevant subreddits for posts mentioning the specified symbol and returns posts based on the selected asset class (e.g., stock, forex, crypto). The function uses the PRAW library to interact with Reddit's API.

Parameters:

Name Type Description Default
symbol str

The financial asset's symbol or name to search for.

 required
client_id str

Reddit API client ID for authentication.

 None
client_secret str

Reddit API client secret.

 None
user_agent str

Reddit API user agent.

 None
asset_class str

The type of financial asset. Defaults to "stock". - "stock": Searches in stock-related subreddits (e.g., wallstreetbets, stocks). - "forex": Searches in forex-related subreddits. - "commodities": Searches in commodity-related subreddits (e.g., gold, oil). - "etf": Searches in ETF-related subreddits. - "future": Searches in futures and options trading subreddits. - "crypto": Searches in cryptocurrency-related subreddits. - If an unrecognized asset class is provided, defaults to stock-related subreddits.

 'stock'
n_posts int

The number of posts to return per subreddit. Defaults to 10.

 10

Returns:

Type Description
List[str]

list[str]: A list of Reddit post contents matching the query. Each entry contains the post title and body. If no posts are found or an error occurs, returns an empty list.

Raises:

Type Description
PRAWException

If an error occurs while interacting with Reddit's API.
Example

get_reddit_posts(symbol="AAPL", client_id="your_id", client_secret="your_secret", user_agent="your_agent", asset_class="stock", n_posts=5) ["Apple stock is rallying today due to strong earnings.", "Should I buy $AAPL now?", ...]

Notes
  • Requires valid Reddit API credentials.
Source code in src/bbstrader/core/data.py 
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
def get_reddit_posts(
    self,
    symbol: str,
    client_id=None,
    client_secret=None,
    user_agent=None,
    asset_class="stock",
    n_posts=10,
) -> List[str]:
    """
    Fetches recent Reddit posts related to a financial asset.

    This method queries relevant subreddits for posts mentioning the specified symbol
    and returns posts based on the selected asset class (e.g., stock, forex, crypto).
    The function uses the PRAW library to interact with Reddit's API.

    Args:
        symbol (str): The financial asset's symbol or name to search for.
        client_id (str, optional): Reddit API client ID for authentication.
        client_secret (str, optional): Reddit API client secret.
        user_agent (str, optional): Reddit API user agent.
        asset_class (str, optional): The type of financial asset. Defaults to "stock".
            - "stock": Searches in stock-related subreddits (e.g., wallstreetbets, stocks).
            - "forex": Searches in forex-related subreddits.
            - "commodities": Searches in commodity-related subreddits (e.g., gold, oil).
            - "etf": Searches in ETF-related subreddits.
            - "future": Searches in futures and options trading subreddits.
            - "crypto": Searches in cryptocurrency-related subreddits.
            - If an unrecognized asset class is provided, defaults to stock-related subreddits.
        n_posts (int, optional): The number of posts to return per subreddit. Defaults to 10.

    Returns:
        list[str]: A list of Reddit post contents matching the query.
                Each entry contains the post title and body.
                If no posts are found or an error occurs, returns an empty list.

    Raises:
        praw.exceptions.PRAWException: If an error occurs while interacting with Reddit's API.

    Example:
        >>> get_reddit_posts(symbol="AAPL", client_id="your_id", client_secret="your_secret", user_agent="your_agent", asset_class="stock", n_posts=5)
        ["Apple stock is rallying today due to strong earnings.", "Should I buy $AAPL now?", ...]

    Notes:
        - Requires valid Reddit API credentials.
    """

    praw = _require("praw", "social")
    reddit = praw.Reddit(
        client_id=client_id,
        client_secret=client_secret,
        user_agent=user_agent,
        check_for_updates=False,
        comment_kind="t1",
        message_kind="t4",
        redditor_kind="t2",
        submission_kind="t3",
        subreddit_kind="t5",
        trophy_kind="t6",
        oauth_url="https://oauth.reddit.com",
        reddit_url="https://www.reddit.com",
        short_url="https://redd.it",
        timeout=16,
        ratelimit_seconds=5,
    )
    assert reddit.read_only
    subreddit_mapping = {
        "stock": ["wallstreetbets", "stocks", "investing", "StockMarket"],
        "forex": ["Forex", "ForexTrading", "DayTrading"],
        "etfs": ["ETFs", "investing"],
        "futures": [
            "FuturesTrading",
            "OptionsTrading",
            "DayTrading",
            "Commodities",
            "Gold",
            "Silverbugs",
            "oil",
        ],
        "crypto": ["CryptoCurrency", "Bitcoin", "ethereum", "altcoin"],
    }
    try:
        subreddits = subreddit_mapping.get(asset_class.lower(), ["stocks"])
    except Exception:
        return []

    posts = []
    for sub in subreddits:
        subreddit = reddit.subreddit(sub)
        query = _get_search_query(symbol)
        all_posts = subreddit.search(query, limit=n_posts)
        for post in all_posts:
            text = post.title + " " + post.selftext
            if _find_news(query, text):
                posts.append(text)
    return posts

get_twitter_posts 

get_twitter_posts(query: str, asset_type: str = 'stock', bearer: Optional[str] = None, api_key: Optional[str] = None, api_secret: Optional[str] = None, access_token: Optional[str] = None, access_secret: Optional[str] = None, n_posts: int = 10) -> List[str]

Fetches recent tweets related to a financial asset.

This method queries Twitter for recent posts mentioning the specified asset and filters the results based on the asset type (e.g., stock, forex, crypto). The function uses the Tweepy API to fetch tweets and returns a list of tweet texts.

Parameters:

Name Type Description Default
query str

The main keyword to search for (e.g., a stock ticker or asset name).

 required
asset_type str

The type of financial asset. Defaults to "stock". - "stock": Searches for tweets mentioning the stock or shares. - "forex": Searches for tweets mentioning foreign exchange (forex) or currency. - "crypto": Searches for tweets mentioning cryptocurrency or related terms. - "commodity": Searches for tweets mentioning commodities or futures trading. - "index": Searches for tweets mentioning stock market indices. - "bond": Searches for tweets mentioning bonds or fixed income securities. - If an unrecognized asset type is provided, defaults to general finance-related tweets.

 'stock'
bearer str

Twitter API bearer token for authentication.

 None
api_key str

Twitter API consumer key.

 None
api_secret str

Twitter API consumer secret.

 None
access_token str

Twitter API access token.

 None
access_secret str

Twitter API access token secret.

 None
n_posts int

The number of tweets to return. Defaults to 10.

 10

Returns:

Type Description
List[str]

list[str]: A list of up to n_posts tweet texts matching the query. If no tweets are found or an API error occurs, returns an empty list.

Raises:

Type Description
TweepyException

If an error occurs while making the Twitter API request.
Example

get_twitter_posts(query="AAPL", asset_type="stock", bearer="YOUR_BEARER_TOKEN", n_posts=5) ["Apple stock surges after strong earnings!", "Is $AAPL a buy at this price?", ...]

Source code in src/bbstrader/core/data.py 
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
def get_twitter_posts(
    self,
    query: str,
    asset_type: str = "stock",
    bearer: Optional[str] = None,
    api_key: Optional[str] = None,
    api_secret: Optional[str] = None,
    access_token: Optional[str] = None,
    access_secret: Optional[str] = None,
    n_posts: int = 10,
) -> List[str]:
    """
    Fetches recent tweets related to a financial asset.

    This method queries Twitter for recent posts mentioning the specified asset
    and filters the results based on the asset type (e.g., stock, forex, crypto).
    The function uses the Tweepy API to fetch tweets and returns a list of tweet texts.

    Args:
        query (str): The main keyword to search for (e.g., a stock ticker or asset name).
        asset_type (str, optional): The type of financial asset. Defaults to "stock".
            - "stock": Searches for tweets mentioning the stock or shares.
            - "forex": Searches for tweets mentioning foreign exchange (forex) or currency.
            - "crypto": Searches for tweets mentioning cryptocurrency or related terms.
            - "commodity": Searches for tweets mentioning commodities or futures trading.
            - "index": Searches for tweets mentioning stock market indices.
            - "bond": Searches for tweets mentioning bonds or fixed income securities.
            - If an unrecognized asset type is provided, defaults to general finance-related tweets.
        bearer (str, optional): Twitter API bearer token for authentication.
        api_key (str, optional): Twitter API consumer key.
        api_secret (str, optional): Twitter API consumer secret.
        access_token (str, optional): Twitter API access token.
        access_secret (str, optional): Twitter API access token secret.
        n_posts (int, optional): The number of tweets to return. Defaults to 10.

    Returns:
        list[str]: A list of up to `n_posts` tweet texts matching the query.
                If no tweets are found or an API error occurs, returns an empty list.

    Raises:
        tweepy.TweepyException: If an error occurs while making the Twitter API request.

    Example:
        >>> get_twitter_posts(query="AAPL", asset_type="stock", bearer="YOUR_BEARER_TOKEN", n_posts=5)
        ["Apple stock surges after strong earnings!", "Is $AAPL a buy at this price?", ...]
    """
    tweepy = _require("tweepy", "social")
    client = tweepy.Client(
        bearer_token=bearer,
        consumer_key=api_key,
        consumer_secret=api_secret,
        access_token=access_token,
        access_token_secret=access_secret,
    )
    asset_queries = {
        "stock": f"{query} stock OR {query} shares -is:retweet lang:en",
        "forex": f"{query} forex OR {query} currency -is:retweet lang:en",
        "crypto": f"{query} cryptocurrency OR {query} crypto OR #{query} -is:retweet lang:en",
        "commodity": f"{query} commodity OR {query} futures OR {query} trading -is:retweet lang:en",
        "index": f"{query} index OR {query} market -is:retweet lang:en",
        "bond": f"{query} bonds OR {query} fixed income -is:retweet lang:en",
    }
    # Get the correct query based on the asset type
    search = asset_queries.get(
        asset_type.lower(), f"{query} finance -is:retweet lang:en"
    )
    try:
        tweets = client.search_recent_tweets(
            query=search, max_results=100, tweet_fields=["text"]
        )
        query = _get_search_query(query)
        news = [tweet.text for tweet in tweets.data] if tweets.data else []  # type: ignore
        return _filter_news(news, query)[:n_posts]
    except tweepy.TweepyException:
        return []

get_fmp_news 

get_fmp_news(api: str | None = None) -> FmpNews

Return an :class:FmpNews client for the given API key.

Parameters:

Name Type Description Default
api str | None

The Financial Modeling Prep API key.

 None

Returns:

Name Type Description
FmpNews FmpNews

A news client bound to api.
Source code in src/bbstrader/core/data.py 
689
690
691
692
693
694
695
696
697
698
def get_fmp_news(self, api: str | None = None) -> FmpNews:
    """Return an :class:`FmpNews` client for the given API key.

    Args:
        api (str | None): The Financial Modeling Prep API key.

    Returns:
        FmpNews: A news client bound to ``api``.
    """
    return FmpNews(api=api)  # type: ignore

get_coindesk_news 

get_coindesk_news(query='', lang: Literal['EN', 'ES', 'TR', 'FR', 'JP', 'PT'] = 'EN', limit=10, list_of_str=False) -> List[str] | List[dict]

Fetches and filters recent news articles from CoinDesk's News API.

Parameters:

Name Type Description Default
query

str, optional A search term to filter articles by title, body, or keywords. If empty, all articles are returned without filtering (default is "").

 required
lang

Literal["EN", "ES", "TR", "FR", "JP", "PT"], optional Language in which to fetch news articles. Supported languages: English (EN), Spanish (ES), Turkish (TR), French (FR), Japanese (JP), and Portuguese (PT). Default is "EN".

 required
limit

int, optional Maximum number of articles to retrieve. Default is 50.

 required
list_of_str

bool, optional If True, returns a list of strings (concatenated article content). If False, returns a list of filtered article dictionaries. Default is False.

 required

Returns:

Type Description
List[str] | List[dict]

List[str] | List[dict] - If query is empty: returns a list of filtered article dictionaries. - If query is provided: - Returns a list of strings if list_of_str=True. - Returns a list of filtered article dictionaries otherwise.
Each article dictionary contains the following fields
  • 'published_on': datetime of publication
  • 'title': article headline
  • 'subtitle': secondary headline
  • 'url': direct link to the article
  • 'body': article content
  • 'keywords': associated tags
  • 'sentiment': sentiment label
  • 'status': publication status
Notes
  • Articles marked as sponsored are automatically excluded.
Source code in src/bbstrader/core/data.py 
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
def get_coindesk_news(
    self,
    query="",
    lang: Literal["EN", "ES", "TR", "FR", "JP", "PT"] = "EN",
    limit=10,
    list_of_str=False,
) -> List[str] | List[dict]:
    """
    Fetches and filters recent news articles from CoinDesk's News API.

    Args:
        query : str, optional
            A search term to filter articles by title, body, or keywords.
            If empty, all articles are returned without filtering (default is "").

        lang : Literal["EN", "ES", "TR", "FR", "JP", "PT"], optional
            Language in which to fetch news articles. Supported languages:
            English (EN), Spanish (ES), Turkish (TR), French (FR), Japanese (JP), and Portuguese (PT).
            Default is "EN".

        limit : int, optional
            Maximum number of articles to retrieve. Default is 50.

        list_of_str : bool, optional
            If True, returns a list of strings (concatenated article content).
            If False, returns a list of filtered article dictionaries.
            Default is False.

    Returns:
        List[str] | List[dict]
            - If `query` is empty: returns a list of filtered article dictionaries.
            - If `query` is provided:
                - Returns a list of strings if `list_of_str=True`.
                - Returns a list of filtered article dictionaries otherwise.

    Each article dictionary contains the following fields:
        - 'published_on': datetime of publication
        - 'title': article headline
        - 'subtitle': secondary headline
        - 'url': direct link to the article
        - 'body': article content
        - 'keywords': associated tags
        - 'sentiment': sentiment label
        - 'status': publication status

    Notes:
        - Articles marked as sponsored are automatically excluded.
    """
    maximum = 100
    if limit > maximum:
        raise ValueError(f"Number of total news articles allowed is {maximum}")
    try:
        response = requests.get(
            "https://data-api.coindesk.com/news/v1/article/list",
            params={"lang": lang, "limit": limit},
            headers={"Content-type": "application/json; charset=UTF-8"},
        )
        response.raise_for_status()
        json_response = response.json()
    except requests.exceptions.RequestException:
        return []
    if (
        response.status_code != 200
        or "Data" not in json_response
        or len(json_response["Data"]) == 0
    ):
        return []
    articles = json_response["Data"]
    to_keep = [
        "PUBLISHED_ON",
        "TITLE",
        "SUBTITLE",
        "URL",
        "BODY",
        "KEYWORDS",
        "SENTIMENT",
        "STATUS",
    ]
    filtered_articles = []
    for article in articles:
        keys = article.keys()
        filtered_articles.append(
            {
                k.lower(): article[k]
                if k in keys and k != "PUBLISHED_ON"
                else datetime.fromtimestamp(article[k])
                for k in to_keep
                if article[k] is not None and "sponsored" not in str(article[k])
            }
        )
    if query == "" or len(filtered_articles) == 0:
        return filtered_articles
    to_return = []
    query = _get_search_query(query)
    for article in filtered_articles:
        if not all(k in article for k in ("title", "body", "keywords")):
            continue
        text = article["title"] + " " + article["body"] + " " + article["keywords"]
        if list_of_str and _find_news(query, text=text):
            to_return.append(text)
        if not list_of_str and _find_news(query, text=text):
            to_return.append(article)
    return to_return

FmpData 

FmpData(api_key: str = '', symbols: str | list = 'AAPL')

Bases: Toolkit

FMPData class for fetching data from Financial Modeling Prep API using the Toolkit class from financetoolkit package.

See financetoolkit for more details.

Initialise the FMP-backed toolkit for the given symbols.

Parameters:

Name Type Description Default
api_key str

The Financial Modeling Prep API key.

 ''
symbols str | list

One symbol or a list of symbols to load.

 'AAPL'
Source code in src/bbstrader/core/data.py 
814
815
816
817
818
819
820
821
def __init__(self, api_key: str = "", symbols: str | list = "AAPL"):
    """Initialise the FMP-backed toolkit for the given symbols.

    Args:
        api_key (str): The Financial Modeling Prep API key.
        symbols (str | list): One symbol or a list of symbols to load.
    """
    super().__init__(tickers=symbols, api_key=api_key)

TradeAction

Bases: Enum

An enumeration class for trade actions.

TradeSignal dataclass 

TradeSignal(id: int, symbol: str, action: TradeAction, price: float = None, stoplimit: float = None, sl: float = None, tp: float = None, comment: str = None)

Represents a trading signal generated by a trading system or strategy.

Notes

Attributes:

  • id (int): A unique identifier for the trade signal or the strategy.
  • symbol (str): The trading symbol (e.g., stock ticker, forex pair, crypto asset).
  • action (TradeAction): The trading action to perform. Must be an instance of the TradeAction enum (e.g., BUY, SELL).
  • price (float, optional): The price at which the trade should be executed.
  • stoplimit (float, optional): A stop-limit price for the trade. Must not be set without specifying a price.
  • sl (float, optional): A stop loss price for the trade.
  • tp (float, optional): A take profit price for the trade.
  • comment (str, optional): An optional comment or description related to the trade signal.

TradingMode

Bases: Enum

isbacktest 

isbacktest() -> bool

Return True if this mode is :attr:TradingMode.BACKTEST.

Source code in src/bbstrader/core/strategy.py 
162
163
164
def isbacktest(self) -> bool:
    """Return True if this mode is :attr:`TradingMode.BACKTEST`."""
    return self == TradingMode.BACKTEST

islive 

islive() -> bool

Return True if this mode is :attr:TradingMode.LIVE.

Source code in src/bbstrader/core/strategy.py 
166
167
168
def islive(self) -> bool:
    """Return True if this mode is :attr:`TradingMode.LIVE`."""
    return self == TradingMode.LIVE

Strategy

A Strategy() object encapsulates all calculation on market data that generate advisory signals to a Portfolio object. Thus all of the "strategy logic" resides within this class. We opted to separate out the Strategy and Portfolio objects for this backtester, since we believe this is more amenable to the situation of multiple strategies feeding "ideas" to a larger Portfolio, which then can handle its own risk (such as sector allocation, leverage). In higher frequency trading, the strategy and portfolio concepts will be tightly coupled and extremely hardware dependent.

At this stage in the event-driven backtester development there is no concept of an indicator or filter, such as those found in technical trading. These are also good candidates for creating a class hierarchy.

The strategy hierarchy is relatively simple as it consists of an abstract base class with a single pure virtual method for generating SignalEvent objects. Other methods are provided to check for pending orders, update trades from fills, and get updates from the portfolio.

calculate_signals abstractmethod 

calculate_signals(*args: Any, **kwargs: Any) -> List[TradeSignal] | None

Generate advisory trade signals from market data.

The single abstract method every strategy must implement; backtest and live engines both call it.

Parameters:

Name Type Description Default
args Any

Engine-supplied positional context (for example a market event).

 ()
kwargs Any

Engine-supplied keyword context.

 {}

Returns:

Type Description
List[TradeSignal] | None

List[TradeSignal] | None: The signals to act on, or None.
Source code in src/bbstrader/core/strategy.py 
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
@abstractmethod
def calculate_signals(self, *args: Any, **kwargs: Any) -> List[TradeSignal] | None:
    """Generate advisory trade signals from market data.

    The single abstract method every strategy must implement; backtest and
    live engines both call it.

    Args:
        args: Engine-supplied positional context (for example a market event).
        kwargs: Engine-supplied keyword context.

    Returns:
        List[TradeSignal] | None: The signals to act on, or None.
    """
    raise NotImplementedError("Should implement calculate_signals()")

check_pending_orders 

check_pending_orders(*args: Any, **kwargs: Any) -> None

Evaluate any pending orders for the current bar (optional hook).

Source code in src/bbstrader/core/strategy.py 
209
210
211
def check_pending_orders(self, *args: Any, **kwargs: Any) -> None:
    """Evaluate any pending orders for the current bar (optional hook)."""
    ...

get_update_from_portfolio 

get_update_from_portfolio(*args: Any, **kwargs: Any) -> None

Receive the latest positions/holdings from the portfolio (optional hook).

Source code in src/bbstrader/core/strategy.py 
213
214
215
def get_update_from_portfolio(self, *args: Any, **kwargs: Any) -> None:
    """Receive the latest positions/holdings from the portfolio (optional hook)."""
    ...

update_trades_from_fill 

update_trades_from_fill(*args: Any, **kwargs: Any) -> None

Update trade bookkeeping from a fill event (optional hook).

Source code in src/bbstrader/core/strategy.py 
217
218
219
def update_trades_from_fill(self, *args: Any, **kwargs: Any) -> None:
    """Update trade bookkeeping from a fill event (optional hook)."""
    ...

perform_period_end_checks 

perform_period_end_checks(*args: Any, **kwargs: Any) -> None

Run end-of-period maintenance such as risk checks (optional hook).

Source code in src/bbstrader/core/strategy.py 
221
222
223
def perform_period_end_checks(self, *args: Any, **kwargs: Any) -> None:
    """Run end-of-period maintenance such as risk checks (optional hook)."""
    ...

sma 

sma(values: ArrayLike, window: int) -> NDArray[np.float64]

Simple moving average over window bars (NaN-padded).

Source code in src/bbstrader/core/indicators.py 
64
65
66
67
68
69
70
71
72
73
74
def sma(values: ArrayLike, window: int) -> NDArray[np.float64]:
    """Simple moving average over ``window`` bars (NaN-padded)."""
    _check_window(window)
    arr = _as_float_array(values)
    out = np.full(arr.shape, np.nan)
    if arr.size < window:
        return out
    # Use a cumulative-sum sliding window for an O(n) average.
    cumsum = np.cumsum(np.insert(arr, 0, 0.0))
    out[window - 1 :] = (cumsum[window:] - cumsum[:-window]) / window
    return out

ema 

ema(values: ArrayLike, window: int) -> NDArray[np.float64]

Exponential moving average with span window (NaN until seeded).

The average is seeded with the SMA of the first window values, matching the common charting convention.

Source code in src/bbstrader/core/indicators.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
def ema(values: ArrayLike, window: int) -> NDArray[np.float64]:
    """Exponential moving average with span ``window`` (NaN until seeded).

    The average is seeded with the SMA of the first ``window`` values, matching
    the common charting convention.
    """
    _check_window(window)
    arr = _as_float_array(values)
    out = np.full(arr.shape, np.nan)
    if arr.size < window:
        return out
    alpha = 2.0 / (window + 1.0)
    prev = float(arr[:window].mean())
    out[window - 1] = prev
    for i in range(window, arr.size):
        prev = alpha * arr[i] + (1.0 - alpha) * prev
        out[i] = prev
    return out

wma 

wma(values: ArrayLike, window: int) -> NDArray[np.float64]

Linearly weighted moving average (most recent bar weighted highest).

Source code in src/bbstrader/core/indicators.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
def wma(values: ArrayLike, window: int) -> NDArray[np.float64]:
    """Linearly weighted moving average (most recent bar weighted highest)."""
    _check_window(window)
    arr = _as_float_array(values)
    out = np.full(arr.shape, np.nan)
    if arr.size < window:
        return out
    weights = np.arange(1.0, window + 1.0)
    denom = weights.sum()
    for i in range(window - 1, arr.size):
        out[i] = np.dot(arr[i - window + 1 : i + 1], weights) / denom
    return out

rolling_std 

rolling_std(values: ArrayLike, window: int, ddof: int = 0) -> NDArray[np.float64]

Rolling standard deviation over window bars (NaN-padded).

Source code in src/bbstrader/core/indicators.py 
111
112
113
114
115
116
117
118
119
120
def rolling_std(values: ArrayLike, window: int, ddof: int = 0) -> NDArray[np.float64]:
    """Rolling standard deviation over ``window`` bars (NaN-padded)."""
    _check_window(window)
    arr = _as_float_array(values)
    out = np.full(arr.shape, np.nan)
    if arr.size < window:
        return out
    for i in range(window - 1, arr.size):
        out[i] = arr[i - window + 1 : i + 1].std(ddof=ddof)
    return out

zscore 

zscore(values: ArrayLike, window: int) -> NDArray[np.float64]

Rolling z-score: (price - rolling_mean) / rolling_std.

Bars where the rolling standard deviation is zero yield NaN to avoid a divide-by-zero.

Source code in src/bbstrader/core/indicators.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
def zscore(values: ArrayLike, window: int) -> NDArray[np.float64]:
    """Rolling z-score: ``(price - rolling_mean) / rolling_std``.

    Bars where the rolling standard deviation is zero yield ``NaN`` to avoid a
    divide-by-zero.
    """
    _check_window(window)
    arr = _as_float_array(values)
    mean = sma(arr, window)
    std = rolling_std(arr, window)
    with np.errstate(invalid="ignore", divide="ignore"):
        out = np.where(std > 0, (arr - mean) / std, np.nan)
    return out

roc 

roc(values: ArrayLike, window: int) -> NDArray[np.float64]

Rate of change in percent over window bars.

Source code in src/bbstrader/core/indicators.py 
138
139
140
141
142
143
144
145
146
147
148
149
150
def roc(values: ArrayLike, window: int) -> NDArray[np.float64]:
    """Rate of change in percent over ``window`` bars."""
    _check_window(window)
    arr = _as_float_array(values)
    out = np.full(arr.shape, np.nan)
    if arr.size <= window:
        return out
    prior = arr[:-window]
    with np.errstate(invalid="ignore", divide="ignore"):
        out[window:] = np.where(
            prior != 0, (arr[window:] - prior) / prior * 100.0, np.nan
        )
    return out

rsi 

rsi(values: ArrayLike, window: int = 14) -> NDArray[np.float64]

Wilder's Relative Strength Index over window bars (NaN-padded).

Source code in src/bbstrader/core/indicators.py 
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
def rsi(values: ArrayLike, window: int = 14) -> NDArray[np.float64]:
    """Wilder's Relative Strength Index over ``window`` bars (NaN-padded)."""
    _check_window(window)
    arr = _as_float_array(values)
    out = np.full(arr.shape, np.nan)
    if arr.size <= window:
        return out
    delta = np.diff(arr)
    gains = np.where(delta > 0, delta, 0.0)
    losses = np.where(delta < 0, -delta, 0.0)
    # Wilder's smoothing: seed with the simple average of the first window.
    avg_gain = gains[:window].mean()
    avg_loss = losses[:window].mean()

    def _rsi_from(avg_gain: float, avg_loss: float) -> float:
        """Return the RSI value for a smoothed average gain and loss.

        Args:
            avg_gain (float): The smoothed average up-move over the window.
            avg_loss (float): The smoothed average down-move over the window.

        Returns:
            float: The RSI in [0, 100]; 100 when there are no losses.
        """
        if avg_loss == 0:
            return 100.0
        rs = avg_gain / avg_loss
        return 100.0 - (100.0 / (1.0 + rs))

    out[window] = _rsi_from(avg_gain, avg_loss)
    for i in range(window + 1, arr.size):
        avg_gain = (avg_gain * (window - 1) + gains[i - 1]) / window
        avg_loss = (avg_loss * (window - 1) + losses[i - 1]) / window
        out[i] = _rsi_from(avg_gain, avg_loss)
    return out

true_range 

true_range(high: ArrayLike, low: ArrayLike, close: ArrayLike) -> NDArray[np.float64]

True range: max(high-low, |high-prev_close|, |low-prev_close|).

Source code in src/bbstrader/core/indicators.py 
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
def true_range(
    high: ArrayLike, low: ArrayLike, close: ArrayLike
) -> NDArray[np.float64]:
    """True range: ``max(high-low, |high-prev_close|, |low-prev_close|)``."""
    h = _as_float_array(high)
    low_arr = _as_float_array(low)
    c = _as_float_array(close)
    if not (h.size == low_arr.size == c.size):
        raise ValueError("high, low and close must have the same length.")
    out = np.full(h.shape, np.nan)
    if h.size == 0:
        return out
    out[0] = h[0] - low_arr[0]
    prev_close = c[:-1]
    hl = h[1:] - low_arr[1:]
    hc = np.abs(h[1:] - prev_close)
    lc = np.abs(low_arr[1:] - prev_close)
    out[1:] = np.maximum.reduce([hl, hc, lc])
    return out

atr 

atr(high: ArrayLike, low: ArrayLike, close: ArrayLike, window: int = 14) -> NDArray[np.float64]

Average True Range using Wilder's smoothing (NaN-padded).

Source code in src/bbstrader/core/indicators.py 
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
def atr(
    high: ArrayLike, low: ArrayLike, close: ArrayLike, window: int = 14
) -> NDArray[np.float64]:
    """Average True Range using Wilder's smoothing (NaN-padded)."""
    _check_window(window)
    tr = true_range(high, low, close)
    out = np.full(tr.shape, np.nan)
    if tr.size < window:
        return out
    prev = float(np.nanmean(tr[:window]))
    out[window - 1] = prev
    for i in range(window, tr.size):
        prev = (prev * (window - 1) + tr[i]) / window
        out[i] = prev
    return out

bollinger_bands 

bollinger_bands(values: ArrayLike, window: int = 20, num_std: float = 2.0) -> Tuple[NDArray[np.float64], NDArray[np.float64], NDArray[np.float64]]

Bollinger Bands; returns (lower, middle, upper) arrays.

Source code in src/bbstrader/core/indicators.py 
228
229
230
231
232
233
234
235
236
237
238
def bollinger_bands(
    values: ArrayLike, window: int = 20, num_std: float = 2.0
) -> Tuple[NDArray[np.float64], NDArray[np.float64], NDArray[np.float64]]:
    """Bollinger Bands; returns ``(lower, middle, upper)`` arrays."""
    _check_window(window)
    arr = _as_float_array(values)
    middle = sma(arr, window)
    std = rolling_std(arr, window)
    upper = middle + num_std * std
    lower = middle - num_std * std
    return lower, middle, upper

macd 

macd(values: ArrayLike, fast: int = 12, slow: int = 26, signal: int = 9) -> Tuple[NDArray[np.float64], NDArray[np.float64], NDArray[np.float64]]

MACD; returns (macd_line, signal_line, histogram) arrays.

Source code in src/bbstrader/core/indicators.py 
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
def macd(
    values: ArrayLike,
    fast: int = 12,
    slow: int = 26,
    signal: int = 9,
) -> Tuple[NDArray[np.float64], NDArray[np.float64], NDArray[np.float64]]:
    """MACD; returns ``(macd_line, signal_line, histogram)`` arrays."""
    if fast >= slow:
        raise ValueError(f"fast ({fast}) must be smaller than slow ({slow}).")
    arr = _as_float_array(values)
    macd_line = ema(arr, fast) - ema(arr, slow)
    # The signal line is an EMA of the MACD line over its valid (non-NaN) tail.
    signal_line = np.full(arr.shape, np.nan)
    valid = ~np.isnan(macd_line)
    if valid.any():
        start = int(np.argmax(valid))
        tail = ema(macd_line[start:], signal)
        signal_line[start:] = tail
    histogram = macd_line - signal_line
    return macd_line, signal_line, histogram

stochastic 

stochastic(high: ArrayLike, low: ArrayLike, close: ArrayLike, k_window: int = 14, d_window: int = 3) -> Tuple[NDArray[np.float64], NDArray[np.float64]]

Stochastic oscillator; returns (%K, %D) arrays.

Source code in src/bbstrader/core/indicators.py 
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
def stochastic(
    high: ArrayLike,
    low: ArrayLike,
    close: ArrayLike,
    k_window: int = 14,
    d_window: int = 3,
) -> Tuple[NDArray[np.float64], NDArray[np.float64]]:
    """Stochastic oscillator; returns ``(%K, %D)`` arrays."""
    _check_window(k_window, "k_window")
    _check_window(d_window, "d_window")
    h = _as_float_array(high)
    low_arr = _as_float_array(low)
    c = _as_float_array(close)
    if not (h.size == low_arr.size == c.size):
        raise ValueError("high, low and close must have the same length.")
    percent_k = np.full(c.shape, np.nan)
    for i in range(k_window - 1, c.size):
        window_high = h[i - k_window + 1 : i + 1].max()
        window_low = low_arr[i - k_window + 1 : i + 1].min()
        span = window_high - window_low
        if span > 0:
            percent_k[i] = (c[i] - window_low) / span * 100.0
    percent_d = sma(percent_k, d_window)
    return percent_k, percent_d

donchian 

donchian(high: ArrayLike, low: ArrayLike, window: int = 20) -> Tuple[NDArray[np.float64], NDArray[np.float64]]

Donchian channel; returns (lower, upper) arrays over window bars.

The channel at bar i uses bars [i-window+1, i] (inclusive), so it is safe to compare the previous bar's channel against the current price for a breakout without look-ahead.

Source code in src/bbstrader/core/indicators.py 
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
def donchian(
    high: ArrayLike, low: ArrayLike, window: int = 20
) -> Tuple[NDArray[np.float64], NDArray[np.float64]]:
    """Donchian channel; returns ``(lower, upper)`` arrays over ``window`` bars.

    The channel at bar ``i`` uses bars ``[i-window+1, i]`` (inclusive), so it is
    safe to compare the *previous* bar's channel against the current price for a
    breakout without look-ahead.
    """
    _check_window(window)
    h = _as_float_array(high)
    low_arr = _as_float_array(low)
    if h.size != low_arr.size:
        raise ValueError("high and low must have the same length.")
    upper = np.full(h.shape, np.nan)
    lower = np.full(h.shape, np.nan)
    for i in range(window - 1, h.size):
        upper[i] = h[i - window + 1 : i + 1].max()
        lower[i] = low_arr[i - window + 1 : i + 1].min()
    return lower, upper

generate_signal 

generate_signal(id: int, symbol: str, action: TradeAction, price: float = None, stoplimit: float = None, sl: float = None, tp: float = None, comment: str = None) -> TradeSignal

Generates a trade signal for MetaTrader 5.

Parameters:

Name Type Description Default
id int

Unique identifier for the trade signal.

 required
symbol str

The symbol for which the trade signal is generated.

 required
action TradeAction

The action to be taken (e.g., BUY, SELL).

 required
price float

The price at which to execute the trade.

 None
stoplimit float

The stop limit price for the trade.

 None
sl float

The stop loss price for the trade.

 None
tp float

The take profit price for the trade.

 None
comment str

Additional comments for the trade.

 None

Returns:

Name Type Description
TradeSignal TradeSignal

A TradeSignal object containing the details of the trade signal.
Source code in src/bbstrader/core/strategy.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
def generate_signal(
    id: int,
    symbol: str,
    action: TradeAction,
    price: float = None,
    stoplimit: float = None,
    sl: float = None,
    tp: float = None,
    comment: str = None,
) -> TradeSignal:
    """
    Generates a trade signal for MetaTrader 5.

    Args:
        id (int): Unique identifier for the trade signal.
        symbol (str): The symbol for which the trade signal is generated.
        action (TradeAction): The action to be taken (e.g., BUY, SELL).
        price (float, optional): The price at which to execute the trade.
        stoplimit (float, optional): The stop limit price for the trade.
        sl (float, optional): The stop loss price for the trade.
        tp (float, optional): The take profit price for the trade.
        comment (str, optional): Additional comments for the trade.

    Returns:
        TradeSignal: A TradeSignal object containing the details of the trade signal.
    """
    return TradeSignal(
        id=id,
        symbol=symbol,
        action=action,
        price=price,
        stoplimit=stoplimit,
        sl=sl,
        tp=tp,
        comment=comment,
    )

broker

Broker-neutral execution abstraction.

A thin Broker interface decouples strategy/execution logic from any specific venue, so the same strategy can target MT5 today and IBKR / a crypto exchange later by swapping the adapter. PaperBroker is a fully in-memory simulated adapter useful for paper trading, tests, and as the reference implementation of the contract. Live adapters (e.g. an MT5 adapter over :mod:bbstrader.metatrader) implement the same methods.

BrokerOrder dataclass 

BrokerOrder(symbol: str, side: OrderSide, quantity: float, order_type: OrderType = OrderType.MARKET, price: Optional[float] = None, id: Optional[int] = None)

A venue-neutral order request and its assigned id once submitted.

BrokerPosition dataclass 

BrokerPosition(symbol: str, quantity: float, avg_price: float)

An open position: signed quantity and volume-weighted average price.

AccountInfo dataclass 

AccountInfo(cash: float, equity: float, currency: str = 'USD')

A snapshot of account cash, mark-to-market equity and currency.

Broker

Bases: ABC

The execution contract every venue adapter implements.

connect abstractmethod 
connect() -> bool

Open the connection to the venue; return True on success.

Source code in src/bbstrader/core/broker.py 
73
74
75
76
@abstractmethod
def connect(self) -> bool:
    """Open the connection to the venue; return True on success."""
    ...
disconnect abstractmethod 
disconnect() -> None

Close the connection to the venue.

Source code in src/bbstrader/core/broker.py 
78
79
80
81
@abstractmethod
def disconnect(self) -> None:
    """Close the connection to the venue."""
    ...
account abstractmethod 
account() -> AccountInfo

Return the current account snapshot (cash, equity, currency).

Source code in src/bbstrader/core/broker.py 
83
84
85
86
@abstractmethod
def account(self) -> AccountInfo:
    """Return the current account snapshot (cash, equity, currency)."""
    ...
get_price abstractmethod 
get_price(symbol: str) -> float

Return the latest market price for symbol.

Parameters:

Name Type Description Default
symbol str

The instrument to price.

 required

Returns:

Name Type Description
float float

The latest price.
Source code in src/bbstrader/core/broker.py 
88
89
90
91
92
93
94
95
96
97
98
@abstractmethod
def get_price(self, symbol: str) -> float:
    """Return the latest market price for ``symbol``.

    Args:
        symbol (str): The instrument to price.

    Returns:
        float: The latest price.
    """
    ...
submit_order abstractmethod 
submit_order(order: BrokerOrder) -> BrokerOrder

Submit order to the venue and return it with venue fields set.

Parameters:

Name Type Description Default
order BrokerOrder

The order to submit.

 required

Returns:

Name Type Description
BrokerOrder BrokerOrder

The submitted order, populated with its assigned id.
Source code in src/bbstrader/core/broker.py 
100
101
102
103
104
105
106
107
108
109
110
@abstractmethod
def submit_order(self, order: BrokerOrder) -> BrokerOrder:
    """Submit ``order`` to the venue and return it with venue fields set.

    Args:
        order (BrokerOrder): The order to submit.

    Returns:
        BrokerOrder: The submitted order, populated with its assigned id.
    """
    ...
positions abstractmethod 
positions() -> List[BrokerPosition]

Return the currently open positions.

Source code in src/bbstrader/core/broker.py 
112
113
114
115
@abstractmethod
def positions(self) -> List[BrokerPosition]:
    """Return the currently open positions."""
    ...
orders abstractmethod 
orders() -> List[BrokerOrder]

Return the currently open (resting) orders.

Source code in src/bbstrader/core/broker.py 
117
118
119
120
@abstractmethod
def orders(self) -> List[BrokerOrder]:
    """Return the currently open (resting) orders."""
    ...

PaperBroker 

PaperBroker(cash: float = 100000.0, currency: str = 'USD')

Bases: Broker

An in-memory simulated broker with immediate market fills.

Maintains cash, positions (volume-weighted average price) and an order log. Prices are set with :meth:set_price; market orders fill at the current price, limit/stop orders rest until :meth:set_price crosses their level.

Initialise the paper broker with starting cash.

Parameters:

Name Type Description Default
cash float

The opening cash balance.

 100000.0
currency str

The account currency code.

 'USD'
Source code in src/bbstrader/core/broker.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
def __init__(self, cash: float = 100000.0, currency: str = "USD") -> None:
    """Initialise the paper broker with starting cash.

    Args:
        cash (float): The opening cash balance.
        currency (str): The account currency code.
    """
    self._cash = float(cash)
    self.currency = currency
    self._positions: Dict[str, BrokerPosition] = {}
    self._orders: List[BrokerOrder] = []
    self._open_orders: List[BrokerOrder] = []
    self._prices: Dict[str, float] = {}
    self._next_id = 1
    self._connected = False
    self.realized_pnl = 0.0
connect 
connect() -> bool

Mark the broker connected; always succeeds for the paper broker.

Source code in src/bbstrader/core/broker.py 
149
150
151
152
def connect(self) -> bool:
    """Mark the broker connected; always succeeds for the paper broker."""
    self._connected = True
    return True
disconnect 
disconnect() -> None

Mark the broker disconnected.

Source code in src/bbstrader/core/broker.py 
154
155
156
def disconnect(self) -> None:
    """Mark the broker disconnected."""
    self._connected = False
set_price 
set_price(symbol: str, price: float) -> None

Update the market price and trigger any resting orders it crosses.

Source code in src/bbstrader/core/broker.py 
159
160
161
162
def set_price(self, symbol: str, price: float) -> None:
    """Update the market price and trigger any resting orders it crosses."""
    self._prices[symbol] = float(price)
    self._check_open_orders(symbol)
get_price 
get_price(symbol: str) -> float

Return the last price set for symbol.

Parameters:

Name Type Description Default
symbol str

The instrument to price.

 required

Returns:

Name Type Description
float float

The most recently set price.

Raises:

Type Description
KeyError

If no price has been set for symbol.
Source code in src/bbstrader/core/broker.py 
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
def get_price(self, symbol: str) -> float:
    """Return the last price set for ``symbol``.

    Args:
        symbol (str): The instrument to price.

    Returns:
        float: The most recently set price.

    Raises:
        KeyError: If no price has been set for ``symbol``.
    """
    if symbol not in self._prices:
        raise KeyError(f"No price set for {symbol}.")
    return self._prices[symbol]
account 
account() -> AccountInfo

Return the account snapshot (cash, mark-to-market equity, currency).

Source code in src/bbstrader/core/broker.py 
181
182
183
184
185
def account(self) -> AccountInfo:
    """Return the account snapshot (cash, mark-to-market equity, currency)."""
    return AccountInfo(
        cash=self._cash, equity=self.equity(), currency=self.currency
    )
equity 
equity() -> float

Return cash plus the mark-to-market value of all open positions.

Source code in src/bbstrader/core/broker.py 
187
188
189
190
191
192
193
def equity(self) -> float:
    """Return cash plus the mark-to-market value of all open positions."""
    market_value = sum(
        pos.quantity * self._prices.get(sym, pos.avg_price)
        for sym, pos in self._positions.items()
    )
    return self._cash + market_value
positions 
positions() -> List[BrokerPosition]

Return the open (non-zero quantity) positions.

Source code in src/bbstrader/core/broker.py 
195
196
197
def positions(self) -> List[BrokerPosition]:
    """Return the open (non-zero quantity) positions."""
    return [p for p in self._positions.values() if p.quantity != 0]
orders 
orders() -> List[BrokerOrder]

Return the resting (not yet filled) orders.

Source code in src/bbstrader/core/broker.py 
199
200
201
def orders(self) -> List[BrokerOrder]:
    """Return the resting (not yet filled) orders."""
    return list(self._open_orders)
submit_order 
submit_order(order: BrokerOrder) -> BrokerOrder

Submit an order, filling market orders immediately at the set price.

Market orders fill at order.price or the current market price; limit/stop orders rest and fill when :meth:set_price crosses them.

Parameters:

Name Type Description Default
order BrokerOrder

The order to submit. Its id is assigned here.

 required

Returns:

Name Type Description
BrokerOrder BrokerOrder

The same order with its assigned id.

Raises:

Type Description
ValueError

If order.quantity is not positive.
Source code in src/bbstrader/core/broker.py 
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
def submit_order(self, order: BrokerOrder) -> BrokerOrder:
    """Submit an order, filling market orders immediately at the set price.

    Market orders fill at ``order.price`` or the current market price;
    limit/stop orders rest and fill when :meth:`set_price` crosses them.

    Args:
        order (BrokerOrder): The order to submit. Its ``id`` is assigned here.

    Returns:
        BrokerOrder: The same order with its assigned ``id``.

    Raises:
        ValueError: If ``order.quantity`` is not positive.
    """
    if order.quantity <= 0:
        raise ValueError("order quantity must be positive.")
    order.id = self._next_id
    self._next_id += 1
    self._orders.append(order)
    if order.order_type is OrderType.MARKET:
        price = order.price or self.get_price(order.symbol)
        self._fill(order, price)
    else:
        self._open_orders.append(order)
        # A resting order may fill immediately if already crossed.
        if order.symbol in self._prices:
            self._check_open_orders(order.symbol)
    return order

data

FmpNews 

FmpNews(api: str)

Bases: object

FmpNews is responsible for retrieving financial news, press releases, and articles from Financial Modeling Prep (FMP).

FmpNews provides methods to fetch the latest stock, crypto, forex, and general financial news, as well as financial articles and press releases.

Parameters:

Name Type Description Default
api str

The API key for accessing FMP's news data.

 required
Example

fmp_news = FmpNews(api="your_api_key_here")

Source code in src/bbstrader/core/data.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
def __init__(self, api: str) -> None:
    """
    Args:
        api (str): The API key for accessing FMP's news data.

    Example:
        fmp_news = FmpNews(api="your_api_key_here")
    """
    if api is None:
        raise ValueError("API key is required For FmpNews")
    self.__api = api
get_articles 
get_articles(**kwargs: Any) -> List[Dict[str, Any]]

Fetch FMP articles with their HTML content stripped to plain text.

Parameters:

Name Type Description Default
kwargs Any

Optional page/limit paging parameters.

 {}

Returns:

Type Description
List[Dict[str, Any]]

List[Dict[str, Any]]: Records with title, date, content
List[Dict[str, Any]]

(plain text) and tickers.
Source code in src/bbstrader/core/data.py 
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
def get_articles(self, **kwargs: Any) -> List[Dict[str, Any]]:
    """Fetch FMP articles with their HTML content stripped to plain text.

    Args:
        kwargs (Any): Optional ``page``/``limit`` paging parameters.

    Returns:
        List[Dict[str, Any]]: Records with ``title``, ``date``, ``content``
        (plain text) and ``tickers``.
    """

    def html_parser(content: str) -> str:
        """Strip HTML markup from ``content`` to a single line of text.

        Args:
            content (str): The raw HTML article body.

        Returns:
            str: The extracted text with newlines removed.
        """
        soup = BeautifulSoup(content, "html.parser")
        text = soup.get_text(separator="\n")
        return text.replace("\n", "")

    articles = self._load_news("articles", **kwargs)
    df = pd.DataFrame(articles)
    df = df[["title", "date", "content", "tickers"]]
    df["content"] = df["content"].apply(html_parser)
    return df.to_dict(orient="records")  # type: ignore
get_releases 
get_releases(symbol: Optional[str] = None, **kwargs: Any) -> List[Dict[str, Any]]

Fetch the latest FMP press releases, optionally for one symbol.

Parameters:

Name Type Description Default
symbol Optional[str]

Restrict to a single symbol.

 None
kwargs Any

Optional paging/date parameters (see :meth:_load_news).

 {}

Returns:

Type Description
List[Dict[str, Any]]

List[Dict[str, Any]]: The press-release records.
Source code in src/bbstrader/core/data.py 
182
183
184
185
186
187
188
189
190
191
192
193
194
def get_releases(
    self, symbol: Optional[str] = None, **kwargs: Any
) -> List[Dict[str, Any]]:
    """Fetch the latest FMP press releases, optionally for one symbol.

    Args:
        symbol (Optional[str]): Restrict to a single symbol.
        kwargs (Any): Optional paging/date parameters (see :meth:`_load_news`).

    Returns:
        List[Dict[str, Any]]: The press-release records.
    """
    return self._load_news("press-releases", symbol, **kwargs)
get_stock_news 
get_stock_news(symbol: Optional[str] = None, **kwargs: Any) -> List[Dict[str, Any]]

Fetch the latest FMP stock news, optionally for one symbol.

Parameters:

Name Type Description Default
symbol Optional[str]

Restrict to a single symbol.

 None
kwargs Any

Optional paging/date parameters (see :meth:_load_news).

 {}

Returns:

Type Description
List[Dict[str, Any]]

List[Dict[str, Any]]: The stock-news records.
Source code in src/bbstrader/core/data.py 
196
197
198
199
200
201
202
203
204
205
206
207
208
def get_stock_news(
    self, symbol: Optional[str] = None, **kwargs: Any
) -> List[Dict[str, Any]]:
    """Fetch the latest FMP stock news, optionally for one symbol.

    Args:
        symbol (Optional[str]): Restrict to a single symbol.
        kwargs (Any): Optional paging/date parameters (see :meth:`_load_news`).

    Returns:
        List[Dict[str, Any]]: The stock-news records.
    """
    return self._load_news("stock", symbol, **kwargs)
get_crypto_news 
get_crypto_news(symbol: Optional[str] = None, **kwargs: Any) -> List[Dict[str, Any]]

Fetch the latest FMP crypto news, optionally for one symbol.

Parameters:

Name Type Description Default
symbol Optional[str]

Restrict to a single symbol.

 None
kwargs Any

Optional paging/date parameters (see :meth:_load_news).

 {}

Returns:

Type Description
List[Dict[str, Any]]

List[Dict[str, Any]]: The crypto-news records.
Source code in src/bbstrader/core/data.py 
210
211
212
213
214
215
216
217
218
219
220
221
222
def get_crypto_news(
    self, symbol: Optional[str] = None, **kwargs: Any
) -> List[Dict[str, Any]]:
    """Fetch the latest FMP crypto news, optionally for one symbol.

    Args:
        symbol (Optional[str]): Restrict to a single symbol.
        kwargs (Any): Optional paging/date parameters (see :meth:`_load_news`).

    Returns:
        List[Dict[str, Any]]: The crypto-news records.
    """
    return self._load_news("crypto", symbol, **kwargs)
get_forex_news 
get_forex_news(symbol: Optional[str] = None, **kwargs: Any) -> List[Dict[str, Any]]

Fetch the latest FMP forex news, optionally for one symbol.

Parameters:

Name Type Description Default
symbol Optional[str]

Restrict to a single symbol.

 None
kwargs Any

Optional paging/date parameters (see :meth:_load_news).

 {}

Returns:

Type Description
List[Dict[str, Any]]

List[Dict[str, Any]]: The forex-news records.
Source code in src/bbstrader/core/data.py 
224
225
226
227
228
229
230
231
232
233
234
235
236
def get_forex_news(
    self, symbol: Optional[str] = None, **kwargs: Any
) -> List[Dict[str, Any]]:
    """Fetch the latest FMP forex news, optionally for one symbol.

    Args:
        symbol (Optional[str]): Restrict to a single symbol.
        kwargs (Any): Optional paging/date parameters (see :meth:`_load_news`).

    Returns:
        List[Dict[str, Any]]: The forex-news records.
    """
    return self._load_news("forex", symbol, **kwargs)
parse_news 
parse_news(news: List[Dict[str, Any]], symbol: Optional[str] = None, **kwargs: Any) -> List[str]

Flatten and date-filter raw news records into text strings.

Keeps records published within the start/end window and, when a symbol is given, only those mentioning it.

Parameters:

Name Type Description Default
news List[Dict[str, Any]]

The raw records to parse.

 required
symbol Optional[str]

Restrict to records mentioning this symbol.

 None
kwargs Any

Optional start and end date bounds as YYYY-MM-DD HH:MM:SS strings.

 {}

Returns:

Type Description
List[str]

List[str]: One flattened text string per matching record.
Source code in src/bbstrader/core/data.py 
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
def parse_news(
    self, news: List[Dict[str, Any]], symbol: Optional[str] = None, **kwargs: Any
) -> List[str]:
    """Flatten and date-filter raw news records into text strings.

    Keeps records published within the ``start``/``end`` window and, when a
    ``symbol`` is given, only those mentioning it.

    Args:
        news (List[Dict[str, Any]]): The raw records to parse.
        symbol (Optional[str]): Restrict to records mentioning this symbol.
        kwargs (Any): Optional ``start`` and ``end`` date bounds as
            ``YYYY-MM-DD HH:MM:SS`` strings.

    Returns:
        List[str]: One flattened text string per matching record.
    """
    start = kwargs.get("start")
    end = kwargs.get("end")
    end_date = self._last_date(end) if end is not None else datetime.now().date()

    def parse_record(record: Dict[str, Any]) -> str:
        """Flatten a news record's text fields into a single string.

        Args:
            record (Dict[str, Any]): A raw news record.

        Returns:
            str: The symbol, title, text, content and tickers joined by spaces.
        """
        return " ".join(
            [
                record.pop("symbol", ""),
                record.pop("title", ""),
                record.pop("text", ""),
                record.pop("content", ""),
                record.pop("tickers", ""),
            ]
        )

    parsed_news = []
    for record in news:
        date = record.get("publishedDate")
        published_date = self._last_date(record.get("date", date)).date()  # type: ignore
        start_date = (
            self._last_date(start).date() if start is not None else published_date
        )
        if published_date >= start_date and published_date <= end_date:
            if symbol is not None:
                if record.get("symbol", "") == symbol or symbol in record.get(
                    "tickers", ""
                ):
                    parsed_news.append(parse_record(record))
            else:
                parsed_news.append(parse_record(record))
    return parsed_news
get_latest_articles 
get_latest_articles(articles: Optional[List[Dict[str, Any]]] = None, save: bool = False, **kwargs: Any) -> List[Dict[str, Any]]

Return the latest FMP articles, using a local CSV cache when fresh.

Reads latest_fmp_articles.csv if present and recent enough; otherwise downloads fresh articles and, when save is set, refreshes the cache.

Parameters:

Name Type Description Default
articles Optional[List[Dict[str, Any]]]

Pre-fetched articles to use instead of reading the cache or downloading.

 None
save bool

When True, persist the fetched articles to the cache CSV.

 False
kwargs Any

Optional end bound and paging parameters forwarded to :meth:get_articles.

 {}

Returns:

Type Description
List[Dict[str, Any]]

List[Dict[str, Any]]: The latest article records.
Source code in src/bbstrader/core/data.py 
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
def get_latest_articles(
    self,
    articles: Optional[List[Dict[str, Any]]] = None,
    save: bool = False,
    **kwargs: Any,
) -> List[Dict[str, Any]]:
    """Return the latest FMP articles, using a local CSV cache when fresh.

    Reads ``latest_fmp_articles.csv`` if present and recent enough; otherwise
    downloads fresh articles and, when ``save`` is set, refreshes the cache.

    Args:
        articles (Optional[List[Dict[str, Any]]]): Pre-fetched articles to
            use instead of reading the cache or downloading.
        save (bool): When True, persist the fetched articles to the cache CSV.
        kwargs (Any): Optional ``end`` bound and paging parameters forwarded
            to :meth:`get_articles`.

    Returns:
        List[Dict[str, Any]]: The latest article records.
    """
    end = kwargs.get("end")
    now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    end_date = self._last_date(end) if end is not None else self._last_date(now)
    if articles is None:
        try:
            articles = pd.read_csv("latest_fmp_articles.csv")  # type: ignore
            articles = articles.to_dict(orient="records")  # type: ignore
            if self._last_date(articles[0]["date"]).hour < end_date.hour:  # type: ignore
                articles = self.get_articles(**kwargs)
            else:
                return articles  # type: ignore
        except FileNotFoundError:
            articles = self.get_articles(**kwargs)

    if save and len(articles) > 0:
        df = pd.DataFrame(articles)
        df.to_csv("latest_fmp_articles.csv", index=False)
    return articles
get_news 
get_news(query: str, source: str = 'articles', articles: Optional[List[Dict[str, Any]]] = None, symbol: Optional[str] = None, **kwargs: Any) -> List[str]

Retrieves relevant financial news based on the specified source.

Parameters:

Name Type Description Default
query str

The search query or keyword for filtering news, may also be a ticker.

 required
source str

The news source to retrieve from. Defaults to "articles". Available options: "articles", "releases", "stock", "crypto", "forex".

 'articles'
articles list

List of pre-fetched articles to use when source="articles". Defaults to None.

 None
symbol str

The financial asset symbol (e.g., "AAPL" for stocks, "BTC" for crypto). Defaults to None.

 None
**kwargs dict

Additional arguments required for fetching news data. May include: - start (str): The start period for news retrieval (YYY-MM-DD) - end (str): The end period for news retrieval (YYY-MM-DD) - page (int): The number of page to load for each news - limit (int): Maximum Responses per API Call

 {}

Returns:

Type Description
List[str]

list[dict]: A list of filtered news articles relevant to the query. Returns an empty list if no relevant news is found.
Source code in src/bbstrader/core/data.py 
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
def get_news(
    self,
    query: str,
    source: str = "articles",
    articles: Optional[List[Dict[str, Any]]] = None,
    symbol: Optional[str] = None,
    **kwargs: Any,
) -> List[str]:
    """
    Retrieves relevant financial news based on the specified source.

    Args:
        query (str): The search query or keyword for filtering news, may also be a ticker.
        source (str, optional): The news source to retrieve from. Defaults to "articles".
                                Available options: "articles", "releases", "stock", "crypto", "forex".
        articles (list, optional): List of pre-fetched articles to use when source="articles". Defaults to None.
        symbol (str, optional): The financial asset symbol (e.g., "AAPL" for stocks, "BTC" for crypto). Defaults to None.
        **kwargs (dict):
            Additional arguments required for fetching news data. May include:
            - start (str): The start period for news retrieval (YYY-MM-DD)
            - end (str): The end period for news retrieval (YYY-MM-DD)
            - page (int): The number  of page to load  for each news
            - limit (int): Maximum Responses per API Call

    Returns:
        list[dict]: A list of filtered news articles relevant to the query.
                    Returns an empty list if no relevant news is found.
    """
    query = _get_search_query(query)
    if symbol is not None:
        symbol = symbol.replace("-", "").split("=")[
            0
        ]  # if symbol is a yahoo finance ticker
    source_methods = {
        "articles": lambda: self.get_latest_articles(
            articles=articles, save=True, **kwargs
        ),
        "releases": lambda: self.get_releases(symbol=symbol, **kwargs),
        "stock": lambda: self.get_stock_news(symbol=symbol, **kwargs),
        "crypto": lambda: self.get_crypto_news(symbol=symbol, **kwargs),
        "forex": lambda: self.get_forex_news(symbol=symbol, **kwargs),
    }
    news_source = source_methods.get(source, lambda: [])()
    if source == "articles":
        symbol = None  # Articles do not require a symbol filter
    news = self.parse_news(news_source, symbol=symbol, **kwargs)
    return _filter_news(news, query)

FinancialNews

Bases: object

The FinancialNews class provides methods to fetch financial news, articles, and discussions from various sources such as Yahoo Finance, Google Finance, Reddit, Coindesk and Twitter. It also supports retrieving news using Financial Modeling Prep (FMP).

get_yahoo_finance_news 
get_yahoo_finance_news(query: str, asset_type: str = 'stock', n_news: int = 10) -> List[str]

Fetches recent Yahoo Finance news headlines for a given financial asset.

Parameters:

Name Type Description Default
query str

The asset symbol or name (e.g., "AAPL").

 required
asset_type str

The type of asset (e.g., "stock", "etf"). Defaults to "stock", supported types include: - "stock": Stock symbols (e.g., AAPL, MSFT) - "etf": Exchange-traded funds (e.g., SPY, QQQ) - "future": Futures contracts (e.g., CL=F for crude oil) - "forex": Forex pairs (e.g., EURUSD=X, USDJPY=X) - "crypto": Cryptocurrency pairs (e.g., BTC-USD, ETH-USD) - "index": Stock market indices (e.g., ^GSPC for S&P 500)

 'stock'
n_news int

The number of news headlines to return. Defaults to 10.

 10
Note

For commotities and bonds, use the "Future" asset type.

Returns:

Type Description
List[str]

list[str]: A list of Yahoo Finance news headlines relevant to the query.
Source code in src/bbstrader/core/data.py 
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
def get_yahoo_finance_news(
    self, query: str, asset_type: str = "stock", n_news: int = 10
) -> List[str]:
    """
    Fetches recent Yahoo Finance news headlines for a given financial asset.

    Args:
        query (str): The asset symbol or name (e.g., "AAPL").
        asset_type (str, optional): The type of asset (e.g., "stock", "etf"). Defaults to "stock",
            supported types include:
            - "stock": Stock symbols (e.g., AAPL, MSFT)
            - "etf": Exchange-traded funds (e.g., SPY, QQQ)
            - "future": Futures contracts (e.g., CL=F for crude oil)
            - "forex": Forex pairs (e.g., EURUSD=X, USDJPY=X)
            - "crypto": Cryptocurrency pairs (e.g., BTC-USD, ETH-USD)
            - "index": Stock market indices (e.g., ^GSPC for S&P 500)
        n_news (int, optional): The number of news headlines to return. Defaults to 10.

    Note:
        For commotities and bonds, use the "Future" asset type.

    Returns:
        list[str]: A list of Yahoo Finance news headlines relevant to the query.
    """
    if asset_type == "forex" or asset_type == "future":
        assert "=" in query, (
            "Forex query must contain '=' for currency pairs (e.g., EURUSD=X, CL=F)"
        )
    if asset_type == "crypto":
        assert "-" in query, (
            "Crypto query must contain '-' for crypto pairs (e.g., BTC-USD, ETH-USD)"
        )
    if asset_type == "index":
        assert query.startswith("^"), (
            "Index query must start with '^' (e.g., ^GSPC for S&P 500)"
        )
    url = (
        f"https://finance.yahoo.com/quote/{query}/news"
        if asset_type in ["stock", "etf", "index", "future", "forex"]
        else "https://finance.yahoo.com/news"
    )
    return self._fetch_news(url, query, n_news, "h3")
get_google_finance_news 
get_google_finance_news(query: str, asset_type: str = 'stock', n_news: int = 10) -> List[str]

Fetches recent Google Finance news headlines for a given financial asset.

Parameters:

Name Type Description Default
query str

The asset symbol or name (e.g., "AAPL").

 required
asset_type str

The type of asset (e.g., "stock", "crypto"). Defaults to "stock". Supported types include: - "stock": Stock symbols (e.g., AAPL, MSFT) - "etf": Exchange-traded funds (e.g., SPY, QQQ) - "future": Futures contracts (e.g., CL=F or crude oil) - "forex": Forex pairs (e.g., EURUSD, USDJPY) - "crypto": Cryptocurrency pairs (e.g., BTCUSD, ETHUSD)

 'stock'
n_news int

The number of news headlines to return. Defaults to 10.

 10

Returns:

Type Description
List[str]

list[str]: A list of Google Finance news headlines relevant to the query.
Source code in src/bbstrader/core/data.py 
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
def get_google_finance_news(
    self, query: str, asset_type: str = "stock", n_news: int = 10
) -> List[str]:
    """
    Fetches recent Google Finance news headlines for a given financial asset.

    Args:
        query (str): The asset symbol or name (e.g., "AAPL").
        asset_type (str, optional): The type of asset (e.g., "stock", "crypto"). Defaults to "stock".
            Supported types include:
            - "stock": Stock symbols (e.g., AAPL, MSFT)
            - "etf": Exchange-traded funds (e.g., SPY, QQQ)
            - "future": Futures contracts (e.g., CL=F or crude oil)
            - "forex": Forex pairs (e.g., EURUSD, USDJPY)
            - "crypto": Cryptocurrency pairs (e.g., BTCUSD, ETHUSD)
        n_news (int, optional): The number of news headlines to return. Defaults to 10.

    Returns:
        list[str]: A list of Google Finance news headlines relevant to the query.
    """
    search_terms = {
        "stock": f"{query} stock OR {query} shares OR {query} market",
        "etf": f"{query} ETF OR {query} fund OR {query} exchange-traded fund",
        "future": f"{query} futures OR {query} price OR {query} market",
        "forex": f"{query} forex OR {query} exchange rate OR {query} market",
        "crypto": f"{query} cryptocurrency OR {query} price OR {query} market",
        "index": f"{query} index OR {query} stock market OR {query} performance",
    }
    search_query = search_terms.get(asset_type, query)
    url = f"https://news.google.com/search?q={search_query.replace(' ', '+')}"
    return self._fetch_news(url, query, n_news, "a")
get_reddit_posts 
get_reddit_posts(symbol: str, client_id=None, client_secret=None, user_agent=None, asset_class='stock', n_posts=10) -> List[str]

Fetches recent Reddit posts related to a financial asset.

This method queries relevant subreddits for posts mentioning the specified symbol and returns posts based on the selected asset class (e.g., stock, forex, crypto). The function uses the PRAW library to interact with Reddit's API.

Parameters:

Name Type Description Default
symbol str

The financial asset's symbol or name to search for.

 required
client_id str

Reddit API client ID for authentication.

 None
client_secret str

Reddit API client secret.

 None
user_agent str

Reddit API user agent.

 None
asset_class str

The type of financial asset. Defaults to "stock". - "stock": Searches in stock-related subreddits (e.g., wallstreetbets, stocks). - "forex": Searches in forex-related subreddits. - "commodities": Searches in commodity-related subreddits (e.g., gold, oil). - "etf": Searches in ETF-related subreddits. - "future": Searches in futures and options trading subreddits. - "crypto": Searches in cryptocurrency-related subreddits. - If an unrecognized asset class is provided, defaults to stock-related subreddits.

 'stock'
n_posts int

The number of posts to return per subreddit. Defaults to 10.

 10

Returns:

Type Description
List[str]

list[str]: A list of Reddit post contents matching the query. Each entry contains the post title and body. If no posts are found or an error occurs, returns an empty list.

Raises:

Type Description
PRAWException

If an error occurs while interacting with Reddit's API.
Example

get_reddit_posts(symbol="AAPL", client_id="your_id", client_secret="your_secret", user_agent="your_agent", asset_class="stock", n_posts=5) ["Apple stock is rallying today due to strong earnings.", "Should I buy $AAPL now?", ...]

Notes
  • Requires valid Reddit API credentials.
Source code in src/bbstrader/core/data.py 
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
def get_reddit_posts(
    self,
    symbol: str,
    client_id=None,
    client_secret=None,
    user_agent=None,
    asset_class="stock",
    n_posts=10,
) -> List[str]:
    """
    Fetches recent Reddit posts related to a financial asset.

    This method queries relevant subreddits for posts mentioning the specified symbol
    and returns posts based on the selected asset class (e.g., stock, forex, crypto).
    The function uses the PRAW library to interact with Reddit's API.

    Args:
        symbol (str): The financial asset's symbol or name to search for.
        client_id (str, optional): Reddit API client ID for authentication.
        client_secret (str, optional): Reddit API client secret.
        user_agent (str, optional): Reddit API user agent.
        asset_class (str, optional): The type of financial asset. Defaults to "stock".
            - "stock": Searches in stock-related subreddits (e.g., wallstreetbets, stocks).
            - "forex": Searches in forex-related subreddits.
            - "commodities": Searches in commodity-related subreddits (e.g., gold, oil).
            - "etf": Searches in ETF-related subreddits.
            - "future": Searches in futures and options trading subreddits.
            - "crypto": Searches in cryptocurrency-related subreddits.
            - If an unrecognized asset class is provided, defaults to stock-related subreddits.
        n_posts (int, optional): The number of posts to return per subreddit. Defaults to 10.

    Returns:
        list[str]: A list of Reddit post contents matching the query.
                Each entry contains the post title and body.
                If no posts are found or an error occurs, returns an empty list.

    Raises:
        praw.exceptions.PRAWException: If an error occurs while interacting with Reddit's API.

    Example:
        >>> get_reddit_posts(symbol="AAPL", client_id="your_id", client_secret="your_secret", user_agent="your_agent", asset_class="stock", n_posts=5)
        ["Apple stock is rallying today due to strong earnings.", "Should I buy $AAPL now?", ...]

    Notes:
        - Requires valid Reddit API credentials.
    """

    praw = _require("praw", "social")
    reddit = praw.Reddit(
        client_id=client_id,
        client_secret=client_secret,
        user_agent=user_agent,
        check_for_updates=False,
        comment_kind="t1",
        message_kind="t4",
        redditor_kind="t2",
        submission_kind="t3",
        subreddit_kind="t5",
        trophy_kind="t6",
        oauth_url="https://oauth.reddit.com",
        reddit_url="https://www.reddit.com",
        short_url="https://redd.it",
        timeout=16,
        ratelimit_seconds=5,
    )
    assert reddit.read_only
    subreddit_mapping = {
        "stock": ["wallstreetbets", "stocks", "investing", "StockMarket"],
        "forex": ["Forex", "ForexTrading", "DayTrading"],
        "etfs": ["ETFs", "investing"],
        "futures": [
            "FuturesTrading",
            "OptionsTrading",
            "DayTrading",
            "Commodities",
            "Gold",
            "Silverbugs",
            "oil",
        ],
        "crypto": ["CryptoCurrency", "Bitcoin", "ethereum", "altcoin"],
    }
    try:
        subreddits = subreddit_mapping.get(asset_class.lower(), ["stocks"])
    except Exception:
        return []

    posts = []
    for sub in subreddits:
        subreddit = reddit.subreddit(sub)
        query = _get_search_query(symbol)
        all_posts = subreddit.search(query, limit=n_posts)
        for post in all_posts:
            text = post.title + " " + post.selftext
            if _find_news(query, text):
                posts.append(text)
    return posts
get_twitter_posts 
get_twitter_posts(query: str, asset_type: str = 'stock', bearer: Optional[str] = None, api_key: Optional[str] = None, api_secret: Optional[str] = None, access_token: Optional[str] = None, access_secret: Optional[str] = None, n_posts: int = 10) -> List[str]

Fetches recent tweets related to a financial asset.

This method queries Twitter for recent posts mentioning the specified asset and filters the results based on the asset type (e.g., stock, forex, crypto). The function uses the Tweepy API to fetch tweets and returns a list of tweet texts.

Parameters:

Name Type Description Default
query str

The main keyword to search for (e.g., a stock ticker or asset name).

 required
asset_type str

The type of financial asset. Defaults to "stock". - "stock": Searches for tweets mentioning the stock or shares. - "forex": Searches for tweets mentioning foreign exchange (forex) or currency. - "crypto": Searches for tweets mentioning cryptocurrency or related terms. - "commodity": Searches for tweets mentioning commodities or futures trading. - "index": Searches for tweets mentioning stock market indices. - "bond": Searches for tweets mentioning bonds or fixed income securities. - If an unrecognized asset type is provided, defaults to general finance-related tweets.

 'stock'
bearer str

Twitter API bearer token for authentication.

 None
api_key str

Twitter API consumer key.

 None
api_secret str

Twitter API consumer secret.

 None
access_token str

Twitter API access token.

 None
access_secret str

Twitter API access token secret.

 None
n_posts int

The number of tweets to return. Defaults to 10.

 10

Returns:

Type Description
List[str]

list[str]: A list of up to n_posts tweet texts matching the query. If no tweets are found or an API error occurs, returns an empty list.

Raises:

Type Description
TweepyException

If an error occurs while making the Twitter API request.
Example

get_twitter_posts(query="AAPL", asset_type="stock", bearer="YOUR_BEARER_TOKEN", n_posts=5) ["Apple stock surges after strong earnings!", "Is $AAPL a buy at this price?", ...]

Source code in src/bbstrader/core/data.py 
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
def get_twitter_posts(
    self,
    query: str,
    asset_type: str = "stock",
    bearer: Optional[str] = None,
    api_key: Optional[str] = None,
    api_secret: Optional[str] = None,
    access_token: Optional[str] = None,
    access_secret: Optional[str] = None,
    n_posts: int = 10,
) -> List[str]:
    """
    Fetches recent tweets related to a financial asset.

    This method queries Twitter for recent posts mentioning the specified asset
    and filters the results based on the asset type (e.g., stock, forex, crypto).
    The function uses the Tweepy API to fetch tweets and returns a list of tweet texts.

    Args:
        query (str): The main keyword to search for (e.g., a stock ticker or asset name).
        asset_type (str, optional): The type of financial asset. Defaults to "stock".
            - "stock": Searches for tweets mentioning the stock or shares.
            - "forex": Searches for tweets mentioning foreign exchange (forex) or currency.
            - "crypto": Searches for tweets mentioning cryptocurrency or related terms.
            - "commodity": Searches for tweets mentioning commodities or futures trading.
            - "index": Searches for tweets mentioning stock market indices.
            - "bond": Searches for tweets mentioning bonds or fixed income securities.
            - If an unrecognized asset type is provided, defaults to general finance-related tweets.
        bearer (str, optional): Twitter API bearer token for authentication.
        api_key (str, optional): Twitter API consumer key.
        api_secret (str, optional): Twitter API consumer secret.
        access_token (str, optional): Twitter API access token.
        access_secret (str, optional): Twitter API access token secret.
        n_posts (int, optional): The number of tweets to return. Defaults to 10.

    Returns:
        list[str]: A list of up to `n_posts` tweet texts matching the query.
                If no tweets are found or an API error occurs, returns an empty list.

    Raises:
        tweepy.TweepyException: If an error occurs while making the Twitter API request.

    Example:
        >>> get_twitter_posts(query="AAPL", asset_type="stock", bearer="YOUR_BEARER_TOKEN", n_posts=5)
        ["Apple stock surges after strong earnings!", "Is $AAPL a buy at this price?", ...]
    """
    tweepy = _require("tweepy", "social")
    client = tweepy.Client(
        bearer_token=bearer,
        consumer_key=api_key,
        consumer_secret=api_secret,
        access_token=access_token,
        access_token_secret=access_secret,
    )
    asset_queries = {
        "stock": f"{query} stock OR {query} shares -is:retweet lang:en",
        "forex": f"{query} forex OR {query} currency -is:retweet lang:en",
        "crypto": f"{query} cryptocurrency OR {query} crypto OR #{query} -is:retweet lang:en",
        "commodity": f"{query} commodity OR {query} futures OR {query} trading -is:retweet lang:en",
        "index": f"{query} index OR {query} market -is:retweet lang:en",
        "bond": f"{query} bonds OR {query} fixed income -is:retweet lang:en",
    }
    # Get the correct query based on the asset type
    search = asset_queries.get(
        asset_type.lower(), f"{query} finance -is:retweet lang:en"
    )
    try:
        tweets = client.search_recent_tweets(
            query=search, max_results=100, tweet_fields=["text"]
        )
        query = _get_search_query(query)
        news = [tweet.text for tweet in tweets.data] if tweets.data else []  # type: ignore
        return _filter_news(news, query)[:n_posts]
    except tweepy.TweepyException:
        return []
get_fmp_news 
get_fmp_news(api: str | None = None) -> FmpNews

Return an :class:FmpNews client for the given API key.

Parameters:

Name Type Description Default
api str | None

The Financial Modeling Prep API key.

 None

Returns:

Name Type Description
FmpNews FmpNews

A news client bound to api.
Source code in src/bbstrader/core/data.py 
689
690
691
692
693
694
695
696
697
698
def get_fmp_news(self, api: str | None = None) -> FmpNews:
    """Return an :class:`FmpNews` client for the given API key.

    Args:
        api (str | None): The Financial Modeling Prep API key.

    Returns:
        FmpNews: A news client bound to ``api``.
    """
    return FmpNews(api=api)  # type: ignore
get_coindesk_news 
get_coindesk_news(query='', lang: Literal['EN', 'ES', 'TR', 'FR', 'JP', 'PT'] = 'EN', limit=10, list_of_str=False) -> List[str] | List[dict]

Fetches and filters recent news articles from CoinDesk's News API.

Parameters:

Name Type Description Default
query

str, optional A search term to filter articles by title, body, or keywords. If empty, all articles are returned without filtering (default is "").

 required
lang

Literal["EN", "ES", "TR", "FR", "JP", "PT"], optional Language in which to fetch news articles. Supported languages: English (EN), Spanish (ES), Turkish (TR), French (FR), Japanese (JP), and Portuguese (PT). Default is "EN".

 required
limit

int, optional Maximum number of articles to retrieve. Default is 50.

 required
list_of_str

bool, optional If True, returns a list of strings (concatenated article content). If False, returns a list of filtered article dictionaries. Default is False.

 required

Returns:

Type Description
List[str] | List[dict]

List[str] | List[dict] - If query is empty: returns a list of filtered article dictionaries. - If query is provided: - Returns a list of strings if list_of_str=True. - Returns a list of filtered article dictionaries otherwise.
Each article dictionary contains the following fields
  • 'published_on': datetime of publication
  • 'title': article headline
  • 'subtitle': secondary headline
  • 'url': direct link to the article
  • 'body': article content
  • 'keywords': associated tags
  • 'sentiment': sentiment label
  • 'status': publication status
Notes
  • Articles marked as sponsored are automatically excluded.
Source code in src/bbstrader/core/data.py 
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
def get_coindesk_news(
    self,
    query="",
    lang: Literal["EN", "ES", "TR", "FR", "JP", "PT"] = "EN",
    limit=10,
    list_of_str=False,
) -> List[str] | List[dict]:
    """
    Fetches and filters recent news articles from CoinDesk's News API.

    Args:
        query : str, optional
            A search term to filter articles by title, body, or keywords.
            If empty, all articles are returned without filtering (default is "").

        lang : Literal["EN", "ES", "TR", "FR", "JP", "PT"], optional
            Language in which to fetch news articles. Supported languages:
            English (EN), Spanish (ES), Turkish (TR), French (FR), Japanese (JP), and Portuguese (PT).
            Default is "EN".

        limit : int, optional
            Maximum number of articles to retrieve. Default is 50.

        list_of_str : bool, optional
            If True, returns a list of strings (concatenated article content).
            If False, returns a list of filtered article dictionaries.
            Default is False.

    Returns:
        List[str] | List[dict]
            - If `query` is empty: returns a list of filtered article dictionaries.
            - If `query` is provided:
                - Returns a list of strings if `list_of_str=True`.
                - Returns a list of filtered article dictionaries otherwise.

    Each article dictionary contains the following fields:
        - 'published_on': datetime of publication
        - 'title': article headline
        - 'subtitle': secondary headline
        - 'url': direct link to the article
        - 'body': article content
        - 'keywords': associated tags
        - 'sentiment': sentiment label
        - 'status': publication status

    Notes:
        - Articles marked as sponsored are automatically excluded.
    """
    maximum = 100
    if limit > maximum:
        raise ValueError(f"Number of total news articles allowed is {maximum}")
    try:
        response = requests.get(
            "https://data-api.coindesk.com/news/v1/article/list",
            params={"lang": lang, "limit": limit},
            headers={"Content-type": "application/json; charset=UTF-8"},
        )
        response.raise_for_status()
        json_response = response.json()
    except requests.exceptions.RequestException:
        return []
    if (
        response.status_code != 200
        or "Data" not in json_response
        or len(json_response["Data"]) == 0
    ):
        return []
    articles = json_response["Data"]
    to_keep = [
        "PUBLISHED_ON",
        "TITLE",
        "SUBTITLE",
        "URL",
        "BODY",
        "KEYWORDS",
        "SENTIMENT",
        "STATUS",
    ]
    filtered_articles = []
    for article in articles:
        keys = article.keys()
        filtered_articles.append(
            {
                k.lower(): article[k]
                if k in keys and k != "PUBLISHED_ON"
                else datetime.fromtimestamp(article[k])
                for k in to_keep
                if article[k] is not None and "sponsored" not in str(article[k])
            }
        )
    if query == "" or len(filtered_articles) == 0:
        return filtered_articles
    to_return = []
    query = _get_search_query(query)
    for article in filtered_articles:
        if not all(k in article for k in ("title", "body", "keywords")):
            continue
        text = article["title"] + " " + article["body"] + " " + article["keywords"]
        if list_of_str and _find_news(query, text=text):
            to_return.append(text)
        if not list_of_str and _find_news(query, text=text):
            to_return.append(article)
    return to_return

FmpData 

FmpData(api_key: str = '', symbols: str | list = 'AAPL')

Bases: Toolkit

FMPData class for fetching data from Financial Modeling Prep API using the Toolkit class from financetoolkit package.

See financetoolkit for more details.

Initialise the FMP-backed toolkit for the given symbols.

Parameters:

Name Type Description Default
api_key str

The Financial Modeling Prep API key.

 ''
symbols str | list

One symbol or a list of symbols to load.

 'AAPL'
Source code in src/bbstrader/core/data.py 
814
815
816
817
818
819
820
821
def __init__(self, api_key: str = "", symbols: str | list = "AAPL"):
    """Initialise the FMP-backed toolkit for the given symbols.

    Args:
        api_key (str): The Financial Modeling Prep API key.
        symbols (str | list): One symbol or a list of symbols to load.
    """
    super().__init__(tickers=symbols, api_key=api_key)

indicators

Vectorized technical indicators.

A small, dependency-free indicator library built on NumPy so that it can be used identically from backtest strategies (BacktestStrategy) and live strategies (LiveStrategy). Every function operates on a 1-D array of prices (or OHLC arrays) and returns an array of the same length as the input, left-padded with NaN where there is not yet enough history. This lines up with the output of BaseStrategy.get_asset_values(...) so an indicator value at index -1 corresponds to the latest bar.

The implementations are plain NumPy (no third-party TA dependency), which keeps the install lean and leaves the door open to JIT-compiling the hot paths later.

sma 

sma(values: ArrayLike, window: int) -> NDArray[np.float64]

Simple moving average over window bars (NaN-padded).

Source code in src/bbstrader/core/indicators.py 
64
65
66
67
68
69
70
71
72
73
74
def sma(values: ArrayLike, window: int) -> NDArray[np.float64]:
    """Simple moving average over ``window`` bars (NaN-padded)."""
    _check_window(window)
    arr = _as_float_array(values)
    out = np.full(arr.shape, np.nan)
    if arr.size < window:
        return out
    # Use a cumulative-sum sliding window for an O(n) average.
    cumsum = np.cumsum(np.insert(arr, 0, 0.0))
    out[window - 1 :] = (cumsum[window:] - cumsum[:-window]) / window
    return out

ema 

ema(values: ArrayLike, window: int) -> NDArray[np.float64]

Exponential moving average with span window (NaN until seeded).

The average is seeded with the SMA of the first window values, matching the common charting convention.

Source code in src/bbstrader/core/indicators.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
def ema(values: ArrayLike, window: int) -> NDArray[np.float64]:
    """Exponential moving average with span ``window`` (NaN until seeded).

    The average is seeded with the SMA of the first ``window`` values, matching
    the common charting convention.
    """
    _check_window(window)
    arr = _as_float_array(values)
    out = np.full(arr.shape, np.nan)
    if arr.size < window:
        return out
    alpha = 2.0 / (window + 1.0)
    prev = float(arr[:window].mean())
    out[window - 1] = prev
    for i in range(window, arr.size):
        prev = alpha * arr[i] + (1.0 - alpha) * prev
        out[i] = prev
    return out

wma 

wma(values: ArrayLike, window: int) -> NDArray[np.float64]

Linearly weighted moving average (most recent bar weighted highest).

Source code in src/bbstrader/core/indicators.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
def wma(values: ArrayLike, window: int) -> NDArray[np.float64]:
    """Linearly weighted moving average (most recent bar weighted highest)."""
    _check_window(window)
    arr = _as_float_array(values)
    out = np.full(arr.shape, np.nan)
    if arr.size < window:
        return out
    weights = np.arange(1.0, window + 1.0)
    denom = weights.sum()
    for i in range(window - 1, arr.size):
        out[i] = np.dot(arr[i - window + 1 : i + 1], weights) / denom
    return out

rolling_std 

rolling_std(values: ArrayLike, window: int, ddof: int = 0) -> NDArray[np.float64]

Rolling standard deviation over window bars (NaN-padded).

Source code in src/bbstrader/core/indicators.py 
111
112
113
114
115
116
117
118
119
120
def rolling_std(values: ArrayLike, window: int, ddof: int = 0) -> NDArray[np.float64]:
    """Rolling standard deviation over ``window`` bars (NaN-padded)."""
    _check_window(window)
    arr = _as_float_array(values)
    out = np.full(arr.shape, np.nan)
    if arr.size < window:
        return out
    for i in range(window - 1, arr.size):
        out[i] = arr[i - window + 1 : i + 1].std(ddof=ddof)
    return out

zscore 

zscore(values: ArrayLike, window: int) -> NDArray[np.float64]

Rolling z-score: (price - rolling_mean) / rolling_std.

Bars where the rolling standard deviation is zero yield NaN to avoid a divide-by-zero.

Source code in src/bbstrader/core/indicators.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
def zscore(values: ArrayLike, window: int) -> NDArray[np.float64]:
    """Rolling z-score: ``(price - rolling_mean) / rolling_std``.

    Bars where the rolling standard deviation is zero yield ``NaN`` to avoid a
    divide-by-zero.
    """
    _check_window(window)
    arr = _as_float_array(values)
    mean = sma(arr, window)
    std = rolling_std(arr, window)
    with np.errstate(invalid="ignore", divide="ignore"):
        out = np.where(std > 0, (arr - mean) / std, np.nan)
    return out

roc 

roc(values: ArrayLike, window: int) -> NDArray[np.float64]

Rate of change in percent over window bars.

Source code in src/bbstrader/core/indicators.py 
138
139
140
141
142
143
144
145
146
147
148
149
150
def roc(values: ArrayLike, window: int) -> NDArray[np.float64]:
    """Rate of change in percent over ``window`` bars."""
    _check_window(window)
    arr = _as_float_array(values)
    out = np.full(arr.shape, np.nan)
    if arr.size <= window:
        return out
    prior = arr[:-window]
    with np.errstate(invalid="ignore", divide="ignore"):
        out[window:] = np.where(
            prior != 0, (arr[window:] - prior) / prior * 100.0, np.nan
        )
    return out

rsi 

rsi(values: ArrayLike, window: int = 14) -> NDArray[np.float64]

Wilder's Relative Strength Index over window bars (NaN-padded).

Source code in src/bbstrader/core/indicators.py 
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
def rsi(values: ArrayLike, window: int = 14) -> NDArray[np.float64]:
    """Wilder's Relative Strength Index over ``window`` bars (NaN-padded)."""
    _check_window(window)
    arr = _as_float_array(values)
    out = np.full(arr.shape, np.nan)
    if arr.size <= window:
        return out
    delta = np.diff(arr)
    gains = np.where(delta > 0, delta, 0.0)
    losses = np.where(delta < 0, -delta, 0.0)
    # Wilder's smoothing: seed with the simple average of the first window.
    avg_gain = gains[:window].mean()
    avg_loss = losses[:window].mean()

    def _rsi_from(avg_gain: float, avg_loss: float) -> float:
        """Return the RSI value for a smoothed average gain and loss.

        Args:
            avg_gain (float): The smoothed average up-move over the window.
            avg_loss (float): The smoothed average down-move over the window.

        Returns:
            float: The RSI in [0, 100]; 100 when there are no losses.
        """
        if avg_loss == 0:
            return 100.0
        rs = avg_gain / avg_loss
        return 100.0 - (100.0 / (1.0 + rs))

    out[window] = _rsi_from(avg_gain, avg_loss)
    for i in range(window + 1, arr.size):
        avg_gain = (avg_gain * (window - 1) + gains[i - 1]) / window
        avg_loss = (avg_loss * (window - 1) + losses[i - 1]) / window
        out[i] = _rsi_from(avg_gain, avg_loss)
    return out

true_range 

true_range(high: ArrayLike, low: ArrayLike, close: ArrayLike) -> NDArray[np.float64]

True range: max(high-low, |high-prev_close|, |low-prev_close|).

Source code in src/bbstrader/core/indicators.py 
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
def true_range(
    high: ArrayLike, low: ArrayLike, close: ArrayLike
) -> NDArray[np.float64]:
    """True range: ``max(high-low, |high-prev_close|, |low-prev_close|)``."""
    h = _as_float_array(high)
    low_arr = _as_float_array(low)
    c = _as_float_array(close)
    if not (h.size == low_arr.size == c.size):
        raise ValueError("high, low and close must have the same length.")
    out = np.full(h.shape, np.nan)
    if h.size == 0:
        return out
    out[0] = h[0] - low_arr[0]
    prev_close = c[:-1]
    hl = h[1:] - low_arr[1:]
    hc = np.abs(h[1:] - prev_close)
    lc = np.abs(low_arr[1:] - prev_close)
    out[1:] = np.maximum.reduce([hl, hc, lc])
    return out

atr 

atr(high: ArrayLike, low: ArrayLike, close: ArrayLike, window: int = 14) -> NDArray[np.float64]

Average True Range using Wilder's smoothing (NaN-padded).

Source code in src/bbstrader/core/indicators.py 
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
def atr(
    high: ArrayLike, low: ArrayLike, close: ArrayLike, window: int = 14
) -> NDArray[np.float64]:
    """Average True Range using Wilder's smoothing (NaN-padded)."""
    _check_window(window)
    tr = true_range(high, low, close)
    out = np.full(tr.shape, np.nan)
    if tr.size < window:
        return out
    prev = float(np.nanmean(tr[:window]))
    out[window - 1] = prev
    for i in range(window, tr.size):
        prev = (prev * (window - 1) + tr[i]) / window
        out[i] = prev
    return out

bollinger_bands 

bollinger_bands(values: ArrayLike, window: int = 20, num_std: float = 2.0) -> Tuple[NDArray[np.float64], NDArray[np.float64], NDArray[np.float64]]

Bollinger Bands; returns (lower, middle, upper) arrays.

Source code in src/bbstrader/core/indicators.py 
228
229
230
231
232
233
234
235
236
237
238
def bollinger_bands(
    values: ArrayLike, window: int = 20, num_std: float = 2.0
) -> Tuple[NDArray[np.float64], NDArray[np.float64], NDArray[np.float64]]:
    """Bollinger Bands; returns ``(lower, middle, upper)`` arrays."""
    _check_window(window)
    arr = _as_float_array(values)
    middle = sma(arr, window)
    std = rolling_std(arr, window)
    upper = middle + num_std * std
    lower = middle - num_std * std
    return lower, middle, upper

macd 

macd(values: ArrayLike, fast: int = 12, slow: int = 26, signal: int = 9) -> Tuple[NDArray[np.float64], NDArray[np.float64], NDArray[np.float64]]

MACD; returns (macd_line, signal_line, histogram) arrays.

Source code in src/bbstrader/core/indicators.py 
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
def macd(
    values: ArrayLike,
    fast: int = 12,
    slow: int = 26,
    signal: int = 9,
) -> Tuple[NDArray[np.float64], NDArray[np.float64], NDArray[np.float64]]:
    """MACD; returns ``(macd_line, signal_line, histogram)`` arrays."""
    if fast >= slow:
        raise ValueError(f"fast ({fast}) must be smaller than slow ({slow}).")
    arr = _as_float_array(values)
    macd_line = ema(arr, fast) - ema(arr, slow)
    # The signal line is an EMA of the MACD line over its valid (non-NaN) tail.
    signal_line = np.full(arr.shape, np.nan)
    valid = ~np.isnan(macd_line)
    if valid.any():
        start = int(np.argmax(valid))
        tail = ema(macd_line[start:], signal)
        signal_line[start:] = tail
    histogram = macd_line - signal_line
    return macd_line, signal_line, histogram

stochastic 

stochastic(high: ArrayLike, low: ArrayLike, close: ArrayLike, k_window: int = 14, d_window: int = 3) -> Tuple[NDArray[np.float64], NDArray[np.float64]]

Stochastic oscillator; returns (%K, %D) arrays.

Source code in src/bbstrader/core/indicators.py 
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
def stochastic(
    high: ArrayLike,
    low: ArrayLike,
    close: ArrayLike,
    k_window: int = 14,
    d_window: int = 3,
) -> Tuple[NDArray[np.float64], NDArray[np.float64]]:
    """Stochastic oscillator; returns ``(%K, %D)`` arrays."""
    _check_window(k_window, "k_window")
    _check_window(d_window, "d_window")
    h = _as_float_array(high)
    low_arr = _as_float_array(low)
    c = _as_float_array(close)
    if not (h.size == low_arr.size == c.size):
        raise ValueError("high, low and close must have the same length.")
    percent_k = np.full(c.shape, np.nan)
    for i in range(k_window - 1, c.size):
        window_high = h[i - k_window + 1 : i + 1].max()
        window_low = low_arr[i - k_window + 1 : i + 1].min()
        span = window_high - window_low
        if span > 0:
            percent_k[i] = (c[i] - window_low) / span * 100.0
    percent_d = sma(percent_k, d_window)
    return percent_k, percent_d

donchian 

donchian(high: ArrayLike, low: ArrayLike, window: int = 20) -> Tuple[NDArray[np.float64], NDArray[np.float64]]

Donchian channel; returns (lower, upper) arrays over window bars.

The channel at bar i uses bars [i-window+1, i] (inclusive), so it is safe to compare the previous bar's channel against the current price for a breakout without look-ahead.

Source code in src/bbstrader/core/indicators.py 
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
def donchian(
    high: ArrayLike, low: ArrayLike, window: int = 20
) -> Tuple[NDArray[np.float64], NDArray[np.float64]]:
    """Donchian channel; returns ``(lower, upper)`` arrays over ``window`` bars.

    The channel at bar ``i`` uses bars ``[i-window+1, i]`` (inclusive), so it is
    safe to compare the *previous* bar's channel against the current price for a
    breakout without look-ahead.
    """
    _check_window(window)
    h = _as_float_array(high)
    low_arr = _as_float_array(low)
    if h.size != low_arr.size:
        raise ValueError("high and low must have the same length.")
    upper = np.full(h.shape, np.nan)
    lower = np.full(h.shape, np.nan)
    for i in range(window - 1, h.size):
        upper[i] = h[i - window + 1 : i + 1].max()
        lower[i] = low_arr[i - window + 1 : i + 1].min()
    return lower, upper

strategy

TradeAction

Bases: Enum

An enumeration class for trade actions.

TradeSignal dataclass 

TradeSignal(id: int, symbol: str, action: TradeAction, price: float = None, stoplimit: float = None, sl: float = None, tp: float = None, comment: str = None)

Represents a trading signal generated by a trading system or strategy.

Notes

Attributes:

  • id (int): A unique identifier for the trade signal or the strategy.
  • symbol (str): The trading symbol (e.g., stock ticker, forex pair, crypto asset).
  • action (TradeAction): The trading action to perform. Must be an instance of the TradeAction enum (e.g., BUY, SELL).
  • price (float, optional): The price at which the trade should be executed.
  • stoplimit (float, optional): A stop-limit price for the trade. Must not be set without specifying a price.
  • sl (float, optional): A stop loss price for the trade.
  • tp (float, optional): A take profit price for the trade.
  • comment (str, optional): An optional comment or description related to the trade signal.

TradingMode

Bases: Enum

isbacktest 
isbacktest() -> bool

Return True if this mode is :attr:TradingMode.BACKTEST.

Source code in src/bbstrader/core/strategy.py 
162
163
164
def isbacktest(self) -> bool:
    """Return True if this mode is :attr:`TradingMode.BACKTEST`."""
    return self == TradingMode.BACKTEST
islive 
islive() -> bool

Return True if this mode is :attr:TradingMode.LIVE.

Source code in src/bbstrader/core/strategy.py 
166
167
168
def islive(self) -> bool:
    """Return True if this mode is :attr:`TradingMode.LIVE`."""
    return self == TradingMode.LIVE

Strategy

A Strategy() object encapsulates all calculation on market data that generate advisory signals to a Portfolio object. Thus all of the "strategy logic" resides within this class. We opted to separate out the Strategy and Portfolio objects for this backtester, since we believe this is more amenable to the situation of multiple strategies feeding "ideas" to a larger Portfolio, which then can handle its own risk (such as sector allocation, leverage). In higher frequency trading, the strategy and portfolio concepts will be tightly coupled and extremely hardware dependent.

At this stage in the event-driven backtester development there is no concept of an indicator or filter, such as those found in technical trading. These are also good candidates for creating a class hierarchy.

The strategy hierarchy is relatively simple as it consists of an abstract base class with a single pure virtual method for generating SignalEvent objects. Other methods are provided to check for pending orders, update trades from fills, and get updates from the portfolio.

calculate_signals abstractmethod 
calculate_signals(*args: Any, **kwargs: Any) -> List[TradeSignal] | None

Generate advisory trade signals from market data.

The single abstract method every strategy must implement; backtest and live engines both call it.

Parameters:

Name Type Description Default
args Any

Engine-supplied positional context (for example a market event).

 ()
kwargs Any

Engine-supplied keyword context.

 {}

Returns:

Type Description
List[TradeSignal] | None

List[TradeSignal] | None: The signals to act on, or None.
Source code in src/bbstrader/core/strategy.py 
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
@abstractmethod
def calculate_signals(self, *args: Any, **kwargs: Any) -> List[TradeSignal] | None:
    """Generate advisory trade signals from market data.

    The single abstract method every strategy must implement; backtest and
    live engines both call it.

    Args:
        args: Engine-supplied positional context (for example a market event).
        kwargs: Engine-supplied keyword context.

    Returns:
        List[TradeSignal] | None: The signals to act on, or None.
    """
    raise NotImplementedError("Should implement calculate_signals()")
check_pending_orders 
check_pending_orders(*args: Any, **kwargs: Any) -> None

Evaluate any pending orders for the current bar (optional hook).

Source code in src/bbstrader/core/strategy.py 
209
210
211
def check_pending_orders(self, *args: Any, **kwargs: Any) -> None:
    """Evaluate any pending orders for the current bar (optional hook)."""
    ...
get_update_from_portfolio 
get_update_from_portfolio(*args: Any, **kwargs: Any) -> None

Receive the latest positions/holdings from the portfolio (optional hook).

Source code in src/bbstrader/core/strategy.py 
213
214
215
def get_update_from_portfolio(self, *args: Any, **kwargs: Any) -> None:
    """Receive the latest positions/holdings from the portfolio (optional hook)."""
    ...
update_trades_from_fill 
update_trades_from_fill(*args: Any, **kwargs: Any) -> None

Update trade bookkeeping from a fill event (optional hook).

Source code in src/bbstrader/core/strategy.py 
217
218
219
def update_trades_from_fill(self, *args: Any, **kwargs: Any) -> None:
    """Update trade bookkeeping from a fill event (optional hook)."""
    ...
perform_period_end_checks 
perform_period_end_checks(*args: Any, **kwargs: Any) -> None

Run end-of-period maintenance such as risk checks (optional hook).

Source code in src/bbstrader/core/strategy.py 
221
222
223
def perform_period_end_checks(self, *args: Any, **kwargs: Any) -> None:
    """Run end-of-period maintenance such as risk checks (optional hook)."""
    ...

BaseStrategy 

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

Bases: Strategy

Base class containing shared logic for both Backtest and Live MT5 strategies. This class handles configuration, logging, and common utility calculations.

Initialise shared strategy configuration.

Parameters:

Name Type Description Default
symbol_list List[str]

The symbols the strategy trades.

 required
kwargs Any

Common options, including risk_weights (risk budget), max_trades (per-symbol trade cap, default 1), time_frame (default "D1") and logger.

 {}
Source code in src/bbstrader/core/strategy.py 
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
def __init__(
    self,
    symbol_list: List[str],
    **kwargs: Any,
) -> None:
    """Initialise shared strategy configuration.

    Args:
        symbol_list (List[str]): The symbols the strategy trades.
        kwargs (Any): Common options, including ``risk_weights`` (risk
            budget), ``max_trades`` (per-symbol trade cap, default 1),
            ``time_frame`` (default ``"D1"``) and ``logger``.
    """
    self.symbols = symbol_list
    self.risk_budget = self._check_risk_budget(**kwargs)
    self.max_trades = kwargs.get("max_trades", {s: 1 for s in self.symbols})
    self.tf = kwargs.get("time_frame", "D1")
    self.logger = kwargs.get("logger") or logger
    self.kwargs = kwargs
    self.periodes = 0
cash abstractmethod property 
cash: float

Returns the available cash (virtual or real).

calculate_signals 
calculate_signals(*args: Any, **kwargs: Any) -> List[TradeSignal] | None

Provides the mechanisms to calculate signals for the strategy. This methods should return a list of signals for the strategy For Live mode and None For Backtest mode.

Each signal must be a TradeSignal object with the following attributes: - id: The unique identifier for the strategy or order. - action: The order to execute on the symbol (LONG, SHORT, EXIT, etc.), see bbstrader.core.utils.TradeAction. - symbol: The trading symbol (e.g., stock ticker, forex pair, crypto asset). - See bbstrader.core.strategy.TradeSignal for other optionnal arguments.

Source code in src/bbstrader/core/strategy.py 
301
302
303
304
305
306
307
308
309
310
311
312
def calculate_signals(self, *args: Any, **kwargs: Any) -> List[TradeSignal] | None:
    """
    Provides the mechanisms to calculate signals for the strategy.
    This methods should return a list of signals for the strategy For Live mode and None For Backtest mode.

    Each signal must be a ``TradeSignal`` object with the following attributes:
    - ``id``: The unique identifier for the strategy or order.
    - ``action``: The order to execute on the symbol (LONG, SHORT, EXIT, etc.), see `bbstrader.core.utils.TradeAction`.
    - ``symbol``: The trading symbol (e.g., stock ticker, forex pair, crypto asset).
    - See ``bbstrader.core.strategy.TradeSignal`` for other optionnal arguments.
    """
    raise NotImplementedError("Should implement calculate_signals()")
perform_period_end_checks 
perform_period_end_checks(*args: Any, **kwargs: Any) -> None

Some strategies may require additional checks at the end of the period, such as closing all positions or orders or tracking the performance of the strategy etc.

This method is called at the end of the period to perform such checks.

Source code in src/bbstrader/core/strategy.py 
314
315
316
317
318
319
320
321
def perform_period_end_checks(self, *args: Any, **kwargs: Any) -> None:
    """
    Some strategies may require additional checks at the end of the period,
    such as closing all positions or orders or tracking the performance of the strategy etc.

    This method is called at the end of the period to perform such checks.
    """
    pass
get_asset_values abstractmethod 
get_asset_values(symbol_list: List[str], window: int, value_type: str = 'returns', array: bool = True, **kwargs) -> Optional[Dict[str, Union[np.typing.NDArray, pd.Series]]]

Get the historical OHLCV value or returns or custum value based on the DataHandker of the assets in the symbol list.

Parameters:

Name Type Description Default
symbol_list

List of ticker symbols for the pairs trading strategy.

 required
window

The lookback period for resquesting the data.

 required
value_type

The type of value to get (e.g., returns, open, high, low, close, adjclose, volume).

 required
array

If True, return the values as numpy arrays, otherwise as pandas Series.

 required
error

The error handling method for the function.

 required

Returns:

Name Type Description
asset_values Optional[Dict[str, Union[NDArray, Series]]]

Historical values of the assets in the symbol list.
Note

In Live mode, the bbstrader.metatrader.rates.Rates class is used to get the historical data so the value_type must be 'returns', 'open', 'high', 'low', 'close', 'adjclose', 'volume'.

Source code in src/bbstrader/core/strategy.py 
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
@abstractmethod
def get_asset_values(
    self,
    symbol_list: List[str],
    window: int,
    value_type: str = "returns",
    array: bool = True,
    **kwargs,
) -> Optional[Dict[str, Union[np.typing.NDArray, pd.Series]]]:
    """
    Get the historical OHLCV value or returns or custum value
    based on the DataHandker of the assets in the symbol list.

    Args:
        symbol_list : List of ticker symbols for the pairs trading strategy.
        window : The lookback period for resquesting the data.
        value_type : The type of value to get (e.g., returns, open, high, low, close, adjclose, volume).
        array : If True, return the values as numpy arrays, otherwise as pandas Series.
        error : The error handling method for the function.

    Returns:
        asset_values : Historical values of the assets in the symbol list.

    Note:
        In Live mode, the `bbstrader.metatrader.rates.Rates` class is used to get the historical data
        so the value_type must be 'returns', 'open', 'high', 'low', 'close', 'adjclose', 'volume'.
    """
    raise NotImplementedError
apply_risk_management 
apply_risk_management(optimizer: str, symbols: Optional[List[str]] = None, freq: int = 252) -> Optional[Dict[str, float]]

Apply risk management optimization.

Source code in src/bbstrader/core/strategy.py 
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
def apply_risk_management(
    self,
    optimizer: str,
    symbols: Optional[List[str]] = None,
    freq: int = 252,
) -> Optional[Dict[str, float]]:
    """Apply risk management optimization."""
    if optimizer is None:
        return None
    symbols = symbols or self.symbols

    prices = self.get_asset_values(
        symbol_list=symbols,
        window=freq,
        value_type="close",
        array=False,
    )

    if prices is None:
        return None
    prices = pd.DataFrame(prices)
    prices = prices.dropna(axis=0, how="any")
    try:
        weights = optimized_weights(prices=prices, freq=freq, method=optimizer)
        return {symbol: abs(weight) for symbol, weight in weights.items()}
    except Exception:
        return {symbol: 0.0 for symbol in symbols}
get_quantity 
get_quantity(symbol: str, weight: float, price: Optional[float] = None, volume: Optional[float] = None, maxqty: Optional[int] = None) -> int

Calculate the quantity to buy or sell for a given symbol based on the dollar value provided. The quantity calculated can be used to evalute a strategy's performance for each symbol given the fact that the dollar value is the same for all symbols.

Parameters:

Name Type Description Default
symbol

The symbol for the trade.

 required

Returns:

Name Type Description
qty int

The quantity to buy or sell for the symbol.
Source code in src/bbstrader/core/strategy.py 
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
def get_quantity(
    self,
    symbol: str,
    weight: float,
    price: Optional[float] = None,
    volume: Optional[float] = None,
    maxqty: Optional[int] = None,
) -> int:
    """
    Calculate the quantity to buy or sell for a given symbol based on the dollar value provided.
    The quantity calculated can be used to evalute a strategy's performance for each symbol
    given the fact that the dollar value is the same for all symbols.

    Args:
        symbol : The symbol for the trade.

    Returns:
        qty : The quantity to buy or sell for the symbol.
    """
    current_cash = self.cash

    if (
        current_cash is None
        or weight == 0
        or current_cash == 0
        or np.isnan(current_cash)
    ):
        return 0
    if price is None:
        vals = self.get_asset_values(
            [symbol], window=1, value_type="close", array=True
        )
        if vals and symbol in vals and len(vals[symbol]) > 0:
            price = float(vals[symbol][-1])
        else:
            price = None

    if volume is None:
        volume = round(current_cash * weight)

    if (
        price is None
        or not isinstance(price, (int, float, np.number))
        or volume is None
        or not isinstance(volume, (int, float, np.number))
        or np.isnan(float(price))
        or np.isnan(float(volume))
    ):
        if weight != 0:
            return 1
        return 0

    qty = round(volume / price, 2)
    qty = max(qty, 0) / self.max_trades.get(symbol, 1)
    if maxqty is not None:
        qty = min(qty, maxqty)
    return int(max(round(qty, 2), 0))
get_quantities 
get_quantities(quantities: Optional[Union[Dict[str, int], int]]) -> Dict[str, Optional[int]]

Get the quantities to buy or sell for the symbols in the strategy. This method is used when whe need to assign different quantities to the symbols.

Parameters:

Name Type Description Default
quantities

The quantities for the symbols in the strategy.

 required
Source code in src/bbstrader/core/strategy.py 
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
def get_quantities(
    self, quantities: Optional[Union[Dict[str, int], int]]
) -> Dict[str, Optional[int]]:
    """
    Get the quantities to buy or sell for the symbols in the strategy.
    This method is used when whe need to assign different quantities to the symbols.

    Args:
        quantities : The quantities for the symbols in the strategy.
    """
    if quantities is None:
        return {symbol: None for symbol in self.symbols}
    if isinstance(quantities, dict):
        return quantities
    elif isinstance(quantities, int):
        return {symbol: quantities for symbol in self.symbols}
    raise TypeError(f"Unsupported type for quantities: {type(quantities)}")
calculate_pct_change staticmethod 
calculate_pct_change(current_price: float, lh_price: float) -> float

Return the percentage change from lh_price to current_price.

Parameters:

Name Type Description Default
current_price float

The current price.

 required
lh_price float

The reference (look-back/historical) price.

 required

Returns:

Name Type Description
float float

The change as a percentage (for example 5.0 for +5%).
Source code in src/bbstrader/core/strategy.py 
456
457
458
459
460
461
462
463
464
465
466
467
@staticmethod
def calculate_pct_change(current_price: float, lh_price: float) -> float:
    """Return the percentage change from ``lh_price`` to ``current_price``.

    Args:
        current_price (float): The current price.
        lh_price (float): The reference (look-back/historical) price.

    Returns:
        float: The change as a percentage (for example 5.0 for +5%).
    """
    return ((current_price - lh_price) / lh_price) * 100
is_signal_time staticmethod 
is_signal_time(period_count: Optional[int], signal_inverval: int) -> bool

Check if we can generate a signal based on the current period count. We use the signal interval as a form of periodicity or rebalancing period.

Parameters:

Name Type Description Default
period_count

The current period count (e.g., number of bars).

 required
signal_inverval

The signal interval for generating signals (e.g., every 5 bars).

 required

Returns:

Name Type Description
bool bool

True if we can generate a signal, False otherwise
Source code in src/bbstrader/core/strategy.py 
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
@staticmethod
def is_signal_time(period_count: Optional[int], signal_inverval: int) -> bool:
    """
    Check if we can generate a signal based on the current period count.
    We use the signal interval as a form of periodicity or rebalancing period.

    Args:
        period_count : The current period count (e.g., number of bars).
        signal_inverval : The signal interval for generating signals (e.g., every 5 bars).

    Returns:
        bool : True if we can generate a signal, False otherwise
    """
    if period_count == 0 or period_count is None:
        return True
    return period_count % signal_inverval == 0
get_current_dt staticmethod 
get_current_dt(time_zone: str = 'US/Eastern') -> datetime

Return the current time in the given timezone.

Parameters:

Name Type Description Default
time_zone str

The IANA timezone name (default "US/Eastern").

 'US/Eastern'

Returns:

Name Type Description
datetime datetime

The timezone-aware current datetime.
Source code in src/bbstrader/core/strategy.py 
486
487
488
489
490
491
492
493
494
495
496
@staticmethod
def get_current_dt(time_zone: str = "US/Eastern") -> datetime:
    """Return the current time in the given timezone.

    Args:
        time_zone (str): The IANA timezone name (default ``"US/Eastern"``).

    Returns:
        datetime: The timezone-aware current datetime.
    """
    return datetime.now(pytz.timezone(time_zone))
convert_time_zone staticmethod 
convert_time_zone(dt: Union[datetime, int, Timestamp], from_tz: str = 'UTC', to_tz: str = 'US/Eastern') -> pd.Timestamp

Convert datetime from one timezone to another.

Parameters:

Name Type Description Default
dt

The datetime to convert.

 required
from_tz

The timezone to convert from.

 required
to_tz

The timezone to convert to.

 required

Returns:

Name Type Description
dt_to Timestamp

The converted datetime.
Source code in src/bbstrader/core/strategy.py 
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
@staticmethod
def convert_time_zone(
    dt: Union[datetime, int, pd.Timestamp],
    from_tz: str = "UTC",
    to_tz: str = "US/Eastern",
) -> pd.Timestamp:
    """
    Convert datetime from one timezone to another.

    Args:
        dt : The datetime to convert.
        from_tz : The timezone to convert from.
        to_tz : The timezone to convert to.

    Returns:
        dt_to : The converted datetime.
    """
    from_tz_pytz = pytz.timezone(from_tz)
    if isinstance(dt, (datetime, int)):
        dt_ts = pd.to_datetime(dt, unit="s")
    else:
        dt_ts = dt
    if dt_ts.tzinfo is None:
        dt_ts = dt_ts.tz_localize(from_tz_pytz)
    else:
        dt_ts = dt_ts.tz_convert(from_tz_pytz)
    return dt_ts.tz_convert(pytz.timezone(to_tz))
stop_time staticmethod 
stop_time(time_zone: str, stop_time: str) -> bool

Return True once the current time has reached stop_time.

Parameters:

Name Type Description Default
time_zone str

The IANA timezone the times are evaluated in.

 required
stop_time str

The cut-off time as "HH:MM".

 required

Returns:

Name Type Description
bool bool

True if the current time is at or past stop_time.
Source code in src/bbstrader/core/strategy.py 
526
527
528
529
530
531
532
533
534
535
536
537
538
539
@staticmethod
def stop_time(time_zone: str, stop_time: str) -> bool:
    """Return True once the current time has reached ``stop_time``.

    Args:
        time_zone (str): The IANA timezone the times are evaluated in.
        stop_time (str): The cut-off time as ``"HH:MM"``.

    Returns:
        bool: True if the current time is at or past ``stop_time``.
    """
    now = datetime.now(pytz.timezone(time_zone)).time()
    stop_time_dt = datetime.strptime(stop_time, "%H:%M").time()
    return now >= stop_time_dt

TWSStrategy

Bases: Strategy

Placeholder strategy base for the Interactive Brokers (TWS) adapter.

calculate_signals 
calculate_signals(*args: Any, **kwargs: Any) -> List[TradeSignal]

Generate trade signals for a TWS strategy (must be implemented).

Parameters:

Name Type Description Default
args Any

Engine-supplied positional context.

 ()
kwargs Any

Engine-supplied keyword context.

 {}

Returns:

Type Description
List[TradeSignal]

List[TradeSignal]: The signals to act on.

Raises:

Type Description
NotImplementedError

Always, until a concrete subclass implements it.
Source code in src/bbstrader/core/strategy.py 
545
546
547
548
549
550
551
552
553
554
555
556
557
558
def calculate_signals(self, *args: Any, **kwargs: Any) -> List[TradeSignal]:
    """Generate trade signals for a TWS strategy (must be implemented).

    Args:
        args: Engine-supplied positional context.
        kwargs: Engine-supplied keyword context.

    Returns:
        List[TradeSignal]: The signals to act on.

    Raises:
        NotImplementedError: Always, until a concrete subclass implements it.
    """
    raise NotImplementedError("Should implement calculate_signals()")

generate_signal 

generate_signal(id: int, symbol: str, action: TradeAction, price: float = None, stoplimit: float = None, sl: float = None, tp: float = None, comment: str = None) -> TradeSignal

Generates a trade signal for MetaTrader 5.

Parameters:

Name Type Description Default
id int

Unique identifier for the trade signal.

 required
symbol str

The symbol for which the trade signal is generated.

 required
action TradeAction

The action to be taken (e.g., BUY, SELL).

 required
price float

The price at which to execute the trade.

 None
stoplimit float

The stop limit price for the trade.

 None
sl float

The stop loss price for the trade.

 None
tp float

The take profit price for the trade.

 None
comment str

Additional comments for the trade.

 None

Returns:

Name Type Description
TradeSignal TradeSignal

A TradeSignal object containing the details of the trade signal.
Source code in src/bbstrader/core/strategy.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
def generate_signal(
    id: int,
    symbol: str,
    action: TradeAction,
    price: float = None,
    stoplimit: float = None,
    sl: float = None,
    tp: float = None,
    comment: str = None,
) -> TradeSignal:
    """
    Generates a trade signal for MetaTrader 5.

    Args:
        id (int): Unique identifier for the trade signal.
        symbol (str): The symbol for which the trade signal is generated.
        action (TradeAction): The action to be taken (e.g., BUY, SELL).
        price (float, optional): The price at which to execute the trade.
        stoplimit (float, optional): The stop limit price for the trade.
        sl (float, optional): The stop loss price for the trade.
        tp (float, optional): The take profit price for the trade.
        comment (str, optional): Additional comments for the trade.

    Returns:
        TradeSignal: A TradeSignal object containing the details of the trade signal.
    """
    return TradeSignal(
        id=id,
        symbol=symbol,
        action=action,
        price=price,
        stoplimit=stoplimit,
        sl=sl,
        tp=tp,
        comment=comment,
    )