mlm_insights.core.metrics.drift_metrics package

Submodules

mlm_insights.core.metrics.drift_metrics.chi_square module

class mlm_insights.core.metrics.drift_metrics.chi_square.ChiSquare(config: ~typing.Dict[str, ~mlm_insights.constants.definitions.ConfigParameter] = <factory>, epsilon_value: float = 0.0001, _max_size_k: int = 7)

Bases: MetricBase

Data Drift Metric to compute Chi-square goodness of fit test
The chi-square tests the null hypothesis that the categorical data has the given frequencies.
It can process only categorical data types (nominal, ordinal, binary).
It is an approximate metric
This is used for Model Drift computation, taking into consideration reference and current profiles

Configuration

epsilon_value: float, default = 0.0001

  • This function replaces the 0 values in an array with a smaller value. If the array contains any elements <= smaller value, it replaces with epsilon. This is required for certain drift algorithms to ensure the value generated is not an invalid one. For eg: if a denominator is a zero, this leads to division by zero error

_max_size_k: int, default = 7

  • Maximum size, in log2, of k. The value must be between 7 and 21, inclusive

Returns

  • algorithm: string: Drift Algorithm Name

    • “Chi Squared Goodness of Fit Test”

  • test_statistic: float: Test Statistic

    • The chi-squared test statistic

  • p_value: float: p value

    • The P-value is the area under the density curve of this chi-square distribution to the right of the value of the test statistic.

Examples

import pandas as pd

from mlm_insights.builder.builder_component import MetricDetail, EngineDetail
from mlm_insights.builder.insights_builder import InsightsBuilder
from mlm_insights.constants.types import FeatureType, DataType, VariableType, ColumnType
from mlm_insights.core.metrics.drift_metrics.chi_square import ChiSquare
from mlm_insights.core.metrics.metric_metadata import MetricMetadata

input_schema = {
    'mode_of_transport': FeatureType(
        data_type=DataType.TEXT,
        variable_type=VariableType.NOMINAL
        column_type=ColumnType.INPUT)
}


def get_metrics():
    uni_variate_metrics = {
        "mode_of_transport": [MetricMetadata(klass=ChiSquare)]

    }
    metric_details = MetricDetail(univariate_metric=uni_variate_metrics,
                                  dataset_metrics=[])
    return metric_details


def do_run(data_frame):
    runner = InsightsBuilder().                 with_input_schema(input_schema).                 with_data_frame(data_frame=data_frame).                 with_metrics(metrics=get_metrics()).                 with_engine(engine=EngineDetail(engine_name="native")).                 build()
    return runner.run().profile


def main():
    reference_data_frame = pd.DataFrame({'mode_of_transport': ['bus', 'bus', 'train', 'walk', 'bus', 'car']})
    target_data_frame = pd.DataFrame({'mode_of_transport': ['bus', 'bus', 'bus', 'cycle', 'bus', 'car']})

    # do a reference run
    reference_profile = do_run(data_frame=reference_data_frame)
    target_profile = do_run(data_frame=target_data_frame)

    profile_json = target_profile.to_json(reference_profile=reference_profile)
    feature_metrics = profile_json['feature_metrics']
    print(feature_metrics['mode_of_transport']["ChiSquare"])


if __name__ == "__main__":
    main()

Returns the standard metric result as:
{
    'metric_name': 'ChiSquare',
    'metric_description': 'Data Drift Metric to compute Chi-square goodness of fit test',
    'variable_count': 3,
    'variable_names': ['algorithm', 'test_statistic', 'p_value'],
    'variable_types': [TEXT, CONTINUOUS, CONTINUOUS],
    'variable_dtypes': [STRING, FLOAT, FLOAT],
    'variable_dimensions': [0, 0, 0],
    'metric_data': ['ChiSquare', 0.5, 0.5],
    'metadata': {},
    'error': None
}
classmethod create(config: Dict[str, ConfigParameter] | None = None) → ChiSquare

