Fix: Corrected the bug where metrics were averaged without accounting for missing values​

Author: yisongfu

basicts/metrics/__init__.py

@@ -1,6 +1,8 @@
+from .cls_metrics import accuracy, f1_score, precision, recall
 from .corr import masked_corr
 from .mae import masked_mae
 from .mape import masked_mape
+from .metric_meter import AvgMeter, RMSEMeter
 from .mse import masked_mse
 from .r_square import masked_r2
 from .rmse import masked_rmse
@@ -15,17 +17,32 @@
             'WAPE': masked_wape,
             'SMAPE': masked_smape,
             'R2': masked_r2,
-            'CORR': masked_corr
+            'CORR': masked_corr,
+            "accuracy": accuracy,
+            "precision": precision,
+            "recall": recall,
+            "f1": f1_score
             }
 
+METRIC_METER = {
+    'RMSE': RMSEMeter,
+    'default': AvgMeter
+}
+
 __all__ = [
     'masked_mae',
     'masked_mse',
     'masked_rmse',
+    'incremental_masked_rmse',
     'masked_mape',
     'masked_wape',
     'masked_smape',
     'masked_r2',
     'masked_corr',
-    'ALL_METRICS'
+    'accuracy',
+    'precision',
+    'recall',
+    'f1_score',
+    'ALL_METRICS',
+    'METRIC_METER'
 ]

basicts/metrics/metric_meter.py

@@ -0,0 +1,76 @@
+class AvgMeter:
+    """Average meter.
+    """
+
+    def __init__(self):
+        self._sum: float = 0.
+        self._count: int = 0
+
+    def reset(self):
+        """Reset counter.
+        """
+
+        self._sum = 0.
+        self._count = 0
+
+    def update(self, value: float, n: int = 1):
+        """Update sum and count.
+
+        Args:
+            value (float): value.
+            n (int): number.
+        """
+
+        self._sum += value * n
+        self._count += n
+
+    @property
+    def value(self) -> float:
+        """Get average value.
+
+        Returns:
+            avg (float)
+        """
+
+        return self._sum / self._count if self._count != 0 else 0
+
+
+class RMSEMeter:
+    """
+    RMSE meter.
+    This meter maintains **MSE** and calculate **RMSE** in the post process.
+    """
+
+    def __init__(self):
+        self._mse: float = 0.
+        self._count: int = 0
+
+    def reset(self):
+        """Reset counter.
+        """
+
+        self._mse = 0.
+        self._count = 0
+
+    def update(self, value: float, n: int = 1):
+        """Update sum and count.
+
+        Args:
+            value (float): value.
+            n (int): number.
+        """
+
+        self._mse += value ** 2 * n
+        self._count += n
+
+    @property
+    def value(self) -> float:
+        """Get average value.
+
+        Returns:
+            avg (float)
+        """
+
+        mse = self._mse / self._count if self._count != 0 else 0
+
+        return mse ** 0.5

basicts/runners/base_epoch_runner.py

@@ -10,7 +10,6 @@
 from easytorch.core.checkpoint import (backup_last_ckpt, clear_ckpt, load_ckpt,
                                        save_ckpt)
 from easytorch.core.data_loader import build_data_loader, build_data_loader_ddp
-from easytorch.core.meter_pool import MeterPool
 from easytorch.device import to_device
 from easytorch.utils import (TimePredictor, get_local_rank, get_logger,
                              is_master, master_only, set_env)
@@ -22,7 +21,7 @@
 from torch.utils.tensorboard import SummaryWriter
 from tqdm import tqdm
 
-from ..utils import get_dataset_name
+from ..utils import MeterPool, get_dataset_name
 from . import optim
 
 
