Real-Time Product Classification

Abstract

Products arrive as a stream, so the classifier is trained with partial_fit and scored before each batch is used for training — prequential evaluation, the only honest way to score a model that learns as it goes. Halfway through, a fourth category appears. Accuracy holds at 0.9990, drops to 0.9600 when the catalogue shifts, and recovers to within 0.0010 of where it started.

Prerequisites

Python 3.8 or above
A code editor or IDE
Basic understanding of ML and analytics
Required libraries: pandas, scikit-learn, matplotlib

Before you Start

Install Python and the required libraries:

Install dependencies

pip install pandas scikit-learn matplotlib

Getting Started

Create a Project

Create a folder named real-time-product-classification.
Open the folder in your code editor or IDE.
Create a file named real_time_product_classification.py.
Copy the code below into your file.

Write the Code

Real-Time Product Classification

pch.viewSource

Real-Time Product Classification

"""Real-time product classification.

Products arrive as a stream, so the classifier is trained with `partial_fit`
and scored **before** each batch is used for training -- prequential
evaluation, which is the only honest way to score a model that learns as it
goes. Halfway through, the catalogue shifts and a new category appears, so the
accuracy curve shows both learning and forgetting.
"""

import matplotlib.pyplot as plt
import numpy as np
from sklearn.linear_model import SGDClassifier

FEATURES = 12
CATEGORIES = ("tools", "clothing", "grocery", "electronics")


def product_stream(batches=60, batch=40, seed=0, drift_at=30):
    """Feature vectors per category, with a fourth category appearing later."""
    rng = np.random.default_rng(seed)
    centres = rng.normal(0, 2.0, (len(CATEGORIES), FEATURES))
    for index in range(batches):
        active = 3 if index < drift_at else 4
        labels = rng.integers(0, active, batch)
        rows = centres[labels] + rng.normal(0, 1.0, (batch, FEATURES))
        yield index, rows.astype("float32"), labels


class StreamingClassifier:
    def __init__(self):
        self.model = SGDClassifier(loss="log_loss", random_state=0)
        self.started = False

    def predict(self, rows):
        if not self.started:
            return np.zeros(len(rows), dtype=int)
        return self.model.predict(rows)

    def learn(self, rows, labels):
        self.model.partial_fit(rows, labels,
                               classes=np.arange(len(CATEGORIES)))
        self.started = True


def main():
    classifier = StreamingClassifier()
    accuracies, seen, drift_at = [], 0, 30

    for index, rows, labels in product_stream(drift_at=drift_at):
        # Score before training on this batch: the model has not seen it yet.
        predicted = classifier.predict(rows)
        accuracies.append(float((predicted == labels).mean()))
        classifier.learn(rows, labels)
        seen += len(rows)

    accuracies = np.asarray(accuracies)
    warm = accuracies[5:drift_at]
    after = accuracies[drift_at:drift_at + 5]
    recovered = accuracies[-5:]

    print("Real-Time Product Classification")
    print(f"  products seen              : {seen:,}")
    print(f"  batches                    : {len(accuracies)}")
    print(f"  prequential accuracy, warm : {warm.mean():.4f}")
    print(f"  first 5 batches after drift: {after.mean():.4f}")
    print(f"  final 5 batches            : {recovered.mean():.4f}")
    print(f"  drop when a new category appeared: "
          f"{warm.mean() - after.mean():.4f}")
    print(f"  recovered to within "
          f"{abs(warm.mean() - recovered.mean()):.4f} of the pre-drift rate")
    print("\n  every score above was taken BEFORE the batch was trained on,")
    print("  so nothing here is measured on data the model had already seen")

    plt.figure(figsize=(9, 3.4))
    plt.plot(accuracies, linewidth=1.3, label="prequential accuracy")
    plt.axvline(drift_at, linestyle="--", linewidth=1.2,
                label="4th category appears")
    plt.axhline(warm.mean(), linestyle=":", linewidth=1.0,
                label=f"pre-drift mean {warm.mean():.3f}")
    plt.xlabel("batch")
    plt.ylabel("accuracy on unseen batch")
    plt.title("learning, then forgetting, then relearning")
    plt.legend(fontsize=8)
    plt.savefig("real_time_product_classification.png", dpi=120,
                bbox_inches="tight")
    print("saved real_time_product_classification.png")


if __name__ == "__main__":
    main()

Example Usage

Run product classification

python real_time_product_classification.py

What it produces

Running the file exactly as it ships takes 4.0 s and prints:

python real_time_product_classification.py

Real-Time Product Classification
  products seen              : 2,400
  batches                    : 60
  prequential accuracy, warm : 0.9990
  first 5 batches after drift: 0.9600
  final 5 batches            : 1.0000
  drop when a new category appeared: 0.0390
  recovered to within 0.0010 of the pre-drift rate
 
  every score above was taken BEFORE the batch was trained on,
  so nothing here is measured on data the model had already seen
saved real_time_product_classification.png

Output of real_time_product_classification.py, produced by running the file. — Written by the run above. If the project stops producing it, the page's figure asset goes missing and check_docs reports it — which is the point of generating it rather than drawing it.

How it fits together

Read from the top: this is what runs when you execute the file, and which function calls which. It is generated from the code, so it cannot drift from it.

