Bot API Reference

tradingbot.utils.botclass.Bot(name: str, symbol: str | None = None, tickers: list[str] | None = None, interval: str = '1m', period: str = '1d', benchmark_tickers: list[str] | None = None, **kwargs)

Base class for trading bots.

Provides common functionality for data fetching, trading operations, portfolio management, and database interactions.

Data Caching: - Each Bot instance has its own DataService instance with per-instance caching. - For cross-run data reuse (e.g., hyperparameter tuning), data is persisted to the database when saveToDB=True. Subsequent fetches check the database first and only call yfinance if data is missing or stale. - Best practice: Use saveToDB=True for historical backtests to enable efficient data reuse across multiple runs or parameter combinations.

Subclasses should implement either: - decisionFunction(row) -> int: Returns -1 (sell), 0 (hold), or 1 (buy) - makeOneIteration() -> int: Custom iteration logic

Initialize a trading bot.

Parameters:

Name	Type	Description	Default
name	str	Unique name for the bot (used for database identification)	required
symbol	str | None	Trading symbol (e.g., "EURUSD=X", "^XAU", "QQQ") Optional for multi-asset bots that override makeOneIteration()	None
tickers	list[str] | None	List of trading symbols for multi-ticker strategies. If provided, symbol is ignored and set to None. e.g. tickers=["SPY", "QQQ", "GLD"]	None
interval	str	Data interval (e.g., "1m", "5m", "1h", "1d") - default: "1m"	'1m'
period	str	Data period (e.g., "1d", "5d", "1mo", "1y") - default: "1d"	'1d'
benchmark_tickers	list[str] | None	Subset of tickers that is loaded for data but NEVER traded, and excluded from the equal-weight divisor. For strategies that need a relative-strength baseline (e.g. GoldenButterflyMomBot carries SPY for its RRG computation). Without this, such a bot's capital is divided by a count that includes an asset it can never hold, permanently stranding that share in cash.	None
**kwargs		Arbitrary hyperparameters that will be stored in self.params and can be accessed by subclasses for flexible parameterization	{}

Source code in tradingbot/utils/botclass.py

def __init__(
    self,
    name: str,
    symbol: str | None = None,
    tickers: list[str] | None = None,
    interval: str = "1m",
    period: str = "1d",
    benchmark_tickers: list[str] | None = None,
    **kwargs,
):
    """
    Initialize a trading bot.

    Args:
        name: Unique name for the bot (used for database identification)
        symbol: Trading symbol (e.g., "EURUSD=X", "^XAU", "QQQ")
                Optional for multi-asset bots that override makeOneIteration()
        tickers: List of trading symbols for multi-ticker strategies.
                 If provided, symbol is ignored and set to None.
                 e.g. tickers=["SPY", "QQQ", "GLD"]
        interval: Data interval (e.g., "1m", "5m", "1h", "1d") - default: "1m"
        period: Data period (e.g., "1d", "5d", "1mo", "1y") - default: "1d"
        benchmark_tickers: Subset of `tickers` that is loaded for data but
                 NEVER traded, and excluded from the equal-weight divisor.
                 For strategies that need a relative-strength baseline (e.g.
                 GoldenButterflyMomBot carries SPY for its RRG computation).
                 Without this, such a bot's capital is divided by a count
                 that includes an asset it can never hold, permanently
                 stranding that share in cash.
        **kwargs: Arbitrary hyperparameters that will be stored in self.params
                 and can be accessed by subclasses for flexible parameterization
    """
    setup_logging()
    self.bot_name = name  # Store name separately to avoid DetachedInstanceError

    # Resolve the universe BEFORE touching the database: a misconfigured bot
    # must fail without first materialising a $10k row in `bots`.
    if tickers is not None:
        # Guard: accept a bare string as a single-element list
        if isinstance(tickers, str):
            tickers = [tickers]
        # Dedupe: a repeated ticker would both inflate the divisor and get
        # traded twice.
        self.tickers: list[str] = list(dict.fromkeys(tickers))
        self.symbol: str | None = None
    elif symbol is not None:
        self.tickers = [symbol]
        self.symbol = symbol
    else:
        self.tickers = []
        self.symbol = None

    if isinstance(benchmark_tickers, str):
        benchmark_tickers = [benchmark_tickers]
    self.benchmark_tickers: list[str] = list(dict.fromkeys(benchmark_tickers or []))

    unknown = [t for t in self.benchmark_tickers if t not in self.tickers]
    if unknown:
        raise ValueError(
            f"{name}: benchmark_tickers {unknown} are not in tickers {self.tickers}. "
            "A benchmark must also be listed in tickers, or its data is never fetched."
        )
    if self.tickers and not self.tradeable_tickers:
        raise ValueError(f"{name}: every ticker is a benchmark — nothing left to trade.")
    if self.USE_OPTIONS and (len(self.tickers) > 1 or type(self).targetWeights is not Bot.targetWeights):
        # The multi-ticker and targetWeights paths size through weight-based
        # rebalancePortfolio, which cannot target contracts. Refuse rather
        # than silently trade the stock the bot asked not to hold.
        raise ValueError(f"{name}: USE_OPTIONS supports single-symbol bots and direct buy()/sell() only.")

    init_db()  # Ensure database is initialized before first access
    self._bot_repository = BotRepository  # staticmethod namespace, never instantiated
    # Registers the bot's own row on first run. Nothing ORM is kept: dbBot
    # reads a fresh snapshot on every access.
    self._bot_repository.create_or_get_bot(name, initial_usd=self.INITIAL_CAPITAL)
    self.interval = interval
    self.period = period

    # Store hyperparameters in a dictionary for flexible access
    self.params = kwargs.copy() if kwargs else {}

    # Initialize services
    self._data_service = DataService()
    self._portfolio_manager = PortfolioManager(
        bot_name=self.bot_name,
        data_service=self._data_service,
        bot_repository=self._bot_repository,
    )

    # Maintain backward compatibility for data caching
    self.data: pd.DataFrame | None = None
    self.datas: dict[str, pd.DataFrame | None] = {}  # per-ticker cache for multi-ticker bots
    self.datasettings: tuple[str | None, str | None] = (None, None)

backtest_type: str property

Classify the bot's backtesting mode.

Note this counts self.tickers, NOT tradeable_tickers: a bot with one tradeable ticker plus a benchmark must still take the multi-ticker path, because the single-asset path fetches only self.symbol (None for a tickers= bot) and would never load the benchmark's data.

Precedence is targetWeights > decisionFunction > makeOneIteration, and it is deliberate: targetWeights is strictly more expressive than a -1/0/1 signal, so a bot defining both is one whose weights function calls its own decisionFunction as a helper. Ticker count does not enter into it — a single-ticker targetWeights bot is legal, because sizing a lone position between 0 and 1 is a strategy in its own right.

Returns:

Type	Description
str	"target_weights" — targetWeights overridden (any ticker count) → backtestable
str	"single_asset" — decisionFunction overridden + single ticker → backtestable
str	"multi_asset" — decisionFunction overridden + multiple tickers → backtestable
str	"event_driven" — only makeOneIteration overridden → NOT backtestable
str	"unknown" — neither method overridden → NOT backtestable

