Position Buffer Methods for Portfolio Trading Systems
Summary
This code describes position buffers used in a trading system’s position sizing and portfolio processes. It supports three configured methods: forecast-based buffers, position-based buffers, and a nominal small buffer when buffering is disabled or an unrecognized method is selected. It also includes a helper that scales existing buffers by a capital multiplier, forward-filling values to align dates.
The forecast method sets buffer size as a configured fraction of an average position proxy, calculated from the absolute product of a volatility scalar, instrument weight, and instrument diversification multiplier. Missing weights and diversification multipliers default to one, and input series are forward-filled onto the position index. The position method instead takes the absolute position times the configured buffer fraction. A separate helper adds and subtracts the buffer from a position to produce upper and lower position thresholds.
The excerpt specifies implementation mechanics but gives no parameter guidance, trading rationale, performance evidence, or tests. Its practical behavior depends on the surrounding system’s configuration and on how downstream components use the thresholds.
Key ideas
- The code offers forecast-based, position-based, and effectively unbuffered position buffer methods.
- Forecast buffers scale a volatility scalar by instrument weight and a diversification multiplier before applying the configured buffer fraction.
- Position-based buffers use the absolute position multiplied by the configured buffer fraction.
- Buffer series and scaling inputs are aligned with forward filling, while missing weight and diversification inputs default to one.
- A helper forms upper and lower position thresholds by adding and subtracting the buffer.
Tags
Full text
# buffering.py
```py
## Buffer class used in both position sizing and portfolio
import pandas as pd
from sysdata.config.configdata import Config
from syslogging.logger import *
from syscore.constants import arg_not_supplied
def calculate_actual_buffers(
buffers: pd.DataFrame, cap_multiplier: pd.Series
) -> pd.DataFrame:
"""
Used when rescaling capital for accumulation
"""
cap_multiplier = cap_multiplier.reindex(buffers.index).ffill()
cap_multiplier = pd.concat([cap_multiplier, cap_multiplier], axis=1)
cap_multiplier.columns = buffers.columns
actual_buffers_for_position = buffers * cap_multiplier
return actual_buffers_for_position
def apply_buffers_to_position(position: pd.Series, buffer: pd.Series) -> pd.DataFrame:
top_position = position.ffill() + buffer.ffill()
bottom_position = position.ffill() - buffer.ffill()
pos_buffers = pd.concat([top_position, bottom_position], axis=1)
pos_buffers.columns = ["top_pos", "bot_pos"]
return pos_buffers
def calculate_buffers(
instrument_code: str,
position: pd.Series,
config: Config,
vol_scalar: pd.Series,
instr_weights: pd.DataFrame = arg_not_supplied,
idm: pd.Series = arg_not_supplied,
log=get_logger(""),
) -> pd.Series:
log.debug(
"Calculating buffers for %s" % instrument_code,
instrument_code=instrument_code,
)
buffer_method = config.buffer_method
if buffer_method == "forecast":
log.debug(
"Calculating forecast method buffers for %s" % instrument_code,
instrument_code=instrument_code,
)
if instr_weights is arg_not_supplied:
instr_weight_this_code = arg_not_supplied
else:
instr_weight_this_code = instr_weights[instrument_code]
buffer = get_forecast_method_buffer(
instr_weight_this_code=instr_weight_this_code,
vol_scalar=vol_scalar,
idm=idm,
position=position,
config=config,
)
elif buffer_method == "position":
log.debug(
"Calculating position method buffer for %s" % instrument_code,
instrument_code=instrument_code,
)
buffer = get_position_method_buffer(config=config, position=position)
elif buffer_method == "none":
log.debug(
"None method, no buffering for %s" % instrument_code,
instrument_code=instrument_code,
)
buffer = get_buffer_if_not_buffering(position=position)
else:
log.critical("Buffer method %s not recognised - not buffering" % buffer_method)
buffer = get_buffer_if_not_buffering(position=position)
return buffer
def get_forecast_method_buffer(
position: pd.Series,
vol_scalar: pd.Series,
config: Config,
instr_weight_this_code: pd.Series = arg_not_supplied,
idm: pd.Series = arg_not_supplied,
) -> pd.Series:
"""
Gets the buffers for positions, using proportion of average forecast method
:param instrument_code: instrument to get values for
:type instrument_code: str
:returns: Tx1 pd.DataFrame
"""
buffer_size = config.buffer_size
buffer = _calculate_forecast_buffer_method(
buffer_size=buffer_size,
position=position,
idm=idm,
instr_weight_this_code=instr_weight_this_code,
vol_scalar=vol_scalar,
)
return buffer
def get_position_method_buffer(
position: pd.Series,
config: Config,
) -> pd.Series:
"""
Gets the buffers for positions, using proportion of position method
"""
buffer_size = config.buffer_size
abs_position = abs(position)
buffer = abs_position * buffer_size
buffer.columns = ["buffer"]
return buffer
def get_buffer_if_not_buffering(position: pd.Series) -> pd.Series:
EPSILON_POSITION = 0.001
buffer = pd.Series([EPSILON_POSITION] * position.shape[0], index=position.index)
return buffer
def _calculate_forecast_buffer_method(
position: pd.Series,
buffer_size: float,
vol_scalar: pd.Series,
idm: pd.Series = arg_not_supplied,
instr_weight_this_code: pd.Series = arg_not_supplied,
):
if instr_weight_this_code is arg_not_supplied:
instr_weight_this_code_indexed = 1.0
else:
instr_weight_this_code_indexed = instr_weight_this_code.reindex(
position.index
).ffill()
if idm is arg_not_supplied:
idm_indexed = 1.0
else:
idm_indexed = idm.reindex(position.index).ffill()
vol_scalar_indexed = vol_scalar.reindex(position.index).ffill()
average_position = abs(
vol_scalar_indexed * instr_weight_this_code_indexed * idm_indexed
)
buffer = average_position * buffer_size
return buffer
```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.