diagram Diagram mermaid

  flowchart TD
  RUN(["python real_time_product_classification.py"])
  product_stream("product_stream")
  StreamingClassifier["StreamingClassifier
class"]
  main("main")
  RUN --> main
  main --> StreamingClassifier
  main --> product_stream

Explanation

Key Features

Prequential scoring: test-then-train, so no score is ever taken on data the model has already seen.
Incremental learning: SGDClassifier.partial_fit with the class list declared up front, so a category that has not appeared yet still has a slot.
Concept drift, on purpose: a fourth category at batch 30, and the measured 0.0390 accuracy drop it causes.
Recovery, measured: the curve comes back, which a single final accuracy would have hidden entirely.

Code Breakdown

What it imports (lines 10–12)

real_time_product_classification.py

import matplotlib.pyplot as plt
import numpy as np
from sklearn.linear_model import SGDClassifier

product_stream — the function (lines 18–26)

real_time_product_classification.py

def product_stream(batches=60, batch=40, seed=0, drift_at=30):
    """Feature vectors per category, with a fourth category appearing later."""
    rng = np.random.default_rng(seed)
    centres = rng.normal(0, 2.0, (len(CATEGORIES), FEATURES))
    for index in range(batches):
        active = 3 if index < drift_at else 4
        labels = rng.integers(0, active, batch)
        rows = centres[labels] + rng.normal(0, 1.0, (batch, FEATURES))
        yield index, rows.astype("float32"), labels

StreamingClassifier — the class (lines 29–42)

real_time_product_classification.py

class StreamingClassifier:
    def __init__(self):
        self.model = SGDClassifier(loss="log_loss", random_state=0)
        self.started = False
 
    def predict(self, rows):
        if not self.started:
            return np.zeros(len(rows), dtype=int)
        return self.model.predict(rows)
 
    def learn(self, rows, labels):
        self.model.partial_fit(rows, labels,
                               classes=np.arange(len(CATEGORIES)))
        self.started = True

main — the function (lines 45–86)

real_time_product_classification.py

def main():
    classifier = StreamingClassifier()
    accuracies, seen, drift_at = [], 0, 30
 
    for index, rows, labels in product_stream(drift_at=drift_at):
        # Score before training on this batch: the model has not seen it yet.
        predicted = classifier.predict(rows)
        accuracies.append(float((predicted == labels).mean()))
        classifier.learn(rows, labels)
        seen += len(rows)
 
    accuracies = np.asarray(accuracies)
    warm = accuracies[5:drift_at]
    after = accuracies[drift_at:drift_at + 5]
    recovered = accuracies[-5:]
 
    print("Real-Time Product Classification")
    print(f"  products seen              : {seen:,}")
    # ... 18 more lines in the file ...
    plt.ylabel("accuracy on unseen batch")
    plt.title("learning, then forgetting, then relearning")
    plt.legend(fontsize=8)
    plt.savefig("real_time_product_classification.png", dpi=120,
                bbox_inches="tight")
    print("saved real_time_product_classification.png")

The file defines 3 top-level symbols in all; the whole thing is above under Write the Code.

Features

Product Classification: Real-time data preprocessing and classification
Modular Design: Separate functions for each task
Error Handling: Manages invalid inputs and exceptions
Production-Ready: Scalable and maintainable code

Next Steps

Enhance the project by:

Integrating with more product APIs
Supporting advanced ML models
Creating a GUI for classification
Adding real-time analytics
Unit testing for reliability

Educational Value

This project teaches:

Streaming evaluation: why test-then-train is the right protocol for an online model.
Concept drift: what it looks like in a metric, and how long recovery takes.
Incremental learners: partial_fit, and why classes must be declared in advance.

Real-World Applications

E-commerce Platforms
Analytics Tools
Classification Engines

Conclusion

Real-Time Product Classification demonstrates how to build a scalable and accurate product classification tool using Python. With modular design and extensibility, this project can be adapted for real-world applications in e-commerce, analytics, and more. For more advanced projects, visit Python Central Hub.

Pitfalls

Scoring after training on the same batch measures memory, not prediction. Measured in the exercise below: train-then-test reports 1.0000 from the first batch onwards; test-then-train reports 0.9667 and has to earn every point on unseen data.
A single final accuracy hides everything that happened on the way. When a new category appears mid-stream the prequential score drops 0.0750 and recovers; a score computed only at the end reports the recovered number and shows none of it. The project measures the same effect: a 0.0390 drop, recovering to within 0.0010 of the pre-drift rate.
Recovery is not the same as detection. The model recovered because it kept learning, not because anything noticed. The dip is the alert, and something has to be watching for it.
The warm-up score is not a failure. The first batch scores 0 because the model has seen nothing. Averaging it into a headline number understates a system that was never going to know the answer.
Prequential scores are not comparable across different batch sizes. Smaller batches mean more frequent learning, so the same model looks better; the batch size belongs next to the score.

Recap

Measured: 2,400 products, 60 batches, prequential accuracy once warm 0.9990.
After a new category appears: first 5 batches 0.9600, final 5 batches 1.0000, drop 0.0390, recovered to within 0.0010.
Every score was taken before the batch was trained on, so nothing is measured on data the model had already seen.
Test-then-train is the standard evaluation for streaming models precisely because there is no held-out set in a stream — each example is test data once and training data afterwards.

pch.quizTag pch.quizDefaultTitle

Why does the project score each batch before training on it rather than after?

pch.quizShowAnswer

B — Scoring after training measures whether the model remembers examples it was just shown, which is not what a deployed model does — it always predicts before the label arrives
A new product category appears and the accuracy drops 0.0390 before recovering. What is the drop worth?

pch.quizShowAnswer

B — It is the only visible signal that the world changed — a final accuracy reports the recovered value and hides the event entirely
Two streaming models report prequential accuracy, one with batches of 10 and one with batches of 200. Are the numbers comparable?

pch.quizShowAnswer

B — No — smaller batches mean the model learns more often before being tested again, so batch size has to be reported alongside the score

Try it yourself

pch.coffeeTagline

pch.coffeeCta

pch.feedbackHeading

pch.feedbackSubheading