Factory Method to create an object.

Returns

Object: number of items that are duplicate of another item in the data and percentage of duplicate count out of the total count.

epsilon_value: float = 0.0001
get_required_shareable_feature_components() → List[SFCMetaData]

Returns a list of Shareable Feature Components containing 1 SFC that is Frequent Items SFC.

Returns

List of SFCMetadata, containing only 1 SFC i.e. Frequent Items SFC

get_result(**kwargs: Any) → Dict[str, Any]

Returns the computed value of the metric Shareable Feature Component(s) can be accessed using kwargs

Returns

Dict[str, Any]: Dictionary with key as string and value as any metric property.

get_standard_metric_result(**kwargs: Any) → StandardMetricResult

This method returns metric output in standard format.

Returns

StandardMetricResult

merge(other_metric: ChiSquare, **kwargs: Any) → ChiSquare

Merge two ChiSquare into one, without mutating the others.

Parameters

other_metricChiSquare

Other ChiSquare that need be merged.

Returns

TypeMetric

A new instance of ChiSquare

mlm_insights.core.metrics.drift_metrics.drift_metrics_helper module

mlm_insights.core.metrics.drift_metrics.drift_metrics_helper.get_quantiles_sfcs(metric_metadata: MetricMetadata, metric: MetricBase, kwargs: Any) → Tuple[Any, Any, float, float]
mlm_insights.core.metrics.drift_metrics.drift_metrics_helper.validate_metric_can_be_computed(current_profile: Profile, reference_profile: Profile, feature_name: str, metric_metadata: MetricMetadata) → None

mlm_insights.core.metrics.drift_metrics.jensen_shannon module

class mlm_insights.core.metrics.drift_metrics.jensen_shannon.JensenShannon(config: ~typing.Dict[str, ~mlm_insights.constants.definitions.ConfigParameter] = <factory>, bins: str | int | ~typing.List[float] = 'sturges')

Bases: MetricBase

Data Drift Metric to compute Jensen Shannon distance between 2 probability distributions
This is the square root of the Jensen-Shannon divergence.
It can process only numerical data types (int, float).
It is an approximate metric
This is used for Model Drift computation, taking into consideration reference and current profiles

Configuration

bin: Union[str, int, List[float]], default=’sturges’
One of the following values
- Number of bins
- Binning algorithm. Default is Sturges
- Bins: List of floats

Returns

  • algorithm: string: Drift Algorithm Name

    • “Jensen Shannon Distance”

  • drift_score: float: Drift Score

    • The Jensen-Shannon distances between 2 probability distributions

Examples

import pandas as pd
from mlm_insights.builder.builder_component import MetricDetail, EngineDetail
from mlm_insights.builder.insights_builder import InsightsBuilder
from mlm_insights.constants.types import FeatureType, DataType, VariableType, ColumnType
from mlm_insights.core.metrics.drift_metrics.jensen_shannon import JensenShannon
from mlm_insights.core.metrics.metric_metadata import MetricMetadata

input_schema = {
    'square_feet': FeatureType(
        data_type=DataType.FLOAT,
        variable_type=VariableType.CONTINUOUS,
        column_type=ColumnType.INPUT)
}


def get_metrics():
    uni_variate_metrics = {
        "square_feet": [MetricMetadata(klass=JensenShannon)]

    }
    metric_details = MetricDetail(univariate_metric=uni_variate_metrics,
                                  dataset_metrics=[])
    return metric_details


def do_run(data_frame):
    runner = InsightsBuilder().                 with_input_schema(input_schema).                 with_data_frame(data_frame=data_frame).                 with_metrics(metrics=get_metrics()).                 with_engine(engine=EngineDetail(engine_name="native")).                 build()
    return runner.run().profile


