Source code for tradeexecutor.strategy.qstrader.order_sizer

"""Convert portfolio weightings to US dollar sized buy and sell trades."""

import logging
from decimal import Decimal
from typing import Dict, Tuple

import pandas as pd
import numpy as np

from qstrader.portcon.order_sizer.order_sizer import OrderSizer
from tradeexecutor.state.state import State
from tradeexecutor.strategy.pricing_model import PricingModel

logger = logging.getLogger(__name__)


[docs]class CashBufferedOrderSizer(OrderSizer): """ Creates a target portfolio of quantities for each Asset using its provided weight and total equity available in the Broker portfolio. Includes an optional cash buffer due to the non-fractional amount of share/unit sizes. The cash buffer defaults to 5% of the total equity, but can be modified. Parameters ---------- cash_buffer_percentage : `float`, optional The percentage of the portfolio equity to retain in cash to avoid generating Orders that exceed account equity (assuming no margin available). """
[docs] def __init__( self, state: State, pricing_model: PricingModel, cash_buffer_percentage=0.05 ): self.state = state self.pricing_model = pricing_model self.cash_buffer_percentage = self._check_set_cash_buffer(cash_buffer_percentage)
def _check_set_cash_buffer(self, cash_buffer_percentage): """ Checks and sets the cash buffer percentage value. Parameters ---------- cash_buffer_percentage : `float` The percentage of the portfolio equity to retain in cash to avoid generating Orders that exceed account equity (assuming no margin available). Returns ------- `float` The cash buffer percentage value. """ if ( cash_buffer_percentage < 0.0 or cash_buffer_percentage > 1.0 ): raise ValueError( 'Cash buffer percentage "%s" provided to dollar-weighted ' 'execution algorithm is negative or ' 'exceeds 100%.' % cash_buffer_percentage ) else: return cash_buffer_percentage def _normalise_weights(self, weights): """ Rescale provided weight values to ensure weight vector sums to unity. Parameters ---------- weights : `dict{Asset: float}` The un-normalised weight vector. Returns ------- `dict{Asset: float}` The unit sum weight vector. """ if any([weight < 0.0 for weight in weights.values()]): raise ValueError( 'Dollar-weighted cash-buffered order sizing does not support ' 'negative weights. All positions must be long-only.' ) weight_sum = sum(weight for weight in weights.values()) # If the weights are very close or equal to zero then rescaling # is not possible, so simply return weights unscaled if np.isclose(weight_sum, 0.0): return weights return { asset: (weight / weight_sum) for asset, weight in weights.items() }
[docs] def __call__(self, dt: pd.Timestamp, weights: Dict[int, float], debug_details: dict) -> Tuple[Dict, Dict]: """ Creates a dollar-weighted cash-buffered target portfolio from the provided target weights at a particular timestamp. Parameters ---------- dt : `pd.Timestamp` The current date-time timestamp. weights : `dict{Asset: float}` The (potentially unnormalised) target weights. """ assert isinstance(debug_details, dict) total_equity = self.state.portfolio.get_total_equity() cash_buffered_total_equity = total_equity * ( 1.0 - self.cash_buffer_percentage ) debug_details["cash_buffer_percentage"] = self.cash_buffer_percentage debug_details["cash_buffered_total_equity"] = cash_buffered_total_equity # logger.trade(f"Calculating US dollar weights for the new portfolio. Total portfolio equity is {total_equity:,.2f} USD, the cash buffered total equity {cash_buffered_total_equity:,.2f} USD") assert cash_buffered_total_equity > 0, "No cash or token holdings" # Pre-cost dollar weight N = len(weights) if N == 0: # No forecasts so portfolio remains in cash # or is fully liquidated return {}, {} # Ensure weight vector sums to unity normalised_weights = self._normalise_weights(weights) # Expose internals to unit testing debug_details["normalised_weights"] = normalised_weights target_portfolio = {} target_prices = {} total_spend = 0 for asset, weight in sorted(normalised_weights.items()): pre_cost_dollar_weight = cash_buffered_total_equity * weight # Estimate broker fees for this asset est_quantity = 0 # TODO: Needs to be added for IB est_costs = 0 # self.broker.fee_model.calc_total_cost(asset, est_quantity, pre_cost_dollar_weight, broker=self.broker) # Calculate integral target asset quantity assuming broker costs after_cost_dollar_weight = pre_cost_dollar_weight - est_costs asset_quantity = 0 if weight > 0: trading_pair = self.pricing_model.get_pair_for_id(asset) price_structure = self.pricing_model.get_buy_price(dt, trading_pair, Decimal(after_cost_dollar_weight)) asset_price = price_structure.price if asset_price is not None: if after_cost_dollar_weight > 0: if np.isnan(asset_price): raise ValueError( 'Asset price for "%s" at timestamp "%s" is Not-a-Number (NaN). ' 'This can occur if the chosen backtest start date is earlier ' 'than the first available price for a particular asset. Try ' 'modifying the backtest start date and re-running.' % (asset, dt) ) asset_quantity = after_cost_dollar_weight / asset_price asset_quantity = self.pricing_model.quantize_base_quantity(trading_pair, asset_quantity) # Add to the target portfolio target_portfolio[asset] = {"quantity": asset_quantity} target_prices[asset] = asset_price total_spend += (asset_quantity * Decimal(asset_price)) + est_costs else: logger.warning("Skipping asset %s because of the price issue", asset) logger.info(f"Total new portfolio cost {total_spend:,.2f}") return target_portfolio, target_prices