Implementation:Online ml River Base Wrapper
| Knowledge Sources | |
|---|---|
| Domains | Online_Learning, Meta_Learning, Base_Classes |
| Last Updated | 2026-02-08 16:00 GMT |
Overview
The Wrapper class is an abstract base class for meta-models that wrap and enhance single estimators with additional functionality.
Description
The Wrapper class provides the foundation for creating meta-estimators that wrap a single base estimator to add capabilities like bagging, boosting, or other transformations. It is a generic class parameterized by the type T of the wrapped estimator, ensuring type safety. The class requires implementing the _wrapped_model property to provide access to the underlying estimator, and it automatically delegates property queries like _tags, _supervised, and _multiclass to the wrapped model. The _labelloc property controls visualization placement in pipeline diagrams, and the __str__ method provides consistent string representation showing both the wrapper and wrapped model.
Usage
Use Wrapper as the parent class when implementing meta-estimators that enhance or modify the behavior of a single base model, such as adaptive learning rate wrappers, feature selection wrappers, or ensemble methods that wrap one model. The wrapper automatically inherits properties from the wrapped model, ensuring consistent behavior with River's type system and pipeline infrastructure.
Code Reference
Source Location
- Repository: Online_ml_River
- File: river/base/wrapper.py
Signature
class Wrapper(ABC, Generic[T]):
"""A wrapper model."""
@property
@abstractmethod
def _wrapped_model(self) -> T
@property
def _labelloc(self) -> str
def __str__(self) -> str
def _more_tags(self) -> set[str]
@property
def _supervised(self) -> bool
@property
def _multiclass(self) -> bool
# Type variable
T = TypeVar("T", bound=Estimator)
Import
from river.base import Wrapper
I/O Contract
Required Property
| Property | Type | Description |
|---|---|---|
| _wrapped_model | T | Must return the wrapped estimator instance |
Automatic Properties
| Property | Type | Description |
|---|---|---|
| _labelloc | str | Location of wrapper name in pipeline visualization (default: "t" for top) |
| _tags | set[str] | Tags from the wrapped model |
| _supervised | bool | Whether the wrapped model is supervised |
| _multiclass | bool | Whether the wrapped model supports multiclass (if it's a classifier) |
Usage Examples
from river.base import Wrapper, Classifier
from river import tree
from river import datasets
# Example 1: Simple wrapper that adds logging
class LoggingWrapper(Wrapper, Classifier):
def __init__(self, model):
self.model = model
self.learn_count = 0
self.predict_count = 0
@property
def _wrapped_model(self):
return self.model
def learn_one(self, x, y):
self.learn_count += 1
print(f"Learning example {self.learn_count}")
return self.model.learn_one(x, y)
def predict_one(self, x):
self.predict_count += 1
return self.model.predict_one(x)
def predict_proba_one(self, x):
return self.model.predict_proba_one(x)
# Use the wrapper
base_model = tree.HoeffdingTreeClassifier()
wrapped = LoggingWrapper(base_model)
print(wrapped) # Output: LoggingWrapper(HoeffdingTreeClassifier)
print(wrapped._supervised) # True (inherited from wrapped model)
for x, y in datasets.Phishing().take(5):
wrapped.learn_one(x, y)
# Example 2: Wrapper that adds caching
from river.base import Wrapper, Regressor
class CachingWrapper(Wrapper, Regressor):
def __init__(self, model, cache_size=100):
self.model = model
self.cache_size = cache_size
self.cache = {}
@property
def _wrapped_model(self):
return self.model
def learn_one(self, x, y):
# Clear cache when learning
self.cache.clear()
return self.model.learn_one(x, y)
def predict_one(self, x):
# Create cache key from features
cache_key = tuple(sorted(x.items()))
if cache_key in self.cache:
return self.cache[cache_key]
# Get prediction from wrapped model
prediction = self.model.predict_one(x)
# Cache the result
if len(self.cache) < self.cache_size:
self.cache[cache_key] = prediction
return prediction
# Use the caching wrapper
from river import linear_model
base_model = linear_model.LinearRegression()
cached_model = CachingWrapper(base_model, cache_size=50)
# Example 3: Wrapper that adds input validation
class ValidatingWrapper(Wrapper, Classifier):
def __init__(self, model, required_features):
self.model = model
self.required_features = required_features
@property
def _wrapped_model(self):
return self.model
def _validate_input(self, x):
missing = self.required_features - set(x.keys())
if missing:
raise ValueError(f"Missing required features: {missing}")
def learn_one(self, x, y):
self._validate_input(x)
return self.model.learn_one(x, y)
def predict_one(self, x):
self._validate_input(x)
return self.model.predict_one(x)
def predict_proba_one(self, x):
self._validate_input(x)
return self.model.predict_proba_one(x)
# Use the validating wrapper
base_model = tree.HoeffdingTreeClassifier()
validated_model = ValidatingWrapper(
base_model,
required_features={'feature1', 'feature2'}
)
# Example 4: Check that wrapper inherits properties
print(f"Is supervised: {wrapped._supervised}")
print(f"Tags: {wrapped._tags}")
print(f"Is multiclass: {wrapped._multiclass}")