def main():
    reference_data_frame = pd.DataFrame({'square_feet': [10, 10, 10, 10]})
    target_data_frame = pd.DataFrame({'square_feet': [20, 21.2, 10, 11.3]})

    # do a reference run
    reference_profile = do_run(data_frame=reference_data_frame)
    target_profile = do_run(data_frame=target_data_frame)

    profile_json = target_profile.to_json(reference_profile=reference_profile)
    feature_metrics = profile_json['feature_metrics']
    print(feature_metrics['square_feet']["JensenShannon"])


if __name__ == "__main__":
    main()

Returns the standard metric result as:
{
    'metric_name': 'JensenShannon',
    'metric_description': 'Data Drift Metric to compute Jensen Shannon distance between 2 probability distributions',
    'variable_count': 2,
    'variable_names': ['algorithm', 'drift_score'],
    'variable_types': [TEXT, CONTINUOUS],
    'variable_dtypes': [STRING, FLOAT],
    'variable_dimensions': [0, 0],
    'metric_data': ['JensenShannon', 0.5],
    'metadata': {},
    'error': None
}
bins: str | int | List[float] = 'sturges'
classmethod create(config: Dict[str, ConfigParameter] | None = None) → JensenShannon

Factory Method to create an object. The configuration will be available in config.

Returns

MetricBase

An Instance of MetricBase.

get_required_shareable_feature_components() → List[SFCMetaData]

Returns list of SFCs required to compute KL metric.

Returns

List: list of SFCs

get_result(**kwargs: Any) → Dict[str, Any]

Returns the computed value of the metric Shareable Feature Component(s) can be accessed using kwargs

Returns

Dict[str, Any]: Dictionary with key as string and value as any metric property.

get_standard_metric_result(**kwargs: Any) → StandardMetricResult

This method returns metric output in standard format.

Returns

StandardMetricResult

merge(other_metric: JensenShannon, **kwargs: Any) → JensenShannon

Merge two JensenShannon into one, without mutating the others.

Parameters

other_metricJensenShannon

Other JensenShannon that need be merged.

Returns

TypeMetric

A new instance of JensenShannon

mlm_insights.core.metrics.drift_metrics.kolmogorov_smirnov module

class mlm_insights.core.metrics.drift_metrics.kolmogorov_smirnov.KolmogorovSmirnov(config: ~typing.Dict[str, ~mlm_insights.constants.definitions.ConfigParameter] = <factory>, bins: str | int | ~typing.List[float] = 100, _kll_k: int = 500)

Bases: MetricBase

Performs the two-sample Kolmogorov-Smirnov test for goodness of fit.
The asymptotic Kolmogorov-Smirnov distribution is used to compute an approximate p-value.
Kolmogorov Smirnov Test is Nonparametric statistical test to identify whether 2 probability distributions differ
or whether the two data samples come from the same distribution
Test Statistic for the 2-sample test is the greatest distance between the CDFs (Cumulative Distribution Function) of each sample
Null Hypothesis: samples are drawn from the same distribution
It can process only numerical data types (int, float).
It is an approximate metric
Internally, it uses a sketch data structure with a default K value of 500.
This is used for Model Drift computation, taking into consideration reference and current profiles

Configuration

bin: Union[str, int, List[float]], default=’sturges’
One of the following values
- Number of bins
- Binning algorithm. Default is Sturges
- Bins: List of floats
_KLL_K: int, default= 500

  • buffer size for kll sketch

Returns

  • algorithm: string: Drift Algorithm Name

    • “Kolmogorov Smirnov”

  • test_statistic: float: Test Statistic

    • The KS test statistic

  • p_value: float: p value

    • show us the chance of getting the two samples, assuming the null hypothesis is true

Examples

import pandas as pd

from mlm_insights.builder.builder_component import MetricDetail, EngineDetail
from mlm_insights.builder.insights_builder import InsightsBuilder
from mlm_insights.constants.types import FeatureType, DataType, VariableType, ColumnType
from mlm_insights.core.metrics.drift_metrics.kolmogorov_smirnov import KolmogorovSmirnov
from mlm_insights.core.metrics.metric_metadata import MetricMetadata

