PyPOTS

https://github.com/wenjiedu/pypots
Admin
PyPOTS is a Python toolbox for machine learning on partially-observed time series, providing...
Tokens:390,271
Snippets:1,411
Trust Score:9.2
License:BSD-3-Clause
Update:3 weeks ago
Show doc for...
Tokens:
Raw
### Environment Setup Commands

Source: https://github.com/wenjiedu/pypots/blob/main/docs/dev_testing.rst

Commands to clone the repository, install in development mode, and generate test data required before running any test.

```bash
# Clone the repository
git clone https://github.com/WenjieDu/PyPOTS.git
cd PyPOTS

# Install in development mode
pip install -e ".[dev]"

# Generate test data (required before running any test)
python tests/global_test_config.py
```

--------------------------------

### Test Environment Setup

Source: https://github.com/wenjiedu/pypots/blob/main/docs/dev_testing.rst

Command to set up the test environment, matching CI.

```bash
python tests/global_test_config.py
```

--------------------------------

### PyPOTS Quick-Start: SAITS Imputation Workflow

Source: https://github.com/wenjiedu/pypots/blob/main/docs/tutorials.rst

Use this snippet to get started with PyPOTS. It prepares the PhysioNet 2012 dataset, organizes it into PyPOTS input format, initializes a SAITS model, trains it, imputes missing values, evaluates with MAE, and saves/loads the model. Requires the pygrinder, benchpots, and sklearn libraries.

```python
import numpy as np
from sklearn.preprocessing import StandardScaler
from pygrinder import mcar, calc_missing_rate
from benchpots.datasets import preprocess_physionet2012

# prepare the dataset
data = preprocess_physionet2012(subset='set-a',rate=0.1) # Our ecosystem libs will automatically download and extract it
train_X, val_X, test_X = data["train_X"], data["val_X"], data["test_X"]
print(train_X.shape)  # (n_samples, n_steps, n_features)
print(val_X.shape)  # samples (n_samples) in train set and val set are different, but they have the same sequence len (n_steps) and feature dim (n_features)
print(f"We have {calc_missing_rate(train_X):.1%} values missing in train_X")

# organize the dataset for PyPOTS model input
train_set = {"X": train_X}  # in training set, simply put the incomplete time series into it
val_set = {
    "X": val_X,
    "X_ori": data["val_X_ori"],  # in validation set, we need ground truth for evaluation and picking the best model checkpoint
}
test_set = {"X": test_X}  # in test set, only give the testing incomplete time series for model to impute

# the test set for final evaluation
test_X_ori = data["test_X_ori"]  # test_X_ori bears ground truth for evaluation
indicating_mask = np.isnan(test_X) ^ np.isnan(test_X_ori)  # mask indicates the values that are missing in X but not in X_ori, i.e. where the gt values are

# initialize the model
_, n_steps, n_features = train_X.shape
saits = SAITS(
    n_steps=n_steps,
    n_features=n_features,
    n_layers=2,
    d_model=256,
    d_ffn=128,
    n_heads=4,
    d_k=64,
    d_v=64,
    dropout=0.1,
    epochs=10,
    saving_path="examples/saits", # set the path for saving tensorboard logging file and model checkpoint
    model_saving_strategy="best", # only save the model with the best validation performance
)

# train the model. You can also omit the val_set if you don't need to validate the model during training
saits.fit(train_set, val_set)
# impute the originally-missing values and artificially-missing values
imputation = saits.impute(test_set)
mae = calc_mae(imputation, np.nan_to_num(test_X_ori), indicating_mask)  # calculate mean absolute error on the ground truth (artificially-missing values)

# the best model has been already saved, but you can still manually save it with function save_model() as below
saits.save(saving_path="examples/saits/manually_saved_saits_model")
# you can load the saved model into a new initialized model
saits.load("examples/saits/manually_saved_saits_model.pypots")
```

--------------------------------

### Install CPU acceleration packages for Intel chips

Source: https://github.com/wenjiedu/pypots/blob/main/docs/install.rst

Conda command to install MKL distribution optimized for multi-core Intel CPUs.

```bash
conda install numpy scipy scikit-learn numexpr "libblas=*=*mkl"
```

--------------------------------

### Set up development environment with pip

Source: https://github.com/wenjiedu/pypots/blob/main/docs/dev_getting_started.rst

