Skip to content
All library documents

Volatility Quantiles for Scaling Trading Forecasts

Code pysystemtrade

Summary

This code describes a volatility-sensitive adjustment to trading forecasts. It calculates daily percentage volatility, compares it with a rolling ten-year average, and converts the normalized volatility observations into quantile ranks. A multiplier decreases as the quantile rises, so forecasts are attenuated when current volatility is high relative to its history. The attenuation series is smoothed with an exponential moving average before being aligned to forecast dates.

The adjustment is applied only to configured rule variations; other forecasts receive a neutral multiplier. The document provides implementation logic but no backtest, performance evidence, or guidance for choosing the multiplier curve. Its rolling baseline requires sufficient historical observations, and the code notes that the calculation may be slow. The approach is therefore a forecast-scaling technique, not evidence that volatility attenuation improves trading results.

Key ideas

  • The method ranks current daily volatility against a long rolling average to estimate its relative level.
  • Higher volatility quantiles receive lower forecast multipliers under the stated linear mapping.
  • An exponential moving average smooths the multiplier series before application.
  • Attenuation is enabled only for forecast rules named in the configuration.
  • The document provides code but no empirical evaluation of trading outcomes.

Tags

Full text
# vol_attenuation_forecast_scale_cap.py


```py
from syscore.pandas.strategy_functions import quantile_of_points_in_data_series
from syscore.pandas.pdutils import from_scalar_values_to_ts
from systems.forecast_scale_cap import *


class volAttenForecastScaleCap(ForecastScaleCap):
    @diagnostic()
    def get_vol_quantile_points(self, instrument_code):
        ## More properly this would go in raw data perhaps
        self.log.debug("Calculating vol quantile for %s" % instrument_code)
        daily_vol = self.parent.rawdata.get_daily_percentage_volatility(instrument_code)
        ten_year_vol = daily_vol.rolling(2500, min_periods=10).mean()
        normalised_vol = daily_vol / ten_year_vol

        normalised_vol_q = quantile_of_points_in_data_series(normalised_vol)

        return normalised_vol_q

    @diagnostic()
    def get_vol_attenuation(self, instrument_code):
        normalised_vol_q = self.get_vol_quantile_points(instrument_code)
        vol_attenuation = normalised_vol_q.apply(multiplier_function)

        smoothed_vol_attenuation = vol_attenuation.ewm(span=10).mean()

        return smoothed_vol_attenuation

    @input
    def get_raw_forecast_before_attenuation(self, instrument_code, rule_variation_name):
        ## original code for get_raw_forecast
        raw_forecast = self.parent.rules.get_raw_forecast(
            instrument_code, rule_variation_name
        )

        return raw_forecast

    @diagnostic()
    def get_raw_forecast(self, instrument_code, rule_variation_name):
        ## overridden method this will be called downstream so don't change name
        raw_forecast_before_atten = self.get_raw_forecast_before_attenuation(
            instrument_code, rule_variation_name
        )
        vol_attenutation_reindex = (
            self.get_attenuation_for_rule_and_instrument_indexed_to_forecast(
                instrument_code=instrument_code, rule_variation_name=rule_variation_name
            )
        )

        attenuated_forecast = raw_forecast_before_atten * vol_attenutation_reindex

        return attenuated_forecast

    @diagnostic()
    def get_attenuation_for_rule_and_instrument_indexed_to_forecast(
        self, instrument_code, rule_variation_name
    ) -> pd.Series:
        raw_forecast_before_atten = self.get_raw_forecast_before_attenuation(
            instrument_code, rule_variation_name
        )

        use_attenuation = self.config.get_element_or_default("use_attenuation", [])

        if rule_variation_name not in use_attenuation:
            forecast_ts = raw_forecast_before_atten.index
            return from_scalar_values_to_ts(1.0, forecast_ts)

        vol_attenutation = self.get_vol_attenuation(instrument_code)
        vol_attenutation_reindex = vol_attenutation.reindex(
            raw_forecast_before_atten.index, method="ffill"
        )

        return vol_attenutation_reindex


# this is a little slow so suggestions for speeding up are welcome


def multiplier_function(vol_quantile):
    if np.isnan(vol_quantile):
        return 1.0

    return 2 - 1.5 * vol_quantile

```

Shown in full with attribution under the source's licence. Licence: GPL-3.0

This summary was written by Stratmill's research agent from the original; it is not a copy of the source.