tessl/pypi-stheno

Implementation of Gaussian processes in Python with support for AutoGrad, TensorFlow, PyTorch, and JAX

Overview

Eval results

Files

Observations and Conditioning

Name: tessl/pypi-stheno
Author: tessl

Structured observation handling for Gaussian process conditioning, including standard exact observations and sparse approximations using inducing points. This module provides various observation types for scalable GP inference with different approximation methods.

Capabilities

Standard Observations

Exact observations that represent direct evaluations of Gaussian processes at specific input points. These provide exact Bayesian conditioning without approximations.

class Observations(AbstractObservations):
    def __init__(self, fdd, y):
        """
        Create observations from FDD and values.
        
        Parameters:
        - fdd: Finite-dimensional distribution
        - y: Observed values corresponding to fdd
        """

    def __init__(self, *pairs):
        """
        Create observations from multiple (FDD, values) pairs.
        
        Parameters:
        - *pairs: Sequence of (fdd, y) tuples
        """

    fdd: FDD    # FDD of observations
    y: Any      # Values of observations

# Convenient alias
Obs = Observations

Pseudo-Observations (Sparse Approximations)

Approximate observations using inducing points for scalable GP inference. Supports multiple approximation methods including VFE, FITC, and DTC.

class PseudoObservations(AbstractPseudoObservations):
    def __init__(self, u, fdd, y):
        """
        Create pseudo-observations with inducing points.
        
        Parameters:
        - u: Inducing points (FDD)
        - fdd: Finite-dimensional distribution of observations
        - y: Observed values
        """

    def __init__(self, us, *pairs):
        """
        Create pseudo-observations with multiple observation pairs.
        
        Parameters:
        - us: Inducing points (FDD)
        - *pairs: Sequence of (fdd, y) tuples
        """

    def elbo(self, measure):
        """
        Compute Evidence Lower BOund (ELBO) for the approximation.
        
        Parameters:
        - measure: Measure containing the prior
        
        Returns:
        - scalar: ELBO value for optimization
        """

    u: FDD      # Inducing points
    fdd: FDD    # FDD of observations  
    y: Any      # Values of observations
    method: str = "vfe"  # Approximation method

FITC Approximation

Fully Independent Training Conditional approximation that assumes conditional independence between observations given inducing points.

class PseudoObservationsFITC(PseudoObservations):
    def __init__(self, u, fdd, y):
        """FITC pseudo-observations."""
        
    def __init__(self, us, *pairs):
        """FITC pseudo-observations with multiple pairs."""
        
    method: str = "fitc"

DTC Approximation

Deterministic Training Conditional approximation that uses a low-rank approximation to the prior covariance.

class PseudoObservationsDTC(PseudoObservations):
    def __init__(self, u, fdd, y):
        """DTC pseudo-observations."""
        
    def __init__(self, us, *pairs):
        """DTC pseudo-observations with multiple pairs."""
        
    method: str = "dtc"

Convenient Aliases

Shortened names for common observation types for more concise code.

# Standard observations
Obs = Observations

# Pseudo-observations  
PseudoObs = PseudoObservations
SparseObs = PseudoObservations  # Backward compatibility

# FITC approximation
PseudoObsFITC = PseudoObservationsFITC

# DTC approximation  
PseudoObsDTC = PseudoObservationsDTC

# Backward compatibility
SparseObservations = PseudoObservations

Observation Methods

Common methods available on all observation types for posterior inference and model evaluation.

class AbstractObservations:
    def posterior_kernel(self, measure, p_i, p_j):
        """
        Get posterior kernel between processes.
        
        Parameters:
        - measure: Measure containing the processes
        - p_i: First process
        - p_j: Second process
        
        Returns:
        - Posterior kernel function
        """

    def posterior_mean(self, measure, p):
        """
        Get posterior mean for process.
        
        Parameters:
        - measure: Measure containing the process
        - p: Process to get posterior mean for
        
        Returns:
        - Posterior mean function
        """

Pseudo-Observation Specific Methods

Additional methods available on pseudo-observations for sparse approximation evaluation and optimization.

