"""
A collection of Algos used to create Strategy logic.
"""
import abc
import math
import random
import re
import numpy as np
import pandas as pd
import sklearn.covariance
import bt
from bt.core import Algo, AlgoStack, SecurityBase, is_zero
_DEFAULT_LOOKBACK = pd.DateOffset(months=3)
_DEFAULT_LAG = pd.DateOffset(days=0)
[docs]
def run_always(f):
"""
Run always decorator to be used with Algo
to ensure stack runs the decorated Algo
on each pass, regardless of failures in the stack.
"""
f.run_always = True
return f
[docs]
class PrintDate(Algo):
"""
This Algo simply print's the current date.
Can be useful for debugging purposes.
"""
def __call__(self, target):
print(target.now)
return True
[docs]
class PrintTempData(Algo):
"""
This Algo prints the temp data.
Useful for debugging.
Args:
* fmt_string (str): A string that will later be formatted with the
target's temp dict. Therefore, you should provide
what you want to examine within curly braces ( { } )
"""
def __init__(self, fmt_string=None):
super().__init__()
self.fmt_string = fmt_string
def __call__(self, target):
if self.fmt_string:
print(self.fmt_string.format(**target.temp))
else:
print(target.temp)
return True
[docs]
class PrintInfo(Algo):
"""
Prints out info associated with the target strategy. Useful for debugging
purposes.
Args:
* fmt_string (str): A string that will later be formatted with the
target object's __dict__ attribute. Therefore, you should provide
what you want to examine within curly braces ( { } )
Ex:
PrintInfo('Strategy {name} : {now}')
This will print out the name and the date (now) on each call.
Basically, you provide a string that will be formatted with target.__dict__
"""
def __init__(self, fmt_string="{name} {now}"):
super().__init__()
self.fmt_string = fmt_string
def __call__(self, target):
print(self.fmt_string.format(**target.__dict__))
return True
[docs]
class Debug(Algo):
"""
Utility Algo that calls pdb.set_trace when triggered.
In the debug session, 'target' is available and can be examined through the
StrategyBase interface.
"""
def __call__(self, target):
import pdb # noqa: T100
pdb.set_trace() # noqa: T100
return True
[docs]
class RunOnce(Algo):
"""
Returns True on first run then returns False.
Args:
* run_on_first_call: bool which determines if it runs the first time the algo is called
As the name says, the algo only runs once. Useful in situations
where we want to run the logic once (buy and hold for example).
"""
def __init__(self):
super().__init__()
self.has_run = False
def __call__(self, target):
# if it hasn't run then we will
# run it and set flag
if not self.has_run:
self.has_run = True
return True
# return false to stop future execution
return False
[docs]
class RunPeriod(Algo):
"""
Schedule from period boundaries in the supplied dates.
The first supplied date uses ``run_on_first_date``, even if it is also
the last. On other final dates, ``run_on_last_date=False`` skips execution
and ``True`` forces it. ``None`` uses the preceding date in beginning-of-period
mode, but returns False in end-of-period mode because no following date exists.
"""
def __init__(self, run_on_first_date=True, run_on_end_of_period=False, run_on_last_date=False):
super().__init__()
self._run_on_first_date = run_on_first_date
self._run_on_end_of_period = run_on_end_of_period
self._run_on_last_date = run_on_last_date
def __call__(self, target):
# get last date
now = target.now
# if none nothing to do - return false
if now is None:
return False
# not a known date in our universe
if now not in target.data.index:
return False
# get index of the current date
index = target.data.index.get_loc(target.now)
result = False
# index 0 is a date added by the Backtest Constructor
if index == 0:
return False
# first date
if index == 1:
if self._run_on_first_date:
result = True
# None can compare the preceding bar, but cannot infer a missing successor.
elif index == (len(target.data.index) - 1) and (self._run_on_last_date is not None or self._run_on_end_of_period):
if self._run_on_last_date:
result = True
else:
# create pandas.Timestamp for useful .week,.quarter properties
now = pd.Timestamp(now)
index_offset = -1
if self._run_on_end_of_period:
index_offset = 1
date_to_compare = target.data.index[index + index_offset]
date_to_compare = pd.Timestamp(date_to_compare)
result = self.compare_dates(now, date_to_compare)
return result
[docs]
@abc.abstractmethod
def compare_dates(self, now, date_to_compare):
raise (NotImplementedError("RunPeriod Algo is an abstract class!"))
[docs]
class RunDaily(RunPeriod):
"""
Returns True on day change.
Args:
* run_on_first_date (bool): determines if it runs the first time the algo is called
* run_on_end_of_period (bool): determines if it should run at the end of the period
or the beginning
* run_on_last_date (bool or None): False (default) skips the final supplied date;
True forces execution; None checks the preceding date, or returns False in
end-of-period mode. The first-date override takes precedence (see RunPeriod).
Returns True if the target.now's day has changed
compared to the last(or next if run_on_end_of_period) date, if not returns False.
Useful for daily rebalancing strategies.
"""
[docs]
def compare_dates(self, now, date_to_compare):
return now.date() != date_to_compare.date()
[docs]
class RunWeekly(RunPeriod):
"""
Returns True on week change.
Args:
* run_on_first_date (bool): determines if it runs the first time the algo is called
* run_on_end_of_period (bool): determines if it should run at the end of the period
or the beginning
* run_on_last_date (bool or None): False (default) skips the final supplied date;
True forces execution; None checks the preceding date, or returns False in
end-of-period mode. The first-date override takes precedence (see RunPeriod).
Returns True if the target.now's week has changed
since relative to the last(or next) date, if not returns False. Useful for
weekly rebalancing strategies.
"""
[docs]
def compare_dates(self, now, date_to_compare):
return now.isocalendar()[:2] != date_to_compare.isocalendar()[:2]
[docs]
class RunMonthly(RunPeriod):
"""
Returns True on month change.
Args:
* run_on_first_date (bool): determines if it runs the first time the algo is called
* run_on_end_of_period (bool): determines if it should run at the end of the period
or the beginning
* run_on_last_date (bool or None): False (default) skips the final supplied date;
True forces execution; None checks the preceding date, or returns False in
end-of-period mode. The first-date override takes precedence (see RunPeriod).
Returns True if the target.now's month has changed
since relative to the last(or next) date, if not returns False. Useful for
monthly rebalancing strategies.
"""
[docs]
def compare_dates(self, now, date_to_compare):
return bool(now.year != date_to_compare.year or now.month != date_to_compare.month)
[docs]
class RunQuarterly(RunPeriod):
"""
Returns True on quarter change.
Args:
* run_on_first_date (bool): determines if it runs the first time the algo is called
* run_on_end_of_period (bool): determines if it should run at the end of the period
or the beginning
* run_on_last_date (bool or None): False (default) skips the final supplied date;
True forces execution; None checks the preceding date, or returns False in
end-of-period mode. The first-date override takes precedence (see RunPeriod).
Returns True if the target.now's quarter has changed
since relative to the last(or next) date, if not returns False. Useful for
quarterly rebalancing strategies.
"""
[docs]
def compare_dates(self, now, date_to_compare):
return bool(now.year != date_to_compare.year or now.quarter != date_to_compare.quarter)
[docs]
class RunYearly(RunPeriod):
"""
Returns True on year change.
Args:
* run_on_first_date (bool): determines if it runs the first time the algo is called
* run_on_end_of_period (bool): determines if it should run at the end of the period
or the beginning
* run_on_last_date (bool or None): False (default) skips the final supplied date;
True forces execution; None checks the preceding date, or returns False in
end-of-period mode. The first-date override takes precedence (see RunPeriod).
Returns True if the target.now's year has changed
since relative to the last(or next) date, if not returns False. Useful for
yearly rebalancing strategies.
"""
[docs]
def compare_dates(self, now, date_to_compare):
return now.year != date_to_compare.year
[docs]
class RunOnDate(Algo):
"""
Returns True on a specific set of dates.
Args:
* dates (list): List of dates to run Algo on.
"""
def __init__(self, *dates):
"""
Args:
* dates (*args): A list of dates. Dates will be parsed
by pandas.to_datetime so pass anything that it can
parse. Typically, you will pass a string 'yyyy-mm-dd'.
"""
super().__init__()
# parse dates and save
self.dates = [pd.to_datetime(d) for d in dates]
def __call__(self, target):
return target.now in self.dates
[docs]
class RunAfterDate(Algo):
"""
Returns True after a date has passed
Args:
* date: Date after which to start trading
Note:
This is useful for algos that rely on trailing averages where you
don't want to start trading until some amount of data has been built up
"""
def __init__(self, date):
"""
Args:
* date: Date after which to start trading
"""
super().__init__()
# parse dates and save
self.date = pd.to_datetime(date)
def __call__(self, target):
return target.now > self.date
[docs]
class RunAfterDays(Algo):
"""
Returns True after a specific number of 'warmup' trading days have passed
Repeated calls on the same date count as one trading day.
Args:
* days (int): Number of trading days to wait before starting
Note:
This is useful for algos that rely on trailing averages where you
don't want to start trading until some amount of data has been built up
"""
def __init__(self, days):
"""
Args:
* days (int): Number of trading days to wait before starting
"""
super().__init__()
self.days = days
self.last_date = None
def __call__(self, target):
if self.days > 0:
if target.now != self.last_date:
self.days -= 1
self.last_date = target.now
return False
return self.last_date is None or bool(target.now != self.last_date)
[docs]
class RunIfOutOfBounds(Algo):
"""
This algo returns true if any current or target weight deviates by an amount
greater than tolerance. For example, it will be run if the tolerance is set to
0.5 and a security grows from a target weight of 0.2 to greater than 0.3.
A missing current or target weight is treated as zero.
A zero target weight is out of bounds whenever the current weight is nonzero.
For ordinary strategies, an optional ``cash`` fraction scales the supplied
weights to the investable portion of total value and is checked against the
strategy's actual cash fraction. Fixed-income strategies ignore ``cash``.
A zero-value ordinary strategy is within its cash target only when its
capital is also zero.
A strategy where rebalancing is performed quarterly or whenever any
security's weight deviates by more than 20% could be implemented by:
Or([runQuarterlyAlgo,runIfOutOfBoundsAlgo(0.2)])
Args:
* tolerance (float): Allowed deviation of each security weight.
Requires:
* Weights
"""
def __init__(self, tolerance):
self.tolerance = float(tolerance)
super().__init__()
def __call__(self, target):
if "weights" not in target.temp:
return True
targets = target.temp["weights"]
target_scale = 1.0
if "cash" in target.temp and not target.fixed_income:
target_scale -= target.temp["cash"]
# Rebalance treats names missing from either side as zero weight.
for cname in target.children.keys() | targets.keys():
current_weight = target.children[cname].weight if cname in target.children else 0.0
target_weight = targets.get(cname, 0.0) * target_scale
if target_weight == 0:
if current_weight != 0:
return True
continue
deviation = (current_weight - target_weight) / target_weight
if deviation > self.tolerance or deviation < -self.tolerance:
return True
if "cash" in target.temp and not target.fixed_income:
value = target.value
if is_zero(value):
return not is_zero(target.capital)
cash_deviation = abs(target.capital / value - target.temp["cash"])
if cash_deviation > self.tolerance:
return True
return False
[docs]
class RunEveryNPeriods(Algo):
"""
This algo runs every n periods.
Args:
* n (int): Run each n periods
* offset (int): Applies to the first run. If 0, this algo will run the
first time it is called.
This Algo can be useful for the following type of strategy:
Each month, select the top 5 performers. Hold them for 3 months.
You could then create 3 strategies with different offsets and create a
master strategy that would allocate equal amounts of capital to each.
"""
def __init__(self, n, offset=0):
super().__init__()
self.n = n
self.offset = offset
self.idx = n - offset - 1
self.lcall = 0
def __call__(self, target):
# ignore multiple calls on same period
if self.lcall == target.now:
return False
else:
self.lcall = target.now
# run when idx == (n-1)
if self.idx == (self.n - 1):
self.idx = 0
return True
else:
self.idx += 1
return False
def _select_by_price(prices, candidates, include_no_data, include_negative):
"""Return labels allowed by the independent data and price-sign flags."""
if include_no_data:
prices = prices.reindex(candidates)
else:
prices = prices.loc[candidates].dropna()
if not include_negative:
prices = prices[prices.isna() | (prices > 0)]
return list(prices.index)
[docs]
class SelectAll(Algo):
"""
Sets temp['selected'] with all securities (based on universe).
Selects all the securities and saves them in temp['selected'].
By default, SelectAll does not include securities that have no
data (nan) on current date or those whose price is zero or negative.
Args:
* include_no_data (bool): Include securities that do not have data?
* include_negative (bool): Include securities that have negative
or zero prices?
Sets:
* selected
"""
def __init__(self, include_no_data=False, include_negative=False):
super().__init__()
self.include_no_data = include_no_data
self.include_negative = include_negative
def __call__(self, target):
if self.include_no_data and self.include_negative:
target.temp["selected"] = target.universe.columns
else:
prices = target.universe.loc[target.now]
target.temp["selected"] = _select_by_price(
prices,
target.universe.columns,
self.include_no_data,
self.include_negative,
)
return True
[docs]
class SelectThese(Algo):
"""
Sets temp['selected'] with a set list of tickers.
Args:
* ticker (list): List of tickers to select.
* include_no_data (bool): Include securities that do not have data?
* include_negative (bool): Include securities that have negative
or zero prices?
Sets:
* selected
"""
def __init__(self, tickers, include_no_data=False, include_negative=False):
super().__init__()
self.tickers = tickers
self.include_no_data = include_no_data
self.include_negative = include_negative
def __call__(self, target):
if self.include_no_data and self.include_negative:
target.temp["selected"] = self.tickers
else:
prices = target.universe.loc[target.now]
target.temp["selected"] = _select_by_price(prices, self.tickers, self.include_no_data, self.include_negative)
return True
[docs]
class SelectHasData(Algo):
"""
Sets temp['selected'] based on all items in universe that meet
data requirements.
This is a more advanced version of SelectAll. Useful for selecting
tickers that need a certain amount of data for future algos to run
properly.
For example, if we need the items with 3 months of data or more,
we could use this Algo with a lookback period of 3 months.
When providing a lookback period, it is also wise to provide a min_count.
This is basically the number of data points needed within the lookback
period for a series to be considered valid. For example, in our 3 month
lookback above, we might want to specify the min_count as being
57 -> a typical trading month has give or take 20 trading days. If we
factor in some holidays, we can use 57 or 58. It's really up to you.
If you don't specify min_count, min_count will default to ffn's
get_num_days_required.
Args:
* lookback (DateOffset): A DateOffset that determines the lookback
period.
* min_count (int): Minimum number of days required for a series to be
considered valid. If not provided, ffn's get_num_days_required is
used to estimate the number of points required.
* include_no_data (bool): Include securities that do not have data?
* include_negative (bool): Include securities that have negative
or zero prices?
Sets:
* selected
"""
def __init__(
self,
lookback=_DEFAULT_LOOKBACK,
min_count=None,
include_no_data=False,
include_negative=False,
):
super().__init__()
self.lookback = lookback
if min_count is None:
min_count = bt.ffn.get_num_days_required(lookback)
self.min_count = min_count
self.include_no_data = include_no_data
self.include_negative = include_negative
def __call__(self, target):
if "selected" in target.temp:
selected = target.temp["selected"]
else:
selected = target.universe.columns
filt = target.universe.loc[target.now - self.lookback :, selected]
cnt = filt.count()
cnt = cnt[cnt >= self.min_count]
prices = target.universe.loc[target.now]
target.temp["selected"] = _select_by_price(prices, cnt.index, self.include_no_data, self.include_negative)
return True
[docs]
class SelectN(Algo):
"""
Sets temp['selected'] based on ranking temp['stat'].
Selects the top or botton N items based on temp['stat'].
This is usually some kind of metric that will be computed in a
previous Algo and will be used for ranking purposes. Can select
top or bottom N based on sort_descending parameter.
Args:
* n (int): select top n items.
* sort_descending (bool): Should the stat be sorted in descending order
before selecting the first n items?
* all_or_none (bool): If true, only populates temp['selected'] if we
have n items. If we have less than n, then temp['selected'] = [].
* filter_selected (bool): If True, will only select from the existing
'selected' list.
Sets:
* selected
Requires:
* stat
"""
def __init__(self, n, sort_descending=True, all_or_none=False, filter_selected=False):
super().__init__()
if n < 0:
raise ValueError("n cannot be negative")
self.n = n
self.ascending = not sort_descending
self.all_or_none = all_or_none
self.filter_selected = filter_selected
def __call__(self, target):
stat = target.temp["stat"].dropna()
if self.filter_selected and "selected" in target.temp:
stat = stat.loc[stat.index.intersection(target.temp["selected"])]
stat.sort_values(ascending=self.ascending, inplace=True)
# handle percent n
keep_n = self.n
if self.n < 1:
keep_n = int(self.n * len(stat))
sel = list(stat[:keep_n].index)
if self.all_or_none and len(sel) < keep_n:
sel = []
target.temp["selected"] = sel
return True
[docs]
class SelectMomentum(AlgoStack):
"""
Sets temp['selected'] based on a simple momentum filter.
Selects the top n securities based on the total return over
a given lookback period. This is just a wrapper around an
AlgoStack with two algos: StatTotalReturn and SelectN.
Note, that SelectAll() or similar should be called before
SelectMomentum(), as StatTotalReturn uses values of temp['selected']
Args:
* n (int): select first N elements
* lookback (DateOffset|sequence): lookback period for total return
calculation, or a sequence of periods whose returns are averaged
* lag (DateOffset): Lag interval for total return calculation
* sort_descending (bool): Sort descending (highest return is best)
* all_or_none (bool): If true, only populates temp['selected'] if we
have n items. If we have less than n, then temp['selected'] = [].
* weights (sequence): Optional finite, non-negative weights for multiple
lookback periods. Their sum must be finite and positive.
Sets:
* selected
Requires:
* selected
"""
def __init__(
self,
n,
lookback=_DEFAULT_LOOKBACK,
lag=_DEFAULT_LAG,
sort_descending=True,
all_or_none=False,
weights=None,
):
if isinstance(lookback, (list, tuple, np.ndarray, pd.Index)):
stat = StatMultiPeriodReturn(lookbacks=lookback, weights=weights, lag=lag)
else:
if weights is not None:
raise ValueError("weights require multiple lookback periods")
stat = StatTotalReturn(lookback=lookback, lag=lag)
super().__init__(
stat,
SelectN(n=n, sort_descending=sort_descending, all_or_none=all_or_none),
)
[docs]
class SelectWhere(Algo):
"""
Selects securities based on an indicator DataFrame.
Selects securities where the value is True on the current date
(target.now). If the date is absent from the signal DataFrame, returns
False to stop the AlgoStack. A present all-False row remains a valid empty
selection and returns True.
For example, this could be the result of a pandas boolean comparison such
as data > 100.
Warning:
Signals derived from fundamental data are only point-in-time safe when
the data contains values known on each historical date. Data containing
later restatements introduces look-ahead bias. SelectWhere evaluates the
supplied signal as-is and cannot detect or correct restated values.
Args:
* signal (str|DataFrame): Boolean DataFrame containing selection logic.
If a string is passed, frame is accessed using target.get_data
This is the preferred way of using the algo.
* include_no_data (bool): Include securities that do not have data?
* include_negative (bool): Include securities that have negative
or zero prices?
Sets:
* selected
"""
def __init__(self, signal, include_no_data=False, include_negative=False):
super().__init__()
if isinstance(signal, pd.DataFrame):
self.signal_name = None
self.signal = signal
else:
self.signal_name = signal
self.signal = None
self.include_no_data = include_no_data
self.include_negative = include_negative
def __call__(self, target):
# get signal Series at target.now
if self.signal_name is None:
signal = self.signal
else:
signal = target.get_data(self.signal_name)
if target.now in signal.index:
sig = signal.loc[target.now]
# get tickers where True
# selected = sig.index[sig]
selected = sig[sig == True].index
# save as list
prices = target.universe.loc[target.now]
selected = _select_by_price(prices, selected, self.include_no_data, self.include_negative)
target.temp["selected"] = list(selected)
return True
# Stop downstream Algos when no signal row exists for the current date.
return False
[docs]
class SelectRandomly(AlgoStack):
"""
Sets temp['selected'] based on a random subset of
the items currently in temp['selected'].
Selects n random elements from the list stored in temp['selected'].
This is useful for benchmarking against a strategy where we believe
the selection algorithm is adding value.
For example, if we are testing a momentum strategy and we want to see if
selecting securities based on momentum is better than just selecting
securities randomly, we could use this Algo to create a random Strategy
used for random benchmarking.
Note:
Another selection algorithm should be use prior to this Algo to
populate temp['selected']. This will typically be SelectAll.
Args:
* n (int): Select N elements randomly.
* include_no_data (bool): Include securities that do not have data?
* include_negative (bool): Include securities that have negative
or zero prices?
Sets:
* selected
Requires:
* selected
"""
def __init__(self, n=None, include_no_data=False, include_negative=False):
super().__init__()
self.n = n
self.include_no_data = include_no_data
self.include_negative = include_negative
def __call__(self, target):
if "selected" in target.temp:
sel = target.temp["selected"]
else:
sel = list(target.universe.columns)
if not (self.include_no_data and self.include_negative):
prices = target.universe.loc[target.now]
sel = _select_by_price(prices, sel, self.include_no_data, self.include_negative)
if self.n is not None:
n = min(len(sel), self.n)
sel = random.sample(list(sel), int(n))
target.temp["selected"] = sel
return True
[docs]
class SelectRegex(Algo):
"""
Sets temp['selected'] based on a regex on their names.
Useful when working with a large universe of different kinds of securities
Args:
* regex (str): regular expression on the name
Sets:
* selected
Requires:
* selected
"""
def __init__(self, regex):
super().__init__()
self.regex = re.compile(regex)
def __call__(self, target):
selected = target.temp["selected"]
selected = [s for s in selected if self.regex.search(s)]
target.temp["selected"] = selected
return True
[docs]
class ResolveOnTheRun(Algo):
"""
Looks at securities set in temp['selected'] and searches for names that
match the names of "aliases" for on-the-run securities in the provided
data. Then replaces the alias with the name of the underlying security
appropriate for the given date, and sets it back on temp['selected']
Args:
* on_the_run (str): Name of a Data frame with
- columns set to "on the run" ticker names
- index set to the timeline for the backtest
- values are the actual security name to use for the given date
* include_no_data (bool): Include securities that do not have data?
* include_negative (bool): Include securities that have negative
or zero prices?
Requires:
* selected
Sets:
* selected
"""
def __init__(self, on_the_run, include_no_data=False, include_negative=False):
super().__init__()
self.on_the_run = on_the_run
self.include_no_data = include_no_data
self.include_negative = include_negative
def __call__(self, target):
# Resolve real tickers based on OTR
on_the_run = target.get_data(self.on_the_run)
selected = target.temp["selected"]
aliases = [s for s in selected if s in on_the_run.columns]
if not aliases:
return True
resolved = on_the_run.loc[target.now, aliases].tolist()
prices = target.universe.loc[target.now]
resolved = _select_by_price(prices, resolved, self.include_no_data, self.include_negative)
target.temp["selected"] = resolved + [s for s in selected if s not in on_the_run.columns]
return True
[docs]
class SetStat(Algo):
"""
Sets temp['stat'] for use by downstream algos (such as SelectN).
Args:
* stat (str|DataFrame): A dataframe of the same dimension as target.universe
If a string is passed, frame is accessed using target.get_data
This is the preferred way of using the algo.
* lag (DateOffset): Lag interval. The stat used today is the one calculated
at today - lag
Sets:
* stat
"""
def __init__(self, stat, lag=_DEFAULT_LAG):
super().__init__()
self.lag = lag
if isinstance(stat, pd.DataFrame):
self.stat_name = None
self.stat = stat
else:
self.stat_name = stat
self.stat = None
def __call__(self, target):
if self.stat_name is None:
stat = self.stat
else:
stat = target.get_data(self.stat_name)
t0 = target.now - self.lag
if t0 not in stat.index:
return False
target.temp["stat"] = stat.loc[t0]
return True
[docs]
class StatTotalReturn(Algo):
"""
Sets temp['stat'] with total returns over a given period.
Sets the 'stat' based on the total return of each element in
temp['selected'] over a given lookback period. The total return
is determined by ffn's calc_total_return.
Args:
* lookback (DateOffset): lookback period.
* lag (DateOffset): Lag interval. Total return is calculated in
the inteval [now - lookback - lag, now - lag]
Sets:
* stat
Requires:
* selected
"""
def __init__(self, lookback=_DEFAULT_LOOKBACK, lag=_DEFAULT_LAG):
super().__init__()
self.lookback = lookback
self.lag = lag
def __call__(self, target):
selected = target.temp["selected"]
t0 = target.now - self.lag
if target.universe[selected].index[0] > t0:
return False
prc = target.universe.loc[t0 - self.lookback : t0, selected]
target.temp["stat"] = prc.calc_total_return()
return True
[docs]
class StatMultiPeriodReturn(Algo):
"""
Sets temp['stat'] to the weighted average of multiple total returns.
Args:
* lookbacks (sequence): DateOffset lookback periods.
* weights (sequence): Optional finite, non-negative weight for each
lookback. Their sum must be finite and positive. Equal weights are
used by default and normalized to sum to one.
* lag (DateOffset): Lag interval. Each total return ends at now - lag.
Sets:
* stat
Requires:
* selected
"""
def __init__(self, lookbacks, weights=None, lag=_DEFAULT_LAG):
super().__init__()
self.lookbacks = list(lookbacks)
if not self.lookbacks:
raise ValueError("lookbacks cannot be empty")
if weights is None:
weights = np.ones(len(self.lookbacks))
elif len(weights) != len(self.lookbacks):
raise ValueError("weights and lookbacks must have the same length")
weights = np.asarray(weights, dtype=float)
if not np.all(np.isfinite(weights)):
raise ValueError("weights and their sum must be finite")
# Check the aggregate too because finite values can overflow when summed.
with np.errstate(over="ignore"):
weight_sum = weights.sum()
if not np.isfinite(weight_sum):
raise ValueError("weights and their sum must be finite")
if np.any(weights < 0):
raise ValueError("weights must be non-negative")
if weight_sum == 0:
raise ValueError("weights must not sum to zero")
self.weights = weights / weight_sum
self.lag = lag
def __call__(self, target):
selected = target.temp["selected"]
t0 = target.now - self.lag
universe = target.universe
prices = universe[selected]
if prices.index[0] > t0:
return False
# Reuse selected columns to avoid repeating wide-universe indexing for each
# lookback. Other inputs retain observable selector, offset, and universe reads.
# Metadata-bearing columns also retain per-lookup pandas warnings.
if (
type(target) in (bt.Strategy, bt.FixedIncomeStrategy)
and type(universe) is pd.DataFrame
and not isinstance(universe.columns, (pd.MultiIndex, pd.CategoricalIndex))
and type(selected) is list
and all(type(label) is str for label in selected)
and type(self.lookbacks) is list
and all(type(lookback) is pd.DateOffset for lookback in self.lookbacks)
):
returns = [prices.loc[t0 - lookback : t0].calc_total_return() for lookback in self.lookbacks]
else:
returns = [target.universe.loc[t0 - lookback : t0, selected].calc_total_return() for lookback in self.lookbacks]
target.temp["stat"] = pd.concat(returns, axis=1).dot(self.weights)
return True
[docs]
class WeighEqually(Algo):
"""
Sets temp['weights'] by calculating equal weights for all items in
selected.
Equal weight Algo. Sets the 'weights' to 1/n for each item in 'selected'.
Sets:
* weights
Requires:
* selected
"""
def __init__(self):
super().__init__()
def __call__(self, target):
selected = target.temp["selected"]
n = len(selected)
if n == 0:
target.temp["weights"] = {}
else:
w = 1.0 / n
target.temp["weights"] = {x: w for x in selected}
return True
[docs]
class WeighSpecified(Algo):
"""
Sets temp['weights'] based on a provided dict of ticker:weights.
Sets the weights based on pre-specified targets.
Args:
* weights (dict): target weights -> ticker: weight
Sets:
* weights
"""
def __init__(self, **weights):
super().__init__()
self.weights = weights
def __call__(self, target):
# added copy to make sure these are not overwritten
target.temp["weights"] = self.weights.copy()
return True
[docs]
class ScaleWeights(Algo):
"""
Sets temp['weights'] based on a scaled version of itself.
Useful for going short, or scaling up/down when using
:class:`FixedIncomeStrategy <bt.core.FixedIncomeStrategy>`.
Args:
* scale (float): the scaling factor
Sets:
* weights
Requires:
* weights
"""
def __init__(self, scale):
super().__init__()
self.scale = scale
def __call__(self, target):
target.temp["weights"] = {k: self.scale * w for k, w in target.temp["weights"].items()}
return True
[docs]
class WeighTarget(Algo):
"""
Sets target weights based on a target weight DataFrame.
If the target weight dataFrame is of same dimension
as the target.universe, the portfolio will effectively be rebalanced on
each period. For example, if we have daily data and the target DataFrame
is of the same shape, we will have daily rebalancing.
However, if we provide a target weight dataframe that has only month end
dates, then rebalancing only occurs monthly.
Basically, if a weight is provided on a given date, the target weights are
set and the algo moves on (presumably to a Rebalance algo). If not, not
target weights are set.
Args:
* weights (str|DataFrame): DataFrame containing the target weights
If a string is passed, frame is accessed using target.get_data
This is the preferred way of using the algo.
Sets:
* weights
"""
def __init__(self, weights):
super().__init__()
if isinstance(weights, pd.DataFrame):
self.weights_name = None
self.weights = weights
else:
self.weights_name = weights
self.weights = None
def __call__(self, target):
# get current target weights
if self.weights_name is None:
weights = self.weights
else:
weights = target.get_data(self.weights_name)
if target.now in weights.index:
w = weights.loc[target.now].dropna()
# Promote the contiguous exact-integer range for fractional modifiers.
# Python-int bounds avoid rounding uint64 comparisons in NumPy.
if (
type(weights) is pd.DataFrame
and type(w) is pd.Series
and pd.api.types.is_integer_dtype(w.dtype)
and not w.empty
and -(2**53) <= int(w.min())
and int(w.max()) <= 2**53
):
w = w.astype(float)
target.temp["weights"] = w
return True
else:
return False
[docs]
class WeighInvVol(Algo):
"""
Sets temp['weights'] based on the inverse volatility Algo.
Sets the target weights based on ffn's calc_inv_vol_weights. This
is a commonly used technique for risk parity portfolios. The least
volatile elements receive the highest weight under this scheme. Weights
are proportional to the inverse of their volatility. Each asset's
volatility estimate uses its own available returns.
Args:
* lookback (DateOffset or int): volatility estimation window. Defaults
to three calendar months. A positive integer counts price rows
(N rows give at most N-1 returns), ending at the last row on or before
now minus lag. Uses all available rows if fewer than N exist.
* lag (DateOffset): offset from now to the window's end (default zero).
Sets:
* weights
Requires:
* selected
"""
def __init__(self, lookback=_DEFAULT_LOOKBACK, lag=_DEFAULT_LAG):
super().__init__()
if isinstance(lookback, (bool, np.bool_)) or (isinstance(lookback, (int, np.integer)) and lookback <= 0):
raise ValueError("lookback must be a positive integer when counting price rows")
self.lookback = lookback
self.lag = lag
def __call__(self, target):
selected = target.temp["selected"]
if len(selected) == 0:
target.temp["weights"] = {}
return True
if len(selected) == 1:
target.temp["weights"] = {selected[0]: 1.0}
return True
t0 = target.now - self.lag
if isinstance(self.lookback, (int, np.integer)):
prc = target.universe.loc[:t0].iloc[-int(self.lookback) :].loc[:, selected]
else:
prc = target.universe.loc[t0 - self.lookback : t0, selected]
returns = prc.to_returns()
# Explicit axis=0 to compute per-column std and avoid FutureWarning
# from pandas (axis=None will reduce over both axes in a future version)
# pandas 1.x reduces nullable columns to an object Series containing pd.NA.
# Normalize missing values before NumPy's isinf check.
vol = (1.0 / returns.std(axis=0, ddof=1)).fillna(np.nan).astype(float)
vol[np.isinf(vol)] = np.nan
tw = vol / vol.sum()
target.temp["weights"] = tw.dropna()
return True
[docs]
class WeighERC(Algo):
"""
Sets temp['weights'] based on equal risk contribution algorithm.
Sets the target weights based on ffn's calc_erc_weights. This
is an extension of the inverse volatility risk parity portfolio in
which the correlation of asset returns is incorporated into the
calculation of risk contribution of each asset.
The resulting portfolio is similar to a minimum variance portfolio
subject to a diversification constraint on the weights of its components
and its volatility is located between those of the minimum variance and
equally-weighted portfolios (Maillard 2008).
See:
https://en.wikipedia.org/wiki/Risk_parity
Args:
* lookback (DateOffset): lookback period for estimating covariance
* initial_weights (list): Starting asset weights [default inverse vol].
* risk_weights (list): Risk target weights [default equal weight].
* covar_method (str): method used to estimate the covariance. See ffn's
calc_erc_weights for more details. (default ledoit-wolf).
* risk_parity_method (str): Risk parity estimation method. see ffn's
calc_erc_weights for more details. (default ccd).
* maximum_iterations (int): Maximum iterations in iterative solutions
(default 100).
* tolerance (float): Tolerance level in iterative solutions (default 1E-8).
Sets:
* weights
Requires:
* selected
"""
def __init__(
self,
lookback=_DEFAULT_LOOKBACK,
initial_weights=None,
risk_weights=None,
covar_method="ledoit-wolf",
risk_parity_method="ccd",
maximum_iterations=100,
tolerance=1e-8,
lag=_DEFAULT_LAG,
):
super().__init__()
self.lookback = lookback
self.initial_weights = initial_weights
self.risk_weights = risk_weights
self.covar_method = covar_method
self.risk_parity_method = risk_parity_method
self.maximum_iterations = maximum_iterations
self.tolerance = tolerance
self.lag = lag
def __call__(self, target):
selected = target.temp["selected"]
if len(selected) == 0:
target.temp["weights"] = {}
return True
if len(selected) == 1:
target.temp["weights"] = {selected[0]: 1.0}
return True
t0 = target.now - self.lag
prc = target.universe.loc[t0 - self.lookback : t0, selected]
tw = bt.ffn.calc_erc_weights(
prc.to_returns().dropna(),
initial_weights=self.initial_weights,
risk_weights=self.risk_weights,
covar_method=self.covar_method,
risk_parity_method=self.risk_parity_method,
maximum_iterations=self.maximum_iterations,
tolerance=self.tolerance,
)
target.temp["weights"] = tw.dropna()
return True
[docs]
class WeighMeanVar(Algo):
"""
Sets temp['weights'] based on mean-variance optimization.
Sets the target weights based on ffn's calc_mean_var_weights. This is a
Python implementation of Markowitz's mean-variance optimization.
See:
http://en.wikipedia.org/wiki/Modern_portfolio_theory#The_efficient_frontier_with_no_risk-free_asset
Args:
* lookback (DateOffset): lookback period for estimating volatility
* bounds ((min, max)): tuple specifying the min and max weights for
each asset in the optimization.
* covar_method (str): method used to estimate the covariance. See ffn's
calc_mean_var_weights for more details.
* rf (float): risk-free rate used in optimization.
Sets:
* weights
Requires:
* selected
Raises:
* ValueError: If a single selected asset cannot satisfy the weight bounds.
"""
def __init__(
self,
lookback=_DEFAULT_LOOKBACK,
bounds=(0.0, 1.0),
covar_method="ledoit-wolf",
rf=0.0,
lag=_DEFAULT_LAG,
):
super().__init__()
self.lookback = lookback
self.lag = lag
self.bounds = bounds
self.covar_method = covar_method
self.rf = rf
def __call__(self, target):
selected = target.temp["selected"]
if len(selected) == 0:
target.temp["weights"] = {}
return True
if len(selected) == 1:
if self.bounds is not None:
# Match the solver's scalar bounds; a fully invested singleton must
# weigh one even when no return history is available.
lower, upper = (float(bound.item() if isinstance(bound, np.ndarray) else bound) if bound is not None else None for bound in self.bounds)
if (lower is not None and lower > 1.0) or (upper is not None and upper < 1.0):
raise ValueError("Weight bounds must include 1 for a single selected asset.")
target.temp["weights"] = {selected[0]: 1.0}
return True
t0 = target.now - self.lag
prc = target.universe.loc[t0 - self.lookback : t0, selected]
tw = bt.ffn.calc_mean_var_weights(
prc.to_returns().dropna(),
weight_bounds=self.bounds,
covar_method=self.covar_method,
rf=self.rf,
)
target.temp["weights"] = tw.dropna()
return True
[docs]
class WeighRandomly(Algo):
"""
Sets temp['weights'] based on a random weight vector.
Sets random target weights for each security in 'selected'.
This is useful for benchmarking against a strategy where we believe
the weighing algorithm is adding value.
For example, if we are testing a low-vol strategy and we want to see if
our weighing strategy is better than just weighing
securities randomly, we could use this Algo to create a random Strategy
used for random benchmarking.
This is an Algo wrapper around ffn's random_weights function.
Args:
* bounds ((low, high)): Tuple including low and high bounds for each
security
* weight_sum (float): What should the weights sum up to?
Sets:
* weights
Requires:
* selected
Raises:
* ValueError: If the bounds cannot produce the requested weight sum for
the selected securities.
"""
def __init__(self, bounds=(0.0, 1.0), weight_sum=1):
super().__init__()
self.bounds = bounds
self.weight_sum = weight_sum
def __call__(self, target):
sel = target.temp["selected"]
n = len(sel)
# No selection is a valid target state, independent of the requested sum.
if n == 0:
target.temp["weights"] = {}
return True
rw = bt.ffn.random_weights(n, self.bounds, self.weight_sum)
target.temp["weights"] = dict(zip(sel, rw))
return True
[docs]
class LimitDeltas(Algo):
"""
Modifies temp['weights'] based on weight delta limits.
Basically, this can be used if we want to restrict how much a security's
target weight can change from day to day. Useful when we want to be more
conservative about how much we could actually trade on a given day without
affecting the market.
For example, if we have a strategy that is currently long 100% one
security, and the weighing Algo sets the new weight to 0%, but we
use this Algo with a limit of 0.1, the new target weight will
be 90% instead of 0%.
For ordinary strategies reserving cash, targets and limits are relative to
the investable portion, as in Rebalance. Fixed-income strategies ignore cash.
A full cash reserve still closes positions immediately when rebalanced.
Args:
* limit (float, dict): Weight delta limit. If float, this will be a
global limit for all securities. If dict, you may specify by-ticker
limit.
Sets:
* weights
Requires:
* weights
"""
def __init__(self, limit=0.1):
super().__init__()
self.limit = limit
# determine if global or specific
self.global_limit = True
if isinstance(limit, dict):
self.global_limit = False
def __call__(self, target):
tw = target.temp["weights"]
all_keys = set(list(target.children.keys()) + list(tw.keys()))
# Use the same investable-slice coordinates as RebalanceOverTime.
scale = 1.0
if "cash" in target.temp and not target.fixed_income:
scale -= target.temp["cash"]
# Rebalance handles full cash by closing positions immediately.
if scale == 0.0:
scale = 1.0
for k in all_keys:
tgt = tw.get(k, 0.0)
cur = target.children[k].weight / scale if k in target.children else 0.0
delta = tgt - cur
# check if we need to limit
if self.global_limit:
if abs(delta) > self.limit:
tw[k] = cur + (self.limit * np.sign(delta))
else:
# make sure we have a limit defined in case of limit dict
if k in self.limit:
lmt = self.limit[k]
if abs(delta) > lmt:
tw[k] = cur + (lmt * np.sign(delta))
return True
[docs]
class LimitWeights(Algo):
"""
Modifies temp['weights'] based on weight limits.
This is an Algo wrapper around ffn's limit_weights. The purpose of this
Algo is to limit the weight of any one specifc asset. For example, some
Algos will set some rather extreme weights that may not be acceptable.
Therefore, we can use this Algo to limit the extreme weights. The excess
weight is then redistributed to the other assets, proportionally to
their current weights.
See ffn's limit_weights for more information.
Args:
* limit (float): Weight limit.
Sets:
* weights
Requires:
* weights
"""
def __init__(self, limit=0.1):
super().__init__()
self.limit = limit
def __call__(self, target):
if "weights" not in target.temp:
return True
tw = target.temp["weights"]
if len(tw) == 0:
return True
# if the limit < equal weight then set weights to 0
if self.limit < 1.0 / len(tw):
tw = {}
else:
tw = bt.ffn.limit_weights(tw, self.limit)
target.temp["weights"] = tw
return True
def _estimate_covariance(algo, returns):
"""Share pairwise and complete-case sampling policy between volatility Algos."""
# Keep branch-local selector reads observable to Algo subclasses.
if algo.covar_method == "ledoit-wolf":
covar = pd.DataFrame(
sklearn.covariance.ledoit_wolf(returns.dropna())[0],
index=returns.columns,
columns=returns.columns,
)
elif algo.covar_method == "standard":
covar = returns.cov()
else:
raise NotImplementedError("covar_method not implemented")
return covar
[docs]
class TargetVol(Algo):
"""
Updates temp['weights'] based on the target annualized volatility desired.
Args:
* target_volatility: annualized volatility to target
* lookback (DateOffset): lookback period for estimating volatility
* lag (DateOffset): amount of time to wait to calculate the covariance
* covar_method: method of calculating volatility
* annualization_factor: number of periods to annualize by.
It is assumed that target volatility is already annualized by this factor.
Updates:
* weights
Requires:
* temp['weights']
Raises:
* ValueError: if an applicable target or the estimated portfolio
volatility is non-finite, or if the estimated volatility is zero
when an applicable non-zero target is requested.
"""
def __init__(
self,
target_volatility,
lookback=_DEFAULT_LOOKBACK,
lag=_DEFAULT_LAG,
covar_method="standard",
annualization_factor=252,
):
super().__init__()
self.target_volatility = target_volatility
self.lookback = lookback
self.lag = lag
self.covar_method = covar_method
self.annualization_factor = annualization_factor
def __call__(self, target):
current_weights = target.temp["weights"]
selected = current_weights.keys()
# if there were no weights already set then skip
if len(selected) == 0:
return True
t0 = target.now - self.lag
prc = target.universe.loc[t0 - self.lookback : t0, selected]
returns = bt.ffn.to_returns(prc)
covar = _estimate_covariance(self, returns)
weights = pd.Series([current_weights[x] for x in covar.columns], index=covar.columns)
vol = np.sqrt(np.matmul(weights.values.T, np.matmul(covar.values, weights.values)) * self.annualization_factor)
# Series iteration yields values; target lookup and scaling require labels.
if isinstance(self.target_volatility, (float, int)):
target_volatility = {k: self.target_volatility for k in selected}
else:
target_volatility = self.target_volatility
# A target mapping may intentionally omit some or all current weights.
applicable_target_volatilities: list[float] = [target_volatility[k] for k in selected if k in target_volatility]
if not applicable_target_volatilities:
return True
if not np.isfinite(applicable_target_volatilities).all():
raise ValueError("Cannot target a non-finite target volatility")
if not np.isfinite(vol):
raise ValueError("Cannot target volatility from a non-finite portfolio volatility")
if vol == 0:
# Scaling cannot create volatility from zero; zero is already a valid target.
if any(value != 0 for value in applicable_target_volatilities):
raise ValueError("Cannot target a non-zero volatility from a zero-volatility portfolio")
return True
for k in selected:
if k in target_volatility:
target.temp["weights"][k] = target.temp["weights"][k] * target_volatility[k] / vol
return True
[docs]
class PTE_Rebalance(Algo):
"""
Triggers a rebalance when PTE from static weights is past a level.
Current weights sum each Security's multiplier-aware exposure by name,
including same-name holdings in different strategy sleeves.
Args:
* PTE_volatility_cap: annualized volatility to target
* target_weights: dataframe of weights that needs to have the same index as the price dataframe
* lookback (DateOffset): lookback period for estimating volatility
* lag (DateOffset): amount of time to wait to calculate the covariance
* covar_method: method of calculating volatility
* annualization_factor: number of periods to annualize by.
It is assumed that target volatility is already annualized by this factor.
"""
def __init__(
self,
PTE_volatility_cap,
target_weights,
lookback=_DEFAULT_LOOKBACK,
lag=_DEFAULT_LAG,
covar_method="standard",
annualization_factor=252,
):
super().__init__()
self.PTE_volatility_cap = PTE_volatility_cap
self.target_weights = target_weights
self.lookback = lookback
self.lag = lag
self.covar_method = covar_method
self.annualization_factor = annualization_factor
def __call__(self, target):
if target.now is None:
return False
if target.positions.shape == (0, 0):
return True
positions = target.positions.loc[target.now]
if positions is None:
return True
prices = target.universe.loc[target.now, positions.index]
if prices is None:
return True
# Same-name holdings can have different contract sizes; value each before summing.
current_values = {}
for security in target.securities:
value = security.position * prices[security.name] * security.multiplier
current_values[security.name] = current_values.get(security.name, 0.0) + value
current_weights = pd.Series(current_values) / target.value
target_weights = self.target_weights.loc[target.now, :]
cols = list(current_weights.index.copy())
for c in target_weights.keys(): # noqa: SIM118
if c not in cols:
cols.append(c)
weights = pd.Series(np.zeros(len(cols)), index=cols)
for c in cols:
if c in current_weights:
weights[c] = current_weights[c]
if c in target_weights:
weights[c] -= target_weights[c]
t0 = target.now - self.lag
prc = target.universe.loc[t0 - self.lookback : t0, cols]
returns = bt.ffn.to_returns(prc)
covar = _estimate_covariance(self, returns)
PTE_vol = np.sqrt(np.matmul(weights.values.T, np.matmul(covar.values, weights.values)) * self.annualization_factor)
if pd.isnull(PTE_vol):
return False
# vol is too high
return PTE_vol > self.PTE_volatility_cap
return True
[docs]
class CapitalFlow(Algo):
"""
Used to model capital flows. Flows can either be inflows or outflows.
This Algo can be used to model capital flows. For example, a pension
fund might have inflows every month or year due to contributions. This
Algo adjusts the capital of the target node. External flows are intended
to be excluded from investment returns, subject to the limitations below.
Since this is modeled as an adjustment, the capital will remain in the
strategy until a re-allocation/rebalancement is made.
The amount must be finite. Nonfinite amounts raise ValueError when the
Algo runs, before the target's accounting state changes.
Args:
* amount (float): Amount of adjustment
Note:
Known calculation defect: In ordinary value-based strategies, external
capital flows can incorrectly change reported investment returns.
These flows should be return-neutral. Backtest marks prices before
running Algos, and the current calculation can misstate returns when
capital is added or withdrawn after that mark.
For example, if 1000 invested grows to 1100, a subsequent contribution
of 500 leaves account value at 1600 but reports approximately 6.67%
return rather than 10%. Withdrawing the original 1000 instead raises
ZeroDivisionError even though account value remains positive at 100.
Use ``strategy.values`` to inspect account value separately; it does
not correct price-based performance statistics.
"""
def __init__(self, amount):
"""
CapitalFlow constructor.
Args:
* amount (float): Amount to adjust by
"""
super().__init__()
self.amount = float(amount)
def __call__(self, target):
target.adjust(self.amount)
return True
[docs]
class CorporateActions(Algo):
"""
Used to model dividends and share splits.
This Algo can be used to model dividend flows and share splits. This
allows the backtest to be run using unadjusted price data and to get
historically correct transactions (position sizes and commissions).
Security positions are adjusted on dates where a split value other than
1.0 is given. A value above 1.0 causes the position size to increase
(compared to the previous date) and vice versa. Split-adjusted values and
weights are available to subsequent Algos in the same iteration.
On a given date, a position on a security causes a cash inflow (when
long) or outflow (when short) given by the size of the position times
the amount in the `dividends` dataframe. Cash adjustments are made on
the `ex` date given as input, since payment date information is not
easily obtainable.
Applicable split ratios and dividends must be finite. Nonfinite values
raise ValueError before either action changes portfolio state. Missing
values retain the no-action meanings described below.
This Algo must run at every iteration to be able to change security
positions as required by splits. All dates in the `dividends`and
`splits` dataframes must exist in the price data for the calculations
to work correctly.
Args:
* dividends (dataframe): dataframe of dividend amounts per unit of
security, indexed by `ex` date. Values should be 0.0 or NaN when
there is no dividend.
* splits (dataframe): dataframe of split ratios from previous date.
Values should be 1.0 or NaN when there is no split.
"""
def __init__(self, dividends, splits):
super().__init__()
self.dividends = dividends.fillna(0.0)
self.splits = splits.fillna(1.0)
def __call__(self, target):
split_row = dividend_row = None
if target.now in self.splits.index:
split_row = self.splits.loc[target.now]
# Mixed-dtype rows can promote scalars and change position arithmetic.
if not self.splits.dtypes.eq(split_row.dtype).all():
split_row = {c: self.splits.loc[target.now, c] for c in target.children if c in self.splits.columns}
if target.now in self.dividends.index:
dividend_row = self.dividends.loc[target.now]
if not self.dividends.dtypes.eq(dividend_row.dtype).all():
dividend_row = {c: self.dividends.loc[target.now, c] for c in target.children if c in self.dividends.columns}
# Check both action sets before the first split can change portfolio state.
for c in target.children:
if split_row is not None and c in self.splits.columns and not math.isfinite(split_row[c]):
raise ValueError("Corporate action split ratio must be finite")
if dividend_row is not None and c in self.dividends.columns and not math.isfinite(dividend_row[c]):
raise ValueError("Corporate action dividend must be finite")
# A split changes positions without a transaction, so invalidate cached tree values.
if split_row is not None:
for c in target.children:
if c in self.splits.columns:
spl = split_row[c]
if spl != 1.0:
target.children[c]._position *= spl
target.root.stale = True
# adjust capital due to dividends
if dividend_row is not None:
div_inflow = 0.0
for c in target.children:
if c in self.dividends.columns:
div = dividend_row[c]
if div != 0.0:
div_inflow += div * target.children[c]._position
target.adjust(div_inflow, flow=False)
return True
[docs]
class CloseDead(Algo):
"""
Closes all positions for which prices are equal to zero (we assume
that these stocks are dead) and removes them from temp['weights'] if
they enter it by any chance.
To be called before Rebalance().
In a normal workflow it is not needed, as those securities will not
be selected by SelectAll(include_no_data=False) or similar method, and
Rebalance() closes positions that are not in temp['weights'] anyway.
However in case when for some reasons include_no_data=False could not
be used or some modified weighting method is used, CloseDead() will
allow to avoid errors.
Requires:
* weights
"""
def __init__(self):
super().__init__()
def __call__(self, target):
if "weights" not in target.temp:
return True
targets = target.temp["weights"]
for c in target.children:
if target.universe.at[target.now, c] <= 0:
target.close(c)
if c in targets:
del targets[c]
return True
[docs]
class SetNotional(Algo):
"""
Sets the notional_value to use as the base for rebalancing for
:class:`FixedIncomeStrategy <bt.core.FixedIncomeStrategy>` targets
Args:
* notional_value (str): Name of a pd.Series object containing the
target notional values of the strategy over time.
Sets:
* notional_value
"""
def __init__(self, notional_value):
self.notional_value = notional_value
super().__init__()
def __call__(self, target):
notional_value = target.get_data(self.notional_value)
if target.now in notional_value.index:
target.temp["notional_value"] = notional_value.loc[target.now]
return True
else:
return False
[docs]
class Rebalance(Algo):
"""
Rebalances capital based on temp['weights']
Rebalances capital based on temp['weights']. Also closes
positions if open but not in target_weights. This is typically
the last Algo called once the target weights have been set. Direct
security targets that require an adjustment are checked for missing or infinite
prices before mutation. Capital allocations also require a nonzero price;
quantity-based fixed-income trades and closing positions allow zero prices.
Requires:
* weights
* cash (optional): You can set a 'cash' value on temp. This should be a
number between 0-1 and determines the amount of cash to set aside.
For example, if cash=0.3, the strategy will allocate 70% of its
value among the provided weights, and the remaining 30% will be kept
in cash. If this value is not provided (default), the full value
of the strategy is allocated to securities. Fixed-income strategies
ignore this value.
* notional_value (optional): Required only for fixed_income targets. This is the base
balue of total notional that will apply to the weights.
"""
def __init__(self):
super().__init__()
@staticmethod
def _validate_target_prices(target, targets, base, target_scale):
"""Reject invalid direct target prices before portfolio mutation.
Mirror StrategyBase.rebalance's effective delta while inspecting lazy and
default targets without materializing them.
"""
for cname, target_weight in targets.items():
child = target.children.get(cname)
if child is None:
child = target._lazy_children.get(cname)
if child is not None and not isinstance(child, SecurityBase):
continue
current_weight = child._weight if cname in target.children else 0.0
scaled_weight = target_weight * target_scale
# StrategyBase.rebalance routes zero weights directly to close.
if is_zero(scaled_weight):
continue
if target.fixed_income:
delta = scaled_weight * base - current_weight * target.notional_value
else:
delta = (scaled_weight - current_weight) * base
# Zero deltas bypass allocation.
if is_zero(delta):
continue
# Idle children may still cache the bootstrap-row price; allocation
# would refresh them from the current universe row before trading.
if cname in target._universe.columns and (cname not in target.children or child.now != target.now):
price = target._universe.loc[target.now, cname]
elif cname in target.children:
price = child._price
else:
price = np.nan
quantity_trade = target.fixed_income and child is not None and child.fixed_income
if pd.isna(price) or math.isinf(price) or (is_zero(price) and not quantity_trade):
raise ValueError(f"Cannot allocate capital to {cname} because price is {price} as of {target.now}")
def __call__(self, target):
if "weights" not in target.temp:
return True
targets = target.temp["weights"]
# save value because it will change after each call to allocate
# use it as base in rebalance calls
# call it before de-allocation so that notional_value is correct
if target.fixed_income:
if "notional_value" in target.temp:
base = target.temp["notional_value"]
else:
base = target.notional_value
else:
base = target.value
# Child weights describe the investable slice; retain total value as the
# stable base so repeated cash-aware rebalances remain idempotent.
target_scale = 1.0
if "cash" in target.temp and not target.fixed_income:
target_scale -= target.temp["cash"]
# Reject direct target-price failures before holdings or registries change.
self._validate_target_prices(target, targets, base, target_scale)
# De-allocate open children that are not in targets.
for cname in target.children:
# if this child is in our targets, we don't want to close it out
if cname in targets:
continue
# get child and value
c = target.children[cname]
if target.fixed_income:
v = c.notional_value
else:
v = c.value
# Zero value can hide live ordinary positions directly or in a child strategy.
zero_value_child = not target.fixed_income and v == 0.0
if (v != 0.0 and not np.isnan(v)) or zero_value_child:
target.close(cname, update=False)
# Turn off updating while we rebalance each child
for item in targets.items():
target.rebalance(item[1] * target_scale, child=item[0], base=base, update=False)
# Now update
target.root.update(target.now)
return True
[docs]
class RebalanceOverTime(Algo):
"""
Similar to Rebalance but rebalances to target
weight over n periods.
Rebalances towards a target weight over a n periods. Splits up the weight
delta over n periods. Current children with nonzero weights omitted from the
targets are phased out over the same periods. Omitted zero-weight hedges
remain untouched. For ordinary strategies reserving cash, target weights refer
to the investable slice, as in Rebalance. A full cash reserve still closes
positions immediately.
This can be useful if we want to make more conservative rebalacing
assumptions. Some strategies can produce large swings in allocations. It
might not be reasonable to assume that this rebalancing can occur at the
end of one specific period. Therefore, this algo can be used to simulate
rebalancing over n periods.
This has typically been used in monthly strategies where we want to spread
out the rebalancing over 5 or 10 days.
Note:
This Algo will require the run_always wrapper in the above case. For
example, the RunMonthly will return True on the first day, and
RebalanceOverTime will be 'armed'. However, RunMonthly will return
False the rest days of the month. Therefore, we must specify that we
want to always run this algo.
Args:
* n (int): number of periods over which rebalancing takes place.
Requires:
* weights
* cash (optional): fraction to reserve for the current period, as in
Rebalance. Fixed-income strategies ignore this value.
* notional_value (optional): fixed-income base for interpreting weights,
as in Rebalance. Supply a positive base on each phase when phasing
towards an explicit notional amount.
"""
def __init__(self, n=10):
super().__init__()
self.n = float(n)
self._rb = Rebalance()
self._weights = None
self._days_left = None
def __call__(self, target):
# new weights specified - update rebalance data
if "weights" in target.temp:
self._weights = target.temp["weights"]
self._days_left = self.n
# if _weights are not None, we have some work to do
if self._weights is not None:
# Normalize NAV weights to the investable-slice units Rebalance expects.
scale = 1.0
if "cash" in target.temp and not target.fixed_income:
scale -= target.temp["cash"]
# Full cash delegates to Rebalance's immediate close without division.
if scale == 0.0:
scale = 1.0
# FI current weights use today's total, not the requested notional base.
# A NumPy float64 numerator prevents low-precision bases from downcasting it.
notional_base = target.temp.get("notional_value", 0.0) if target.fixed_income else 0.0
tgt = {}
# Read mapping entries by key and preserve their supplied order.
for cname, target_weight in self._weights.items():
curr = 0.0
if cname in target.children:
child = target.children[cname]
curr = np.float64(child.weight * target.notional_value) / notional_base if notional_base > 0 else child.weight / scale
dlt = (target_weight - curr) / self._days_left
tgt[cname] = curr + dlt
# Rebalance would otherwise close omitted holdings immediately.
for cname in target.children:
if cname not in tgt:
child = target.children[cname]
curr = np.float64(child.weight * target.notional_value) / notional_base if notional_base > 0 else child.weight / scale
if curr == 0.0:
continue
dlt = -curr / self._days_left
tgt[cname] = curr + dlt
# mock weights and call real Rebalance
target.temp["weights"] = tgt
self._rb(target)
# dec _days_left. If 0, set to None & set _weights to None
self._days_left -= 1
if self._days_left == 0:
self._days_left = None
self._weights = None
return True
[docs]
class Require(Algo):
"""
Flow control Algo.
This algo returns the value of a predicate
on an temp entry. Useful for controlling
flow.
For example, we might want to make sure we have some items selected.
We could pass a lambda function that checks the len of 'selected':
pred=lambda x: len(x) == 0
item='selected'
Args:
* pred (Algo): Function that returns a Bool given the strategy. This
is the definition of an Algo. However, this is typically used
with a simple lambda function.
* item (str): An item within temp.
* if_none (bool): Result if the item required is not in temp or if it's
value if None
"""
def __init__(self, pred, item, if_none=False):
super().__init__()
self.item = item
self.pred = pred
self.if_none = if_none
def __call__(self, target):
if self.item not in target.temp:
return self.if_none
item = target.temp[self.item]
if item is None:
return self.if_none
return self.pred(item)
[docs]
class Not(Algo):
"""
Flow control Algo
It is usful for "inverting" other flow control algos,
For example Not( RunAfterDate(...) ), Not( RunAfterDays(...) ), etc
Args:
* list_of_algos (Algo): The algo to run and invert the return value of
"""
def __init__(self, algo):
super().__init__()
self._algo = algo
def __call__(self, target):
return not self._algo(target)
[docs]
class Or(Algo):
"""
Flow control Algo
It useful for combining multiple signals into one signal.
For example, we might want two different rebalance signals to work together:
runOnDateAlgo = bt.algos.RunOnDate(pdf.index[0]) # where pdf.index[0] is the first date in our time series
runMonthlyAlgo = bt.algos.RunMonthly()
orAlgo = Or([runMonthlyAlgo,runOnDateAlgo])
orAlgo will return True if it is the first date or if it is 1st of the month
Args:
* list_of_algos: Iterable list of algos.
Runs each algo and
returns true if any algo returns true.
"""
def __init__(self, list_of_algos):
super().__init__()
self._list_of_algos = list_of_algos
def __call__(self, target):
res = False
for algo in self._list_of_algos:
tempRes = algo(target)
res = res | tempRes
return res
[docs]
class SelectTypes(Algo):
"""
Sets temp['selected'] based on node type.
If temp['selected'] is already set, it will filter the existing
selection.
Args:
* include_types (list): Types of nodes to include
* exclude_types (list): Types of nodes to exclude
Sets:
* selected
"""
def __init__(self, include_types=(bt.core.Node,), exclude_types=()):
super().__init__()
self.include_types = include_types
self.exclude_types = exclude_types or (type(None),)
def __call__(self, target):
selected = [sec_name for sec_name, sec in target.children.items() if isinstance(sec, self.include_types) and not isinstance(sec, self.exclude_types)]
if selected and "selected" in target.temp:
prior = target.temp["selected"]
# A local set avoids rescanning a long prior list for every child in
# wide universes, while filtering still preserves child order.
# Keep short or sparse selections on the list path; set construction
# costs more than a few membership checks. Only plain strings are interchangeable.
if (
type(prior) is list
and len(selected) >= 32
and 32 <= len(prior) <= 4 * len(selected)
and all(type(s) is str for s in prior)
and all(type(s) is str for s in selected)
):
prior = set(prior)
selected = [s for s in selected if s in prior]
else:
# Preserve per-child reads for caller-defined membership.
selected = [s for s in selected if s in target.temp["selected"]]
target.temp["selected"] = selected
return True
[docs]
class ClosePositionsAfterDates(Algo):
"""
Close positions on securities after a given date.
This can be used to make sure positions on matured/redeemed securities are
closed. It can also be used as part of a strategy to, i.e. make sure
the strategy doesn't hold any securities with time to maturity less than a year
Note that if placed after a RunPeriod algo in the stack, that the actual
closing of positions will occur after the provided date. For this to work,
the "price" of the security (even if matured) must exist up until that date.
Alternatively, run this with the @run_always decorator to close the positions
immediately.
Also note that this algo does not operate using temp['weights'] and Rebalance.
This is so that hedges (which are excluded from that workflow) will also be
closed as necessary.
Args:
* close_dates (str): the name of a dataframe indexed by security name, with columns
"date": the date after which we want to close the position ASAP
Sets:
* target.perm['closed'] : to keep track of which securities have already closed
"""
def __init__(self, close_dates):
super().__init__()
self.close_dates = close_dates
def __call__(self, target):
if "closed" not in target.perm:
target.perm["closed"] = set()
close_dates = target.get_data(self.close_dates)["date"]
# Find securities that are candidate for closing
sec_names = [
sec_name for sec_name, sec in target.children.items() if isinstance(sec, SecurityBase) and sec_name in close_dates.index and sec_name not in target.perm["closed"]
]
# Check whether closed
is_closed = close_dates.loc[sec_names] <= target.now
# Close position
for sec_name in is_closed[is_closed].index:
target.close(sec_name, update=False)
target.perm["closed"].add(sec_name)
# Now update
target.root.update(target.now)
return True
[docs]
class RollPositionsAfterDates(Algo):
"""
Roll securities based on the provided map.
This can be used for any securities which have "On-The-Run" and "Off-The-Run"
versions (treasury bonds, index swaps, etc).
Also note that this algo does not operate using temp['weights'] and Rebalance.
This is so that hedges (which are excluded from that workflow) will also be
rolled as necessary.
Args:
* roll_data (str): the name of a dataframe indexed by security name, with columns
- "date": the first date at which the roll can occur
- "target": the security name we are rolling into
- "factor": the conversion factor. One unit of the original security
rolls into "factor" units of the new one. Applicable factors and
resulting quantities must be finite and are validated before
portfolio state changes.
Known execution limitation: roll execution is not atomic. An execution
error can leave earlier trades and ``target.perm['rolled']`` entries
applied; completed changes are not rolled back.
If a source close fails with its nonzero position unchanged, the marker
newly added for that close is removed. Other effects are not undone,
and a partially executed roll is not guaranteed to be safe to retry.
Sets:
* target.perm['rolled'] : to keep track of which securities have already rolled
"""
def __init__(self, roll_data):
super().__init__()
self.roll_data = roll_data
def __call__(self, target):
roll_data = target.get_data(self.roll_data)
rolled = target.perm.get("rolled", set())
transactions = {}
# Find securities that are candidate for roll
sec_names = [sec_name for sec_name, sec in target.children.items() if isinstance(sec, SecurityBase) and sec_name in roll_data.index and sec_name not in rolled]
# Build and validate the complete due transaction plan before mutation.
due_rolls = roll_data.loc[sec_names]
due_rolls = due_rolls[due_rolls["date"] <= target.now]
for sec_name, sec_fields in due_rolls.iterrows():
factor = sec_fields["factor"]
factor_array = np.asarray(factor)
if factor_array.ndim != 0 or factor_array.dtype.kind not in "biuf" or not np.isfinite(factor_array).item():
raise ValueError("Roll factors and resulting quantities must be finite")
with np.errstate(over="raise", invalid="raise"):
try:
new_quantity = float(factor * target[sec_name].position)
new_sec = sec_fields["target"]
if new_sec in transactions:
new_quantity += transactions[new_sec]
except FloatingPointError as exc:
raise ValueError("Roll factors and resulting quantities must be finite") from exc
if not np.isfinite(new_quantity):
raise ValueError("Roll factors and resulting quantities must be finite")
transactions[new_sec] = new_quantity
# Due sources are closed before transfers; other destinations retain their positions.
for new_sec, quantity in transactions.items():
child = target.children.get(new_sec)
if isinstance(child, SecurityBase) and new_sec not in due_rolls.index and not np.isfinite(float(child.position) + quantity):
raise ValueError("Roll factors and resulting quantities must be finite")
if "rolled" not in target.perm:
target.perm["rolled"] = set()
# Apply the validated plan only after every due row has succeeded.
for sec_name in due_rolls.index:
security = target[sec_name]
position = security.position
rolled = target.perm["rolled"]
already_rolled = sec_name in rolled
# Keep the marker visible to callbacks that invoke this Algo again.
rolled.add(sec_name)
try:
target.close(sec_name, update=False)
except Exception:
# Do not clear a completed recursive roll or infer rollback after a trade.
if not already_rolled and position != 0 and security.position == position:
rolled.discard(sec_name)
raise
# Do all the new transactions at the end, to do any necessary aggregations first
for new_sec, quantity in transactions.items():
target.transact(quantity, new_sec, update=False)
# Now update
target.root.update(target.now)
return True
[docs]
class SelectActive(Algo):
"""
Sets temp['selected'] based on filtering temp['selected'] to exclude
those securities that have been closed or rolled after a certain date
using ClosePositionsAfterDates or RollPositionsAfterDates. This makes sure
not to select them again for weighting (even if they have prices).
Requires:
* selected
* perm['closed'] or perm['rolled']
Sets:
* selected
"""
def __call__(self, target):
selected = target.temp["selected"]
rolled = target.perm.get("rolled", set())
closed = target.perm.get("closed", set())
# Avoid rebuilding large exclusion sets for each ordinary selected label.
# Keep short selections on the original path: registry label scans cost more
# than a few unions. Only inspect label types when that preparation can pay off.
# Other containers or label hooks retain per-label evaluation, including no-work calls.
if (
type(selected) is list
and len(selected) >= 32
and type(rolled) is set
and type(closed) is set
and all(type(s) is str for s in selected)
and all(type(s) is str for s in rolled)
and all(type(s) is str for s in closed)
):
excluded = set.union(rolled, closed)
selected = [s for s in selected if s not in excluded]
else:
selected = [s for s in selected if s not in set.union(rolled, closed)]
target.temp["selected"] = selected
return True
def _validate_transaction_security(target, security):
"""Validate a declared or market-backed transaction target without creating it."""
if security in target.children:
child = target.children[security]
elif security in target._lazy_children:
child = target._lazy_children[security]
elif security in target.universe.columns:
return
else:
raise KeyError(security)
if not isinstance(child, SecurityBase):
raise TypeError(f"Transaction target {security!r} is not a security")
def _transaction_quantity_is_noop(quantity):
"""Validate a transaction quantity and report whether dispatch is a no-op."""
if np.iscomplexobj(quantity):
raise ValueError("Transaction quantity must be real")
try:
if math.isnan(quantity):
return True
if not math.isfinite(quantity):
raise ValueError("Transaction quantity must be finite")
return is_zero(quantity)
except (TypeError, OverflowError) as exc:
raise ValueError("Transaction quantity must be numeric") from exc
def _validate_transaction_price(price):
"""Validate a custom transaction price used by an actionable quantity."""
if np.iscomplexobj(price):
raise ValueError("Transaction price must be real")
try:
finite = math.isfinite(price)
except (TypeError, ValueError, OverflowError) as exc:
raise ValueError("Transaction price must be numeric") from exc
if not finite:
raise ValueError("Transaction price must be finite")
def _preflight_transactions(target, transactions):
"""Validate a complete transaction batch before dispatch can mutate state."""
prepared = []
for (_, security), transaction in transactions.iterrows():
_validate_transaction_security(target, security)
quantity = transaction["quantity"]
price = transaction["price"]
if not _transaction_quantity_is_noop(quantity) and price is not None:
_validate_transaction_price(price)
price = float(price)
prepared.append((security, float(quantity), price))
return prepared
def _resolve_transaction_security(target, security):
"""Resolve a preflighted security for transaction dispatch."""
_validate_transaction_security(target, security)
target._create_child_if_needed(security)
return target[security]
[docs]
class ReplayTransactions(Algo):
"""
Replay a list of transactions that were executed.
This is useful for taking a blotter of actual trades that occurred,
and measuring performance against hypothetical strategies.
In particular, one can replay the outputs of backtest.Result.get_transactions.
Securities may be existing children, lazily declared children, or available
columns in the target's market-data universe.
The selected batch is validated before dispatch so an invalid target,
actionable quantity, or required custom price cannot partially replay it.
Note that this allows the timestamps and prices of the reported transactions
to be completely arbitrary, so while the strategy may track performance
on a daily basis, it will accurately account for the actual PNL of
the trades based on where they actually traded, and the bidofferpaid
attribute on the strategy will capture the "slippage" as measured
against the daily prices.
Args:
* transactions (str): name of a MultiIndex dataframe with format
Date, Security | quantity, price.
Note this schema follows the output of backtest.Result.get_transactions
"""
def __init__(self, transactions):
super().__init__()
self.transactions = transactions
def __call__(self, target):
timeline = target.data.index
index = timeline.get_loc(target.now)
end = target.now
if index == 0:
# UTC makes the minimum comparable to aware clocks without underflowing
# when their local timezone has a positive offset.
start = pd.Timestamp.min if getattr(end, "tzinfo", None) is None else pd.Timestamp.min.tz_localize("UTC")
else:
start = timeline[index - 1]
# Get the transactions since the last update
all_transactions = target.get_data(self.transactions)
timestamps = all_transactions.index.get_level_values("Date")
transactions = all_transactions[(timestamps > start) & (timestamps <= end)]
for security, quantity, price in _preflight_transactions(target, transactions):
c = _resolve_transaction_security(target, security)
c.transact(quantity, price=price, update=False)
# Now update
target.root.update(target.now)
return True
[docs]
class SimulateRFQTransactions(Algo):
"""
An algo that simulates the outcomes from RFQs (Request for Quote)
using a "model" that determines which ones becomes transactions and at what price
those transactions happen. This can be used from the perspective of the sender of the
RFQ or the receiver.
Transaction securities may be existing children, lazily declared children, or
available columns in the target's market-data universe.
Returned batches are validated before dispatch so an invalid target,
actionable quantity, or required custom price cannot partially execute them.
Args:
* rfqs (str): name of a dataframe with columns
Date, Security | quantity, and additional columns as required by model
* model (object): a function/callable object with arguments
- rfqs : data frame of rfqs to respond to
- target : the strategy object, for access to position and value data
and which returns a set of transactions, a MultiIndex DataFrame with:
Date, Security | quantity, price
"""
def __init__(self, rfqs, model):
super().__init__()
self.rfqs = rfqs
self.model = model
def __call__(self, target):
timeline = target.data.index
index = timeline.get_loc(target.now)
end = target.now
if index == 0:
# UTC makes the minimum comparable to aware clocks without underflowing
# when their local timezone has a positive offset.
start = pd.Timestamp.min if getattr(end, "tzinfo", None) is None else pd.Timestamp.min.tz_localize("UTC")
else:
start = timeline[index - 1]
# Get the RFQs since the last update
all_rfqs = target.get_data(self.rfqs)
timestamps = all_rfqs.index.get_level_values("Date")
rfqs = all_rfqs[(timestamps > start) & (timestamps <= end)]
# Turn the RFQs into transactions
transactions = self.model(rfqs, target)
for security, quantity, price in _preflight_transactions(target, transactions):
c = _resolve_transaction_security(target, security)
c.transact(quantity, price=price, update=False)
# Now update
target.root.update(target.now)
return True
def _get_unit_risk(security, data, index=None):
try:
unit_risks = data[security]
unit_risk = unit_risks.iloc[index]
except Exception: # noqa: BLE001
# No risk data, assume zero
unit_risk = 0.0
return unit_risk
[docs]
class UpdateRisk(Algo):
"""
Tracks a risk measure on all nodes of the strategy. To use this node, the
``additional_data`` argument on :class:`Backtest <bt.backtest.Backtest>` must
have a "unit_risk" key. The value should be a dictionary, keyed
by risk measure, of DataFrames with a column per security that is sensitive to that measure.
Infinite unit risk for positions not treated as zero raises ValueError; exposure arithmetic overflow
raises ValueError or FloatingPointError before current risk or history is changed. NaN risk
still propagates for downstream handling, and missing security columns still contribute zero.
Args:
* name (str): the name of the risk measure (IR01, PVBP, IsIndustials, etc).
The name must coincide with the keys of the dictionary passed to additional_data as the
"unit_risk" argument.
* history (int): The level of depth in the tree at which to track the time series of risk numbers.
i.e. 0=no tracking, 1=first level only, etc. More levels is more expensive.
Modifies:
* The "risk" attribute on the target and all its children
* If history==True, the "risks" attribute on the target and all its children
"""
def __init__(self, measure, history=0):
super().__init__(name=f"UpdateRisk>{measure}")
self.measure = measure
self.history = history
def _setup_risk(self, target, set_history):
"""Setup risk attributes on the node in question"""
target.risk = {}
if set_history:
target.risks = pd.DataFrame(index=target.data.index)
def _setup_measure(self, target, set_history):
"""Setup a risk measure within the risk attributes on the node in question"""
target.risk[self.measure] = np.nan
if set_history:
target.risks[self.measure] = np.nan
def _calculate_risk(self, target, depth, unit_risk_frame, risks):
if isinstance(target, bt.core.SecurityBase):
# Use target.root.now as non-traded securities may not have been updated yet
# and there is no need to update them here as we only use position
index = unit_risk_frame.index.get_loc(target.root.now)
unit_risk = _get_unit_risk(target.name, unit_risk_frame, index)
if is_zero(target.position):
risk = 0.0
else:
if not pd.isna(unit_risk) and math.isinf(unit_risk):
raise ValueError(f"Infinite unit risk for {target.name} in {self.measure}")
risk = unit_risk * target.position * target.multiplier
else:
risk = 0.0
for child in target.children.values():
risk += self._calculate_risk(child, depth + 1, unit_risk_frame, risks)
if not pd.isna(risk) and math.isinf(risk):
raise ValueError(f"Infinite risk for {target.name} in {self.measure}")
if not pd.isna(risk) and math.isinf(risk):
raise ValueError(f"Infinite risk for {target.name} in {self.measure}")
risks.append((target, depth, risk))
return risk
def __call__(self, target):
unit_risk_frame = target.get_data("unit_risk")[self.measure]
risks = []
# Reject bad leaves and totals before initializing or publishing any risk state.
with np.errstate(over="raise"):
self._calculate_risk(target, 0, unit_risk_frame, risks)
for node, depth, risk in risks:
set_history = depth < self.history
if not hasattr(node, "risk"):
self._setup_risk(node, set_history)
# Different measures may track history at different tree depths.
if set_history and not hasattr(node, "risks"):
node.risks = pd.DataFrame(index=node.data.index)
if self.measure not in node.risk:
self._setup_measure(node, set_history)
node.risk[self.measure] = risk
if set_history:
# Idle securities can retain an older clock; risk is calculated at root.now.
# Scalar writes avoid general selection for each tracked node; other
# history representations retain their original indexer semantics.
history = node.risks
if type(history) is pd.DataFrame and isinstance(self.measure, str):
history.at[node.root.now, self.measure] = risk
else:
history.loc[node.root.now, self.measure] = risk
return True
[docs]
class PrintRisk(Algo):
"""
This Algo prints the risk data.
Args:
* fmt_string (str): A string that will later be formatted with the
target object's risk attributes. Therefore, you should provide
what you want to examine within curly braces ( { } )
If not provided, will print the entire dictionary with no formatting.
"""
def __init__(self, fmt_string=""):
super().__init__()
self.fmt_string = fmt_string
def __call__(self, target):
if hasattr(target, "risk"):
if self.fmt_string:
print(self.fmt_string.format(**target.risk))
else:
print(target.risk)
return True
[docs]
class HedgeRisks(Algo):
"""
Hedges risk measures with selected instruments.
Unit risk is scaled by each selected security's multiplier to obtain its
per-contract sensitivity. Infinite hedge notionals are rejected before any
hedge transaction. NaN hedge notionals are also rejected before trading when
``throw_nan`` is True.
Make sure that the UpdateRisk algo has been called beforehand.
Args:
* measures (list): the names of the risk measures to hedge
* pseudo (bool): whether to use the pseudo-inverse to compute
the inverse Jacobian. If False, will fail if the number
of selected instruments is not equal to the number of
measures, or if the Jacobian is singular
* strategy (StrategyBase): If provided, will hedge the risk
from this strategy in addition to the risk from target.
This is to allow separate tracking of hedged and unhedged
performance. Note that risk_strategy must occur earlier than
'target' in a depth-first traversal of the children of the root,
otherwise hedging will occur before positions of risk_strategy are
updated.
* throw_nan (bool): Whether to throw on nan hedge notionals, rather
than simply not hedging.
Requires:
* selected
"""
def __init__(self, measures, pseudo=False, strategy=None, throw_nan=True):
super().__init__()
if len(measures) == 0:
raise ValueError("Must pass in at least one measure to hedge")
self.measures = measures
self.pseudo = pseudo
self.strategy = strategy
self.throw_nan = throw_nan
def _get_target_risk(self, target, measure):
if not hasattr(target, "risk"):
raise ValueError(f"risk not set up on target {target.name}")
if measure not in target.risk:
raise ValueError(f"measure {measure} not set on target {target.name}")
return target.risk[measure]
def __call__(self, target):
securities = target.temp["selected"]
# Get target risk
target_risk = np.array([self._get_target_risk(target, m) for m in self.measures])
if self.strategy is not None:
# Add the target risk of the strategy to the risk of the target
# (which contains existing hedges)
target_risk += np.array([self._get_target_risk(self.strategy, m) for m in self.measures])
# Turn target_risk into a column array
target_risk = target_risk.reshape(len(self.measures), 1)
# Get hedge risk as a Jacobian matrix
data = []
for m in self.measures:
d = target.get_data("unit_risk").get(m)
if d is None:
raise ValueError(f"unit_risk for {m} not present in temp on {target.name}")
i = d.index.get_loc(target.now)
data.append((i, d))
hedge_risk = np.array([[_get_unit_risk(s, d, i) for (i, d) in data] for s in securities])
hedge_risk = hedge_risk.astype(float, copy=False)
for row in range(hedge_risk.shape[0]):
security = securities[row]
child = target.children.get(security)
if child is None:
child = target._lazy_children.get(security)
# Undeclared names become multiplier-one Securities when transacted.
if isinstance(child, bt.core.SecurityBase):
hedge_risk[row] *= child.multiplier
# Get hedge ratios
if self.pseudo:
inv = np.linalg.pinv(hedge_risk).T
else:
inv = np.linalg.inv(hedge_risk).T
notionals = np.matmul(inv, -target_risk).flatten()
# Validate the whole plan before dispatch: a later invalid quantity must
# not leave earlier fills or consume a lazily declared hedge security.
for notional, security in zip(notionals, securities):
if np.isinf(notional):
raise ValueError(f"{security} has infinite hedge notional")
if np.isnan(notional) and self.throw_nan:
raise ValueError(f"{security} has nan hedge notional")
# Preserve native NaN dispatch when throw_nan is disabled.
for notional, security in zip(notionals, securities):
target.transact(notional, security)
return True
[docs]
class Margin(Algo):
"""
Model margin lending by periodically charging interest and liquidating a leveraged long
portfolio when it falls below a specified maintenance requirement.
For a portfolio with positive equity, liquidation targets the maintenance requirement when
no transaction costs apply. Integer-position constraints may liquidate beyond that boundary.
Args:
* rate (float): the margin interest rate. i.e: 0.05 -> 5%
* requirement (float): the maintenance requirement. The algorithm will liquidate the portfolio if the
equity falls below this percentage.
"""
def __init__(self, rate, requirement):
super().__init__(Margin.__name__)
self.rate = rate
self.requirement = requirement
self._last_date = None
def _daily_rate(self):
return math.pow(1 + self.rate, 1 / 365.25) - 1
def __call__(self, target):
if self._last_date is None:
self._last_date = target.now
return True
# how many days since we calculated interest
diff = target.now - self._last_date
# the margin amount is the difference
margin = -target.capital
if margin > 0:
# get the total value of this portfolio
port_val = 0
for child in target.children.values():
port_val += child.value
# calculate the interest on the margin
f = math.pow(1 + self._daily_rate(), diff.total_seconds() / 86400) - 1
fee = margin * f
# charge it
target.adjust(-fee, flow=False, fee=fee)
equity_ratio = target.value / port_val
# check and see if we are below our margin requirement
if equity_ratio < self.requirement:
max_value = target.value * (1 / self.requirement)
# Strategy.allocate applies the amount through child weights, which sum to port_val / equity.
deficit = max_value - port_val
target.allocate(deficit * equity_ratio)
# update our date
self._last_date = target.now
return True
[docs]
class TaxLiabilities(Algo):
"""
Model capital gains taxes by charging the portfolio for gains realized since the last run.
Each run reads net position changes per bar since the previous run and matches sales against
each security node's open tax lots first-in, first-out. A lot held for longer than ``holding_period`` produces a
long-term gain or loss. Otherwise, the result is short-term. Gains from covering short
positions are always short-term.
This is a simplified calendar-year model. A net loss in one bucket first offsets a net gain in the other. Each
remaining net gain is taxed at its bucket's rate. A remaining net loss is deducted from
ordinary income, up to ``loss_deduction_limit`` per calendar year, and earns a credit at
``short_term_rate``. Short-term losses are deducted first. Losses beyond the limit carry
forward, keep their bucket, and offset later gains. Each run reconciles the year's cumulative
liability, charging or refunding its change. Trades are assigned to their actual calendar
years, including when a run spans multiple years. Carryforwards assume sufficient ordinary
income to use the annual deduction. Wash sales and commissions are not modeled.
Net position histories cannot recover individual executions or intrabar round trips.
Use split-adjusted prices; tax-basis adjustments for ``CorporateActions`` splits are not modeled.
The tax is charged as a fee, so it reduces performance and appears in ``target.fees``. If the portfolio
loses money in a period a credit is paid to the portfolio and appears as a negative fee.
Place this Algo after any Algos that trade on the same bar (for example, after
``Rebalance``). Trades made on a bar after this Algo has run are not taxed.
Tax charges reduce cash while holdings stay in place. After a full allocation,
``target.capital`` can become negative, increasing investment exposure relative to
``target.value`` until cash is replenished. This Algo does not charge financing costs
on that negative balance.
With fractional positions and no execution costs, a full value-based ``Rebalance``
whose weights sum to 1 restores cash to zero before the next tax charge by sizing
holdings to the reduced ``target.value``. Hold a sufficient cash buffer if negative
cash is undesirable, for example by using weights that sum to less than 1.
A credit, such as a deduction of carried-forward losses on a year's first run, is added
to cash and becomes available for subsequent trading Algos to invest.
Args:
* short_term_rate (float): tax rate on short-term gains. i.e: 0.35 -> 35%
* long_term_rate (float): tax rate on long-term gains. i.e: 0.15 -> 15%
* loss_deduction_limit (float): the most net loss per calendar year that can be
deducted from ordinary income. Defaults to 3000. None removes the
limit, and 0 means losses only offset gains.
* holding_period (DateOffset): how long a lot must be held before its gain is
long-term. Defaults to one year.
Attributes:
* liabilities (DataFrame): realized gains and tax charged on each run, indexed by date.
``deduction`` is the change in ordinary-income deduction; it can be negative when later
gains reverse an earlier deduction. Carryforwards are the current year-to-date estimates.
Sets:
* target.perm['tax_liabilities'] : the same DataFrame as ``liabilities``. Backtests
copy their strategy, so read it from ``backtest.strategy.perm`` after a run.
"""
def __init__(self, short_term_rate, long_term_rate, loss_deduction_limit=3000.0, holding_period=None):
super().__init__(TaxLiabilities.__name__)
for name, rate in (("short_term_rate", short_term_rate), ("long_term_rate", long_term_rate)):
if not math.isfinite(rate) or rate < 0:
raise ValueError(f"{name} must be a finite, non-negative number")
if loss_deduction_limit is not None and (math.isnan(loss_deduction_limit) or loss_deduction_limit < 0):
raise ValueError("loss_deduction_limit must be None or a non-negative number")
self.short_term_rate = float(short_term_rate)
self.long_term_rate = float(long_term_rate)
self.loss_deduction_limit = math.inf if loss_deduction_limit is None else float(loss_deduction_limit)
self.holding_period = pd.DateOffset(years=1) if holding_period is None else holding_period
# unused losses carried forward, stored as non-negative amounts
self._short_term_carryforward = 0.0
self._long_term_carryforward = 0.0
# calendar year -> loss already deducted from ordinary income
self._deducted = {}
# security node -> list of open lots, oldest first: [date, signed quantity, price]
self._lots = {}
self._last_date = None
self._records = []
self._tax_year = None
self._year_gains = [0.0, 0.0]
self._year_opening_losses = (0.0, 0.0)
self._year_tax = 0.0
self._year_deduction = 0.0
@property
def liabilities(self):
columns = ["short_term_gain", "long_term_gain", "deduction", "tax", "short_term_carryforward", "long_term_carryforward"]
if not self._records:
return pd.DataFrame(columns=columns, index=pd.DatetimeIndex([], name="Date"))
res = pd.DataFrame(self._records).set_index("Date")
return res[columns]
def _process_trade(self, security, date, quantity, price, multiplier):
"""Apply one trade to the open lots and return its realized (short_term, long_term) gain."""
lots = self._lots.setdefault(security, [])
short_term = 0.0
long_term = 0.0
remaining = quantity
# a trade in the opposite direction of the open lots closes them, oldest first
while lots and not is_zero(remaining) and np.sign(lots[0][1]) != np.sign(remaining):
lot_date, lot_qty, lot_price = lots[0]
closed = min(abs(remaining), abs(lot_qty))
direction = np.sign(lot_qty)
gain = closed * direction * (price - lot_price) * multiplier
if direction > 0 and date > lot_date + self.holding_period:
long_term += gain
else:
short_term += gain
lots[0][1] -= direction * closed
remaining += direction * closed
if is_zero(lots[0][1]):
lots.pop(0)
# whatever is left opens a new lot
if not is_zero(remaining):
lots.append([date, remaining, price])
return short_term, long_term
def _calculate_tax(self, short_term_gains, long_term_gains):
"""
Net realized gains and return ``(tax, net_short_term, net_long_term)``.
1. Carryover: losses that could not be deducted in earlier periods are subtracted
from gains in the same bucket (short-term from short-term, long-term from
long-term). See https://www.irs.gov/publications/p550
2. Offset: if one bucket is a net loss and the other a net gain, the loss reduces
the gain, but never past zero. For example, a -100 short-term and a +300
long-term become 0 and +200.
3. Each remaining positive bucket is taxed at its own rate.
Negative net amounts are returned so ``_deduct_losses`` can deduct or carry them
forward. This method does not change any state.
"""
# apply losses carried forward, then net the two buckets against each other
st = short_term_gains - self._short_term_carryforward
lt = long_term_gains - self._long_term_carryforward
if st < 0 < lt:
offset = min(-st, lt)
st += offset
lt -= offset
elif lt < 0 < st:
offset = min(-lt, st)
lt += offset
st -= offset
tax = max(st, 0.0) * self.short_term_rate + max(lt, 0.0) * self.long_term_rate
return tax, st, lt
def _deduct_losses(self, target, tax, st, lt, year=None):
"""
Deduct net losses from ordinary income and return ``(tax, deduction)``.
``st`` and ``lt`` are the net amounts from ``_calculate_tax``. A negative bucket is
a loss that can be deducted from ordinary income, up to ``loss_deduction_limit``
per calendar year ($3,000 in the US). Short-term losses use the allowance first.
Each deducted dollar earns a credit at ``short_term_rate``, so ``tax`` goes down
and can become negative.
Any loss above the allowance carries over to later periods and keeps its bucket.
This method replaces the stored carryforward rather than adding to it, because
``_calculate_tax`` already used up the previous carryforward. See
https://www.irs.gov/publications/p550
"""
if year is None:
year = target.now.year
allowance = max(self.loss_deduction_limit - self._deducted.get(year, 0.0), 0.0)
st_loss = max(-st, 0.0)
lt_loss = max(-lt, 0.0)
st_deduction = min(st_loss, allowance)
lt_deduction = min(lt_loss, allowance - st_deduction)
deduction = st_deduction + lt_deduction
self._deducted[year] = self._deducted.get(year, 0.0) + deduction
tax -= deduction * self.short_term_rate
self._short_term_carryforward = st_loss - st_deduction
self._long_term_carryforward = lt_loss - lt_deduction
return tax, deduction
def _settle_year(self, target, year, short_term_gains, long_term_gains):
"""Recompute a year's liability and return changes since its last settlement."""
if self._tax_year != year:
self._tax_year = year
self._year_gains = [0.0, 0.0]
self._year_opening_losses = (self._short_term_carryforward, self._long_term_carryforward)
self._year_tax = self._year_deduction = 0.0
self._year_gains[0] += short_term_gains
self._year_gains[1] += long_term_gains
self._short_term_carryforward, self._long_term_carryforward = self._year_opening_losses
self._deducted[year] = 0.0
tax, st, lt = self._calculate_tax(*self._year_gains)
tax, deduction = self._deduct_losses(target, tax, st, lt, year=year)
tax_change = tax - self._year_tax
deduction_change = deduction - self._year_deduction
self._year_tax, self._year_deduction = tax, deduction
return tax_change, deduction_change
def __call__(self, target):
if target.root.stale:
target.root.update(target.root.now)
trades = []
for security in target.securities:
positions = security.positions
prices = security.prices
if self._last_date is not None:
# Retain the previous bar to compute the first unprocessed position change.
positions = positions.loc[self._last_date :]
prices = prices.loc[self._last_date :]
if positions.empty:
continue
quantities = positions.diff()
if self._last_date is None:
quantities.iloc[0] = positions.iloc[0]
changes = pd.DataFrame({"quantity": quantities, "price": prices})
changes = changes[changes["quantity"].notna() & changes["quantity"].ne(0)]
if security._bidoffer_set and security.multiplier != 0:
changes["price"] += security.bidoffers_paid.reindex(changes.index) / (changes["quantity"] * security.multiplier)
for date, row in changes.iterrows():
trades.append((date, security, row["quantity"], row["price"]))
gains_by_year = {}
for date, security, quantity, price in sorted(trades, key=lambda trade: trade[0]):
st, lt = self._process_trade(security, date, quantity, price, security.multiplier)
gains = gains_by_year.setdefault(date.year, [0.0, 0.0])
gains[0] += st
gains[1] += lt
first_year = self._tax_year if self._tax_year is not None else min(gains_by_year, default=target.now.year)
tax = deduction = 0.0
for year in range(first_year, target.now.year + 1):
year_tax, year_deduction = self._settle_year(target, year, *gains_by_year.get(year, (0.0, 0.0)))
tax += year_tax
deduction += year_deduction
short_term_gains = sum(gains[0] for gains in gains_by_year.values())
long_term_gains = sum(gains[1] for gains in gains_by_year.values())
if not is_zero(tax):
# if tax is not 0 charge the portfolio a fee to simulate the
# tax liability
target.adjust(-tax, flow=False, fee=tax)
self._records.append(
{
"Date": target.now,
"short_term_gain": short_term_gains,
"long_term_gain": long_term_gains,
"deduction": deduction,
"tax": tax,
"short_term_carryforward": self._short_term_carryforward,
"long_term_carryforward": self._long_term_carryforward,
}
)
target.perm["tax_liabilities"] = self.liabilities
self._last_date = target.now
return True