can_backtest: bool property

True if this bot can be backtested with local_backtest() / local_optimize().

Only data-driven bots that implement decisionFunction() or targetWeights() are backtestable. Event-driven bots (makeOneIteration only) must use run() for live execution.

dbBot: BotSnapshot property

This bot's row, read fresh on every access, as plain data.

It used to be the ORM row itself, re-fetched after every trade by hand. Any row outlives its session and goes stale or detached, and a missed re-fetch surfaced later as DetachedInstanceError or as last run's cash. Reading on access makes both impossible. Bind it to a local when a method needs one consistent view.

tradeable_tickers: list[str] property

Tickers this bot may actually hold: self.tickers minus benchmark_tickers.

Benchmarks are fetched into self.datas (a strategy may need one as a relative-strength baseline) but are excluded from the equal-weight divisor and are never bought or sold. Defaults to self.tickers, so bots that declare no benchmarks are unaffected.

Written defensively because tests stub out Bot.init entirely and backtest_bot() accepts arbitrary instances. It is also called from init during validation, before the attributes are guaranteed set.

addPdDFToDb(df: pd.DataFrame, interval: str | None = None) -> None

Add DataFrame rows to database, skipping duplicates.

Only inserts rows newer than the latest stored for this (symbol, interval).

Parameters:

Name	Type	Description	Default
df	DataFrame	DataFrame with columns: symbol, timestamp, open, high, low, close, volume	required
interval	str | None	Bar size these rows were fetched at; defaults to this bot's own interval. Filing rows under the wrong bar size corrupts every later read (see HistoricData's docstring).	None

Source code in tradingbot/utils/botclass.py

def addPdDFToDb(self, df: pd.DataFrame, interval: str | None = None) -> None:
    """
    Add DataFrame rows to database, skipping duplicates.

    Only inserts rows newer than the latest stored for this (symbol, interval).

    Args:
        df: DataFrame with columns: symbol, timestamp, open, high, low, close, volume
        interval: Bar size these rows were fetched at; defaults to this bot's
                  own interval. Filing rows under the wrong bar size corrupts
                  every later read (see HistoricData's docstring).
    """
    self._data_service.add_pd_df_to_db(df, interval=interval or self.interval)

buy(symbol: str, quantity_usd: float = -1, option: bool | str | None = None, *, delta: float | None = None, dte: int | None = None) -> None

Buy a quantity of the specified symbol.

Parameters:

Name	Type	Description	Default
symbol	str	Trading symbol to buy	required
quantity_usd	float	Amount in USD to spend (-1 means use all available cash)	-1
option	bool | str | None	True / "call" / "put" buys a ~OPTION_TARGET_DTE-day at-the-money contract on symbol instead of the symbol itself. None uses the class's USE_OPTIONS (default False: plain stock).	None
delta	float | None	Option strike by delta (0.70 = a 70-delta call) instead of ATM. Defaults to OPTION_TARGET_DELTA.	None
dte	int | None	Minimum days to expiry. Defaults to OPTION_TARGET_DTE.	None

Source code in tradingbot/utils/botclass.py

def buy(
    self,
    symbol: str,
    quantity_usd: float = -1,
    option: bool | str | None = None,
    *,
    delta: float | None = None,
    dte: int | None = None,
) -> None:
    """
    Buy a quantity of the specified symbol.

    Args:
        symbol: Trading symbol to buy
        quantity_usd: Amount in USD to spend (-1 means use all available cash)
        option: True / "call" / "put" buys a ~OPTION_TARGET_DTE-day at-the-money
            contract on `symbol` instead of the symbol itself. None uses the
            class's USE_OPTIONS (default False: plain stock).
        delta: Option strike by delta (0.70 = a 70-delta call) instead of
            ATM. Defaults to OPTION_TARGET_DELTA.
        dte: Minimum days to expiry. Defaults to OPTION_TARGET_DTE.
    """
    # Use per-ticker cache if available
    cached = self.datas.get(symbol, self.data)
    self._portfolio_manager.buy(
        symbol,
        quantity_usd=quantity_usd,
        cached_data=cached,
        option=self._option_mode(option),
        target_dte=dte if dte is not None else getattr(self, "OPTION_TARGET_DTE", 30),
        target_delta=delta if delta is not None else getattr(self, "OPTION_TARGET_DELTA", None),
    )

close_options(underlying: str, include_stock: bool = False) -> float

Close every option leg on underlying (long and short); include_stock also flattens the shares held beside them. Returns net cash.

Source code in tradingbot/utils/botclass.py

def close_options(self, underlying: str, include_stock: bool = False) -> float:
    """
    Close every option leg on `underlying` (long and short); include_stock
    also flattens the shares held beside them. Returns net cash.
    """
    cash = self._portfolio_manager.close_options(underlying, include_stock=include_stock)
    return cash

convertToWideFormat(data_long: pd.DataFrame, value_column: str = 'close', fill_method: str = 'both') -> pd.DataFrame

Convert long-format DataFrame to wide format for portfolio optimization.

Parameters:

Name	Type	Description	Default
data_long	DataFrame	DataFrame in long format with columns: symbol, timestamp, open, high, low, close, volume	required
value_column	str	Column name to use as values (default: "close")	'close'
fill_method	str	How to handle missing values - "forward", "backward", "both", or None	'both'

Returns:

Type	Description
DataFrame	DataFrame with timestamp as index, symbols as columns, and specified value column as values

Source code in tradingbot/utils/botclass.py

def convertToWideFormat(
    self,
    data_long: pd.DataFrame,
    value_column: str = "close",
    fill_method: str = "both",
) -> pd.DataFrame:
    """
    Convert long-format DataFrame to wide format for portfolio optimization.

    Args:
        data_long: DataFrame in long format with columns: symbol, timestamp, open, high, low, close, volume
        value_column: Column name to use as values (default: "close")
        fill_method: How to handle missing values - "forward", "backward", "both", or None

    Returns:
        DataFrame with timestamp as index, symbols as columns, and specified value column as values
    """
    return self._data_service.convert_to_wide_format(
        data_long=data_long,
        value_column=value_column,
        fill_method=fill_method,
    )

decisionFunction(row: pd.Series) -> int

Decision function that determines trading action based on market data row.

Must be overridden by subclasses (unless using makeOneIteration() instead).

This is the preferred approach for most bots. The base class will: 1. Apply this function to each row in the DataFrame 2. Average the last N decisions (default: 1) 3. Execute trades based on the final decision

Parameters:

Name	Type	Description	Default
row	Series	Pandas Series containing: - Market data: symbol, timestamp, open, high, low, close, volume - Technical indicators: ~150+ indicators (e.g., momentum_rsi, trend_macd, etc.) - Access via: row["indicator_name"]	required

Returns:

Type	Description
int	-1: Sell signal (will sell holdings if any exist) 0: Hold (no action taken) 1: Buy signal (will buy if cash available)

Example