input_schema = {
    'square_feet': FeatureType(
        data_type=DataType.FLOAT,
        variable_type=VariableType.CONTINUOUS,
        column_type=ColumnType.INPUT)
}


def get_metrics():
    uni_variate_metrics = {
        "square_feet": [MetricMetadata(klass=KolmogorovSmirnov)]

    }
    metric_details = MetricDetail(univariate_metric=uni_variate_metrics,
                                  dataset_metrics=[])
    return metric_details


def do_run(data_frame):
    runner = InsightsBuilder().                 with_input_schema(input_schema).                 with_data_frame(data_frame=data_frame).                 with_metrics(metrics=get_metrics()).                 with_engine(engine=EngineDetail(engine_name="native")).                 build()
    return runner.run().profile


def main():
    reference_data_frame = pd.DataFrame({'square_feet': [10, 10, 10, 10]})
    target_data_frame = pd.DataFrame({'square_feet': [20, 21.2, 10, 11.3]})

    # do a reference run
    reference_profile = do_run(data_frame=reference_data_frame)
    target_profile = do_run(data_frame=target_data_frame)

    profile_json = target_profile.to_json(reference_profile=reference_profile)
    feature_metrics = profile_json['feature_metrics']
    print(feature_metrics['square_feet']["KolmogorovSmirnov"])


if __name__ == "__main__":
    main()

Returns the standard metric result as:
{
    'metric_name': 'KolmogorovSmirnov',
    'metric_description': 'Data Drift Metric to compute two-sample Kolmogorov-Smirnov test for goodness of fit',
    'variable_count': 3,
    'variable_names': ['algorithm', 'test_statistic', 'p_value'],
    'variable_types': [TEXT, CONTINUOUS, CONTINUOUS],
    'variable_dtypes': [STRING, FLOAT, FLOAT],
    'variable_dimensions': [0, 0, 0],
    'metric_data': ['KolmogorovSmirnov', 0.5, 0.5],
    'metadata': {},
    'error': None
}
bins: str | int | List[float] = 100
classmethod create(config: Dict[str, ConfigParameter] | None = None) → KolmogorovSmirnov

Factory Method to create an object. The configuration will be available in config.

Returns

MetricBase

An Instance of MetricBase.

get_required_shareable_feature_components() → List[SFCMetaData]

Returns list of SFCs required to compute KS metric.

Returns

List: list of SFCs

get_result(**kwargs: Any) → Dict[str, Any]

Returns the computed value of the metric Shareable Feature Component(s) can be accessed using kwargs

Returns

Dict[str, Any]: Dictionary with key as string and value as any metric property.

get_standard_metric_result(**kwargs: Any) → StandardMetricResult

This method returns metric output in standard format.

Returns

StandardMetricResult

merge(other_metric: KolmogorovSmirnov, **kwargs: Any) → KolmogorovSmirnov

Merge two KolmogorovSmirnov metric into one, without mutating the others.

Parameters

other_metricKolmogorovSmirnov

Other KolmogorovSmirnov metric that need be merged.

Returns

TypeMetric

A new instance of KolmogorovSmirnov

mlm_insights.core.metrics.drift_metrics.kullback_leibler module

class mlm_insights.core.metrics.drift_metrics.kullback_leibler.KullbackLeibler(config: ~typing.Dict[str, ~mlm_insights.constants.definitions.ConfigParameter] = <factory>, bins: str | int | ~typing.List[float] = 'sturges')

Bases: MetricBase

Metric to compute Kullback-Leibler divergence between 2 probability distributions
It is an approximate metric. It can process only numerical data types (int, float).
This is used for Model Drift computation, taking into consideration reference and current profiles

Configuration

bin: Union[str, int, List[float]], default=’sturges’
One of the following values
- Number of bins
- Binning algorithm. Default is Sturges
- Bins: List of floats