@@ -597,7 +596,7 @@ def inference_pipeline(self, cfg: Optional[Dict] = None, input_data: Union[str,
         result = self.inference(save_result_path=output_data_file_path)
 
         inference_end_time = time.time()
-        self.update_epoch_meter('inference/time', inference_end_time - inference_start_time)
+        self.update_epoch_meter('inference/time', 'inference', inference_end_time - inference_start_time)
 
         self.print_epoch_meters('inference')
 
@@ -924,7 +923,7 @@ def save_best_model(self, epoch: int, metric_name: str, greater_best: bool = Tru
                 `False` means lower value is best, such as `loss`. Defaults to True.
         """
 
-        metric = self.meter_pool.get_avg(metric_name)
+        metric = self.meter_pool.get_value(metric_name)
         best_metric = self.best_metrics.get(metric_name)
         if best_metric is None or (metric > best_metric if greater_best else metric < best_metric):
             self.best_metrics[metric_name] = metric

basicts/runners/base_tsf_runner.py

@@ -74,7 +74,7 @@ def __init__(self, cfg: Dict):
         # define metrics
         self.metrics = cfg.get('METRICS', {}).get('FUNCS', {
                                                             'MAE': masked_mae, 
-                                                            'RMSE': masked_rmse, 
+                                                            'RMSE': masked_rmse,
                                                             'MAPE': masked_mape, 
                                                             'WAPE': masked_wape, 
                                                             'MSE': masked_mse
@@ -376,7 +376,7 @@ def train_iters(self, epoch: int, iter_index: int, data: Union[torch.Tensor, Tup
 
         for metric_name, metric_func in self.metrics.items():
             metric_item = self.metric_forward(metric_func, forward_return)
-            self.update_epoch_meter(f'train/{metric_name}', metric_item.item())
+            self.update_epoch_meter(f'train/{metric_name}', metric_item.item(), weight)
         return loss
 
     def val_iters(self, iter_index: int, data: Union[torch.Tensor, Tuple]):
@@ -432,22 +432,23 @@ def test(self, train_epoch: Optional[int] = None, save_metrics: bool = False, sa
             for i in self.evaluation_horizons:
                 pred_h = pred[:, i, :, :]
                 target_h = target[:, i, :, :]
+                weight_h = self._get_metric_weight(target_h)
 
                 for metric_name, metric_func in self.metrics.items():
                     if metric_name.lower() == 'mase':
                         continue  # MASE needs to be calculated after all horizons
                     metric_val = self.metric_forward(metric_func, {'prediction': pred_h, 'target': target_h})
-                    self.update_epoch_meter(f'test/{metric_name}@h{i+1}', metric_val.item(), weight)
+                    self.update_epoch_meter(f'test/{metric_name}@h{i+1}', metric_val.item(), weight_h)
 
             for metric_name, metric_func in self.metrics.items():
                 metric_item = self.metric_forward(metric_func, {'prediction': pred, 'target': target})
                 self.update_epoch_meter(f'test/{metric_name}', metric_item.item(), weight)
 
         if save_metrics:
             metrics_results = {}
-            metrics_results['overall'] = {k: self.meter_pool.get_avg(f'test/{k}') for k in self.metrics.keys()}
+            metrics_results['overall'] = {k: self.meter_pool.get_value(f'test/{k}') for k in self.metrics.keys()}
             for i in self.evaluation_horizons:
-                metrics_results[f'horizon_{i+1}'] = {k: self.meter_pool.get_avg(f'test/{k}@h{i+1}') for k in self.metrics.keys()}
+                metrics_results[f'horizon_{i+1}'] = {k: self.meter_pool.get_value(f'test/{k}@h{i+1}') for k in self.metrics.keys()}
 
             # save metrics_results to self.ckpt_save_dir/test_metrics.json
             with open(os.path.join(self.ckpt_save_dir, 'test_metrics.json'), 'w') as f:
@@ -553,18 +554,14 @@ def _save_test_results(self, batch_idx: int, batch_data: Dict[str, np.ndarray])
     def _get_metric_weight(self, x: torch.Tensor) -> int:
         """
         Get the weight for calculating metrics.
-        1. Since the last batch may be smaller (`drop_last=False`), it is necessary to perform a weighted average based on the batch size.
-        2. Since the number of valid values in each batch may vary, a weighted average based on the valid value count is also required.
-           Valid value count is the total count minus the number of missing values.
-        The weight is the product of the batch size and the valid value count.
+        Since the number of valid values in each batch may vary, it is necessary to perform a weighted average based on the valid value count.
+        The valid value count is the total count minus the number of missing values.
         """
 
-        batch_size = x.shape[0]
-
         if self.null_val == np.nan:
             valid_num = (~torch.isnan(x)).sum().item()
         else:
             eps = 5e-5
             valid_num = (~torch.isclose(x, torch.tensor(self.null_val).expand_as(x).to(x.device), atol=eps, rtol=0.0)).sum().item()
 
-        return batch_size * valid_num
+        return valid_num

basicts/utils/meter_pool.py

@@ -0,0 +1,104 @@
+import logging
+from typing import Any, Dict, Tuple, Union
+
+from torch.utils.tensorboard import SummaryWriter
+
+from ..metrics import METRIC_METER
+
+
+class MeterPool:
+    """Meter container
+    """
+
+    def __init__(self):
+        self._pool: Dict[str, Dict[str, Any]] = {}
+
+    def register(self, name: str, meter_type: str, fmt: str = '{:f}', plt: bool = True):
+        """Add a meter to meter pool.
+        Args:
+            name (str): meter name.
+            meter_type (str): meter type.
+            fmt (str): meter output format.
+            plt (bool): set ```True``` to plot it in tensorboard
+                when calling ```plt_meters```.
+        """
+
+        if name in self._pool:
+            raise ValueError(f'Meter {name} already existed.')
+
+        # name: type/metric or type/metric@h{i}
+        metric = name.split('/')[1].split('@')[0] # get the metric name
+        handle_meter = 'default' if metric not in METRIC_METER else metric
+
+        self._pool[name] = {
+            'meter': METRIC_METER[handle_meter](),
+            'index': len(self._pool.keys()),
+            'format': fmt,
+            'type': meter_type,
+            'plt': plt
+        }
+
+    def update(self, name: str, value: Union[float, Tuple[float]] , n: int = 1):
+        """Update average meter.
+
+        Args:
+            name (str): meter name.
+            value (Union[float, Tuple[float]]): value.
+            n: (int): num.
+        """
+
+        self._pool[name]['meter'].update(value, n)
+
+    def get_value(self, name: str) -> float:
+        """Get value.
+
+        Args:
+            name (str): meter name.
+
+        Returns:
+            avg (float)
+        """
+
+        return self._pool[name]['meter'].value
+
+    def print_meters(self, meter_type: str, logger: logging.Logger = None):
+        """Print the specified type of meters.
+
+        Args:
+            meter_type (str): meter type
+            logger (logging.Logger): logger
+        """
+
+        print_list = []
+        for i in range(len(self._pool.keys())):
+            for name, value in self._pool.items():
+                if value['index'] == i and value['type'] == meter_type:
+                    print_list.append(
+                        ('{}: ' + value['format']).format(name, value['meter'].value)
+                    )
+        print_str = 'Result <{}>: [{}]'.format(meter_type, ', '.join(print_list))
+        if logger is None:
+            print(print_str)
+        else:
+            logger.info(print_str)
+
+    def plt_meters(self, meter_type: str, step: int, tensorboard_writer: SummaryWriter):
+        """Plot the specified type of meters in tensorboard.
+
+        Args:
+            meter_type (str): meter type.
+            step (int): Global step value to record
+            tensorboard_writer (SummaryWriter): tensorboard SummaryWriter
+        """
+
+        for name, value in self._pool.items():
+            if value['plt'] and value['type'] == meter_type:
+                tensorboard_writer.add_scalar(name, value['meter'].value, global_step=step)
+        tensorboard_writer.flush()
+
+    def reset(self):
+        """Reset all meters.
+        """
+
+        for _, value in self._pool.items():
+            value['meter'].reset()