prescriptive-analytics

处方分析

你是供应链处方分析和决策支持系统方面的专家。你的目标是帮助构建不仅能够预测将会发生什么，还能推荐应采取哪些行动以实现预期结果、优化绩效并降低风险的系统。

初步评估

构建处方分析系统之前，请先了解：

1. 决策背景

   • 需要针对哪些决策提供建议？（库存订单、路线规划、定价、分配）
   • 谁负责做出决策？（计划人员、采购人员、经理、自动化系统）
   • 目标是什么？（降低成本、提高服务水平、平衡多个目标）
   • 决策频率如何？（实时、每日、每周、战略性）

2. 当前状态

   • 目前如何做出决策？（手动、电子表格、基本规则、凭直觉）
   • 使用哪些信息？（历史数据、预测、约束条件）
   • 痛点有哪些？（决策不一致、结果欠佳、速度太慢）
   • 对建议的信任程度如何？（持怀疑态度或予以接受）

3. 分析成熟度

   • 是否已具备描述性分析？（发生了什么）
   • 是否具备诊断性分析？（为什么会发生）
   • 预测性分析是否有效？（将会发生什么）
   • 是否已准备好采用处方分析？（应该怎么做）

4. 技术要求

   • 需要实时建议还是批量建议？
   • 对可解释性有什么要求？（黑盒还是透明）
   • 是否需要与现有系统集成？
   • 需要人工参与还是完全自动化？

---

处方分析框架

分析演进金字塔

                    ┌─────────────────────────┐
                    │   PRESCRIPTIVE          │
                    │   What should we do?    │
                    │   • Recommendations     │
                    │   • Optimization        │
                    │   • Decision automation │
                    └─────────────────────────┘
                            ▲
                    ┌───────────────────────────┐
                    │   PREDICTIVE              │
                    │   What will happen?       │
                    │   • Forecasting           │
                    │   • Classification        │
                    │   • Risk scoring          │
                    └───────────────────────────┘
                            ▲
                    ┌─────────────────────────────┐
                    │   DIAGNOSTIC                │
                    │   Why did it happen?        │
                    │   • Root cause analysis     │
                    │   • Correlation             │
                    │   • Pattern recognition     │
                    └─────────────────────────────┘
                            ▲
                    ┌───────────────────────────────┐
                    │   DESCRIPTIVE                 │
                    │   What happened?              │
                    │   • Historical reporting      │
                    │   • KPIs & dashboards         │
                    │   • Data aggregation          │
                    └───────────────────────────────┘

处方分析的组成部分

1. 预测模型

• 预测未来状态（需求、交货周期、成本）
• 预测概率（延迟、缺陷、缺货）
• 为优化和建议提供输入

2. 优化引擎

• 数学优化（LP、MIP、NLP）
• 约束规划
• 启发式算法和元启发式算法
• 多目标优化

3. 决策规则与逻辑

• 业务规则引擎
• 策略约束
• 可行性检查
• 审批工作流

4. 解释与置信度

• 为什么提出此建议？
• 置信度
• 敏感性分析
• 替代方案情景

5. 行动执行

• API 集成
• 自动化工作流
• 人工审核界面
• 监控与反馈

---

库存补货建议

智能补货建议系统

import pandas as pd
import numpy as np
from scipy import stats
from datetime import datetime, timedelta
import joblib

