套餐设计

场景

系统运行稳定，用户持续增长。

六个月后：

• 注册用户：3000 个
• 日活用户：800 个
• 日调用量：60 万次

但运营团队找到我：

当前问题：
- 服务器成本：¥2000/月
- 外部 API 成本：¥1000/月
- 总成本：¥3000/月
- 收入：¥0/月

我们需要尽快实现盈利！

商业化需求

运营团队提出了需求：

目标：
1. 设计差异化套餐
2. 吸引部分用户付费
3. 覆盖服务器成本
4. 最终实现盈利

用户分层：
- 个人开发者：用量小，对价格敏感
- 初创公司：用量中等，需要稳定性
- 大型企业：用量大，需要专属服务

套餐设计

我调研了竞品的定价策略：

竞品分析

竞品 A 定价：
- 免费版：100 次/天
- 基础版：¥99/月，10000 次/天
- 专业版：¥499/月，100000 次/天
- 企业版：联系销售

竞品 B 定价：
- 免费版：1000 次/月
- 基础版：¥49/月，50000 次/月
- 专业版：¥199/月，500000 次/月
- 企业版：¥999/月，无限制

我的定价策略

基于成本分析和竞品调研，我设计了这样的套餐：

数据库设计

套餐表

CREATE TABLE pricing_plans (
  id INT PRIMARY KEY AUTO_INCREMENT,
  name VARCHAR(50) NOT NULL,  -- free, basic, pro, enterprise
  display_name VARCHAR(100) NOT NULL,
  price_monthly DECIMAL(10, 2) DEFAULT 0,
  price_yearly DECIMAL(10, 2) DEFAULT 0,

  -- 限额配置
  daily_limit INT DEFAULT 1000,
  rate_limit_per_second INT DEFAULT 1,
  burst_capacity INT DEFAULT 10,

  -- 功能配置
  max_apis INT DEFAULT 3,
  data_retention_days INT DEFAULT 7,

  -- 支持配置
  support_type VARCHAR(50) DEFAULT 'community',  -- community, email, priority, dedicated
  sla_guarantee DECIMAL(5, 4) DEFAULT NULL,

  -- 其他
  custom_domain BOOLEAN DEFAULT FALSE,
  features JSON,  -- 其他功能配置

  created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
  UNIQUE KEY unique_name (name)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

-- 插入套餐数据
INSERT INTO pricing_plans (name, display_name, price_monthly, price_yearly, daily_limit, rate_limit_per_second, burst_capacity, max_apis, data_retention_days, support_type, sla_guarantee, custom_domain) VALUES
('free', '免费版', 0, 0, 1000, 1, 10, 3, 7, 'community', NULL, FALSE),
('basic', '基础版', 99, 990, 10000, 10, 100, 10, 30, 'email', 0.995, FALSE),
('pro', '专业版', 499, 4990, 100000, 100, 1000, -1, 90, 'priority', 0.999, TRUE),
('enterprise', '企业版', NULL, NULL, -1, 1000, 10000, -1, -1, 'dedicated', 0.9995, TRUE);

用户订阅表

CREATE TABLE user_subscriptions (
  id INT PRIMARY KEY AUTO_INCREMENT,
  user_id INT NOT NULL,
  plan_id INT NOT NULL,

  -- 计费周期
  billing_cycle VARCHAR(20) DEFAULT 'monthly',  -- monthly, yearly

  -- 订阅状态
  status VARCHAR(20) DEFAULT 'active',  -- active, cancelled, expired, past_due

  -- 时间
  current_period_start TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
  current_period_end TIMESTAMP NOT NULL,
  cancel_at_period_end BOOLEAN DEFAULT FALSE,

  -- 支付信息
  payment_method_id VARCHAR(255),

  created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
  updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,

  UNIQUE KEY unique_user_subscription (user_id),
  FOREIGN KEY (user_id) REFERENCES users(id),
  FOREIGN KEY (plan_id) REFERENCES pricing_plans(id),
  INDEX idx_status (status),
  INDEX idx_period_end (current_period_end)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

套餐管理 API

获取套餐列表

@app.route('/api/pricing/plans')
def get_pricing_plans():
    """获取所有套餐"""

    with get_db_connection() as conn:
        cursor = conn.cursor()
        cursor.execute('SELECT * FROM pricing_plans ORDER BY price_monthly ASC')
        plans = cursor.fetchall()

        # 格式化返回
        result = []
        for plan in plans:
            result.append({
                'id': plan['id'],
                'name': plan['name'],
                'display_name': plan['display_name'],
                'price_monthly': float(plan['price_monthly']),
                'price_yearly': float(plan['price_yearly']),
                'limits': {
                    'daily_limit': plan['daily_limit'],
                    'rate_limit_per_second': plan['rate_limit_per_second'],
                    'burst_capacity': plan['burst_capacity'],
                    'max_apis': plan['max_apis'],
                    'data_retention_days': plan['data_retention_days']
                },
                'features': {
                    'support_type': plan['support_type'],
                    'sla_guarantee': float(plan['sla_guarantee']) if plan['sla_guarantee'] else None,
                    'custom_domain': plan['custom_domain']
                }
            })

        return jsonify(result)

获取用户当前套餐

@app.route('/api/pricing/my-plan')
@require_api_key
def get_my_plan():
    """获取用户当前套餐"""

    with get_db_connection() as conn:
        cursor = conn.cursor()

        # 获取用户订阅
        cursor.execute(
            '''SELECT us.*, p.* FROM user_subscriptions us
               JOIN pricing_plans p ON us.plan_id = p.id
               WHERE us.user_id = ? AND us.status = 'active'
               ORDER BY us.created_at DESC
               LIMIT 1''',
            (request.user['id'],)
        )

        subscription = cursor.fetchone()

        if not subscription:
            # 默认免费版
            return jsonify({
                'plan': 'free',
                'status': 'active',
                'current_period_end': None
            })

        return jsonify({
            'plan': subscription['name'],
            'display_name': subscription['display_name'],
            'status': subscription['status'],
            'current_period_start': subscription['current_period_start'].isoformat(),
            'current_period_end': subscription['current_period_end'].isoformat(),
            'billing_cycle': subscription['billing_cycle'],
            'cancel_at_period_end': subscription['cancel_at_period_end']
        })

升级/降级套餐

@app.route('/api/pricing/change-plan', methods=['POST'])
@require_api_key
def change_plan():
    """更改套餐"""

    data = request.get_json()
    new_plan_name = data.get('plan')
    billing_cycle = data.get('billing_cycle', 'monthly')

    # 验证套餐是否存在
    with get_db_connection() as conn:
        cursor = conn.cursor()

        cursor.execute(
            'SELECT * FROM pricing_plans WHERE name = ?',
            (new_plan_name,)
        )
        new_plan = cursor.fetchone()

        if not new_plan:
            return jsonify({'error': 'Invalid plan'}), 400

        if new_plan['name'] == 'enterprise':
            return jsonify({'error': 'Please contact sales for enterprise plan'}), 400

        # 获取用户当前订阅
        cursor.execute(
            '''SELECT * FROM user_subscriptions
               WHERE user_id = ? AND status = 'active'
               ORDER BY created_at DESC
               LIMIT 1''',
            (request.user['id'],)
        )
        current_subscription = cursor.fetchone()

        if current_subscription:
            current_plan_id = current_subscription['plan_id']

            cursor.execute(
                'SELECT * FROM pricing_plans WHERE id = ?',
                (current_plan_id,)
            )
            current_plan = cursor.fetchone()

            # 检查是否降级
            if new_plan['price_monthly'] < current_plan['price_monthly']:
                # 降级：本期结束生效
                cursor.execute(
                    '''UPDATE user_subscriptions
                       SET plan_id = ?, cancel_at_period_end = FALSE
                       WHERE user_id = ? AND status = 'active'
                       ORDER BY created_at DESC
                       LIMIT 1''',
                    (new_plan['id'], request.user['id'])
                )

                # 创建新的订阅记录（下期生效）
                cursor.execute(
                    '''INSERT INTO user_subscriptions
                       (user_id, plan_id, billing_cycle, status, current_period_start, current_period_end)
                       VALUES (?, ?, ?, 'pending', ?, ?)''',
                    (
                        request.user['id'],
                        new_plan['id'],
                        billing_cycle,
                        current_subscription['current_period_end'],
                        current_subscription['current_period_end'] + timedelta(days=30)
                    )
                )

                conn.commit()

                return jsonify({
                    'message': 'Plan will be downgraded at the end of current period',
                    'effective_date': current_subscription['current_period_end'].isoformat()
                })

            else:
                # 升级：立即生效
                cursor.execute(
                    '''UPDATE user_subscriptions
                       SET plan_id = ?, billing_cycle = ?
                       WHERE user_id = ? AND status = 'active'
                       ORDER BY created_at DESC
                       LIMIT 1''',
                    (new_plan['id'], billing_cycle, request.user['id'])
                )

                conn.commit()

                # 更新缓存
                update_user_plan_cache(request.user['id'], new_plan['name'])

                return jsonify({
                    'message': 'Plan upgraded successfully',
                    'effective': 'immediately'
                })

套餐对比页面

前端展示

<!-- pricing-plans.html -->
<div class="pricing-plans">
  <div class="plan free">
    <h3>免费版</h3>
    <div class="price">¥0/月</div>
    <ul class="features">
      <li>✅ 1000 次/天</li>
      <li>✅ 1 次/秒</li>
      <li>✅ 3 个 API</li>
      <li>❌ 无技术支持</li>
    </ul>
    <button class="select-plan" data-plan="free">当前套餐</button>
  </div>

  <div class="plan basic">
    <h3>基础版</h3>
    <div class="price">¥99/月</div>
    <ul class="features">
      <li>✅ 10000 次/天</li>
      <li>✅ 10 次/秒</li>
      <li>✅ 10 个 API</li>
      <li>✅ 邮件支持</li>
    </ul>
    <button class="select-plan" data-plan="basic">升级</button>
  </div>

  <div class="plan pro">
    <h3>专业版</h3>
    <div class="price">¥499/月</div>
    <ul class="features">
      <li>✅ 100000 次/天</li>
      <li>✅ 100 次/秒</li>
      <li>✅ 所有 API</li>
      <li>✅ 优先支持</li>
    </ul>
    <button class="select-plan" data-plan="pro">升级</button>
  </div>

  <div class="plan enterprise">
    <h3>企业版</h3>
    <div class="price">联系销售</div>
    <ul class="features">
      <li>✅ 无限制</li>
      <li>✅ 专属经理</li>
      <li>✅ SLA 保证</li>
      <li>✅ 定制服务</li>
    </ul>
    <button class="contact-sales">联系销售</button>
  </div>
</div>

限流集成

根据套餐设置限流

def get_user_plan_limits(user_id):
    """获取用户套餐限额"""

    # 先查缓存
    cache_key = f'user_plan:{user_id}'
    cached = redis_client.get(cache_key)

    if cached:
        return json.loads(cached)

    # 查询数据库
    with get_db_connection() as conn:
        cursor = conn.cursor()

        cursor.execute(
            '''SELECT p.* FROM user_subscriptions us
               JOIN pricing_plans p ON us.plan_id = p.id
               WHERE us.user_id = ? AND us.status = 'active'
               ORDER BY us.created_at DESC
               LIMIT 1''',
            (user_id,)
        )

        plan = cursor.fetchone()

        if not plan:
            # 默认免费版
            limits = {
                'plan': 'free',
                'rate_limit': 1,
                'burst_capacity': 10
            }
        else:
            limits = {
                'plan': plan['name'],
                'rate_limit': plan['rate_limit_per_second'],
                'burst_capacity': plan['burst_capacity']
            }

        # 存入缓存（1 小时）
        redis_client.setex(cache_key, 3600, json.dumps(limits))

        return limits

# 在限流中间件中使用
@app.route('/api/weather')
@require_api_key
def get_weather():
    # 获取用户套餐限额
    limits = get_user_plan_limits(request.user['id'])

    # 检查限流
    result = rate_limiter.is_allowed(
        request.user['id'],
        limits['rate_limit'],
        limits['burst_capacity']
    )

    if not result['allowed']:
        return jsonify({
            'error': 'Rate limit exceeded',
            'plan': limits['plan'],
            'retry_after': 1
        }), 429

    # ... 业务逻辑

本节小结

✅ 完成的工作：

• 设计了 4 个套餐（免费版、基础版、专业版、企业版）
• 创建了套餐和订阅数据表
• 实现了套餐管理 API
• 集成了套餐限流机制

✅ 商业价值：

• 明确了定价策略
• 差异化了免费和付费用户
• 为后续支付和计费打下基础

⚠️ 下一步： 我需要实现支付和账单系统

🎯 下一步： 月底需要生成账单，如何统计月度调用量？