SaaS 行业观察：AI 驱动的动态定价与货币化策略

class WillingnessToPayPredictor:
    def __init__(self):
        self.model = self.load_wtp_model()
        self.feature_engineer = WTPFeatureEngineer()
    
    def predict_wtp(self, customer, product):
        # 提取特征
        features = self.feature_engineer.extract_features(customer, product)
        
        # 预测支付意愿
        wtp_prediction = self.model.predict(features)
        
        # 生成置信区间
        confidence_interval = self.calculate_confidence_interval(features)
        
        # 识别影响支付意愿的关键因素
        key_factors = self.identify_key_factors(features)
        
        return {
            "wtp_mean": wtp_prediction["mean"],
            "wtp_range": confidence_interval,
            "confidence": wtp_prediction["confidence"],
            "key_factors": key_factors,
            "recommendations": self.generate_pricing_recommendations(wtp_prediction, key_factors)
        }
    
    def extract_features(self, customer, product):
        features = {
            # 客户特征
            "company_size": customer.company_size,
            "industry": customer.industry,
            "location": customer.location,
            "revenue": customer.annual_revenue,
            "growth_rate": customer.growth_rate,
            "tech_maturity": customer.technology_maturity_score,
            
            # 使用特征
            "usage_intensity": customer.usage_intensity,
            "feature_adoption": customer.feature_adoption_rate,
            "engagement_score": customer.engagement_score,
            "value_realized": customer.value_realized,
            "roi_achieved": customer.roi_achieved,
            
            # 行为特征
            "price_sensitivity": customer.price_sensitivity_score,
            "competitor_usage": customer.competitor_usage,
            "upgrade_history": customer.upgrade_history,
            "churn_risk": customer.churn_risk_score,
            
            # 产品特征
            "product_complexity": product.complexity_score,
            "value_proposition": product.value_score,
            "competitive_position": product.competitive_position,
            "market_demand": product.market_demand_score
        }
        
        return features

class WTPAnalysis:
    def analyze_wtp_segments(self, customers, product):
        # 按支付意愿分群
        segments = {
            "high_wtp": [],
            "medium_wtp": [],
            "low_wtp": []
        }
        
        for customer in customers:
            wtp = self.predict_wtp(customer, product)
            
            if wtp["wtp_mean"] > product.base_price * 1.5:
                segments["high_wtp"].append({
                    "customer": customer,
                    "wtp": wtp
                })
            elif wtp["wtp_mean"] > product.base_price * 0.8:
                segments["medium_wtp"].append({
                    "customer": customer,
                    "wtp": wtp
                })
            else:
                segments["low_wtp"].append({
                    "customer": customer,
                    "wtp": wtp
                })
        
        # 分析每个细分市场的特征
        segment_analysis = {}
        for segment_name, segment_customers in segments.items():
            segment_analysis[segment_name] = {
                "count": len(segment_customers),
                "avg_wtp": np.mean([c["wtp"]["wtp_mean"] for c in segment_customers]),
                "key_characteristics": self.identify_segment_characteristics(segment_customers),
                "pricing_strategy": self.recommend_segment_strategy(segment_name, segment_customers)
            }
        
        return segment_analysis