class IntelligentReplenishmentAdvisor:
    """
    Prescriptive system for inventory replenishment recommendations

    Combines:
    - Demand forecasting (predictive)
    - Inventory optimization (prescriptive)
    - Business rules (constraints)
    - Explainability (trust)
    """

    def __init__(self):
        self.forecast_model = None
        self.optimization_config = {}
        self.business_rules = {}
        self.recommendations = []

    def load_models(self):
        """Load pre-trained ML models"""
        self.forecast_model = joblib.load('demand_forecast_model.pkl')

    def forecast_demand(self, sku, horizon_days=30):
        """
        Predict demand for SKU over horizon

        Returns distribution (mean, std) not just point estimate
        """

        # Get historical data and features
        features = self.get_forecast_features(sku)

        # Predict mean demand
        mean_demand = self.forecast_model.predict(features)[0]

        # Estimate uncertainty (from model or historical error)
        forecast_error = 0.15  # 15% coefficient of variation
        std_demand = mean_demand * forecast_error

        return {
            'sku': sku,
            'horizon_days': horizon_days,
            'mean_daily_demand': mean_demand / horizon_days,
            'std_daily_demand': std_demand / np.sqrt(horizon_days),
            'total_forecast': mean_demand,
            'confidence_interval_95': (
                mean_demand - 1.96 * std_demand,
                mean_demand + 1.96 * std_demand
            )
        }

    def get_forecast_features(self, sku):
        """Create feature vector for demand forecasting"""
        # Placeholder - would pull from database and engineer features
        return np.array([[1.0] * 20])  # Example feature vector

    def calculate_optimal_reorder(self, sku, current_inventory, forecast,
                                  lead_time, service_level=0.95):
        """
        Calculate optimal reorder quantity and timing

        Uses:
        - EOQ for quantity optimization
        - Service level for safety stock
        - Lead time for reorder point
        """

        # Get SKU parameters
        unit_cost = self.get_unit_cost(sku)
        order_cost = 100  # $ per order
        holding_cost_rate = 0.25  # 25% annually
        holding_cost = unit_cost * holding_cost_rate

        # Economic Order Quantity (EOQ)
        annual_demand = forecast['mean_daily_demand'] * 365
        eoq = np.sqrt((2 * annual_demand * order_cost) / holding_cost)

        # Safety stock based on service level
        z_score = stats.norm.ppf(service_level)
        std_during_lt = forecast['std_daily_demand'] * np.sqrt(lead_time)
        safety_stock = z_score * std_during_lt

        # Reorder point
        expected_demand_lt = forecast['mean_daily_demand'] * lead_time
        reorder_point = expected_demand_lt + safety_stock

        # Inventory position
        inventory_position = current_inventory  # Simplified (+ on_order in practice)

        # Calculate recommended order quantity
        if inventory_position < reorder_point:
            # Order up to EOQ (or more if severely understocked)
            target_inventory = reorder_point + eoq
            recommended_order = max(0, target_inventory - inventory_position)
        else:
            recommended_order = 0

        return {
            'sku': sku,
            'current_inventory': current_inventory,
            'reorder_point': reorder_point,
            'safety_stock': safety_stock,
            'eoq': eoq,
            'recommended_order_quantity': recommended_order,
            'inventory_position': inventory_position,
            'days_of_supply': inventory_position / forecast['mean_daily_demand'],
            'stockout_risk': self.calculate_stockout_risk(
                inventory_position, forecast, lead_time
            )
        }

    def calculate_stockout_risk(self, inventory_position, forecast, lead_time):
        """
        Calculate probability of stockout during lead time

        P(Demand during LT > Inventory Position)
        """

        mean_demand_lt = forecast['mean_daily_demand'] * lead_time
        std_demand_lt = forecast['std_daily_demand'] * np.sqrt(lead_time)

        if std_demand_lt == 0:
            return 0 if inventory_position >= mean_demand_lt else 1

        z = (inventory_position - mean_demand_lt) / std_demand_lt
        stockout_prob = 1 - stats.norm.cdf(z)

        return stockout_prob

    def apply_business_rules(self, recommendation):
        """
        Apply business rules and constraints

        Examples:
        - Minimum order quantity (MOQ)
        - Case pack constraints
        - Budget limits
        - Supplier capacity
        - Shelf life considerations
        """

        sku = recommendation['sku']

        # MOQ constraint
        moq = self.get_moq(sku)
        if recommendation['recommended_order_quantity'] > 0 and \
           recommendation['recommended_order_quantity'] < moq:
            recommendation['recommended_order_quantity'] = moq
            recommendation['constraints_applied'].append('MOQ')

        # Case pack rounding
        case_pack = self.get_case_pack(sku)
        if case_pack > 0:
            recommended = recommendation['recommended_order_quantity']
            rounded = np.ceil(recommended / case_pack) * case_pack
            recommendation['recommended_order_quantity'] = rounded
            if rounded != recommended:
                recommendation['constraints_applied'].append('Case Pack')

        # Budget check
        unit_cost = self.get_unit_cost(sku)
        order_value = recommendation['recommended_order_quantity'] * unit_cost

        if order_value > self.business_rules.get('max_order_value', float('inf')):
            recommendation['requires_approval'] = True
            recommendation['approval_reason'] = f"Order value ${order_value:,.0f} exceeds limit"

        return recommendation

    def generate_explanation(self, recommendation):
        """
        Generate human-readable explanation for recommendation

        Critical for user trust and adoption
        """

        sku = recommendation['sku']
        order_qty = recommendation['recommended_order_quantity']

        if order_qty == 0:
            explanation = f"No order needed. Current inventory ({recommendation['current_inventory']} units) " \
                         f"is above reorder point ({recommendation['reorder_point']:.0f} units). " \
                         f"You have {recommendation['days_of_supply']:.1f} days of supply."

            return {
                'action': 'DO NOT ORDER',
                'explanation': explanation,
                'confidence': 'High',
                'risk_level': 'Low'
            }

        else:
            stockout_risk = recommendation['stockout_risk']

            explanation = f"Order {order_qty:.0f} units. " \
                         f"Current inventory ({recommendation['current_inventory']} units) " \
                         f"is below reorder point ({recommendation['reorder_point']:.0f} units). " \
                         f"Without ordering, stockout risk is {stockout_risk:.1%}. " \
                         f"Recommended quantity optimizes ordering and holding costs (EOQ)."

            # Risk assessment
            if stockout_risk > 0.2:
                risk_level = 'HIGH'
                urgency = 'URGENT'
            elif stockout_risk > 0.05:
                risk_level = 'MEDIUM'
                urgency = 'Soon'
            else:
                risk_level = 'LOW'
                urgency = 'Normal'

            return {
                'action': 'ORDER',
                'quantity': order_qty,
                'explanation': explanation,
                'urgency': urgency,
                'risk_level': risk_level,
                'stockout_risk': f"{stockout_risk:.1%}",
                'days_until_stockout': recommendation['days_of_supply']
            }

    def generate_recommendations(self, sku_list):
        """
        Generate recommendations for list of SKUs

        Returns prioritized action list
        """

        recommendations = []

        for sku in sku_list:
            # Get current state
            current_inventory = self.get_current_inventory(sku)
            lead_time = self.get_lead_time(sku)

            # Forecast demand
            forecast = self.forecast_demand(sku, horizon_days=30)

            # Calculate optimal reorder
            recommendation = self.calculate_optimal_reorder(
                sku, current_inventory, forecast, lead_time
            )

            # Apply business rules
            recommendation['constraints_applied'] = []
            recommendation['requires_approval'] = False
            recommendation = self.apply_business_rules(recommendation)

            # Generate explanation
            explanation = self.generate_explanation(recommendation)
            recommendation['explanation'] = explanation

            recommendations.append(recommendation)

        # Prioritize recommendations
        recommendations_df = pd.DataFrame(recommendations)

        # Sort by urgency and risk
        recommendations_df['priority_score'] = (
            recommendations_df['stockout_risk'] * 100 +
            (1 / recommendations_df['days_of_supply']) * 10
        )

        recommendations_df = recommendations_df.sort_values(
            'priority_score',
            ascending=False
        )

        return recommendations_df

    def get_current_inventory(self, sku):
        """Get current inventory from system"""
        # In production: query WMS/ERP
        return np.random.randint(50, 500)

    def get_lead_time(self, sku):
        """Get supplier lead time"""
        return 14  # days

    def get_unit_cost(self, sku):
        """Get unit cost"""
        return 25.00

    def get_moq(self, sku):
        """Get minimum order quantity"""
        return 100

    def get_case_pack(self, sku):
        """Get case pack quantity"""
        return 50

