Cookbook tehnic

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. filtram produsele interzise, fara stoc sau sub marja minima
    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. daca dupa filtrare lista este prea scurta, completam cu fallback intern (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. sortam dupa scorul AI și limitam
    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 -- sau prag parametrizabil
  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     # ex: "WMS", "ERP", "Manual"
    updated_at: datetime
    idempotency_key: str = Field(min_length=8)

    #aici se poate adauga logica de Feature (atribut) pentru care apoi se defineste pipeline-ul
    
    

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 de mapare scenariu -> model
MODEL_REGISTRY = {
    Scenario.SIMILAR_ITEMS:  "models/similar_items_v3",
    Scenario.CART_ADDONS:    "models/cart_addons_v5",
    Scenario.SUBSTITUTES:    "models/substitutes_v2",
}

# Daca folosesti un Feature Store, poti avea un registry separat pentru versiunile serviciilor scenariilor
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()
    # interactiune cu Feature store (conceptual): mai intai obtii detaliile din store, apoi hidratezi contextul
    # 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)

    # Apel catre un model local sau un endpoint din cloud (cu toate datele necesare)
    scores = call_model(model_path, account_id=account_id, inputs=full_context)
    

    # scores e lista de (sku, score_ai), ex: [("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. Datele: fiecare rand este o tranzactie cu un array de SKU
# schema: [order_id: string, items: array<string>]
df_orders = spark.table("sales_history_clean")

# 2. Configuram algoritmul FPGrowth
# minSupport: produsele apar impreuna in X% din tranzactii
# minConfidence: cine ia A, are X% sanse sa ia si B
fp_growth = FPGrowth(itemsCol="items", minSupport=0.001, minConfidence=0.2)

# 3. Antrenam modelul distribuit
model = fp_growth.fit(df_orders)

# 4. Extragem regulile de asociere (antecedent -> consequent)
# schema: [antecedent: [SKU_A], consequent: [SKU_B], confidence: 0.85, lift: 2.4]
association_rules = model.associationRules

# Salvam intr-un tabel temporar pentru procesare ulterioara
association_rules.createOrReplaceTempView("raw_rules")
    
    

WITH rule_financials AS (
    SELECT 
        r.antecedent
        , r.consequent
        , r.confidence
        -- COALESCE pentru a evita NULL
        , SUM(COALESCE(p_ant.margin_eur, 0)) as antecedent_total_margin        
        -- Verificam daca am gasit toate produsele din antecedent
        , 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 pentru antecedent
    JOIN products p_ant ON a.ant_sku = p_ant.sku  
    -- JOIN pentru consequent cu filtrare directa de stoc > 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
    -- Calcul scor
    , (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 
    -- la JOIN nu s-au pierdut produse din antecedent 
    found_ant_products = expected_ant_products
    -- eliminam marjele negative sau zero (recomandat)
    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"
# completati
account_id = "HASH_ID_ERP_CLIENT"
user_id = "HASH_ID_ERP_REPREZENTANT_CLIENT"
device_id = "HASH_ID_DEVICE_UNIC"

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 din cos cu cantitatile
        retail_v2.ProductDetail(product=retail_v2.Product(id="SKU-HOLSURUB-36"), quantity=1),
        retail_v2.ProductDetail(product=retail_v2.Product(id="SKU-SURUB-35"), quantity=2)
    ]
)
# Custom attributes sunt esentiale 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, #nr maxim recomandari
    filter='(availability: ANY("IN_STOCK")) AND NOT (categories: ANY("Hobby-DIY"))', #reguli de business
    params={
        "filterSyntaxV2": True, #permite filtre avansate
        "strictFiltering": True, # respecta filtrele scrise, fara fallback
        "diversityLevel": "medium-diversity",
        "returnProduct": True, #returneaza metadatele produselor
        "returnScore": True, #returneaza scorul AI,
        "priceRerankLevel": "low-price-reranking", # prioritizeaza usor produsele cu pret mare (nu 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,  # vezi Exemplul 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}"
    )
    # construim request-ul de baza
    request = retail_v2.PredictRequest(
        placement=serving_config,
        user_event=user_event,
        page_size=page_size,  # nr maxim recomandari
        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)
    
    # ID-uri produse recomandate
    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], # Vezi Exemplul 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 preia chei camelCase si le mapează pe campuri proto
    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-HOLSURUB-36",
  "title": "Holsurub RTE 6x60 mm",
  "availability": "IN_STOCK", 
  "priceInfo": {
    "price": 45.90,
    "currencyCode": "EUR"
  },
  "fulfillmentInfo": [
    {
      "type": "pickup-in-store",
      "placeIds": ["store_cluj_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_DEVICE_UNIC
    account_id: str, #HASH_ID_ERP_CLIENT (compania ca user)
    project_id: str = "PROJECT_ID",
    catalog_id: str = "default_catalog",
    serving_config_id: str = "bundle_recs",
    page_size: int = 6 
) -> list[str]:
   
    
    # 1. Creare user-event, folosim varianta de account ca 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("...") pentru exact match, atributele sunt de tip text si setate ca filtrabile
    filter_expr = (
        f'(availability: ANY("IN_STOCK")) AND '
        f'(attributes.is_bundle: ANY("true")) AND '
        f'(attributes.bundle_component_skus: ANY("{item_sku}"))'
    )
    
    # 3. Vezi Exemplul 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