class DynamicPricingEngine:
    def __init__(self):
        self.wtp_predictor = WillingnessToPayPredictor()
        self.market_analyzer = MarketAnalyzer()
        self.competitor_monitor = CompetitorMonitor()
        self.elasticity_model = PriceElasticityModel()
    
    def calculate_optimal_price(self, customer, product, context):
        # 预测支付意愿
        wtp = self.wtp_predictor.predict_wtp(customer, product)
        
        # 分析市场条件
        market_conditions = self.market_analyzer.analyze_current_conditions()
        
        # 监控竞争对手价格
        competitor_prices = self.competitor_monitor.get_competitor_prices(product)
        
        # 计算价格弹性
        elasticity = self.elasticity_model.calculate_elasticity(customer, product)
        
        # 生成最优价格
        optimal_price = self.optimize_price(
            wtp=wtp,
            market_conditions=market_conditions,
            competitor_prices=competitor_prices,
            elasticity=elasticity,
            business_objectives=context["business_objectives"]
        )
        
        return {
            "base_price": optimal_price["base"],
            "discounts": optimal_price["discounts"],
            "final_price": optimal_price["final"],
            "rationale": optimal_price["rationale"],
            "expected_conversion": optimal_price["conversion_probability"],
            "expected_revenue": optimal_price["expected_revenue"],
            "confidence": optimal_price["confidence"]
        }
    
    def optimize_price(self, wtp, market_conditions, competitor_prices, elasticity, business_objectives):
        # 定义优化目标
        if business_objectives["priority"] == "revenue":
            objective = "maximize_revenue"
        elif business_objectives["priority"] == "conversion":
            objective = "maximize_conversion"
        elif business_objectives["priority"] == "profit":
            objective = "maximize_profit"
        else:
            objective = "balance"
        
        # 约束条件
        constraints = {
            "min_price": self.calculate_minimum_viable_price(),
            "max_price": wtp["wtp_range"]["upper"],
            "competitor_range": self.calculate_competitive_range(competitor_prices),
            "price_fairness": self.ensure_price_fairness(wtp)
        }
        
        # 优化算法
        if objective == "maximize_revenue":
            optimal = self.maximize_revenue(wtp, elasticity, constraints)
        elif objective == "maximize_conversion":
            optimal = self.maximize_conversion(wtp, elasticity, constraints)
        elif objective == "maximize_profit":
            optimal = self.maximize_profit(wtp, elasticity, constraints)
        else:
            optimal = self.balance_objectives(wtp, elasticity, constraints)
        
        return optimal
    
    def maximize_revenue(self, wtp, elasticity, constraints):
        # 收入 = 价格 × 转化概率
        
        # 搜索最优价格点
        price_range = np.linspace(constraints["min_price"], constraints["max_price"], 100)
        
        best_price = None
        best_revenue = 0
        
        for price in price_range:
            # 计算在该价格下的转化概率
            conversion_prob = self.calculate_conversion_probability(price, wtp, elasticity)
            
            # 计算预期收入
            expected_revenue = price * conversion_prob
            
            if expected_revenue > best_revenue:
                best_revenue = expected_revenue
                best_price = price
        
        return {
            "base": best_price,
            "discounts": [],
            "final": best_price,
            "rationale": f"Maximizes expected revenue at ${best_price}",
            "conversion_probability": self.calculate_conversion_probability(best_price, wtp, elasticity),
            "expected_revenue": best_revenue,
            "confidence": 0.85
        }