class AbstractPseudoObservations:
    def K_z(self, measure):
        """
        Get kernel matrix of inducing points.
        
        Parameters:
        - measure: Measure to evaluate kernel with
        
        Returns:
        - Kernel matrix of inducing points
        """

    def elbo(self, measure):
        """
        Compute evidence lower bound (ELBO) for variational inference.
        
        Parameters:
        - measure: Measure to compute ELBO with
        
        Returns:
        - Evidence lower bound value
        """

    def mu(self, measure):
        """
        Mean of optimal approximating distribution.
        
        Parameters:
        - measure: Measure for computation
        
        Returns:
        - Mean vector of optimal distribution
        """

    def A(self, measure):
        """
        Corrective variance parameter for approximation.
        
        Parameters:
        - measure: Measure for computation
        
        Returns:
        - Corrective variance matrix
        """

Utility Functions

Helper functions for combining and manipulating observations.

def combine(*observations):
    """
    Combine multiple FDDs or observation pairs.
    
    Parameters:
    - *observations: FDD objects or (FDD, values) pairs to combine
    
    Returns:
    - Combined observations object
    """

Usage Examples

Standard Exact Observations

import stheno
import numpy as np

# Create GP and generate synthetic data
gp = stheno.GP(kernel=stheno.EQ())
x = np.linspace(0, 2, 10)
y = np.sin(x) + 0.1 * np.random.randn(len(x))

# Create observations
fdd = gp(x, noise=0.1)
obs = stheno.Observations(fdd, y)
# or equivalently:
obs = stheno.Obs(fdd, y)

# Condition GP on observations
posterior = gp.condition(obs)

# Make predictions
x_pred = np.linspace(0, 2, 50)
pred = posterior(x_pred)
mean, lower, upper = pred.marginal_credible_bounds()

Multiple Observation Sets

# Create multiple observation sets
x1 = np.linspace(0, 1, 5)
x2 = np.linspace(1.5, 2.5, 5)
y1 = np.sin(x1) + 0.1 * np.random.randn(len(x1))
y2 = np.cos(x2) + 0.1 * np.random.randn(len(x2))

fdd1 = gp(x1, noise=0.1)
fdd2 = gp(x2, noise=0.1)

# Combine into single observation set
obs = stheno.Observations((fdd1, y1), (fdd2, y2))

# Condition on all observations simultaneously
posterior = gp.condition(obs)

Sparse Approximations with Inducing Points

# Large dataset requiring sparse approximation
n_obs = 1000
n_inducing = 50

x_obs = np.random.uniform(0, 10, n_obs)
y_obs = np.sin(x_obs) + 0.2 * np.random.randn(n_obs)

# Select inducing point locations
x_inducing = np.linspace(0, 10, n_inducing)

# Create FDDs
fdd_obs = gp(x_obs, noise=0.2)
fdd_inducing = gp(x_inducing)

# VFE approximation (default)
pseudo_obs_vfe = stheno.PseudoObservations(fdd_inducing, fdd_obs, y_obs)

# FITC approximation  
pseudo_obs_fitc = stheno.PseudoObservationsFITC(fdd_inducing, fdd_obs, y_obs)

# DTC approximation
pseudo_obs_dtc = stheno.PseudoObservationsDTC(fdd_inducing, fdd_obs, y_obs)

# Condition using sparse approximation
posterior_vfe = gp.condition(pseudo_obs_vfe)
posterior_fitc = gp.condition(pseudo_obs_fitc)
posterior_dtc = gp.condition(pseudo_obs_dtc)

Model Selection with ELBO

# Compare different numbers of inducing points using ELBO
inducing_counts = [10, 25, 50, 100]
elbos = []

for m in inducing_counts:
    x_inducing = np.linspace(0, 10, m)
    fdd_inducing = gp(x_inducing)
    pseudo_obs = stheno.PseudoObservations(fdd_inducing, fdd_obs, y_obs)
    
    # Compute ELBO for this configuration
    elbo = pseudo_obs.elbo(gp.measure)
    elbos.append(elbo)
    print(f"Inducing points: {m}, ELBO: {elbo:.3f}")

# Select best configuration
best_m = inducing_counts[np.argmax(elbos)]
print(f"Best number of inducing points: {best_m}")

Advanced Sparse GP Usage

# Multi-output sparse GP
gp1 = stheno.GP(kernel=stheno.EQ(), name="output1")
gp2 = stheno.GP(kernel=stheno.Matern52(), name="output2")