# Example usage
advisor = IntelligentReplenishmentAdvisor()
advisor.load_models()

# Generate recommendations
sku_list = ['SKU_001', 'SKU_002', 'SKU_003', 'SKU_004', 'SKU_005']
recommendations = advisor.generate_recommendations(sku_list)

# Display top recommendations
print("\n" + "="*80)
print("INTELLIGENT REPLENISHMENT RECOMMENDATIONS")
print("="*80 + "\n")

for idx, row in recommendations.head(10).iterrows():
    print(f"SKU: {row['sku']}")
    print(f"  Action: {row['explanation']['action']}")
    if row['explanation']['action'] == 'ORDER':
        print(f"  Quantity: {row['explanation']['quantity']:.0f} units")
        print(f"  Urgency: {row['explanation']['urgency']}")
        print(f"  Risk: {row['explanation']['risk_level']} (Stockout risk: {row['explanation']['stockout_risk']})")
    print(f"  Explanation: {row['explanation']['explanation']}")
    if row['requires_approval']:
        print(f"  ⚠️  Requires Approval: {row['approval_reason']}")
    print()

---

路线优化建议

智能路线规划系统

from scipy.spatial import distance_matrix
from sklearn.cluster import KMeans
import numpy as np

class RouteOptimizationAdvisor:
    """
    Prescriptive system for delivery route recommendations

    Provides:
    - Optimal route sequences
    - Vehicle assignments
    - Time window compliance
    - Real-time adjustments
    """

    def __init__(self, config):
        self.config = config
        self.routes = []

    def generate_route_recommendations(self, orders, vehicles, constraints):
        """
        Generate optimal routing recommendations

        orders: list of delivery orders with locations and time windows
        vehicles: available vehicles with capacities
        constraints: max driving time, time windows, capacity
        """

        # Step 1: Cluster orders by geography
        clusters = self.cluster_orders_geographically(
            orders,
            n_clusters=len(vehicles)
        )

        # Step 2: Assign clusters to vehicles
        vehicle_assignments = self.assign_clusters_to_vehicles(
            clusters, vehicles
        )

        # Step 3: Optimize sequence within each route
        optimized_routes = []

        for vehicle_id, cluster_orders in vehicle_assignments.items():
            route = self.optimize_route_sequence(
                cluster_orders,
                vehicles[vehicle_id],
                constraints
            )

            route['vehicle_id'] = vehicle_id
            route['recommendations'] = self.generate_route_explanation(route)

            optimized_routes.append(route)

        return optimized_routes

    def cluster_orders_geographically(self, orders, n_clusters):
        """
        Cluster orders by location for vehicle assignment

        Uses K-means clustering
        """

        # Extract coordinates
        coords = np.array([[o['lat'], o['lon']] for o in orders])

        # Cluster
        kmeans = KMeans(n_clusters=n_clusters, random_state=42)
        labels = kmeans.fit_predict(coords)

        # Group orders by cluster
        clusters = {}
        for i, label in enumerate(labels):
            if label not in clusters:
                clusters[label] = []
            clusters[label].append(orders[i])

        return clusters

    def assign_clusters_to_vehicles(self, clusters, vehicles):
        """
        Assign clusters to vehicles based on capacity and constraints
        """

        assignments = {}

        for i, (cluster_id, orders) in enumerate(clusters.items()):
            vehicle_id = list(vehicles.keys())[i]

            # Check capacity
            total_volume = sum(o['volume'] for o in orders)
            total_weight = sum(o['weight'] for o in orders)

            vehicle = vehicles[vehicle_id]

            if total_volume > vehicle['capacity_volume'] or \
               total_weight > vehicle['capacity_weight']:
                # Need to split cluster or use larger vehicle
                # Simplified: just assign and flag
                print(f"WARNING: Cluster {cluster_id} exceeds vehicle {vehicle_id} capacity")

            assignments[vehicle_id] = orders

        return assignments

    def optimize_route_sequence(self, orders, vehicle, constraints):
        """
        Optimize stop sequence using nearest neighbor heuristic

        For production: use proper VRP solver (OR-Tools, Gurobi)
        """

        if not orders:
            return {
                'stops': [],
                'total_distance': 0,
                'total_time': 0,
                'utilization': 0
            }

        # Start from depot
        depot = {'lat': self.config['depot_lat'],
                'lon': self.config['depot_lon'],
                'order_id': 'DEPOT'}

        current_location = depot
        remaining_orders = orders.copy()
        route_sequence = [depot]
        total_distance = 0

        # Nearest neighbor heuristic
        while remaining_orders:
            # Find nearest order
            nearest_order = min(
                remaining_orders,
                key=lambda o: self.calculate_distance(
                    current_location['lat'], current_location['lon'],
                    o['lat'], o['lon']
                )
            )

            distance = self.calculate_distance(
                current_location['lat'], current_location['lon'],
                nearest_order['lat'], nearest_order['lon']
            )

            total_distance += distance
            route_sequence.append(nearest_order)
            remaining_orders.remove(nearest_order)
            current_location = nearest_order

        # Return to depot
        distance_to_depot = self.calculate_distance(
            current_location['lat'], current_location['lon'],
            depot['lat'], depot['lon']
        )
        total_distance += distance_to_depot
        route_sequence.append(depot)

        # Calculate metrics
        total_time = total_distance / vehicle.get('avg_speed_mph', 30) * 60  # minutes
        total_time += len(orders) * constraints.get('stop_time_minutes', 15)

        total_volume = sum(o['volume'] for o in orders)
        utilization = total_volume / vehicle['capacity_volume'] * 100

        return {
            'stops': route_sequence,
            'total_distance_miles': total_distance,
            'total_time_minutes': total_time,
            'utilization_pct': utilization,
            'num_stops': len(orders),
            'feasible': total_time <= constraints.get('max_driving_time_minutes', 480)
        }

    def calculate_distance(self, lat1, lon1, lat2, lon2):
        """Calculate distance between two coordinates (Haversine)"""

        from math import radians, sin, cos, sqrt, atan2

        R = 3959  # Earth radius in miles

        lat1, lon1, lat2, lon2 = map(radians, [lat1, lon1, lat2, lon2])

        dlat = lat2 - lat1
        dlon = lon2 - lon1

        a = sin(dlat/2)**2 + cos(lat1) * cos(lat2) * sin(dlon/2)**2
        c = 2 * atan2(sqrt(a), sqrt(1-a))

        return R * c

    def generate_route_explanation(self, route):
        """
        Generate explanation and recommendations for route
        """

        recommendations = {
            'summary': f"{route['num_stops']} stops, "
                      f"{route['total_distance_miles']:.1f} miles, "
                      f"{route['total_time_minutes']/60:.1f} hours",
            'utilization': f"{route['utilization_pct']:.0f}% capacity used",
            'feasibility': 'FEASIBLE' if route['feasible'] else 'INFEASIBLE',
            'improvements': []
        }

        # Identify improvement opportunities
        if route['utilization_pct'] < 70:
            recommendations['improvements'].append(
                f"Low utilization ({route['utilization_pct']:.0f}%). "
                "Consider consolidating with another route."
            )

        if route['total_time_minutes'] > 420:  # 7 hours
            recommendations['improvements'].append(
                "Long route duration. Consider splitting into two routes."
            )

        if not route['feasible']:
            recommendations['improvements'].append(
                "Route exceeds maximum driving time. MUST split or reschedule."
            )

        return recommendations