class PersonalizedDiscountSystem:
    def __init__(self):
        self.discount_rules = self.load_discount_rules()
        self.customer_analyzer = CustomerAnalyzer()
    
    def calculate_personalized_discount(self, customer, product, context):
        # 分析客户特征
        customer_profile = self.customer_analyzer.analyze(customer)
        
        # 评估折扣资格
        eligible_discounts = []
        
        # 忠诚度折扣
        if customer_profile["tenure"] > 24:  # 2年以上
            loyalty_discount = self.calculate_loyalty_discount(customer_profile)
            eligible_discounts.append({
                "type": "loyalty",
                "percentage": loyalty_discount,
                "rationale": f"Rewarding {customer_profile['tenure']} months of loyalty"
            })
        
        # 成长折扣（针对快速成长的初创公司）
        if customer_profile["growth_rate"] > 0.5:  # 年增长率>50%
            growth_discount = self.calculate_growth_discount(customer_profile)
            eligible_discounts.append({
                "type": "growth",
                "percentage": growth_discount,
                "rationale": "Supporting high-growth startups"
            })
        
        # 年度承诺折扣
        if context["commitment"] == "annual":
            annual_discount = 0.20  # 20%年度折扣
            eligible_discounts.append({
                "type": "annual_commitment",
                "percentage": annual_discount,
                "rationale": "Discount for annual commitment"
            })
        
        # 批量折扣
        if context["quantity"] > 10:
            volume_discount = self.calculate_volume_discount(context["quantity"])
            eligible_discounts.append({
                "type": "volume",
                "percentage": volume_discount,
                "rationale": f"Volume discount for {context['quantity']} seats"
            })
        
        # 计算总折扣（不能叠加过多）
        total_discount = self.calculate_total_discount(eligible_discounts)
        
        # 确保折扣不超过上限
        max_discount = 0.50  # 最多50%折扣
        total_discount = min(total_discount, max_discount)
        
        return {
            "total_discount": total_discount,
            "breakdown": eligible_discounts,
            "final_price": product.base_price * (1 - total_discount),
            "savings": product.base_price * total_discount,
            "valid_until": self.calculate_discount_validity()
        }
    
    def calculate_loyalty_discount(self, customer_profile):
        tenure = customer_profile["tenure"]
        
        if tenure >= 60:  # 5年以上
            return 0.30  # 30%折扣
        elif tenure >= 36:  # 3年以上
            return 0.20  # 20%折扣
        elif tenure >= 24:  # 2年以上
            return 0.15  # 15%折扣
        elif tenure >= 12:  # 1年以上
            return 0.10  # 10%折扣
        else:
            return 0.05  # 5%折扣
    
    def calculate_total_discount(self, discounts):
        # 不是简单相加，而是递减叠加
        # 例如：30% + 20% = 30% + (70% * 20%) = 30% + 14% = 44%
        
        total = 0
        remaining = 1.0
        
        # 按折扣大小排序，大的先应用
        sorted_discounts = sorted(discounts, key=lambda x: x["percentage"], reverse=True)
        
        for discount in sorted_discounts:
            total += remaining * discount["percentage"]
            remaining *= (1 - discount["percentage"])
        
        return total

class IntelligentPackagingSystem:
    def __init__(self):
        self.feature_analyzer = FeatureAnalyzer()
        self.usage_analyzer = UsageAnalyzer()
        self.cluster_model = CustomerClusteringModel()
    
    def generate_personalized_packages(self, customer):
        # 分析客户的使用模式
        usage_patterns = self.usage_analyzer.analyze_usage_patterns(customer)
        
        # 识别客户最重视的功能
        valued_features = self.feature_analyzer.identify_valued_features(customer)
        
        # 将客户聚类到相似客户群
        customer_cluster = self.cluster_model.predict_cluster(customer)
        
        # 生成个性化套餐
        packages = self.create_custom_packages(
            customer=customer,
            usage_patterns=usage_patterns,
            valued_features=valued_features,
            cluster=customer_cluster
        )
        
        return packages
    
    def create_custom_packages(self, customer, usage_patterns, valued_features, cluster):
        packages = []
        
        # 基础套餐：满足基本需求
        basic_package = {
            "name": "Starter",
            "description": "Perfect for getting started",
            "features": self.select_basic_features(valued_features),
            "limits": self.calculate_basic_limits(usage_patterns),
            "price": self.calculate_basic_price(customer, cluster),
            "recommended_for": "Individual users and small teams"
        }
        packages.append(basic_package)
        
        # 成长套餐：支持业务增长
        growth_package = {
            "name": "Growth",
            "description": "Scale your operations",
            "features": self.select_growth_features(valued_features, usage_patterns),
            "limits": self.calculate_growth_limits(usage_patterns),
            "price": self.calculate_growth_price(customer, cluster),
            "recommended_for": "Growing teams and businesses",
            "popular": True
        }
        packages.append(growth_package)
        
        # 专业套餐：高级功能
        pro_package = {
            "name": "Professional",
            "description": "Advanced features for power users",
            "features": self.select_pro_features(valued_features),
            "limits": self.calculate_pro_limits(usage_patterns),
            "price": self.calculate_pro_price(customer, cluster),
            "recommended_for": "Large teams and enterprises"
        }
        packages.append(pro_package)
        
        # 定制套餐：完全个性化
        custom_package = {
            "name": "Custom",
            "description": "Tailored to your exact needs",
            "features": "All features",
            "limits": "Custom limits",
            "price": "Contact us",
            "recommended_for": "Enterprises with specific requirements"
        }
        packages.append(custom_package)
        
        # 为每个套餐添加个性化说明
        for package in packages:
            package["personalization"] = self.generate_personalization_notes(package, customer)
            package["value_highlights"] = self.highlight_value(package, customer)
            package["roi_projection"] = self.project_roi(package, customer)
        
        return packages
    
    def select_basic_features(self, valued_features):
        # 选择最核心的功能
        core_features = [f for f in valued_features if f["importance"] > 0.8]
        
        # 限制数量，保持简洁
        return core_features[:5]
    
    def calculate_basic_limits(self, usage_patterns):
        # 基于当前使用量的 120% 设置限制
        limits = {}
        
        for metric, current_usage in usage_patterns["current"].items():
            limits[metric] = int(current_usage * 1.2)
        
        return limits

