Optimized Custom Functions

Introduction

The Optimized Custom Functions format represents the next generation of Quicklizard's custom pricing functions. This new version can process an entire array of products and return an array of results, providing a broader perspective and deeper insights for your pricing strategies.

Function Types and Capabilities

Pricing Rule Functions

Pricing Rule Functions are used to calculate and set product prices based on custom logic. These functions allow you to implement complex pricing strategies tailored to your business needs.

Capabilities

• Calculate custom prices based on product attributes, competition data, and other factors
• Apply business-specific pricing rules and logic
• Implement dynamic pricing strategies
• Set recommendation prices for products
• Provide pricing reasons for audit and transparency
• Set product attributes for further processing

Cost Functions

Cost Functions are used when the customer does not provide a "cost" for the products. These functions calculate the cost based on product data.

Capabilities

• Calculate product costs based on available data
• Support cost-plus pricing strategies
• Enable margin-based pricing decisions
• Provide cost data for profitability analysis

Before Pricing (Pre-Processor) Functions

Pre-Processor Functions prepare data for the pricing process. These functions run before the main pricing calculations and can set up the necessary context for pricing decisions.

Capabilities

• Prepare and transform data before pricing calculations
• Set and modify product attributes
• Filter products based on custom criteria
• Group products for bulk processing

After Pricing (Post-Processor) Functions

Post-Processor Functions handle calculations that must be performed at the end of the process after "limits" and "price rounding" have been applied. These functions can also confirm recommendations.

Capabilities

• Apply final adjustments to prices
• Confirm or reject pricing recommendations
• Add explanations or reasons for pricing decisions
• Set custom attributes based on final prices
• Trigger callbacks for dependent products
• Set products as read-only

Dynamic Attributes Functions

Dynamic Attributes Functions calculate and set dynamic attributes for products. These attributes can be used for display, filtering, or further processing.

Capabilities

• Calculate and set dynamic attributes based on product data
• Remove attributes by setting empty values
• Support display and filtering in the UI

Function Structure Requirements

All custom functions of type "optimized" must contain a main function that receives payload and params. The main function is the entry point for the function execution.

def main(client_key, channel, params, payload):
    # Your code here
    return result

Input Parameters:

• client_key: The client identifier
• channel: The pricing channel
• params: Function parameters configured in the UI
• payload: Array of product data

The params parameter contains the variables configured for the function in the UI, while the payload contains the array of products to process.

Output Format

The output should be an array where each element in the array is an object containing one product identifier (product_id/cid/uid) + channel, and each function has what is relevant to it, for example cost for cost function.

• product_id / cid / uid: Unique identifier for the product (Required for all function types)
• channel: The sales channel for the product (Required for all function types)

Context API

The Context API provides access to various helper functions and data sources within your custom functions. These include:

Logging

• ctx.log(msg): Log messages for debugging and monitoring

Redis Operations

• ctx.kv_get(key, scope=None): Get value from Redis
• ctx.kv_set(key, val, ttl=None, scope=None): Set value in Redis with TTL
• ctx.kv_del(key, scope=None): Delete value from Redis

Parameter Sheets

• ctx.get_pricing_parameters(sheet_name, key): Access parameter sheets

Variables and Configuration

• ctx.get_variables(): Get function variables from UI configuration

Widget Management

• ctx.add_widget(uid, channel, kind, title, position, data, ttl)
• ctx.del_widget(uid, channel, position)
• ctx.get_widget(uid, channel, position)
• ctx.get_widgets(uid, channel)

Internal Helper Functions

• ctx.get_elasticity_data(cid): Get elasticity data by product cid
• ctx.get_products_from_payload(): Extract products from payload

Code Examples

This section provides practical code examples for implementing various types of optimized custom functions.

Basic Function Structure

Here's a recommended structure for your optimized custom functions:

def run(client_key=None, channel=None, params={}, payload=[]):
    """Main processing logic"""
    resp = []
    products = get_data(payload, "products")
    variables = ctx.get_variables()

    for product in products:
        # Your processing logic here
        # Process product data and create response
        resp.append(format_recommendation(product))

    return resp