# Example usage
config = {
    'depot_lat': 34.0522,
    'depot_lon': -118.2437
}

advisor = RouteOptimizationAdvisor(config)

# Sample orders
orders = [
    {'order_id': 'O1', 'lat': 34.1, 'lon': -118.3, 'volume': 100, 'weight': 50},
    {'order_id': 'O2', 'lat': 34.2, 'lon': -118.1, 'volume': 150, 'weight': 75},
    {'order_id': 'O3', 'lat': 34.0, 'lon': -118.4, 'volume': 80, 'weight': 40},
]

vehicles = {
    'V1': {'capacity_volume': 1000, 'capacity_weight': 2000, 'avg_speed_mph': 30},
    'V2': {'capacity_volume': 800, 'capacity_weight': 1500, 'avg_speed_mph': 30}
}

constraints = {
    'max_driving_time_minutes': 480,  # 8 hours
    'stop_time_minutes': 15
}

routes = advisor.generate_route_recommendations(orders, vehicles, constraints)

for route in routes:
    print(f"\nVehicle {route['vehicle_id']}:")
    print(f"  {route['recommendations']['summary']}")
    print(f"  {route['recommendations']['utilization']}")
    print(f"  Status: {route['recommendations']['feasibility']}")
    if route['recommendations']['improvements']:
        print("  Improvements:")
        for imp in route['recommendations']['improvements']:
            print(f"    - {imp}")