class PackageRecommendationEngine:
    def __init__(self):
        self.package_analyzer = PackageAnalyzer()
        self.customer_predictor = CustomerValuePredictor()
    
    def recommend_package(self, customer, packages):
        # 为每个套餐计算匹配度
        package_scores = []
        
        for package in packages:
            score = self.calculate_package_score(customer, package)
            package_scores.append({
                "package": package,
                "score": score["overall"],
                "breakdown": score["breakdown"],
                "rationale": score["rationale"]
            })
        
        # 排序
        sorted_packages = sorted(package_scores, key=lambda x: x["score"], reverse=True)
        
        # 选择最佳推荐
        best_package = sorted_packages[0]
        
        # 生成推荐说明
        recommendation = {
            "recommended_package": best_package["package"],
            "confidence": best_package["score"],
            "rationale": best_package["rationale"],
            "alternative_options": sorted_packages[1:3],
            "upgrade_path": self.suggest_upgrade_path(customer, best_package["package"]),
            "expected_value": self.calculate_expected_value(customer, best_package["package"])
        }
        
        return recommendation
    
    def calculate_package_score(self, customer, package):
        scores = {}
        
        # 功能匹配度（40%权重）
        feature_match = self.calculate_feature_match(customer, package)
        scores["feature_match"] = feature_match * 0.4
        
        # 价格合理性（30%权重）
        price_fit = self.calculate_price_fit(customer, package)
        scores["price_fit"] = price_fit * 0.3
        
        # 成长空间（20%权重）
        growth_room = self.calculate_growth_room(customer, package)
        scores["growth_room"] = growth_room * 0.2
        
        # 价值实现速度（10%权重）
        value_speed = self.calculate_value_speed(customer, package)
        scores["value_speed"] = value_speed * 0.1
        
        # 计算总分
        overall_score = sum(scores.values())
        
        # 生成说明
        rationale = self.generate_rationale(customer, package, scores)
        
        return {
            "overall": overall_score,
            "breakdown": scores,
            "rationale": rationale
        }
    
    def calculate_feature_match(self, customer, package):
        # 分析客户需要的功能
        needed_features = self.identify_needed_features(customer)
        
        # 分析套餐提供的功能
        provided_features = package["features"]
        
        # 计算匹配度
        match_count = len(set(needed_features) & set(provided_features))
        total_needed = len(needed_features)
        
        if total_needed == 0:
            return 1.0
        
        return match_count / total_needed
    
    def calculate_price_fit(self, customer, package):
        # 预测客户的支付意愿
        wtp = self.predict_wtp(customer, package)
        
        # 计算价格与支付意愿的比率
        price_ratio = package["price"] / wtp
        
        # 理想比率是 0.7-0.9（客户觉得物有所值）
        if 0.7 <= price_ratio <= 0.9:
            return 1.0
        elif price_ratio < 0.7:
            # 价格太低，可能被认为质量不高
            return 0.8
        else:
            # 价格太高
            return max(0, 1.0 - (price_ratio - 0.9) * 2)

