Dynamic Pricing Agent Crew

Pricing is one of the most impactful decisions in e-commerce. A 1% improvement in price can boost profits by 11% on average. But getting pricing right requires analyzing competitors, understanding demand, and maintaining healthy margins - all while moving fast.

Let's build a Dynamic Pricing Crew that makes intelligent pricing recommendations.

The Pricing Challenge

Pricing Crew Architecture

Our Dynamic Pricing Crew has four specialized agents that work together:

┌─────────────────────────────────────────────────────────────────────┐
│                    DYNAMIC PRICING CREW                             │
│                                                                     │
│  ┌──────────────┐    ┌──────────────┐    ┌──────────────┐          │
│  │  Competitor  │    │    Demand    │    │    Risk      │          │
│  │   Monitor    │    │   Analyst    │    │   Analyst    │          │
│  └──────┬───────┘    └──────┬───────┘    └──────┬───────┘          │
│         │                   │                   │                   │
│         └───────────────────┼───────────────────┘                   │
│                             ▼                                       │
│                    ┌──────────────┐                                 │
│                    │   Pricing    │                                 │
│                    │  Strategist  │                                 │
│                    └──────────────┘                                 │
│                             │                                       │
│                             ▼                                       │
│                    PRICING RECOMMENDATION                           │
└─────────────────────────────────────────────────────────────────────┘

Complete Implementation

Step 1: Define Pricing Tools

from crewai import Agent, Task, Crew, Process
from crewai.tools import tool
from typing import List, Dict, Optional
from datetime import datetime, timedelta
import json

@tool("Competitor Price Monitor")
def get_competitor_prices(product_id: str) -> str:
    """
    Fetches current competitor prices for a product.

    Args:
        product_id: Our internal product identifier

    Returns:
        JSON with competitor names, prices, availability, and last updated time.
    """
    competitor_data = {
        "product_id": product_id,
        "our_current_price": 129.99,
        "competitors": [
            {
                "name": "Amazon",
                "price": 134.99,
                "shipping": "Free (Prime)",
                "in_stock": True,
                "last_updated": "2025-03-20T10:30:00Z",
                "price_history_7d": [139.99, 139.99, 134.99, 134.99, 134.99, 134.99, 134.99]
            },
            {
                "name": "Walmart",
                "price": 127.99,
                "shipping": "Free over $35",
                "in_stock": True,
                "last_updated": "2025-03-20T10:30:00Z",
                "price_history_7d": [129.99, 129.99, 127.99, 127.99, 127.99, 127.99, 127.99]
            },
            {
                "name": "Target",
                "price": 139.99,
                "shipping": "$5.99",
                "in_stock": True,
                "last_updated": "2025-03-20T10:30:00Z",
                "price_history_7d": [139.99, 139.99, 139.99, 139.99, 139.99, 139.99, 139.99]
            },
            {
                "name": "Dick's Sporting Goods",
                "price": 129.99,
                "shipping": "Free over $49",
                "in_stock": False,
                "last_updated": "2025-03-20T10:30:00Z",
                "price_history_7d": [129.99, 129.99, 129.99, 129.99, 129.99, 129.99, 129.99]
            }
        ],
        "market_summary": {
            "lowest_price": 127.99,
            "highest_price": 139.99,
            "average_price": 133.24,
            "our_position": "Middle (2nd lowest)"
        }
    }

    return json.dumps(competitor_data, indent=2)


