"""
Contains the core building blocks of the framework.
"""
from __future__ import division
import math
from copy import deepcopy
import cython as cy
import numpy as np
import pandas as pd
PAR = 100.0
TOL = 1e-16
[docs]@cy.locals(x=cy.double)
def is_zero(x):
"""
Test for zero that is robust against floating point precision errors
"""
return abs(x) < TOL
[docs]class Node(object):
"""
The Node is the main building block in bt's tree structure design.
Both StrategyBase and SecurityBase inherit Node. It contains the
core functionality of a tree node.
Args:
* name (str): The Node name
* parent (Node): The parent Node
* children (dict, list): A collection of children. If dict,
the format is {name: child}, if list then list of children.
Children can be any type of Node or str.
String values correspond to children which will be lazily created
with that name when needed.
Attributes:
* name (str): Node name
* parent (Node): Node parent
* root (Node): Root node of the tree (topmost node)
* children (dict): Node's children
* now (datetime): Used when backtesting to store current date
* stale (bool): Flag used to determine if Node is stale and need
updating
* prices (TimeSeries): Prices of the Node. Prices for a security will
be the security's price, for a strategy it will be an index that
reflects the value of the strategy over time.
* price (float): last price
* value (float): last value
* notional_value (float): last notional value. Notional value is used
when fixed_income=True. It is always positive for strategies, but
is signed for securities (and typically set to either market value,
position, or zero).
* weight (float): weight in parent
* full_name (str): Name including parents' names
* members (list): Current Node + node's children
* fixed_income (bool): Whether the node corresponds to a fixed income
component, which would use notional-weighting instead of market
value weighing. See also :class:`FixedIncomeStrategy <bt.core.FixedIncomeStrategy>`
for more details.
"""
_capital = cy.declare(cy.double)
_price = cy.declare(cy.double)
_value = cy.declare(cy.double)
_notl_value = cy.declare(cy.double)
_weight = cy.declare(cy.double)
_issec = cy.declare(cy.bint)
_has_strat_children = cy.declare(cy.bint)
_fixed_income = cy.declare(cy.bint)
_bidoffer_set = cy.declare(cy.bint)
_bidoffer_paid = cy.declare(cy.double)
def __init__(self, name, parent=None, children=None):
self.name = name
# children helpers
self.children = {}
self._lazy_children = {}
self._universe_tickers = []
self._childrenv = [] # Shortcut to self.children.values()
self._original_children_are_present = (children is not None) and (
len(children) >= 1
)
# strategy children helpers
self._has_strat_children = False
self._strat_children = []
if parent is None:
self.parent = self
self.root = self
# by default all positions are integer
self.integer_positions = True
else:
self.parent = parent
parent._add_children([self], dc=False)
self._add_children(children, dc=True)
# set default value for now
self.now = 0
# make sure root has stale flag
# used to avoid unnecessary update
# sometimes we change values in the tree and we know that we will need
# to update if another node tries to access a given value (say weight).
# This avoid calling the update until it is actually needed.
self.root.stale = False
# helper vars
self._price = 0
self._value = 0
self._notl_value = 0
self._weight = 0
self._capital = 0
# is security flag - used to avoid updating 0 pos securities
self._issec = False
# fixed income flag - used to turn on notional weighing
self._fixed_income = False
# flag for whether to do bid/offer accounting
self._bidoffer_set = False
self._bidoffer_paid = 0
def __getitem__(self, key):
return self.children[key]
def _add_children(self, children, dc):
"""
Add the collection of children to the current node, where
children is either an iterable of children objects/strings, or
a dictionary
Args:
dc (bool): Whether or not to deepcopy nodes before adding them.
"""
# if at least 1 children is specified
if children is not None:
if isinstance(children, dict):
# Preserve the names from the dictionary by renaming the nodes
tmp = []
for name, c in children.items():
if isinstance(c, str):
tmp.append(name)
else:
if dc:
c = deepcopy(c)
c.name = name
tmp.append(c)
children = tmp
for c in children:
if dc: # deepcopy object for possible later reuse
c = deepcopy(c)
if type(c) == str:
if c in self._universe_tickers:
raise ValueError("Child %s already exists" % c)
# Create default security with lazy_add
c = Security(c, lazy_add=True)
if getattr(c, "lazy_add", False):
self._lazy_children[c.name] = c
else:
if c.name in self.children:
raise ValueError("Child %s already exists" % c)
c.parent = self
c._set_root(self.root)
c.use_integer_positions(self.integer_positions)
self.children[c.name] = c
self._childrenv.append(c)
# if strategy, turn on flag and add name to list
# strategy children have special treatment
if isinstance(c, StrategyBase):
self._has_strat_children = True
self._strat_children.append(c.name)
# if not strategy, then we will want to add this to
# universe_tickers to filter on setup
elif c.name not in self._universe_tickers:
self._universe_tickers.append(c.name)
def _set_root(self, root):
self.root = root
for c in self._childrenv:
c._set_root(root)
[docs] def use_integer_positions(self, integer_positions):
"""
Set indicator to use (or not) integer positions for a given strategy or
security.
By default all positions in number of stocks should be integer.
However this may lead to unexpected results when working with adjusted
prices of stocks. Because of series of reverse splits of stocks, the
adjusted prices back in time might be high. Thus rounding of desired
amount of stocks to buy may lead to having 0, and thus ignoring this
stock from backtesting.
"""
self.integer_positions = integer_positions
for c in self._childrenv:
c.use_integer_positions(integer_positions)
@property
def fixed_income(self):
"""
Whether the node is a fixed income node (using notional weighting).
"""
return self._fixed_income
@property
def prices(self):
"""
A TimeSeries of the Node's price.
"""
# can optimize depending on type -
# securities don't need to check stale to
# return latest prices, whereas strategies do...
raise NotImplementedError()
@property
def price(self):
"""
Current price of the Node
"""
# can optimize depending on type -
# securities don't need to check stale to
# return latest prices, whereas strategies do...
raise NotImplementedError()
@property
def value(self):
"""
Current value of the Node
"""
if self.root.stale:
self.root.update(self.root.now, None)
return self._value
@property
def notional_value(self):
"""
Current notional value of the Node
"""
if self.root.stale:
self.root.update(self.root.now, None)
return self._notl_value
@property
def weight(self):
"""
Current weight of the Node (with respect to the parent).
"""
if self.root.stale:
self.root.update(self.root.now, None)
return self._weight
[docs] def setup(self, universe, **kwargs):
"""
Setup method used to initialize a Node with a universe, and potentially other information.
"""
raise NotImplementedError()
[docs] def update(self, date, data=None, inow=None):
"""
Update Node with latest date, and optionally some data.
"""
raise NotImplementedError()
[docs] def adjust(self, amount, update=True, flow=True):
"""
Adjust Node value by amount.
"""
raise NotImplementedError()
[docs] def allocate(self, amount, update=True):
"""
Allocate capital to Node.
"""
raise NotImplementedError()
@property
def members(self):
"""
Node members. Members include current node as well as Node's
children.
"""
res = [self]
for c in list(self.children.values()):
res.extend(c.members)
return res
@property
def full_name(self):
if self.parent == self:
return self.name
else:
return "%s>%s" % (self.parent.full_name, self.name)
def __repr__(self):
return "<%s %s>" % (self.__class__.__name__, self.full_name)
[docs] def to_dot(self, root=True):
"""
Represent the node structure in DOT format.
"""
name = lambda x: x.name or repr(self) # noqa: E731
edges = "\n".join(
'\t"%s" -> "%s"' % (name(self), name(c)) for c in self.children.values()
)
below = "\n".join(c.to_dot(False) for c in self.children.values())
body = "\n".join([edges, below]).rstrip()
if root:
return "\n".join(["digraph {", body, "}"])
return body
[docs]class StrategyBase(Node):
"""
Strategy Node. Used to define strategy logic within a tree.
A Strategy's role is to allocate capital to it's children
based on a function.
Args:
* name (str): Strategy name
* children (dict, list): A collection of children. If dict,
the format is {name: child}, if list then list of children.
Children can be any type of Node or str.
String values correspond to children which will be lazily created
with that name when needed.
* parent (Node): The parent Node
Attributes:
* name (str): Strategy name
* parent (Strategy): Strategy parent
* root (Strategy): Root node of the tree (topmost node)
* children (dict): Strategy's children
* now (datetime): Used when backtesting to store current date
* stale (bool): Flag used to determine if Strategy is stale and need
updating
* prices (TimeSeries): Prices of the Strategy - basically an index that
reflects the value of the strategy over time.
* outlays (DataFrame): Outlays for each SecurityBase child
* price (float): last price
* value (float): last value
* notional_value (float): last notional value
* weight (float): weight in parent
* full_name (str): Name including parents' names
* members (list): Current Strategy + strategy's children
* securities (list): List of strategy children that are of type
SecurityBase
* commission_fn (fn(quantity, price)): A function used to determine the
commission (transaction fee) amount. Could be used to model
slippage (implementation shortfall). Note that often fees are
symmetric for buy and sell and absolute value of quantity should
be used for calculation.
* capital (float): Capital amount in Strategy - cash
* universe (DataFrame): Data universe available at the current time.
Universe contains the data passed in when creating a Backtest. Use
this data to determine strategy logic.
"""
_net_flows = cy.declare(cy.double)
_last_value = cy.declare(cy.double)
_last_notl_value = cy.declare(cy.double)
_last_price = cy.declare(cy.double)
_last_fee = cy.declare(cy.double)
_paper_trade = cy.declare(cy.bint)
bankrupt = cy.declare(cy.bint)
_last_chk = cy.declare(cy.bint)
def __init__(self, name, children=None, parent=None):
Node.__init__(self, name, children=children, parent=parent)
self._weight = 1
self._value = 0
self._notl_value = 0
self._price = PAR
# helper vars
self._net_flows = 0
self._last_value = 0
self._last_notl_value = 0
self._last_price = PAR
self._last_fee = 0
self._last_chk = 0
# default commission function
self.commission_fn = self._dflt_comm_fn
self._paper_trade = False
self._positions = None
self.bankrupt = False
@property
def price(self):
"""
Current price.
"""
if self.root.stale:
self.root.update(self.now, None)
return self._price
@property
def prices(self):
"""
TimeSeries of prices.
"""
if self.root.stale:
self.root.update(self.now, None)
return self._prices.loc[: self.now]
@property
def values(self):
"""
TimeSeries of values.
"""
if self.root.stale:
self.root.update(self.now, None)
return self._values.loc[: self.now]
@property
def notional_values(self):
"""
TimeSeries of notional values.
"""
if self.root.stale:
self.root.update(self.now, None)
return self._notl_values.loc[: self.now]
@property
def capital(self):
"""
Current capital - amount of unallocated capital left in strategy.
"""
# no stale check needed
return self._capital
@property
def cash(self):
"""
TimeSeries of unallocated capital.
"""
# no stale check needed
return self._cash
@property
def fees(self):
"""
TimeSeries of fees.
"""
if self.root.stale:
self.root.update(self.now, None)
return self._fees.loc[: self.now]
@property
def flows(self):
"""
TimeSeries of flows.
"""
if self.root.stale:
self.root.update(self.now, None)
return self._all_flows.loc[: self.now]
@property
def bidoffer_paid(self):
"""
Bid/offer spread paid on transactions in the current step
"""
if self._bidoffer_set:
if self.root.stale:
self.root.update(self.now, None)
return self._bidoffer_paid
else:
raise Exception(
"Bid/offer accounting not turned on: "
'"bidoffer" argument not provided during setup'
)
@property
def bidoffers_paid(self):
"""
TimeSeries of bid/offer spread paid on transactions in each step
"""
if self._bidoffer_set:
if self.root.stale:
self.root.update(self.now, None)
return self._bidoffers_paid.loc[: self.now]
else:
raise Exception(
"Bid/offer accounting not turned on: "
'"bidoffer" argument not provided during setup'
)
@property
def universe(self):
"""
Data universe available at the current time.
Universe contains the data passed in when creating a Backtest.
Use this data to determine strategy logic.
"""
# avoid windowing every time
# if calling and on same date return
# cached value
if self.now == self._last_chk:
return self._funiverse
else:
self._last_chk = self.now
self._funiverse = self._universe.loc[: self.now]
return self._funiverse
@property
def securities(self):
"""
Returns a list of children that are of type SecurityBase
"""
return [x for x in self.members if isinstance(x, SecurityBase)]
@property
def outlays(self):
"""
Returns a DataFrame of outlays for each child SecurityBase
"""
if self.root.stale:
self.root.update(self.root.now, None)
outlays = pd.DataFrame()
for x in self.securities:
if x.name in outlays.columns:
outlays[x.name] += x.outlays
else:
outlays[x.name] = x.outlays
return outlays
@property
def positions(self):
"""
TimeSeries of positions.
"""
# if accessing and stale - update first
if self.root.stale:
self.root.update(self.root.now, None)
vals = pd.DataFrame()
for x in self.members:
if isinstance(x, SecurityBase):
if x.name in vals.columns:
vals[x.name] += x.positions
else:
vals[x.name] = x.positions
self._positions = vals
return vals
[docs] def setup(self, universe, **kwargs):
"""
Setup strategy with universe. This will speed up future calculations
and updates.
"""
# save full universe in case we need it
self._original_data = universe
self._setup_kwargs = kwargs
# Guard against fixed income children of regular
# strategies as the "price" is just a reference
# value and should not be used for capital allocation
if self.fixed_income and not self.parent.fixed_income:
raise ValueError(
"Cannot have fixed income "
"strategy child (%s) of non-"
"fixed income strategy (%s)" % (self.name, self.parent.name)
)
# determine if needs paper trading
# and setup if so
if self is not self.parent:
self._paper_trade = True
self._paper_amount = 1000000
paper = deepcopy(self)
paper.parent = paper
paper.root = paper
paper._paper_trade = False
paper.setup(self._original_data, **kwargs)
paper.adjust(self._paper_amount)
self._paper = paper
# setup universe
funiverse = universe.copy()
# filter only if the node has any children specified as input,
# otherwise we use the full universe. If all children are strategies,
# funiverse will be empty, to signal that no other ticker should be
# used in addition to the strategies
if self._original_children_are_present:
# if we have universe_tickers defined, limit universe to
# those tickers
valid_filter = list(
set(universe.columns).intersection(self._universe_tickers)
)
funiverse = universe[valid_filter].copy()
# if we have strat children, we will need to create their columns
# in the new universe
if self._has_strat_children:
for c in self._strat_children:
funiverse[c] = np.nan
# must create to avoid pandas warning
funiverse = pd.DataFrame(funiverse)
self._universe = funiverse
# holds filtered universe
self._funiverse = funiverse
self._last_chk = None
# We're not bankrupt yet
self.bankrupt = False
# setup internal data
self.data = pd.DataFrame(
index=funiverse.index,
columns=["price", "value", "notional_value", "cash", "fees", "flows"],
data=0.0,
)
self._prices = self.data["price"]
self._values = self.data["value"]
self._notl_values = self.data["notional_value"]
self._cash = self.data["cash"]
self._fees = self.data["fees"]
self._all_flows = self.data["flows"]
if "bidoffer" in kwargs:
self._bidoffer_set = True
self.data["bidoffer_paid"] = 0.0
self._bidoffers_paid = self.data["bidoffer_paid"]
# setup children as well - use original universe here - don't want to
# pollute with potential strategy children in funiverse
if self.children is not None:
[c.setup(universe, **kwargs) for c in self._childrenv]
[docs] def setup_from_parent(self, **kwargs):
"""
Setup a strategy from the parent. Used when dynamically creating
child strategies.
Args:
* kwargs: additional arguments that will be passed to setup
(potentially overriding those from the parent)
"""
all_kwargs = self.parent._setup_kwargs.copy()
all_kwargs.update(kwargs)
self.setup(self.parent._original_data, **all_kwargs)
if self.name not in self.parent._universe:
self.parent._universe[self.name] = np.nan
[docs] def get_data(self, key):
"""
Returns additional data that was passed to the setup function via kwargs,
for use in the algos. This allows algos to reference data sources "by name",
where the binding of the data to the name happens at Backtest creation
time rather than at Strategy definition time, allowing the same strategies
to be run against different data sets more easily.
"""
return self._setup_kwargs[key]
[docs] @cy.locals(
newpt=cy.bint,
val=cy.double,
ret=cy.double,
coupons=cy.double,
notl_val=cy.double,
bidoffer_paid=cy.double,
)
def update(self, date, data=None, inow=None):
"""
Update strategy. Updates prices, values, weight, etc.
"""
# resolve stale state
self.root.stale = False
# update helpers on date change
# also set newpt flag
newpt = False
if self.now == 0:
newpt = True
elif date != self.now:
self._net_flows = 0
self._last_price = self._price
self._last_value = self._value
self._last_notl_value = self._notl_value
self._last_fee = 0.0
newpt = True
# update now
self.now = date
if inow is None:
if self.now == 0:
inow = 0
else:
inow = self.data.index.get_loc(date)
# update children if any and calculate value
val = self._capital # default if no children
notl_val = 0.0 # Capital doesn't count towards notional value
bidoffer_paid = 0.0
coupons = 0
if self.children:
for c in self._childrenv:
# Sweep up cash from the security nodes (from coupon payments, etc)
if c._issec and newpt:
coupons += c._capital
c._capital = 0
# avoid useless update call
if c._issec and not c._needupdate:
continue
c.update(date, data, inow)
val += c.value
# Strategies always have positive notional value
notl_val += abs(c.notional_value)
if self._bidoffer_set:
bidoffer_paid += c.bidoffer_paid
self._capital += coupons
val += coupons
if self.root == self:
if (
(val < 0)
and not self.bankrupt
and not self.fixed_income
and not is_zero(val)
):
# Declare a bankruptcy
self.bankrupt = True
self.flatten()
# update data if this value is different or
# if now has changed - avoid all this if not since it
# won't change
if (
newpt
or not is_zero(self._value - val)
or not is_zero(self._notl_value - notl_val)
):
self._value = val
self._values.values[inow] = val
self._notl_value = notl_val
self._notl_values.values[inow] = notl_val
if self._bidoffer_set:
self._bidoffer_paid = bidoffer_paid
self._bidoffers_paid.values[inow] = bidoffer_paid
if self.fixed_income:
# For notional weights, we compute additive return
pnl = self._value - (self._last_value + self._net_flows)
if not is_zero(self._last_notl_value):
ret = pnl / self._last_notl_value * PAR
elif not is_zero(self._notl_value):
# This case happens when paying bid/offer or fees when building an initial position
ret = pnl / self._notl_value * PAR
else:
if is_zero(pnl):
ret = 0
else:
raise ZeroDivisionError(
"Could not update %s on %s. Last notional value "
"was %s and pnl was %s. Therefore, "
"we are dividing by zero to obtain the pnl "
"per unit notional for the period."
% (self.name, self.now, self._last_notl_value, pnl)
)
self._price = self._last_price + ret
self._prices.values[inow] = self._price
else:
bottom = self._last_value + self._net_flows
if not is_zero(bottom):
ret = self._value / (self._last_value + self._net_flows) - 1
else:
if is_zero(self._value):
ret = 0
else:
raise ZeroDivisionError(
"Could not update %s on %s. Last value "
"was %s and net flows were %s. Current"
"value is %s. Therefore, "
"we are dividing by zero to obtain the return "
"for the period."
% (
self.name,
self.now,
self._last_value,
self._net_flows,
self._value,
)
)
self._price = self._last_price * (1 + ret)
self._prices.values[inow] = self._price
# update children weights
if self.children:
for c in self._childrenv:
# avoid useless update call
if c._issec and not c._needupdate:
continue
if self.fixed_income:
if not is_zero(notl_val):
c._weight = c.notional_value / notl_val
else:
c._weight = 0.0
else:
if not is_zero(val):
c._weight = c.value / val
else:
c._weight = 0.0
# if we have strategy children, we will need to update them in universe
if self._has_strat_children:
for c in self._strat_children:
# TODO: optimize ".loc" here as well
self._universe.loc[date, c] = self.children[c].price
# Cash should track the unallocated capital at the end of the day, so
# we should update it every time we call "update".
# Same for fees and flows
self._cash.values[inow] = self._capital
self._fees.values[inow] = self._last_fee
self._all_flows.values[inow] = self._net_flows
# update paper trade if necessary
if self._paper_trade:
if newpt:
self._paper.update(date)
self._paper.run()
self._paper.update(date)
# update price
self._price = self._paper.price
self._prices.values[inow] = self._price
[docs] @cy.locals(amount=cy.double, update=cy.bint, flow=cy.bint, fees=cy.double)
def adjust(self, amount, update=True, flow=True, fee=0.0):
"""
Adjust capital - used to inject capital to a Strategy. This injection
of capital will have no effect on the children.
Args:
* amount (float): Amount to adjust by.
* update (bool): Force update?
* flow (bool): Is this adjustment a flow? A flow will not have an
impact on the performance (price index). Example of flows are
simply capital injections (say a monthly contribution to a
portfolio). This should not be reflected in the returns. A
non-flow (flow=False) does impact performance. A good example
of this is a commission, or a dividend.
"""
# adjust capital
self._capital += amount
self._last_fee += fee
# if flow - increment net_flows - this will not affect
# performance. Commissions and other fees are not flows since
# they have a performance impact
if flow:
self._net_flows += amount
if update:
# indicates that data is now stale and must
# be updated before access
self.root.stale = True
[docs] @cy.locals(amount=cy.double, update=cy.bint)
def allocate(self, amount, child=None, update=True):
"""
Allocate capital to Strategy. By default, capital is allocated
recursively down the children, proportionally to the children's
weights. If a child is specified, capital will be allocated
to that specific child.
Allocation also have a side-effect. They will deduct the same amount
from the parent's "account" to offset the allocation. If there is
remaining capital after allocation, it will remain in Strategy.
Args:
* amount (float): Amount to allocate.
* child (str): If specified, allocation will be directed to child
only. Specified by name.
* update (bool): Force update.
"""
# allocate to child
if child is not None:
self._create_child_if_needed(child)
# allocate to child
self.children[child].allocate(amount)
# allocate to self
else:
# adjust parent's capital
# no need to update now - avoids repetition
if self.parent == self:
self.parent.adjust(-amount, update=False, flow=True)
else:
# do NOT set as flow - parent will be another strategy
# and therefore should not incur flow
self.parent.adjust(-amount, update=False, flow=False)
# adjust self's capital
self.adjust(amount, update=False, flow=True)
# push allocation down to children if any
# use _weight to avoid triggering an update
if self.children is not None:
[c.allocate(amount * c._weight, update=False) for c in self._childrenv]
# mark as stale if update requested
if update:
self.root.stale = True
[docs] @cy.locals(q=cy.double, update=cy.bint)
def transact(self, q, child=None, update=True):
"""
Transact a notional amount q in the Strategy. By default, it is allocated
recursively down the children, proportionally to the children's
weights. Recursive allocation only works for fixed income strategies.
If a child is specified, notional will be allocated
to that specific child.
Args:
* q (float): Notional quantity to allocate.
* child (str): If specified, allocation will be directed to child
only. Specified by name.
* update (bool): Force update.
"""
# allocate to child
if child is not None:
self._create_child_if_needed(child)
# allocate to child
self.children[child].transact(q)
# allocate to self
else:
# push allocation down to children if any
# use _weight to avoid triggering an update
if self.children is not None:
[c.transact(q * c._weight, update=False) for c in self._childrenv]
# mark as stale if update requested
if update:
self.root.stale = True
[docs] @cy.locals(delta=cy.double, weight=cy.double, base=cy.double, update=cy.bint)
def rebalance(self, weight, child, base=np.nan, update=True):
"""
Rebalance a child to a given weight.
This is a helper method to simplify code logic. This method is used
when we want to see the weight of a particular child to a set amount.
It is similar to allocate, but it calculates the appropriate allocation
based on the current weight. For fixed income strategies, it uses
transact to rebalance based on notional value instead of capital.
Args:
* weight (float): The target weight. Usually between -1.0 and 1.0.
* child (str): child to allocate to - specified by name.
* base (float): If specified, this is the base amount all weight
delta calculations will be based off of. This is useful when we
determine a set of weights and want to rebalance each child
given these new weights. However, as we iterate through each
child and call this method, the base (which is by default the
current value) will change. Therefore, we can set this base to
the original value before the iteration to ensure the proper
allocations are made.
* update (bool): Force update?
"""
# if weight is 0 - we want to close child
if is_zero(weight):
if child in self.children:
return self.close(child, update=update)
else:
return
# if no base specified use self's value
if np.isnan(base):
if self.fixed_income:
base = self.notional_value
else:
base = self.value
# else make sure we have child
self._create_child_if_needed(child)
# allocate to child
# figure out weight delta
c = self.children[child]
if self.fixed_income:
# In fixed income strategies, the provided "base" value can be used
# to upscale/downscale the notional_value of the strategy, whereas
# in normal strategies the total capital is fixed. Thus, when
# rebalancing, we must take care to account for differences between
# previous notional value and passed base value. Note that for
# updating many weights in sequence, one must pass update=False so
# that the existing weights and notional_value are not recalculated
# before finishing.
if c.fixed_income:
delta = weight * base - c.weight * self.notional_value
c.transact(delta, update=update)
else:
delta = weight * base - c.weight * self.notional_value
c.allocate(delta, update=update)
else:
delta = weight - c.weight
c.allocate(delta * base, update=update)
[docs] @cy.locals(update=cy.bint)
def close(self, child, update=True):
"""
Close a child position - alias for rebalance(0, child). This will also
flatten (close out all) the child's children.
Args:
* child (str): Child, specified by name.
"""
c = self.children[child]
# flatten if children not None
if c.children is not None and len(c.children) != 0:
c.flatten()
if self.fixed_income:
if c.position != 0.0:
c.transact(-c.position, update=update)
else:
if c.value != 0.0 and not np.isnan(c.value):
c.allocate(-c.value, update=update)
[docs] def flatten(self):
"""
Close all child positions.
"""
# go right to base alloc
if self.fixed_income:
[
c.transact(-c.position, update=False)
for c in self._childrenv
if c.position != 0
]
else:
[
c.allocate(-c.value, update=False)
for c in self._childrenv
if c.value != 0
]
self.root.stale = True
[docs] def run(self):
"""
This is the main logic method. Override this method to provide some
algorithm to execute on each date change. This method is called by
backtester.
"""
pass
[docs] def set_commissions(self, fn):
"""
Set commission (transaction fee) function.
Args:
fn (fn(quantity, price)): Function used to determine commission
amount.
"""
self.commission_fn = fn
for c in self._childrenv:
if isinstance(c, StrategyBase):
c.set_commissions(fn)
[docs] def get_transactions(self):
"""
Helper function that returns the transactions in the following format:
Date, Security | quantity, price
The result is a MultiIndex DataFrame.
"""
# get prices for each security in the strategy & create unstacked
# series
prc = pd.DataFrame({x.name: x.prices for x in self.securities}).unstack()
# get security positions
positions = pd.DataFrame()
for x in self.securities:
if x.name in positions.columns:
positions[x.name] += x.positions
else:
positions[x.name] = x.positions
# trades are diff
trades = positions.diff()
# must adjust first row
trades.iloc[0] = positions.iloc[0]
# now convert to unstacked series, dropping nans along the way
trades = trades[trades != 0].unstack().dropna()
# Adjust prices for bid/offer paid if needed
if self._bidoffer_set:
bidoffer = pd.DataFrame(
{x.name: x.bidoffers_paid for x in self.securities}
).unstack()
prc += bidoffer / trades
res = pd.DataFrame({"price": prc, "quantity": trades}).dropna(
subset=["quantity"]
)
# set names
res.index.names = ["Security", "Date"]
# swap levels so that we have (date, security) as index and sort
res = res.swaplevel().sort_index()
return res
@cy.locals(q=cy.double, p=cy.double)
def _dflt_comm_fn(self, q, p):
return 0.0
def _create_child_if_needed(self, child):
if child not in self.children:
# Look up name in lazy children, or create a default security
c = self._lazy_children.pop(child, Security(child))
c.lazy_add = False
# add child to tree
self._add_children([c], dc=False)
c.setup(self._universe, **self._setup_kwargs)
# update to bring up to speed
c.update(self.now)
[docs]class SecurityBase(Node):
"""
Security Node. Used to define a security within a tree.
A Security's has no children. It simply models an asset that can be bought
or sold.
Args:
* name (str): Security name
* multiplier (float): security multiplier - typically used for
derivatives or to trade in lots. The quantity of the Security will
always be multiplied by this to determine the underlying amount.
* lazy_add (bool): Flag to control whether instrument should be added
to strategy children lazily, i.e. only when there is a transaction
on the instrument. This improves performance of strategies which
transact on a sparse set of children.
Attributes:
* name (str): Security name
* parent (Security): Security parent
* root (Security): Root node of the tree (topmost node)
* now (datetime): Used when backtesting to store current date
* stale (bool): Flag used to determine if Security is stale and need
updating
* prices (TimeSeries): Security prices.
* price (float): last price
* outlays (TimeSeries): Series of outlays. Positive outlays mean
capital was allocated to security and security consumed that
amount. Negative outlays are the opposite. This can be useful for
calculating turnover at the strategy level.
* value (float): last value - basically position * price * multiplier
* weight (float): weight in parent
* full_name (str): Name including parents' names
* members (list): Current Security + strategy's children
* position (float): Current position (quantity).
* bidoffer (float): Current bid/offer spread
* bidoffers (TimeSeries): Series of bid/offer spreads
* bidoffer_paid (TimeSeries): Series of bid/offer paid on transactions
"""
_last_pos = cy.declare(cy.double)
_position = cy.declare(cy.double)
multiplier = cy.declare(cy.double)
_prices_set = cy.declare(cy.bint)
_needupdate = cy.declare(cy.bint)
_outlay = cy.declare(cy.double)
_bidoffer = cy.declare(cy.double)
@cy.locals(multiplier=cy.double)
def __init__(self, name, multiplier=1, lazy_add=False):
Node.__init__(self, name, parent=None, children=None)
self._value = 0
self._price = 0
self._weight = 0
self._position = 0
self.multiplier = multiplier
self.lazy_add = lazy_add
# opt
self._last_pos = 0
self._issec = True
self._needupdate = True
self._outlay = 0
self._bidoffer = 0
@property
def price(self):
"""
Current price.
"""
# if accessing and stale - update first
if self._needupdate or self.now != self.parent.now:
self.update(self.root.now)
return self._price
@property
def prices(self):
"""
TimeSeries of prices.
"""
# if accessing and stale - update first
if self._needupdate or self.now != self.parent.now:
self.update(self.root.now)
return self._prices.loc[: self.now]
@property
def values(self):
"""
TimeSeries of values.
"""
# if accessing and stale - update first
if self._needupdate or self.now != self.parent.now:
self.update(self.root.now)
if self.root.stale:
self.root.update(self.root.now, None)
return self._values.loc[: self.now]
@property
def notional_values(self):
"""
TimeSeries of notional values.
"""
# if accessing and stale - update first
if self._needupdate or self.now != self.parent.now:
self.update(self.root.now)
if self.root.stale:
self.root.update(self.root.now, None)
return self._notl_values.loc[: self.now]
@property
def position(self):
"""
Current position
"""
# no stale check needed
return self._position
@property
def positions(self):
"""
TimeSeries of positions.
"""
# if accessing and stale - update first
if self._needupdate:
self.update(self.root.now)
if self.root.stale:
self.root.update(self.root.now, None)
return self._positions.loc[: self.now]
@property
def outlays(self):
"""
TimeSeries of outlays. Positive outlays (buys) mean this security
received and consumed capital (capital was allocated to it). Negative
outlays are the opposite (the security close/sold, and returned capital
to parent).
"""
# if accessing and stale - update first
if self._needupdate or self.now != self.parent.now:
self.update(self.root.now)
if self.root.stale:
self.root.update(self.root.now, None)
return self._outlays.loc[: self.now]
@property
def bidoffer(self):
"""
Current bid/offer spread.
"""
# if accessing and stale - update first
if self._needupdate or self.now != self.parent.now:
self.update(self.root.now)
return self._bidoffer
@property
def bidoffers(self):
"""
TimeSeries of bid/offer spread
"""
if self._bidoffer_set:
# if accessing and stale - update first
if self._needupdate or self.now != self.parent.now:
self.update(self.root.now)
return self._bidoffers.loc[: self.now]
else:
raise Exception(
"Bid/offer accounting not turned on: "
'"bidoffer" argument not provided during setup'
)
@property
def bidoffer_paid(self):
"""
TimeSeries of bid/offer spread paid on transactions in the current step
"""
# if accessing and stale - update first
if self._needupdate or self.now != self.parent.now:
self.update(self.root.now)
return self._bidoffer_paid
@property
def bidoffers_paid(self):
"""
TimeSeries of bid/offer spread paid on transactions in the current step
"""
if self._bidoffer_set:
# if accessing and stale - update first
if self._needupdate or self.now != self.parent.now:
self.update(self.root.now)
if self.root.stale:
self.root.update(self.root.now, None)
return self._bidoffers_paid.loc[: self.now]
else:
raise Exception(
"Bid/offer accounting not turned on: "
'"bidoffer" argument not provided during setup'
)
[docs] def setup(self, universe, **kwargs):
"""
Setup Security with universe. Speeds up future runs.
Args:
* universe (DataFrame): DataFrame of prices with security's name as
one of the columns.
* bidoffer (DataFrame): Optional argument that represents the
bid/offer spread on each security across time. If provided, the
strategy will account for these costs when rebalancing.
* kwargs (dict): Dictionary of additional information needed by
the strategy. In particular, often takes the form of a DataFrame
of security level information (i.e. signals, risk, etc).
"""
# if we already have all the prices, we will store them to speed up
# future updates
try:
prices = universe[self.name]
except KeyError:
prices = None
# setup internal data
if prices is not None:
self._prices = prices
self.data = pd.DataFrame(
index=universe.index,
columns=["value", "position", "notional_value"],
data=0.0,
)
self._prices_set = True
else:
self.data = pd.DataFrame(
index=universe.index,
columns=["price", "value", "position", "notional_value"],
)
self._prices = self.data["price"]
self._prices_set = False
self._values = self.data["value"]
self._notl_values = self.data["notional_value"]
self._positions = self.data["position"]
# add _outlay
self.data["outlay"] = 0.0
self._outlays = self.data["outlay"]
# save bidoffer, if provided
if "bidoffer" in kwargs:
self._bidoffer_set = True
self._bidoffers = kwargs["bidoffer"]
try:
bidoffers = self._bidoffers[self.name]
except KeyError:
bidoffers = None
if bidoffers is not None:
if bidoffers.index.equals(universe.index):
self._bidoffers = bidoffers
else:
raise ValueError("Index of bidoffer must match universe data")
else:
self.data["bidoffer"] = 0.0
self._bidoffers = self.data["bidoffer"]
self.data["bidoffer_paid"] = 0.0
self._bidoffers_paid = self.data["bidoffer_paid"]
[docs] @cy.locals(prc=cy.double)
def update(self, date, data=None, inow=None):
"""
Update security with a given date and optionally, some data.
This will update price, value, weight, etc.
"""
# filter for internal calls when position has not changed - nothing to
# do. Internal calls (stale root calls) have None data. Also want to
# make sure date has not changed, because then we do indeed want to
# update.
if date == self.now and self._last_pos == self._position:
return
if inow is None:
if date == 0:
inow = 0
else:
inow = self.data.index.get_loc(date)
# date change - update price
if date != self.now:
# update now
self.now = date
if self._prices_set:
self._price = self._prices.values[inow]
# traditional data update
elif data is not None:
prc = data[self.name]
self._price = prc
self._prices.values[inow] = prc
# update bid/offer
if self._bidoffer_set:
self._bidoffer = self._bidoffers.values[inow]
self._bidoffer_paid = 0.0
self._positions.values[inow] = self._position
self._last_pos = self._position
if np.isnan(self._price):
if is_zero(self._position):
self._value = 0
else:
raise Exception(
"Position is open (non-zero: %s) and latest price is NaN "
"for security %s on %s. Cannot update node value."
% (self._position, self.name, date)
)
else:
self._value = self._position * self._price * self.multiplier
self._notl_value = self._value
self._values.values[inow] = self._value
self._notl_values.values[inow] = self._notl_value
if is_zero(self._weight) and is_zero(self._position):
self._needupdate = False
# save outlay to outlays
if self._outlay != 0:
self._outlays.values[inow] += self._outlay
# reset outlay back to 0
self._outlay = 0
if self._bidoffer_set:
self._bidoffers_paid.values[inow] = self._bidoffer_paid
[docs] @cy.locals(
amount=cy.double, update=cy.bint, q=cy.double, outlay=cy.double, i=cy.int
)
def allocate(self, amount, update=True):
"""
This allocates capital to the Security. This is the method used to
buy/sell the security.
A given amount of shares will be determined on the current price, a
commission will be calculated based on the parent's commission fn, and
any remaining capital will be passed back up to parent as an
adjustment.
Args:
* amount (float): Amount of adjustment.
* update (bool): Force update?
"""
# will need to update if this has been idle for a while...
# update if needupdate or if now is stale
# fetch parent's now since our now is stale
if self._needupdate or self.now != self.parent.now:
self.update(self.parent.now)
# ignore 0 alloc
# Note that if the price of security has dropped to zero, then it
# should never be selected by SelectAll, SelectN etc. I.e. we should
# not open the position at zero price. At the same time, we are able
# to close it at zero price, because at that point amount=0.
# Note also that we don't erase the position in an asset which price
# has dropped to zero (though the weight will indeed be = 0)
if is_zero(amount):
return
if self.parent is self or self.parent is None:
raise Exception("Cannot allocate capital to a parentless security")
if is_zero(self._price) or np.isnan(self._price):
raise Exception(
"Cannot allocate capital to "
"%s because price is %s as of %s"
% (self.name, self._price, self.parent.now)
)
# buy/sell
# determine quantity - must also factor in commission
# closing out?
if is_zero(amount + self._value):
q = -self._position
else:
q = amount / (self._price * self.multiplier)
if self.integer_positions:
if (self._position > 0) or (is_zero(self._position) and (amount > 0)):
# if we're going long or changing long position
q = math.floor(q)
else:
# if we're going short or changing short position
q = math.ceil(q)
# if q is 0 nothing to do
if is_zero(q) or np.isnan(q):
return
# unless we are closing out a position (q == -position)
# we want to ensure that
#
# - In the event of a positive amount, this indicates the maximum
# amount a given security can use up for a purchase. Therefore, if
# commissions push us above this amount, we cannot buy `q`, and must
# decrease its value
#
# - In the event of a negative amount, we want to 'raise' at least the
# amount indicated, no less. Therefore, if we have commission, we must
# sell additional units to fund this requirement. As such, q must once
# again decrease.
#
if not q == -self._position:
full_outlay, _, _, _ = self.outlay(q)
# if full outlay > amount, we must decrease the magnitude of `q`
# this can potentially lead to an infinite loop if the commission
# per share > price per share. However, we cannot really detect
# that in advance since the function can be non-linear (say a fn
# like max(1, abs(q) * 0.01). Nevertheless, we want to avoid these
# situations.
# cap the maximum number of iterations to 1e4 and raise exception
# if we get there
# if integer positions then we know we are stuck if q doesn't change
# if integer positions is false then we want full_outlay == amount
# if integer positions is true then we want to be at the q where
# if we bought 1 more then we wouldn't have enough cash
i = 0
last_q = q
last_amount_short = full_outlay - amount
while not np.isclose(full_outlay, amount, rtol=0.0) and q != 0:
dq_wout_considering_tx_costs = (full_outlay - amount) / (
self._price * self.multiplier
)
q = q - dq_wout_considering_tx_costs
if self.integer_positions:
q = math.floor(q)
full_outlay, _, _, _ = self.outlay(q)
# if our q is too low and we have integer positions
# then we know that the correct quantity is the one where
# the outlay of q + 1 < amount. i.e. if we bought one more
# position then we wouldn't have enough cash
if self.integer_positions:
full_outlay_of_1_more, _, _, _ = self.outlay(q + 1)
if full_outlay < amount and full_outlay_of_1_more > amount:
break
# if not integer positions then we should keep going until
# full_outlay == amount or is close enough
i = i + 1
if i > 1e4:
raise Exception(
"Potentially infinite loop detected. This occurred "
"while trying to reduce the amount of shares purchased"
" to respect the outlay <= amount rule. This is most "
"likely due to a commission function that outputs a "
"commission that is greater than the amount of cash "
"a short sale can raise."
)
if self.integer_positions and last_q == q:
raise Exception(
"Newton Method like root search for quantity is stuck!"
" q did not change in iterations so it is probably a bug"
" but we are not entirely sure it is wrong! Consider "
" changing to warning."
)
last_q = q
if np.abs(full_outlay - amount) > np.abs(last_amount_short):
raise Exception(
"The difference between what we have raised with q and"
" the amount we are trying to raise has gotten bigger since"
" last iteration! full_outlay should always be approaching"
" amount! There may be a case where the commission fn is"
" not smooth"
)
last_amount_short = full_outlay - amount
self.transact(q, update, False)
[docs] @cy.locals(
q=cy.double,
update=cy.bint,
update_self=cy.bint,
outlay=cy.double,
bidoffer=cy.double,
)
def transact(self, q, update=True, update_self=True, price=None):
"""
This transacts the Security. This is the method used to
buy/sell the security for a given quantity.
The amount of shares is explicitly provided, a
commission will be calculated based on the parent's commission fn, and
any remaining capital will be passed back up to parent as an
adjustment.
Args:
* amount (float): Amount of adjustment.
* update (bool): Force update on parent due to transaction proceeds
* update_self (bool): Check for update on self
* price (float): Optional price if the transaction happens at a bespoke level
"""
# will need to update if this has been idle for a while...
# update if needupdate or if now is stale
# fetch parent's now since our now is stale
if update_self and (self._needupdate or self.now != self.parent.now):
self.update(self.parent.now)
# if q is 0 nothing to do
if is_zero(q) or np.isnan(q):
return
if price is not None and not self._bidoffer_set:
raise ValueError(
'Cannot transact at custom prices when "bidoffer" has '
"not been passed during setup to enable bid-offer tracking."
)
# this security will need an update, even if pos is 0 (for example if
# we close the positions, value and pos is 0, but still need to do that
# last update)
self._needupdate = True
# adjust position & value
self._position += q
# calculate proper adjustment for parent
# parent passed down amount so we want to pass
# -outlay back up to parent to adjust for capital
# used
full_outlay, outlay, fee, bidoffer = self.outlay(q, p=price)
# store outlay for future reference
self._outlay += outlay
self._bidoffer_paid += bidoffer
# call parent
self.parent.adjust(-full_outlay, update=update, flow=False, fee=fee)
[docs] @cy.locals(q=cy.double, p=cy.double)
def commission(self, q, p):
"""
Calculates the commission (transaction fee) based on quantity and
price. Uses the parent's commission_fn.
Args:
* q (float): quantity
* p (float): price
"""
return self.parent.commission_fn(q, p)
[docs] @cy.locals(q=cy.double)
def outlay(self, q, p=None):
"""
Determines the complete cash outlay (including commission) necessary
given a quantity q.
Second returning parameter is a commission itself.
Args:
* q (float): quantity
* p (float): price override
"""
if p is None:
fee = self.commission(q, self._price * self.multiplier)
bidoffer = abs(q) * 0.5 * self._bidoffer * self.multiplier
else:
# price override provided: custom transaction
fee = self.commission(q, p * self.multiplier)
bidoffer = q * (p - self._price) * self.multiplier
outlay = q * self._price * self.multiplier + bidoffer
return outlay + fee, outlay, fee, bidoffer
[docs] def run(self):
"""
Does nothing - securities have nothing to do on run.
"""
pass
[docs]class Security(SecurityBase):
"""
A standard security with no special features, and where notional value
is measured based on market value (notional times price).
It exists to be able to identify standard securities from nonstandard
ones via isinstance, i.e. isinstance( sec, Security ) would only return
True for a vanilla security, whereas SecurityBase would return True for
all securities.
"""
pass
[docs]class FixedIncomeSecurity(SecurityBase):
"""
A Fixed Income Security is a security where notional value is
measured only based on the quantity (par value) of the security.
Only relevant when using :class:`FixedIncomeStrategy <bt.core.FixedIncomeStrategy>`.
"""
[docs] @cy.locals(coupon=cy.double)
def update(self, date, data=None, inow=None):
"""
Update security with a given date and optionally, some data.
This will update price, value, weight, etc.
"""
if inow is None:
if date == 0:
inow = 0
else:
inow = self.data.index.get_loc(date)
super(FixedIncomeSecurity, self).update(date, data, inow)
# For fixed income securities (bonds, swaps), notional value is position size, not value!
self._notl_value = self._position
self._notl_values.values[inow] = self._notl_value
[docs]class CouponPayingSecurity(FixedIncomeSecurity):
"""
CouponPayingSecurity expands on SecurityBase to handle securities which
pay (possibly irregular) coupons (or other forms of cash disbursement).
More generally, this can include instruments with any sort of carry,
including (potentially asymmetric) holding costs.
Args:
* name (str): Security name
* multiplier (float): security multiplier - typically used for
derivatives.
* fixed_income (bool): Flag to control whether notional_value is based
only on quantity, or on market value (like an equity).
Defaults to notional weighting for coupon paying instruments.
* lazy_add (bool): Flag to control whether instrument should be added
to strategy children lazily, i.e. only when there is a transaction
on the instrument. This improves performance of strategies which
transact on a sparse set of children.
Attributes:
* SecurityBase attributes
* coupon (float): Current coupon payment (quantity).
* holding_cost (float): Current holding cost (quantity).
Represents a coupon-paying security, where coupon payments adjust
the capital of the parent. Coupons and costs must be passed in during setup.
"""
_coupon = cy.declare(cy.double)
_holding_cost = cy.declare(cy.double)
@cy.locals(multiplier=cy.double)
def __init__(self, name, multiplier=1, fixed_income=True, lazy_add=False):
super(CouponPayingSecurity, self).__init__(name, multiplier)
self._coupon = 0
self._holding_cost = 0
self._fixed_income = fixed_income
self.lazy_add = lazy_add
[docs] def setup(self, universe, **kwargs):
"""
Setup Security with universe and coupon data. Speeds up future runs.
Args:
* universe (DataFrame): DataFrame of prices with security's name as
one of the columns.
* coupons (DataFrame): Manatory DataFrame of coupon/carry amount with
the same schema as universe.
* cost_long (DataFrame): Optional DataFrame containing the cost of
holding a unit long position in the security (i.e. funding).
* cost_short (DataFrame): Optional DataFrame containing the cost of
holding a unit short position in the security (i.e. repo).
* kwargs (dict): Dictionary of additional information needed by
the strategy. In particular, often takes the form of a DataFrame
of security level information (i.e. signals, risk, etc).
"""
super(CouponPayingSecurity, self).setup(universe, **kwargs)
# Handle coupons
if "coupons" not in kwargs:
raise Exception(
'"coupons" must be passed to setup for a CouponPayingSecurity'
)
try:
self._coupons = kwargs["coupons"][self.name]
except KeyError:
self._coupons = None
if self._coupons is None or not self._coupons.index.equals(universe.index):
raise ValueError("Index of coupons must match universe data")
# Handle holding costs
try:
self._cost_long = kwargs["cost_long"][self.name]
except KeyError:
self._cost_long = None
try:
self._cost_short = kwargs["cost_short"][self.name]
except KeyError:
self._cost_short = None
self.data["coupon"] = 0.0
self.data["holding_cost"] = 0.0
self._coupon_income = self.data["coupon"]
self._holding_costs = self.data["holding_cost"]
[docs] @cy.locals(coupon=cy.double, cost=cy.double)
def update(self, date, data=None, inow=None):
"""
Update security with a given date and optionally, some data.
This will update price, value, weight, etc.
"""
if inow is None:
if date == 0:
inow = 0
else:
inow = self.data.index.get_loc(date)
if self._coupons is None:
raise Exception("coupons have not been set for security %s" % self.name)
# Standard update
super(CouponPayingSecurity, self).update(date, data, inow)
coupon = self._coupons.values[inow]
# If we were to call self.parent.adjust, then all the child weights would
# need to be updated. If each security pays a coupon, then this happens for
# each child. Instead, we store the coupon on self._capital, and it gets
# swept up as part of the strategy update
if np.isnan(coupon):
if is_zero(self._position):
self._coupon = 0.0
else:
raise Exception(
"Position is open (non-zero) and latest coupon is NaN "
"for security %s on %s. Cannot update node value."
% (self.name, date)
)
else:
self._coupon = self._position * coupon
if self._position > 0 and self._cost_long is not None:
cost = self._cost_long.values[inow]
self._holding_cost = self._position * cost
elif self._position < 0 and self._cost_short is not None:
cost = self._cost_short.values[inow]
self._holding_cost = -self._position * cost
else:
self._holding_cost = 0.0
self._capital = self._coupon - self._holding_cost
self._coupon_income.values[inow] = self._coupon
self._holding_costs.values[inow] = self._holding_cost
@property
def coupon(self):
"""
Current coupon payment (scaled by position)
"""
if (
self.root.stale
): # Stale check needed because coupon paid depends on position
self.root.update(self.root.now, None)
return self._coupon
@property
def coupons(self):
"""
TimeSeries of coupons paid (scaled by position)
"""
if (
self.root.stale
): # Stale check needed because coupon paid depends on position
self.root.update(self.root.now, None)
return self._coupon_income.loc[: self.now]
@property
def holding_cost(self):
"""
Current holding cost (scaled by position)
"""
if (
self.root.stale
): # Stale check needed because coupon paid depends on position
self.root.update(self.root.now, None)
return self._holding_cost
@property
def holding_costs(self):
"""
TimeSeries of coupons paid (scaled by position)
"""
if (
self.root.stale
): # Stale check needed because coupon paid depends on position
self.root.update(self.root.now, None)
return self._holding_costs.loc[: self.now]
[docs]class HedgeSecurity(SecurityBase):
"""
HedgeSecurity is a SecurityBase where the notional value is set to zero, and thus
does not count towards the notional value of the strategy. It is intended for use
in fixed income strategies.
For example in a corporate bond strategy, the notional value might refer to the size
of the corporate bond portfolio, and exclude the notional of treasury bonds or interest
rate swaps used as hedges.
"""
[docs] def update(self, date, data=None, inow=None):
"""
Update security with a given date and optionally, some data.
This will update price, value, weight, etc.
"""
super(HedgeSecurity, self).update(date, data, inow)
self._notl_value = 0.0
self._notl_values.values.fill(0.0)
[docs]class CouponPayingHedgeSecurity(CouponPayingSecurity):
"""
CouponPayingHedgeSecurity is a CouponPayingSecurity where the notional value is set to zero, and thus
does not count towards the notional value of the strategy. It is intended for use
in fixed income strategies.
For example in a corporate bond strategy, the notional value might refer to the size
of the corporate bond portfolio, and exclude the notional of treasury bonds or interest
rate swaps used as hedges.
"""
[docs] def update(self, date, data=None, inow=None):
"""
Update security with a given date and optionally, some data.
This will update price, value, weight, etc.
"""
super(CouponPayingHedgeSecurity, self).update(date, data, inow)
self._notl_value = 0.0
self._notl_values.values.fill(0.0)
[docs]class Algo(object):
"""
Algos are used to modularize strategy logic so that strategy logic becomes
modular, composable, more testable and less error prone. Basically, the
Algo should follow the unix philosophy - do one thing well.
In practice, algos are simply a function that receives one argument, the
Strategy (referred to as target) and are expected to return a bool.
When some state preservation is necessary between calls, the Algo
object can be used (this object). The __call___ method should be
implemented and logic defined therein to mimic a function call. A
simple function may also be used if no state preservation is necessary.
Args:
* name (str): Algo name
"""
def __init__(self, name=None):
self._name = name
@property
def name(self):
"""
Algo name.
"""
if self._name is None:
self._name = self.__class__.__name__
return self._name
def __call__(self, target):
raise NotImplementedError("%s not implemented!" % self.name)
[docs]class AlgoStack(Algo):
"""
An AlgoStack derives from Algo runs multiple Algos until a
failure is encountered.
The purpose of an AlgoStack is to group a logic set of Algos together. Each
Algo in the stack is run. Execution stops if one Algo returns False.
Args:
* algos (list): List of algos.
"""
def __init__(self, *algos):
super(AlgoStack, self).__init__()
self.algos = algos
self.check_run_always = any(hasattr(x, "run_always") for x in self.algos)
def __call__(self, target):
# normal running mode
if not self.check_run_always:
for algo in self.algos:
if not algo(target):
return False
return True
# run mode when at least one algo has a run_always attribute
else:
# store result in res
# allows continuation to check for and run
# algos that have run_always set to True
res = True
for algo in self.algos:
if res:
res = algo(target)
elif hasattr(algo, "run_always"):
if algo.run_always:
algo(target)
return res
[docs]class Strategy(StrategyBase):
"""
Strategy expands on the StrategyBase and incorporates Algos.
Basically, a Strategy is built by passing in a set of algos. These algos
will be placed in an Algo stack and the run function will call the stack.
Furthermore, two class attributes are created to pass data between algos.
perm for permanent data, temp for temporary data.
Args:
* name (str): Strategy name
* algos (list): List of Algos to be passed into an AlgoStack
* children (dict, list): Children - useful when you want to create
strategies of strategies
Children can be any type of Node or str.
String values correspond to children which will be lazily created
with that name when needed.
* parent (Node): The parent Node
Attributes:
* stack (AlgoStack): The stack
* temp (dict): A dict containing temporary data - cleared on each call
to run. This can be used to pass info to other algos.
* perm (dict): Permanent data used to pass info from one algo to
another. Not cleared on each pass.
"""
def __init__(self, name, algos=None, children=None, parent=None):
super(Strategy, self).__init__(name, children=children, parent=parent)
if algos is None:
algos = []
self.stack = AlgoStack(*algos)
self.temp = {}
self.perm = {}
[docs] def run(self):
# clear out temp data
self.temp = {}
# run algo stack
self.stack(self)
# run children
for c in self._childrenv:
c.run()
[docs]class FixedIncomeStrategy(Strategy):
"""
FixedIncomeStrategy is an alias for Strategy where the fixed_income flag
is set to True.
For this type of strategy:
- capital allocations are not necessary, and initial capital is not used
- bankruptcy is disabled (and should be modeled explicitly via an Algo)
- weights are based off notional_value rather than value
- strategy price is computed from additive PNL returns
per unit of current notional_value, with a reference price of PAR.
:class:`RenormalizedFixedIncomeResult<bt.backtest.RenormalizedFixedIncomeResult>`
can be used to re-calculate the price-based performance statistics
using different normalization schemes on total pnl.
- "transact" assumes the role of "allocate", in order to buy/sell
children on a weighted notional basis
- "rebalance" adjusts notionals rather than capital allocations based
on weights
"""
def __init__(self, name, algos=None, children=None):
super(FixedIncomeStrategy, self).__init__(name, algos=algos, children=children)
self._fixed_income = True