From eace0c97cee58c6f3e68889cc108263f10541481 Mon Sep 17 00:00:00 2001 From: =?utf8?q?J=C3=A9r=C3=B4me=20Benoit?= Date: Fri, 3 Jul 2026 01:49:44 +0200 Subject: [PATCH] refactor(reforcexy): make optuna sampler dispatch exhaustive --- ReforceXY/user_data/freqaimodels/ReforceXY.py | 55 ++++++------ .../tests/test_reforcexy_sampler_dispatch.py | 87 +++++++++++++++++++ 2 files changed, 117 insertions(+), 25 deletions(-) create mode 100644 ReforceXY/user_data/tests/test_reforcexy_sampler_dispatch.py diff --git a/ReforceXY/user_data/freqaimodels/ReforceXY.py b/ReforceXY/user_data/freqaimodels/ReforceXY.py index 0cf5a38..1321753 100644 --- a/ReforceXY/user_data/freqaimodels/ReforceXY.py +++ b/ReforceXY/user_data/freqaimodels/ReforceXY.py @@ -20,6 +20,7 @@ from typing import ( Tuple, Type, Union, + assert_never, cast, ) @@ -1368,35 +1369,39 @@ class ReforceXY(BaseReinforcementLearningModel): return False def create_sampler(self) -> BaseSampler: - sampler: SamplerType = self.rl_config_optuna.get( + sampler_config = self.rl_config_optuna.get( "sampler", ReforceXY._SAMPLER_TYPES[0] - ) # "tpe" - seed = self.rl_config_optuna.get("seed", 42) - # "auto" - if sampler == ReforceXY._SAMPLER_TYPES[1]: - logger.info( - "Hyperopt [global]: using AutoSampler (seed=%d)", - seed, - ) - return optunahub.load_module("samplers/auto_sampler").AutoSampler(seed=seed) - # "tpe" - elif sampler == ReforceXY._SAMPLER_TYPES[0]: - logger.info( - "Hyperopt [global]: using TPESampler (n_startup_trials=%d, multivariate=True, group=True, seed=%d)", - self.optuna_n_startup_trials, - seed, - ) - return TPESampler( - n_startup_trials=self.optuna_n_startup_trials, - multivariate=True, - group=True, - seed=seed, - ) - else: + ) + if sampler_config not in ReforceXY._SAMPLER_TYPES: raise ValueError( - f"Hyperopt [global]: unsupported sampler '{sampler}'. " + f"Hyperopt [global]: unsupported sampler '{sampler_config}'. " f"Valid: {', '.join(ReforceXY._SAMPLER_TYPES)}" ) + sampler = cast(SamplerType, sampler_config) + seed = self.rl_config_optuna.get("seed", 42) + match sampler: + case "tpe": + logger.info( + "Hyperopt [global]: using TPESampler (n_startup_trials=%d, multivariate=True, group=True, seed=%d)", + self.optuna_n_startup_trials, + seed, + ) + return TPESampler( + n_startup_trials=self.optuna_n_startup_trials, + multivariate=True, + group=True, + seed=seed, + ) + case "auto": + logger.info( + "Hyperopt [global]: using AutoSampler (seed=%d)", + seed, + ) + return optunahub.load_module("samplers/auto_sampler").AutoSampler( + seed=seed + ) + case _: + assert_never(sampler) @staticmethod def create_pruner( diff --git a/ReforceXY/user_data/tests/test_reforcexy_sampler_dispatch.py b/ReforceXY/user_data/tests/test_reforcexy_sampler_dispatch.py new file mode 100644 index 0000000..daae9ba --- /dev/null +++ b/ReforceXY/user_data/tests/test_reforcexy_sampler_dispatch.py @@ -0,0 +1,87 @@ +import importlib +import inspect +import sys +import types + +from optuna.samplers import TPESampler + + +sys.path.insert(0, "/freqtrade/user_data") +reforcexy_module = importlib.import_module("freqaimodels.ReforceXY") +ReforceXY = reforcexy_module.ReforceXY + + +class AutoSampler: + def __init__(self, seed: int) -> None: + self.seed = seed + + +def make_model(sampler: str) -> ReforceXY: + model = object.__new__(ReforceXY) + model.rl_config_optuna = {"sampler": sampler, "seed": 7} + model.optuna_n_startup_trials = 3 + return model + + +def test_create_sampler_returns_tpe_when_sampler_is_tpe() -> None: + # Given: a ReforceXY model configured for the public tpe sampler. + model = make_model("tpe") + + # When: the sampler factory is invoked. + sampler = model.create_sampler() + + # Then: the factory returns Optuna's TPE sampler. + assert isinstance(sampler, TPESampler) + + +def test_create_sampler_returns_auto_when_sampler_is_auto() -> None: + # Given: a ReforceXY model configured for the public auto sampler. + model = make_model("auto") + original_load_module = reforcexy_module.optunahub.load_module + reforcexy_module.optunahub.load_module = lambda _: types.SimpleNamespace( + AutoSampler=AutoSampler + ) + + try: + # When: the sampler factory is invoked. + sampler = model.create_sampler() + finally: + reforcexy_module.optunahub.load_module = original_load_module + + # Then: the factory returns the AutoSampler provided by optunahub. + assert isinstance(sampler, AutoSampler) + + +def test_create_sampler_rejects_invalid_public_sampler() -> None: + # Given: a ReforceXY model configured with an unsupported public sampler. + model = make_model("invalid") + + try: + # When: the sampler factory is invoked. + model.create_sampler() + except ValueError as exc: + # Then: public invalid config fails before typed exhaustive dispatch. + message = str(exc) + assert "Hyperopt [global]: unsupported sampler 'invalid'." in message + assert "Valid: tpe, auto" in message + else: + raise AssertionError("unsupported sampler did not raise ValueError") + + +def test_create_sampler_uses_exhaustive_match_dispatch() -> None: + # Given: ReforceXY exposes a finite SamplerType surface. + source = inspect.getsource(ReforceXY.create_sampler) + + # When/Then: dispatch is encoded as a match with assert_never for type-level exhaustiveness. + assert "match sampler:" in source + assert 'case "tpe":' in source + assert 'case "auto":' in source + assert "assert_never(sampler)" in source + + +if __name__ == "__main__": + test_create_sampler_returns_tpe_when_sampler_is_tpe() + test_create_sampler_returns_auto_when_sampler_is_auto() + test_create_sampler_rejects_invalid_public_sampler() + test_create_sampler_uses_exhaustive_match_dispatch() + print("sampler dispatch checks passed") -- 2.53.0