Clone the repository and install the package in editable mode with development dependencies.

```bash
git clone https://github.com/WenjieDu/PyPOTS.git
cd PyPOTS
pip install -e ".[dev]"
```

--------------------------------

### Install PyPOTS via pip, conda, and Docker

Source: https://github.com/wenjiedu/pypots/blob/main/docs/install.rst

Commands to install PyPOTS using pip, conda, or Docker. Includes first-time installation, updates, and running a Docker container.

```bash
# via pip
pip install pypots            # the first time installation
pip install pypots --upgrade  # update pypots to the latest version
# install from the latest source code with the latest features but may be not officially released yet
pip install https://github.com/WenjieDu/PyPOTS/archive/main.zip

# via conda
conda install conda-forge::pypots  # the first time installation
conda update  conda-forge::pypots  # update pypots to the latest version

# via docker
docker run -it --name pypots wenjiedu/pypots  # docker will auto pull our built image and run a instance for you
# after things settled, you can run python in the container to access the well-configured environment for running pypots
# if you'd like to detach from the container, press ctrl-P + ctrl-Q
# run `docker attach pypots` to enter the container again.
```

--------------------------------

### Example PR Evidence

Source: https://github.com/wenjiedu/pypots/blob/main/docs/dev_testing.rst

Example text block showing the evidence format for a review-ready PR, including environment, commands run, and results.

```text
## Testing Evidence

### Environment
- Python 3.10, PyTorch 2.1, CUDA 12.1

### Commands Run
```
python tests/global_test_config.py
pytest -rA tests/imputation/my_model.py -n 1
flake8 .
```

### Results
- All 5 tests passed
- No lint errors
- Scope: targeted (only my_model changed)
```

--------------------------------

### StepLR usage example

Source: https://github.com/wenjiedu/pypots/blob/main/docs/pypots.optim.rst

Shows how to create a StepLR scheduler and pass it to a pypots.optim.Adam optimizer. The scheduler decays the learning rate by gamma every step_size epochs. The example is marked with xdoctest: +SKIP.

```pycon
>>> # Assuming optimizer uses lr = 0.05 for all groups
>>> # lr = 0.05     if epoch < 30
>>> # lr = 0.005    if 30 <= epoch < 60
>>> # lr = 0.0005   if 60 <= epoch < 90
>>> # ...
>>> # xdoctest: +SKIP
>>> scheduler = StepLR(step_size=30, gamma=0.1)
>>> adam = pypots.optim.Adam(lr=1e-3, lr_scheduler=scheduler)
```

--------------------------------

### Implementing BaseModel

Source: https://github.com/wenjiedu/pypots/blob/main/docs/dev_base_classes.rst

Example of creating a custom model by inheriting from BaseModel and implementing the required fit and predict methods.

```python
from pypots.base import BaseModel

class MyModel(BaseModel):
    def __init__(self, device=None, saving_path=None):
        super().__init__(device=device, saving_path=saving_path)

    def fit(self, train_set, val_set=None, file_type="hdf5"):
        # Your training logic
        ...

    def predict(self, test_set, file_type="hdf5"):
        # Your inference logic
        ...
```

--------------------------------

### Set up development environment with conda

Source: https://github.com/wenjiedu/pypots/blob/main/docs/dev_getting_started.rst

Create a new conda environment, activate it, clone the repository, and install the package in editable mode with development dependencies.

```bash
conda create -n pypots python=3.10
conda activate pypots
git clone https://github.com/WenjieDu/PyPOTS.git
cd PyPOTS
pip install -e ".[dev]"
```

--------------------------------

### MultiStepLR usage example

Source: https://github.com/wenjiedu/pypots/blob/main/docs/pypots.optim.rst

Shows how to create a MultiStepLR scheduler and pass it to a pypots.optim.Adam optimizer. The scheduler decays the learning rate by gamma at the specified milestones. The example is marked with xdoctest: +SKIP.

```pycon
>>> # Assuming optimizer uses lr = 0.05 for all groups
>>> # lr = 0.05     if epoch < 30
>>> # lr = 0.005    if 30 <= epoch < 80
>>> # lr = 0.0005   if epoch >= 80
>>> # xdoctest: +SKIP
>>> scheduler = MultiStepLR(milestones=[30,80], gamma=0.1)
>>> adam = pypots.optim.Adam(lr=1e-3, lr_scheduler=scheduler)
```

