Building a Dynamic Pricing Microservice with Vertex AI and MongoDB Atlas

1
git clone https://github.com/mongodb-industry-solutions/retail-store.git

1
cd retail-store/microservices/dynamicPricing

1
pip install -r requirements.txt

1
MONGODB_URI=mongodb+srv://username:password@clusterName.mongodb.net/
2
GOOGLE_APPLICATION_CREDENTIALS=my-google-credentials
3
GOOGLE_CLOUD_PROJECT=my-google-cloud
4
PUBSUB_TOPIC_ID=my-topic-id

1
conn_string = "your_mongodb_connection_string_here"
2
client = MongoClient(conn_string)

1
db = client["your_database_name"]
2
collection = db["your_collection_name"]

1
```python
2
!pip install pymongo
3
!pip install 'pymongo[srv]'
4
!pip install pandas
5
from pymongo import MongoClient
6
import pandas as pd
7
import keras

1
# Replace the below connection string with your MongoDB connection URI
2
conn_string = "your_mongodb_connection_string_here"
3
client = MongoClient(conn_string)

1
# Specify the database and collection
2
db = client["your_database_name"]
3
collection = db["your_collection_name"]

1
# Get all the documents
2
documents = collection.find()
3
4
# Convert the documents into a list and then into a DataFrame
5
df = pd.DataFrame(list(documents))
6
7
# Drop unnecessary columns
8
df = df.drop(columns=['product_name', 'product_id', 'timestamp', 'tensor'])
9
10
# Extracting the 'amount' from the 'price' column and converting it to float
11
df['price'] = df['price'].apply(lambda x: float(x['amount']) if isinstance(x, dict) and 'amount' in x else None)
12
13
df = df.dropna()

1
import tensorflow as tf
2
from sklearn.model_selection import train_test_split
3
from sklearn.preprocessing import LabelEncoder, MinMaxScaler, StandardScaler
4
from tensorflow.keras.models import Sequential
5
from tensorflow.keras.layers import Dense
6
from tensorflow.keras.layers import Dropout

1
# Encode categorical variables
2
label_encoders = {}
3
for column in ['action', 'encoded_name']:
4
    le = LabelEncoder()
5
    df[column] = le.fit_transform(df[column])
6
    label_encoders[column] = le
7
8
df.head()

1
# Standardizing
2
scaler = StandardScaler()
3
df[['action', 'encoded_name']] = scaler.fit_transform(df[['action', 'encoded_name']])
4
5
df.head()

1
!pip install google-cloud-storage
2
from google.cloud import storage
3
import joblib
4
import io

1
# Initialize a client
2
storage_client = storage.Client()
3
4
# The name of your GCP bucket
5
bucket_name = 'dyn_pricing_scaler'
6
7
# The path within your bucket to save the scaler object
8
destination_blob_name = 'labelEncoder.joblib'
9
10
# Create a buffer
11
buffer = io.BytesIO()
12
13
# Dump the scaler object to the buffer
14
joblib.dump(label_encoders, buffer)
15
16
# Now upload the buffer content to GCS
17
bucket = storage_client.bucket(bucket_name)
18
blob = bucket.blob(destination_blob_name)
19
20
# Rewind the buffer's file pointer to the beginning of the file
21
buffer.seek(0)
22
23
# Upload the contents of the buffer
24
blob.upload_from_file(buffer, content_type='application/octet-stream')
25
26
print(f"Uploaded scaler to gs://{bucket_name}/{destination_blob_name}")

1
from tensorflow.keras.models import Sequential
2
from tensorflow.keras.layers import Dense
3
from tensorflow.keras.optimizers import Adam
4
5
# Splitting data into training and testing sets
6
X = df.drop('price', axis=1)
7
y = df['price']
8
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)
9
10
# Define the model
11
model = Sequential([
12
    Dense(64, activation='relu', input_shape=(X_train.shape[1],)),
13
    Dense(64, activation='relu'),
14
    Dense(1)  # Output layer for regression
15
])
16
17
# Compile the model
18
model.compile(optimizer='adam', loss='mean_squared_error', metrics=['mae'])
19
20
# Print the model summary to check the architecture
21
model.summary()
22
23
# Train the model
24
history = model.fit(X_train, y_train,
25
                    validation_split=0.2,  # Further split the training set for validation
26
                    epochs=10,  # Number of epochs to train for
27
                    verbose=1,  # Show training output
28
                    )
29
30
# Evaluate the model on the test set
31
test_loss, test_mae = model.evaluate(X_test, y_test, verbose=1)
32
print(f"Test Loss: {test_loss}, Test MAE: {test_mae}")

1
import numpy as np
2
3
# Example new data point converted into a tensor
4
new_data = np.array([[1.225999, -0.957973]])
5
6
predicted_price = model.predict(new_data)
7
print("Predicted Price:", predicted_price[0])

1
from google.colab import auth
2
auth.authenticate_user()
3
4
project_id = 'your-gcp-project-id'
5
!gcloud config set project {project_id}
6
7
model_dir = 'your-model-directory'
8
9
# Save the model to GCS
10
model_dir = f'gs://{bucket_name}/{model_dir}
11
model.save(model_dir)
12
13
bucket_name = 'your-cloud-storage-bucket-name'
14
model_dir = 'your-model-directory'
15
model_path = 'your-model-path'
16
!gsutil mb -l us-central1 gs://{bucket_name}  # Create the bucket if necessary
17
!gsutil cp -r {model_dir} gs://{bucket_name}/{model_path}

1
from google.cloud import aiplatform
2
3
aiplatform.init(project='your-gcp-project-id', location='your-gcp-region')
4
5
#Model registry
6
7
model_display_name = 'dyn_pricingv1'
8
model_description = 'TensorFlow dynamic pricing model'
9
bucket_name = 'your-gcp-bucket-name'
10
model_path = 'your-model-path'
11
12
model = aiplatform.Model.upload(
13
    display_name=model_display_name,
14
    artifact_uri=f'gs://{bucket_name}/{model_path}',
15
    serving_container_image_uri='us-docker.pkg.dev/vertex-ai/prediction/tf2-cpu.2-6:latest',
16
    description=model_description,
17
)

1
from pymongo import MongoClient
2
from datetime import datetime, timedelta
3
import json
4
import numpy as np
5
import pandas as pd
6
import joblib
7
import os
8
from google.cloud import storage
9
import functions_framework
10
import base64
11
from bson import ObjectId
12
from google.cloud import aiplatform
13
14
@functions_framework.cloud_event
15
def hello_pubsub(cloud_event):
16
  event_data = base64.b64decode(cloud_event.data["message"]["data"])
17
  print(event_data)
18
  event_data = json.loads(event_data)
19
  print(event_data)
20
  event_data = pd.DataFrame([event_data])
21
  print(event_data)
22
23
# Correctly handle the '_id' field when it's a string, getting the event_id and product_id data
24
  event_id = ObjectId(str(event_data['_id'].loc[0]))
25
  print(event_id)
26
  product_id = event_data['product_id'].loc[0]
27
  print(product_id)
28
29
  # Set your GCS bucket and file path
30
  bucket_name = 'dyn_pricing_scaler'
31
  scaler_file_name = 'scaler.joblib'
32
  label_encoder_name = 'labelEncoder.joblib'
33
34
  # Initialize a GCS client
35
  storage_client = storage.Client()
36
37
  # Get the bucket
38
  bucket = storage_client.bucket(bucket_name)
39
40
  # Get the blob (file) containing the scaler
41
  blob1 = bucket.blob(scaler_file_name)
42
43
  # Download the scaler file to a temporary location
44
  scaler_temp_file_path = '/tmp/' + scaler_file_name
45
  blob1.download_to_filename(scaler_temp_file_path)
46
47
  # Download the label encoder to a temporary location
48
  blob2 = bucket.blob(label_encoder_name)
49
  label_encoder_temp = '/tmp/' + label_encoder_name
50
  blob2.download_to_filename(label_encoder_temp)
51
52
  # Clean unnecessary columns
53
  if '_id' in event_data.columns:
54
    event_data = event_data.drop(columns=['_id'])
55
  print(event_data)
56
  event_data.drop('price', axis=1, inplace=True)
57
  event_data.drop('timestamp', axis=1, inplace=True)
58
  print(event_data)
59
  event_data.drop('product_id', axis=1, inplace=True)
60
  print(event_data)
61
  event_data.drop('product_name', axis=1, inplace=True)
62
  print(event_data)
63
64
  #Load label encoder
65
  label_encoders = joblib.load(label_encoder_temp)
66
  print(label_encoders)
67
68
  #Encode categorical fields
69
  for column, encoder in label_encoders.items():
70
    event_data[column] = encoder.transform(event_data[column])
71
  print(event_data)
72
73
74
  # Load the scaler using joblib
75
  scaler = joblib.load(scaler_temp_file_path)
76
  print(scaler)
77
78
79
  # Use scaler
80
  columns_to_scale = ['action', 'encoded_name']
81
  event_data[columns_to_scale] = scaler.transform(event_data[columns_to_scale])
82
  print(event_data)
83
84
  #Prepare data as tensors
85
  #first_row_scaled = event_data[0, :].reshape(1, -1)
86
  #print(first_row_scaled)
87
  event_data = event_data.to_numpy()
88
89
  # Prepare input data for VertexAI
90
  input_data = {"instances": event_data}
91
  print(input_data)
92
  input_data['instances'] = input_data['instances'].tolist()
93
  print(input_data)
94
  input_data_json = json.dumps(input_data)
95
  print(input_data_json)
96
97
  # Call VertexAI endpoint
98
  endpoint_id = "your-endpoint-id"
99
  project_id = "your-project-id"
100
  location = "your-project-location"
101
  endpoint_url = f"https://us-central1-aiplatform.googleapis.com/v1/projects/{project_id}/locations/us-central1/endpoints/{endpoint_id}:predict"
102
103
  aiplatform.init(project=project_id, location=location)
104
105
  endpoint = aiplatform.Endpoint(endpoint_id)
106
107
  prediction = endpoint.predict(instances=input_data['instances'])
108
109
  print(prediction)
110
  
111
  pred_price = prediction.predictions[0][0]
112
113
  print(pred_price)
114
  
115
  #Start MongoClient
116
  mongo_uri = "your-mongo-db-connection-uri" #remember you can set it up in a .env file
117
  client = MongoClient(mongo_uri)
118
119
  db = client["dotLocalStore"]  # Replace "your_database" with your actual database name
120
  collection = db["products"]
121
  feature_store = db["events"]
122
123
  # Update products collection document with 'pred_price'
124
  collection.update_one({"id": int(product_id)}, {"$set": {"pred_price": pred_price}})
125
126
  # Update events collection (feature_store) with event tensor
127
  tensor = event_data.tolist()
128
  print(tensor)
129
  feature_store.update_one({"_id": event_id}, {"$set": {"tensor": tensor }})
130
131
  # Clean up: Delete the temporary files
132
  os.remove(scaler_temp_file_path)
133
  os.remove(label_encoder_temp)
134
  print(pred_price)
135
136
  return pred_price

1
functions-framework==3.*
2
pymongo
3
scikit-learn==1.2.2
4
numpy
5
google-cloud-aiplatform
6
google-auth
7
google-cloud-storage
8
pandas
9
joblib

1
import random
2
import pymongo
3
from pymongo import MongoClient
4
import os
5
import time
6
from google.cloud import pubsub_v1
7
from dotenv import load_dotenv
8
from faker import Faker
9
from datetime import datetime, timedelta
10
import json
11
from bson import ObjectId
12
import hashlib
13
14
# Load environment variables
15
load_dotenv()
16
MONGODB_URI = os.getenv('MONGODB_URI')
17
GOOGLE_APPLICATION_CREDENTIALS = os.getenv('GOOGLE_APPLICATION_CREDENTIALS')
18
GOOGLE_CLOUD_PROJECT = os.getenv('GOOGLE_CLOUD_PROJECT')
19
PUBSUB_TOPIC_ID = os.getenv('PUBSUB_TOPIC_ID')
20
21
# Initialize MongoDB client
22
mongo_client = pymongo.MongoClient(MONGODB_URI)
23
db = mongo_client["dotLocalStore"]
24
behaviors_collection = db["events"]
25
products_collection = db["products"]
26
27
# Initialize Google Cloud Pub/Sub publisher
28
publisher = pubsub_v1.PublisherClient()
29
topic_path = publisher.topic_path(GOOGLE_CLOUD_PROJECT, PUBSUB_TOPIC_ID)
30
31
def fetch_products():
32
    """Fetch products from MongoDB"""
33
    return list(products_collection.find({}))
34
35
def generate_ecommerce_behavior(product):
36
    """Generate a single synthetic ecommerce behavior data for a given product"""
37
    # Encode product_name into a numerical field
38
    encoded_name = int(hashlib.sha256(product["name"].encode('utf-8')).hexdigest(), 16) % 10**8
39
    behavior = {
40
        "product_name": product["name"],
41
        "product_id": product["id"],
42
        "action": random.choice(["view", "add_to_cart", "purchase"]),
43
        "price": product["price"],
44
        "timestamp": datetime.now().isoformat(),  # Format timestamp for JSON serialization
45
        "encoded_name": encoded_name
46
    }
47
    return behavior
48
49
# Custom JSON Encoder that converts ObjectId to str
50
class JSONEncoder(json.JSONEncoder):
51
    def default(self, o):
52
        if isinstance(o, ObjectId):
53
            return str(o)
54
        return json.JSONEncoder.default(self, o)
55
56
def push_event_to_mongodb(behavior):
57
    """Push a single ecommerce behavior data to MongoDB"""
58
    behaviors_collection.insert_one(behavior)
59
    print("Pushed an event to MongoDB.")
60
61
def push_event_to_pubsub(event):
62
    """Push a single event to Google Cloud Pub/Sub"""
63
    try:
64
        # Attempt to serialize the event to JSON
65
        data = json.dumps(event, cls=JSONEncoder).encode("utf-8")
66
        # Attempt to publish the serialized data to Pub/Sub
67
        future = publisher.publish(topic_path, data=data)
68
        print(f"Published id:{event['product_id']} product:{event['product_name']} to Pub/Sub.")
69
        future.result()  # Block until the publish completes
70
    except Exception as e:
71
        print(f"An error occurred: {e}")
72
73
if __name__ == "__main__":
74
    products = fetch_products()
75
    num_behaviors_per_cycle = 150
76
77
    try:
78
        while True:
79
            for _ in range(num_behaviors_per_cycle):
80
                # Select a random product
81
                selected_product = random.choice(products)
82
                # Generate behavior for the selected product
83
                behavior = generate_ecommerce_behavior(selected_product)
84
                # Push the behavior to MongoDB
85
                push_event_to_mongodb(behavior)
86
                # Push the behavior to Pub/Sub
87
                push_event_to_pubsub(behavior)
88
                # Wait for 3 seconds before generating the next behavior
89
            time.sleep(3)
90
    except KeyboardInterrupt:
91
        print("Stopped by the user.")

1
python3 generator.py