def main(client_key, channel, params, payload):
    """Entry point - calls run function"""
    return run(client_key, channel, params, payload)

Helper Functions

These helper functions can be used across different function types:

def get_data(payload=[], kind="products"):
    """Extract specific data type from payload"""
    products = []
    for item in payload:
        if item['name'] == kind:
            products = item['value']
    return products

def get_attr(attr_name, attrs):
    """Get attribute value by name, handling 'delete!' values"""
    for attr in attrs:
        if attr['name'] == attr_name:
            return attr['value'] if attr['value'] != 'delete!' else None
    return None

def update_attr_item(attrs, name, value):
    """Update or add attribute in attrs list"""
    # Check if the attribute exists
    for attr in attrs:
        if attr['name'] == name:
            attr['value'] = value  # Update the existing attribute
            return attrs
    # If the attribute does not exist, append it
    attrs.append({'name': name, 'value': value})
    return attrs

def format_recommendation(item={}):
    """Format recommendation response with all possible return fields"""
    item_hash = {
        'product_id': item['product_id'],
        'channel': item['channel'],
        'override_reason': 'true'  # Common requirement
    }

    # Optional fields - only include if they exist
    if 'new_recommendation' in item:
        item_hash['recommendation'] = round(item['new_recommendation'], 2)

    if 'new_reason' in item:
        item_hash['reason'] = item['new_reason']

    if 'new_attrs' in item:
        item_hash['attrs'] = item['new_attrs']

    if 'new_accept' in item:
        item_hash['accepted'] = item['new_accept']

    if 'callbacks' in item:
        item_hash['callbacks'] = item['callbacks']

    if 'pricing_error' in item:
        item_hash['pricing_error'] = item['pricing_error']

    return item_hash

Working with Competition Data

Competition data in optimized format requires JSON parsing:

def get_competitor_prices(product):
    """Extract and parse competitor prices"""
    competitions = product.get('competition', [])
    competitor_prices = []

    for comp in competitions:
        # Competition data may be JSON string, parse if needed
        if isinstance(comp, str):
            try:
                comp_data = json.loads(comp)
                competitor_prices.append(comp_data)
            except:
                ctx.log("Failed to parse competition data: %s" % comp)
        else:
            competitor_prices.append(comp)

    return competitor_prices

Callback System (Replacing send_to_pricing)

def send_followers(cpie_productid, channel):
    """Example callback implementation for triggering dependent pricing"""
    key = "%s_followers" % (cpie_productid)
    ctx.log("key %s " % key)
    followers_data = ctx.kv_get(key)
    ctx.log("followers_data %s " % followers_data)

    if followers_data and str(followers_data) != 'None':
        # Parse JSON string to get the followers value
        followers_json = json.loads(followers_data)
        followers_str = followers_json.get('followers', '')
        followers = followers_str.split("$@$") if followers_str else []
        ctx.log("Followers sent to pricing: %s" % followers)

        if followers:
            callbacks = []
            for follower in followers:
                callback = {
                    'group_by': 'cpie_productid',
                    'group_on': str(follower),  # Each follower gets its own bulk group
                    'channels': [channel],
                    'mode': 'bulk'
                }
                callbacks.append(callback)
            return callbacks, len(followers)  # Return list of all bulk callbacks
        else:
           return [], 0
    else:       
        return [], 0

# Usage in main function:
callbacks, count = send_followers(cpie_productid, channel)
if callbacks:
    product['callbacks'] = callbacks

Handling Dynamic Attributes

Special considerations for Dynamic Attributes functions:

