Technical Cookbook

from dataclasses import dataclass

@dataclass
class Candidate:
    sku: str
    score_ai: float
    stock: int
    margin_pct: float
    forbidden: bool = False

def apply_guardrails(
    candidates: list[Candidate],
    min_margin_pct: float,
    min_score_ai: float,
    max_items: int = 10,
) -> list[Candidate]:
    # 1. filtering forbidden products, without stock or under minimum margin
    filtered = [
        c for c in candidates
        if not c.forbidden
        and c.stock > 0
        and c.margin_pct >= min_margin_pct
        and c.score_ai >= min_score_ai
    ]

    # 2. if the list is too short after filtering, fill with internal fallback (get_top_selling_products)
    if len(filtered) < max_items:
        missing = max_items - len(filtered)
        exclude_skus = {c.sku for c in filtered}
        filtered.extend(get_top_selling_products(exclude_skus=exclude_skus, limit=missing))
    # 3. sort by AI score and limit
    filtered.sort(key=lambda c: c.score_ai, reverse=True)
    return filtered[:max_items]
    
    

SELECT
  r.account_id
  , r.scenario
  , r.sku_recom
  , r.score_ai
  , p.margin_pct
  , r.created_at
FROM recommendations_log r
JOIN products p ON r.sku_recom = p.sku
WHERE 
  p.margin_pct <= 0.15 -- or a parametrizable threshold
  AND r.created_at > (NOW() - INTERVAL '1 DAY')
ORDER BY r.created_at DESC
LIMIT 100;
    
    

from pydantic import BaseModel, Field
from datetime import datetime
from typing import Optional

class InventoryUpdateEvent(BaseModel):
    sku: str
    new_stock: int = Field(ge=0)
    source: Optional[str] = None     # e.g. "WMS", "ERP", "Manual"
    updated_at: datetime
    idempotency_key: str = Field(min_length=8)

    # Here you can add Feature logic for which a pipeline should be defined
    
    

INSERT INTO inventory (sku, stock, updated_at)
VALUES (:sku, :stock, :updated_at)
ON CONFLICT (sku)
DO UPDATE SET
  stock = EXCLUDED.stock
  , updated_at = EXCLUDED.updated_at
WHERE 
    inventory.updated_at IS NULL 
    OR inventory.updated_at < EXCLUDED.updated_at;
    
    

SELECT
  i.sku
  , i.stock AS stock_official
  , c.stock_cache AS stock_recommender
FROM inventory i
JOIN recommender_cache c ON i.sku = c.sku
WHERE i.stock <> c.stock_cache;
    
    

from enum import Enum

class Scenario(Enum):
    SIMILAR_ITEMS = "similar_items"
    CART_ADDONS = "cart_addons"
    SUBSTITUTES = "substitutes"

# Registry mapping scenario -> model
MODEL_REGISTRY = {
    Scenario.SIMILAR_ITEMS:  "models/similar_items_v3",
    Scenario.CART_ADDONS:    "models/cart_addons_v5",
    Scenario.SUBSTITUTES:    "models/substitutes_v2",
}

# If you are using a Feature Store, you can have a separate registry for scenario service versions
FS_SERVICE_REGISTRY = {
    Scenario.SIMILAR_ITEMS:  "similar_items_service_v1",
    Scenario.CART_ADDONS:    "cart_addons_service_v2",
    Scenario.SUBSTITUTES:    "substitutes_service_v1",
}

def recommend_for_scenario(
    scenario: Scenario,
    account_id: str,
    context: dict,
) -> list[str]:
    model_path = MODEL_REGISTRY[scenario]
    
    full_context = context.copy()
    # Interaction with Feature Store (conceptual): first get details from the store, then hydrate the context
    # fs_service = FS_SERVICE_REGISTRY[scenario]
    # account_features = call_fs_to_get_features(account_id=account_id, service=fs_service)
    # full_context.update(account_features)

    # Call to a local model or a cloud endpoint (with all necessary data)
    scores = call_model(model_path, account_id=account_id, inputs=full_context)
    

    # scores is a list of (sku, ai_score), e.g., [("SKU1", 0.92), ("SKU2", 0.87)...]
    ordered = sorted(scores, key=lambda x: x[1], reverse=True)
    return [sku for sku, _ in ordered]
    
    

scenarios:
  similar_items:
    model: "models/similar_items_v3"
    max_items: 10

  cart_addons:
    model: "models/cart_addons_v5"
    max_items: 6

  substitutes:
    model: "models/substitutes_v2"
    max_items: 5
    
    