def decisionFunction(self, row): if row["momentum_rsi"] < 30: return 1 # Oversold, buy elif row["momentum_rsi"] > 70: return -1 # Overbought, sell return 0 # Hold

Source code in tradingbot/utils/botclass.py

def decisionFunction(self, row: pd.Series) -> int:
    """
    Decision function that determines trading action based on market data row.

    **Must be overridden by subclasses** (unless using makeOneIteration() instead).

    This is the preferred approach for most bots. The base class will:
    1. Apply this function to each row in the DataFrame
    2. Average the last N decisions (default: 1)
    3. Execute trades based on the final decision

    Args:
        row: Pandas Series containing:
            - Market data: symbol, timestamp, open, high, low, close, volume
            - Technical indicators: ~150+ indicators (e.g., momentum_rsi, trend_macd, etc.)
            - Access via: row["indicator_name"]

    Returns:
        -1: Sell signal (will sell holdings if any exist)
         0: Hold (no action taken)
         1: Buy signal (will buy if cash available)

    Example:
        def decisionFunction(self, row):
            if row["momentum_rsi"] < 30:
                return 1  # Oversold, buy
            elif row["momentum_rsi"] > 70:
                return -1  # Overbought, sell
            return 0  # Hold
    """
    raise NotImplementedError("You need to overwrite the decisionFunction!!!!")

delta_hedge(underlying: str, band_usd: float, ww: tuple[float, float] | None = None) -> float

Trade shares to bring the options-plus-shares delta back to ~0 beyond band_usd.

ww=(cost_frac, risk_aversion) switches to a Whalley-Wilmott band, rehedged to its edge.

Source code in tradingbot/utils/botclass.py

def delta_hedge(self, underlying: str, band_usd: float, ww: tuple[float, float] | None = None) -> float:
    """Trade shares to bring the options-plus-shares delta back to ~0 beyond `band_usd`.

    ww=(cost_frac, risk_aversion) switches to a Whalley-Wilmott band, rehedged to its edge.
    """
    shares = self._portfolio_manager.delta_hedge(underlying, band_usd, ww=ww)
    return shares

dividend_threatened_calls(book: options.OptionBook) -> list[str]

Short ITM calls in book likely to be assigned early before the next ex-dividend date (time value below the dividend, within option_rules.EX_DIVIDEND_CLOSE_BDAYS sessions). Each is logged.

Source code in tradingbot/utils/botclass.py

def dividend_threatened_calls(self, book: options.OptionBook) -> list[str]:
    """
    Short ITM calls in `book` likely to be assigned early before the next
    ex-dividend date (time value below the dividend, within
    option_rules.EX_DIVIDEND_CLOSE_BDAYS sessions). Each is logged.
    """
    return options.dividend_threatened_calls(book, options.next_dividend(book.underlying, book.today))

getDataFromDB(symbol: str, start_date: pd.Timestamp | None = None, end_date: pd.Timestamp | None = None, interval: str | None = None) -> pd.DataFrame

Load data from database for a symbol.

Parameters:

Name	Type	Description	Default
symbol	str	Trading symbol to query	required
start_date	Timestamp | None	Optional start date (timezone-aware UTC)	None
end_date	Timestamp | None	Optional end date (timezone-aware UTC)	None
interval	str | None	Bar size to read; defaults to this bot's own interval. Rows stored at other bar sizes are never returned.	None

Returns:

Type	Description
DataFrame	DataFrame with columns: symbol, timestamp, open, high, low, close, volume
DataFrame	Empty DataFrame if no data found

Source code in tradingbot/utils/botclass.py

def getDataFromDB(
    self,
    symbol: str,
    start_date: pd.Timestamp | None = None,
    end_date: pd.Timestamp | None = None,
    interval: str | None = None,
) -> pd.DataFrame:
    """
    Load data from database for a symbol.

    Args:
        symbol: Trading symbol to query
        start_date: Optional start date (timezone-aware UTC)
        end_date: Optional end date (timezone-aware UTC)
        interval: Bar size to read; defaults to this bot's own interval.
                  Rows stored at other bar sizes are never returned.

    Returns:
        DataFrame with columns: symbol, timestamp, open, high, low, close, volume
        Empty DataFrame if no data found
    """
    # Keyword args, not positional: get_data_from_db takes `interval` second,
    # so a positional call would silently pass start_date as the interval.
    return self._data_service.get_data_from_db(
        symbol=symbol,
        interval=interval or self.interval,
        start_date=start_date,
        end_date=end_date,
    )

getLatestDecision(data: pd.DataFrame, nrMedianLatest: int = 1) -> int

Get the latest trading decision by applying decisionFunction to data.

Parameters:

Name	Type	Description	Default
data	DataFrame	DataFrame with market data	required
nrMedianLatest	int	Number of latest rows to average (default: 1)	1

Returns:

Type	Description
int	Averaged decision signal (-1, 0, or 1)

Source code in tradingbot/utils/botclass.py

def getLatestDecision(self, data: pd.DataFrame, nrMedianLatest: int = 1) -> int:
    """
    Get the latest trading decision by applying decisionFunction to data.

    Args:
        data: DataFrame with market data
        nrMedianLatest: Number of latest rows to average (default: 1)

    Returns:
        Averaged decision signal (-1, 0, or 1)
    """
    if not isinstance(data, pd.DataFrame):
        raise ValueError("Data must be a pandas DataFrame")
    if len(data) == 0:
        return 0  # No data, hold

    # Work on a copy to avoid mutating the original DataFrame
    data_copy = data.copy()
    data_copy["signal"] = data_copy.apply(self.decisionFunction, axis=1)

    # Ensure we don't try to access more rows than available
    nrMedianLatest = min(nrMedianLatest, len(data_copy))
    if nrMedianLatest <= 0:
        return 0

    # Get the last nrMedianLatest signals and return their mean
    latest_signals = data_copy["signal"].iloc[-nrMedianLatest:]
    return int(latest_signals.mean())

getLatestPrice(symbol: str) -> float

Get the latest price for a symbol, using TTL cache and checking DB first.

Parameters:

Name	Type	Description	Default
symbol	str	Trading symbol to get price for	required

Returns:

Type	Description
float	Latest price as float

Raises:

Type	Description
ValueError	If no price data is available

Source code in tradingbot/utils/botclass.py

def getLatestPrice(self, symbol: str) -> float:
    """
    Get the latest price for a symbol, using TTL cache and checking DB first.

    Args:
        symbol: Trading symbol to get price for

    Returns:
        Latest price as float

    Raises:
        ValueError: If no price data is available
    """
    # Pass the per-ticker cache if available, else fallback to self.data
    cached = self.datas.get(symbol, self.data)
    return self._data_service.get_latest_price(symbol, cached_data=cached)

getLatestPricesBatch(symbols: list[str]) -> dict[str, float]

Get latest prices for multiple symbols in a single DB query.

Parameters:

Name	Type	Description	Default
symbols	list[str]	List of trading symbols to get prices for	required

Returns:

Type	Description
dict[str, float]	Dictionary mapping symbol to latest price

