Diagnosing Forecast Scaling and Exporting Trading System Parameters
Code pysystemtrade
Summary
This Python module provides diagnostics and configuration helpers for a systematic trading system. It compares each rule’s capped forecasts and each instrument’s combined forecasts with a target average forecast magnitude, ranking the largest discrepancies first. The error measure is the larger of the absolute deviations in mean absolute forecast and standard deviation.
Other methods estimate forecast mappings, expose the latest forecast scalars and diversification multipliers, retrieve forecast and instrument weights, and assemble those values into a configuration that can be written as YAML. A calculation breakdown traces how a position relates to combined forecasts, volatility scaling, instrument value volatility, portfolio weights, diversification, and buffers. The code offers implementation-level inspection tools, not a trading strategy or empirical validation. Some estimates depend on system state and configuration, and the mapping logic handles absent positions by assigning zero; users should interpret exported settings in light of those dependencies.
Key ideas
- Forecast scaling diagnostics rank rules and instruments by deviations from a target forecast magnitude.
- The error measure considers both mean absolute forecast size and standard deviation.
- The module extracts current forecast scalars, weights, and diversification multipliers for configuration.
- Position explanations connect forecasts and volatility inputs to portfolio sizing.
- The code provides diagnostics and configuration output, not evidence of trading performance.
Tags
Full text
# diagoutput.py
```py
"""
Suite of functions to analyse a system, and produce configuration that can be saved to a yaml file
"""
from syscore.dateutils import ROOT_BDAYS_INYEAR
from systems.forecast_mapping import estimate_mapping_params
import yaml
import numpy as np
class systemDiag(object):
def __init__(self, system):
self.system = system
def instrument_list(self):
return self.system.get_instrument_list()
def trading_rules(self):
return self.system.rules.trading_rules().keys()
def target_forecast_value(self):
return self.system.config.average_absolute_forecast
def check_forecast_scaling(self):
"""
Check forecast scaling
Returns a list of tuples, ordered with largest error first
:param system:
:return: list of tuples
"""
system = self.system
instrument_list = system.get_instrument_list()
rule_list = self.trading_rules()
target_forecast_value = self.target_forecast_value()
results_list = []
for rule in rule_list:
for instrument in instrument_list:
forecast = system.forecastScaleCap.get_capped_forecast(instrument, rule)
error = forecast_error(forecast, target_forecast_value)
results_list.append((instrument, rule, error))
sorted_by_max_error = sorted(results_list, key=lambda tup: tup[2], reverse=True)
return sorted_by_max_error
def check_combined_forecast_scaling(self, forecast_type="raw"):
"""
Check combined forecast scaling
Returns a list of tuples, ordered with largest error first
:param system:
:param forecast_type: raw or final. If raw is specified will be before any forecast scaling is applied
:return: list of tuples
"""
system = self.system
instrument_list = system.get_instrument_list()
target_forecast_value = self.target_forecast_value()
try:
attr_name_dict = dict(
raw="_get_raw_combined_forecast", final="get_combined_forecast"
)
attr_name = attr_name_dict[forecast_type]
except KeyError:
raise Exception(
"forecast_type must be one of %s" % str(attr_name_dict.keys())
)
try:
forecast_func = getattr(system.combForecast, attr_name)
except BaseException:
raise Exception("%s not a method system.combForecast" % attr_name)
results_list = []
for instrument in instrument_list:
forecast = forecast_func(instrument)
error = forecast_error(forecast, target_forecast_value)
results_list.append((instrument, error))
sorted_by_max_error = sorted(results_list, key=lambda tup: tup[1], reverse=True)
return sorted_by_max_error
def forecast_mapping(self, target_position_at_avg_forecast=2.0):
"""
Fit threshold values for forecasts
:return: dict, suitable for dropping into a config object or yaml file
"""
system = self.system
instrument_list = self.instrument_list()
avg_forecast = self.target_forecast_value()
forecast_mapping = {}
for instrument in instrument_list:
position = system.portfolio.get_notional_position(instrument)
forecast = system.combForecast.get_combined_forecast(instrument)
scalar = position / forecast
scalar_ewma = scalar.ewm(500).mean()
position_at_avg_forecast = avg_forecast * scalar_ewma.values[-1]
if np.isnan(
position_at_avg_forecast
): # In case no position was open for a given instrument
position_at_avg_forecast = 0.0
a_param = 0.0
else:
a_param = target_position_at_avg_forecast / position_at_avg_forecast
print("%s avg position %.2f" % (instrument, position_at_avg_forecast))
if a_param < 1.2:
# no need to do anything
print("Forecast scaling not required for %s" % instrument)
elif a_param > 1.7:
print(
"Warning! Position at avg forecast of %.2f is too small for mapping to work for %s "
% (position_at_avg_forecast, instrument)
)
else:
(
a_param,
b_param,
threshold_value,
capped_value,
) = estimate_mapping_params(a_param)
map_dict = dict(
a_param=float(a_param),
b_param=float(b_param),
threshold=float(threshold_value),
)
forecast_mapping[instrument] = map_dict
return forecast_mapping
def forecast_scalars(self):
"""
Returns final estimated values for forecast scalars, so they can be written into a config as fixed values
:return: dict of forecast scalars
"""
system = self.system
instrument_list = self.instrument_list()
rule_list = self.trading_rules()
use_estimates = system.config.use_forecast_scale_estimates
if not use_estimates:
print("Can't output forecast scalar estimates, as they weren't estimated")
pooling = system.config.forecast_scalar_estimate["pool_instruments"]
if not pooling:
print(
"WARNING: No way of putting different forecast scalars for different instruments into config"
)
scalar_results = dict()
for rule in rule_list:
if not pooling:
scalar_results[rule] = dict()
for instrument in instrument_list:
scalar = float(
system.forecastScaleCap.get_forecast_scalar(instrument, rule).iloc[
-1
]
)
if pooling:
# will be overwritten for each instrument
scalar_results[rule] = scalar
else:
scalar_results[rule][instrument] = scalar
return scalar_results
def forecast_div_multiplier(self):
"""
Returns final estimated values for FDM, so they can be written into a config as fixed values
:return: dict
"""
system = self.system
instrument_list = self.instrument_list()
fdm_results = dict()
for instrument in instrument_list:
fdm = system.combForecast.get_forecast_diversification_multiplier(
instrument
).values[-1]
fdm_results[instrument] = float(fdm)
return fdm_results
def forecast_weights(self):
"""
Returns final estimated values for forecast weights, so they can be written into a config as fixed values
:return: dict of dicts
"""
# forecast weights
system = self.system
instrument_list = self.instrument_list()
forecast_weights = dict()
for instrument in instrument_list:
weights = dict(
system.combForecast.get_forecast_weights(instrument).iloc[-1]
)
weights = dict(
(str(rule_name), float(weight)) for rule_name, weight in weights.items()
)
forecast_weights[instrument] = weights
return forecast_weights
def instrument_weights(self):
"""
Returns final estimated values for instrument weights, so they can be written into a config as fixed values
:return: dict
"""
system = self.system
instrument_weights = system.portfolio.get_instrument_weights().iloc[-1]
instrument_weights = dict(
(str(key), float(value)) for key, value in instrument_weights.items()
)
return instrument_weights
def instrument_div_multiplier(self):
"""
Returns final estimated values for instrument diversification multiplier, so it can be written into a config as fixed values
:return: dict
"""
system = self.system
instrument_div_multiplier = float(
system.portfolio.get_instrument_diversification_multiplier().values[-1]
)
return instrument_div_multiplier
def output_config_with_estimated_parameters(
self,
attr_names=[
"forecast_scalars",
"forecast_weights",
"forecast_div_multiplier",
"forecast_mapping",
"instrument_weights",
"instrument_div_multiplier",
],
):
output_dict = {}
for config_item in attr_names:
dict_function = getattr(self, config_item)
try:
dict_value = dict_function()
output_dict[config_item] = dict_value
except BaseException:
print("Couldn't get %s will exclude from output" % config_item)
return output_dict
def yaml_config_with_estimated_parameters(
self,
yaml_filename,
attr_names=[
"forecast_scalars",
"forecast_weights",
"forecast_div_multiplier",
"forecast_mapping",
"instrument_weights",
"instrument_div_multiplier",
],
):
output_dict = self.output_config_with_estimated_parameters(
attr_names=attr_names
)
with open(yaml_filename, "w") as f:
yaml.dump(output_dict, f, default_flow_style=False)
def calculation_details(self, instrument_code):
"""
Explain how the position is calculated for a given instrument
:return: pd.Series
"""
system = self.system
attributes_last_ts = [
"combForecast.get_combined_forecast",
"rawdata.daily_denominator_price",
"rawdata.daily_returns_volatility",
"positionSize.get_price_volatility",
"positionSize.get_block_value",
"positionSize.get_instrument_currency_vol",
"positionSize.get_fx_rate",
"positionSize.get_instrument_value_vol",
"positionSize.get_average_position_at_subsystem_level",
"positionSize.get_subsystem_position",
"portfolio.get_notional_position",
]
attributes_names = [
"Fcast",
"Price",
"S(P_d)",
"S(%daily)",
"Blck val",
"ICV",
"FX",
"IVV",
"Vol scalar",
"SS Pos",
"Pos.",
]
results = dict()
for attribute, name in zip(attributes_last_ts, attributes_names):
stage, method = attribute.split(".")
stage_object = getattr(system, stage)
stage_method = getattr(stage_object, method)
result = stage_method(instrument_code).ffill().iloc[-1]
results[name] = result
attributes_dict = ["portfolio.get_instrument_weights"]
attributes_names = ["Instr.Wt"]
for attribute, name in zip(attributes_dict, attributes_names):
stage, method = attribute.split(".")
stage_object = getattr(system, stage)
stage_method = getattr(stage_object, method)
result_dict = stage_method()
result = result_dict[instrument_code].ffill().iloc[-1]
results[name] = result
attributes_all = ["portfolio.get_instrument_diversification_multiplier"]
attributes_names = ["IDM"]
for attribute, name in zip(attributes_all, attributes_names):
stage, method = attribute.split(".")
stage_object = getattr(system, stage)
stage_method = getattr(stage_object, method)
result = stage_method().ffill().iloc[-1]
results[name] = result
attributes_scalar = ["data.get_value_of_block_price_move"]
attributes_names = ["Blc size"]
for attribute, name in zip(attributes_scalar, attributes_names):
stage, method = attribute.split(".")
stage_object = getattr(system, stage)
stage_method = getattr(stage_object, method)
result = stage_method(instrument_code)
results[name] = result
results["Daily VolTgt"] = system.positionSize.get_vol_target_dict()[
"daily_cash_vol_target"
]
buffers = system.portfolio.get_buffers_for_position(instrument_code).iloc[-1]
results["Bfr+"], results["Bfr-"] = buffers.values
return results
def explain_calculator_for_code(self, instrument_code):
results = self.calculation_details(instrument_code)
explainers = [
"Position = Subsystem position * Instrument weight * IDM = %.2f * %.4f * %.2f = %.1f"
% (results["SS Pos"], results["Instr.Wt"], results["IDM"], results["Pos."]),
"Subsystem position = Combined forecast * Vol scalar / 10 = %.2f * %.2f / 10.0 = %.2f"
% (results["Fcast"], results["Vol scalar"], results["SS Pos"]),
"Vol scalar = Daily cash vol target / Instrument value vol = %.1f / %.1f = %.2f"
% (results["Daily VolTgt"], results["IVV"], results["Vol scalar"]),
"Instrument Value Vol = Instrument currency vol * FX rate = %.2f * %.6f = %.2f"
% (results["ICV"], results["FX"], results["IVV"]),
"Instrument currency vol = Block value * Daily %% Price vol = %.2f * %.4f = %.2f"
% (results["Blck val"], results["S(%daily)"], results["ICV"]),
"Daily %% Price vol = 100* Return difference vol / Price = %.6f / %.6f = %.4f (%.2f%% per year)"
% (
results["S(P_d)"],
results["Price"],
results["S(%daily)"],
results["S(%daily)"] * ROOT_BDAYS_INYEAR,
),
"Block value = Price * Block size * 0.01 = %.6f * %.1f * 0.01 = %.2f"
% (results["Price"], results["Blc size"], results["Blck val"]),
"OR Instrument currency vol = Return difference vol * Block size = %.6f * %.1f = %.2f"
% (results["S(P_d)"], results["Blc size"], results["ICV"]),
]
return explainers
def forecast_error(forecast, target_forecast_value):
abs_size = forecast.abs().mean()
std_size = forecast.std()
abs_error = abs(abs_size - target_forecast_value)
std_error = abs(std_size - target_forecast_value)
max_error = max([abs_error, std_error])
return max_error
```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.