Module engine

Classes

class Portfolio (transaction_cost: float = 10.0, transaction_relative: bool = False, debt_record: Optional[Dict[str, float]] = None)

This class will have two sub portfolios: long and short. When 'trade' is called, it will add or remove in the appropriate portfolio the security. It will also remember each trades (in a list) to recover the state at each days (For debugging purpose).

Args:

transaction_cost: The cost of a transaction (buy or sell) in $ or in %
transaction_relative: Whether the transaction cost is in percentage relative to transaction cost or fix price
debt_record: The amount of debt used to buy securities: {security: amount}. Passed by reference from broker.

Static methods

def load_state(data: dict, debt_record: Dict[str, float])

Load the state of the portfolio from a dictionary

Args:

data: The dictionary to load from
debt_record: The amount of debt used to buy securities: {security: amount}. Passed by reference from broker.

Returns:

The portfolio object

Instance variables

var long : Dict[str, Position]

Return the long positions that are not empty (amount > 0)

var short : Dict[str, Position]

Return the short positions that are not empty (amount > 0)

Methods

def empty(self) ‑> bool

To check if the portfolio is empty

Returns:

True if the portfolio is empty, False otherwise

def estimateCost(self, price: float, amount: int, sell: bool = False) ‑> float

This method can e used by external classes such as Borker to estimate the cost of a trade.

Args:

price: The price of the security
amount: The amount of security in the transaction (Trade)
sell: Whether the trade is a sell or not. If sell, we subtract the transaction cost from the worth.

Returns:

Worth of transaction - transaction cost

def getLong(self) ‑> Dict[str, Position]

To get what securities are bought long to later calculate interests (When using margin)

Returns:

Dictionary[security, amount]

def getLongProfit(self, average_buy_price: float, average_sell_price: float, qty: int, ratio_owned: float, debt: float, number_of_entry: int) ‑> Tuple[float, float]

Calculate the profit made on a long trade

Args:

average_buy_price: The price at which the security was bought
average_sell_price: The price at which the security was sold
qty: The number of shares in the trade
ratio_owned: The ratio of the security held in account that are owned by the investor. (not bought on margin)
debt: The amount of debt used to buy the security
number_of_entry: The number of time the position was entered. (Used only if transaction cost is absolute)

Returns:

The profit (positive) or loss (negative) made on the trade, relative profit in percentage

def getShort(self) ‑> Dict[str, Position]

To get what securities are sold short to later calculate interests

Returns:

Dictionary[security, amount]

def getShortProfit(self, average_sell_price: float, average_buy_price: float, qty: int) ‑> Tuple[float, float]

Calculate the profit made on a short trade

Args:

average_sell_price: The price at which the security was sold
average_buy_price: The price at which the security was bought
qty: The number of shares in the trade

Returns:

The profit (positive) or loss (negative) made on the trade, relative profit

def get_state(self) ‑> dict

Return the state of the portfolio. (Make a deep copy)

Returns:

The state of the portfolio as a dict

def get_trade_count(self, exit_only: bool = True, cutoff: Optional[datetime.datetime] = None) ‑> int

To get the number of trades made.

Args:

exit_only: If True, Only the trade that exit a position are counted

Returns:

The number of trades

def get_trade_stats(self, cutoff: Optional[datetime.datetime] = None) ‑> dict

To get the stats for each trade. (Make a deep copy)

Returns:

The dictionary of trades: { "best_trade": float, # In percentage "worst_trade": float, # In percentage "win_rate": float, # In percentage "avg_trade": float, # In percentage "max_trade_duration": float, # In days "min_trade_duration": float, # In days "avg_trade_duration": float, # In days "profit_factor": float, # Total gains / Total losses "SQN": float, # System Quality Number}

def get_trades(self)

To get the list of trades. (Make a deep copy)

Returns:

the list of rades

def split(self, ticker: str, ratio: float)

Call this method when a stock split occurs. It will update the average price and the amount of the position.

Args:

ticker: The ticker of the security
ratio: The ratio of the split. let's say it is 7:1, then the ratio is 7, 2:3, then the ratio is 2/3

Returns:

None

def trade(self, trade: Trade, *, correction: float = 0.0) ‑> float

Make a trade and add it to the portfolio.

Note

This class will handle the debt record. It will add the amount borrowed to the debt record when buying long and remove it when selling long.

Args:

trade: Can be BuyLong, SellLong, SellShort, BuyShort
correction: The correction to apply to the price of the trade. It is used to correct the price onlywhile logging the transaction. It could be used when a split happens at the next tick. If the correction is 0,it is ignored.

Returns:

The cash change in account. If negative, withdraw. If positive, deposit.

Raises:

RuntimeError: If the transaction ID is already used

def update_time_stock_idx(self, timestep_elapsed: int = 1)

Call this method to update the time_stock_idx for each position. It will be used to calculate the dividends due to the shareholders.

Args:

timestep_elapsed: The number of timestep elapsed since the last call to this method

Returns:

None

class Position (ticker: str, amount: int, long: bool, average_price: float, average_filled_time: datetime.datetime, ratio_owned: float = 1.0)

Data class holding info about a position

Static methods

def load(data: dict)

Load the object from a dictionary

Args:

data: The dictionary to load from

Returns:

The object

Methods

def dividends_got_paid(self, timestamp: datetime.datetime)

Call this method to reset last dividends date. Useful when calculating how much dividends the user should have

Args:

timestamp: The timestamp of dividend payment

Returns:

None

def export(self) ‑> dict

Export the position to a dictionary JSONable

Returns:

The object as a dictionary

def split(self, ratio: float) ‑> None

Call this method when a stock split occurs. It will update the average price and the amount.

• $new_price = old_price / ratio$
• $new_amount = old_amount * ratio$

Args:

ratio: The ratio of the split. let's say it is 7:1, then the ratio is 7, 2:3, then the ratio is 2/3

Returns:

The two new positions

def update_time_stock_idx(self, timestep_elapsed: int = 1)

Call this method to update the time_stock_idx. It will be used to calculate the dividends due to the shareholders. Call this method at the end of each day.

Args:

timestep_elapsed: The number of seconds elapsed since the last call to this method

Returns:

None

class TradeStats (trade: Trade, duration: datetime.timedelta, profit: float, rel_profit: float, ratio_owned)

This class will hold the stats for a given trade. It will be used to calculate the stats for the whole portfolio

Args:

trade: The trade object (SellLong or BuyShort)
duration: The duration of the trade
profit: The profit made on the trade
rel_profit: The profit made on the trade relative to the amount invested NOT IN PERCENTAGE
ratio_owned: The ratio of the security held in account that are owned by the investor. (not bought on margin)

Raises:

ValueError: If the trade type is not SellLong or BuyShort (Doesn't exit a position)

Static methods

def load(data: dict)

Load the object from a dictionary

Args:

data: The dictionary to load from

Returns:

The object

Methods

def export(self)

Export the object to a JSONable dictionary Note: The duration is saved in seconds

Returns:

A dictionary corresponding top the object's state