from pyspark.ml.fpm import FPGrowth
from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()

# 1. Data: each row is a transaction with an array of SKUs
# schema: [order_id: string, items: array]
df_orders = spark.table("sales_history_clean")

# 2. Configure the FPGrowth algorithm
# minSupport: products appear together in X% of transactions
# minConfidence: if a customer buys A, there is X% chance they will also buy B
fp_growth = FPGrowth(itemsCol="items", minSupport=0.001, minConfidence=0.2)

# 3. Train the distributed model
model = fp_growth.fit(df_orders)

# 4. Extract association rules (antecedent -> consequent)
# schema: [antecedent: [SKU_A], consequent: [SKU_B], confidence: 0.85, lift: 2.4]
association_rules = model.associationRules

# Save to a temporary table for further processing
association_rules.createOrReplaceTempView("raw_rules")
    
    

WITH rule_financials AS (
    SELECT 
        r.antecedent
        , r.consequent
        , r.confidence
        -- COALESCE to avoid NULL
        , SUM(COALESCE(p_ant.margin_eur, 0)) as antecedent_total_margin        
        -- Check if all products in the antecedent are found
        , COUNT(p_ant.sku) as found_ant_products
        , SIZE(r.antecedent) as expected_ant_products
        , MAX(COALESCE(p_cons.margin_eur, 0)) as consequent_margin
        , MAX(p_cons.days_in_inventory) as cons_days_inv
    FROM raw_rules r
    LATERAL VIEW explode(r.antecedent) a AS ant_sku
    -- JOIN for antecedent
    JOIN products p_ant ON a.ant_sku = p_ant.sku  
    -- JOIN for consequent with direct stock filter > 10 
    JOIN products p_cons ON r.consequent[0] = p_cons.sku AND p_cons.stock_quantity > 10
    GROUP BY 
        r.antecedent
        , r.consequent
        , r.confidence
)

SELECT
    antecedent as current_basket
    , consequent[0] as recommended_upsell
    , confidence
    , (antecedent_total_margin + consequent_margin) as total_bundle_margin
    -- Score calculation
    , (confidence 
        * (antecedent_total_margin + consequent_margin) 
        * (CASE WHEN cons_days_inv > 120 THEN 1.5 ELSE 1.0 END)
        ) as priority_score
FROM rule_financials
WHERE 
    -- in the JOIN, no products from the antecedent were lost
    found_ant_products = expected_ant_products
    -- eliminate negative or zero margins (recommended)
    AND (antecedent_total_margin + consequent_margin) > 0
ORDER BY 
    priority_score DESC;
    
    

def suggest_bundles_for_quote(
    current_skus: list[str],
    bundles: dict[tuple[str, str], int],
    max_suggestions: int = 5,
) -> list[tuple[str, str]]:
    suggestions: list[tuple[str, str]] = []
    seen: set[tuple[str, str]] = set()
    current = set(current_skus)

    for sku in current_skus:
        for (a, b), _count in bundles.items():
            pair = None

            if sku == a and b not in current:
                pair = (a, b)
            elif sku == b and a not in current:
                pair = (b, a)

            if pair and pair not in seen:
                seen.add(pair)
                suggestions.append(pair)

                if len(suggestions) >= max_suggestions:
                    break

        if len(suggestions) >= max_suggestions:
            break

    return suggestions
    
    

from google.cloud import retail_v2

project_id = "PROJECT_ID"
catalog_id = "default_catalog"
serving_config_id = "cart_upsell"
# fill in
account_id = "HASH_ID_ERP_CUSTOMER"
user_id = "HASH_ID_ERP_CUSTOMER_REP"
device_id = "HASH_ID_UNIQUE_DEVICE"

serving_config = (
    f"projects/{project_id}/locations/global/catalogs/{catalog_id}"
    f"/servingConfigs/{serving_config_id}"
)

client = retail_v2.PredictionServiceClient()

user_event = retail_v2.UserEvent(
    event_type="shopping-cart-page-view",
    visitor_id=device_id,
    user_info=retail_v2.UserInfo(user_id=user_id),
    product_details=[
        # SKU-urile from the cart, including quantity
        retail_v2.ProductDetail(product=retail_v2.Product(id="SKU-SCREW-36"), quantity=1),
        retail_v2.ProductDetail(product=retail_v2.Product(id="SKU-PIN-35"), quantity=2)
    ]
)
# Custom attributes are essential in B2B
user_event.attributes["account_id"] = retail_v2.CustomAttribute(text=[account_id])