---

供应商选择建议

import pandas as pd
import numpy as np
from sklearn.preprocessing import MinMaxScaler

class SupplierSelectionAdvisor:
    """
    Prescriptive system for supplier selection recommendations

    Multi-criteria decision analysis considering:
    - Cost
    - Quality
    - Delivery performance
    - Capacity
    - Risk
    """

    def __init__(self):
        self.scaler = MinMaxScaler()

    def evaluate_suppliers(self, rfq_data, suppliers_data, criteria_weights):
        """
        Evaluate and rank suppliers for RFQ

        rfq_data: RFQ details (product, quantity, requirements)
        suppliers_data: Supplier profiles and historical performance
        criteria_weights: importance weights for each criterion
        """

        evaluations = []

        for supplier in suppliers_data:
            evaluation = self.evaluate_single_supplier(
                supplier, rfq_data, criteria_weights
            )
            evaluations.append(evaluation)

        # Convert to DataFrame
        eval_df = pd.DataFrame(evaluations)

        # Calculate weighted score
        eval_df['total_score'] = 0

        for criterion, weight in criteria_weights.items():
            if f'{criterion}_score' in eval_df.columns:
                eval_df['total_score'] += eval_df[f'{criterion}_score'] * weight

        # Rank
        eval_df = eval_df.sort_values('total_score', ascending=False)

        # Generate recommendations
        recommendations = self.generate_supplier_recommendations(
            eval_df, rfq_data
        )

        return eval_df, recommendations

    def evaluate_single_supplier(self, supplier, rfq_data, criteria_weights):
        """
        Evaluate single supplier across multiple criteria
        """

        evaluation = {
            'supplier_id': supplier['supplier_id'],
            'supplier_name': supplier['name']
        }

        # 1. Cost Score (lower cost = higher score)
        quoted_price = supplier.get('quoted_price', float('inf'))
        if quoted_price < float('inf'):
            # Normalize: score = 100 * (1 - (price - min_price) / (max_price - min_price))
            evaluation['quoted_price'] = quoted_price
            evaluation['cost_score'] = 100 - min(quoted_price / 100, 100)  # Simplified
        else:
            evaluation['cost_score'] = 0

        # 2. Quality Score (based on historical defect rate)
        defect_rate = supplier.get('historical_defect_rate', 0.05)
        evaluation['defect_rate'] = defect_rate
        evaluation['quality_score'] = max(0, 100 - defect_rate * 1000)

        # 3. Delivery Performance (OTD rate)
        otd_rate = supplier.get('on_time_delivery_rate', 0.85)
        evaluation['otd_rate'] = otd_rate
        evaluation['delivery_score'] = otd_rate * 100

        # 4. Capacity Match
        supplier_capacity = supplier.get('available_capacity', 0)
        required_quantity = rfq_data['quantity']
        capacity_ratio = min(supplier_capacity / required_quantity, 1.0)
        evaluation['capacity_ratio'] = capacity_ratio
        evaluation['capacity_score'] = capacity_ratio * 100

        # 5. Risk Score (inverse - lower risk = higher score)
        risk_factors = {
            'financial_health': supplier.get('financial_rating', 'BBB'),
            'geographic_risk': supplier.get('country_risk', 'Low'),
            'single_source': supplier.get('is_sole_source', False)
        }

        risk_score = self.calculate_risk_score(risk_factors)
        evaluation['risk_score'] = 100 - risk_score

        # 6. Lead Time (shorter = better)
        lead_time = supplier.get('lead_time_days', 30)
        evaluation['lead_time_days'] = lead_time
        evaluation['lead_time_score'] = max(0, 100 - lead_time)

        # 7. Past Relationship
        past_orders = supplier.get('past_orders_count', 0)
        evaluation['relationship_score'] = min(past_orders / 10, 1.0) * 100

        return evaluation

    def calculate_risk_score(self, risk_factors):
        """
        Calculate overall risk score (0-100, higher = more risk)
        """

        risk_score = 0

        # Financial risk
        financial_ratings = {'AAA': 0, 'AA': 10, 'A': 20, 'BBB': 30,
                           'BB': 50, 'B': 70, 'CCC': 90}
        risk_score += financial_ratings.get(risk_factors['financial_health'], 40)

        # Geographic risk
        geo_risk = {'Low': 0, 'Medium': 20, 'High': 40}
        risk_score += geo_risk.get(risk_factors['geographic_risk'], 20)

        # Single source risk
        if risk_factors['single_source']:
            risk_score += 30

        return min(risk_score, 100)

    def generate_supplier_recommendations(self, eval_df, rfq_data):
        """
        Generate actionable supplier recommendations
        """

        recommendations = []

        # Top recommendation
        top_supplier = eval_df.iloc[0]

        rec = {
            'rank': 1,
            'supplier': top_supplier['supplier_name'],
            'action': 'AWARD',
            'rationale': f"Best overall score ({top_supplier['total_score']:.0f}/100). "
                        f"Strong performance in key areas: "
                        f"Quality ({top_supplier['quality_score']:.0f}), "
                        f"Delivery ({top_supplier['delivery_score']:.0f}), "
                        f"Price (${top_supplier['quoted_price']:.2f}).",
            'confidence': 'High' if top_supplier['total_score'] > 80 else 'Medium'
        }

        recommendations.append(rec)

        # Secondary recommendation (backup supplier)
        if len(eval_df) > 1:
            second_supplier = eval_df.iloc[1]

            rec = {
                'rank': 2,
                'supplier': second_supplier['supplier_name'],
                'action': 'BACKUP',
                'rationale': f"Strong alternative ({second_supplier['total_score']:.0f}/100). "
                            "Recommend as backup supplier for risk mitigation.",
                'confidence': 'Medium'
            }

            recommendations.append(rec)

        # Dual sourcing recommendation
        if rfq_data['quantity'] > 1000 and len(eval_df) >= 2:
            recommendations.append({
                'supplier': f"{eval_df.iloc[0]['supplier_name']} + {eval_df.iloc[1]['supplier_name']}",
                'action': 'DUAL SOURCE',
                'rationale': f"Large volume ({rfq_data['quantity']} units) justifies dual sourcing "
                            "for risk mitigation. Recommend 70/30 split.",
                'split': {eval_df.iloc[0]['supplier_name']: 0.7,
                         eval_df.iloc[1]['supplier_name']: 0.3}
            })

        return recommendations

