Description:

datarobotmodelmetrics 0.6.4

DataRobot Model Metrics Overview
The datarobot-model-metrics (dmm) library provides the tools necessary to create custom metrics,
including the following operations:

Set up the environment
Create a data source
Create a custom metric
Evaluate a custom metric
Submit results to DataRobot

Review the following documentation to learn more about the datarobot-model-metrics interfaces used to
perform custom metric operations:

Environment Setup
Data Sources
Custom Metrics
Metric Evaluator
DR Custom Metrics

For end-to-end examples, you can review the ./examples directory.
Environment setup
There are two primary ecosystems where you can develop a custom metric:

Within the DataRobot application (via notebook or scheduled job).
In a local development environment.

Environmental considerations
Python modules:

When running from within the DataRobot application, the ecosystem has all the required Python modules.
When running locally, you need to install the dmm module (built in this repository—it will update the Python environment with all the required modules.

Setting parameters:

When running from within the DataRobot application, parameters get set through environment variables.
When running locally, it is easier to pass values using arguments to set parameters than to set environment variables.

Initializing your application, as show below, allows setting parameters using either method without significant changes.
Initializing your application
The CustomMetricArgumentParser is a class that wraps the standard argparser.ArgumentParser.
It provides some convenience functions that allow reading values from the environment or normal argument parsing.
When the CustomMetricArgumentParser.parse_args() gets called, it checks for missing values.
The log_manager provides a set of functions to help with logging.
The DMM and the DataRobot public API client use standard Python logging primitives.
A complete list of log classes with their current levels is available using get_log_levels().
The initialize_loggers() function takes care of initializing all the loggers with a format like shown below:
2024-08-09 02:19:50 PM - dmm.data_source.datarobot_source - INFO - fetching the next predictions dataframe... 2024-07-15 00:00:00 - 2024-08-09 14:19:46.643722
2024-08-09 02:19:56 PM - urllib3.connectionpool - DEBUG - https://staging.datarobot.com:443 "POST /api/v2/deployments/66a90a712ad81645df8c469c/predictionDataExports/ HTTP/1.1" 202 368

The following snippet shows how to set up your runtime environment using the previously mentioned classes:
import sys
from dmm import CustomMetricArgumentParser
from dmm.log_manager import initialize_loggers

parser = CustomMetricArgumentParser(description="My new custom metric")
parser.add_base_args() # adds standard arguments
# Add more with standard ArgumentParser primitives, or some convenience functions such as add_environment_arg()

# Parse the program arguments (if any) to an argparse.Namespace.
args = parser.parse_args(sys.argv[1:])

# Initialize the logging based on the 'LOG' environment variable, or the --log option
initialize_loggers(args.log)

The standard/base arguments include:

BASE_URL - URL to the public API
API_KEY - token used for authentication to server at BASE_URL
DEPLOYMENT_ID - deployment identifier from application
CUSTOM_METRIC_ID - custom-metric identifier from application
DRY_RUN - flag to indicate whether to really report the custom-metric result
START_TS - start of time for calculations
END_TS - end of time for calculations
MAX_ROWS - maximum number of rows to process
LOG - initialization of logging, defaults to setting all dmm and datarobot modules to WARNING.

Here's an example of the help when using the CustomMetricArgumentParser:
(model-runner) $ python3 custom.py --help
usage: custom.py [-h] [--api-key KEY] [--base-url URL] [--deployment-id ID] [--custom-metric-id ID] [--dry-run] [--start-ts TIMESTAMP] [--end-ts TIMESTAMP] [--max-rows ROWS] [--required] [--log [[NAME:]LEVEL ...]]

My new custom metric

optional arguments:
-h, --help show this help message and exit
--api-key KEY API key used to authenticate to server. Settable via 'API_KEY', required.
--base-url URL URL for server. Settable via 'BASE_URL' (default: https://staging.datarobot.com/api/v2), required.
--deployment-id ID Deployment ID. Settable via 'DEPLOYMENT_ID' (default: None), required.
--custom-metric-id ID
Custom metric ID. Settable via 'CUSTOM_METRIC_ID' (default: None), required.
--dry-run Dry run. Settable via 'DRY_RUN' (default: False).
--start-ts TIMESTAMP Start timestamp. Settable with 'START_TS', or 'LAST_SUCCESSFUL_RUN_TS' (when not dry run). Default is 2024-08-08 14:27:55.493027
--end-ts TIMESTAMP End timestamp. Settable with 'END_TS' or 'CURRENT_RUN_TS'. Default is 2024-08-09 14:27:55.493044.
--max-rows ROWS Maximum number of rows. Settable via 'MAX_ROWS' (default: 100000).
--required List the required properties and exit.
--log [[NAME:]LEVEL ...]
Logging level list. Settable via 'LOG' (default: WARNING).
(model-runner) $

Other utilities
This section mentions some utilities that may help you develop.
save_to_csv()
During development, it is common to run your code over the same data multiple times to see how your changes impact the results.
The save_to_csv() utility allows you to save your results to a CSV file, so you can compare the results between successive runs on the same data.
DataRobotSource
The most commonly used data source is DataRobotSource. This data source
connects to DataRobot to fetch selected data from the DataRobot platform.
DataRobotSource Initialization
Initialize DataRobotSource with the following mandatory parameters:
from dmm.data_source import DataRobotSource

source = DataRobotSource(
base_url=DATAROBOT_ENDPOINT,
token=DATAROBOT_API_TOKEN,
deployment_id=deployment_id,
start=start_of_export_window,
end=end_of_export_window,
)

You can also provide the base_url and token parameters as environment variables:
os.environ['DATAROBOT_ENDPOINT'] and os.environ['BASE_URL']
from dmm.data_source import DataRobotSource

source = DataRobotSource(
deployment_id=deployment_id,
start=start_of_export_window,
end=end_of_export_window,
)

The following example initializes DataRobotSource with all parameters:
from dmm.data_source import DataRobotSource

source = DataRobotSource(
base_url=DATAROBOT_ENDPOINT,
token=DATAROBOT_API_TOKEN,
client=None,
deployment_id=deployment_id,
model_id=model_id,
start=start_of_export_window,
end=end_of_export_window,
max_rows=10000,
delete_exports=False,
use_cache=False,
actuals_with_matched_predictions=True,
)




Parameter
Description




base_url: str
The DataRobot API URL; for example, https://app.datarobot.com/api/v2.


token: str
A DataRobot API token from Developer Tools.


client: Optional[dr.Client]
Use the dr.Client object instead of base_url and token.


deployment_id: str
The ID of the deployment evaluated by the custom metric.


model_id: Optional[str]
The ID of the model evaluated by the custom metric. If you don't specify a model ID, the champion model ID is used.


start: datetime
The start of the export window. Define the date you want to start to retrieving data from.


end: datetime
The end of the export window. Define the date you want to retrieve data until.


max_rows: Optional[int]
The maximum number of rows to fetch at once when the requested data doesn't fit into memory.


delete_exports: Optional[bool]
If True, datasets with exported data created in the AI Catalog are automatically deleted. The default value is False.


use_cache: Optional[bool]
If True, use existing datasets stored in the AI Catalog for time ranges included in previous exports. The default value is False.


actuals_with_matched_predictions: Optional[bool]
If False, allow actuals export without matched predictions. The default value is True.



DataRobotSource Usage
Prediction Data Export
The get_prediction_data method returns a chunk of prediction data with the appropriate chunk ID;
the returned data chunk is a pandas DataFrame with the number of rows respecting the max_rows parameter.
This method returns data until the data source is exhausted.
prediction_df_1, prediction_chunk_id_1 = source.get_prediction_data()

print(prediction_df_1.head(5).to_string())
print(f"chunk id: {prediction_chunk_id_1}")

DR_RESERVED_PREDICTION_TIMESTAMP DR_RESERVED_PREDICTION_VALUE_high DR_RESERVED_PREDICTION_VALUE_low date_non_unique date_random id 年月日
0 2023-09-13 11:02:51.248000+00:00 0.697782 0.302218 1950-10-01 1949-01-27 1 1949-01-01
1 2023-09-13 11:02:51.252000+00:00 0.581351 0.418649 1959-04-01 1949-02-03 2 1949-02-01
2 2023-09-13 11:02:51.459000+00:00 0.639347 0.360653 1954-05-01 1949-03-28 3 1949-03-01
3 2023-09-13 11:02:51.459000+00:00 0.627727 0.372273 1951-09-01 1949-04-07 4 1949-04-01
4 2023-09-13 11:02:51.664000+00:00 0.591612 0.408388 1951-03-01 1949-05-16 5 1949-05-01
chunk id: 0

When the data source is exhausted, None and -1 are returned:
prediction_df_2, prediction_chunk_id_2 = source.get_prediction_data()

print(prediction_df_2)
print(prediction_chunk_id_2)

None
chunk id: -1

The reset method resets the exhausted data source, allowing it to iterate from the beginning:
source.reset()

The get_all_prediction_data method returns all prediction data available for a data source object in a single DataFrame:
prediction_df = source.get_all_prediction_data()

Actuals Data Export
The get_actuals_data method returns a chunk of actuals data with the appropriate chunk ID
the returned data chunk is a pandas DataFrame with the number of rows respecting the max_rows parameter.
This method returns data until the data source is exhausted.
actuals_df_1, actuals_chunk_id_1 = source.get_actuals_data()

print(actuals_df_1.head(5).to_string())
print(f"chunk id: {actuals_chunk_id_1}")

association_id timestamp label actuals predictions predicted_class
0 1 2023-09-13 11:00:00+00:00 low 0 0.302218 high
194 57 2023-09-13 11:00:00+00:00 low 1 0.568564 low
192 56 2023-09-13 11:00:00+00:00 low 1 0.569865 low
190 55 2023-09-13 11:00:00+00:00 low 0 0.473282 high
196 58 2023-09-13 11:00:00+00:00 low 1 0.573861 low
chunk id: 0

To return raw data in the format of data from postgresql, set the return_original_column_names parameter to True:
actuals_df_1, actuals_chunk_id_1 = source.get_actuals_data()

print(actuals_df_1.head(5).to_string())
print(f"chunk id: {actuals_chunk_id_1}")

id timestamp label actuals y predicted_class
0 1 2023-09-13 11:00:00+00:00 low 0 0.302218 high
194 57 2023-09-13 11:00:00+00:00 low 1 0.568564 low
192 56 2023-09-13 11:00:00+00:00 low 1 0.569865 low
190 55 2023-09-13 11:00:00+00:00 low 0 0.473282 high
196 58 2023-09-13 11:00:00+00:00 low 1 0.573861 low
chunk id: 0

To return all actuals data available for a source object in a single DataFrame, use the get_all_actuals_data method:
actuals_df = source.get_all_actuals_data()

When the data source is exhausted, None and -1 are returned:
actuals_df_2, actuals_chunk_id_2 = source.get_actuals_data()

print(actuals_df_2)
print(actuals_chunk_id_2)

None
chunk id: -1

The reset method resets the exhausted data source, allowing it to iterate from the beginning:
source.reset()

Training Data Export
The get_training_data method returns all data used for training in one call. The returned data is a pandas DataFrame:
train_df = source.get_training_data()
print(train_df.head(5).to_string())

y date_random date_non_unique 年月日
0 high 1949-01-27 1950-10-01 1949-01-01
1 high 1949-02-03 1959-04-01 1949-02-01
2 low 1949-03-28 1954-05-01 1949-03-01
3 high 1949-04-07 1951-09-01 1949-04-01
4 high 1949-05-16 1951-03-01 1949-05-01

Combined data export
The get_data method returns combined_data, which includes merged scoring data, predictions, and matched actuals:
This Metric Evaluator uses this method as the main data export method.
df, chunk_id_1 = source.get_data()
print(df.head(5).to_string())
print(f"chunk id: {chunk_id_1}")

timestamp predictions date_non_unique date_random association_id 年月日 predicted_class label actuals
0 2023-09-13 11:02:51.248000+00:00 0.302218 1950-10-01 1949-01-27 1 1949-01-01 high low 0
1 2023-09-13 11:02:51.252000+00:00 0.418649 1959-04-01 1949-02-03 2 1949-02-01 high low 0
2 2023-09-13 11:02:51.459000+00:00 0.360653 1954-05-01 1949-03-28 3 1949-03-01 high low 1
3 2023-09-13 11:02:51.459000+00:00 0.372273 1951-09-01 1949-04-07 4 1949-04-01 high low 0
4 2023-09-13 11:02:51.664000+00:00 0.408388 1951-03-01 1949-05-16 5 1949-05-01 high low 0
chunk id: 0

The get_all_data returns all combined data available for that source object in a single DataFrame:
df = source.get_all_data()

BatchDataRobotSource
The BatchDataRobotSource interface is for batch deployments.
The following example initializes BatchDataRobotSource with all parameters:
from dmm.data_source import BatchDataRobotSource

source = BatchDataRobotSource(
base_url=DATAROBOT_ENDPOINT,
token=DATAROBOT_API_TOKEN,
client=None,
deployment_id=deployment_id,
model_id=model_id,
batch_ids=batch_ids,
max_rows=10000,
delete_exports=False,
use_cache=False,
)

The parameters for this method are analogous to those for DataRobotSource.
The most important difference is that instead of the time range (start and end), you must provide batch IDs.
In addition, a batch source doesn't support actuals export.
The get_prediction_data method returns a chunk of prediction data with the appropriate chunk ID;
the returned data chunk is a pandas DataFrame with the number of rows respecting the max_rows parameter.
This method returns data until the data source is exhausted.
prediction_df_1, prediction_chunk_id_1 = source.get_prediction_data()
print(prediction_df_1.head(5).to_string())
print(f"chunk id: {prediction_chunk_id_1}")

AGE B CHAS CRIM DIS batch_id DR_RESERVED_BATCH_NAME timestamp INDUS LSTAT MEDV NOX PTRATIO RAD RM TAX ZN id
0 65.2 396.90 0 0.00632 4.0900 <batch_id> batch1 2023-06-23 09:47:47.060000+00:00 2.31 4.98 24.0 0.538 15.3 1 6.575 296 18.0 1
1 78.9 396.90 0 0.02731 4.9671 <batch_id> batch1 2023-06-23 09:47:47.060000+00:00 7.07 9.14 21.6 0.469 17.8 2 6.421 242 0.0 2
2 61.1 392.83 0 0.02729 4.9671 <batch_id> batch1 2023-06-23 09:47:47.060000+00:00 7.07 4.03 34.7 0.469 17.8 2 7.185 242 0.0 3
3 45.8 394.63 0 0.03237 6.0622 <batch_id> batch1 2023-06-23 09:47:47.060000+00:00 2.18 2.94 33.4 0.458 18.7 3 6.998 222 0.0 4
4 54.2 396.90 0 0.06905 6.0622 <batch_id> batch1 2023-06-23 09:47:47.060000+00:00 2.18 5.33 36.2 0.458 18.7 3 7.147 222 0.0 5
chunk id: 0

prediction_df = source.get_all_prediction_data()

source.reset()

df, chunk_id_1 = source.get_data()

The get_training_data method returns all data used for training in one call. The returned data is a pandas DataFrame:
train_df = source.get_training_data()


Note:: actuals export for batches is not implemented yet.

DataFrameSource
If you aren't exporting data directly from DataRobot, and instead have it downloaded locally (for example),
you can load the dataset into DataFrameSource.
The DataFrameSource method wraps any pd.DataFrame to create a library-compatible source.
This is the easiest way to interact with the library when bringing your own data:
source = DataFrameSource(
df=pd.read_csv("./data_hour_of_week.csv"),
max_rows=10000,
timestamp_col="date"
)

df, chunk_id_1 = source.get_data()
print(df.head(5).to_string())
print(f"chunk id: {chunk_id_1}")

date y
0 1959-12-31 23:59:57 -0.183669
1 1960-01-01 01:00:02 0.283993
2 1960-01-01 01:59:52 0.020663
3 1960-01-01 03:00:14 0.404304
4 1960-01-01 03:59:58 1.005252
chunk id: 0

In addition, it is possible to create new data source definitions. To define a new data source, you
can customize and implement the DataSourceBase interface.
TimeBucket
The TimeBucket enum defines the required data aggregation granularity over time . By default, TimeBucket is set to TimeBucket.ALL.
You can specify any of the following values: SECOND, MINUTE, HOUR, DAY, WEEK, MONTH, QUARTER, or ALL.
To change the TimeBucket value, use the init method: source.init(time_bucket):
# let's generate a dummy DataFrame with 2 rows per time bucket (Hour in this scenario)
test_df = gen_dataframe_for_accuracy_metric(
nr_rows=10,
rows_per_time_bucket=2,
prediction_value=1,
with_actuals=True,
with_predictions=True,
time_bucket=TimeBucket.HOUR,
)
print(test_df)
timestamp predictions actuals
0 01/06/2005 13:00:00.000000 1 0.999
1 01/06/2005 13:00:00.000000 1 0.999
2 01/06/2005 14:00:00.000000 1 0.999
3 01/06/2005 14:00:00.000000 1 0.999
4 01/06/2005 15:00:00.000000 1 0.999
5 01/06/2005 15:00:00.000000 1 0.999
6 01/06/2005 16:00:00.000000 1 0.999
7 01/06/2005 16:00:00.000000 1 0.999
8 01/06/2005 17:00:00.000000 1 0.999
9 01/06/2005 17:00:00.000000 1 0.999

# let's use DataFrameSource and load created DataFrame
source = DataFrameSource(
df=test_df,
max_rows=10000,
timestamp_col="timestamp",
)
# init source with the selected TimeBucket
source.init(TimeBucket.HOUR)
df, _ = source.get_data()
print(df)
timestamp predictions actuals
0 01/06/2005 13:00:00.000000 1 0.999
1 01/06/2005 13:00:00.000000 1 0.999
df, _ = source.get_data()
print(df)
timestamp predictions actuals
2 01/06/2005 14:00:00.000000 1 0.999
3 01/06/2005 14:00:00.000000 1 0.999

source.init(TimeBucket.DAY)
df, _ = source.get_data()
print(df)
timestamp predictions actuals
0 01/06/2005 13:00:00.000000 1 0.999
1 01/06/2005 13:00:00.000000 1 0.999
2 01/06/2005 14:00:00.000000 1 0.999
3 01/06/2005 14:00:00.000000 1 0.999
4 01/06/2005 15:00:00.000000 1 0.999
5 01/06/2005 15:00:00.000000 1 0.999
6 01/06/2005 16:00:00.000000 1 0.999
7 01/06/2005 16:00:00.000000 1 0.999
8 01/06/2005 17:00:00.000000 1 0.999
9 01/06/2005 17:00:00.000000 1 0.999

The returned data chunks follow the selected TimeBucket. This is helpful in the
MetricEvaluator. In addition to TimeBucket, the source respects the max_rows
parameter when generating data chunks; for example, using the same dataset as in the example above (but with max_rows set to 3):
source = DataFrameSource(
df=test_df,
max_rows=3,
timestamp_col="timestamp",
)
source.init(TimeBucket.DAY)
df, chunk_id = source.get_data()
print(df)
timestamp predictions actuals
0 01/06/2005 13:00:00.000000 1 0.999
1 01/06/2005 13:00:00.000000 1 0.999
2 01/06/2005 14:00:00.000000 1 0.999

In DataRobotSource, you can specify the TimeBucket and max_rows parameters for all export types except
training data export, which is returned in one piece.
Deployment
The Deployment class is a helper class which provides access to relevant deployment properties. This class is used
inside the DataRobotSource to select the appropriate workflow to work with data.
import datarobot as dr
from dmm.data_source.datarobot.deployment import Deployment
dr.Client()
deployment = Deployment(deployment_id=deployment_id)

deployment_type = deployment.type()
target_column = deployment.target_column()
positive_class_label = deployment.positive_class_label()
negative_class_label = deployment.negative_class_label()
prediction_threshold = deployment.prediction_threshold()
.
.
.

Custom Metrics
The MetricBase provides an interface to define your own custom metrics.
See the examples of custom metrics located in the metric directory.
MetricBase
In MetricBase, you must define the type of data a metric requires, and the custom metric must inherit that definition:
class MetricBase(object):
def __init__(
self,
name: str,
description: str = None,
need_predictions: bool = False,
need_actuals: bool = False,
need_scoring_data: bool = False,
need_training_data: bool = False,
):
self.name = name
self.description = description
self._need_predictions = need_predictions
self._need_actuals = need_actuals
self._need_scoring_data = need_scoring_data
self._need_training_data = need_training_data

In addition, you must implement the scoring and reduction methods in MetricBase:

Scoring (score): Uses initialized data types to calculate a metric.
Reduction (reduce_func): Reduces multiple values in the same TimeBucket to one value.

def score(
self,
scoring_data: pd.DataFrame,
predictions: np.array,
actuals: np.array,
fit_ctx=None,
metadata=None,
) -> float:
raise NotImplemented

def reduce_func(self) -> callable:
return np.mean

ModelMetricBase and DataMetricBase
Two default classes can help you create your own custom metrics: ModelMetricBase and DataMetricBase.
ModelMetricBase is the base class for metrics that require actuals and predictions for metric calculation.
class ModelMetricBase(MetricBase):
def __init__(
self, name: str, description: str = None, need_training_data: bool = False
):
super().__init__(
name=name,
description=description,
need_scoring_data=False,
need_predictions=True,
need_actuals=True,
need_training_data=need_training_data,
)

def score(
self,
prediction: np.array,
actuals: np.array,
fit_context=None,
metadata=None,
scoring_data=None,
) -> float:
raise NotImplemented

DataMetricBase is the base class for metrics that require scoring data for metric calculation.
class DataMetricBase(MetricBase):
def __init__(
self, name: str, description: str = None, need_training_data: bool = False
):
super().__init__(
name=name,
description=description,
need_scoring_data=True,
need_predictions=False,
need_actuals=False,
need_training_data=need_training_data,
)

def score(
self,
scoring_data: pd.DataFrame,
fit_ctx=None,
metadata=None,
predictions=None,
actuals=None,
) -> float:
raise NotImplemented

LLM metrics
In the case of LLM support, a new type of metric was introduced
LLMMetricBase is the base class for metrics that require scoring data and predictions for metric calculation.
Which in the LLM world can be translated into prompts (user input) and completions (LLM response).
class LLMMetricBase(MetricBase):
def __init__(
self, name: str, description: str = None, need_training_data: bool = False
):
super().__init__(
name=name,
description=description,
need_scoring_data=True,
need_predictions=True,
need_actuals=False,
need_training_data=need_training_data,
)

def score(
self,
scoring_data: pd.DataFrame,
predictions: np.array,
fit_ctx=None,
metadata=None,
actuals=None,
) -> float:
raise NotImplemented

SklearnMetric
To accelerate the implementation of custom metrics, you can use ready-made, proven metrics from
Sklearn.
Provide the name of a metric, using the SklearnMetric class as the base class, to create a custom metric this way.
See the example below:
from dmm.metric.sklearn_metric import SklearnMetric


class MedianAbsoluteError(SklearnMetric):
"""
Metric that calculates the median absolute error of the difference between predictions and actuals
"""

def __init__(self):
super().__init__(
metric="median_absolute_error",
)

PromptSimilarityMetricBase
The PromptSimilarityMetricBase is designed to make for easy comparison of LLM prompt and context vectors.
This class is generally used with TextGen models where the prompt and context vectors are populated as described below.
The base class takes care of pulling the vectors from the scoring_data, and iterating over each entry.
The prompt vector is pulled from the prompt_column (which defaults to _LLM_PROMPT_VECTOR) of the scoring_data.
The context vectors are pulled from the context_column (which defaults to _LLM_CONTEXT) of the scoring_data. The context column contains a list of context dictionaries, and each context needs to have a vector element.
Both the prompt_column and context_column are expected to be JSON encoded data.
A derived class must implement calculate_distance() -- for this class, the score() is already implemented.
The calculate_distance function returns a single floating point value based on a single prompt_vector, and a list of context_vectors.
Using the PromptSimilarityMetricBase to calculate the minimum Euclidean distance is shown below:
from dmm.metric import PromptSimilarityMetricBase

class EuclideanMinMetric(PromptSimilarityMetricBase):
"""Calculate the minimum Euclidean distance between a prompt vector, and a list of context vectors"""
def calculate_distance(self, prompt_vector: np.array, context_vectors: List[np.array]) -> float:
distances = [
np.linalg.norm(prompt_vector - context_vector)
for context_vector in context_vectors
]
return min(distances)

# an instantiation would potentially look like this
scorer = EuclideanMinMetric(name=custom_metric.name, description="Euclidean minimum distance between prompt and context vectors")

Submit custom metric values with datarobot-model-metrics
The metrics mentioned above can provide the source of the custom metric definitions in the DataRobot platform.
The CustomMetric interface retrieves the metadata of an existing custom metric
in DataRobot to report data to that custom metric.
We can initialize the metric by providing the parameters explicitly (metric_id, deployment_id, model_id, dr.Client()):
from dmm.custom_metric import CustomMetric


cm = CustomMetric.from_id(metric_id=METRIC_ID, deployment_id=DEPLOYMENT_ID, model_id=MODEL_ID, client=CLIENT)

You can also define these parameters as environment variables: os.environ["DEPLOYMENT_ID"], os.environ["CUSTOM_METRIC_ID"] os.environ['BASE_URL'], and os.environ['DATAROBOT_ENDPOINT']:
from dmm.custom_metric import CustomMetric


cm = CustomMetric.from_id()

In the case of batch mode, it is required to specify it:
from dmm.custom_metric import CustomMetric


cm = CustomMetric.from_id(is_batch=True)

The report method submits custom metric values to a custom metric defined in DataRobot.
To use this method, report a dataframe in the shape of the output from the metric evaluator.
For more information, see MetricEvaluator.
print(aggregated_metric_per_time_bucket.to_string())

timestamp samples median_absolute_error
1 01/06/2005 14:00:00.000000 2 0.001

response = cm.report(df=aggregated_metric_per_time_bucket)
print(response.status_code)
202

The dry_run parameter determines if the custom metric values transfer is a dry run
(where the values aren't saved in the database) or if it is a production data transfer.
This parameter is set to false by default.
response = cm.report(df=aggregated_metric_per_time_bucket, dry_run=True)
print(response.status_code)
202

Metric Evaluator
The MetricEvaluator class calculates metric values over time using the selected source.
This class is used to "stream" data through the metric object, generating metric values.
Metric Evaluator Initialization
Initialize the MetricEvaluator with the following mandatory parameters:
from dmm import MetricEvaluator, TimeBucket
from dmm.data_source.datarobot_source import DataRobotSource
from dmm.metric import MedianAbsoluteError

source = DataRobotSource(
deployment_id=DEPLOYMENT_ID,
start=datetime.utcnow() - timedelta(weeks=1),
end=datetime.utcnow(),
)

metric = MedianAbsoluteError()

metric_evaluator = MetricEvaluator(metric=metric, source=source, time_bucket=TimeBucket.MINUTE)

To use MetricEvaluator, create a metric class implementing the MetricBase interface, a source implementing
DataSourceBase, and then specify the level of aggregation granularity.
Initialize MetricEvaluator with all parameters:
from dmm import ColumnName, MetricEvaluator, TimeBucket

metric_evaluator = MetricEvaluator(
metric=metric,
source=source,
time_bucket=TimeBucket.HOUR,
prediction_col=ColumnName.PREDICTIONS,
actuals_col=ColumnName.ACTUALS,
timestamp_col=ColumnName.TIMESTAMP,
filter_actuals=False,
filter_predictions=False,
filter_scoring_data=False,
segment_attribute=None,
segment_value=None,
)




Parameter
Description




metric: Union[str, MetricBase, List[str], List[MetricBase]]
If a string or list of strings is passed,then MetricEvaluator will look for matched sklearn metrics, in case a metrics or list of objects is passed they must implement MetricBase interface.


source: DataSourceBase
Source to pull the data from, DataRobotSource or DataFrameSource or other sources that implement DataSourceBase interface.


time_bucket: TimeBucket
Time bucket size to use for evaluating metrics, determines the granularity of aggregation.


prediction_col: Optional[str]
The name of the column that contains predictions.


actuals_col: Optional[str]
The name of the column that contains actuals.


timestamp_col: Optional[str]
The name of the column that contains timestamps.


filter_actuals: Optional[bool]
If True metric evaluator removes missing actuals values before scoring. The default value is False.


filter_predictions: Optional[bool]
If True metric evaluator removes missing predictions values before scoring. The default value is False.


filter_scoring_data: Optional[bool]
If True metric evaluator removes missing scoring values before scoring. The default value is False.


segment_attribute: Optional[str]
The name of the column with segment values.


segment_value: Optional[Union[str or List[str]]]
Single value or a list of values of the segment attribute to segment on.



Metric Evaluator Usage
The score method returns a metric aggregated as defined by `TimeBucket, The output returned as a pandas DataFrame contains
the results per time bucket for all data from the source.
source = DataRobotSource(
deployment_id=DEPLOYMENT_ID,
start=datetime.utcnow() - timedelta(hours=3),
end=datetime.utcnow(),
)
metric = LogLossFromSklearn()

me = MetricEvaluator(metric=metric, source=source, time_bucket=TimeBucket.HOUR)

aggregated_metric_per_time_bucket = me.score()
print(aggregated_metric_per_time_bucket.to_string())

timestamp samples log_loss
0 2023-09-14 13:29:48.065000+00:00 499 0.539315
1 2023-09-14 14:01:51.484000+00:00 499 0.539397

# we can see the evaluator's statistics
stats = me.stats()
print(stats)
total rows: 998, score calls: 2, reduce calls: 2

To pass more than one metric at a time, you can do the following:
metrics = [LogLossFromSklearn(), AsymmetricError(), RocAuc()]
me = MetricEvaluator(metric=metric, source=source, time_bucket=TimeBucket.HOUR)

aggregated_metric_per_time_bucket = me.score()
stats = me.stats()
print(aggregated_metric_per_time_bucket.to_string())
print(stats)

timestamp samples log_loss Asymmetric Error roc_auc_score
0 2023-09-14 13:29:48.065000+00:00 499 0.539315 0.365571 0.787030
1 2023-09-14 14:01:51.484000+00:00 499 0.539397 0.365636 0.786837
total rows: 998, score calls: 6, reduce calls: 6

For your own data, you can provide the names of the columns to evaluate:
test_df = gen_dataframe_for_accuracy_metric(
nr_rows=5,
rows_per_time_bucket=1,
prediction_value=1,
time_bucket=TimeBucket.DAY,
prediction_col="my_pred_col",
actuals_col="my_actuals_col",
timestamp_col="my_timestamp_col"
)
print(test_df)
my_timestamp_col my_pred_col my_actuals_col
0 01/06/2005 13:00:00.000000 1 0.999
1 02/06/2005 13:00:00.000000 1 0.999
2 03/06/2005 13:00:00.000000 1 0.999
3 04/06/2005 13:00:00.000000 1 0.999
4 05/06/2005 13:00:00.000000 1 0.999

source = DataFrameSource(
df=test_df,
max_rows=10000,
timestamp_col="timestamp",
)

metric = LogLossFromSklearn()

me = MetricEvaluator(metric=metric,
source=source,
time_bucket=TimeBucket.DAY,
prediction_col="my_pred_col",
actuals_col="my_actuals_col",
timestamp_col="my_timestamp_col"
)
aggregated_metric_per_time_bucket = me.score()

Metric Evaluator Data Filtering
If some data is missing, use filtering flags. For example, the following example is for data with missing actuals.
In this scenario without a flag, an exception is raised:
test_df = gen_dataframe_for_accuracy_metric(
nr_rows=10,
rows_per_time_bucket=5,
prediction_value=1,
time_bucket=TimeBucket.HOUR,
)
test_df["actuals"].loc[2] = None
test_df["actuals"].loc[5] = None
print(test_df)
timestamp predictions actuals
0 01/06/2005 13:00:00.000000 1 0.999
1 01/06/2005 13:00:00.000000 1 0.999
2 01/06/2005 13:00:00.000000 1 NaN
3 01/06/2005 13:00:00.000000 1 0.999
4 01/06/2005 13:00:00.000000 1 0.999
5 01/06/2005 14:00:00.000000 1 NaN
6 01/06/2005 14:00:00.000000 1 0.999
7 01/06/2005 14:00:00.000000 1 0.999
8 01/06/2005 14:00:00.000000 1 0.999
9 01/06/2005 14:00:00.000000 1 0.999

source = DataFrameSource(df=test_df)

metric = MedianAbsoluteError()

me = MetricEvaluator(metric=metric, source=source, time_bucket=TimeBucket.HOUR)

aggregated_metric_per_time_bucket = me.score()
"ValueError: Could not apply metric median_absolute_error, make sure you are passing the right data (see the sklearn docs).
The error message was: Input contains NaN."

For the same dataset, compare the previous result with the result when you enable the filter_actuals flag:
me = MetricEvaluator(metric=metric, source=source, time_bucket=TimeBucket.HOUR, filter_actuals=True)

aggregated_metric_per_time_bucket = me.score()
"removed 1 rows out of 5 in the data chunk before scoring, due to missing values in ['actuals'] data"
"removed 1 rows out of 5 in the data chunk before scoring, due to missing values in ['actuals'] data"

print(aggregated_metric_per_time_bucket.to_string())
timestamp samples median_absolute_error
0 01/06/2005 13:00:00.000000 4 0.001
1 01/06/2005 14:00:00.000000 4 0.001

Using the filter_actuals, filter_predictions, filter_scoring_data flags, you can filter out missing values from
the data before calculating the metric. By default, these flags are set to False.
If all data needed to calculate the metric is missing in the data chunk, we skip this data chunk with the appropriate log:
test_df = gen_dataframe_for_accuracy_metric(
nr_rows=4,
rows_per_time_bucket=2,
prediction_value=1,
time_bucket=TimeBucket.HOUR,
)
test_df["actuals"].loc[0] = None
test_df["actuals"].loc[1] = None
print(test_df)
timestamp predictions actuals
0 01/06/2005 13:00:00.000000 1 NaN
1 01/06/2005 13:00:00.000000 1 NaN
2 01/06/2005 14:00:00.000000 1 0.999
3 01/06/2005 14:00:00.000000 1 0.999

source = DataFrameSource(df=test_df)

metric = MedianAbsoluteError()

me = MetricEvaluator(metric=metric, source=source, time_bucket=TimeBucket.HOUR, filter_actuals=True)

aggregated_metric_per_time_bucket = me.score()
"removed 2 rows out of 2 in the data chunk before scoring, due to missing values in ['actuals'] data"
"data chunk is empty, skipping scoring..."

print(aggregated_metric_per_time_bucket.to_string())
timestamp samples median_absolute_error
1 01/06/2005 14:00:00.000000 2 0.001

Metric Evaluator Segmented Analysis
Perform segmented analysis by defining the segment_attribute and each segment_value:
metrics = LogLossFromSklearn()
me = MetricEvaluator(metric=metric,
source=source,
time_bucket=TimeBucket.HOUR,
segment_attribute="insulin",
segment_value="Down",
)

aggregated_metric_per_time_bucket = me.score()
print(aggregated_metric_per_time_bucket.to_string())
timestamp samples log_loss [Down]
0 2023-09-14 13:29:49.737000+00:00 49 0.594483
1 2023-09-14 14:01:52.437000+00:00 49 0.594483

# passing more than one segment value
me = MetricEvaluator(metric=metric,
source=source,
time_bucket=TimeBucket.HOUR,
segment_attribute="insulin",
segment_value=["Down", "Steady"],
)

aggregated_metric_per_time_bucket = me.score()
print(aggregated_metric_per_time_bucket.to_string())
timestamp samples log_loss [Down] log_loss [Steady]
0 2023-09-14 13:29:48.502000+00:00 199 0.594483 0.515811
1 2023-09-14 14:01:51.758000+00:00 199 0.594483 0.515811

# passing more than one segment value and more than one metric
me = MetricEvaluator(metric=[LogLossFromSklearn(), RocAuc()],
source=source,
time_bucket=TimeBucket.HOUR,
segment_attribute="insulin",
segment_value=["Down", "Steady"],
)

aggregated_metric_per_time_bucket = me.score()
print(aggregated_metric_per_time_bucket.to_string())
timestamp samples log_loss [Down] log_loss [Steady] roc_auc_score [Down] roc_auc_score [Steady]
0 2023-09-14 13:29:48.502000+00:00 199 0.594483 0.515811 0.783333 0.826632
1 2023-09-14 14:01:51.758000+00:00 199 0.594483 0.515811 0.783333 0.826632

Batch Metric Evaluator
The BatchMetricEvaluator class uses aggregation per batch instead of aggregation over time.
For batches, you don't define TimeBucket:
from dmm.batch_metric_evaluator import BatchMetricEvaluator
from dmm.data_source.datarobot_source import BatchDataRobotSource
from dmm.metric import MissingValuesFraction

source = BatchDataRobotSource(
deployment_id=DEPLOYMENT_ID,
batch_ids=BATCH_IDS,
model_id=MODEL_ID,
)

feature_name = 'RAD'
metric = MissingValuesFraction(feature_name=feature_name)

missing_values_fraction_evaluator = BatchMetricEvaluator(metric=metric, source=source)

aggregated_metric_per_batch = missing_values_fraction_evaluator.score()
print(aggregated_metric_per_batch.to_string())
batch_id samples Missing Values Fraction
0 <batch_id> 506 0.0
1 <batch_id> 506 0.0
2 <batch_id> 506 0.0


Note: For batches, actuals and multiple segments are not supported.

Individual Metric Evaluator
The IndividualMetricEvaluator class is used to evaluate metrics without data aggregation.
Perform metric calculations on all exported data, return a list of individual results.
This evaluator allows submitting individual data points with a corresponding association id.
This is useful for the cases when you want to visualize your metric results alongside predictions and actuals.
To use this evaluator with custom metric, it is necessary to provide score method that
contains, among others, the following parameters: 'timestamps' and 'association_ids'.
from itertools import zip_longest
from typing import List
from datetime import datetime
from datetime import timedelta

from dmm.individual_metric_evaluator import IndividualMetricEvaluator
from dmm.custom_metric import CustomMetric
from dmm.custom_metric import SingleMetricResult
from dmm.data_source import DataRobotSource
from dmm.metric.metric_base import LLMMetricBase
from nltk import sent_tokenize
import numpy as np
import pandas as pd

source = DataRobotSource(
deployment_id=DEPLOYMENT_ID,
start=datetime.utcnow() - timedelta(weeks=1),
end=datetime.utcnow(),
)

custom_metric = CustomMetric.from_id()

class SentenceCount(LLMMetricBase):
"""
Calculates the total number of sentences created while working with the LLM model.
Returns the sum of the number of sentences from prompts and completions.
"""

def __init__(self):
super().__init__(
name=custom_metric.name,
description="Calculates the total number of sentences created while working with the LLM model.",
need_training_data=False,
)
self.prompt_column = "promptColumn"

def score(
self,
scoring_data: pd.DataFrame,
predictions: np.array,
timestamps: np.array,
association_ids: np.array,
**kwargs,
) -> List[SingleMetricResult]:
if self.prompt_column not in scoring_data.columns:
raise ValueError(
f"Prompt column {self.prompt_column} not found in the exported data, "
f"modify 'PROMPT_COLUMN' runtime parameter"
)
prompts = scoring_data[self.prompt_column].to_numpy()

sentence_count = []
for prompt, completion, ts, a_id in zip_longest(
prompts, predictions, timestamps, association_ids
):
if not isinstance(prompt, str) or not isinstance(completion, str):
continue
value = len(sent_tokenize(prompt)) + len(sent_tokenize(completion))
sentence_count.append(
SingleMetricResult(value=value, timestamp=ts, association_id=a_id)
)
return sentence_count


sentence_count_evaluator = IndividualMetricEvaluator(
metric=SentenceCount(),
source=source,
)
metric_results = sentence_count_evaluator.score()

DR Custom Metrics
The DR Custom Metrics module allows better synchronization with existing
metrics on the DR side. The logic of this module is based on unique names for custom metrics, so you can operate
on metrics without knowing their IDs. Thanks to this solution, we can define the metric earlier (e.g. before creating
the deployment) and synchronize it with DR at the appropriate time.
DRCustomMetric:
This class DRCustomMetric allows you to create new or fetch existing metrics from DR.
the logic is as follows:

you can provide custom metrics configuration in the selected format: YAML, dict, YAML file, JSON file.
the configuration contains custom metrics metadata that describe them.
DRCustomMetric.sync() method retrieves information about existing custom metrics on the DR side, if a metric is
defined locally but is not on the DR side, it will be created on the DR side.
DRCustomMetric.report() method allows you to report a single value based on a unique name.

Example:
dr_cm = DRCustomMetric(
dr_client=client, deployment_id=deployment_id, model_package_id=model_package_id
)

metric_config_yaml = f"""
customMetrics:
- name: new metric
description: foo bar
type: average
timeStep: hour
units: count
directionality: lowerIsBetter
isModelSpecific: yes
baselineValue: 0
"""

dr_cm.set_config(config_yaml=metric_config_yaml)
dr_cm.sync()
dr_cm.get_dr_custom_metrics()
> [{"name": "existing metric", "id": "65ef19410239ff8015f05a94", ...},
> {"name": "new metric", "id": "65ef197ce5d7b2176ceecf3a", ...}]

dr_cm.report_value("existing metric", 1)
dr_cm.report_value("new metric", 9)

License

For personal and professional use. You cannot resell or redistribute these repositories in their original state.

Share

• Share on Facebook
• Share on Twitter