Source code in tradingbot/utils/botclass.py

def getLatestPricesBatch(self, symbols: list[str]) -> dict[str, float]:
    """
    Get latest prices for multiple symbols in a single DB query.

    Args:
        symbols: List of trading symbols to get prices for

    Returns:
        Dictionary mapping symbol to latest price
    """
    return self._data_service.get_latest_prices_batch(symbols)

getYFData(symbol: str | None = None, interval: str = '1m', period: str = '1d', saveToDB: bool = False) -> pd.DataFrame

Fetch market data from Yahoo Finance, checking database first.

Parameters:

Name	Type	Description	Default
symbol	str | None	Trading symbol (defaults to self.symbol)	None
interval	str	Data interval (e.g., "1m", "5m", "1h", "1d")	'1m'
period	str	Data period (e.g., "1d", "5d", "1mo", "1y")	'1d'
saveToDB	bool	Whether to save fetched data to database	False

Returns:

Type	Description
DataFrame	DataFrame with columns: symbol, timestamp, open, high, low, close, volume

Source code in tradingbot/utils/botclass.py

def getYFData(
    self,
    symbol: str | None = None,
    interval: str = "1m",
    period: str = "1d",
    saveToDB: bool = False,
) -> pd.DataFrame:
    """
    Fetch market data from Yahoo Finance, checking database first.

    Args:
        symbol: Trading symbol (defaults to self.symbol)
        interval: Data interval (e.g., "1m", "5m", "1h", "1d")
        period: Data period (e.g., "1d", "5d", "1mo", "1y")
        saveToDB: Whether to save fetched data to database

    Returns:
        DataFrame with columns: symbol, timestamp, open, high, low, close, volume
    """
    is_primary = False
    if not symbol:
        if self.symbol is None:
            raise ValueError("symbol parameter is required when self.symbol is None (multi-asset bot)")
        symbol = self.symbol
        is_primary = True
    elif symbol == self.symbol:
        is_primary = True

    data = self._data_service.get_yf_data(
        symbol=symbol,
        interval=interval,
        period=period,
        save_to_db=saveToDB,
        use_cache=True,
    )

    # Update cache for backward compatibility: only if it's the primary symbol
    # or it's a single-ticker bot.
    if (is_primary or len(self.tickers) <= 1) and (interval, period) == self.datasettings:
        self.data = data

    return data

getYFDataMultiple(symbols: list[str], interval: str = '1d', period: str = '3mo', saveToDB: bool = True) -> pd.DataFrame

Fetch market data for multiple symbols efficiently, checking database first.

Parameters:

Name	Type	Description	Default
symbols	list[str]	List of trading symbols to fetch	required
interval	str	Data interval (e.g., "1m", "5m", "1h", "1d")	'1d'
period	str	Data period (e.g., "1d", "5d", "1mo", "3mo", "1y")	'3mo'
saveToDB	bool	Whether to save fetched data to database for each symbol	True

Returns:

Type	Description
DataFrame	DataFrame with columns: symbol, timestamp, open, high, low, close, volume
DataFrame	Combined data from all symbols in long format

Source code in tradingbot/utils/botclass.py

def getYFDataMultiple(
    self,
    symbols: list[str],
    interval: str = "1d",
    period: str = "3mo",
    saveToDB: bool = True,
) -> pd.DataFrame:
    """
    Fetch market data for multiple symbols efficiently, checking database first.

    Args:
        symbols: List of trading symbols to fetch
        interval: Data interval (e.g., "1m", "5m", "1h", "1d")
        period: Data period (e.g., "1d", "5d", "1mo", "3mo", "1y")
        saveToDB: Whether to save fetched data to database for each symbol

    Returns:
        DataFrame with columns: symbol, timestamp, open, high, low, close, volume
        Combined data from all symbols in long format
    """
    return self._data_service.get_yf_data_multiple(
        symbols=symbols,
        interval=interval,
        period=period,
        save_to_db=saveToDB,
    )

getYFDataWithTA(symbol: str | None = None, interval: str = '1m', period: str = '1d', saveToDB: bool = False, features: list[str] | None = None) -> pd.DataFrame

Fetch market data with technical analysis indicators.

Data fetching strategy: - Checks database first for existing data - Only fetches from yfinance if data is missing or stale - If saveToDB=True, saves fetched data to database for future reuse

Note: For repeated backtests or hyperparameter tuning, set saveToDB=True to enable efficient data reuse.

Parameters:

Name	Type	Description	Default
symbol	str | None	Trading symbol (defaults to self.symbol)	None
interval	str	Data interval (e.g., "1m", "5m", "1h", "1d")	'1m'
period	str	Data period (e.g., "1d", "5d", "1mo", "1y")	'1d'
saveToDB	bool	Whether to save fetched data to database. Set to True for historical backtests to enable data reuse.	False
features	list[str] | None	Optional list of specific TA indicator column names to keep. If provided, drops other TA columns to save memory.	None

Returns:

Type	Description
DataFrame	DataFrame with market data and technical analysis features

Source code in tradingbot/utils/botclass.py

def getYFDataWithTA(
    self,
    symbol: str | None = None,
    interval: str = "1m",
    period: str = "1d",
    saveToDB: bool = False,
    features: list[str] | None = None,
) -> pd.DataFrame:
    """
    Fetch market data with technical analysis indicators.

    Data fetching strategy:
    - Checks database first for existing data
    - Only fetches from yfinance if data is missing or stale
    - If saveToDB=True, saves fetched data to database for future reuse

    Note: For repeated backtests or hyperparameter tuning, set saveToDB=True
    to enable efficient data reuse.

    Args:
        symbol: Trading symbol (defaults to self.symbol)
        interval: Data interval (e.g., "1m", "5m", "1h", "1d")
        period: Data period (e.g., "1d", "5d", "1mo", "1y")
        saveToDB: Whether to save fetched data to database. Set to True for
                 historical backtests to enable data reuse.
        features: Optional list of specific TA indicator column names to keep.
                  If provided, drops other TA columns to save memory.

    Returns:
        DataFrame with market data and technical analysis features
    """
    is_primary = False
    if not symbol:
        if self.symbol is None:
            raise ValueError("symbol parameter is required when self.symbol is None (multi-asset bot)")
        symbol = self.symbol
        is_primary = True
    elif symbol == self.symbol:
        is_primary = True

    data = self._data_service.get_yf_data_with_ta(
        symbol=symbol,
        interval=interval,
        period=period,
        save_to_db=saveToDB,
        features=features,
    )

    # Update cache for backward compatibility: only if it's the primary symbol
    # or it's a single-ticker bot.
    if (is_primary or len(self.tickers) <= 1) and (interval, period) == self.datasettings:
        self.data = data

    return data

get_ai_tools() -> list[Any]

Return custom LangChain tools for this bot. Override in subclasses to add bot-specific tools (e.g. get_tradeable_symbols, run_optimization). These are merged with the base tools when calling run_ai().

Source code in tradingbot/utils/botclass.py