# Example usage
rfq_data = {
    'rfq_id': 'RFQ-2025-001',
    'product': 'Widget A',
    'quantity': 10000,
    'delivery_date': '2025-03-01'
}

suppliers_data = [
    {
        'supplier_id': 'S1',
        'name': 'Acme Corp',
        'quoted_price': 45.00,
        'historical_defect_rate': 0.02,
        'on_time_delivery_rate': 0.95,
        'available_capacity': 15000,
        'financial_rating': 'AA',
        'country_risk': 'Low',
        'is_sole_source': False,
        'lead_time_days': 21,
        'past_orders_count': 25
    },
    {
        'supplier_id': 'S2',
        'name': 'GlobalTech Inc',
        'quoted_price': 42.00,
        'historical_defect_rate': 0.05,
        'on_time_delivery_rate': 0.88,
        'available_capacity': 12000,
        'financial_rating': 'BBB',
        'country_risk': 'Medium',
        'is_sole_source': False,
        'lead_time_days': 35,
        'past_orders_count': 10
    },
    {
        'supplier_id': 'S3',
        'name': 'Reliable Parts Co',
        'quoted_price': 48.00,
        'historical_defect_rate': 0.01,
        'on_time_delivery_rate': 0.98,
        'available_capacity': 20000,
        'financial_rating': 'AAA',
        'country_risk': 'Low',
        'is_sole_source': False,
        'lead_time_days': 14,
        'past_orders_count': 50
    }
]

criteria_weights = {
    'cost': 0.30,
    'quality': 0.25,
    'delivery': 0.20,
    'capacity': 0.10,
    'risk': 0.10,
    'lead_time': 0.05
}

advisor = SupplierSelectionAdvisor()
evaluations, recommendations = advisor.evaluate_suppliers(
    rfq_data, suppliers_data, criteria_weights
)

print("\n" + "="*80)
print("SUPPLIER SELECTION RECOMMENDATIONS")
print("="*80 + "\n")

print("Supplier Scores:\n")
print(evaluations[['supplier_name', 'total_score', 'quoted_price',
                  'quality_score', 'delivery_score']].to_string(index=False))

print("\n\nRecommendations:\n")
for rec in recommendations:
    print(f"Rank {rec.get('rank', 'N/A')}: {rec['supplier']}")
    print(f"  Action: {rec['action']}")
    print(f"  Rationale: {rec['rationale']}")
    if 'split' in rec:
        print(f"  Split: {rec['split']}")
    print()

---

实时决策支持

动态定价建议

import numpy as np
from datetime import datetime, timedelta

