ml-slop: Simple offer prediction service using mlcore patterns

deploy.py

"""
Deploy offer prediction service using mlcore.

Per mlcore guide sections 5.1-5.3:
- model_path is required for ContainerBuilder
- s3_model_uri must be a .tar.gz file
- model_name must follow <inference_type>-<model_framework>-<model_type>

Since our "model" is parquet data loaded at runtime, we create a placeholder
model artifact to satisfy SageMaker requirements.
"""

import tarfile
import tempfile
from pathlib import Path

import boto3

S3_BUCKET = "even-ml"
S3_MODEL_KEY = "models/offer-prediction/model.tar.gz"


def create_placeholder_model() -> Path:
    """
    Create a placeholder model file for SageMaker.

    The actual data is loaded from S3 parquet at runtime by the context.
    This just satisfies the model artifact requirement.
    """
    tmpdir = Path(tempfile.mkdtemp())
    model_dir = tmpdir / "model"
    model_dir.mkdir()

    # Create a simple config file as the "model"
    config_file = model_dir / "config.json"
    config_file.write_text('{"type": "parquet_lookup", "version": "1.0"}')

    return config_file


def package_and_upload_model() -> str:
    """
    Package placeholder model as tar.gz and upload to S3.

    Returns:
        S3 URI of uploaded model artifact
    """
    model_file = create_placeholder_model()

    with tempfile.NamedTemporaryFile(suffix=".tar.gz", delete=False) as tmp:
        tar_path = Path(tmp.name)

    with tarfile.open(tar_path, "w:gz") as tar:
        tar.add(model_file, arcname=model_file.name)

    s3 = boto3.client("s3")
    s3.upload_file(str(tar_path), S3_BUCKET, S3_MODEL_KEY)

    s3_uri = f"s3://{S3_BUCKET}/{S3_MODEL_KEY}"
    print(f"Model uploaded to {s3_uri}")
    return s3_uri


def deploy_online_endpoint():
    """
    Deploy as online endpoint per mlcore guide section 5.2.

    Uses custom model context (section 5.3).
    """
    from mlcore import (
        ContainerBuilder,
        InferenceMode,
        OnlineEndpointDeployment,
        OnlineEndpointDeployOptions,
        InstanceType,
    )

    # 1. Create and upload placeholder model artifact
    s3_model_uri = package_and_upload_model()

    # 2. Get path to placeholder model for ContainerBuilder
    model_file = create_placeholder_model()

    # 3. Build container with custom model context
    container = ContainerBuilder(
        model_path=str(model_file),
        output_dir="build_context",
        mode=InferenceMode.ONLINE_ENDPOINT,
        pyproject_path="pyproject.toml",
        uv_lock_path="uv.lock",
        model_context_path="offer_prediction_context.py",
        model_context_class="OfferPredictionContext",
    )

    container.prepare_container_context()
    container.build(repo_name="offer-prediction", tag="latest")
    container.push("offer-prediction:latest")

    # 4. Configure deployment per section 5.2.3
    # Model name follows convention: <inference_type>-<model_framework>-<model_type>
    deploy_options = OnlineEndpointDeployOptions(
        model_name="loanofferrecommendation-sklearn-classifier",
        image_uri="617208391173.dkr.ecr.us-east-1.amazonaws.com/offer-prediction:latest",
        s3_model_uri=s3_model_uri,
        production_variants=[
            OnlineEndpointDeployOptions.ProductionVariants(
                variant_name="primary",
                instance_type=InstanceType.ML_C5_XLARGE,
                initial_instance_count=1,
                max_concurrency=4,
            )
        ],
        container_builder=container,
    )

    # 5. Deploy
    deployment = OnlineEndpointDeployment.from_options(deploy_options)
    result = deployment.deploy()

    if result.success:
        print(f"Endpoint deployed: {result.endpoint_name}")
        print(f"Endpoint ARN: {result.endpoint_arn}")
    else:
        print(f"Failed: {result.error_message}")

    return result


if __name__ == "__main__":
    deploy_online_endpoint()

offer_prediction_context.py

"""
Custom ModelContext for offer predictions.

Loads pre-computed CTR/RPI/RPC metrics from parquet.
Implements mlcore BaseModelContext interface per section 5.3.
"""

import logging
import tempfile
from pathlib import Path
from typing import Any, Union

import boto3
import pandas as pd

logger = logging.getLogger(__name__)

S3_BUCKET = "even-ml"
S3_PREFIX = "dbt/generic_offer_predictions/"