def get_ai_tools(self) -> list[Any]:
    """
    Return custom LangChain tools for this bot. Override in subclasses to add
    bot-specific tools (e.g. get_tradeable_symbols, run_optimization).
    These are merged with the base tools when calling run_ai().
    """
    return []

local_backtest(initial_capital: float = 10000.0) -> dict[str, Any]

Local-only helper: run a backtest with current instance parameters.

Parameters:

Name	Type	Description	Default
initial_capital	float	Starting capital for backtest	10000.0

Returns:

Type	Description
dict[str, Any]	Backtest results dictionary

Source code in tradingbot/utils/botclass.py

def local_backtest(self, initial_capital: float = 10000.0) -> dict[str, Any]:
    """
    Local-only helper: run a backtest with current instance parameters.

    Args:
        initial_capital: Starting capital for backtest

    Returns:
        Backtest results dictionary
    """
    from .backtest import backtest_bot

    self._assert_backtestable()
    if getattr(self, "USE_OPTIONS", False):
        logger.warning(
            "%s trades options live, but this backtest holds the UNDERLYING: "
            "yfinance has no historical chains, so these numbers are not the option P&L.",
            self.bot_name,
        )
    results = backtest_bot(self, initial_capital=initial_capital)
    logger.info(f"\n--- Backtest Results: {self.bot_name} ---")
    logger.info(f"Yearly Return: {results['yearly_return']:.2%}")
    logger.info(f"Buy & Hold Return: {results['buy_hold_return']:.2%}")
    logger.info(f"Outperformance vs B&H: {(results['yearly_return'] - results['buy_hold_return']):+.2%}")
    logger.info(f"Sharpe Ratio: {results['sharpe_ratio']:.2f}")
    if results.get("alpha") is not None:
        logger.info(
            f"Alpha vs QQQ: {results['alpha']:+.2%}/yr (t={results['alpha_t']:.2f}), "
            f"Beta: {results['beta']:.2f}, Corr: {results['benchmark_corr']:.2f}"
        )
    else:
        logger.info("Alpha vs QQQ: n/a (no QQQ data overlapping the backtest)")
    logger.info(f"Number of Trades: {results['nrtrades']}")
    logger.info(f"Max Drawdown: {results['maxdrawdown']:.2%}")
    return results

local_development(param_grid: dict[str, list[Any]] | None = None, objective: str = 'alpha_t', initial_capital: float = 10000.0, n_jobs: int | None = None, param_sample_ratio: float = 1.0) -> dict[str, Any]

Convenience wrapper for the typical local development workflow:

1) Optimize hyperparameters for this bot's class 2) Backtest a bot instance constructed with the best parameters

Does NOT modify init defaults; you still paste them manually.

Parameters:

Name	Type	Description	Default
param_grid	dict[str, list[Any]] | None	Optional parameter grid to use. If None, uses self.param_grid or class attribute.	None
objective	str	Metric to maximize, one of hyperparameter_tuning.OBJECTIVES (default "alpha_t": t-stat of alpha vs QQQ)	'alpha_t'
initial_capital	float	Starting capital for backtests	10000.0
n_jobs	int | None	Number of parallel jobs (None = auto-detect)	None
param_sample_ratio	float	Fraction of param combinations to test (0.0–1.0). 1.0 = all (default). e.g. 0.2 = randomly test 20% of the grid.	1.0

Returns:

Type	Description
dict[str, Any]	Optimization results dictionary with 'best_params' and performance metrics

Example

bot = MyBot() results = bot.local_development()

Prints best parameters in copy-paste format

Then backtests with those parameters

Copy the printed parameters into init defaults

Source code in tradingbot/utils/botclass.py

def local_development(
    self,
    param_grid: dict[str, list[Any]] | None = None,
    objective: str = "alpha_t",
    initial_capital: float = 10000.0,
    n_jobs: int | None = None,
    param_sample_ratio: float = 1.0,
) -> dict[str, Any]:
    """
    Convenience wrapper for the typical local development workflow:

    1) Optimize hyperparameters for this bot's class
    2) Backtest a bot instance constructed with the best parameters

    Does NOT modify __init__ defaults; you still paste them manually.

    Args:
        param_grid: Optional parameter grid to use. If None, uses self.param_grid or class attribute.
        objective: Metric to maximize, one of hyperparameter_tuning.OBJECTIVES
                   (default "alpha_t": t-stat of alpha vs QQQ)
        initial_capital: Starting capital for backtests
        n_jobs: Number of parallel jobs (None = auto-detect)
        param_sample_ratio: Fraction of param combinations to test (0.0–1.0). 1.0 = all (default).
                           e.g. 0.2 = randomly test 20% of the grid.

    Returns:
        Optimization results dictionary with 'best_params' and performance metrics

    Example:
        bot = MyBot()
        results = bot.local_development()
        # Prints best parameters in copy-paste format
        # Then backtests with those parameters
        # Copy the printed parameters into __init__ defaults
    """
    self._assert_backtestable()

    # Step 1: Optimize
    opt_results = self.local_optimize(
        param_grid=param_grid,
        objective=objective,
        initial_capital=initial_capital,
        n_jobs=n_jobs,
        param_sample_ratio=param_sample_ratio,
    )

    # Step 2: Backtest with best parameters
    logger.info("\n" + "=" * 60)
    logger.info("Backtesting with best parameters...")
    logger.info("=" * 60)
    best_params = opt_results["best_params"]
    for key, value in best_params.items():
        logger.info(f"  {key}: {value}")
    best_bot = self.__class__(**best_params)
    best_bot.local_backtest(initial_capital=initial_capital)

    return opt_results

local_optimize(param_grid: dict[str, list[Any]] | None = None, objective: str = 'alpha_t', initial_capital: float = 10000.0, n_jobs: int | None = None, param_sample_ratio: float = 1.0) -> dict[str, Any]

Local-only helper: run hyperparameter optimization for this bot's class.

Uses either the provided param_grid or self.param_grid (if defined as class attribute). Prints the best combination in a format easy to copy-paste into init defaults.

Parameters:

Name	Type	Description	Default
param_grid	dict[str, list[Any]] | None	Optional parameter grid to use. If None, uses self.param_grid or class attribute.	None
objective	str	Metric to maximize, one of hyperparameter_tuning.OBJECTIVES (default "alpha_t": t-stat of alpha vs QQQ)	'alpha_t'
initial_capital	float	Starting capital for backtests	10000.0
n_jobs	int | None	Number of parallel jobs (None = auto-detect)	None
param_sample_ratio	float	Fraction of param combinations to test (0.0–1.0). 1.0 = all (default).	1.0

Returns:

Type	Description
dict[str, Any]	Full optimization results dictionary

Raises:

Type	Description
ValueError	If no param_grid is defined

Source code in tradingbot/utils/botclass.py

