Module engine

Classes

class Broker (bank_account: Account, commission: float = None, relative_commission: float = None, margin_interest: float = 0, min_initial_margin: float = 0.5, min_maintenance_margin: float = 0.25, liquidation_delay: int = 2, min_initial_margin_short: float = 0.5, min_maintenance_margin_short: float = 0.25)

When using margin, you need to add a column in the data called 'marginable' that is bool corresponding if a stock can have margin or not. Note: If there is not enough cash in the account when monthly fees are deduced, the strategy fills a bankruptcy, and the simulation stops.

Args:

bank_account: The bank account holding cash and collateral with the initial cash set.
commission: The commission to be paid for each trade. This parameter cannot be set if relative_commission is set. And vice versa. If commission and relative_commission are both None, the commission is set to 0.
relative_commission: The commission to be paid for each trade as a ratio (0-1) of the trade value. This parameter cannot be set if commission is set. And vice versa.
margin_interest: The margin interest rate. It is the interest rate charged on borrowed money. It is calculated daily and charged monthly.
min_initial_margin: The initial margin required to buy a stock on margin. It is a ratio (0-1) of the stock's value. Example: 50% initial margin is 0.5
min_maintenance_margin: The minimum margin required to keep a stock on margin before a margin call. It is a ratio (0-1) of the stock's value. Example: 25% maintenance margin is 0.25
liquidation_delay: The time a margin call has to be active before the broker liquidate positions to pay the margin call. (Must be grater than 0, and an int)
min_initial_margin_short: The minimum initial margin required to sell a stock short. It is a ratio (0-1)
min_maintenance_margin_short: The minimum maintenance margin required to keep a stock short before a margin call. It is a ratio (0-1) of the stock's value.

Raises:

ValueError: If both commission and relative_commission are set.
ValueError: If relative_commission is not None and not between 0 and 1
ValueError: If commission is not None and smaller than 0
ValueError: If margin_interest is not between 0 and 1
ValueError: min_initial_margin and min_maintenance_margin are not between 0 and 1
ValueError: If min_maintenance_margin_short and min_initial_margin_short are not between 0 and 1
ValueError: If min_maintenance_margin is greater than or equal to min_initial_margin
ValueError: If min_maintenance_margin_short is greater than or equal to min_initial_margin_short

Static methods

def compute_dividend_payout(position: Position, div_freq: DividendFrequency, dividends: float) ‑> float

Compute the dividend payout of a long position

Args:

position: The position to calculate dividend payout
div_freq: The frequency that the security is paying dividends.
dividends: The actual dividends payout for 1 share and for the period in div_freq.

Returns:

Dividend payout

Raises:

RuntimeError: If the position is not long

def load_state(data: dict, account: Account)

Load a broker from a state dictionary

Args:

data: The state dictionary
account: The account to which the broker is linked (Loaded by the BacktestResult object)

Returns:

The broker

Instance variables

var pending_orders : List[TradeOrder]

Return a copy of the pending orders

Methods

def buy_long(self, ticker: str, amount: int, amount_borrowed: int = 0, expiry: Optional[datetime.datetime] = None, price_limit: Tuple[Optional[float], Optional[float]] = (None, None))

The strategy calls this method to open a long position. The position can be bought on margin.

Args:

