From 9040c75e72f411ee66682018a640b5381e2525dc Mon Sep 17 00:00:00 2001 From: Rodion Lim Date: Mon, 19 Jan 2026 22:54:16 +0800 Subject: [PATCH 1/3] tests: add forecasting minimal tests --- src/quantlib_st/systems/basesystem.py | 11 +++++- .../provided/futures_chapter15/basesystem.py | 15 +++++++- src/quantlib_st/systems/rawdata.py | 2 +- src/quantlib_st/systems/tests/__init__.py | 0 .../systems/tests/test_forecasting.py | 38 +++++++++++++++++++ 5 files changed, 62 insertions(+), 4 deletions(-) create mode 100644 src/quantlib_st/systems/tests/__init__.py create mode 100644 src/quantlib_st/systems/tests/test_forecasting.py diff --git a/src/quantlib_st/systems/basesystem.py b/src/quantlib_st/systems/basesystem.py index eb33571..cda1167 100644 --- a/src/quantlib_st/systems/basesystem.py +++ b/src/quantlib_st/systems/basesystem.py @@ -1,4 +1,9 @@ -from typing import Optional +from typing import Optional, TYPE_CHECKING + +if TYPE_CHECKING: + from quantlib_st.systems.forecasting import Rules + from quantlib_st.systems.accounts.accounts_stage import Account + from quantlib_st.systems.rawdata import RawData from quantlib_st.config.configdata import Config from quantlib_st.config.instruments import ( @@ -40,6 +45,10 @@ class System(object): """ + rules: "Rules" + accounts: "Account" + rawdata: "RawData" + def __init__( self, stage_list: list, diff --git a/src/quantlib_st/systems/provided/futures_chapter15/basesystem.py b/src/quantlib_st/systems/provided/futures_chapter15/basesystem.py index 7bbc842..31f3f16 100644 --- a/src/quantlib_st/systems/provided/futures_chapter15/basesystem.py +++ b/src/quantlib_st/systems/provided/futures_chapter15/basesystem.py @@ -6,14 +6,16 @@ from quantlib_st.config.configdata import Config from quantlib_st.sysdata.sim.csv_futures_sim_test_data import CsvFuturesSimTestData -from quantlib_st.systems.basesystem import System from quantlib_st.systems.accounts.accounts_stage import Account +from quantlib_st.systems.basesystem import System +from quantlib_st.systems.forecasting import Rules from quantlib_st.systems.rawdata import RawData def futures_system( data=None, config=None, + trading_rules=None, ): """ Minimal system with only Account stage. @@ -31,8 +33,17 @@ def futures_system( if config is None: config = Config("systems.provided.config.test_account_config.yaml") + stage_list = [ + Account(), + RawData(), + ] + + if trading_rules is not None: + rules = Rules(trading_rules) + stage_list.append(rules) + system = System( - [Account(), RawData()], + stage_list, data, config, ) diff --git a/src/quantlib_st/systems/rawdata.py b/src/quantlib_st/systems/rawdata.py index c0d1abc..2d50f36 100644 --- a/src/quantlib_st/systems/rawdata.py +++ b/src/quantlib_st/systems/rawdata.py @@ -2,11 +2,11 @@ import pandas as pd -from systems.stage import SystemStage from quantlib_st.core.objects import resolve_function from quantlib_st.core.dateutils import ROOT_BDAYS_INYEAR from quantlib_st.core.genutils import list_intersection from quantlib_st.core.exceptions import missingData +from quantlib_st.systems.stage import SystemStage from quantlib_st.systems.system_cache import input, diagnostic, output from quantlib_st.sysdata.sim.futures_sim_data import futuresSimData diff --git a/src/quantlib_st/systems/tests/__init__.py b/src/quantlib_st/systems/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/quantlib_st/systems/tests/test_forecasting.py b/src/quantlib_st/systems/tests/test_forecasting.py new file mode 100644 index 0000000..65dc9bf --- /dev/null +++ b/src/quantlib_st/systems/tests/test_forecasting.py @@ -0,0 +1,38 @@ +import pytest + +from quantlib_st.sysdata.sim.csv_futures_sim_test_data import CsvFuturesSimTestData +from quantlib_st.config.configdata import Config +from quantlib_st.systems.provided.futures_chapter15.basesystem import futures_system +from quantlib_st.systems.basesystem import System + + +@pytest.fixture +def system(request) -> System: + """Minimal system fixture that can be parametrized with trading rules.""" + trading_rules = getattr(request, "param", None) + system = futures_system( + trading_rules=trading_rules, + data=CsvFuturesSimTestData(), + config=Config("systems.provided.config.test_account_config.yaml"), + ) + return system + + +@pytest.mark.parametrize( + "system", + [ + dict( + rule0="systems.provided.rules.ewmac.ewmac_forecast_with_defaults", + rule1="systems.provided.rules.breakout.breakout", + rule2="systems.provided.rules.carry.carry", + ) + ], + indirect=True, +) +def test_get_raw_forecast_minimal(system: System): + """Minimal smoke test for forecasting rule.""" + ans = system.rules.get_raw_forecast("EDOLLAR", "rule0") + assert abs(ans.iloc[-1] - (-2.784115)) < 0.00001 + + ans = system.rules.get_raw_forecast("EDOLLAR", "rule1") + assert abs(ans.iloc[-1] - (-19.617574)) < 0.00001 From 35b997c83930671d0f05718d9394dfd5b1197b59 Mon Sep 17 00:00:00 2001 From: Rodion Lim Date: Mon, 19 Jan 2026 22:59:30 +0800 Subject: [PATCH 2/3] feat(forecast): add forecast_scale_cap stage --- src/quantlib_st/systems/forecast_scale_cap.py | 519 ++++++++++++++++++ 1 file changed, 519 insertions(+) create mode 100644 src/quantlib_st/systems/forecast_scale_cap.py diff --git a/src/quantlib_st/systems/forecast_scale_cap.py b/src/quantlib_st/systems/forecast_scale_cap.py new file mode 100644 index 0000000..873be9f --- /dev/null +++ b/src/quantlib_st/systems/forecast_scale_cap.py @@ -0,0 +1,519 @@ +from copy import copy + +import numpy as np +import pandas as pd + +from quantlib_st.config.configdata import Config + +from quantlib_st.systems.basesystem import ALL_KEYNAME +from quantlib_st.systems.stage import SystemStage +from quantlib_st.systems.system_cache import input, dont_cache, diagnostic, output + +from quantlib_st.core.genutils import str2Bool +from quantlib_st.core.objects import resolve_function + + +class ForecastScaleCap(SystemStage): + """ + Stage for scaling and capping + + This is a 'switching' class which selects either the fixed or the + estimated flavours + + """ + + @property + def name(self): + return "forecastScaleCap" + + @output() + def get_capped_forecast( + self, instrument_code: str, rule_variation_name: str + ) -> pd.Series: + """ + + Return the capped, scaled, forecast + + KEY OUTPUT + + + :param instrument_code: + :type str: + + :param rule_variation_name: + :type str: name of the trading rule variation + + :returns: Tx1 pd.DataFrame, same size as forecast + + >>> from systems.tests.testdata import get_test_object_futures_with_rules + >>> from systems.basesystem import System + >>> (rules, rawdata, data, config)=get_test_object_futures_with_rules() + >>> config.forecast_cap=0.2 + >>> system=System([rawdata, rules, ForecastScaleCapFixed()], data, config) + >>> system.forecastScaleCap.get_capped_forecast("EDOLLAR", "ewmac8").tail(2) + ewmac8 + 2015-12-10 -0.190583 + 2015-12-11 0.200000 + """ + + self.log.debug( + "Calculating capped forecast for %s %s" + % (instrument_code, rule_variation_name), + instrument_code=instrument_code, + ) + + scaled_forecast = self.get_scaled_forecast(instrument_code, rule_variation_name) + upper_cap = self.get_forecast_cap() + lower_floor = self.get_forecast_floor() + + capped_scaled_forecast = scaled_forecast.clip( + upper=upper_cap, lower=lower_floor + ) + + return capped_scaled_forecast + + @diagnostic() + def get_scaled_forecast(self, instrument_code, rule_variation_name): + """ + Return the scaled forecast + + :param instrument_code: + :type str: + + :param rule_variation_name: + :type str: name of the trading rule variation + + :returns: Tx1 pd.DataFrame, same size as forecast + + >>> from systems.tests.testdata import get_test_object_futures_with_rules + >>> from systems.basesystem import System + >>> (rules, rawdata, data, config)=get_test_object_futures_with_rules() + >>> system=System([rawdata, rules, ForecastScaleCapFixed()], data, config) + >>> system.forecastScaleCap.get_scaled_forecast("EDOLLAR", "ewmac8").tail(2) + ewmac8 + 2015-12-10 -0.190583 + 2015-12-11 0.871231 + """ + + raw_forecast = self.get_raw_forecast(instrument_code, rule_variation_name) + forecast_scalar = self.get_forecast_scalar( + instrument_code, rule_variation_name + ) # will either be a scalar or a timeseries + + scaled_forecast = raw_forecast * forecast_scalar + + return scaled_forecast + + @input + def get_raw_forecast( + self, instrument_code: str, rule_variation_name: str + ) -> pd.Series: + """ + Convenience method as we use the raw forecast several times + + :param instrument_code: + :type str: + + :param rule_variation_name: + :type str: name of the trading rule variation + + :returns: Tx1 pd.DataFrame, same size as forecast + + >>> from systems.tests.testdata import get_test_object_futures_with_rules + >>> from systems.basesystem import System + >>> (rules, rawdata, data, config)=get_test_object_futures_with_rules() + >>> system=System([rawdata, rules, ForecastScaleCapFixed()], data, config) + >>> system.forecastScaleCap.get_raw_forecast("EDOLLAR","ewmac8").tail(2) + ewmac8 + 2015-12-10 -0.035959 + 2015-12-11 0.164383 + """ + + raw_forecast = self.rules_stage.get_raw_forecast( + instrument_code, rule_variation_name + ) + + return raw_forecast + + @property + def rules_stage(self): + return self.parent.rules + + @dont_cache + def get_forecast_scalar( + self, instrument_code: str, rule_variation_name: str + ) -> pd.Series: + if self._use_estimated_weights(): + forecast_scalar = self._get_forecast_scalar_estimated( + instrument_code, rule_variation_name + ) + else: + forecast_scalar = self._get_forecast_scalar_fixed_as_series( + instrument_code, rule_variation_name + ) + + return forecast_scalar + + @dont_cache + def _use_estimated_weights(self) -> bool: + return str2Bool(self.config.use_forecast_scale_estimates) + + @property + def config(self) -> Config: + return self.parent.config + + # protected in cache as slow to estimate + @diagnostic(protected=True) + def _get_forecast_scalar_estimated( + self, instrument_code: str, rule_variation_name: str + ) -> pd.Series: + """ + Get the scalar to apply to raw forecasts + + If not cached, these are estimated from past forecasts + + If configuration variable pool_forecasts_for_scalar is "True", then we + do this across instruments. + + :param instrument_code: + :type str: + + :param rule_variation_name: + :type str: name of the trading rule variation + + :returns: float + + >>> from systems.tests.testdata import get_test_object_futures_with_rules + >>> from systems.basesystem import System + >>> (rules, rawdata, data, config)=get_test_object_futures_with_rules() + >>> system1=System([rawdata, rules, ForecastScaleCapEstimated()], data, config) + >>> + >>> ## From default + >>> system1.forecastScaleCap.get_forecast_scalar("EDOLLAR", "ewmac8").tail(3) + scale_factor + 2015-12-09 5.849888 + 2015-12-10 5.850474 + 2015-12-11 5.851091 + >>> system1.forecastScaleCap.get_capped_forecast("EDOLLAR", "ewmac8").tail(3) + ewmac8 + 2015-12-09 0.645585 + 2015-12-10 -0.210377 + 2015-12-11 0.961821 + >>> + >>> ## From config + >>> scale_config=dict(pool_instruments=False) + >>> config.forecast_scalar_estimate=scale_config + >>> system3=System([rawdata, rules, ForecastScaleCapEstimated()], data, config) + >>> system3.forecastScaleCap.get_forecast_scalar("EDOLLAR", "ewmac8").tail(3) + scale_factor + 2015-12-09 5.652174 + 2015-12-10 5.652833 + 2015-12-11 5.653444 + >>> + """ + # Get some useful stuff from the config + forecast_scalar_config = copy(self.config.forecast_scalar_estimate) + + instrument_code_to_pass = _get_instrument_code_depending_on_pooling_status( + instrument_code=instrument_code, + forecast_scalar_config=forecast_scalar_config, + ) + + scaling_factor = self._get_forecast_scalar_estimated_from_instrument_code( + instrument_code=instrument_code_to_pass, + rule_variation_name=rule_variation_name, + forecast_scalar_config=forecast_scalar_config, + ) + + forecast = self.get_raw_forecast(instrument_code, rule_variation_name) + forecast_scalar = scaling_factor.reindex(forecast.index, method="ffill") + + return forecast_scalar + + # protected in cache as slow to estimate + @diagnostic(protected=True) + def _get_forecast_scalar_estimated_from_instrument_code( + self, + instrument_code: str, + rule_variation_name: str, + forecast_scalar_config: dict, + ) -> pd.Series: + """ + Get the scalar to apply to raw forecasts + + If not cached, these are estimated from past forecasts + + + :param instrument_code: instrument code, or ALL_KEYNAME if pooling + :type str: + + :param rule_variation_name: + :type str: name of the trading rule variation + + :param forecast_scalar_config: + :type dict: relevant part of the config + + :returns: float + + """ + + # The config contains 'func' and some other arguments + # we turn func which could be a string into a function, and then + # call it with the other args + + cs_forecasts = self._get_cross_sectional_forecasts_for_instrument( + instrument_code, rule_variation_name + ) + scalar_function = resolve_function(forecast_scalar_config.pop("func")) + + # an example of a scaling function is sysquant.estimators.forecast_scalar.forecast_scalar + # must return thing the same size as cs_forecasts + + # This we get from here to avoid possible inconsistency + target_abs_forecast = self.target_abs_forecast() + + scaling_factor = scalar_function( + cs_forecasts, + target_abs_forecast=target_abs_forecast, + **forecast_scalar_config, + ) + + return scaling_factor + + @dont_cache + def target_abs_forecast(self) -> float: + return self.config.average_absolute_forecast + + @diagnostic() + def _get_cross_sectional_forecasts_for_instrument( + self, instrument_code: str, rule_variation_name: str + ) -> pd.DataFrame: + """ + instrument_list contains multiple things, might pool everything across + all instruments + """ + + if instrument_code == ALL_KEYNAME: + # pool data across all instruments using this trading rule + instrument_list = self._list_of_instruments_for_trading_rule( + rule_variation_name + ) + + else: + ## not pooled + instrument_list = [instrument_code] + + self.log.debug( + "Getting cross sectional forecasts for scalar calculation for %s over %s" + % (rule_variation_name, ", ".join(instrument_list)) + ) + + forecast_list = [ + self.get_raw_forecast(instrument_code, rule_variation_name) + for instrument_code in instrument_list + ] + + cs_forecasts = pd.concat(forecast_list, axis=1) + cs_forecasts.columns = instrument_list + + return cs_forecasts + + @diagnostic() + def _list_of_instruments_for_trading_rule(self, rule_variation_name: str) -> list: + """ + Return the list of instruments associated with a given rule + + If we don't have a combForecast this will be all of our instruments + + :param rule_variation_name: + :return: list + """ + + instrument_list = self.parent.get_instrument_list() + instruments_with_rule = [ + instrument_code + for instrument_code in instrument_list + if rule_variation_name in self._get_trading_rule_list(instrument_code) + ] + + if len(instruments_with_rule) == 0: + return instrument_list + else: + return instruments_with_rule + + @input + def _get_trading_rule_list(self, instrument_code: str) -> list: + """ + Get a list of trading rules which apply to a particular instrument + + :param instrument_code: + :return: list of trading rules + """ + + try: + getattr(self.parent, "combForecast") + except AttributeError: + return [] + else: + return self.comb_forecast_stage.get_trading_rule_list(instrument_code) + + @property + def comb_forecast_stage(self): + # no use of -> as would cause circular import + return self.parent.combForecast + + @diagnostic() + def _get_forecast_scalar_fixed_as_series( + self, instrument_code: str, rule_variation_name: str + ) -> pd.Series: + """ + Get the scalar to apply to raw forecasts + + In this simple version it's the same for all instruments, and fixed + + We get the scalars from: (a) configuration file in parent system + (b) or if missing: uses the scalar from systems.defaults.py + + :param instrument_code: + :type str: + + :param rule_variation_name: + :type str: name of the trading rule variation + + :returns: Series + + """ + scalar = self._get_forecast_scalar_fixed( + instrument_code=instrument_code, rule_variation_name=rule_variation_name + ) + raw_forecast = self.get_raw_forecast( + instrument_code=instrument_code, rule_variation_name=rule_variation_name + ) + forecast_scalar = pd.Series( + np.full(raw_forecast.shape[0], scalar), index=raw_forecast.index + ) + + return forecast_scalar + + @diagnostic() + def _get_forecast_scalar_fixed( + self, instrument_code: str, rule_variation_name: str + ) -> pd.Series: + """ + Get the scalar to apply to raw forecasts + + In this simple version it's the same for all instruments, and fixed + + We get the scalars from: (a) configuration file in parent system + (b) or if missing: uses the scalar from systems.defaults.py + + :param instrument_code: + :type str: + + :param rule_variation_name: + :type str: name of the trading rule variation + + :returns: float + + >>> from systems.tests.testdata import get_test_object_futures_with_rules + >>> from systems.basesystem import System + >>> (rules, rawdata, data, config)=get_test_object_futures_with_rules() + >>> system1=System([rawdata, rules, ForecastScaleCapFixed()], data, config) + >>> + >>> ## From config + >>> system1.forecastScaleCap.get_forecast_scalar("EDOLLAR", "ewmac8") + 5.3 + >>> + >>> ## default + >>> unused=config.trading_rules['ewmac8'].pop('forecast_scalar') + >>> system3=System([rawdata, rules, ForecastScaleCapFixed()], data, config) + >>> system3.forecastScaleCap.get_forecast_scalar("EDOLLAR", "ewmac8") + 1.0 + >>> + >>> ## other config location + >>> setattr(config, 'forecast_scalars', dict(ewmac8=11.0)) + >>> system4=System([rawdata, rules, ForecastScaleCapFixed()], data, config) + >>> system4.forecastScaleCap.get_forecast_scalar("EDOLLAR", "ewmac8") + 11.0 + """ + + config = self.config + try: + scalar = config.trading_rules[rule_variation_name]["forecast_scalar"] + except: + try: + # can also put somewhere else ... + scalar = config.forecast_scalars[rule_variation_name] + except: + # just one global default + scalar = config.get_element("forecast_scalar") + + return scalar + + @diagnostic() + def get_forecast_cap(self) -> float: + """ + Get forecast cap + + We get the cap from: + (a) configuration object in parent system + (c) or if missing: uses the forecast_cap from systems.default.py + :returns: float + + >>> from systems.tests.testdata import get_test_object_futures_with_rules + >>> from systems.basesystem import System + >>> (rules, rawdata, data, config)=get_test_object_futures_with_rules() + >>> system=System([rawdata, rules, ForecastScaleCapFixed()], data, config) + >>> + >>> ## From config + >>> system.forecastScaleCap.get_forecast_cap() + 21.0 + >>> + >>> ## default + >>> del(config.forecast_cap) + >>> system3=System([rawdata, rules, ForecastScaleCapFixed()], data, config) + >>> system3.forecastScaleCap.get_forecast_cap() + 20.0 + + """ + + return self.config.forecast_cap + + @diagnostic() + def get_forecast_floor(self) -> float: + """ + Get forecast floor + + We get the cap from: + (a) configuration object in parent system + (c) or if missing: uses the the cap with a minus sign in front of it + :returns: float + + """ + + forecast_cap = self.get_forecast_cap() + minus_forecast_cap = -forecast_cap + forecast_floor = getattr(self.config, "forecast_floor", minus_forecast_cap) + + return forecast_floor + + +def _get_instrument_code_depending_on_pooling_status( + instrument_code: str, forecast_scalar_config: dict +) -> str: + # this determines whether we pool or not + pool_instruments = str2Bool(forecast_scalar_config.pop("pool_instruments")) + + if pool_instruments: + # pooled, same for all instruments + instrument_code_to_pass = ALL_KEYNAME + else: + instrument_code_to_pass = copy(instrument_code) + + return instrument_code_to_pass + + +if __name__ == "__main__": + import doctest + + doctest.testmod() From a4e91af815acf25d48f7963419211ddad6ef05f2 Mon Sep 17 00:00:00 2001 From: Rodion Lim Date: Tue, 20 Jan 2026 00:23:37 +0800 Subject: [PATCH 3/3] test(forecast): add tests for forecast_scale_cap --- src/quantlib_st/config/configdata.py | 18 +++- src/quantlib_st/systems/basesystem.py | 9 +- src/quantlib_st/systems/forecast_scale_cap.py | 8 +- src/quantlib_st/systems/forecasting.py | 11 +-- .../provided/config/test_forecast_config.yaml | 20 +++++ .../provided/futures_chapter15/basesystem.py | 14 ++- src/quantlib_st/systems/stage.py | 2 +- .../systems/tests/test_forecast_scale_cap.py | 88 +++++++++++++++++++ 8 files changed, 155 insertions(+), 15 deletions(-) create mode 100644 src/quantlib_st/systems/provided/config/test_forecast_config.yaml create mode 100644 src/quantlib_st/systems/tests/test_forecast_scale_cap.py diff --git a/src/quantlib_st/config/configdata.py b/src/quantlib_st/config/configdata.py index 3acf201..73a20ce 100644 --- a/src/quantlib_st/config/configdata.py +++ b/src/quantlib_st/config/configdata.py @@ -13,7 +13,7 @@ """ from pathlib import Path -from typing import Optional, Union +from typing import Optional, Union, Any import os import yaml @@ -45,6 +45,14 @@ class Config(object): + # Common configuration attributes (type hints for IDE) + trading_rules: Any + forecast_scalar_estimate: Any + forecast_scalar_fixed: Any + instruments: Any + parameters: Any + base_currency: Any + def __init__( self, config_object: Optional[Union[str, dict, list]] = None, @@ -110,6 +118,14 @@ def add_single_element(self, element_name): elements.append(element_name) self._elements = elements + def __getattr__(self, name: str) -> Any: + # This allows Pylance to know that Config can have dynamic attributes + # and prevents "unknown attribute" errors. + # We only get here if the attribute isn't already defined in __dict__ + raise AttributeError( + f"'{type(self).__name__}' object has no attribute '{name}'" + ) + def get_element(self, element_name): try: result = getattr(self, element_name) diff --git a/src/quantlib_st/systems/basesystem.py b/src/quantlib_st/systems/basesystem.py index cda1167..78d794c 100644 --- a/src/quantlib_st/systems/basesystem.py +++ b/src/quantlib_st/systems/basesystem.py @@ -5,6 +5,9 @@ from quantlib_st.systems.accounts.accounts_stage import Account from quantlib_st.systems.rawdata import RawData + # from quantlib_st.systems.forecast_combine import ForecastCombine + from quantlib_st.systems.forecast_scale_cap import ForecastScaleCap + from quantlib_st.config.configdata import Config from quantlib_st.config.instruments import ( get_duplicate_list_of_instruments_to_remove_from_config, @@ -48,6 +51,8 @@ class System(object): rules: "Rules" accounts: "Account" rawdata: "RawData" + # combForecast: "ForecastCombine" + forecastScaleCap: "ForecastScaleCap" def __init__( self, @@ -138,11 +143,11 @@ def log(self): return self._log @property - def data(self): + def data(self) -> simData: return self._data @property - def config(self): + def config(self) -> Config: return self._config @property diff --git a/src/quantlib_st/systems/forecast_scale_cap.py b/src/quantlib_st/systems/forecast_scale_cap.py index 873be9f..0fab541 100644 --- a/src/quantlib_st/systems/forecast_scale_cap.py +++ b/src/quantlib_st/systems/forecast_scale_cap.py @@ -1,4 +1,8 @@ from copy import copy +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from quantlib_st.systems.forecast_combine import ForecastCombine import numpy as np import pandas as pd @@ -272,6 +276,7 @@ def _get_forecast_scalar_estimated_from_instrument_code( # This we get from here to avoid possible inconsistency target_abs_forecast = self.target_abs_forecast() + assert scalar_function is not None scaling_factor = scalar_function( cs_forecasts, target_abs_forecast=target_abs_forecast, @@ -358,8 +363,7 @@ def _get_trading_rule_list(self, instrument_code: str) -> list: return self.comb_forecast_stage.get_trading_rule_list(instrument_code) @property - def comb_forecast_stage(self): - # no use of -> as would cause circular import + def comb_forecast_stage(self) -> "ForecastCombine": return self.parent.combForecast @diagnostic() diff --git a/src/quantlib_st/systems/forecasting.py b/src/quantlib_st/systems/forecasting.py index 8c5e75d..624587e 100644 --- a/src/quantlib_st/systems/forecasting.py +++ b/src/quantlib_st/systems/forecasting.py @@ -66,11 +66,12 @@ def passed_trading_rules(self): return self._passed_trading_rules def __repr__(self): - trading_rules = self.trading_rules() - - rule_names = ", ".join(trading_rules.keys()) - - return "Rules object with rules " + rule_names + try: + trading_rules = self.trading_rules() + rule_names = ", ".join(trading_rules.keys()) + return "Rules object with rules " + rule_names + except Exception: + return "Rules object (uninitialized or error during initialization)" @output() def get_raw_forecast( diff --git a/src/quantlib_st/systems/provided/config/test_forecast_config.yaml b/src/quantlib_st/systems/provided/config/test_forecast_config.yaml new file mode 100644 index 0000000..bea904c --- /dev/null +++ b/src/quantlib_st/systems/provided/config/test_forecast_config.yaml @@ -0,0 +1,20 @@ +trading_rules: + ewmac8: + function: systems.provided.rules.ewmac.ewmac_forecast_with_defaults_no_vol + data: + - "rawdata.get_daily_prices" + - "rawdata.daily_returns_volatility" + other_args: + Lfast: 8 + Lslow: 32 + forecast_scalar: 5.3 + ewmac16: + function: systems.provided.rules.ewmac.ewmac_forecast_with_defaults_no_vol + data: + - "rawdata.get_daily_prices" + - "rawdata.daily_returns_volatility" + other_args: + Lfast: 16 + Lslow: 64 + forecast_scalar: 7.5 +forecast_cap: 21.0 diff --git a/src/quantlib_st/systems/provided/futures_chapter15/basesystem.py b/src/quantlib_st/systems/provided/futures_chapter15/basesystem.py index 31f3f16..0df52df 100644 --- a/src/quantlib_st/systems/provided/futures_chapter15/basesystem.py +++ b/src/quantlib_st/systems/provided/futures_chapter15/basesystem.py @@ -5,9 +5,12 @@ """ from quantlib_st.config.configdata import Config + from quantlib_st.sysdata.sim.csv_futures_sim_test_data import CsvFuturesSimTestData + from quantlib_st.systems.accounts.accounts_stage import Account from quantlib_st.systems.basesystem import System +from quantlib_st.systems.forecast_scale_cap import ForecastScaleCap from quantlib_st.systems.forecasting import Rules from quantlib_st.systems.rawdata import RawData @@ -33,15 +36,18 @@ def futures_system( if config is None: config = Config("systems.provided.config.test_account_config.yaml") + if isinstance(trading_rules, Rules): + rules_stage = trading_rules + else: + rules_stage = Rules(trading_rules) + stage_list = [ Account(), + ForecastScaleCap(), RawData(), + rules_stage, ] - if trading_rules is not None: - rules = Rules(trading_rules) - stage_list.append(rules) - system = System( stage_list, data, diff --git a/src/quantlib_st/systems/stage.py b/src/quantlib_st/systems/stage.py index 9359184..88b92f0 100644 --- a/src/quantlib_st/systems/stage.py +++ b/src/quantlib_st/systems/stage.py @@ -21,7 +21,7 @@ class SystemStage(object): """ @property - def name(self): + def name(self) -> str: return "Need to replace method when inheriting" def __repr__(self): diff --git a/src/quantlib_st/systems/tests/test_forecast_scale_cap.py b/src/quantlib_st/systems/tests/test_forecast_scale_cap.py new file mode 100644 index 0000000..4107202 --- /dev/null +++ b/src/quantlib_st/systems/tests/test_forecast_scale_cap.py @@ -0,0 +1,88 @@ +import pytest + +from quantlib_st.config.configdata import Config + +from quantlib_st.sysdata.sim.csv_futures_sim_test_data import CsvFuturesSimTestData + +from quantlib_st.systems.basesystem import System +from quantlib_st.systems.forecasting import Rules +from quantlib_st.systems.provided.futures_chapter15.basesystem import futures_system + + +@pytest.fixture +def system() -> System: + """Minimal system fixture that can be parametrized with trading rules.""" + system = futures_system( + trading_rules=Rules(), + data=CsvFuturesSimTestData(), + config=Config("systems.provided.config.test_forecast_config.yaml"), + ) + return system + + +@pytest.fixture +def system_defaults() -> System: + """Fetch system defaults for forecast scaling and capping config.""" + cfg = Config("systems.provided.config.test_forecast_config.yaml") + del cfg.forecast_cap + system = futures_system( + trading_rules=Rules(), + data=CsvFuturesSimTestData(), + config=cfg, + ) + return system + + +def test_get_raw_forecast(system): + # Check 2015-12-11 to ensure consistency with historical pysystemtrade benchmarks + ans = system.forecastScaleCap.get_raw_forecast("EDOLLAR", "ewmac8") + val_2015 = ans.loc["2015-12-11"] + assert abs(val_2015 - 0.191395) < 1e-6 + + +def test_get_forecast_cap(system, system_defaults): + ans = system.forecastScaleCap.get_forecast_cap() + assert ans == 21.0 + + ans = system_defaults.forecastScaleCap.get_forecast_cap() + assert ans == 20.0 + + +def test_get_forecast_scalar(system): + # 1. From config (ewmac8 has 5.3 in the yaml) + ans = system.forecastScaleCap.get_forecast_scalar("EDOLLAR", "ewmac8") + assert (ans == 5.3).all() + + # 2. default (if missing from config, should be 1.0) + cfg_no_scalar = Config( + "quantlib_st.systems.provided.config.test_forecast_config.yaml" + ) + del cfg_no_scalar.trading_rules["ewmac8"]["forecast_scalar"] + + system_default_scalar = futures_system( + trading_rules=Rules(), + data=CsvFuturesSimTestData(), + config=cfg_no_scalar, + ) + + ans = system_default_scalar.forecastScaleCap.get_forecast_scalar( + "EDOLLAR", "ewmac8" + ) + assert (ans == 1.0).all() + + # 3. other config location (using .forecast_scalars dict) + cfg_other_loc = Config( + "quantlib_st.systems.provided.config.test_forecast_config.yaml" + ) + # Must remove from trading_rules first as it takes precedence + del cfg_other_loc.trading_rules["ewmac8"]["forecast_scalar"] + cfg_other_loc.forecast_scalars = dict(ewmac8=11.0) + + system_other_loc = futures_system( + trading_rules=Rules(), + data=CsvFuturesSimTestData(), + config=cfg_other_loc, + ) + + ans = system_other_loc.forecastScaleCap.get_forecast_scalar("EDOLLAR", "ewmac8") + assert (ans == 11.0).all()