Returns

  • algorithm: string: Drift Algorithm Name

    • “Kullback Leibler Divergence”

  • drift_score: float: Drift Score

    • The KL distances between 2 probability distributions

Examples

import pandas as pd

from mlm_insights.builder.builder_component import MetricDetail, EngineDetail
from mlm_insights.builder.insights_builder import InsightsBuilder
from mlm_insights.constants.types import FeatureType, DataType, VariableType, ColumnType
from mlm_insights.core.metrics.drift_metrics.kullback_leibler import KullbackLeibler
from mlm_insights.core.metrics.metric_metadata import MetricMetadata

input_schema = {
    'square_feet': FeatureType(
        data_type=DataType.FLOAT,
        variable_type=VariableType.CONTINUOUS,
        column_type=ColumnType.INPUT)
}


def get_metrics():
    uni_variate_metrics = {
        "square_feet": [MetricMetadata(klass=KullbackLeibler)]

    }
    metric_details = MetricDetail(univariate_metric=uni_variate_metrics,
                                  dataset_metrics=[])
    return metric_details


def do_run(data_frame):
    runner = InsightsBuilder().                 with_input_schema(input_schema).                 with_data_frame(data_frame=data_frame).                 with_metrics(metrics=get_metrics()).                 with_engine(engine=EngineDetail(engine_name="native")).                 build()
    return runner.run().profile


def main():
    reference_data_frame = pd.DataFrame({'square_feet': [10, 10, 10, 10]})
    target_data_frame = pd.DataFrame({'square_feet': [20, 21.2, 10, 11.3]})

    # do a reference run
    reference_profile = do_run(data_frame=reference_data_frame)
    target_profile = do_run(data_frame=target_data_frame)

    profile_json = target_profile.to_json(reference_profile=reference_profile)
    feature_metrics = profile_json['feature_metrics']
    print(feature_metrics['square_feet']["KullbackLeibler"])


if __name__ == "__main__":
    main()

Returns the standard metric result as:
{
    'metric_name': 'KullbackLeibler',
    'metric_description': 'Data Drift Metric to compute Kullback-Leibler divergence between 2 probability distributions',
    'variable_count': 2,
    'variable_names': ['algorithm', 'drift_score'],
    'variable_types': [TEXT, CONTINUOUS],
    'variable_dtypes': [STRING, FLOAT],
    'variable_dimensions': [0, 0],
    'metric_data': ['KullbackLeibler', 0.5],
    'metadata': {},
    'error': None
}
bins: str | int | List[float] = 'sturges'
classmethod create(config: Dict[str, ConfigParameter] | None = None) → KullbackLeibler

Factory Method to create an object. The configuration will be available in config.

Returns

MetricBase

An Instance of MetricBase.

get_required_shareable_feature_components() → List[SFCMetaData]

Returns list of SFCs required to compute KL metric.

Returns

List: list of SFCs

get_result(**kwargs: Any) → Dict[str, Any]

Returns the computed value of the metric Shareable Feature Component(s) can be accessed using kwargs

Returns

Dict[str, Any]: Dictionary with key as string and value as any metric property.

get_standard_metric_result(**kwargs: Any) → StandardMetricResult

This method returns metric output in standard format.

Returns

StandardMetricResult

merge(other_metric: KullbackLeibler, **kwargs: Any) → KullbackLeibler

Merge two KullbackLeibler into one, without mutating the others.

Parameters

other_metricKullbackLeibler

Other KullbackLeibler that need be merged.

Returns

TypeMetric

A new instance of KullbackLeibler

mlm_insights.core.metrics.drift_metrics.population_stability_index module

class mlm_insights.core.metrics.drift_metrics.population_stability_index.PopulationStabilityIndex(config: ~typing.Dict[str, ~mlm_insights.constants.definitions.ConfigParameter] = <factory>, bins: str | int | ~typing.List[float] = 'sturges', _kll_k: int = 500)

Bases: MetricBase