request = retail_v2.PredictRequest(
    placement=serving_config,
    user_event=user_event,
    page_size=10, # maximum number of recommendations
    filter='(availability: ANY("IN_STOCK")) AND NOT (categories: ANY("Hobby-DIY"))', # business rules
    params={
        "filterSyntaxV2": True, # allows advanced filters
        "strictFiltering": True, # respects written filters, no fallback
        "diversityLevel": "medium-diversity",
        "returnProduct": True, # returns product metadata
        "returnScore": True, # returns AI score
        "priceRerankLevel": "low-price-reranking", # slightly prioritizes higher-priced products (not profit)
    },
)

response = client.predict(request=request)
    
    

from typing import Any, Optional
from google.cloud import retail_v2


def recommended_for_placement(
    user_event: retail_v2.UserEvent,  # see Example M1
    serving_config_id: str,  # Ex: "homepage_recs", "cart_upsell"
    project_id: str,
    catalog_id: str,         # Ex: "default_catalog"
    page_size: int = 10,
    filter_expr: Optional[str] = None,
    params: Optional[dict[str, Any]] = None = None,
) -> list[str]:
    client = retail_v2.PredictionServiceClient()

    serving_config = (
        f"projects/{project_id}/locations/global/catalogs/{catalog_id}/"
        f"servingConfigs/{serving_config_id}"
    )
    # we build the base request
    request = retail_v2.PredictRequest(
        placement=serving_config,
        user_event=user_event,
        page_size=page_size,  # maximum number of recommendations
        params={
            "filterSyntaxV2": True,
            "returnProduct": True,
            "returnScore": True,
        },
    )

    if filter_expr:
        request.filter = filter_expr

    if params:
        request.params.update(params)

    response = client.predict(request=request)
    
    # Recommended product IDs
    return [result.id for result in response.results]
    
    

from typing import Any
from google.cloud import retail_v2
from google.protobuf.json_format import ParseDict

def write_event_from_json(
    user_event_json: dict[str, Any], # See Example M3
    project_id: str,
    catalog_id: str, # Ex: "default_catalog"    
) -> retail_v2.UserEvent:
    
    client = retail_v2.UserEventServiceClient()

    parent = f"projects/{project_id}/locations/global/catalogs/{catalog_id}"

    # ParseDict takes camelCase keys and maps them to proto fields
    user_event = ParseDict(user_event_json, retail_v2.UserEvent())

    request = retail_v2.WriteUserEventRequest(
        parent=parent,
        user_event=user_event,
    )

    return client.write_user_event(request=request)
    
    

{
  "id": "SKU-SCREW-36",
  "title": "Screw RTE 6x60 mm",
  "availability": "IN_STOCK", 
  "priceInfo": {
    "price": 45.90,
    "currencyCode": "USD"
  },
  "fulfillmentInfo": [
    {
      "type": "pickup-in-store",
      "placeIds": ["store_london_1"]
    }
  ]
}

    
    

from google.cloud import retail_v2
from collections.abc import Iterable

def recommend_bundles_for_cart_item(
    item_sku: str,
    cart_skus: Iterable[str],
    device_id: str, #HASH_ID_UNIQUE_DEVICE
    account_id: str, #HASH_ID_ERP_CUSTOMER (company as user)
    project_id: str = "PROJECT_ID",
    catalog_id: str = "default_catalog",
    serving_config_id: str = "bundle_recs",
    page_size: int = 6 
) -> list[str]:
   
    
    # 1. Create user-event, using the variant company as user
    user_event = retail_v2.UserEvent(
        event_type="shopping-cart-page-view",
        visitor_id=device_id,
        user_info=retail_v2.UserInfo(user_id=account_id),
        product_details=[
            retail_v2.ProductDetail(product=retail_v2.Product(id=str(sku)), quantity=1)
            for sku in cart_skus
        ],
    )

    # 2. ANY("...") for exact match, attributes are of type text and set as filterable
    filter_expr = (
        f'(availability: ANY("IN_STOCK")) AND '
        f'(attributes.is_bundle: ANY("true")) AND '
        f'(attributes.bundle_component_skus: ANY("{item_sku}"))'
    )
    
    # 3. See Example M2
    ids = recommended_for_placement(
        user_event=user_event,    
        serving_config_id=serving_config_id,
        project_id=project_id,
        catalog_id=catalog_id,
        page_size=page_size,
        filter_expr=filter_expr
    )

    return ids