A flexible backtesting framework for Python designed for quantitative trading strategy development and testing.
—
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
Comprehensive library of 50+ algorithmic building blocks for strategy logic, organized into functional categories. These components can be mixed and matched to create complex trading strategies while maintaining modularity and reusability.
Control when strategy logic executes based on time periods, dates, or other conditions.
class RunOnce(Algo):
"""Returns True on first run, then False."""
def __init__(self): ...
def __call__(self, target): ...
class RunDaily(RunPeriod):
"""Returns True on day change."""
def __init__(self, run_on_first_date=True, run_on_end_of_period=False, run_on_last_date=False): ...
def __call__(self, target): ...
class RunWeekly(RunPeriod):
"""Returns True on week change."""
def __init__(self, run_on_first_date=True, run_on_end_of_period=False, run_on_last_date=False): ...
class RunMonthly(RunPeriod):
"""Returns True on month change."""
def __init__(self, run_on_first_date=True, run_on_end_of_period=False, run_on_last_date=False): ...
class RunQuarterly(RunPeriod):
"""Returns True on quarter change."""
def __init__(self, run_on_first_date=True, run_on_end_of_period=False, run_on_last_date=False): ...
class RunYearly(RunPeriod):
"""Returns True on year change."""
def __init__(self, run_on_first_date=True, run_on_end_of_period=False, run_on_last_date=False): ...
class RunOnDate(Algo):
"""Returns True on specific dates."""
def __init__(self, *dates): ...
class RunAfterDate(Algo):
"""Returns True after specified date."""
def __init__(self, date): ...
class RunAfterDays(Algo):
"""Returns True after specified trading days."""
def __init__(self, days: int): ...
class RunIfOutOfBounds(Algo):
"""Returns True if weights deviate beyond tolerance."""
def __init__(self, tolerance: float): ...
class RunEveryNPeriods(Algo):
"""Runs every N periods."""
def __init__(self, n: int, offset=0): ...Select which securities to include in the strategy based on various criteria.
class SelectAll(Algo):
"""Selects all securities from universe."""
def __init__(self, include_no_data=False, include_negative=False): ...
class SelectThese(Algo):
"""Selects specific list of tickers."""
def __init__(self, tickers: list, include_no_data=False, include_negative=False): ...
class SelectHasData(Algo):
"""Selects securities with sufficient data history."""
def __init__(self, lookback=None, min_count=None, include_no_data=False, include_negative=False): ...
class SelectN(Algo):
"""Selects top/bottom N securities based on ranking."""
def __init__(self, n: int, sort_descending=True, all_or_none=False, filter_selected=False): ...
class SelectMomentum(AlgoStack):
"""Selects securities based on momentum ranking."""
def __init__(self, n: int, lookback=None, lag=None, sort_descending=True, all_or_none=False): ...
class SelectWhere(Algo):
"""Selects securities based on boolean indicator."""
def __init__(self, signal, include_no_data=False, include_negative=False): ...
class SelectRandomly(Algo):
"""Selects random subset of securities."""
def __init__(self, n=None, include_no_data=False, include_negative=False): ...
class SelectRegex(Algo):
"""Selects securities matching regex pattern."""
def __init__(self, regex: str): ...
class SelectTypes(Algo):
"""Selects securities by node type."""
def __init__(self, include_types=None, exclude_types=None): ...
class SelectActive(Algo):
"""Filters out closed or rolled securities."""
def __call__(self, target): ...
class ResolveOnTheRun(Algo):
"""Resolves on-the-run security aliases."""
def __init__(self, on_the_run: dict, include_no_data=False, include_negative=False): ...Determine how to allocate capital among selected securities.
class WeighEqually(Algo):
"""Sets equal weights for selected securities."""
def __call__(self, target): ...
class WeighSpecified(Algo):
"""Sets weights from specified dictionary."""
def __init__(self, **weights): ...
class ScaleWeights(Algo):
"""Scales existing weights by factor."""
def __init__(self, scale: float): ...
class WeighTarget(Algo):
"""Sets weights from target DataFrame."""
def __init__(self, weights): ...
class WeighInvVol(Algo):
"""Sets inverse volatility weights (risk parity)."""
def __init__(self, lookback=None, lag=None): ...
class WeighERC(Algo):
"""Sets equal risk contribution weights."""
def __init__(self, lookback=None, initial_weights=None, risk_weights=None,
covar_method="ledoit-wolf", risk_parity_method="ccd",
maximum_iterations=100, tolerance=1e-8, lag=None): ...
class WeighMeanVar(Algo):
"""Sets mean-variance optimization weights."""
def __init__(self, lookback=None, bounds=(0.0, 1.0), covar_method="ledoit-wolf",
rf=0.0, lag=None): ...
class WeighRandomly(Algo):
"""Sets random weights for benchmarking."""
def __init__(self, bounds=(0.0, 1.0), weight_sum=1): ...Manage portfolio positions, risk, and execution.
class Rebalance(Algo):
"""Main rebalancing algorithm using temp weights."""
def __call__(self, target): ...
class RebalanceOverTime(Algo):
"""Rebalances to target over multiple periods."""
def __init__(self, n=10): ...
class LimitDeltas(Algo):
"""Limits weight changes between periods."""
def __init__(self, limit=0.1): ...
class LimitWeights(Algo):
"""Limits maximum individual asset weights."""
def __init__(self, limit=0.1): ...
class TargetVol(Algo):
"""Scales weights to target volatility."""
def __init__(self, target_volatility: float, lookback=None, lag=None,
covar_method="standard", annualization_factor=252): ...
class PTE_Rebalance(Algo):
"""Triggers rebalance based on tracking error."""
def __init__(self, PTE_volatility_cap: float, target_weights, lookback=None,
lag=None, covar_method="standard", annualization_factor=252): ...
class CapitalFlow(Algo):
"""Models capital inflows/outflows."""
def __init__(self, amount: float): ...
class CloseDead(Algo):
"""Closes positions with zero prices."""
def __call__(self, target): ...
class SetNotional(Algo):
"""Sets notional value for fixed income strategies."""
def __init__(self, notional_value: float): ...Specialized algorithms for complex portfolio operations.
class ClosePositionsAfterDates(Algo):
"""Closes positions after specified dates."""
def __init__(self, close_dates: dict): ...
class RollPositionsAfterDates(Algo):
"""Rolls securities based on mapping."""
def __init__(self, roll_data: dict): ...
class ReplayTransactions(Algo):
"""Replays executed transactions."""
def __init__(self, transactions): ...
class SimulateRFQTransactions(Algo):
"""Simulates RFQ transaction outcomes."""
def __init__(self, rfqs, model): ...Calculate and store statistical measures for use by other algorithms.
class SetStat(Algo):
"""Sets statistic in temp data for downstream use."""
def __init__(self, stat: str, lag=None): ...
class StatTotalReturn(Algo):
"""Calculates total returns over period."""
def __init__(self, lookback=None, lag=None): ...Monitor and manage risk exposures across the strategy.
class UpdateRisk(Algo):
"""Tracks risk measures across strategy nodes."""
def __init__(self, measure: str, history=0): ...
class PrintRisk(Algo):
"""Prints risk data for debugging."""
def __init__(self, fmt_string=""): ...
class HedgeRisks(Algo):
"""Hedges risk measures with instruments."""
def __init__(self, measures: dict, pseudo=False, strategy=None, throw_nan=True): ...Control algorithm execution flow with logical operations.
class Require(Algo):
"""Controls flow based on predicate evaluation."""
def __init__(self, pred, item, if_none=False): ...
class Not(Algo):
"""Inverts return value of another algorithm."""
def __init__(self, algo: Algo): ...
class Or(Algo):
"""Combines multiple signals with OR logic."""
def __init__(self, list_of_algos: list): ...Utility algorithms for debugging and monitoring strategy execution.
class PrintDate(Algo):
"""Prints current date for debugging."""
def __call__(self, target): ...
class PrintTempData(Algo):
"""Prints temporary data with optional formatting."""
def __init__(self, fmt_string=None): ...
class PrintInfo(Algo):
"""Prints strategy information with formatting."""
def __init__(self, fmt_string="{name} {now}"): ...
class Debug(Algo):
"""Triggers debugger (pdb.set_trace)."""
def __call__(self, target): ...def run_always(f):
"""Ensure algorithm runs on each pass regardless of stack failures."""
...# Monthly momentum strategy
strategy = bt.Strategy('Momentum', [
bt.algos.RunMonthly(),
bt.algos.SelectAll(),
bt.algos.SelectMomentum(10, lookback=pd.DateOffset(months=3)),
bt.algos.WeighEqually(),
bt.algos.Rebalance()
])
# Risk parity with volatility targeting
strategy = bt.Strategy('RiskParity', [
bt.algos.RunMonthly(),
bt.algos.SelectAll(),
bt.algos.WeighInvVol(lookback=pd.DateOffset(months=6)),
bt.algos.TargetVol(0.12),
bt.algos.Rebalance()
])# Rebalance only when out of bounds
strategy = bt.Strategy('ConditionalRebalance', [
bt.algos.RunDaily(),
bt.algos.RunIfOutOfBounds(0.05), # Only if >5% deviation
bt.algos.SelectAll(),
bt.algos.WeighEqually(),
bt.algos.Rebalance()
])# Select top 20 momentum stocks with data requirements
strategy = bt.Strategy('TopMomentum', [
bt.algos.RunMonthly(),
bt.algos.SelectHasData(lookback=pd.DateOffset(months=12)),
bt.algos.SelectMomentum(20, lookback=pd.DateOffset(months=3)),
bt.algos.WeighEqually(),
bt.algos.Rebalance()
])Install with Tessl CLI
npx tessl i tessl/pypi-bt