# In Dynamic Attributes functions, you cannot set recommendations
# Only return attribute updates:
def format_dynamic_attr_response(item={}):
    """Format response for Dynamic Attributes functions"""
    item_hash = {
        'product_id': item['product_id'],
        'channel': item['channel'],
    }

    if 'dynamic_attrs' in item:
        item_hash['dynamic_attrs'] = item['dynamic_attrs']

    # You can also create dynamic_attrs directly in the response:
    # item_hash['dynamic_attrs'] = [
    #     {'name': 'attr_name', 'value': 'attr_value', 'type': 'string'},
    #     {'name': 'numeric_attr', 'value': '42', 'type': 'string'},
    #     # To delete an attribute, set an empty value (unlike other contexts where 'delete!' is used)
    #     {'name': 'attr_to_delete', 'value': '', 'type': 'string'}
    # ]

    # Note: 'recommendation' is NOT allowed in Dynamic Attributes functions
    # Note: In dynamic attributes, you can send an empty attribute value to delete it,
    #       unlike in other contexts where 'delete!' is used

    return item_hash

Example: Calculating and Returning Dynamic Attributes

This example demonstrates how to calculate various pricing metrics and return them as dynamic attributes:

import math

def truncate(f, n):
    return math.floor(f * 10**n) / 10**n

def get_attr_value(name, item={}):
    created_val = None
    for attr in item['attrs']:
        if attr['name'] == name:
            created_val = attr['value']
            break
    return created_val

def format_response(item={}):
    # Get the recommendation from the price object
    recom = item['price']['recommendation']

    # Get various attribute values
    variant_size = get_attr_value('variant_size', item)
    product_group = get_attr_value('product_group', item)
    product_category = get_attr_value('product_category', item)
    cost = item['price']['cost']
    comps = item['competition']

    # Calculate price per unit of measure
    recommened_price_per_uom = None
    if variant_size and variant_size != "delete!":
        recommened_price_per_uom = float(recom) / float(variant_size)
        recommened_price_per_uom = truncate(recommened_price_per_uom, 3)

    # Calculate margins and other metrics
    # (Simplified for this example)
    min_margin = "20%"
    min_margin_price = round(cost / (1 - 0.2), 2) if cost else "N/A"
    target_margin = "30%"
    target_margin_price = round(cost / (1 - 0.3), 2) if cost else "N/A"

    # Create flags based on business rules
    flag = '🟢' if recom and recom != 'N/A' and min_margin_price != 'N/A' and float(min_margin_price) < float(recom) else '🔴'

    # Return the response with dynamic attributes
    dynamic_attrs = [
        {'name': 'recommened_price_per_uom', 'value': str(recommened_price_per_uom), 'type': 'string'}, 
        {'name': 'competitors_available', 'value': str(len(comps)), 'type': 'string'},
        {'name': 'min_margin', 'value': min_margin, 'type': 'string'},
        {'name': 'min_margin_price', 'value': str(min_margin_price), 'type': 'string'},
        {'name': 'target_margin', 'value': target_margin, 'type': 'string'},
        {'name': 'target_margin_price', 'value': str(target_margin_price), 'type': 'string'},
        {'name': 'price_status_flag', 'value': flag, 'type': 'string'}
    ]

    # Example: Delete an attribute by setting empty value
    # If we want to remove the 'old_attribute' from the product
    dynamic_attrs.append({'name': 'old_attribute', 'value': '', 'type': 'string'})

    return {
        'product_id': item['product_id'],
        'channel': item['channel'],
        'dynamic_attrs': dynamic_attrs
    }

Complete Example with Cross-Referencing Products

This example demonstrates how to use optimized custom functions to analyze relationships between products and make pricing decisions based on those relationships:

import json

def run(client_key=None, channel=None, params={}, payload=[]):
    resp = []
    products = get_data(payload, "products")
    variables = ctx.get_variables()

    # Group products by category for cross-referencing
    products_by_category = {}
    for product in products:
        category = get_product_category(product)
        if category not in products_by_category:
            products_by_category[category] = []
        products_by_category[category].append(product)

    # Process each product with knowledge of other products in the same category
    for product in products:
        category = get_product_category(product)
        related_products = products_by_category[category]

        # Calculate average price in category
        category_prices = [p['price']['shelf'] for p in related_products if p['product_id'] != product['product_id']]
        avg_category_price = sum(category_prices) / len(category_prices) if category_prices else 0

        # Get competition data
        competitor_prices = get_competitor_prices(product)

        # Make pricing decision based on both competition and related products
        if competitor_prices and avg_category_price > 0:
            lowest_competitor = min(competitor_prices)

            # Cross-reference strategy: position relative to both competitors and category
            if avg_category_price > lowest_competitor:
                # Position between category average and lowest competitor
                new_price = (avg_category_price + lowest_competitor) / 2
            else:
                # Position slightly below category average
                new_price = avg_category_price * 0.95

            product['new_recommendation'] = new_price
            product['new_reason'] = 'Price optimized based on category analysis and competition'

            ctx.log("Set price for %s: %s (category avg: %s, lowest comp: %s)" % 
                   (product['product_id'], new_price, avg_category_price, lowest_competitor))

        resp.append(format_recommendation(product))

    return resp