@tool("Sales History Analyzer")
def get_sales_history(product_id: str, days: int = 30) -> str:
    """
    Retrieves sales history and demand patterns.

    Args:
        product_id: Product identifier
        days: Number of days of history to analyze

    Returns:
        JSON with daily sales, price points, and demand elasticity data.
    """
    sales_data = {
        "product_id": product_id,
        "period": f"Last {days} days",
        "total_units_sold": 847,
        "total_revenue": 109911.53,
        "daily_average": 28.2,
        "price_performance": [
            {"price": 119.99, "avg_daily_sales": 42, "margin": 28.5},
            {"price": 129.99, "avg_daily_sales": 28, "margin": 35.2},
            {"price": 139.99, "avg_daily_sales": 18, "margin": 41.8},
            {"price": 149.99, "avg_daily_sales": 11, "margin": 48.3}
        ],
        "demand_elasticity": -1.8,  # 1% price increase = 1.8% demand decrease
        "peak_demand_days": ["Saturday", "Sunday"],
        "low_demand_days": ["Tuesday", "Wednesday"],
        "seasonal_factor": 1.2,  # 20% above normal (spring running season)
        "inventory_status": {
            "current_stock": 234,
            "reorder_point": 100,
            "days_of_supply": 8.3
        }
    }

    return json.dumps(sales_data, indent=2)


@tool("Margin Calculator")
def calculate_margins(product_id: str, proposed_price: float) -> str:
    """
    Calculates profit margins for a proposed price.

    Args:
        product_id: Product identifier
        proposed_price: The price to evaluate

    Returns:
        JSON with cost breakdown, margin percentages, and profitability metrics.
    """
    # Product costs (would come from your actual systems)
    costs = {
        "cogs": 65.00,  # Cost of goods sold
        "shipping_to_warehouse": 3.50,
        "fulfillment_cost": 5.00,
        "payment_processing": proposed_price * 0.029 + 0.30,  # 2.9% + $0.30
        "returns_reserve": proposed_price * 0.05,  # 5% return rate
        "marketing_allocation": proposed_price * 0.10  # 10% for ads
    }

    total_cost = sum(costs.values())
    gross_margin = proposed_price - costs["cogs"] - costs["shipping_to_warehouse"]
    net_margin = proposed_price - total_cost

    margin_data = {
        "product_id": product_id,
        "proposed_price": proposed_price,
        "cost_breakdown": costs,
        "total_cost": round(total_cost, 2),
        "gross_margin": round(gross_margin, 2),
        "gross_margin_pct": round((gross_margin / proposed_price) * 100, 1),
        "net_margin": round(net_margin, 2),
        "net_margin_pct": round((net_margin / proposed_price) * 100, 1),
        "break_even_price": round(total_cost, 2),
        "minimum_viable_price": round(total_cost * 1.10, 2),  # 10% minimum margin
        "target_margin_price": round(total_cost / 0.70, 2)    # 30% target margin
    }

    return json.dumps(margin_data, indent=2)


@tool("Price History Database")
def get_price_history(product_id: str) -> str:
    """
    Retrieves historical pricing data for the product.

    Args:
        product_id: Product identifier

    Returns:
        JSON with price history, changes, and performance impact.
    """
    history = {
        "product_id": product_id,
        "price_changes": [
            {
                "date": "2025-01-15",
                "old_price": 139.99,
                "new_price": 129.99,
                "reason": "Competitive response",
                "impact": "+35% sales volume, -7% margin"
            },
            {
                "date": "2024-11-28",
                "old_price": 129.99,
                "new_price": 99.99,
                "reason": "Black Friday promotion",
                "impact": "+280% sales volume, -23% margin"
            },
            {
                "date": "2024-11-30",
                "old_price": 99.99,
                "new_price": 129.99,
                "reason": "End of promotion",
                "impact": "-65% sales volume (normalized after 5 days)"
            },
            {
                "date": "2024-09-01",
                "old_price": 149.99,
                "new_price": 139.99,
                "reason": "New model launch positioning",
                "impact": "+22% sales volume"
            }
        ],
        "insights": [
            "Product is price-sensitive (elasticity -1.8)",
            "$10 price drops historically increase volume 30-40%",
            "Post-promotion demand normalizes within 5-7 days",
            "Weekend pricing tests showed no significant difference"
        ]
    }

    return json.dumps(history, indent=2)

Step 2: Define Pricing Agents