# Shared inducing points for both outputs
x_inducing = np.linspace(0, 5, 30)
u1 = gp1(x_inducing)
u2 = gp2(x_inducing)

# Observations for each output
x1_obs = np.random.uniform(0, 5, 200)
x2_obs = np.random.uniform(0, 5, 150)
y1_obs = np.sin(x1_obs) + 0.1 * np.random.randn(len(x1_obs))
y2_obs = np.cos(x2_obs) + 0.1 * np.random.randn(len(x2_obs))

# Create sparse observations for each output
sparse_obs1 = stheno.PseudoObservations(u1, gp1(x1_obs), y1_obs)
sparse_obs2 = stheno.PseudoObservations(u2, gp2(x2_obs), y2_obs)

# Condition both processes
posterior1 = gp1.condition(sparse_obs1)
posterior2 = gp2.condition(sparse_obs2)

Combining Exact and Sparse Observations

# High-quality observations (exact)
x_exact = np.linspace(0, 1, 10)
y_exact = np.sin(x_exact) + 0.05 * np.random.randn(len(x_exact))
obs_exact = stheno.Observations(gp(x_exact, noise=0.05), y_exact)

# Large-scale noisy observations (sparse)
x_sparse = np.random.uniform(1, 5, 500)
y_sparse = np.sin(x_sparse) + 0.2 * np.random.randn(len(x_sparse))
x_inducing = np.linspace(1, 5, 25)
obs_sparse = stheno.PseudoObservations(
    gp(x_inducing), 
    gp(x_sparse, noise=0.2), 
    y_sparse
)

# Combine both types of observations
combined_obs = stheno.combine(obs_exact, obs_sparse)
posterior = gp.condition(combined_obs)

Accessing Approximation Properties

# Create sparse observations
pseudo_obs = stheno.PseudoObservations(fdd_inducing, fdd_obs, y_obs)

# Access approximation properties
print(f"Approximation method: {pseudo_obs.method}")
print(f"Number of inducing points: {len(pseudo_obs.u.x)}")
print(f"Number of observations: {len(pseudo_obs.y)}")

# Get approximation-specific quantities
K_z = pseudo_obs.K_z(gp.measure)  # Inducing point covariances
mu = pseudo_obs.mu(gp.measure)    # Optimal mean
A = pseudo_obs.A(gp.measure)      # Corrective variance
elbo = pseudo_obs.elbo(gp.measure)  # Evidence lower bound

print(f"ELBO: {elbo:.3f}")
print(f"Inducing covariance shape: {K_z.shape}")

Complete Sparse GP Workflow

Sparse GP Regression with ELBO Optimization

import stheno
import numpy as np

# Create GP and generate data
gp = stheno.GP(kernel=stheno.EQ())
x_obs = np.linspace(0, 10, 1000)  # Many observations
y_obs = np.sin(x_obs) + 0.1 * np.random.randn(len(x_obs))

# Choose fewer inducing points for efficiency
x_inducing = np.linspace(0, 10, 50)  # Much fewer inducing points
fdd_inducing = gp(x_inducing)
fdd_obs = gp(x_obs, noise=0.1)

# Create sparse observations using VFE approximation
sparse_obs = stheno.PseudoObs(fdd_inducing, fdd_obs, y_obs)

# Evaluate approximation quality
elbo = sparse_obs.elbo(gp.measure)
print(f"ELBO: {elbo:.2f}")

# Create posterior using sparse approximation
posterior = gp.measure.condition(sparse_obs)
post_gp = posterior(gp)

# Make predictions
x_test = np.linspace(0, 10, 200)
pred = post_gp(x_test)
mean, lower, upper = pred.marginal_credible_bounds()

print(f"Predictions computed using {len(x_inducing)} inducing points")
print(f"Instead of {len(x_obs)} full observations")

Comparing Approximation Methods

# Compare VFE, FITC, and DTC approximations
methods = [
    ("VFE", stheno.PseudoObs),
    ("FITC", stheno.PseudoObsFITC), 
    ("DTC", stheno.PseudoObsDTC)
]

for name, obs_class in methods:
    sparse_obs = obs_class(fdd_inducing, fdd_obs, y_obs)
    elbo = sparse_obs.elbo(gp.measure)
    print(f"{name} ELBO: {elbo:.2f}")

Install with Tessl CLI