monetization

用途

该技能用于数字产品的变现策略与实现，涵盖 Stripe 集成、订阅、定价实验、freemium、升级流程、流失预防、收入优化以及 SaaS 商业模式。当任务涉及集成 Stripe、创建订阅计划、定价策略、升级/降级、支付 webhook、免费试用、流失、LTV/CAC、单位经济模型或商业模式时，可启用该技能。

不适用场景：任务与变现无关；有更简单、更具体的工具可处理请求；用户需要通用协助而非领域专业知识。

主要能力

Stripe 集成：安装依赖、配置 API 密钥与 webhook 密钥、定义计划价格 ID。
客户与订阅管理：创建 Customer、创建 Subscription（支持试用期、默认不完整支付行为、展开最新发票的支付意图）。
Checkout Session：推荐用于提升转化，支持订阅模式、试用期、成功/取消 URL、允许促销码。
客户门户：创建自助服务门户会话。
Webhook 处理：使用 FastAPI 接收并验证 Stripe 事件，处理订阅创建/更新/删除、支付成功/失败、试用即将结束等事件。
订阅状态查询：列出客户订阅并返回层级、状态、试用结束、当前周期结束、是否周期末取消等信息。
定价框架：基于价值的定价（计算经济价值、捕获 10-30%、验证支付意愿、A/B 测试三个价格点）与竞争锚定。
定价心理学：左位数效应、年付折扣、视觉层级、锚定、试用是否需信用卡、"最受欢迎"徽章。
计划结构：三档计划（Free/Pro/Business）示例，含价格、对话数、记忆、专家面板、多用户、API 访问、支持等维度。
流失预防：高/中风险流失信号、反流失序列（第 0/3/7/14/30 天动作）、退出调查原因列表。
单位经济计算：ARPU、流失率、LTV、CAC、LTV/CAC、回收期及健康状态判断。
巴西 B2C SaaS 基准：月度流失、LTV/CAC、回收期、试用转付费、月环比增长的分档参考。
收入仪表盘：MRR 构成（新增、扩展、收缩、流失、净新增）、ARR、流失率、净收入留存。
用量触发升级：当用量接近限额时发送升级提示。
快捷命令：/stripe-setup、/pricing-analysis、/churn-playbook、/unit-economics、/upgrade-flow、/revenue-dashboard、/trial-optimization。

使用方式

安装依赖：

pip install stripe

或

npm install stripe

配置示例：

import stripe
import os

stripe.api_key = os.environ["STRIPE_SECRET_KEY"]
STRIPE_WEBHOOK_SECRET = os.environ["STRIPE_WEBHOOK_SECRET"]

PLANS = {
    "free": None,
    "pro": os.environ["STRIPE_PRICE_PRO"],
    "business": os.environ["STRIPE_PRICE_BIZ"],
}

创建客户与订阅：

def create_customer(email: str, name: str, user_id: str) -> str:
    customer = stripe.Customer.create(
        email=email,
        name=name,
        metadata={"user_id": user_id}
    )
    return customer.id

def create_subscription(customer_id: str, price_id: str, trial_days: int = 14):
    subscription = stripe.Subscription.create(
        customer=customer_id,
        items=[{"price": price_id}],
        trial_period_days=trial_days,
        payment_behavior="default_incomplete",
        expand=["latest_invoice.payment_intent"],
    )
    return {
        "subscription_id": subscription.id,
        "client_secret": subscription.latest_invoice.payment_intent.client_secret,
        "status": subscription.status
    }

Checkout Session：

def create_checkout_session(
    customer_id: str,
    price_id: str,
    success_url: str,
    cancel_url: str,
    trial_days: int = 14
) -> str:
    session = stripe.checkout.Session.create(
        customer=customer_id,
        mode="subscription",
        line_items=[{"price": price_id, "quantity": 1}],
        subscription_data={"trial_period_days": trial_days},
        success_url=success_url + "?session_id={CHECKOUT_SESSION_ID}",
        cancel_url=cancel_url,
        allow_promotion_codes=True,
    )
    return session.url

客户门户：

def create_portal_session(customer_id: str, return_url: str) -> str:
    session = stripe.billing_portal.Session.create(
        customer=customer_id,
        return_url=return_url,
    )
    return session.url

Webhook 处理：

from fastapi import Request, HTTPException
import stripe

async def stripe_webhook(request: Request):
    payload = await request.body()
    sig_header = request.headers.get("stripe-signature")

    try:
        event = stripe.Webhook.construct_event(
            payload, sig_header, STRIPE_WEBHOOK_SECRET
        )
    except ValueError:
        raise HTTPException(status_code=400, detail="Invalid payload")
    except stripe.error.SignatureVerificationError:
        raise HTTPException(status_code=400, detail="Invalid signature")

    handlers = {
        "customer.subscription.created": handle_subscription_created,
        "customer.subscription.updated": handle_subscription_updated,
        "customer.subscription.deleted": handle_subscription_deleted,
        "invoice.payment_succeeded": handle_payment_succeeded,
        "invoice.payment_failed": handle_payment_failed,
        "customer.subscription.trial_will_end": handle_trial_ending,
    }

    handler = handlers.get(event["type"])
    if handler:
        await handler(event["data"]["object"])

    return {"status": "ok"}

查询订阅状态：

def get_subscription_status(customer_id: str) -> dict:
    subscriptions = stripe.Subscription.list(
        customer=customer_id,
        status="all",
        limit=1
    )
    if not subscriptions.data:
        return {"tier": "free", "status": "none"}

    sub = subscriptions.data[0]
    return {
        "tier": get_tier_from_price(sub.items.data[0].price.id),
        "status": sub.status,
        "trial_end": sub.trial_end,
        "current_period_end": sub.current_period_end,
        "cancel_at_period_end": sub.cancel_at_period_end,
    }

适用场景

为数字产品接入 Stripe 支付与订阅。
设计 SaaS 定价策略与计划结构。
优化试用转化与升级流程。
识别并干预流失风险用户。
计算 LTV/CAC 等单位经济指标并监控收入健康度。
搭建收入仪表盘与用量触发升级机制。

限制

仅在任务明确匹配上述范围时使用。
不应将输出视为环境特定验证、测试或专家评审的替代。
若缺少必要输入、权限、安全边界或成功标准，应停止并请求澄清。
资料未提供具体收费、价格或平台支持范围等额外信息。

技能介绍

用途

主要能力

使用方式

适用场景

限制