# Agent 1: Competitor Monitor
competitor_monitor = Agent(
    role="Competitive Intelligence Analyst",

    goal="Monitor competitor prices and identify pricing opportunities "
         "and threats in real-time",

    backstory="""You are a competitive intelligence specialist who has
    tracked pricing for major e-commerce retailers. You understand that
    pricing is a dynamic battlefield.

    Your expertise:
    - Identifying price trends and patterns
    - Predicting competitor pricing moves
    - Spotting promotional cycles
    - Understanding market positioning

    You know that the lowest price doesn't always win - value perception
    matters. You look beyond the number to understand the full offer
    (shipping, availability, reputation).""",

    tools=[get_competitor_prices],
    verbose=True
)


# Agent 2: Demand Analyst
demand_analyst = Agent(
    role="Demand Forecasting Analyst",

    goal="Analyze sales patterns and demand elasticity to predict "
         "how price changes will impact volume and revenue",

    backstory="""You are a data scientist specializing in e-commerce
    demand modeling. You've built pricing models that increased revenue
    by 15%+ for major retailers.

    Your approach:
    - Analyze price-volume relationships
    - Account for seasonality and trends
    - Factor in inventory constraints
    - Consider promotional fatigue

    You understand that demand elasticity varies by segment, season,
    and competitive context. A price that works in January may fail
    in December.""",

    tools=[get_sales_history],
    verbose=True
)


# Agent 3: Risk Analyst
risk_analyst = Agent(
    role="Pricing Risk Analyst",

    goal="Identify potential risks and unintended consequences of "
         "pricing decisions before they cause damage",

    backstory="""You are a pricing strategist who has seen pricing
    decisions go wrong. You've witnessed price wars, margin erosion,
    and brand damage from poor pricing choices.

    You evaluate risks including:
    - Price war triggers (aggressive cuts that spark retaliation)
    - Brand perception damage (pricing too low undermines quality)
    - Margin sustainability (short-term gains vs long-term health)
    - Customer expectations (training customers to wait for sales)
    - Channel conflict (different prices causing issues)

    Your job is to be the voice of caution that prevents costly mistakes.""",

    tools=[calculate_margins, get_price_history],
    verbose=True
)


# Agent 4: Pricing Strategist
pricing_strategist = Agent(
    role="Dynamic Pricing Strategist",

    goal="Synthesize competitive, demand, and risk analysis to recommend "
         "optimal prices that maximize profit while maintaining market position",

    backstory="""You are a senior pricing strategist with experience at
    Amazon and leading DTC brands. You've optimized pricing for billions
    in revenue.

    Your pricing philosophy:
    - Price is a signal of value, not just a number
    - Optimal price balances volume, margin, and positioning
    - Psychological pricing matters ($99.99 vs $100)
    - Dynamic doesn't mean erratic - consistency builds trust
    - Always have a clear rationale for every recommendation

    You make decisions that are defensible, data-driven, and aligned
    with business strategy.""",

    tools=[calculate_margins],
    verbose=True
)

Step 3: Define the Pricing Workflow

# Task 1: Competitive Analysis
competitive_analysis_task = Task(
    description="""
    Analyze the competitive pricing landscape for product: {product_id}

    Your analysis should include:
    1. Current competitor prices and their recent changes
    2. Our price position relative to competitors
    3. Competitor availability and shipping factors
    4. Price trends over the past 7 days
    5. Opportunities (competitors out of stock, price increases)
    6. Threats (aggressive competitor pricing, new entrants)

    Provide actionable intelligence, not just data.
    """,

    expected_output="""
    Competitive intelligence report containing:
    - Price comparison table with all competitors
    - Our current market position (premium/value/parity)
    - Recent competitive moves and likely intentions
    - Recommended competitive response
    - Risk of price war assessment
    """,

    agent=competitor_monitor
)


# Task 2: Demand Analysis
demand_analysis_task = Task(
    description="""
    Analyze demand patterns and price sensitivity for product: {product_id}

    Your analysis should include:
    1. Sales velocity at different price points
    2. Demand elasticity calculation and interpretation
    3. Seasonal factors affecting demand
    4. Inventory considerations (days of supply)
    5. Revenue optimization scenarios

    Calculate expected outcomes for at least 3 price scenarios.
    """,

    expected_output="""
    Demand analysis report containing:
    - Price elasticity assessment
    - Revenue projections at 3+ price points
    - Seasonal adjustment recommendations
    - Inventory-aware pricing considerations
    - Optimal price range for maximum revenue
    """,

    agent=demand_analyst,
    context=[competitive_analysis_task]
)


