Skip to main content

Drift Detection

Model drift detection: predictions, ground truth, baselines, analyses.

Access it as client.drift_detection on a Strongly client, or the same path on AsyncStrongly with await. All methods exist on both with identical signatures.

Methods​

All methods​

activate_baseline​

activate_baseline(baseline_id: str, *, model_id: str) -> Dict[str, Any]

Set a specific baseline as the active one for drift analysis.

active_baselines​

active_baselines(model_id: str) -> Dict[str, Any]

Get the currently active reference baseline for a model.

alerts​

alerts(model_id: str) -> Dict[str, Any]

Get the drift alert configuration for a model.

analyze​

analyze(*, model_id: str, window_days: Optional[int] = None, window_start: Optional[str] = None, window_end: Optional[str] = None) -> Dict[str, Any]

Kick off a drift analysis run over a time window.

run = client.drift_detection.analyze(model_id="model_abc123", window_days=7)

analyze_status​

analyze_status(analysis_id: str, *, model_id: str) -> Dict[str, Any]

Get the status of a drift analysis run. model_id is required alongside the analysis (job) id returned by analyze().

create_baseline​

create_baseline(*, model_id: str, version: str, feature_data: Mapping[str, Sequence[Any]], probabilities: Optional[Sequence[Sequence[float]]] = None, predicted_labels: Optional[Sequence[Any]] = None, actual_labels: Optional[Sequence[Any]] = None) -> Dict[str, Any]

Create a reference baseline from training data.

The model must have been registered with a problem_type (stored at training.problemType); drift detection needs it to tell classification from regression. If it is missing, create_baseline raises an error - set problem_type when you register the model.

Args: model_id: Model id from the model registry. version: Baseline version tag. feature_data: Training samples per feature, {feature_name: [values, ...]}. probabilities: Optional per-sample probability vectors ([[p0, p1], ...]). predicted_labels: Optional model-predicted label per sample. actual_labels: Optional ground-truth label per sample. Supplying these three together unlocks calibrated CBPE and prediction-drift.

client.drift_detection.create_baseline(
    model_id="model_abc123",
    version="v1",
    feature_data={"age": [31, 42, 28], "income": [50000, 82000, 41000]},
    probabilities=[[0.8, 0.2], [0.3, 0.7], [0.9, 0.1]],
    predicted_labels=["good", "bad", "good"],
    actual_labels=["good", "good", "good"],
)

latest_results​

latest_results(model_id: str) -> Dict[str, Any]

Get the most recent drift-analysis results for a model.

list_baselines​

list_baselines(model_id: str) -> Dict[str, Any]

List all reference baselines for a model.

list_predictions​

list_predictions(model_id: str, *, limit: Optional[int] = None, offset: Optional[int] = None, start_date: Optional[str] = None, end_date: Optional[str] = None, has_ground_truth: Optional[bool] = None) -> Dict[str, Any]

List prediction logs for a model.

Args: model_id: Model id from the model registry. limit: Max results (default 100 server-side). offset: Pagination offset. start_date: ISO-8601 lower bound. end_date: ISO-8601 upper bound. has_ground_truth: Restrict to predictions matched (or not) with ground truth.

log_ground_truth​

log_ground_truth(*, model_id: str, entity_id: str, actual_outcome: Any) -> Dict[str, Any]

Add one ground-truth outcome, matched to a prediction by entity id.

client.drift_detection.log_ground_truth(
    model_id="model_abc123",
    entity_id="order-1042",
    actual_outcome="churned",
)

log_ground_truth_batch​

log_ground_truth_batch(*, model_id: str, records: Sequence[Mapping[str, Any]]) -> Dict[str, Any]

Bulk-upload ground-truth outcomes for a model.

Args: model_id: Model id from the model registry. records: A sequence of plain dicts with camelCase wire keys, e.g. {"entityId": ..., "actualOutcome": ..., "outcomeTimestamp": ...}.

client.drift_detection.log_ground_truth_batch(
    model_id="model_abc123",
    records=[
        {"entityId": "order-1042", "actualOutcome": "churned"},
        {"entityId": "order-1043", "actualOutcome": "retained"},
    ],
)

log_prediction​

log_prediction(*, model_id: str, features: Mapping[str, Any], prediction: Any, probabilities: Optional[Sequence[float]] = None, entity_id: Optional[str] = None) -> Dict[str, Any]

Log a production prediction for drift detection.

Args: model_id: Model id from the model registry. features: Input feature values as a name -> value mapping. prediction: The model's prediction output. probabilities: Per-class probability scores (recommended for classification - the max is the confidence CBPE uses). entity_id: Id used to match a later ground-truth record.

client.drift_detection.log_prediction(
    model_id="model_abc123",
    features={"age": 31, "income": 50000},
    prediction="churned",
    probabilities=[0.18, 0.82],
    entity_id="order-1042",
)

performance​

performance(model_id: str) -> Dict[str, Any]

Get model performance metrics derived from matched ground truth.

results​

results(model_id: str, *, limit: Optional[int] = None) -> Dict[str, Any]

Get historical drift-analysis results for a model.

unmatched_predictions​

unmatched_predictions(model_id: str, *, limit: Optional[int] = None) -> Dict[str, Any]

List predictions not yet matched with ground truth for a model.

update_alerts​

update_alerts(*, model_id: str, **thresholds: Any) -> Dict[str, Any]

Update the drift alert configuration.

Args: model_id: Model id from the model registry. **thresholds: Alert threshold settings (algorithm-specific), e.g. enabled=True, psi_threshold=0.2.

client.drift_detection.update_alerts(
    model_id="model_abc123", enabled=True, psi_threshold=0.2,
)