{ "cells": [ { "cell_type": "code", "execution_count": null, "metadata": { "collapsed": false }, "outputs": [], "source": [ "%matplotlib inline" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "\n\n# Custom Optuna Optimizer\n\n

Warning

This example shows more advanced features of `tpcp` when using\n [Optuna](https://optuna.readthedocs.io/en/stable/index.html) for hyperparameter optimization.\n To make this example understandable, you should make yourself familiar with Optuna first and understand\n how it works, before trying to go through this example.

\n\nThe most popular method of (hyper-)parameter optimization is GridSearch (or GridSearchCV for optimizable pipelines).\nThese methods perform an exhaustive search of the parameter space by simply testing every option.\nConsidering that training and testing an algorithm can be very costly, exhaustive gridsearch takes a long time and is\nsometimes not feasible at all due to the required computational load.\n\nFor these cases various alternatives exist like :class:`~sklearn.model_selection.RandomizedSearchCV`,\n:class:`~sklearn.model_selection.HalvingGridSearchCV`, or advanced blackbox optimizer like\n:class:`~optuna.samplers.TPESampler`.\n`tpcp` does not implement all of these methods explicitly, as it would simply be too much work.\nHowever, we try to make it relatively simple to bring such methods into the `tpcp` ecosystem by providing an interface\nfor [Optuna](https://optuna.readthedocs.io/en/stable/index.html).\nOptuna is a state-of-the-art hyperparameter optimization framework that allows to implement any of the methods\nmentioned above (and more), and allows to easily create custom samplers and pruners.\n\nHowever, Optuna uses a (very elegant) functional interface that does not play well with the sklearn-inspired\ninterface of `tpcp`.\nTherefore, we provide the :class:`~tpcp.optimize.optuna.CustomOptunaOptimize` class which you can subclass to create\nyour own Optuna based optimizer.\n\n

Note

There is no need to create a custom subclass if you *only* want to run the hyperparameter optimization and\n *not* nest the optimization into other `tpcp` methods like :func:`~tpcp.validate.cross_validate`.\n For these cases, you can simply use Optuna with its default interface and just call the respective `tpcp`\n methods in the objective function.

\n\nIn this example, we are going to create an optimized gridsearch using custom pruning that terminates trials early\nif we already realise at the first couple of participants that the parameter combination will not work well.\n\nKeep in mind that this example should merely demonstrate the possibility to integrate Optuna with `tpcp`.\nYou are very much encouraged to read through the\n[Optuna documentation](https://optuna.readthedocs.io/en/stable/tutorial/index.html) and create your\nown project-specific optimizers.\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## The Prerequisites\nFirst, we need a dataset and a pipeline we want to optimize.\nFor this example we are using the `QRSDetector` pipeline (the non-trainable version) and the `ECGExampleData` dataset.\nCheck out the other examples to learn more about them.\nWe will simply copy the code over and create an instance of both objects to be used later.\n\n

Note

We make pretty extensive use of Python's optional typing features (in particular generics) in this example.\n This can be a little overwhelming, and you might not need that in your implementation.\n So whenever, you see `TpcpClass[SomeClassName]` and you don't understand what it means, you can safely\n ignore it.\n But just for your understanding, if you see for example `Pipeline[ECGExampleData]` you should mentally\n read it as \"A pipeline that requires a :class:`~tpcp.Dataset` of type `ECGExampleData` internally.\n Whenever you encounter a variable ending with a `T` (e.g. `PipelineT`), these are TypeVar types to type\n generics.\n You should read that as \"Some subclass of :class:`~tpcp.Pipeline`, but we don't know which yet\".

\n\n\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "collapsed": false }, "outputs": [], "source": [ "from pathlib import Path\nfrom typing import Any, Callable, Optional, Sequence, Tuple, Union\n\nimport pandas as pd\n\nfrom examples.algorithms.algorithms_qrs_detection_final import QRSDetector\nfrom examples.datasets.datasets_final_ecg import ECGExampleData\nfrom tpcp import Parameter, Pipeline, cf\n\ntry:\n HERE = Path(__file__).parent\nexcept NameError:\n HERE = Path(\".\").resolve()\ndata_path = HERE.parent.parent / \"example_data/ecg_mit_bih_arrhythmia/data\"\n\n# The dataset\nexample_data = ECGExampleData(data_path)\n\n\nclass MyPipeline(Pipeline[ECGExampleData]):\n algorithm: Parameter[QRSDetector]\n\n r_peak_positions_: pd.Series\n\n def __init__(self, algorithm: QRSDetector = cf(QRSDetector())):\n self.algorithm = algorithm\n\n def run(self, datapoint: ECGExampleData):\n # Note: We need to clone the algorithm instance to make sure we don't leak any data between runs.\n algo = self.algorithm.clone()\n algo.detect(datapoint.data[\"ecg\"], datapoint.sampling_rate_hz)\n\n self.r_peak_positions_ = algo.r_peak_positions_\n return self\n\n\n# The pipeline\npipe = MyPipeline()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## What We Want To Do\nIn the [GridSearch Example](grid_search)_, we already performed a gridsearch using the\ntpcp-:class:`~tpcp.optimize.GridSearch` class.\nHere, we want to do something similar, but improve the gridsearch in two key aspects:\n\n1. Instead of doing an exhaustive gridsearch, we use one of Optuna's advanced samplers\n2. When we encounter a parameter combination that doesn't work, we want to stop testing as early as possible to not\n waste any time on bad parameter combinations.\n\nWe will start by implementing the first aspect and will then make some modifications to enable the second.\n\nThe first thing we need for any gridsearch is a score function that tells us how good our parameter combination works.\nIn the [GridSearch Example](grid_search)_ we used a scorer that returns *accuracy*, *precision* and *f1-score*.\nWe will use basically the same function here, but only return the f1-score, as this is the parameter we want to\noptimize.\nWe could still calculate and return multiple other scores, but this would complicate the implementation of our\nOptimizer and hence, is kept as exercise for the reader ;) .\n\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "collapsed": false }, "outputs": [], "source": [ "from examples.algorithms.algorithms_qrs_detection_final import match_events_with_reference, precision_recall_f1_score\n\n\ndef f1_score(pipeline: MyPipeline, datapoint: ECGExampleData) -> float:\n # We use the `safe_run` wrapper instead of just run. This is always a good idea.\n # We don't need to clone the pipeline here, as GridSearch will already clone the pipeline internally and `run`\n # will clone it again.\n pipeline = pipeline.safe_run(datapoint)\n tolerance_s = 0.02 # We just use 20 ms for this example\n matches = match_events_with_reference(\n pipeline.r_peak_positions_.to_numpy(),\n datapoint.r_peak_positions_.to_numpy(),\n tolerance=tolerance_s * datapoint.sampling_rate_hz,\n )\n *_, f1_score_ = precision_recall_f1_score(matches)\n return f1_score_\n\n\nfrom optuna import Study, Trial" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## The Custom Optimizer\nOptimizers in `tpcp` are nothing magical \u2013 they are simply algorithms that take a pipeline as input parameter and\nhave an action method called `optimize` that takes in a dataset and then optimizes some parameters of the passed\npipeline using this data.\n\nThe :class:`~tpcp.optimize.optuna.CustomOptunaOptimize` class already implements most of that for us and simply\nrequires us to implement our objective function for the optimization like we would need for Optuna anyway.\n\nHere, we define the objective function within the `create_objective` method of our custom optimizer and return it.\nThe objective we define here is slightly different from the pure Optuna objective function, as it also takes a\n:class:`~tpcp.Pipeline` and a :class:`~tpcp.Dataset` as input in addition to the trial-object.\n\nThe content of our objective function is very similar to our score function, but we do not expect just a single\ndatapoint, but an entire dataset. Also, we need to handle getting and applying our parameters within the objective\nfunction.\n\nBecause we define the function nested within another method, we have access to all class parameters.\nHence, if we want to add certain configurations to our objective, we can add parameters to the\n:class:`~tpcp.optimize.Optimizer` itself and then access it in the objective function.\n\nFor the :class:`~tpcp.optimize.Optimizer` we want to build we primarily need two custom pieces of configuration:\n\n1. The score function we want to use. We want to make that configurable and not hard-code \"f1-score\" into our\n optimizer.\n2. The search space for the parameter search. In Optuna the search space is defined by calls to methods on a\n :class:`optuna.Trial` object. Therefore, we take in a callable that gets the trial object passed and returns\n the selected parameters. You will see how this works later on.\n\nWith these two pieces of configuration in place our objective needs to simply do four things:\n1. First, we need to call the search space function to get the parameters.\n2. Then, apply these parameters to our pipeline.\n3. Afterwards, we need to calculate how good the pipeline with the new parameters works for each of the datapoints\n within our test dataset.\n4. Finally, we return the aggregated score.\n\nTo avoid writing our own for-loop (for now) for the third step, we use :class:`~tpcp.validate.Scorer` with our\ncustom score function.\n:class:`~tpcp.validate.Scorer` handles looping and aggregating results over multiple datapoints.\n\nWith that, our implementation looks as follows:\n\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "collapsed": false }, "outputs": [], "source": [ "from tpcp.optimize.optuna import CustomOptunaOptimize\nfrom tpcp.types import DatasetT, PipelineT\nfrom tpcp.validate import Scorer\n\n\nclass OptunaSearch(CustomOptunaOptimize[PipelineT, DatasetT]):\n create_search_space: Parameter[Callable[[Trial], None]]\n score_function: Parameter[Callable[[PipelineT, DatasetT], float]]\n\n def __init__(\n self,\n pipeline: PipelineT,\n study: Study,\n create_search_space: Callable[[Trial], None],\n score_function: Callable[[PipelineT, DatasetT], float],\n *,\n n_trials: Optional[int] = None,\n timeout: Optional[float] = None,\n return_optimized: bool = True,\n ) -> None:\n self.create_search_space = create_search_space\n self.score_function = score_function\n super().__init__(pipeline, study, n_trials=n_trials, timeout=timeout, return_optimized=return_optimized)\n\n def create_objective(self) -> Callable[[Trial, PipelineT, DatasetT], Union[float, Sequence[float]]]:\n # Here we define our objective function\n\n def objective(trial: Trial, pipeline: PipelineT, dataset: DatasetT) -> float:\n # First we need to select parameters for the current trial\n self.create_search_space(trial)\n # Then we apply these parameters to the pipeline\n pipeline = pipeline.set_params(**trial.params)\n\n # We wrap the score function with a scorer to avoid writing our own for-loop to aggregate the results.\n scorer = Scorer(self.score_function)\n\n # In the end, we calculate the results per datapoint.\n # Note that we could expose the `error_score` parameter on an optimizer level.\n # But let's keep it simple for now.\n average_score, single_scores = scorer(pipeline, dataset)\n\n # As a bonus, we use the custom params option of optuna to store the individual scores per datapoint and the\n # respective data labels\n trial.set_user_attr(\"single_scores\", single_scores)\n trial.set_user_attr(\"data_labels\", dataset.groups)\n\n return average_score\n\n return objective" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Running the optimization\nTo run the optimization, we need to create a new Optuna study, a custom sampler and the function that defines our\nsearch space:\n\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "collapsed": false }, "outputs": [], "source": [ "from optuna import create_study, samplers\n\n# We use a simple RandomSampler, but every optuna sampler will work\nsampler = samplers.RandomSampler(seed=42)\n# We use a simple in-memory study with the direction \"maximize\", as we want to optimize for the highest f1-score\nstudy = create_study(direction=\"maximize\", sampler=sampler)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "The search space function requires a little more explanation:\nIn Optuna, we can use the `suggest_...` methods on a trial to get a new value within a given range.\nThis uses our sampler in the background to suggest a new value that makes sense based on the trials that are already\ncompleted.\nThe selected parameters are stored in the trial object so that we can access them after the function was called.\n\nWe use the names of the parameters we want to modify in our pipeline (using the `__` for nested values).\nThis makes applying the parameters to the pipeline later on easy.\n\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "collapsed": false }, "outputs": [], "source": [ "def create_search_space(trial: Trial):\n trial.suggest_float(\"algorithm__min_r_peak_height_over_baseline\", 0.1, 2, step=0.1)\n trial.suggest_float(\"algorithm__high_pass_filter_cutoff_hz\", 0.1, 2, step=0.1)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Finally, we are ready to run the pipeline.\nWe create a new instance and set the stopping criteria (in this case 10 random trials).\nThen we can use the familiar :class:`~tpcp.optimize.Optimize` interface to run everything.\n\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "collapsed": false }, "outputs": [], "source": [ "opti = OptunaSearch(\n pipe,\n study,\n create_search_space=create_search_space,\n score_function=f1_score,\n n_trials=10,\n)\n\nopti = opti.optimize(example_data)\nprint(\n f\"The best performance was achieved with the parameters {opti.best_params_} and an f1-score of {opti.best_score_}.\"\n)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "We can use `opti.search_results_` to get a full overview over all results.\nThese parameters and parameter names are slightly modified compared to the normal Optuna output, to make it similar\nto the output of :class:`~tpcp.optimize.GridSearch`.\n\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "collapsed": false }, "outputs": [], "source": [ "pd.DataFrame(opti.search_results_)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "If you need even more insides, you can access the study object directly.\n\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "collapsed": false }, "outputs": [], "source": [ "opti.study_" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "And like with all Optimizers, we can access the `optimized_pipeline_` and call `run` directly on the Optimizer.\n\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "collapsed": false }, "outputs": [], "source": [ "opti.optimized_pipeline_" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "collapsed": false }, "outputs": [], "source": [ "out_pipe = opti.run(example_data[0])\nout_pipe.r_peak_positions_" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "With this we created a simple random search optimizer (or grid search, or whatever sampler we want to use) using\nOptuna.\nBy using :class:`~tpcp.optimize.optuna.CustomOptunaOptimize` we get compatibility with the `tpcp` optimizer interface.\nThis means we could throw this optimizer into :func:`~tpcp.validate.cross_validate` and things would just work.\n\n## A step further: Custom Pruning\nSimply to demonstrate the power of having full access to all Optuna features, we will implement a custom pruner\nthat stops testing a trial when one datapoint scores below a certain threshold.\nThe idea is that when we iterate through the datapoints and process them one by one, and find one datapoint where\nthe performance is really bad, we already know that this will not be our best choice/a choice we want to use.\nHence, there is no need to compute scores for the remaining datapoints.\n\nIn Optuna we can implement this using a custom Pruner and the *callback* feature of the\n:class:`~tpcp.validate.Scorer` class.\nThe Pruner will be called everytime we report a new result and will tell us if we should stop evaluating the trial.\n\n

Note

This a unusual usage of pruning. Usually, pruning is used to stop after a certain number of training\n epochs of an ML classifier and not to stop half-way through evaluating your dataset.\n But it works and is practical.

\n\nTo create our custom pruner we need a new class, sub-classing :class:`~optuna.pruners.BasePruner`.\nThen we implement a `prune` method that simply checks if the current intermediate value is below a certain threshold.\nIf yes we return True, telling optuna that the trial can be pruned.\n\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "collapsed": false }, "outputs": [], "source": [ "from optuna.pruners import BasePruner\nfrom optuna.study.study import Study\nfrom optuna.trial import FrozenTrial\n\n\nclass MinDatapointPerformancePruner(BasePruner):\n def __init__(self, min_performance: float):\n self.min_performance = min_performance\n\n def prune(self, _: Study, trial: FrozenTrial) -> bool:\n step = trial.last_step\n\n if step is not None:\n score = trial.intermediate_values[step]\n if score < self.min_performance:\n return True\n return False" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Afterwards, we need to modify our optimizer to work with the pruner.\nWe need to report each calculated value from each datapoint to Optuna as soon as it was calculated and not wait\nuntil we ran through the entire dataset.\nWe can do that by passing a callback function to the `Scorer`.\nThis callback will be called after each datapoint is evaluated and allows us to access the most recent score.\n\nWe define this callback within the objective function to have access to the trial object of the outer scope.\nUsing the `trial` object, we can report the most recent score to Optuna using `trial.report`.\nThis will call the pruner and allows us to check afterwards, if the trial should be pruned.\nWe then write some debug information and end the trial by raising a :class:`~optuna.TrialPruned` exception.\n\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "collapsed": false }, "outputs": [], "source": [ "from optuna import TrialPruned\n\n\nclass OptunaSearchEarlyStopping(CustomOptunaOptimize[PipelineT, DatasetT]):\n create_search_space: Parameter[Callable[[Trial], None]]\n score_function: Parameter[Callable[[PipelineT, DatasetT], float]]\n\n def __init__(\n self,\n pipeline: PipelineT,\n study: Study,\n create_search_space: Callable[[Trial], None],\n score_function: Callable[[PipelineT, DatasetT], float],\n *,\n n_trials: Optional[int] = None,\n timeout: Optional[float] = None,\n return_optimized: bool = True,\n ) -> None:\n self.create_search_space = create_search_space\n self.score_function = score_function\n super().__init__(pipeline, study, n_trials=n_trials, timeout=timeout, return_optimized=return_optimized)\n\n def create_objective(self) -> Callable[[Trial, PipelineT, DatasetT], Union[float, Sequence[float]]]:\n def objective(trial: Trial, pipeline: PipelineT, dataset: DatasetT) -> float:\n # First, we need to select parameters for the current trial\n self.create_search_space(trial)\n # Then, we apply these parameters to the pipeline\n pipeline = pipeline.set_params(**trial.params)\n\n def single_score_callback(*, step: int, dataset: DatasetT, scores: Tuple[float, ...], **_: Any):\n # We need to report the new score value.\n # This will call the pruner internally and then tell us if we should stop\n trial.report(float(scores[step]), step)\n if trial.should_prune():\n # Apparently, our last value was bad, and we should abort.\n # However, before we do so, we will save the scores so far as debug information\n trial.set_user_attr(\"single_scores\", scores)\n trial.set_user_attr(\"data_labels\", dataset[: step + 1].groups)\n # And, finally, we abort the trial\n raise TrialPruned(\n f\"Pruned at datapoint {step} ({dataset[step].groups[0]}) with value \" f\"{scores[step]}.\"\n )\n\n # We wrap the score function with a Scorer object to avoid writing our own for-loop to aggregate the\n # results. We pass our callback and `trial` which is passed as a generic kwarg to scorer and hence can be\n # accessed from within our callback.\n scorer = Scorer(self.score_function, single_score_callback=single_score_callback)\n\n # Calculate the results per datapoint.\n # Note that we could expose the `error_score` parameter on an optimizer level.\n # But let's keep it simple for now.\n average_score, single_scores = scorer(pipeline, dataset)\n\n # As a bonus, we use the custom params option of Optuna to store the individual scores per datapoint and the\n # respective data labels.\n trial.set_user_attr(\"single_scores\", single_scores)\n trial.set_user_attr(\"data_labels\", dataset.groups)\n\n return average_score\n\n return objective" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Running the new Optimizer stays the same (we even reuse the search space).\nWe only need to add an instance of our pruner to the study.\n\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "collapsed": false }, "outputs": [], "source": [ "sampler = samplers.RandomSampler(seed=42)\nstudy = create_study(direction=\"maximize\", sampler=sampler, pruner=MinDatapointPerformancePruner(0.3))\n\nopti_early_stop = OptunaSearchEarlyStopping(\n pipe,\n study,\n create_search_space=create_search_space,\n score_function=f1_score,\n n_trials=10,\n)\n\nopti_early_stop.optimize(example_data)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "And then we can inspect the output.\nCompared to our previous run, we can see that many trials report NaN as score and \"PRUNED\" in the `state` column.\nFor each of these values we saved some time.\nFor the other trials, we get the same results as earlier.\n\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "collapsed": false }, "outputs": [], "source": [ "pd.DataFrame(opti_early_stop.search_results_)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Summary\nThe tpcp <-> Optuna interface is a little bit more low-level than many other tpcp features.\nTherefore, here is a short summary of the steps you need:\n\n1. Create a custom optimizer than inherits from `CustomOptunaOptimize`\n2. Overwrite the `create_objective` method so that it returns a Callable.\n3. The returned callable should expect a :class:`~optuna.Trial`, a :class:`~tpcp.Pipeline`,\n and a :class:`~tpcp.Dataset` object as input.\n Otherwise, it is identical to the objective function you would write in \"plain\" Optuna, and hence, should only\n return a single cost value for the optimization.\n4. If your objective function requires parameter, add them as class attributes via the init.\n5. (optional) If you want to report additional values from your optimization, you can do that via the\n `set_user_attr` parameter of the :class:`~optuna.Trial` object.\n6. (optional) Early stopping and other Pruners can be implemented identical to Optuna.\n Using the callback option of :class:`~tpcp.validate.Scorer` you can even hook into the datapoint iteration to\n trigger early stopping.\n\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Next steps\nBuilding a custom optimizer is a little more involved than just using `GridSearch`. However, it allows great\nflexibility with relatively small overhead compared to a pure implementation in Optuna.\n\nIn this example we created an objective function that only makes sense for pipelines that don't have an internal\noptimization.\nHowever, instead of just a simple search, you could also create a cross-validation-based search by using\n:func:`~tpcp.validate.cross_validate` within your objective to split the passed data into multiple train-test sets\nand optimize hyperparameters similar to :class:`~tpcp.optimize.GridSearchCV`.\n\n" ] } ], "metadata": { "kernelspec": { "display_name": "Python 3", "language": "python", "name": "python3" }, "language_info": { "codemirror_mode": { "name": "ipython", "version": 3 }, "file_extension": ".py", "mimetype": "text/x-python", "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", "version": "3.8.13" } }, "nbformat": 4, "nbformat_minor": 0 }