class DynamicPricingAdvisor:
    """
    Real-time prescriptive pricing recommendations

    Considers:
    - Current inventory levels
    - Demand forecast
    - Competitor prices
    - Shelf life / expiration
    - Profit margins
    """

    def __init__(self, config):
        self.config = config

    def recommend_price(self, product, current_price, context):
        """
        Recommend optimal price for product

        context: dict with market conditions, inventory, etc.
        """

        # Extract context
        inventory_level = context['inventory_level']
        days_until_expiry = context.get('days_until_expiry', 365)
        competitor_price = context.get('competitor_price', current_price)
        recent_sales_velocity = context['recent_sales_velocity']
        target_inventory = context.get('target_inventory', 100)

        # Calculate elasticity-based recommendation
        base_recommendation = self.calculate_elasticity_price(
            product, current_price, recent_sales_velocity
        )

        # Adjust for inventory position
        if inventory_level > target_inventory * 1.5:
            # Excess inventory - recommend discount
            price_adjustment = -0.10  # 10% discount
            reason = "Excess inventory"

        elif inventory_level < target_inventory * 0.5:
            # Low inventory - can increase price
            price_adjustment = 0.05  # 5% premium
            reason = "Low inventory"

        else:
            price_adjustment = 0
            reason = "Optimal inventory level"

        # Adjust for perishability
        if days_until_expiry < 7:
            # Urgent - aggressive discount
            price_adjustment = min(price_adjustment, -0.25)
            reason += ", Near expiration"

        elif days_until_expiry < 30:
            # Moderate urgency
            price_adjustment = min(price_adjustment, -0.10)
            reason += ", Approaching expiration"

        # Competitive positioning
        if competitor_price < current_price * 0.95:
            # Competitor significantly cheaper
            price_adjustment = min(price_adjustment, -0.05)
            reason += ", Competitor underpricing"

        # Calculate recommended price
        recommended_price = current_price * (1 + price_adjustment)

        # Enforce min/max constraints
        min_price = product.get('min_price', current_price * 0.7)
        max_price = product.get('max_price', current_price * 1.3)

        recommended_price = np.clip(recommended_price, min_price, max_price)

        # Calculate expected impact
        impact = self.estimate_price_impact(
            current_price, recommended_price,
            recent_sales_velocity, inventory_level
        )

        return {
            'product_id': product['product_id'],
            'current_price': current_price,
            'recommended_price': recommended_price,
            'price_change_pct': (recommended_price - current_price) / current_price * 100,
            'reason': reason,
            'expected_impact': impact,
            'confidence': self.calculate_confidence(context),
            'urgency': self.calculate_urgency(days_until_expiry, inventory_level, target_inventory)
        }

    def calculate_elasticity_price(self, product, current_price, sales_velocity):
        """
        Calculate price based on demand elasticity

        Price elasticity = % change in quantity / % change in price
        """

        # Get historical elasticity (would be learned from data)
        elasticity = product.get('price_elasticity', -1.5)

        # If sales are low, consider lowering price
        if sales_velocity < product.get('target_velocity', 10):
            # Reduce price to stimulate demand
            return current_price * 0.95
        else:
            return current_price

    def estimate_price_impact(self, current_price, new_price, sales_velocity, inventory):
        """
        Estimate impact of price change
        """

        price_change = (new_price - current_price) / current_price

        # Assume price elasticity of -1.5 (10% price cut → 15% sales increase)
        elasticity = -1.5
        demand_change = price_change * elasticity

        new_sales_velocity = sales_velocity * (1 + demand_change)

        # Calculate revenue impact
        current_revenue = current_price * sales_velocity * 30  # 30 days
        new_revenue = new_price * new_sales_velocity * 30

        # Calculate inventory impact (days to sell out)
        current_days_to_sellout = inventory / sales_velocity if sales_velocity > 0 else 999
        new_days_to_sellout = inventory / new_sales_velocity if new_sales_velocity > 0 else 999

        return {
            'revenue_change': new_revenue - current_revenue,
            'revenue_change_pct': (new_revenue - current_revenue) / current_revenue * 100,
            'new_sales_velocity': new_sales_velocity,
            'demand_change_pct': demand_change * 100,
            'current_days_to_sellout': current_days_to_sellout,
            'new_days_to_sellout': new_days_to_sellout
        }

    def calculate_confidence(self, context):
        """Calculate confidence in recommendation"""

        # Confidence based on data quality and market stability
        data_points = context.get('historical_data_points', 0)

        if data_points > 100:
            return 'High'
        elif data_points > 30:
            return 'Medium'
        else:
            return 'Low'

    def calculate_urgency(self, days_until_expiry, inventory, target):
        """Calculate urgency of price action"""

        if days_until_expiry < 7 or inventory > target * 2:
            return 'URGENT'
        elif days_until_expiry < 30 or inventory > target * 1.5:
            return 'High'
        else:
            return 'Normal'

# Example usage
config = {}
advisor = DynamicPricingAdvisor(config)

product = {
    'product_id': 'P001',
    'name': 'Fresh Strawberries',
    'price_elasticity': -1.8,
    'min_price': 3.00,
    'max_price': 6.00
}

context = {
    'inventory_level': 500,
    'target_inventory': 200,
    'days_until_expiry': 5,
    'competitor_price': 4.50,
    'recent_sales_velocity': 30,  # units per day
    'historical_data_points': 150
}

recommendation = advisor.recommend_price(product, current_price=4.99, context=context)

print("\n" + "="*80)
print("DYNAMIC PRICING RECOMMENDATION")
print("="*80 + "\n")

print(f"Product: {product['name']}")
print(f"Current Price: ${recommendation['current_price']:.2f}")
print(f"Recommended Price: ${recommendation['recommended_price']:.2f}")
print(f"Change: {recommendation['price_change_pct']:+.1f}%")
print(f"\nReason: {recommendation['reason']}")
print(f"Urgency: {recommendation['urgency']}")
print(f"Confidence: {recommendation['confidence']}")