class OfferPredictionContext:
    """
    Model context that loads offer predictions from S3 parquet.

    Implements BaseModelContext interface:
    - __init__(model_path): Initialize with model path (required by mlcore)
    - load(): Load the "model" (parquet data from S3)
    - predict(data: pd.DataFrame) -> list[float]: Return predictions
    - preprocess(data: pd.DataFrame) -> pd.DataFrame: Optional preprocessing
    - postprocess(predictions: list[float]) -> list[float]: Optional postprocessing
    """

    def __init__(self, model_path: Union[str, Path] = "/opt/ml/model"):
        """
        Initialize the context.

        Args:
            model_path: Path to model directory (required by mlcore interface)
        """
        self.model_path = Path(model_path)
        self.data: pd.DataFrame | None = None

    def load(self) -> Any:
        """
        Load prediction data from S3 parquet files.

        Returns:
            The loaded DataFrame (the "model")
        """
        logger.info(f"Loading data from s3://{S3_BUCKET}/{S3_PREFIX}")

        s3 = boto3.client("s3")
        response = s3.list_objects_v2(Bucket=S3_BUCKET, Prefix=S3_PREFIX)

        parquet_keys = [
            obj["Key"]
            for obj in response.get("Contents", [])
            if obj["Key"].endswith(".parquet")
        ]

        if not parquet_keys:
            raise ValueError(f"No parquet files at s3://{S3_BUCKET}/{S3_PREFIX}")

        dfs = []
        with tempfile.TemporaryDirectory() as tmpdir:
            for key in parquet_keys:
                local_path = Path(tmpdir) / Path(key).name
                s3.download_file(S3_BUCKET, key, str(local_path))
                dfs.append(pd.read_parquet(local_path))

        self.data = pd.concat(dfs, ignore_index=True)
        logger.info(f"Loaded {len(self.data)} offer predictions")

        return self.data

    def preprocess(self, data: pd.DataFrame) -> pd.DataFrame:
        """
        Preprocess input data before prediction.

        Args:
            data: Input DataFrame with offer_id column

        Returns:
            Preprocessed DataFrame
        """
        logger.info(f"Preprocessing {len(data)} records")
        return data

    def predict(self, data: pd.DataFrame) -> list[float]:
        """
        Return CTR predictions for given offers.

        Args:
            data: DataFrame with offer_id column

        Returns:
            List of CTR values as floats
        """
        if self.data is None:
            raise ValueError("Data not loaded. Call load() first.")

        logger.info(f"Predicting for {len(data)} records")

        processed = self.preprocess(data)

        merged = processed.merge(
            self.data[["offer_id", "ctr"]],
            on="offer_id",
            how="left",
        )

        predictions = merged["ctr"].fillna(0.0).tolist()

        logger.info(f"Returning {len(predictions)} predictions")
        return predictions

    def postprocess(self, predictions: list[float]) -> list[float]:
        """
        Postprocess predictions.

        Args:
            predictions: Raw predictions

        Returns:
            Processed predictions (pass-through)
        """
        return predictions

predict.py

"""
Make predictions using pre-computed offer metrics.

Uses OfferPredictionContext to load and query parquet data.
"""

import pandas as pd

from offer_prediction_context import OfferPredictionContext


def predict(offer_ids: list[int]) -> pd.DataFrame:
    """
    Get predictions for a list of offer IDs.

    Returns DataFrame with offer_id, ctr, rpi, rpc.
    """
    context = OfferPredictionContext()
    context.load()

    input_df = pd.DataFrame({"offer_id": offer_ids})
    return context.get_full_predictions(input_df)


def main():
    """Example usage."""
    context = OfferPredictionContext()
    context.load()

    # Show sample of available data
    print("Sample predictions from parquet:")
    print(context.data[["offer_id", "impression_cnt", "click_cnt", "ctr", "rpi", "rpc"]].head(10))

    # Example lookup
    sample_ids = context.data["offer_id"].head(3).tolist()
    print(f"\nLooking up offers: {sample_ids}")

    input_df = pd.DataFrame({"offer_id": sample_ids})
    predictions = context.predict(input_df)
    print(f"CTR predictions: {predictions}")


if __name__ == "__main__":
    main()

pyproject.toml

[project]
name = "ml-slop"
version = "0.1.0"
description = "Offer prediction service using analytics-dbt parquet data"
requires-python = ">=3.11"
dependencies = [
    "pandas>=2.0",
    "pyarrow>=14.0",
    "boto3>=1.34",
    "numpy>=1.9",
]

[project.optional-dependencies]
dev = [
    "pytest>=7.0",
    "black>=23.0",
    "isort>=5.0",
]

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"