# Task 3: Risk Assessment
risk_assessment_task = Task(
    description="""
    Evaluate the risks of potential pricing changes for product: {product_id}

    Consider the following risk categories:
    1. COMPETITIVE RISKS: Could this trigger a price war?
    2. MARGIN RISKS: Is this sustainable long-term?
    3. BRAND RISKS: Does this align with our positioning?
    4. CUSTOMER RISKS: Will this train bad behaviors?
    5. OPERATIONAL RISKS: Can we fulfill increased demand?

    For each potential price point, assign a risk score and explain.
    """,

    expected_output="""
    Risk assessment report containing:
    - Risk evaluation for each proposed price point
    - Price war probability assessment
    - Margin sustainability analysis
    - Brand perception impact
    - Mitigation strategies for identified risks
    - Risk-adjusted recommendations
    """,

    agent=risk_analyst,
    context=[competitive_analysis_task, demand_analysis_task]
)


# Task 4: Pricing Recommendation
pricing_recommendation_task = Task(
    description="""
    Synthesize all analyses to provide a final pricing recommendation
    for product: {product_id}

    Your recommendation must include:
    1. RECOMMENDED PRICE: The specific price point
    2. RATIONALE: Why this price optimizes our objectives
    3. EXPECTED IMPACT: Volume, revenue, and margin projections
    4. IMPLEMENTATION: When and how to implement
    5. MONITORING: KPIs to track and triggers for adjustment

    Use psychological pricing principles where appropriate.
    Provide a clear, defensible recommendation.
    """,

    expected_output="""
    Pricing recommendation document:
    - Primary recommendation with specific price
    - Alternative scenarios (aggressive/conservative)
    - Financial projections (30-day outlook)
    - Implementation timeline
    - Success metrics and review triggers
    - Executive summary (2-3 sentences)
    """,

    agent=pricing_strategist,
    context=[competitive_analysis_task, demand_analysis_task, risk_assessment_task]
)

Step 4: Assemble the Pricing Crew

# Create the Dynamic Pricing Crew
pricing_crew = Crew(
    agents=[
        competitor_monitor,
        demand_analyst,
        risk_analyst,
        pricing_strategist
    ],
    tasks=[
        competitive_analysis_task,
        demand_analysis_task,
        risk_assessment_task,
        pricing_recommendation_task
    ],
    process=Process.sequential,
    verbose=True
)


def get_pricing_recommendation(product_id: str) -> str:
    """Get a pricing recommendation for a product."""
    result = pricing_crew.kickoff(
        inputs={"product_id": product_id}
    )
    return result


# Run the pricing crew
if __name__ == "__main__":
    recommendation = get_pricing_recommendation("PROD-12345")
    print("\n" + "="*60)
    print("PRICING RECOMMENDATION")
    print("="*60)
    print(recommendation)

Sample Pricing Recommendation Output

Automated Price Monitoring

Set up the crew to run automatically when conditions change:

def should_reprice(product_id: str) -> bool:
    """Determine if repricing analysis is needed."""
    triggers = [
        competitor_price_changed(product_id),
        inventory_below_threshold(product_id),
        sales_velocity_anomaly(product_id),
        scheduled_review_due(product_id)
    ]
    return any(triggers)


def run_pricing_monitor():
    """Continuously monitor and reprice products."""
    products = get_active_products()

    for product_id in products:
        if should_reprice(product_id):
            recommendation = get_pricing_recommendation(product_id)
            if requires_human_approval(recommendation):
                send_for_approval(product_id, recommendation)
            else:
                apply_price_change(product_id, recommendation)

Key Takeaways

Next up: Customer Feedback Analysis Crew - Build agents that extract insights from reviews and generate responses.