def main(client_key, channel, params, payload):
    return run(client_key, channel, params, payload)

def get_product_category(product):
    """Extract product category from attributes"""
    attrs = product.get('attrs', [])
    for attr in attrs:
        if attr['name'] == 'category':
            return attr['value']
    return 'uncategorized'

def get_competitor_prices(product):
    """Extract and parse competitor prices"""
    competitions = product.get('competition', [])
    competitor_prices = []

    for comp in competitions:
        # Competition data may be JSON string, parse if needed
        if isinstance(comp, str):
            try:
                comp_data = json.loads(comp)
                competitor_prices.append(comp_data['price'])
            except:
                ctx.log("Failed to parse competition data: %s" % comp)
        else:
            competitor_prices.append(comp.get('price', 0))

    return competitor_prices

Pre-Processor and Post-Processor Examples

Before Pricing (Pre-Processor) Example:

def run(client_key=None, channel=None, params={}, payload=[]):
    """Pre-processor function to prepare data before pricing"""
    resp = []
    products = get_data(payload, "products")

    for product in products:
        attrs = product.get('attrs', [])
        new_attrs = []

        # Add calculated attributes
        margin = calculate_margin(product)
        new_attrs.append({'name': 'calculated_margin', 'value': str(margin)})

        # Update existing attributes
        for attr in attrs:
            if attr['name'] == 'status':
                attr['value'] = 'processed'
            new_attrs.append(attr)

        product['new_attrs'] = new_attrs

        resp.append(format_recommendation(product))

    return resp

def main(client_key, channel, params, payload):
    return run(client_key, channel, params, payload)

After Pricing (Post-Processor) Example:

def run(client_key=None, channel=None, params={}, payload=[]):
    """Post-processor function to handle results after pricing"""
    resp = []
    products = get_data(payload, "products")

    for product in products:
        recommendation = product['price'].get('recommendation', 0)
        shelf_price = product['price'].get('shelf', 0)

        # Apply rounding rules
        final_price = apply_rounding_rules(recommendation, product['attrs'])

        # Save to Redis for other functions
        key = "%s_%s_final_price" % (product['product_id'], product['channel'])
        ctx.kv_set(key, str(final_price), ttl=60*60*24)

        product['new_recommendation'] = final_price
        product['new_reason'] = 'Applied rounding rules'

        resp.append(format_recommendation(product))

    return resp

def main(client_key, channel, params, payload):
    return run(client_key, channel, params, payload)

Payload Structure Example

Understanding the payload structure is crucial for optimized functions:

# Example payload structure:
payload = [
    {
        "name": "products",
        "value": [
            {
                "product_id": "12345",
                "channel": "web",
                "cid": "abc123",
                "uid": "uid123",
                "price": {
                    "shelf": 10.99,
                    "recommendation": 10.99,
                    "cost": 5.50,
                    "inventory": 100,
                    "accepted": True
                },
                "attrs": [
                    {"name": "brand", "value": "Nike"},
                    {"name": "category", "value": "shoes"},
                    {"name": "qlia_elasticity_data", "value": "{\"error_msg\":\"no_results\"}"}
                ],
                "competition": [],  # May contain JSON strings
                "activity": {
                    "24hr": {"conversions": 5, "revenue": 54.95},
                    "7d": {"conversions": 35, "revenue": 384.65}
                }
            }
        ]
    }
]

Simple Pricing Function Example

Example of a complete pricing function:

import json

def run(client_key=None, channel=None, params={}, payload=[]):
    resp = []
    products = get_data(payload, "products")
    variables = ctx.get_variables()

    # Process each product
    for product in products:
        # Get competition data
        competitor_prices = get_competitor_prices(product)

        # Make pricing decision based on competition
        if competitor_prices:
            lowest_competitor = min([c.get('price', 0) for c in competitor_prices])

            # Position slightly below lowest competitor
            new_price = lowest_competitor * 0.95

            product['new_recommendation'] = new_price
            product['new_reason'] = 'Price set to 95% of lowest competitor'

        resp.append(format_recommendation(product))

    return resp

def main(client_key, channel, params, payload):
    return run(client_key, channel, params, payload)

Best Practices

This section provides best practices and guidelines for developing and maintaining optimized custom functions.

General Best Practices

1. Always use logging: Use ctx.log() for debugging and monitoring function execution. This helps with troubleshooting and understanding the flow of your function.
   Remember to remove or minimize logs when you finish debugging to avoid unnecessary output in production.

2. Handle missing data gracefully: Always check for None values and provide sensible defaults. This prevents errors when dealing with incomplete data.

3. Parse JSON data properly: Competition and other complex data may require JSON parsing. Always use try/except blocks when parsing JSON to handle potential errors.

4. Filter 'delete!' values: Always check for and handle 'delete!' attribute values when reading attributes (note: in dynamic attributes, empty values are used for deletion instead).

5. Use proper return formats: Return dictionary with appropriate keys instead of legacy set methods. This ensures compatibility with the optimized format.

6. Leverage Redis for state management: Use ctx.kv_set/get for sharing data between functions and caching. This allows for efficient data sharing across function calls.

7. Test with sample payloads: Use the built-in test functionality to validate your functions before deploying them to production.

8. Follow Python 2.7 syntax: Ensure compatibility with the execution environment by using Python 2.7 syntax.

9. Consider using run() and main(): While not strictly required, this structure is recommended for better code organization. The main() function serves as the entry point, while run() contains the main logic.

10. Use callbacks instead of send_to_pricing: Replace legacy channel pricing with the callback system for better performance and reliability.

11. Handle different function types appropriately: Dynamic Attributes functions cannot set recommendations and should use 'dynamic_attrs' format.

12. Use helper functions: Leverage provided helper functions for common operations to reduce code duplication and improve maintainability.

Performance Considerations

1. Minimize Redis operations: While Redis is useful for state management, excessive Redis operations can slow down your function. Use Redis only when necessary.

2. Batch processing: Take advantage of the optimized format's ability to process multiple products at once. Group operations when possible instead of processing each product individually.

3. Avoid unnecessary calculations: Only perform calculations that are needed for your pricing logic. Skip calculations for products that don't meet your criteria.

4. Use efficient data structures: Choose appropriate data structures for your use case to minimize memory usage and processing time.

5. Limit logging: While logging is important for debugging, excessive logging can impact performance. Use logging judiciously in production.

Error Handling

1. Validate input data: Always validate input data before processing to avoid errors during execution.

2. Use try/except blocks: Wrap critical sections of your code in try/except blocks to handle potential errors gracefully.

3. Provide meaningful error messages: When setting pricing_error, provide clear and specific error messages that help identify the issue.

4. Handle edge cases: Consider all possible edge cases and handle them appropriately to ensure your function works reliably in all scenarios.

5. Graceful degradation: Design your function to degrade gracefully when certain data is missing or invalid, rather than failing completely.

Testing and Debugging

1. Test with representative data: Test your function with data that represents real-world scenarios, including edge cases.

2. Use logging for debugging: Add strategic log statements to help trace the execution flow and identify issues.

3. Validate output format: Ensure your function returns data in the expected format to avoid integration issues.

4. Test performance: Test your function with large datasets to ensure it performs well at scale.

5. Review logs after deployment: Monitor logs after deployment to catch any issues that weren't apparent during testing.

By following these best practices, you can create robust, efficient, and maintainable optimized custom functions that deliver reliable pricing results.