def local_optimize(
    self,
    param_grid: dict[str, list[Any]] | None = None,
    objective: str = "alpha_t",
    initial_capital: float = 10000.0,
    n_jobs: int | None = None,
    param_sample_ratio: float = 1.0,
) -> dict[str, Any]:
    """
    Local-only helper: run hyperparameter optimization for this bot's class.

    Uses either the provided param_grid or self.param_grid (if defined as class attribute).
    Prints the best combination in a format easy to copy-paste into __init__ defaults.

    Args:
        param_grid: Optional parameter grid to use. If None, uses self.param_grid or class attribute.
        objective: Metric to maximize, one of hyperparameter_tuning.OBJECTIVES
                   (default "alpha_t": t-stat of alpha vs QQQ)
        initial_capital: Starting capital for backtests
        n_jobs: Number of parallel jobs (None = auto-detect)
        param_sample_ratio: Fraction of param combinations to test (0.0–1.0). 1.0 = all (default).

    Returns:
        Full optimization results dictionary

    Raises:
        ValueError: If no param_grid is defined
    """
    from .hyperparameter_tuning import tune_hyperparameters

    self._assert_backtestable()

    # Use provided grid, or fall back to class attribute
    grid = param_grid or getattr(self, "param_grid", None) or self.__class__.param_grid
    if not grid:
        raise ValueError(
            f"No param_grid defined for {self.__class__.__name__}. "
            f"Either define param_grid as a class attribute or pass it to local_optimize()."
        )

    logger.info("=" * 60)
    logger.info(f"Hyperparameter optimization for {self.__class__.__name__}")
    logger.info("=" * 60)

    results = tune_hyperparameters(
        self.__class__,
        grid,
        objective=objective,
        initial_capital=initial_capital,
        verbose=True,
        n_jobs=n_jobs,
        param_sample_ratio=param_sample_ratio,
    )

    logger.info("\n" + "=" * 60)
    logger.info("Best parameters (paste into __init__ defaults):")
    logger.info("=" * 60)
    for key, value in results["best_params"].items():
        logger.info(f"    {key}: {value},")

    return results

makeOneIteration() -> int

Execute one iteration of the trading bot.

Default implementation: 1. Fetches data with technical indicators 2. Gets decision by applying decisionFunction() to data 3. Executes buy/sell based on decision

When to override: - Multi-asset bots (must override if self.symbol is None) - External data sources (e.g., Fear & Greed Index API) - Portfolio optimization strategies - Custom data processing beyond row-by-row logic

When NOT to override: - Simple single-asset strategies (just implement decisionFunction() instead) - Strategies that only need different timeframes (set interval/period in init)

Returns:

Type	Description
int	-1: Sold 0: No action 1: Bought

Raises:

Type	Description
NotImplementedError	If self.symbol is None (multi-asset bot) and method not overridden

Source code in tradingbot/utils/botclass.py

def makeOneIteration(self) -> int:
    """
    Execute one iteration of the trading bot.

    Default implementation:
    1. Fetches data with technical indicators
    2. Gets decision by applying decisionFunction() to data
    3. Executes buy/sell based on decision

    **When to override:**
    - Multi-asset bots (must override if self.symbol is None)
    - External data sources (e.g., Fear & Greed Index API)
    - Portfolio optimization strategies
    - Custom data processing beyond row-by-row logic

    **When NOT to override:**
    - Simple single-asset strategies (just implement decisionFunction() instead)
    - Strategies that only need different timeframes (set interval/period in __init__)

    Returns:
        -1: Sold
         0: No action
         1: Bought

    Raises:
        NotImplementedError: If self.symbol is None (multi-asset bot) and method not overridden
    """
    # Truly uninitialized bot (no symbol and no tickers)
    if self.symbol is None and len(self.tickers) == 0:
        raise NotImplementedError(
            "Bot has no symbol or tickers configured. "
            "Pass symbol= for single-asset or tickers= for multi-asset bots."
        )

    # targetWeights path: takes precedence over the -1/0/1 paths, and works
    # for any ticker count. Must come before the multi-ticker check.
    if type(self).targetWeights is not Bot.targetWeights:
        return self._run_target_weights_iteration()

    # Multi-ticker path: delegate to _run_multi_ticker_iteration
    if len(self.tickers) > 1:
        return self._run_multi_ticker_iteration()

    # Single-asset path
    data = self.getYFDataWithTA(saveToDB=True, interval=self.interval, period=self.period)
    # Make full dataset available so decisionFunction can access history
    # (e.g. Hurst exponent, rolling z-scores) without overriding makeOneIteration.
    self.data = data
    self.datasettings = (self.interval, self.period)
    decision = self.getLatestDecision(data)
    cash = self.dbBot.portfolio.get("USD", 0)
    holding = self._position_qty(self.symbol) if self.symbol is not None else 0
    if decision == 1 and cash > 0 and self.symbol is not None:
        self.buy(self.symbol)
        return 1
    elif decision == -1 and holding > 0 and self.symbol is not None:
        self.sell(self.symbol)
        return -1
    else:
        logger.info("No trade action taken (hold).")
    return 0

open_credit_spread(underlying: str, side: str, short_delta: float = 0.3, width: float = 10.0, dte: int = 35, max_risk_usd: float | None = None, view: options.ChainView | None = None) -> int

Sell a vertical credit spread: side="put" / "bull" is a bull put spread (short put at short_delta, long put width lower), side="call" / "bear" a bear call spread. Sized so that total max loss stays within max_risk_usd (default: all free cash). Returns the number opened. view reuses a chain the bot already loaded (options.load_chain).

Source code in tradingbot/utils/botclass.py

def open_credit_spread(
    self,
    underlying: str,
    side: str,
    short_delta: float = 0.30,
    width: float = 10.0,
    dte: int = 35,
    max_risk_usd: float | None = None,
    view: options.ChainView | None = None,
) -> int:
    """
    Sell a vertical credit spread: side="put" / "bull" is a bull put spread
    (short put at short_delta, long put `width` lower), side="call" /
    "bear" a bear call spread. Sized so that total max loss stays within
    max_risk_usd (default: all free cash). Returns the number opened.
    `view` reuses a chain the bot already loaded (options.load_chain).
    """
    right = {"put": "P", "bull": "P", "call": "C", "bear": "C"}.get(side.lower())
    if right is None:
        raise ValueError(f"side={side!r}: use 'put'/'bull' or 'call'/'bear'")
    pick = options.select_vertical(underlying, right, short_delta, width, dte, view=view)
    return self.open_structure(pick, max_risk_usd)

open_iron_condor(underlying: str, short_delta: float = 0.16, width: float = 10.0, dte: int = 35, max_risk_usd: float | None = None, view: options.ChainView | None = None, call_delta: float | None = None) -> int

Sell a put spread and a call spread on one expiry (calls at call_delta if given). Returns the number opened.

Source code in tradingbot/utils/botclass.py

def open_iron_condor(
    self,
    underlying: str,
    short_delta: float = 0.16,
    width: float = 10.0,
    dte: int = 35,
    max_risk_usd: float | None = None,
    view: options.ChainView | None = None,
    call_delta: float | None = None,
) -> int:
    """Sell a put spread and a call spread on one expiry (calls at call_delta if given). Returns the number opened."""
    pick = options.select_iron_condor(underlying, short_delta, width, dte, view=view, call_delta=call_delta)
    return self.open_structure(pick, max_risk_usd)