print(f"\nExpected Impact:")
impact = recommendation['expected_impact']
print(f"  Revenue Change: ${impact['revenue_change']:+,.0f} ({impact['revenue_change_pct']:+.1f}%)")
print(f"  Demand Change: {impact['demand_change_pct']:+.1f}%")
print(f"  Days to Sellout: {impact['current_days_to_sellout']:.1f} → {impact['new_days_to_sellout']:.1f}")

---

工具与技术

决策支持平台

商业平台：

• IBM Decision Optimization：基于 CPLEX 的处方分析
• SAS Decision Manager：业务规则与优化
• Oracle Prescriptive Analytics：嵌入式优化
• Gurobi Optimizer：数学优化
• River Logic：处方分析平台

供应链专用：

• Blue Yonder：由 AI 驱动的供应链决策
• o9 Solutions：集成式规划与决策智能
• Kinaxis RapidResponse：提供建议的并行规划
• LLamasoft Supply Chain Guru：网络优化
• Coupa：采购与寻源建议

Python 库

优化：

• pulp：线性规划
• pyomo：优化建模
• cvxpy：凸优化
• scipy.optimize：科学优化
• OR-Tools (Google)：组合优化

机器学习：

• scikit-learn：ML 模型
• xgboost、lightgbm：梯度提升
• tensorflow/pytorch：深度学习

决策树：

• dtreeviz：决策树可视化
• sklearn.tree：决策树
• graphviz：图形可视化

可解释性：

• shap：用于模型解释的 SHAP 值
• lime：局部可解释的模型解释
• eli5：模型解释库

---

常见挑战与解决方案

挑战：对建议缺乏信任

问题：

• 用户不信任“黑箱”建议
• 对机器学习／优化持怀疑态度
• 更倾向于手动决策

解决方案：

• 提供清晰的解释（为什么给出此建议）
• 显示置信度
• 允许人工推翻建议，并建立反馈闭环
• 从决策支持入手，而非自动化
• 将权衡因素和替代方案可视化
• 通过 A/B 测试展示价值

挑战：模型漂移与条件变化

问题：

• 业务条件发生变化
• 模型逐渐过时
• 建议不再是最优选择

解决方案：

• 持续监控建议的质量
• 定期重新训练模型
• 根据用户反馈进行自适应学习
• 基于情景给出建议（假设分析）
• 采用人在回路验证

挑战：与现有系统集成

问题：

• 遗留系统不支持规范性分析
• 数据孤岛和数据质量问题
• IT 部门抵制新工具

解决方案：

• 采用 API 优先架构
• 从独立的决策支持工具入手
• 逐步与核心系统集成
• 推行数据质量改进计划
• 获得高管支持并做好变革管理

挑战：目标相互冲突

问题：

• 降低成本与提升服务水平之间的冲突
• 短期与长期优化之间的冲突
• 局部与全局优化之间的冲突

解决方案：

• 多目标优化（帕累托前沿）
• 根据利益相关方的意见进行加权评分
• 通过情景分析展示权衡因素
• 明确定义优先级
• 采用分层决策

---

输出格式

规范性分析报告

执行摘要：

• 关键建议
• 预期影响
• 置信度
• 所需操作

建议：

排名	操作	目标	预期影响	置信度	紧迫程度
1	订购 500 件 SKU_001	库存	避免 50K 美元的缺货损失	高	紧急
2	将 SKU_002 的价格降低 10%	销售	收入增加 25K 美元	中	高
3	更换为供应商 B	成本	每年降低 15K 美元	高	正常

详细分析：

针对每项建议：

• 操作：要做什么
• 理由：为什么（基于数据和模型）
• 预期结果：量化的影响
• 置信度：高／中／低及其原因
• 替代方案：考虑过的其他选项
• 风险：可能出现的问题
• 实施：如何执行

支持数据：

• 模型预测
• 敏感性分析
• 历史表现
• 假设

所需审批：

• 高价值决策
• 高风险决策
• 政策例外

---

需要提出的问题

如果需要更多背景信息：

1. 哪些决策需要规范性建议？
2. 目标是什么？（降低成本、提升服务水平，还是兼顾两者）
3. 可用于预测和优化的数据有哪些？
4. 谁会根据建议采取行动？（自动化执行还是由人类决策者执行）
5. 决策频率如何？（实时、每日、战略性）
6. 可解释性有多重要？（可以接受黑箱，还是必须可解释）
7. 有哪些约束？（预算、产能、政策）
8. 当前的决策流程是什么？

---

相关技能

• optimization-modeling：用于构建优化引擎
• ml-supply-chain：用于预测模型
• supply-chain-analytics：用于描述性和诊断性分析
• digital-twin-modeling：用于基于仿真的建议
• supply-chain-automation：用于自动执行决策
• demand-forecasting：用于需求预测
• inventory-optimization：用于库存建议
• route-optimization：用于路线规划建议
• supplier-selection：用于采购决策