2025-06-05 - Dev Log

Finishing the pipeline

There are different degrees of “Finished” after all. So, I’ve added a final step to the current pipeline that packages the newly generated model up into a docker container running a FastAPI python app providing 1 endpoint that can be queried to get the predicated average bandwidth consumption rate for the next hour.

I chose to shoot for just “next hour” predictions due to observed lack of predictive utility due to the almost complete reliance on 1 and 2-hour lagged features. This ultimately results in oscillating predictions when the model is used recursively to predict further than 2 or 3 hours into the future.

Ideas for improving the model

  • I think it’s possible to create a model that generates predictions for several periods at once. 1 prediction providing values for the next 30 days in one shot for example.
  • Also, I initially switched from the diff columns to the rate columns for predictions since the rate columns accounted for varying row durations. Since I ended up re-sampling the data with pandas anyway, I think it might be better to train the model to predict diff column values.
    • Similarly, I chose to just predict the total column for the sake of simplicity while getting familiar with all of the moving parts. I think the next version of the model will predict both u_diff and d_diff. (t_diff of course being equal to u_diff + d_diff)
  • I’ve been doing a little reading and it sounds like Prophet from Facebook might be worth trying to see how it compares to XGBoost for this application.
  • I need to add the data cap back in as a feature after the re-sampling, I think that would have some predictive value
    • Speaking of, the bandwidth cap was applied monthly when there was one. So probably the best way to go is generating a feature called something like cap_utilization which would be a percentage of whatever the data cap was at that time based on the current cumulative usage for the month at that time.
    • This absolutely did used to influence our behaviour/usage of the internet as we neared the cap; and I still do ‘pretend’ the 6000GB cap is in effect. So I think there is potential here.

Anyway, deploying with Docker

Conceptual Sequence Diagram

sequenceDiagram
  actor Requestor as Requestor
  participant API as API
  participant Postgres as Postgres

  API ->> API: Model, scaler, and feature list<br>loaded from disk at startup
  Requestor ->>+ API: HTTP GET /predict/next-hour
  API ->>+ Postgres: SELECT Most recent 720 hours of historical data
  Postgres ->>- API: Banwdith history data
  API ->> API: Fetched data fed through same<br>feature engineering function from model training
  API ->>- Requestor: Predicted t_rate value for next hour

New Pipeline Step

Relevant directory structure in main project folder

api_deploy/
├─ api_service.py
├─ Dockerfile
├─ requirements.txt
model_files/
├─ scaler.joblib
├─ xgboost_model_optuna.joblib
├─ feature_list.json

I can’t believe I hadn’t ever tried automating the Docker teardown, rebuild, deploy cycle in python before. This is pretty slick. I especially like pulling the postgres connection details from the zenml secrets and passing them directly to the container as env variables.

# disabled caching during dev because changes to the docker container specific files
# do not trigger zenml to re-run this step
@step(enable_cache=False)
def deploy_model_api(
    model_path: str,
    scaler_path: str,
    postgres_secret_name: str = POSTGRES_SECRET_NAME,
    container_name: str = "bandwidth-prediction-api",
    port: int = 19000
) -> str:
    """
    Deploy the trained model as a Docker container with FastAPI
    """
    import docker
    import shutil
    import tempfile
    
    # Create temporary directory for Docker build context
    with tempfile.TemporaryDirectory() as temp_dir:
        # Copy model files
        models_dir = os.path.join(temp_dir, "models")
        os.makedirs(models_dir, exist_ok=True)
        
        shutil.copy(model_path, models_dir) # yeah, this needs re-worked...
        shutil.copy(scaler_path, models_dir)
        shutil.copy(os.path.join(MODEL_OUT, "feature_list.json"), models_dir)
        
        # Copy API files
        api_files = ["api_deploy/api_service.py", "api_deploy/requirements.txt", "api_deploy/Dockerfile"]
        for file in api_files:
            if os.path.exists(file):
                shutil.copy(file, temp_dir)
        
        # Get database credentials, going to pass these as env variables directly to container when we run it below
        secret = Client().get_secret(postgres_secret_name)
        
        # Get reference to docker daemon
        client = docker.from_env()
        
        # Stop and remove existing container if it exists
        try:
            existing_container = client.containers.get(container_name)
            existing_container.stop()
            existing_container.remove()
            print(f"Removed existing container: {container_name}")
        except docker.errors.NotFound:
            pass
        
        # Build new image
        image, logs = client.images.build(
            path=temp_dir,
            tag=f"{container_name}:latest",
            rm=True
        )
        
        # Run container
        container = client.containers.run(
            image.id,
            name=container_name,
            ports={8000: port},
            environment={
                "DB_HOST": secret.secret_values["host"],
                "DB_PORT": secret.secret_values["port"],
                "DB_NAME": secret.secret_values["dbname"],
                "DB_USER": secret.secret_values["username"],
                "DB_PASSWORD": secret.secret_values["password"]
            },
            detach=True,
            restart_policy={"Name": "unless-stopped"}
        )
        
        print(f"Deployed model API container: {container_name}")
        print(f"API available at: http://192.168.88.5:{port}")
        print(f"Health check: http://192.168.88.5:{port}/health")
        print(f"Prediction endpoint: http://192.168.88.5:{port}/predict/next-hour")
        print("Pausing for 5 seconds to let the API container fire up...")
        time.sleep(5)

        try:
            print("Testing the API...")
            response = requests.get(f"http://127.0.0.1:{port}/predict/next-hour")
            response.raise_for_status()
            print("API Response:")
            print(response.json())
        except requests.exceptions.RequestException as e:
            print(f"API request failed: {e}")
        
        return f"http://192.168.88.5:{port}"
Step deploy_model_api has started.
[deploy_model_api] Removed existing container: bandwidth-prediction-api
[deploy_model_api] Deployed model API container: bandwidth-prediction-api
[deploy_model_api] API available at: http://192.168.88.5:19000
[deploy_model_api] Health check: http://192.168.88.5:19000/health
[deploy_model_api] Prediction endpoint: http://192.168.88.5:19000/predict/next-hour
[deploy_model_api] Pausing for 5 seconds to let the API container fire up...
[deploy_model_api] Testing the API...
[deploy_model_api] API Response:
[deploy_model_api] {'prediction_timestamp': '2025-06-05T20:00:00', 'predicted_t_rate': 1840713.25, 'model_version': 'xgboost_optuna_v1'}
Step deploy_model_api has finished in 8.333s.

and there you have it. So now I have the means of:

  • pulling historical data from postgres
  • Validating/cleaning the data up
  • Engineering additional features into the dataframe to aid the model
  • Training an XGBoost model with hyperparameter tuning and time-series cross validation
  • Deploying the newly created model in a docker container behind a fastapi python app so it can be used to generate predictions All in 1 script or pipeline as it’s called in Zenml

Next Steps

  • Adding a call to this new endpoint to the already running python script that is scraping this bandwidth data from the cable company’s bandwidth graph utility. Inserting the predicted values into a new table so the predicted values can be plotted against actual. I want to do this with each iteration of the model going forward as well.
  • Improving the model; there’s so much that could be discussed here. I outlined some ideas above
  • Automate running this pipeline nightly(?) Essentially, re-train the model each night and re-deploy the docker container. New table in postgres should have a column to store the model that generated the prediction

Failure is unique among sin, in that we can turn it into a virtue when it drives us to succeed.

The living diminish, but the machine endures...

Wesley Ray · blog · git · resume