Apply formatter

.flake8

@@ -4,3 +4,4 @@
  exclude =
      venv
      build
+ ignore=E203

.pre-commit-config.yaml

@@ -6,7 +6,7 @@ repos:
         args: [--show-error-codes]
         name: mypy check
         files: tpe
-  - repo: https://gitlab.com/pycqa/flake8
+  - repo: https://github.com/pycqa/flake8
     rev: 3.8.3
     hooks:
       - id: flake8

tpe/optimizer/base_optimizer.py

@@ -1,5 +1,7 @@
+from __future__ import annotations
+
 from abc import ABCMeta, abstractmethod
-from typing import Any, Dict, Optional, Protocol, Tuple
+from typing import Any, Protocol
 
 import ConfigSpace as CS
 
@@ -10,12 +12,12 @@
 
 
 class ObjectiveFunc(Protocol):
-    def __call__(self, eval_config: Dict[str, Any]) -> float:
+    def __call__(self, eval_config: dict[str, Any]) -> float:
         """
         Objective func prototype.
 
         Args:
-            eval_config (Dict[str, Any]):
+            eval_config (dict[str, Any]):
                 A configuration after the reversion.
 
         Returns:
@@ -33,7 +35,7 @@ def __init__(
         resultfile: str,
         n_init: int,
         max_evals: int,
-        seed: Optional[int],
+        seed: int | None,
         metric_name: str,
         only_requirements: bool,
     ):
@@ -45,10 +47,10 @@ def __init__(
             obj_func (Callable): The objective function
             hp_names (List[str]): The list of hyperparameter names
             metric_name (str): The name of the metric (or objective function value)
-            observations (Dict[str, Any]): The storage of the observations
+            observations (dict[str, Any]): The storage of the observations
             config_space (CS.ConfigurationSpace): The searching space of the task
-            is_categoricals (Dict[str, bool]): Whether the given hyperparameter is categorical
-            is_ordinals (Dict[str, bool]): Whether the given hyperparameter is ordinal
+            is_categoricals (dict[str, bool]): Whether the given hyperparameter is categorical
+            is_ordinals (dict[str, bool]): Whether the given hyperparameter is ordinal
             only_requirements (bool): If True, we only save runtime and loss.
         """
 
@@ -70,16 +72,16 @@ def __init__(
             for hp_name in self._hp_names
         }
 
-    def optimize(self, logger_name: Optional[str] = None) -> Tuple[Dict[str, Any], float]:
+    def optimize(self, logger_name: str | None = None) -> tuple[dict[str, Any], float]:
         """
         Optimize obj_func using TPE Sampler and store the results in the end.
 
         Args:
-            logger_name (Optional[str]):
+            logger_name (str | None):
                 The name of logger to write the intermediate results
 
         Returns:
-            best_config (Dict[str, Any]): The configuration that has the best loss
+            best_config (dict[str, Any]): The configuration that has the best loss
             best_loss (float): The best loss value during the optimization
         """
         use_logger = logger_name is not None
@@ -116,46 +118,45 @@ def optimize(self, logger_name: Optional[str] = None) -> Tuple[Dict[str, Any], f
         return best_config, best_loss
 
     @abstractmethod
-    def update(self, eval_config: Dict[str, Any], loss: float) -> None:
+    def update(self, eval_config: dict[str, Any], loss: float) -> None:
         """
         Update of the child sampler.
 
         Args:
-            eval_config (Dict[str, Any]): The configuration to be evaluated
+            eval_config (dict[str, Any]): The configuration to be evaluated
             loss (float):
                 loss function value.
         """
         raise NotImplementedError
 
     @abstractmethod
-    def fetch_observations(self) -> Dict[str, np.ndarray]:
+    def fetch_observations(self) -> dict[str, np.ndarray]:
         """
         Fetch observations of this optimization.
 
         Returns:
-            observations (Dict[str, np.ndarray]):
+            observations (dict[str, np.ndarray]):
                 observations of this optimization.
         """
         raise NotImplementedError
 
     @abstractmethod
-    def sample(self) -> Dict[str, Any]:
+    def sample(self) -> dict[str, Any]:
         """
         Sample a configuration using a child class instance sampler
 
         Returns:
-            eval_config (Dict[str, Any]): A sampled configuration
+            eval_config (dict[str, Any]): A sampled configuration
         """
         raise NotImplementedError
 
-    def initial_sample(self) -> Dict[str, Any]:
+    def initial_sample(self) -> dict[str, Any]:
         """
         Sampling method up to n_init configurations
 
         Returns:
-            samples (Dict[str, Any]):
+            samples (dict[str, Any]):
                 Typically randomly sampled configurations
-
         """
         eval_config = {hp_name: self._get_random_sample(hp_name=hp_name) for hp_name in self._hp_names}
         return self._revert_eval_config(eval_config=eval_config)
@@ -169,7 +170,7 @@ def _get_random_sample(self, hp_name: str) -> NumericType:
             is_ordinal=self._is_ordinals[hp_name],
         )
 