open_structure(pick: options.StructurePick, max_risk_usd: float | None = None) -> int

Open as many units of any options.select_* pick (butterfly, straddle, collar, calendar, diagonal, covered call...) as max_risk_usd of worst-case loss allows (default: all free cash). Returns units opened.

Source code in tradingbot/utils/botclass.py

def open_structure(self, pick: options.StructurePick, max_risk_usd: float | None = None) -> int:
    """
    Open as many units of any options.select_* pick (butterfly, straddle,
    collar, calendar, diagonal, covered call...) as `max_risk_usd` of
    worst-case loss allows (default: all free cash). Returns units opened.
    """
    units = self._portfolio_manager.open_structure(pick, math.inf if max_risk_usd is None else max_risk_usd)
    return units

option_book(underlying: str) -> options.OptionBook

Held options on underlying: legs, marks, entry value, P&L, DTE and net greeks.

Source code in tradingbot/utils/botclass.py

def option_book(self, underlying: str) -> options.OptionBook:
    """Held options on `underlying`: legs, marks, entry value, P&L, DTE and net greeks."""
    return self._portfolio_manager.option_book(underlying)

portfolio_value() -> float

Total book value: cash plus every holding at the latest price (short legs negative).

Source code in tradingbot/utils/botclass.py

def portfolio_value(self) -> float:
    """Total book value: cash plus every holding at the latest price (short legs negative)."""
    return self._portfolio_manager.total_value()

rebalancePortfolio(targetPortfolio: dict[str, float], onlyOver50USD: bool = False) -> None

Rebalance portfolio to match target weights.

Parameters:

Name	Type	Description	Default
targetPortfolio	dict[str, float]	Dictionary mapping symbols to target weights (e.g., {"VWCE": 0.8, "GLD": 0.1, "USD": 0.1}) Weights must sum to 1.0 (100%)	required
onlyOver50USD	bool	If True, filter out assets with target value <= $50 and redistribute weights equally among remaining assets (default: False)	False

Raises:

Type	Description
ValueError	If weights don't sum to 1.0 (within tolerance)

Source code in tradingbot/utils/botclass.py

def rebalancePortfolio(self, targetPortfolio: dict[str, float], onlyOver50USD: bool = False) -> None:
    """
    Rebalance portfolio to match target weights.

    Args:
        targetPortfolio: Dictionary mapping symbols to target weights (e.g., {"VWCE": 0.8, "GLD": 0.1, "USD": 0.1})
                       Weights must sum to 1.0 (100%)
        onlyOver50USD: If True, filter out assets with target value <= $50 and redistribute weights equally
                      among remaining assets (default: False)

    Raises:
        ValueError: If weights don't sum to 1.0 (within tolerance)
    """
    self._portfolio_manager.rebalance_portfolio(targetPortfolio, only_over_50_usd=onlyOver50USD)

run() -> None

Execute one iteration of the bot and log results.

Catches exceptions and logs them to the database before re-raising.

Source code in tradingbot/utils/botclass.py

def run(self) -> None:
    """
    Execute one iteration of the bot and log results.

    Catches exceptions and logs them to the database before re-raising.
    """
    bot_name = self.bot_name
    decision = -2
    try:
        # First of all: a split not yet applied makes the book look 1/ratio
        # its size, and the strategy would "buy the dip" (utils/splits.py).
        splits.apply_splits(bot_name)
        if options.option_legs(self.dbBot.portfolio):
            # Before the strategy runs, so it sees a book with no dead contracts.
            self._portfolio_manager.roll_and_settle_options(
                roll_dte=getattr(self, "OPTION_ROLL_DTE", 7),
                target_dte=getattr(self, "OPTION_TARGET_DTE", 30),
                target_delta=getattr(self, "OPTION_TARGET_DELTA", None),
                settlement=getattr(self, "OPTION_SETTLEMENT", "cash"),
            )
        decision = self.makeOneIteration()
        portfolio = self.dbBot.portfolio  # read after the iteration traded
        cash = portfolio.get("USD", 0)

        # Handle multi-asset bots gracefully
        if self.symbol:
            holding = self._position_qty(self.symbol)
            holding_info = f"Holding: {holding}"
        else:
            # For multi-asset bots, show portfolio summary
            non_usd_holdings = {k: v for k, v in portfolio.items() if k != "USD" and v != 0}
            holding_info = f"Holdings: {len(non_usd_holdings)} assets"

        logger.info("Decision: %s", decision)
        with get_db_session() as session:
            run = RunLog(
                bot_name=bot_name,
                success=True,
                result=f"Decision: {decision}, Cash: {cash}, {holding_info}",
            )
            session.add(run)
            # Context manager will commit automatically
    except Exception as e:
        logger.error("Error in makeOneIteration: %s", e)
        with get_db_session() as session:
            run = RunLog(
                bot_name=bot_name,
                success=False,
                result=str(e),
            )
            session.add(run)
            # Context manager will commit automatically
        raise e

run_ai(system_prompt: str, user_message: str, model: str | None = None, max_tool_rounds: int = 5, extra_tools: list[Any] | None = None, tool_names: list[str] | None = None) -> str

Run the AI with tools (market data, portfolio, recent trades, plus any custom tools) bound to this bot. Uses the main LLM (OPENROUTER_MAIN_MODEL, default deepseek/deepseek-v3.2). Pass model= to override. Merge custom tools via get_ai_tools() or extra_tools=. Optional tool_names= whitelists which base tools to include. Requires OPENROUTER_API_KEY.

Source code in tradingbot/utils/botclass.py

def run_ai(
    self,
    system_prompt: str,
    user_message: str,
    model: str | None = None,
    max_tool_rounds: int = 5,
    extra_tools: list[Any] | None = None,
    tool_names: list[str] | None = None,
) -> str:
    """
    Run the AI with tools (market data, portfolio, recent trades, plus any custom tools) bound to this bot.
    Uses the main LLM (OPENROUTER_MAIN_MODEL, default deepseek/deepseek-v3.2).
    Pass model= to override. Merge custom tools via get_ai_tools() or extra_tools=.
    Optional tool_names= whitelists which base tools to include. Requires OPENROUTER_API_KEY.
    """
    from .aitools import run_ai_with_tools

    merged_extra = list(self.get_ai_tools()) + (extra_tools or [])
    return run_ai_with_tools(
        system_prompt,
        user_message,
        self,
        model=model,
        max_tool_rounds=max_tool_rounds,
        extra_tools=merged_extra if merged_extra else None,
        tool_names=tool_names,
    )

run_ai_simple(system_prompt: str, user_message: str, model: str | None = None) -> str

Run the AI for a single-turn, no-tools task (summarization, extraction, classification, rewriting). Uses the cheap LLM (OPENROUTER_CHEAP_MODEL, default openrouter/free). Use run_ai() when you need tool access.

Source code in tradingbot/utils/botclass.py

