`TinyBlobs`#

class capymoa.anomaly.datasets.TinyBlobs[source]#

Bases: NumpyStream

A tiny stream for running unit tests for anomaly detection.

import matplotlib.pyplot as plt
from capymoa.anomaly.datasets import TinyBlobs

stream = TinyBlobs()
x, y = stream._x_data, stream._y_data
plt.scatter(x[y == 0, 0], x[y == 0, 1], marker=".")
plt.scatter(x[y == 1, 0], x[y == 1, 1], marker="x")

../../_images/capymoa-anomaly-datasets-TinyBlobs-1.png

__init__( in_samples: int = 1000, out_samples: int = 100, features: int = 4, clusters: int = 3, seed: int = 0, center_box: tuple[float, float] = (-10.0, 10.0), cluster_std: float | list[float] = 1.0, )[source]#

Construct TinyBlobs.

Parameters:

in_samples – In distribution samples.
out_samples – Out of distribution samples.
features – Number of features.
clusters – Number of clusters.
seed – Random seed for generating data.
center_box – Range features may take.
cluster_std – Variance of each blob center.

__iter__() → Self[source]#

Get an iterator over the stream.

This will NOT restart the stream if it has already been iterated over. Please use the restart() method to restart the stream.

Yield:: An iterator over the stream.

__next__() → _AnyInstance[source]#

Get the next instance in the stream.

Returns:: The next instance in the stream.

cli_help() → str[source]#: Return a help message

get_moa_stream() → InstanceStream | None[source]#: Get the MOA stream object if it exists.

get_schema()[source]#: Return the schema of the stream.

has_more_instances()[source]#: Return True if the stream have more instances to read.

next_instance() → _AnyInstance[source]#

Return the next instance in the stream.

Raises:: ValueError – If the machine learning task is neither a regression nor a classification task.
Returns:: A labeled instances or a regression depending on the schema.

restart()[source]#: Restart the stream to read instances from the beginning.

TinyBlobs#

`TinyBlobs`#