Volatility-Targeted Position Sizing from Combined Forecasts
Code pysystemtrade
Summary
This position-sizing stage converts a combined trading forecast into a subsystem position. It scales the forecast by an average position size derived from the account’s daily cash volatility target and the instrument’s volatility, then normalizes by the configured average absolute forecast. The resulting position therefore reflects both forecast direction and strength and the amount of exposure needed to meet the volatility target.
Instrument volatility is estimated by multiplying block value by daily percentage volatility, then translating that amount into the account’s base currency using an FX rate. The stage can use raw-data inputs or fall back to price data for some measurements. It also supports configured long-only instruments by setting negative positions to zero and calculates buffers for positions. The examples demonstrate output values and show small differences when raw data is omitted, but provide no performance evaluation. Results depend on the input data, configuration, and volatility estimates; the shown logic describes a sizing mechanism rather than a complete trading strategy.
Key ideas
- Combined forecasts are scaled by volatility-based position capacity and average absolute forecast.
- Instrument volatility is measured in block-value terms and converted into the account’s base currency.
- A daily cash volatility target is used to determine the average subsystem position size.
- Configured long-only instruments have negative positions clipped to zero.
- Position buffers are calculated and applied separately from the scaled position.
Tags
Full text
# positionsizing.py
```py
import pandas as pd
from syscore.dateutils import ROOT_BDAYS_INYEAR
from syscore.exceptions import missingData
from sysdata.config.configdata import Config
from sysdata.sim.sim_data import simData
from systems.buffering import (
calculate_buffers,
apply_buffers_to_position,
)
from systems.stage import SystemStage
from systems.system_cache import input, diagnostic, output
from systems.forecast_combine import ForecastCombine
from systems.rawdata import RawData
class PositionSizing(SystemStage):
"""
Stage for position sizing (take combined forecast; turn into subsystem positions)
KEY INPUTS: a) system.combForecast.get_combined_forecast(instrument_code)
found in self.get_combined_forecast
b) system.rawdata.get_daily_percentage_volatility(instrument_code)
found in self.get_price_volatility(instrument_code)
If not found, uses system.data.daily_prices to calculate
c) system.rawdata.daily_denominator_price((instrument_code)
found in self.get_instrument_sizing_data(instrument_code)
If not found, uses system.data.daily_prices
d) system.data.get_value_of_block_price_move(instrument_code)
found in self.get_instrument_sizing_data(instrument_code)
e) system.data.get_fx_for_instrument(instrument_code, base_currency)
found in self.get_fx_rate(instrument_code)
KEY OUTPUT: system.positionSize.get_subsystem_position(instrument_code)
Name: positionSize
"""
@property
def name(self):
return "positionSize"
@output()
def get_buffers_for_subsystem_position(self, instrument_code: str) -> pd.Series:
"""
Get buffers for subsystem
"""
position = self.get_subsystem_position(instrument_code)
buffer = self.get_subsystem_buffers(instrument_code)
pos_buffers = apply_buffers_to_position(position=position, buffer=buffer)
return pos_buffers
@diagnostic()
def get_subsystem_buffers(self, instrument_code: str) -> pd.Series:
position = self.get_subsystem_position(instrument_code)
vol_scalar = self.get_average_position_at_subsystem_level(instrument_code)
log = self.log
config = self.config
buffer = calculate_buffers(
instrument_code=instrument_code,
position=position,
log=log,
config=config,
vol_scalar=vol_scalar,
)
return buffer
@output()
def get_subsystem_position(self, instrument_code: str) -> pd.Series:
"""
Get scaled position (assuming for now we trade our entire capital for one instrument)
KEY OUTPUT
:param instrument_code: instrument to get values for
:type instrument_code: str
:returns: Tx1 pd.DataFrame
>>> from systems.tests.testdata import get_test_object_futures_with_comb_forecasts
>>> from systems.basesystem import System
>>> (comb, fcs, rules, rawdata, data, config)=get_test_object_futures_with_comb_forecasts()
>>> system=System([rawdata, rules, fcs, comb, PositionSizing()], data, config)
>>>
>>> system.positionSize.get_subsystem_position("EDOLLAR").tail(2)
ss_position
2015-12-10 1.811465
2015-12-11 2.544598
>>>
>>> system2=System([rawdata, rules, fcs, comb, PositionSizing()], data, config)
>>> system2.positionSize.get_subsystem_position("EDOLLAR").tail(2)
ss_position
2015-12-10 1.811465
2015-12-11 2.544598
"""
self.log.debug(
"Calculating subsystem position for %s" % instrument_code,
instrument_code=instrument_code,
)
"""
We don't allow this to be changed in config
"""
avg_abs_forecast = self.avg_abs_forecast()
vol_scalar = self.get_average_position_at_subsystem_level(instrument_code)
forecast = self.get_combined_forecast(instrument_code)
vol_scalar = vol_scalar.reindex(forecast.index, method="ffill")
subsystem_position_raw = vol_scalar * forecast / avg_abs_forecast
subsystem_position = self._apply_long_only_constraint_to_position(
position=subsystem_position_raw, instrument_code=instrument_code
)
return subsystem_position
def _apply_long_only_constraint_to_position(
self, position: pd.Series, instrument_code: str
) -> pd.Series:
instrument_long_only = self._is_instrument_long_only(instrument_code)
if instrument_long_only:
position[position < 0.0] = 0.0
return position
@diagnostic()
def _is_instrument_long_only(self, instrument_code: str) -> bool:
list_of_long_only_instruments = self._get_list_of_long_only_instruments()
return instrument_code in list_of_long_only_instruments
@diagnostic()
def _get_list_of_long_only_instruments(self) -> list:
config = self.config
long_only = config.get_element_or_default("long_only_instruments", [])
return long_only
def avg_abs_forecast(self) -> float:
return self.config.average_absolute_forecast
@property
def config(self) -> Config:
return self.parent.config
@diagnostic()
def get_average_position_at_subsystem_level(
self, instrument_code: str
) -> pd.Series:
"""
Get ratio of required volatility vs volatility of instrument in instrument's own currency
:param instrument_code: instrument to get values for
:type instrument_code: str
:returns: Tx1 pd.DataFrame
>>> from systems.tests.testdata import get_test_object_futures_with_comb_forecasts
>>> from systems.basesystem import System
>>> (comb, fcs, rules, rawdata, data, config)=get_test_object_futures_with_comb_forecasts()
>>> system=System([rawdata, rules, fcs, comb, PositionSizing()], data, config)
>>>
>>> system.positionSize.get_average_position_at_subsystem_level("EDOLLAR").tail(2)
vol_scalar
2015-12-10 11.187869
2015-12-11 10.332930
>>>
>>> ## without raw data
>>> system2=System([ rules, fcs, comb, PositionSizing()], data, config)
>>> system2.positionSize.get_average_position_at_subsystem_level("EDOLLAR").tail(2)
vol_scalar
2015-12-10 11.180444
2015-12-11 10.344278
"""
self.log.debug(
"Calculating volatility scalar for %s" % instrument_code,
instrument_code=instrument_code,
)
instr_value_vol = self.get_instrument_value_vol(instrument_code)
cash_vol_target = self.get_daily_cash_vol_target()
vol_scalar = cash_vol_target / instr_value_vol
return vol_scalar
@diagnostic()
def get_instrument_value_vol(self, instrument_code: str) -> pd.Series:
"""
Get value of volatility of instrument in base currency (used for account value)
:param instrument_code: instrument to get values for
:type instrument_code: str
:returns: Tx1 pd.DataFrame
>>> from systems.tests.testdata import get_test_object_futures_with_comb_forecasts
>>> from systems.basesystem import System
>>> (comb, fcs, rules, rawdata, data, config)=get_test_object_futures_with_comb_forecasts()
>>> system=System([rawdata, rules, fcs, comb, PositionSizing()], data, config)
>>>
>>> system.positionSize.get_instrument_value_vol("EDOLLAR").tail(2)
ivv
2015-12-10 89.382530
2015-12-11 96.777975
>>>
>>> system2=System([rawdata, rules, fcs, comb, PositionSizing()], data, config)
>>> system2.positionSize.get_instrument_value_vol("EDOLLAR").tail(2)
ivv
2015-12-10 89.382530
2015-12-11 96.777975
"""
self.log.debug(
"Calculating instrument value vol for %s" % instrument_code,
instrument_code=instrument_code,
)
instr_ccy_vol = self.get_instrument_currency_vol(instrument_code)
fx_rate = self.get_fx_rate(instrument_code)
fx_rate = fx_rate.reindex(instr_ccy_vol.index, method="ffill")
instr_value_vol = instr_ccy_vol.ffill() * fx_rate
return instr_value_vol
@diagnostic()
def get_instrument_currency_vol(self, instrument_code: str) -> pd.Series:
"""
Get value of volatility of instrument in instrument's own currency
:param instrument_code: instrument to get values for
:type instrument_code: str
:returns: Tx1 pd.DataFrame
>>> from systems.tests.testdata import get_test_object_futures_with_comb_forecasts
>>> from systems.basesystem import System
>>> (comb, fcs, rules, rawdata, data, config)=get_test_object_futures_with_comb_forecasts()
>>> system=System([rawdata, rules, fcs, comb, PositionSizing()], data, config)
>>>
>>> system.positionSize.get_instrument_currency_vol("EDOLLAR").tail(2)
icv
2015-12-10 135.272415
2015-12-11 146.464756
>>>
>>> system2=System([ rules, fcs, comb, PositionSizing()], data, config)
>>> system2.positionSize.get_instrument_currency_vol("EDOLLAR").tail(2)
icv
2015-12-10 135.362246
2015-12-11 146.304072
"""
self.log.debug(
"Calculating instrument currency vol for %s" % instrument_code,
instrument_code=instrument_code,
)
block_value = self.get_block_value(instrument_code)
daily_perc_vol = self.get_price_volatility(instrument_code)
## FIXME WHY NOT RESAMPLE?
(block_value, daily_perc_vol) = block_value.align(daily_perc_vol, join="inner")
instr_ccy_vol = block_value.ffill() * daily_perc_vol
return instr_ccy_vol
@diagnostic()
def get_block_value(self, instrument_code: str) -> pd.Series:
"""
Calculate block value for instrument_code
:param instrument_code: instrument to get values for
:type instrument_code: str
:returns: Tx1 pd.DataFrame
>>> from systems.tests.testdata import get_test_object_futures_with_comb_forecasts
>>> from systems.basesystem import System
>>> (comb, fcs, rules, rawdata, data, config)=get_test_object_futures_with_comb_forecasts()
>>> system=System([rawdata, rules, fcs, comb, PositionSizing()], data, config)
>>>
>>> system.positionSize.get_block_value("EDOLLAR").tail(2)
bvalue
2015-12-10 2447.0000
2015-12-11 2449.6875
>>>
>>> system=System([rules, fcs, comb, PositionSizing()], data, config)
>>> system.positionSize.get_block_value("EDOLLAR").tail(2)
bvalue
2015-12-10 2447.0000
2015-12-11 2449.6875
"""
underlying_price = self.get_underlying_price(instrument_code)
value_of_price_move = self.rawdata_stage.get_value_of_block_price_move(
instrument_code
)
block_value = underlying_price.ffill() * value_of_price_move * 0.01
return block_value
@diagnostic()
def get_underlying_price(self, instrument_code: str) -> pd.Series:
"""
Get various things from data and rawdata to calculate position sizes
KEY INPUT
:param instrument_code: instrument to get values for
:type instrument_code: str
:returns: Tx1 pd.DataFrame: underlying price [as used to work out % volatility],
>>> from systems.tests.testdata import get_test_object_futures_with_comb_forecasts
>>> from systems.basesystem import System
>>> (comb, fcs, rules, rawdata, data, config)=get_test_object_futures_with_comb_forecasts()
>>> system=System([rawdata, rules, fcs, comb, PositionSizing()], data, config)
>>>
>>> ans=system.positionSize.get_underlying_price("EDOLLAR")
>>> ans[0].tail(2)
price
2015-12-10 97.8800
2015-12-11 97.9875
>>>
>>> ans[1]
2500
>>>
>>> system=System([rules, fcs, comb, PositionSizing()], data, config)
>>>
>>> ans=system.positionSize.get_underlying_price("EDOLLAR")
>>> ans[0].tail(2)
price
2015-12-10 97.8800
2015-12-11 97.9875
>>>
>>> ans[1]
2500
"""
try:
rawdata = self.rawdata_stage
except missingData:
underlying_price = self.data.daily_prices(instrument_code)
else:
underlying_price = rawdata.daily_denominator_price(instrument_code)
return underlying_price
@property
def rawdata_stage(self) -> RawData:
try:
rawdata_stage = getattr(self.parent, "rawdata")
except AttributeError as e:
raise missingData from e
return rawdata_stage
@property
def data(self) -> simData:
return self.parent.data
@input
def get_price_volatility(self, instrument_code: str) -> pd.Series:
"""
Get the daily % volatility; If a rawdata stage exists from there; otherwise work it out
:param instrument_code: instrument to get values for
:type instrument_code: str
:returns: Tx1 pd.DataFrame
KEY INPUT
Note as an exception to the normal rule we cache this, as it sometimes comes from data
>>> from systems.tests.testdata import get_test_object_futures_with_comb_forecasts
>>> from systems.basesystem import System
>>> (comb, fcs, rules, rawdata, data, config)=get_test_object_futures_with_comb_forecasts()
>>> system=System([rawdata, rules, fcs, comb, PositionSizing()], data, config)
>>>
>>> system.positionSize.get_price_volatility("EDOLLAR").tail(2)
vol
2015-12-10 0.055281
2015-12-11 0.059789
>>>
>>> system2=System([ rules, fcs, comb, PositionSizing()], data, config)
>>>
>>> system2.positionSize.get_price_volatility("EDOLLAR").tail(2)
vol
2015-12-10 0.055318
2015-12-11 0.059724
"""
daily_perc_vol = self.rawdata_stage.get_daily_percentage_volatility(
instrument_code
)
return daily_perc_vol
@diagnostic()
def get_vol_target_dict(self) -> dict:
# FIXME UGLY REPLACE WITH COMPONENTS
"""
Get the daily cash vol target
Requires: percentage_vol_target, notional_trading_capital, base_currency
To find these, look in (a) in system.config.parameters...
(b).... if not found, in systems.get_defaults.py
:Returns: tuple (str, float): str is base_currency, float is value
>>> from systems.tests.testdata import get_test_object_futures_with_comb_forecasts
>>> from systems.basesystem import System
>>> (comb, fcs, rules, rawdata, data, config)=get_test_object_futures_with_comb_forecasts()
>>> system=System([rawdata, rules, fcs, comb, PositionSizing()], data, config)
>>>
>>> ## from config
>>> system.positionSize.get_vol_target_dict()['base_currency']
'GBP'
>>>
>>> ## from defaults
>>> del(config.base_currency)
>>> system=System([rawdata, rules, fcs, comb, PositionSizing()], data, config)
>>> system.positionSize.get_vol_target_dict()['base_currency']
'USD'
>>>
"""
self.log.debug("Getting vol target")
percentage_vol_target = self.get_percentage_vol_target()
notional_trading_capital = self.get_notional_trading_capital()
base_currency = self.get_base_currency()
annual_cash_vol_target = self.annual_cash_vol_target()
daily_cash_vol_target = self.get_daily_cash_vol_target()
vol_target_dict = dict(
base_currency=base_currency,
percentage_vol_target=percentage_vol_target,
notional_trading_capital=notional_trading_capital,
annual_cash_vol_target=annual_cash_vol_target,
daily_cash_vol_target=daily_cash_vol_target,
)
return vol_target_dict
@diagnostic()
def get_daily_cash_vol_target(self) -> float:
annual_cash_vol_target = self.annual_cash_vol_target()
daily_cash_vol_target = annual_cash_vol_target / ROOT_BDAYS_INYEAR
return daily_cash_vol_target
@diagnostic()
def annual_cash_vol_target(self) -> float:
notional_trading_capital = self.get_notional_trading_capital()
percentage_vol_target = self.get_percentage_vol_target()
annual_cash_vol_target = (
notional_trading_capital * percentage_vol_target / 100.0
)
return annual_cash_vol_target
@diagnostic()
def get_notional_trading_capital(self) -> float:
notional_trading_capital = float(self.config.notional_trading_capital)
return notional_trading_capital
@diagnostic()
def get_percentage_vol_target(self):
return float(self.config.percentage_vol_target)
@diagnostic()
def get_base_currency(self) -> str:
base_currency = self.config.base_currency
return base_currency
@input
def get_fx_rate(self, instrument_code: str) -> pd.Series:
"""
Get FX rate to translate instrument volatility into same currency as account value.
KEY INPUT
:param instrument_code: instrument to get values for
:type instrument_code: str
:returns: Tx1 pd.DataFrame: fx rate
>>> from systems.tests.testdata import get_test_object_futures_with_comb_forecasts
>>> from systems.basesystem import System
>>> (comb, fcs, rules, rawdata, data, config)=get_test_object_futures_with_comb_forecasts()
>>> system=System([rawdata, rules, fcs, comb, PositionSizing()], data, config)
>>>
>>> system.positionSize.get_fx_rate("EDOLLAR").tail(2)
fx
2015-12-09 0.664311
2015-12-10 0.660759
"""
base_currency = self.get_base_currency()
fx_rate = self.rawdata_stage.get_fx_for_instrument(
instrument_code, base_currency
)
return fx_rate
@input
def get_combined_forecast(self, instrument_code: str) -> pd.Series:
"""
Get the combined forecast from previous module
:param instrument_code: instrument to get values for
:type instrument_code: str
:returns: Tx1 pd.DataFrame
KEY INPUT
>>> from systems.tests.testdata import get_test_object_futures_with_comb_forecasts
>>> from systems.basesystem import System
>>> (comb, fcs, rules, rawdata, data, config)=get_test_object_futures_with_comb_forecasts()
>>> system=System([rawdata, rules, fcs, comb, PositionSizing()], data, config)
>>>
>>> system.positionSize.get_combined_forecast("EDOLLAR").tail(2)
comb_forecast
2015-12-10 1.619134
2015-12-11 2.462610
"""
return self.comb_forecast_stage.get_combined_forecast(instrument_code)
@property
def comb_forecast_stage(self) -> ForecastCombine:
return self.parent.combForecast
if __name__ == "__main__":
import doctest
doctest.testmod()
```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.