def run_ai_simple(
    self,
    system_prompt: str,
    user_message: str,
    model: str | None = None,
) -> str:
    """
    Run the AI for a single-turn, no-tools task (summarization, extraction,
    classification, rewriting). Uses the cheap LLM (OPENROUTER_CHEAP_MODEL,
    default openrouter/free). Use run_ai() when you need tool access.
    """
    from .aitools import run_ai_simple as _run_ai_simple

    return _run_ai_simple(system_prompt, user_message, model=model)

run_ai_simple_with_fallback(system_prompt: str, user_message: str, sanity_check: Callable[[str], bool] | None = None, fallback_to_main: bool = True) -> str

Run a simple (no-tools) task with cheap LLM first; verify output for sanity; if validation fails, retry with main LLM. Prefer this over run_ai_simple when you want to save cost but still guarantee sane results.

Optional callable(response) -> bool. If None, uses a default

check (non-empty, no refusal/error prefix).

fallback_to_main: If True and sanity check fails, retry with main model.

Source code in tradingbot/utils/botclass.py

def run_ai_simple_with_fallback(
    self,
    system_prompt: str,
    user_message: str,
    sanity_check: Callable[[str], bool] | None = None,
    fallback_to_main: bool = True,
) -> str:
    """
    Run a simple (no-tools) task with cheap LLM first; verify output for sanity;
    if validation fails, retry with main LLM. Prefer this over run_ai_simple when
    you want to save cost but still guarantee sane results.

    sanity_check: Optional callable(response) -> bool. If None, uses a default
        check (non-empty, no refusal/error prefix).
    fallback_to_main: If True and sanity check fails, retry with main model.
    """
    from .aitools import run_ai_simple_with_fallback as _run_with_fallback

    return _run_with_fallback(
        system_prompt,
        user_message,
        sanity_check=sanity_check,
        fallback_to_main=fallback_to_main,
    )

sell(symbol: str, quantity_usd: float = -1, option: bool | str | None = None) -> None

Sell a quantity of the specified symbol.

Parameters:

Name	Type	Description	Default
symbol	str	Trading symbol to sell	required
quantity_usd	float	Amount in USD to sell (-1 means sell all holdings)	-1
option	bool | str | None	True sells the option contracts held on symbol (only "call" / "put" to restrict the side). None uses USE_OPTIONS.	None

Source code in tradingbot/utils/botclass.py

def sell(self, symbol: str, quantity_usd: float = -1, option: bool | str | None = None) -> None:
    """
    Sell a quantity of the specified symbol.

    Args:
        symbol: Trading symbol to sell
        quantity_usd: Amount in USD to sell (-1 means sell all holdings)
        option: True sells the option contracts held on `symbol` (only
            "call" / "put" to restrict the side). None uses USE_OPTIONS.
    """
    # Use per-ticker cache if available
    cached = self.datas.get(symbol, self.data)
    self._portfolio_manager.sell(
        symbol, quantity_usd=quantity_usd, cached_data=cached, option=self._option_mode(option)
    )

targetWeights(rows: dict[str, pd.Series]) -> dict[str, float] | None

Cross-sectional alternative to decisionFunction: return the target book.

Override this instead of decisionFunction when the strategy sizes positions rather than merely signalling them — vol targeting, risk parity, cross-sectional ranking. decisionFunction can only say -1/0/1, which the framework turns into equal weights; that is a different strategy from "hold each leg at constant ex-ante volatility", and expressing the latter used to require makeOneIteration and thereby forfeit backtesting entirely.

Parameters:

Name	Type	Description	Default
rows	dict[str, Series]	The current bar for EVERY ticker in self.tickers, benchmarks included — cross-sectional strategies need to see the whole universe before weighting any of it. For history, read self.datas[ticker], which the caller has already truncated to bars at or before the current one (so there is no way to peek ahead by accident).	required

Returns:

Type	Description
dict[str, float] | None	Long-only weights over TRADEABLE tickers, summing to <= 1.0.
dict[str, float] | None	"USD" must NOT be a key — cash is the residual, derived by the framework, so a bot cannot accidentally double-count it.
dict[str, float] | None	An omitted ticker means weight 0, i.e. a full exit that bypasses the no-trade band (the same semantics decisionFunction gives -1).
dict[str, float] | None	Weights are validated and clamped by _coerce_target_weights(), so returning something slightly off is corrected and logged rather than silently traded.
dict[str, float] | None	Returning {} goes fully to cash.
dict[str, float] | None	Returning None means "no rebalance this bar": the book is left exactly as it is, live and in the backtest. This is how a weekly strategy runs on daily bars — {} could not express it, because an omitted ticker is an exit.

Example

def targetWeights(self, rows): longs = [t for t in self.tradeable_tickers if rows[t]["close"] > self.datas[t]["close"].iloc[-200:].mean()] return {t: 1.0 / len(longs) for t in longs} if longs else {}

Source code in tradingbot/utils/botclass.py

def targetWeights(self, rows: dict[str, pd.Series]) -> dict[str, float] | None:
    """
    Cross-sectional alternative to decisionFunction: return the target book.

    Override this instead of decisionFunction when the strategy sizes
    positions rather than merely signalling them — vol targeting, risk
    parity, cross-sectional ranking. decisionFunction can only say -1/0/1,
    which the framework turns into equal weights; that is a different
    strategy from "hold each leg at constant ex-ante volatility", and
    expressing the latter used to require makeOneIteration and thereby
    forfeit backtesting entirely.

    Args:
        rows: The current bar for EVERY ticker in self.tickers, benchmarks
            included — cross-sectional strategies need to see the whole
            universe before weighting any of it. For history, read
            self.datas[ticker], which the caller has already truncated to
            bars at or before the current one (so there is no way to peek
            ahead by accident).

    Returns:
        Long-only weights over TRADEABLE tickers, summing to <= 1.0.

        * "USD" must NOT be a key — cash is the residual, derived by the
          framework, so a bot cannot accidentally double-count it.
        * An omitted ticker means weight 0, i.e. a full exit that bypasses
          the no-trade band (the same semantics decisionFunction gives -1).
        * Weights are validated and clamped by _coerce_target_weights(), so
          returning something slightly off is corrected and logged rather
          than silently traded.
        * Returning {} goes fully to cash.
        * Returning None means "no rebalance this bar": the book is left
          exactly as it is, live and in the backtest. This is how a
          weekly strategy runs on daily bars — {} could not express it,
          because an omitted ticker is an exit.

    Example:
        def targetWeights(self, rows):
            longs = [t for t in self.tradeable_tickers
                     if rows[t]["close"] > self.datas[t]["close"].iloc[-200:].mean()]
            return {t: 1.0 / len(longs) for t in longs} if longs else {}
    """
    raise NotImplementedError("You need to overwrite targetWeights!!!!")

trade_option_legs(legs: list[tuple[str, float]]) -> float

Change several legs (contracts in share-equivalents, or the underlying's shares) atomically.

Source code in tradingbot/utils/botclass.py

def trade_option_legs(self, legs: list[tuple[str, float]]) -> float:
    """Change several legs (contracts in share-equivalents, or the underlying's shares) atomically."""
    cash = self._portfolio_manager.trade_option_legs(legs)
    return cash