Source code for qf_lib.backtesting.execution_handler.slippage.square_root_market_impact_slippage

#     Copyright 2016-present CERN – European Organization for Nuclear Research
#
#     Licensed under the Apache License, Version 2.0 (the "License");
#     you may not use this file except in compliance with the License.
#     You may obtain a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#     Unless required by applicable law or agreed to in writing, software
#     distributed under the License is distributed on an "AS IS" BASIS,
#     WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#     See the License for the specific language governing permissions and
#     limitations under the License.
from datetime import datetime
from typing import Sequence, Optional

import numpy as np

from qf_lib.backtesting.execution_handler.slippage.base import Slippage
from qf_lib.backtesting.order.order import Order
from qf_lib.common.enums.frequency import Frequency
from qf_lib.common.enums.price_field import PriceField
from qf_lib.common.tickers.tickers import Ticker
from qf_lib.common.utils.dateutils.relative_delta import RelativeDelta
from qf_lib.common.utils.volatility.get_volatility import get_volatility
from qf_lib.containers.series.prices_series import PricesSeries
from qf_lib.data_providers.data_provider import DataProvider
from qf_lib.data_providers.helpers import cast_data_array_to_proper_type


[docs]class SquareRootMarketImpactSlippage(Slippage): """ Slippage based on the square-root formula for market impact modelling. The price slippage is calculated by multiplying no-slippage-price by (1 + market impact), where the market impact is defined as the product of volatility, square of the volume and volatility ratio(volume traded in bar / average daily volume) and a constant value (price_impact). The direction of the slippage is always making the price worse for the trader (it increases the price when buying and decreases when selling). Parameters ---------- price_impact: float factor which implies how big will be the slippage data_provider: DataProvider DataProvider component max_volume_share_limit: float, None number from range [0,1] which denotes how big (volume-wise) the Order can be i.e. if it's 0.5 and a daily volume for a given asset is 1,000,000 USD, then max volume of the Order can be 500,000 USD. If not provided, no volume checks are performed. """ def __init__(self, price_impact: float, data_provider: DataProvider, max_volume_share_limit: Optional[float] = None): super().__init__(data_provider, max_volume_share_limit) self.price_impact = price_impact self._number_of_samples = 20 def _get_fill_prices(self, date: datetime, orders: Sequence[Order], no_slippage_fill_prices: Sequence[float], fill_volumes: Sequence[float]) -> Sequence[float]: no_slippage_fill_prices = np.array(no_slippage_fill_prices) tickers = [order.ticker for order in orders] market_impact_values = self._compute_market_impact(date, tickers, fill_volumes) fill_prices = no_slippage_fill_prices * np.add(market_impact_values, 1.0) return fill_prices def _compute_market_impact(self, date: datetime, tickers: Sequence[Ticker], fill_volumes: Sequence[float]) \ -> Sequence[float]: """ Market Impact is positive for buys and negative for sells. MI = +/- price_impact * volatility * sqrt (fill volume / average daily volume) """ start_date = date - RelativeDelta(days=60) end_date = date - RelativeDelta(days=1) # Download close price and volume values data_array = self._data_provider.get_price(tickers, [PriceField.Close, PriceField.Volume], start_date, end_date, Frequency.DAILY) close_prices = cast_data_array_to_proper_type(data_array.loc[:, tickers, PriceField.Close]) volumes = cast_data_array_to_proper_type(data_array.loc[:, tickers, PriceField.Volume]) close_prices_volatility = close_prices.apply(self._compute_volatility).values average_volumes = volumes.apply(self._compute_average_volume).values abs_fill_volumes = np.abs(fill_volumes) volatility_volume_ratio = np.divide(abs_fill_volumes, average_volumes) sqrt_volatility_volume_ratio = np.sqrt(volatility_volume_ratio) * np.sign(fill_volumes) return self.price_impact * close_prices_volatility * sqrt_volatility_volume_ratio def _compute_volatility(self, prices_tms) -> float: """Compute the annualised volatility of the last self._number_of_samples days""" prices_tms = prices_tms.dropna().iloc[-self._number_of_samples:] prices_tms = PricesSeries(prices_tms) try: volatility = get_volatility(prices_tms, frequency=Frequency.DAILY, annualise=True) except (AssertionError, AttributeError): volatility = float('nan') return volatility def _compute_average_volume(self, volume_tms) -> float: """Compute the average volume of the last self._number_of_samples days""" volume_tms = volume_tms.dropna().iloc[-self._number_of_samples:] volume_tms = volume_tms[volume_tms >= 0] adv = volume_tms.mean() return adv