Repository: microsoft/qlib
File: qlib/workflow/online/utils.py (Python)

Code before:
        self.logger.info(f"Set {len(recorder)} models to '{tag}'.")

    def get_online_tag(self, recorder: Recorder) -> str:
        """
        Given a model recorder and return its online tag.

        Args:
            recorder (Recorder): an instance of recorder

        Returns:
            str: the online tag
        """
        tags = recorder.list_tags()
        return tags.get(self.ONLINE_KEY, self.OFFLINE_TAG)

    def reset_online_tag(self, recorder: Union[Recorder, List], exp_name: str = None):
        """
        Offline all models and set the recorders to 'online'.

        Args:
            recorder (Union[Recorder, List]):
                the recorder you want to reset to 'online'.
            exp_name (str): the experiment name. If None, then use default_exp_name.

        """
        if exp_name is None:
            if self.default_exp_name is None:
                raise ValueError(
                    "Both default_exp_name and exp_name are None. OnlineToolR needs a specific experiment."
                )
            exp_name = self.default_exp_name
        if isinstance(recorder, Recorder):
            recorder = [recorder]
        recs = list_recorders(exp_name)
        self.set_online_tag(self.OFFLINE_TAG, list(recs.values()))
        self.set_online_tag(self.ONLINE_TAG, recorder)

    def online_models(self, exp_name: str = None) -> list:
        """
        Get current `online` models

        Args:
            exp_name (str): the experiment name. If None, then use default_exp_name.

        Returns:
            list: a list of `online` models.
        """
        if exp_name is None:
            if self.default_exp_name is None:
                raise ValueError(
                    "Both default_exp_name and exp_name are None. OnlineToolR needs a specific experiment."

Review comment (suggestion):
This piece of code appears too many times.
Please create a new function like  `def _get_exp_name(exp_name) -> str`  (error may be raised in this method)

Code after:
        self.logger.info(f"Set {len(recorder)} models to '{tag}'.")

    def get_online_tag(self, recorder: Recorder) -> str:
        """
        Given a model recorder and return its online tag.

        Args:
            recorder (Recorder): an instance of recorder

        Returns:
            str: the online tag
        """
        tags = recorder.list_tags()
        return tags.get(self.ONLINE_KEY, self.OFFLINE_TAG)

    def reset_online_tag(self, recorder: Union[Recorder, List], exp_name: str = None):
        """
        Offline all models and set the recorders to 'online'.

        Args:
            recorder (Union[Recorder, List]):
                the recorder you want to reset to 'online'.
            exp_name (str): the experiment name. If None, then use default_exp_name.

        """
        exp_name = self._get_exp_name(exp_name)
        if isinstance(recorder, Recorder):
            recorder = [recorder]
        recs = list_recorders(exp_name)
        self.set_online_tag(self.OFFLINE_TAG, list(recs.values()))
        self.set_online_tag(self.ONLINE_TAG, recorder)

    def online_models(self, exp_name: str = None) -> list:
        """
        Get current `online` models

        Args:
            exp_name (str): the experiment name. If None, then use default_exp_name.

        Returns:
            list: a list of `online` models.
        """
        exp_name = self._get_exp_name(exp_name)
        return list(list_recorders(exp_name, lambda rec: self.get_online_tag(rec) == self.ONLINE_TAG).values())

    def update_online_pred(self, to_date=None, exp_name: str = None):
        """
        Update the predictions of online models to to_date.

        Args:
            to_date (pd.Timestamp): the pred before this date will be updated. None for updating to latest time in Calendar.