ticker: The ticker of the security to buy
amount: The amount of shares to buy
amount_borrowed: The amount of shares to buy on margin. (Won't be charged in the account)
expiry: The expiry date of the order. If the order is not executed before the expiry date, it is automatically cancelled. If the expiry date is None, the order is never cancelled.
price_limit: The price limit of the order. The first value of the tuple is the maximum price to buy.Meaning the order will be executed if the price is lower or equal to the limit. The second value is the minimumprice to buy. Meaning the order will be executed if the price is higher or equal to the limit. If the limit is(None, None) the order will be executed at the next open price.

Returns:

None

def buy_short(self, ticker: str, amount_borrowed: int = 0, expiry: Optional[datetime.datetime] = None, price_limit: Tuple[Optional[float], Optional[float]] = (None, None))

The strategy calls this method to close a short position.

Args:

ticker: The ticker of the security to buy
amount_borrowed: The amount of shares to buy back.
expiry: The expiry date of the order. If the order is not executed before the expiry date, it is automatically cancelled. If the expiry date is None, the order is never cancelled.
price_limit: The price limit of the order. The first value of the tuple is the maximum price to buy.Meaning the order will be executed if the price is lower or equal to the limit. The second value is the minimumprice to buy. Meaning the order will be executed if the price is higher or equal to the limit. If the limit is(None, None) the order will be executed at the next open price.

Returns:

None

def get_state(self) ‑> dict

Get the current state of the broker as a JSONable dictionary (deepcopy

Note

• It doesn't include the account's state.

Returns:

The broker's state as a dictionary

def get_worth(self, security_names: List[str], current_tick_data: numpy.ndarray) ‑> float

Get the worth of the portfolio. (Doesn't include the transaction cost as if we would sell the positions)

Formula

worth = cash + sum(long_positions) - sum(short_positions)

where: long_positions = sum(long_position_amount * close_price - debt) short_positions = sum(short_position_amount * close_price) cash: The total cash in portfolio excluding all collateral.

Args:

security_names: A list of all securities where the index of each ticker is the index of the data of the corresponding security in the 'current_tick_data' parameter along the first axis (axis=0).
current_tick_data: An array containing prices of each security for the current step shape(n_securities, 4)

Returns:

worth in dollars

def make_trade(self, order: TradeOrder, security_price: Tuple[float, float, float, float, float], timestamp: datetime.datetime, marginable: bool, shortable: bool) ‑> bool

This method is call to make trades (convert tradeOrders to trade). Make the trade if price is within limits

Args:

order: TradeOrder
security_price: The security price (Open, High, Low, Close, Splits)
timestamp: The time where securities will be bought (Usually the next step)
marginable: If the security is marginable.
shortable: If the security is shortable.

Returns:

True if the trade was successful (Needs to remove from pending trades) And False otherwise

Raises:

RuntimeError: If the trade cannot be executed because of margin or cash issues.

def new_margin_call(self, value: float, message: str = 'missing_funds')

Create a new margin call and appends it to the other margin calls. Do not forget to remove money from account if necessary. It also add the value to the debt record.

Args:

value: The value that needs to be added to the account to cover margin call

Returns:

None

def remove_margin_call(self, key: str)

Remove a margin call from the list of margin calls. It also removed the debt in the debt record.

Args:

key: The key of the margin call to remove

Returns:

None

def sell_long(self, ticker: str, amount: int, expiry: Optional[datetime.datetime] = None, price_limit: Tuple[Optional[float], Optional[float]] = (None, None))

The strategy calls this method to close a long position.

Args:

ticker: The ticker of the security to sell
amount: The amount of shares to sell
expiry: The expiry date of the order. If the order is not executed before the expiry date, it is automatically cancelled. If the expiry date is None, the order is never cancelled.
price_limit: The price limit of the order. The first value of the tuple is the minimum price to sell(Stop loss). Meaning the order will be executed if the price is lower or equal to the limit. The second valueis the maximum price to sell. Meaning the order will be executed if the price is higher or equal to the limit.If the limit is (None, None) the order will be executed at the next open price.

Returns:

None

def sell_short(self, ticker: str, amount_borrowed: int = 0, expiry: Optional[datetime.datetime] = None, price_limit: Tuple[Optional[float], Optional[float]] = (None, None))

The strategy calls this method to open a short position.

Args:

ticker: The ticker of the security to sell
amount_borrowed: The amount of shares to sell short.
expiry: The expiry date of the order. If the order is not executed before the expiry date, it is automatically cancelled. If the expiry date is None, the order is never cancelled.
price_limit: The price limit of the order. The first value of the tuple is the minimum price to sell.Meaning the order will be executed if the price is lower or equal to the limit. The second value is the maximum price to sell. Meaning the order will be executed if the price is higher or equal to the limit.If the limit is (None, None) the order will be executed at the next open price.

Returns:

None

def tick(self, timestamp: datetime.datetime, next_timestamp, security_names: List[str], current_tick_data: numpy.ndarray, next_tick_data: numpy.ndarray, marginables: numpy.ndarray[typing.Any, numpy.dtype[bool]], dividends: numpy.ndarray[typing.Any, numpy.dtype[numpy.float32]], div_freq: List[DividendFrequency], short_rates: numpy.ndarray[typing.Any, numpy.dtype[numpy.float32]], last_tick: bool = False)

The simulation calls this method after the strategy has run. It will calculate interests and margin call if applicable. It will do trades that can be done in the trade queue at the next open.

Args:

timestamp: The date and time of the current step
next_timestamp: The date and time of the next step where orders will be evaluated/bought
security_names: A list of all securities where the index of each ticker is the index of the data of the corresponding security in the 'next_tick_data' parameter along the first axis (axis=0).
current_tick_data: An array of prices of each security for the current step(n_securities, 4) The 4 columns of the array are: Open, High, Low, Close of the next step.
next_tick_data: An array containing prices of each security for the next step shape(n_securities, 4) The 5 columns of the array are: Open, High, Low, Close and Stock Splits of the next step.
marginables: A boolean array of shape (n_securities, 2) [Marginable, Shortable) where True means that the security can be bought on margin / sold short and False means that it cannot be bought on margin / sold short.
dividends: A float array of shape (n_securities, ) containing the dividends of each security for the current step.
div_freq: The frequency that the security is paying dividends.
short_rates: The interest rates for each security that the user held overnight. Shape(n_securities, )
last_tick: Whether it is the last tick of the simulation. If it is, the broker will liquidate all positions and charge interests.

Returns:

None

class BrokerState (margin_calls: Dict[str, MarginCall], bankruptcy: bool, bankruptcy_amount: float = 0)

Data class to store message from broker to strategy

Static methods

def load(data: dict)

This method load a broker state object from a dictionary.

Args:

data: The dictionary containing the object state

Returns:

The object state as a dictionary

Methods

def export(self)

This method export the broker state object to a JSONable dictionary.

Returns:

The object state as a dictionary

class MarginCall (time_remaining: int, amount: float)

Data class to store info about margin calls

Args:

time_remaining: Time remaining in number of steps until liquidation
amount: Amount in $ that needs to be added to account to come out of margin call

Static methods

def load(data: dict)

This method load a margin call object from a dictionary.

Args:

data: The dictionary containing the object state

Returns:

The object state as a dictionary

Methods

def export(self) ‑> dict

This method export the margin call object to a JSONable dictionary.

Returns:

The object state as a dictionary

class StepState (timestamp: datetime.datetime, worth: float, pending_orders: List[TradeOrder], filled_orders: List[TradeOrder], margin_calls: Dict[str, MarginCall])

Record the state of the broker at each steps for easier strategy debugging

Args:

timestamp: The date and time of the current step
worth: The worth of the portfolio
pending_orders: The pending orders that were not executed
filled_orders: The orders that were executed at the current step
margin_calls: The margin calls active at the current step

Static methods

def load(data: dict)

This method load a step state object from a dictionary.

Args:

data: The dictionary containing the object state

Returns:

The object state as a dictionary

Methods

def export(self) ‑> dict

This method export the step state object to a JSONable dictionary.

Returns:

The object state as a dictionary