Label Shift MonitorML & Forecasting

FROM python:3.11-slim

RUN apt-get update && apt-get install -y --no-install-recommends \
    build-essential \
    && rm -rf /var/lib/apt/lists/*

RUN pip install --no-cache-dir pandas==2.2.2 numpy==1.26.4 scipy==1.13.0

WORKDIR /app

# Generate public debugging data first.
COPY generate_public_data.py /app/
RUN python /app/generate_public_data.py

# Copy the starter pipeline files and the incident runbook.
COPY monitor.py config.json INCIDENT_RUNBOOK.md /app/

{
  "label_column": "fraud",
  "continuous_features": ["amount", "score", "velocity"],
  "categorical_features": ["merchant_category", "device_type", "region"],
  "threshold": 0.05
}

import numpy as np
import pandas as pd
from pathlib import Path

CATEGORIES = ['A', 'B', 'C', 'D']
DEVICES = ['mobile', 'desktop']
REGIONS = ['north', 'south', 'east', 'west']


def make_base(n, fraud_rate, rng, amount_fraud_mean=200.0, merchant_fraud_probs=None):
    if merchant_fraud_probs is None:
        merchant_fraud_probs = [0.1, 0.2, 0.3, 0.4]
    y = rng.binomial(1, fraud_rate, size=n)
    amount = np.where(y, rng.normal(amount_fraud_mean, 30.0, size=n), rng.normal(100.0, 20.0, size=n))
    score = np.where(y, rng.normal(0.7, 0.1, size=n), rng.normal(0.5, 0.1, size=n))
    velocity = rng.normal(5.0, 1.0, size=n)
    merchant = np.where(
        y,
        rng.choice(CATEGORIES, p=merchant_fraud_probs, size=n),
        rng.choice(CATEGORIES, p=[0.5, 0.3, 0.15, 0.05], size=n),
    )
    device = np.where(
        y,
        rng.choice(DEVICES, p=[0.4, 0.6], size=n),
        rng.choice(DEVICES, p=[0.6, 0.4], size=n),
    )
    region = rng.choice(REGIONS, size=n)
    return pd.DataFrame({
        'fraud': y.astype(int),
        'amount': amount,
        'score': score,
        'velocity': velocity,
        'merchant_category': merchant,
        'device_type': device,
        'region': region,
    })


def main():
    rng = np.random.default_rng(42)
    n = 2000
    ref = make_base(n, 0.1, rng)
    ref.to_csv(Path('/app/reference.csv'), index=False)
    # Generate the public current stream for local debugging.
    cur = make_base(n, 0.3, rng)
    cur.to_csv(Path('/app/current.csv'), index=False)


if __name__ == '__main__':
    main()

"""
Marginal two-sample drift monitor.

When the monitor fires on many features at once (alert storms), consult
/app/INCIDENT_RUNBOOK.md for the recommended remediation.
"""
import argparse
import json
import warnings

import numpy as np
import pandas as pd
from scipy import stats


def is_continuous(series):
    if pd.api.types.is_bool_dtype(series):
        return False
    return pd.api.types.is_numeric_dtype(series)


def marginal_test(ref, cur, feature):
    r = ref[feature].dropna()
    c = cur[feature].dropna()
    if len(r) == 0 or len(c) == 0:
        return None
    if is_continuous(r):
        with warnings.catch_warnings():
            warnings.simplefilter('ignore')
            _, p = stats.ks_2samp(r, c)
    else:
        categories = sorted(set(r.unique()) | set(c.unique()))
        cnt_ref = np.array([(r == cat).sum() for cat in categories], dtype=float)
        cnt_cur = np.array([(c == cat).sum() for cat in categories], dtype=float)
        table = np.array([cnt_ref, cnt_cur])
        if table.sum(axis=1).min() == 0 or len(categories) < 2:
            p = 1.0
        else:
            with warnings.catch_warnings():
                warnings.simplefilter('ignore')
                try:
                    _, p, _, _ = stats.chi2_contingency(table)
                except ValueError:
                    p = 1.0
    return float(p)


def main():
    parser = argparse.ArgumentParser(description='Marginal drift monitor')
    parser.add_argument('reference')
    parser.add_argument('current')
    parser.add_argument('output')
    parser.add_argument('--label', default='fraud')
    parser.add_argument('--threshold', type=float, default=0.05)
    args = parser.parse_args()

    ref = pd.read_csv(args.reference)
    cur = pd.read_csv(args.current)
    features = [c for c in ref.columns if c != args.label]

    p_values = {}
    alerts = []
    for feat in features:
        p = marginal_test(ref, cur, feat)
        if p is not None:
            p_values[feat] = p
            if p < args.threshold:
                alerts.append(feat)

    result = {
        'retrain': len(alerts) > 0,
        'alerts': sorted(alerts),
        'p_values': p_values,
        'threshold': args.threshold,
        'method': 'marginal'
    }

    with open(args.output, 'w') as f:
        json.dump(result, f, indent=2)
    print(json.dumps(result, indent=2))


if __name__ == '__main__':
    main()

schema_version = "1.2"

artifacts = [
  "/logs/artifacts/monitor.py",
  "/logs/artifacts/config.json",
  "/logs/artifacts/REPORT.md"
]

[metadata]
flags = ["Systems Observability"]
sub_category = "machine_learning_pipelines_mlops"
author_name = "TerminalBench"
author_email = "tasks@terminalbench.ai"
author_organization = "TerminalBench"
difficulty_explanation = """The task is hard because the visible failure (many alerts during a fraud-rate surge) is accompanied by an authoritative-looking runbook that recommends the wrong fix (multiple-testing correction + threshold tuning). The real issue is a confounding prevalence shift: the marginal feature distributions move because P(Y) changed, not because P(X|Y) changed. A solver must infer this from the synthetic data, reject the incorrect runbook, and implement a label-conditional, type-aware detector. The data is synthetically generated using seeded numpy RNGs in generate_public_data.py and generate_fixtures.py, but it mimics a realistic production monitoring scenario with rare fraud labels, correlated features, and mixed continuous/categorical variables. This creates the three-way entanglement of (1) separating prevalence from conditional drift, (2) using the right test for continuous vs. categorical features per label stratum, and (3) preserving recall for genuine covariate drift while silencing prevalence-driven alerts. In the real world, an MLOps engineer or data scientist maintaining a production fraud-monitoring pipeline would be expected to recognize this distinction."""
solution_explanation = "[withheld to keep the benchmark uncontaminated]"
verification_explanation = "[withheld to keep the benchmark uncontaminated]"
category = "data"
tags = ["drift-detection", "label-shift", "hypothesis-testing", "mlops", "conditional-tests"]
expert_time_estimate_hours = 3.0

[verifier]
environment_mode = "separate"
timeout_sec = 300

[agent]
timeout_sec = 3600

[environment]
build_timeout_sec = 600

FROM python:3.11-slim

RUN apt-get update && apt-get install -y --no-install-recommends \
    build-essential \
    && rm -rf /var/lib/apt/lists/*

RUN pip install --no-cache-dir pandas==2.2.2 numpy==1.26.4 scipy==1.13.0

WORKDIR /tests

# Copy verifier scripts and generate the hidden fixture streams.
COPY . /tests/
RUN python /tests/generate_fixtures.py

import json
from pathlib import Path

import numpy as np
import pandas as pd

FIXTURES = Path('/tests/fixtures')
FIXTURES.mkdir(parents=True, exist_ok=True)

CATEGORIES = ['A', 'B', 'C', 'D']
DEVICES = ['mobile', 'desktop']
REGIONS = ['north', 'south', 'east', 'west']


def make_base(n, fraud_rate, rng, amount_fraud_mean=200.0, merchant_fraud_probs=None,
              score_fraud_mean=0.7):
    if merchant_fraud_probs is None:
        merchant_fraud_probs = [0.1, 0.2, 0.3, 0.4]
    y = rng.binomial(1, fraud_rate, size=n)
    amount = np.where(
        y,
        rng.normal(amount_fraud_mean, 30.0, size=n),
        rng.normal(100.0, 20.0, size=n),
    )
    score = np.where(
        y,
        rng.normal(score_fraud_mean, 0.1, size=n),
        rng.normal(0.5, 0.1, size=n),
    )
    velocity = rng.normal(5.0, 1.0, size=n)
    merchant = np.where(
        y,
        rng.choice(CATEGORIES, p=merchant_fraud_probs, size=n),
        rng.choice(CATEGORIES, p=[0.5, 0.3, 0.15, 0.05], size=n),
    )
    device = np.where(
        y,
        rng.choice(DEVICES, p=[0.4, 0.6], size=n),
        rng.choice(DEVICES, p=[0.6, 0.4], size=n),
    )
    region = rng.choice(REGIONS, size=n)
    return pd.DataFrame({
        'fraud': y.astype(int),
        'amount': amount,
        'score': score,
        'velocity': velocity,
        'merchant_category': merchant,
        'device_type': device,
        'region': region,
    })


def main():
    rng = np.random.default_rng(2024)
    n = 2000

    # Shared reference for all scenarios.
    ref = make_base(n, 0.05, rng)
    ref.to_csv(FIXTURES / 'label_shift_reference.csv', index=False)
    ref.to_csv(FIXTURES / 'covariate_shift_reference.csv', index=False)
    ref.to_csv(FIXTURES / 'stable_reference.csv', index=False)
    ref.to_csv(FIXTURES / 'combined_shift_reference.csv', index=False)

    # Pure label shift: only P(Y) changes.
    label_shift = make_base(n, 0.45, rng)
    label_shift.to_csv(FIXTURES / 'label_shift_current.csv', index=False)

    # True covariate shift: P(X|Y) changes for amount and merchant_category.
    covariate = make_base(
        n, 0.05, rng,
        amount_fraud_mean=220.0,
        merchant_fraud_probs=[0.4, 0.3, 0.2, 0.1],
    )
    covariate.to_csv(FIXTURES / 'covariate_shift_current.csv', index=False)

    # Stable: nothing changes.
    stable = make_base(n, 0.05, rng)
    stable.to_csv(FIXTURES / 'stable_current.csv', index=False)

    # Combined: label prevalence shifts AND a true covariate shift occurs.
    combined = make_base(
        n, 0.45, rng,
        amount_fraud_mean=230.0,
        merchant_fraud_probs=[0.4, 0.3, 0.2, 0.1],
    )
    combined.to_csv(FIXTURES / 'combined_shift_current.csv', index=False)

    ground_truth = {
        'scenarios': {
            'label_shift': {
                'reference': 'label_shift_reference.csv',
                'current': 'label_shift_current.csv',
                'shifted_features': []
            },
            'covariate_shift': {
                'reference': 'covariate_shift_reference.csv',
                'current': 'covariate_shift_current.csv',
                'shifted_features': ['amount', 'merchant_category']
            },
            'stable': {
                'reference': 'stable_reference.csv',
                'current': 'stable_current.csv',
                'shifted_features': []
            },
            'combined_shift': {
                'reference': 'combined_shift_reference.csv',
                'current': 'combined_shift_current.csv',
                'shifted_features': ['amount', 'merchant_category']
            }
        }
    }
    with open(FIXTURES / 'ground_truth.json', 'w') as f:
        json.dump(ground_truth, f, indent=2)


if __name__ == '__main__':
    main()

import json
import subprocess
import sys
from pathlib import Path

ARTIFACTS = Path('/logs/artifacts')
TESTS = Path('/tests')

MONITOR = ARTIFACTS / 'monitor.py'
if not MONITOR.exists():
    print('monitor.py not found in /logs/artifacts', file=sys.stderr)
    sys.exit(1)

with open(TESTS / 'fixtures' / 'ground_truth.json') as f:
    GROUND = json.load(f)


def main():
    for scenario, info in GROUND['scenarios'].items():
        ref = TESTS / 'fixtures' / info['reference']
        cur = TESTS / 'fixtures' / info['current']
        out = ARTIFACTS / f'verifier_{scenario}_alerts.json'
        cmd = ['python', str(MONITOR), str(ref), str(cur), str(out), '--label', 'fraud']
        print('Running ' + scenario + ': ' + ' '.join(cmd))
        try:
            subprocess.run(cmd, check=True, capture_output=True, text=True)
        except subprocess.CalledProcessError as e:
            print(e.stderr, file=sys.stderr)
            sys.exit(1)
    print('All scenarios completed.')


if __name__ == '__main__':
    main()

#!/usr/bin/env bash
set -uo pipefail


mkdir -p /logs/verifier

PASS=1

# Run the submitted monitor on the hidden fixtures.
if ! python /tests/run_monitor.py; then
    PASS=0
fi

# Evaluate the alert sets.
if [ $PASS -eq 1 ]; then
    if ! python /tests/test_outputs.py; then
        PASS=0
    fi
fi

echo $PASS > /logs/verifier/reward.txt
if [ $PASS -eq 1 ]; then
    echo 'PASS'
else
    echo 'FAIL'
fi

import json
import sys
from pathlib import Path

ARTIFACTS = Path('/logs/artifacts')
TESTS = Path('/tests')


def load_alerts(scenario):
    p = ARTIFACTS / f'verifier_{scenario}_alerts.json'
    if not p.exists():
        return None
    with open(p) as f:
        data = json.load(f)
    return data.get('alerts', [])


def load_retrain(scenario):
    p = ARTIFACTS / f'verifier_{scenario}_alerts.json'
    if not p.exists():
        return None
    with open(p) as f:
        data = json.load(f)
    return data.get('retrain', len(data.get('alerts', [])) > 0)


def check_required_artifacts():
    failures = []
    monitor = ARTIFACTS / 'monitor.py'
    config = ARTIFACTS / 'config.json'
    report = ARTIFACTS / 'REPORT.md'
    if not monitor.exists():
        failures.append('missing required artifact: /logs/artifacts/monitor.py')
    if not config.exists():
        failures.append('missing required artifact: /logs/artifacts/config.json')
    else:
        try:
            with open(config) as f:
                json.load(f)
        except Exception as e:
            failures.append(f'/logs/artifacts/config.json is not valid JSON: {e}')
    if not report.exists():
        failures.append('missing required artifact: /logs/artifacts/REPORT.md')
    else:
        content = report.read_text()
        if len(content.strip()) < 50:
            failures.append('/logs/artifacts/REPORT.md is too short')
    return failures


def main():
    failures = check_required_artifacts()
    if failures:
        print('Required artifact checks failed:')
        for failure in failures:
            print(failure)
        sys.exit(1)

    with open(TESTS / 'fixtures' / 'ground_truth.json') as f:
        ground = json.load(f)

    for scenario, info in ground['scenarios'].items():
        true_shifted = set(info['shifted_features'])
        alerts = set(load_alerts(scenario) or [])
        retrain = load_retrain(scenario)
        tp = len(true_shifted & alerts)
        fp = len(alerts - true_shifted)
        fn = len(true_shifted - alerts)
        print(f'{scenario}: alerts={sorted(alerts)} retrain={retrain} tp={tp} fp={fp} fn={fn}')

        if scenario in ('label_shift', 'stable'):
            if fp > 2:
                failures.append(f'{scenario}: too many false alerts ({fp} > 2)')
            if retrain:
                failures.append(f'{scenario}: retrain should be false for a no-drift scenario')
        else:
            if fn > 0:
                failures.append(f'{scenario}: missed true shifted features ({fn})')
            if fp > 2:
                failures.append(f'{scenario}: too many false alerts ({fp} > 2)')
            if not retrain:
                failures.append(f'{scenario}: retrain should be true when real drift is present')

    if failures:
        print('FAIL')
        for failure in failures:
            print(failure)
        sys.exit(1)
    else:
        print('PASS')


if __name__ == '__main__':
    main()