--------------------------------

### Implement model.py: YourModel wrapper for PyPOTS NN imputer

Source: https://github.com/wenjiedu/pypots/blob/main/docs/dev_standard_nn.rst

Complete example of a PyPOTS NN imputation model wrapper. It inherits BaseNNImputer, instantiates the core model and optimizer, implements the three input assembly methods, and starts the training dataloader in fit(). The snippet is partial: the fit() method is cut off after creating the training dataloader, so the remaining steps (validation set, _train_model call) are not shown.

```python
# pypots/imputation/your_model/model.py

from typing import Union, Optional
import numpy as np
import torch
from torch.utils.data import DataLoader

from .core import _YourModel
from ..base import BaseNNImputer
from ...data.dataset.base import BaseDataset
from ...data.checking import key_in_data_set
from ...nn.modules.loss import Criterion, MAE, MSE
from ...optim.adam import Adam
from ...optim.base import Optimizer


class YourModel(BaseNNImputer):
    """Your model description here.

    Parameters
    ----------
    n_steps :
        The number of time steps in the time-series data sample.

    n_features :
        The number of features in the time-series data sample.

    d_model :
        The dimension of the model's backbone.

    batch_size :
        The batch size for training and evaluating the model.

    epochs :
        The number of epochs for training the model.

    patience :
        The patience for the early-stopping mechanism.

    training_loss :
        The loss function for training. Default: MAE.

    validation_metric :
        The metric function for validation. Default: MSE.

    optimizer :
        The optimizer for model training. Default: Adam.
    """

    def __init__(
        self,
        n_steps: int,
        n_features: int,
        d_model: int = 64,
        batch_size: int = 32,
        epochs: int = 100,
        patience: Optional[int] = None,
        training_loss: Union[Criterion, type] = MAE,
        validation_metric: Union[Criterion, type] = MSE,
        optimizer: Union[Optimizer, type] = Adam,
        num_workers: int = 0,
        device: Optional[Union[str, torch.device, list]] = None,
        saving_path: Optional[str] = None,
        model_saving_strategy: Optional[str] = "best",
        verbose: bool = True,
    ):
        super().__init__(
            training_loss=training_loss,
            validation_metric=validation_metric,
            batch_size=batch_size,
            epochs=epochs,
            patience=patience,
            num_workers=num_workers,
            device=device,
            saving_path=saving_path,
            model_saving_strategy=model_saving_strategy,
            verbose=verbose,
        )

        # Store hyperparameters
        self.n_steps = n_steps
        self.n_features = n_features
        self.d_model = d_model

        # Set up the model
        self.model = _YourModel(
            n_steps=n_steps,
            n_features=n_features,
            d_model=d_model,
            training_loss=self.training_loss,
            validation_metric=self.validation_metric,
        )
        self._print_model_size()
        self._send_model_to_given_device()

        # Set up the optimizer
        if isinstance(optimizer, Optimizer):
            self.optimizer = optimizer
        else:
            self.optimizer = optimizer()
            assert isinstance(self.optimizer, Optimizer)
        self.optimizer.init_optimizer(self.model.parameters())

    def _assemble_input_for_training(self, data: list) -> dict:
        (
            indices,
            X,
            missing_mask,
            X_ori,
            indicating_mask,
        ) = self._send_data_to_given_device(data)

        inputs = {
            "X": X,
            "missing_mask": missing_mask,
            "X_ori": X_ori,
            "indicating_mask": indicating_mask,
        }
        return inputs

    def _assemble_input_for_validating(self, data: list) -> dict:
        return self._assemble_input_for_training(data)

    def _assemble_input_for_testing(self, data: list) -> dict:
        indices, X, missing_mask = self._send_data_to_given_device(data)
        inputs = {
            "X": X,
            "missing_mask": missing_mask,
        }
        return inputs

    def fit(
        self,
        train_set: Union[dict, str],
        val_set: Optional[Union[dict, str]] = None,
        file_type: str = "hdf5",
    ) -> None:
        # Step 1: Create datasets and dataloaders
        training_set = DatasetForYourModel(
            train_set, return_X_ori=False, return_y=False, file_type=file_type
        )
        train_dataloader = DataLoader(
            training_set,
            batch_size=self.batch_size,
            shuffle=True,
            num_workers=self.num_workers,
        )
```