Data Drift Metric to compute Population Stability Index (PSI) distance between 2 probability distributions
It can process only numerical data types (int, float).
It is an approximate metric
This is used for Model Drift computation, taking into consideration reference and current profiles

Configuration

bin: Union[str, int, List[float]], default=’sturges’
One of the following values
- Number of bins
- Binning algorithm. Default is Sturges
- Bins: List of floats

Returns

  • algorithm: string: Drift Algorithm Name

    • “Population Stability Index”

  • drift_score: float: Drift Score

    • The PSI distances between one probability distribution from a reference probability distribution

Examples

import pandas as pd

from mlm_insights.builder.builder_component import MetricDetail, EngineDetail
from mlm_insights.builder.insights_builder import InsightsBuilder
from mlm_insights.constants.types import FeatureType, DataType, VariableType, ColumnType
from mlm_insights.core.metrics.drift_metrics.population_stability_index import PopulationStabilityIndex
from mlm_insights.core.metrics.metric_metadata import MetricMetadata

input_schema = {
    'square_feet': FeatureType(
        data_type=DataType.FLOAT,
        variable_type=VariableType.CONTINUOUS,
        column_type=ColumnType.INPUT)
}


def get_metrics():
    uni_variate_metrics = {
        "square_feet": [MetricMetadata(klass=PopulationStabilityIndex)]

    }
    metric_details = MetricDetail(univariate_metric=uni_variate_metrics,
                                  dataset_metrics=[])
    return metric_details


def do_run(data_frame):
    runner = InsightsBuilder().                 with_input_schema(input_schema).                 with_data_frame(data_frame=data_frame).                 with_metrics(metrics=get_metrics()).                 with_engine(engine=EngineDetail(engine_name="native")).                 build()
    return runner.run().profile


def main():
    reference_data_frame = pd.DataFrame({'square_feet': [10, 10, 10, 10]})
    target_data_frame = pd.DataFrame({'square_feet': [20, 21.2, 10, 11.3]})

    # do a reference run
    reference_profile = do_run(data_frame=reference_data_frame)
    target_profile = do_run(data_frame=target_data_frame)

    profile_json = target_profile.to_json(reference_profile=reference_profile)
    feature_metrics = profile_json['feature_metrics']
    print(feature_metrics['square_feet']["PopulationStabilityIndex"])


if __name__ == "__main__":
    main()

Returns the standard metric result as:
{
    'metric_name': 'PopulationStabilityIndex',
    'metric_description': 'Data Drift Metric to compute Population Stability Index(PSI) distance between 2 probability distributions',
    'variable_count': 2,
    'variable_names': ['algorithm', 'drift_score'],
    'variable_types': [TEXT, CONTINUOUS],
    'variable_dtypes': [STRING, FLOAT],
    'variable_dimensions': [0, 0],
    'metric_data': ['PopulationStabilityIndex', 0.5],
    'metadata': {},
    'error': None
}
bins: str | int | List[float] = 'sturges'
classmethod create(config: Dict[str, ConfigParameter] | None = None) → PopulationStabilityIndex

Factory Method to create an object. The configuration will be available in config.

Returns

MetricBase

An Instance of MetricBase.

get_required_shareable_feature_components() → List[SFCMetaData]

Returns list of SFCs required to compute KL metric.

Returns

List: list of SFCs

get_result(**kwargs: Any) → Dict[str, Any]

Returns the computed value of the metric Shareable Feature Component(s) can be accessed using kwargs

Returns

Dict[str, Any]: Dictionary with key as string and value as any metric property.

get_standard_metric_result(**kwargs: Any) → StandardMetricResult

This method returns metric output in standard format.

Returns

StandardMetricResult

merge(other_metric: PopulationStabilityIndex, **kwargs: Any) → PopulationStabilityIndex

Merge two PopulationStabilityIndex into one, without mutating the others.

Parameters

other_metricPopulationStabilityIndex

Other JensenShannon that need be merged.

Returns

TypeMetric

A new instance of JensenShannon