class RevenueOptimizationSystem:
    def __init__(self):
        self.pricing_engine = DynamicPricingEngine()
        self.churn_predictor = ChurnPredictor()
        self.upsell_recommender = UpsellRecommender()
        self.ltv_calculator = LTVCalculator()
    
    def optimize_revenue(self, customer, current_plan):
        # 预测客户生命周期价值
        current_ltv = self.ltv_calculator.calculate_ltv(customer, current_plan)
        
        # 预测流失风险
        churn_risk = self.churn_predictor.predict_churn_risk(customer)
        
        # 识别追加销售机会
        upsell_opportunities = self.upsell_recommender.identify_opportunities(customer)
        
        # 生成收入优化策略
        strategies = []
        
        # 策略1：如果流失风险高，提供保留优惠
        if churn_risk["probability"] > 0.3:
            retention_offer = self.generate_retention_offer(customer, churn_risk)
            strategies.append({
                "type": "retention",
                "action": retention_offer,
                "expected_impact": self.calculate_retention_impact(customer, retention_offer),
                "priority": "high" if churn_risk["probability"] > 0.5 else "medium"
            })
        
        # 策略2：如果有追加销售机会，推荐升级
        if upsell_opportunities:
            for opportunity in upsell_opportunities[:2]:  # 最多推荐2个
                strategies.append({
                    "type": "upsell",
                    "action": opportunity,
                    "expected_impact": self.calculate_upsell_impact(customer, opportunity),
                    "priority": "medium"
                })
        
        # 策略3：优化当前定价
        pricing_optimization = self.optimize_current_pricing(customer, current_plan)
        if pricing_optimization["recommended_change"]:
            strategies.append({
                "type": "pricing_adjustment",
                "action": pricing_optimization,
                "expected_impact": pricing_optimization["impact"],
                "priority": "low"
            })
        
        # 计算总体收入影响
        total_revenue_impact = sum(s["expected_impact"]["revenue_delta"] for s in strategies)
        
        return {
            "current_ltv": current_ltv,
            "optimized_ltv": current_ltv + total_revenue_impact,
            "strategies": strategies,
            "total_revenue_impact": total_revenue_impact,
            "implementation_timeline": self.generate_implementation_timeline(strategies)
        }
    
    def generate_retention_offer(self, customer, churn_risk):
        # 根据流失风险和客户价值生成保留优惠
        
        customer_value = self.ltv_calculator.calculate_ltv(customer)
        
        # 高价值客户获得更好的优惠
        if customer_value > 10000:  # LTV > $10,000
            discount = 0.30  # 30%折扣
            additional_benefits = ["priority_support", "extended_trial"]
        elif customer_value > 5000:
            discount = 0.20  # 20%折扣
            additional_benefits = ["priority_support"]
        else:
            discount = 0.10  # 10%折扣
            additional_benefits = []
        
        offer = {
            "type": "retention_discount",
            "discount_percentage": discount,
            "duration": "3 months",
            "additional_benefits": additional_benefits,
            "conditions": [
                "Must commit to 12-month contract",
                "Offer valid for 7 days"
            ],
            "estimated_retention_probability": min(0.9, churn_risk["probability"] + 0.4)
        }
        
        return offer
    
    def calculate_retention_impact(self, customer, retention_offer):
        # 计算保留优惠的收入影响
        
        current_mrr = customer.current_mrr
        discounted_mrr = current_mrr * (1 - retention_offer["discount_percentage"])
        
        # 如果不提供优惠，客户可能流失
        churn_probability = self.churn_predictor.predict_churn_risk(customer)["probability"]
        
        # 提供优惠后的保留概率
        retention_probability = retention_offer["estimated_retention_probability"]
        
        # 预期收入影响
        # = (保留概率 × 折扣后MRR × 12个月) - (流失概率 × 0)
        expected_revenue = retention_probability * discounted_mrr * 12
        
        # 对比不采取行动的情况
        baseline_revenue = (1 - churn_probability) * current_mrr * 12
        
        revenue_delta = expected_revenue - baseline_revenue
        
        return {
            "revenue_delta": revenue_delta,
            "retention_probability": retention_probability,
            "discounted_mrr": discounted_mrr,
            "roi": revenue_delta / (current_mrr * retention_offer["discount_percentage"] * 3)
        }