--------------------------------

### Implementing a custom model core with ModelCore

Source: https://github.com/wenjiedu/pypots/blob/main/docs/dev_base_classes.rst

This example demonstrates how to subclass ModelCore, initialize loss and metric criteria, and implement the forward() method to handle input dictionaries and conditional criterion calculation.

```python
from pypots.nn.modules import ModelCore
from pypots.nn.modules.loss import Criterion

class _MyModel(ModelCore):
    def __init__(self, training_loss: Criterion, validation_metric: Criterion):
        super().__init__()
        self.training_loss = training_loss
        self.validation_metric = validation_metric
        # Define your model's components here
        ...

    def forward(self, inputs: dict, calc_criterion: bool = False) -> dict:
        X, missing_mask = inputs["X"], inputs["missing_mask"]

        # Your model computation
        reconstruction = ...

        imputed_data = missing_mask * X + (1 - missing_mask) * reconstruction
        results = {
            "imputation": imputed_data,
            "reconstruction": reconstruction,
        }

        if calc_criterion:
            X_ori = inputs["X_ori"]
            indicating_mask = inputs["indicating_mask"]
            if self.training:
                results["loss"] = self.training_loss(
                    reconstruction, X_ori, indicating_mask
                )
            else:
                results["metric"] = self.validation_metric(
                    reconstruction, X_ori, indicating_mask
                )

        return results
```

--------------------------------

### Install CPU acceleration packages for AMD chips

Source: https://github.com/wenjiedu/pypots/blob/main/docs/install.rst

Conda command to install OpenBLAS distribution for AMD devices.

```bash
conda install -c conda-forge numpy scipy scikit-learn numexpr "libblas=*=*openblas"
```

--------------------------------

### Wire the package with __init__.py

Source: https://github.com/wenjiedu/pypots/blob/main/docs/dev_non_nn.rst

Wire the package by creating an __init__.py file that imports the model and sets __all__.

```python
# pypots/imputation/your_model/__init__.py
from .model import YourModel
__all__ = ["YourModel"]
```

--------------------------------

### Package Build

Source: https://github.com/wenjiedu/pypots/blob/main/docs/dev_testing.rst

Command to build the package, matching CI.

```bash
python -m build
```

--------------------------------

### Install CPU acceleration packages for Apple Silicon

Source: https://github.com/wenjiedu/pypots/blob/main/docs/install.rst

Conda command to install optimized data-science packages for Apple Silicon Macs.

```bash
conda install numpy scipy scikit-learn numexpr "libblas=*=*accelerate"
```

--------------------------------

### Verify result keys against BaseNNImputer contract

Source: https://github.com/wenjiedu/pypots/blob/main/docs/dev_common_mistakes.rst

Demonstrates the difference between using placeholder keys and the correct keys required by the BaseNNImputer contract.

```python
# Wrong: blindly copied from a template
results = {"output": imputed_data}

# Correct: verified against BaseNNImputer contract
results = {"imputation": imputed_data}
```

--------------------------------

### Calculate MAE with calc_mae

Source: https://github.com/wenjiedu/pypots/blob/main/docs/pypots.utils.rst

Example of using calc_mae to compute Mean Absolute Error between predictions and targets. The first example shows basic usage without masks, resulting in mae = 0.6. The second example demonstrates using masks to filter out specific elements, yielding mae = 0.5.

```pycon
>>> import numpy as np
>>> from pypots.nn.functional import calc_mae
>>> targets = np.array([1, 2, 3, 4, 5])
>>> predictions = np.array([1, 2, 1, 4, 6])
>>> mae = calc_mae(predictions, targets)
```

```pycon
>>> masks = np.array([0, 0, 0, 1, 1])
>>> mae = calc_mae(predictions, targets, masks)
```

--------------------------------

### pypots.utils.random.get_random_seed() → int

Source: https://github.com/wenjiedu/pypots/blob/main/docs/pypots.utils.rst

Get the random seed used in PyPOTS.

```APIDOC
## get_random_seed

### Description
Get the random seed used in PyPOTS.

### Method
Function

### Signature
`get_random_seed() -> int`

### Returns
- **random_seed** (int) - The random seed used in PyPOTS.
```
Tokens:
Raw