-    def _revert_eval_config(self, eval_config: Dict[str, NumericType]) -> Dict[str, Any]:
+    def _revert_eval_config(self, eval_config: dict[str, NumericType]) -> dict[str, Any]:
         return revert_eval_config(
             eval_config=eval_config,
             config_space=self._config_space,

tpe/optimizer/tpe.py

@@ -1,4 +1,7 @@
-from typing import Any, Callable, Dict, Literal, Optional, Tuple, Type
+from __future__ import annotations
+
+from collections.abc import Callable
+from typing import Any, Literal
 
 import ConfigSpace as CS
 
@@ -22,15 +25,15 @@ def __init__(
         weight_func: WeightFuncType,
         n_ei_candidates: int,
         metric_name: str,
-        seed: Optional[int],
+        seed: int | None,
         min_bandwidth_factor: float,
-        min_bandwidth_factor_for_discrete: Optional[float],
-        top: Optional[float],
+        min_bandwidth_factor_for_discrete: float | None,
+        top: float | None,
         multivariate: bool,
         magic_clip: bool,
         magic_clip_exponent: float,
         prior: bool,
-        heuristic: Optional[Literal["optuna", "hyperopt"]] = None,
+        heuristic: Literal["optuna", "hyperopt"] | None = None,
     ):
         """
         Attributes:
@@ -39,19 +42,19 @@ def __init__(
             config_space (CS.ConfigurationSpace): The searching space of the task
             hp_names (List[str]): The list of hyperparameter names
             metric_name (str): The name of the metric (or objective function value)
-            observations (Dict[str, Any]): The storage of the observations
-            sorted_observations (Dict[str, Any]): The storage of the observations sorted based on loss
+            observations (dict[str, Any]): The storage of the observations
+            sorted_observations (dict[str, Any]): The storage of the observations sorted based on loss
             min_bandwidth_factor (float): The minimum bandwidth for continuous.
-            min_bandwidth_factor_for_discrete (Optional[float]):
+            min_bandwidth_factor_for_discrete (float | None):
                 The minimum bandwidth factor for discrete.
                 If None, it adapts so that the factor gives 0.1 after the discrete modifications.
-            top (Optional[float]):
+            top (float | None):
                 The hyperparam of the cateogircal kernel. It defines the prob of the top category.
                 If None, it adapts the parameter in a way that Optuna does.
                 top := (1 + 1/N) / (1 + c/N) where
                 c is the number of choices and N is the number of observations.
-            is_categoricals (Dict[str, bool]): Whether the given hyperparameter is categorical
-            is_ordinals (Dict[str, bool]): Whether the given hyperparameter is ordinal
+            is_categoricals (dict[str, bool]): Whether the given hyperparameter is categorical
+            is_ordinals (dict[str, bool]): Whether the given hyperparameter is ordinal
             quantile_func (Callable[[np.ndarray], int]):
                 The function that returns the number of a better group based on the total number of evaluations.
             magic_clip (bool):
@@ -96,7 +99,7 @@ def __init__(
         self._mvpe_upper: MultiVariateParzenEstimator
 
     def _insert_observations(
-        self, key: str, insert_loc: int, val: Any, is_categorical: bool, dtype: Optional[Type[np.number]] = None
+        self, key: str, insert_loc: int, val: Any, is_categorical: bool, dtype: type[np.number] | None = None
     ) -> None:
         data, sorted_data = self._observations, self._sorted_observations
 
@@ -113,12 +116,12 @@ def _insert_observations(
             data[key] = np.append(data[key], val).astype(dtype)
             sorted_data[key] = np.insert(sorted_data[key], insert_loc, val).astype(dtype)
 
-    def update_observations(self, eval_config: Dict[str, NumericType], loss: float) -> None:
+    def update_observations(self, eval_config: dict[str, NumericType], loss: float) -> None:
         """
         Update the observations for the TPE construction
 
         Args:
-            eval_config (Dict[str, NumericType]): The configuration to evaluate (after conversion)
+            eval_config (dict[str, NumericType]): The configuration to evaluate (after conversion)
             loss (float): The loss value as a result of the evaluation
         """
         sorted_losses, losses = (
@@ -147,7 +150,7 @@ def update_observations(self, eval_config: Dict[str, NumericType], loss: float)
 
         self._update_parzen_estimators()
 
-    def _calculate_weights(self) -> Tuple[np.ndarray, np.ndarray]:
+    def _calculate_weights(self) -> tuple[np.ndarray, np.ndarray]:
         sorted_loss_vals = self._sorted_observations[self._metric_name]
         n_lower, n_upper = self._n_lower, sorted_loss_vals.size - self._n_lower
         lower_vals, upper_vals = sorted_loss_vals[:n_lower], sorted_loss_vals[n_lower:]
@@ -171,8 +174,8 @@ def _calculate_weights(self) -> Tuple[np.ndarray, np.ndarray]:
 
     def _update_parzen_estimators(self) -> None:
         n_lower = self._n_lower
-        pe_lower_dict: Dict[str, ParzenEstimatorType] = {}
-        pe_upper_dict: Dict[str, ParzenEstimatorType] = {}
+        pe_lower_dict: dict[str, ParzenEstimatorType] = {}
+        pe_upper_dict: dict[str, ParzenEstimatorType] = {}
         weights_lower, weights_upper = self._calculate_weights()
         for hp_name in self._hp_names:
             is_categorical = self._is_categoricals[hp_name]
@@ -194,7 +197,7 @@ def _update_parzen_estimators(self) -> None:
         self._mvpe_lower = MultiVariateParzenEstimator(pe_lower_dict)
         self._mvpe_upper = MultiVariateParzenEstimator(pe_upper_dict)
 
-    def get_config_candidates(self) -> Dict[str, np.ndarray]:
+    def get_config_candidates(self) -> dict[str, np.ndarray]:
         """
         Since we compute the probability improvement of each objective independently,
         we need to sample the configurations in advance.
@@ -205,19 +208,19 @@ def get_config_candidates(self) -> Dict[str, np.ndarray]:
                 If None is provided, we use n_ei_candidates.
 
         Returns:
-            config_cands (Dict[str, np.ndarray]):
+            config_cands (dict[str, np.ndarray]):
                 A dict of arrays of candidates in each dimension
         """
         return self._mvpe_lower.sample(
             n_samples=self._n_ei_candidates, rng=self._rng, dim_independent=True, return_dict=True
         )
 
-    def compute_probability_improvement(self, config_cands: Dict[str, np.ndarray]) -> np.ndarray:
+    def compute_probability_improvement(self, config_cands: dict[str, np.ndarray]) -> np.ndarray:
         """
         Compute the probability improvement given configurations
 
         Args:
-            config_cands (Dict[str, np.ndarray]):
+            config_cands (dict[str, np.ndarray]):
                 The dict of candidate values for each dimension.
                 The length is the number of dimensions and
                 each array has the length of n_ei_candidates.
@@ -253,7 +256,7 @@ def _get_parzen_estimator(
         weights_upper: np.ndarray,
         hp_name: str,
         is_categorical: bool,
-    ) -> Tuple[ParzenEstimatorType, ParzenEstimatorType]:
+    ) -> tuple[ParzenEstimatorType, ParzenEstimatorType]:
         """
         Construct parzen estimators for the lower and the upper groups and return them
 
@@ -342,7 +345,7 @@ def _calculate_adapted_top(self, config: CategoricalHPType, n_observations: int)
         return (1 + 1 / N) / (1 + n_choices / N)
 
     @property
-    def observations(self) -> Dict[str, np.ndarray]:
+    def observations(self) -> dict[str, np.ndarray]:
         return {hp_name: vals.copy() for hp_name, vals in self._observations.items()}
 
     @property

tpe/optimizer/tpe_optimizer.py

@@ -1,4 +1,7 @@
-from typing import Any, Callable, Dict, Literal, Optional
+from __future__ import annotations
+
+from collections.abc import Callable
+from typing import Any, Literal
 
 import ConfigSpace as CS
 
@@ -17,13 +20,13 @@ def __init__(
         resultfile: str = "temp",
         n_init: int = 10,
         max_evals: int = 100,
-        seed: Optional[int] = None,
+        seed: int | None = None,
         metric_name: str = "loss",
         only_requirements: bool = True,
         n_ei_candidates: int = 24,
         min_bandwidth_factor: float = 1e-2,
-        min_bandwidth_factor_for_discrete: Optional[float] = None,
-        top: Optional[float] = 1.0,
+        min_bandwidth_factor_for_discrete: float | None = None,
+        top: float | None = 1.0,
         quantile_func: Callable[[int], int] = QuantileFunc(),
         weight_func_choice: Literal[
             "uniform", "older-smaller", "older-drop", "weaker-smaller", "expected-improvement"
@@ -32,7 +35,7 @@ def __init__(
         magic_clip: bool = False,
         magic_clip_exponent: float = 1.0,
         prior: bool = True,
-        heuristic: Optional[Literal["optuna", "hyperopt"]] = None,
+        heuristic: Literal["optuna", "hyperopt"] | None = None,
     ):
         """
         Args:
@@ -47,10 +50,10 @@ def __init__(
             only_requirements (bool): If True, we only save runtime and loss.
             n_ei_candidates (int): The number of samplings to optimize the EI value
             min_bandwidth_factor (float): The minimum bandwidth for numerical parameters
-            min_bandwidth_factor_for_discrete (Optional[float]):
+            min_bandwidth_factor_for_discrete (float | None):
                 The minimum bandwidth factor for discrete.
                 If None, it adapts so that the factor gives 0.1 after the discrete modifications.
-            top (Optional[float]):
+            top (float | None):
                 The hyperparam of the cateogircal kernel. It defines the prob of the top category.
                 If None, it adapts the parameter in a way that Optuna does.
                 top := (1 + 1/N) / (1 + c/N) where
@@ -87,29 +90,29 @@ def __init__(
             heuristic=heuristic,
         )
 
-    def update(self, eval_config: Dict[str, Any], loss: float) -> None:
+    def update(self, eval_config: dict[str, Any], loss: float) -> None:
         self._tpe_sampler.update_observations(eval_config=eval_config, loss=loss)
 
-    def fetch_observations(self) -> Dict[str, np.ndarray]:
+    def fetch_observations(self) -> dict[str, np.ndarray]:
         return self._tpe_sampler.observations
 
-    def _get_config_cands(self) -> Dict[str, np.ndarray]:
+    def _get_config_cands(self) -> dict[str, np.ndarray]:
         return self._tpe_sampler.get_config_candidates()
 
-    def _compute_probability_improvement(self, config_cands: Dict[str, np.ndarray]) -> np.ndarray:
+    def _compute_probability_improvement(self, config_cands: dict[str, np.ndarray]) -> np.ndarray:
         return self._tpe_sampler.compute_probability_improvement(config_cands=config_cands)
 
-    def sample(self) -> Dict[str, Any]:
+    def sample(self) -> dict[str, Any]:
         """
         Sample a configuration using tree-structured parzen estimator (TPE)
 
         Returns:
-            eval_config (Dict[str, Any]): A sampled configuration from TPE
+            eval_config (dict[str, Any]): A sampled configuration from TPE
         """
         config_cands = self._get_config_cands()
         pi_config = self._compute_probability_improvement(config_cands)
 
-        eval_config: Dict[str, Any] = {}
+        eval_config: dict[str, Any] = {}
         if self._tpe_sampler._multivariate:
             best_idx = int(np.argmax(pi_config))
             eval_config = {hp_name: config_cands[hp_name][best_idx] for dim, hp_name in enumerate(self._hp_names)}