付费自定义
第一笔收入 💰
收到第一笔 ¥99 的付款时，我激动得截了屏。
终于不再只是烧钱了。
那天是周二的晚上，我正在看服务器账单——每个月 ¥5000 的成本像一块石头压在胸口。突然手机震动了一下，支付宝到账提醒：¥99。
我赶紧打开后台，看到是一个深圳的用户，购买了”专业套餐”，订阅了 3 个自定义别名。他在备注里写道：“太需要这个功能了，公司品牌推广必备！”
那一刻我意识到：这不是玩具，是生意。
套餐设计 📊
为什么收费？
先看看当时的数据：
优质别名申请统计（2026年2月）：
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
"apple" 申请次数：128次
"shop" 申请次数：256次
"go" 申请次数：189次
"news" 申请次数：167次
"vip" 申请次数：143次
3字符短别名：100%被占用
4字符好记词：87%被占用
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
用户留言精选：
• "我愿意付钱买'apple'这个别名！"
• "能不能租给我？我用一个月就行"
• "我们公司需要这个品牌名，价格好商量"
• "求转让'go'，¥100/月可以接受"需求明确，供给稀缺——完美的定价场景。
三档套餐
我设计了三个套餐，定价逻辑是：
class PricingModel:
"""定价模型"""
def __init__(self):
self.tiers = {
'free': {
'name': '免费版',
'price': 0,
'custom_aliases': 0,
'alias_length': 6, # 自动生成，6位以上
'statistics': False,
'priority': 0,
'target': '个人尝鲜用户'
},
'basic': {
'name': '基础版',
'price': 29,
'custom_aliases': 5,
'alias_length': 4, # 最短4位
'statistics': True,
'priority': 1,
'target': '小团队/个人品牌'
},
'pro': {
'name': '专业版',
'price': 99,
'custom_aliases': -1, # 无限
'alias_length': 3, # 最短3位
'statistics': True,
'priority': 2,
'target': '企业用户/重度使用者'
}
}
def get_recommended_tier(self, user):
"""根据用户情况推荐套餐"""
# 新用户：推荐免费版
if user['created_days'] < 7:
return 'free'
# 使用量大的用户：推荐专业版
if user['total_links'] > 50:
return 'pro'
# 有自定义需求的：推荐基础版
if user['alias_applications'] > 0:
return 'basic'
return 'free'
def calculate_annual_discount(self, monthly_price, months):
"""年付折扣"""
if months >= 12:
return monthly_price * 12 * 0.7 # 年付7折
elif months >= 6:
return monthly_price * months * 0.85 # 半年付85折
else:
return monthly_price * months # 月付无折扣
# 定价展示
pricing = PricingModel()
print("=== 短链接服务套餐 ===\n")
for tier_id, tier in pricing.tiers.items():
print(f"【{tier['name']}】")
print(f"价格：¥{tier['price']}/月")
print(f"自定义别名：{'无限' if tier['custom_aliases'] == -1 else tier['custom_aliases']}个")
print(f"最短长度：{tier['alias_length']}位")
print(f"数据统计：{'✓' if tier['statistics'] else '✗'}")
print(f"适用：{tier['target']}")
print()输出：
=== 短链接服务套餐 ===
【免费版】
价格：¥0/月
自定义别名：0个
最短长度：6位
数据统计：✗
适用：个人尝鲜用户
【基础版】
价格：¥29/月
自定义别名：5个
最短长度：4位
数据统计：✓
适用：小团队/个人品牌
【专业版】
价格：¥99/月
自定义别名：无限
最短长度：3位
数据统计：✓
适用：企业用户/重度使用者定价逻辑
为什么是 ¥29 和 ¥99？
心理定价
- ¥29：低于 ¥30，感觉”不到30块”
- ¥99：低于三位数，感觉”不到100块”
锚定效应
- 免费版作为锚点，让基础版显得很便宜
- 专业版¥99与基础版¥29对比，¥70差价换无限别名，值得
成本覆盖
- 单个用户服务器成本：¥5/月
- ¥29 套餐毛利：83%
- ¥99 套餐毛利：95%
转化漏斗
100个免费用户 ↓ 2%转化 2个付费用户 ↓ 分配 1.5个基础版（¥29） + 0.5个专业版（¥99） ↓ 平均客单价：¥46.5
支付接入 💳
为什么选择支付宝+微信支付？
我当时考虑了三个因素：
- 用户习惯：国内用户习惯支付宝/微信
- 接入难度：官方SDK文档完善
- 费率：0.6%交易费，可接受
支付流程
用户选择套餐 → 创建订阅订单 → 调用支付接口 → 跳转支付页面
↓
用户完成支付
↓
支付平台回调通知
↓
验证签名 → 激活订阅 → 创建别名代码实现
from datetime import datetime, timedelta
import hashlib
import hmac
import json
class PaymentService:
"""支付服务"""
def __init__(self):
self.alipay_config = {
'app_id': '2022xxxxxxxx',
'private_key': open('alipay_private_key.pem').read(),
'public_key': open('alipay_public_key.pem').read(),
'gateway': 'https://openapi.alipay.com/gateway.do'
}
self.wechat_config = {
'appid': 'wxXXXXXXXX',
'mch_id': '1XXXXXXXX',
'api_key': 'xxxxxxxxxxxxxxxx',
'gateway': 'https://api.mch.weixin.qq.com/pay/unifiedorder'
}
def create_payment(self, user_id, tier, months, payment_method):
"""创建支付订单"""
# 1. 创建订阅记录
subscription = self._create_subscription(user_id, tier, months)
# 2. 创建支付订单
order = {
'order_id': self._generate_order_id(),
'subscription_id': subscription['id'],
'amount': subscription['total_price'],
'subject': f"短链接{subscription['tier_name']} - {months}个月",
'user_id': user_id
}
# 3. 调用对应支付接口
if payment_method == 'alipay':
payment_url = self._create_alipay_payment(order)
elif payment_method == 'wechat':
payment_url = self._create_wechat_payment(order)
else:
return {'error': '不支持的支付方式'}
# 4. 保存订单
db.execute("""
INSERT INTO payment_orders
(order_id, subscription_id, user_id, amount,
payment_method, status, created_at)
VALUES (?, ?, ?, ?, ?, 'pending', ?)
""", (
order['order_id'], subscription['id'], user_id,
order['amount'], payment_method, datetime.now()
))
return {
'order_id': order['order_id'],
'payment_url': payment_url,
'amount': order['amount'],
'qr_code': self._generate_qr_code(payment_url)
}
def _create_subscription(self, user_id, tier, months):
"""创建订阅记录"""
pricing = PricingModel()
tier_config = pricing.tiers[tier]
# 计算折扣
monthly_price = tier_config['price']
total_price = pricing.calculate_annual_discount(monthly_price, months)
subscription = {
'id': generate_id(),
'user_id': user_id,
'tier': tier,
'tier_name': tier_config['name'],
'months': months,
'monthly_price': monthly_price,
'total_price': total_price,
'start_date': datetime.now(),
'end_date': datetime.now() + timedelta(days=30*months),
'status': 'pending_payment'
}
db.execute("""
INSERT INTO subscriptions
(id, user_id, tier, tier_name, months,
monthly_price, total_price, start_date, end_date, status)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", (
subscription['id'], subscription['user_id'], subscription['tier'],
subscription['tier_name'], subscription['months'],
subscription['monthly_price'], subscription['total_price'],
subscription['start_date'], subscription['end_date'],
subscription['status']
))
return subscription
def _create_alipay_payment(self, order):
"""创建支付宝支付"""
import urllib.parse
# 构造请求参数
biz_content = {
'out_trade_no': order['order_id'],
'product_code': 'FAST_INSTANT_TRADE_PAY',
'total_amount': str(order['amount']),
'subject': order['subject']
}
# 生成签名
params = {
'app_id': self.alipay_config['app_id'],
'method': 'alipay.trade.page.pay',
'charset': 'utf-8',
'sign_type': 'RSA2',
'timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
'version': '1.0',
'biz_content': json.dumps(biz_content)
}
# 签名
sign = self._sign_params(params, self.alipay_config['private_key'])
params['sign'] = sign
# 生成支付URL
url = self.alipay_config['gateway'] + '?' + urllib.parse.urlencode(params)
return url
def _sign_params(self, params, private_key):
"""签名参数"""
from cryptography.hazmat.primitives import hashes
from cryptography.hazat.primitives.asymmetric import padding
from cryptography.hazmat.backends import default_backend
# 排序参数
sorted_params = sorted(params.items())
query_string = '&'.join([f"{k}={v}" for k, v in sorted_params if k != 'sign'])
# RSA签名
private_key_obj = load_pem_private_key(
private_key.encode(),
password=None,
backend=default_backend()
)
signature = private_key_obj.sign(
query_string.encode('utf-8'),
padding.PKCS1v15(),
hashes.SHA256()
)
return base64.b64encode(signature).decode('utf-8')
def verify_alipay_callback(self, params):
"""验证支付宝回调"""
# 提取签名
sign = params.pop('sign')
sign_type = params.pop('sign_type')
# 重新签名
expected_sign = self._sign_params(params, self.alipay_config['public_key'])
return sign == expected_sign支付回调处理
这是最关键的部分——要确保回调的真实性：
from flask import Flask, request
app = Flask(__name__)
@app.route('/payment/alipay/callback', methods=['POST'])
def alipay_callback():
"""支付宝支付回调"""
params = request.form.to_dict()
# 1. 验证签名
if not payment_service.verify_alipay_callback(params.copy()):
app.logger.error(f"Invalid signature: {params}")
return 'fail', 400
# 2. 验证订单状态
order_id = params.get('out_trade_no')
trade_status = params.get('trade_status')
if trade_status != 'TRADE_SUCCESS':
app.logger.info(f"Trade not success: {trade_status}")
return 'success' # 返回success避免重复通知
# 3. 检查订单是否已处理
order = db.query(
"SELECT * FROM payment_orders WHERE order_id = ?",
(order_id,)
)
if not order:
app.logger.error(f"Order not found: {order_id}")
return 'fail', 400
if order[0]['status'] == 'paid':
app.logger.info(f"Order already paid: {order_id}")
return 'success' # 重复通知
# 4. 更新订单状态
db.execute("""
UPDATE payment_orders
SET status = 'paid', paid_at = ?,
trade_no = ?, total_amount = ?
WHERE order_id = ?
""", (
datetime.now(),
params.get('trade_no'),
params.get('total_amount'),
order_id
))
# 5. 激活订阅
db.execute("""
UPDATE subscriptions
SET status = 'active', activated_at = ?
WHERE id = ?
""", (datetime.now(), order[0]['subscription_id']))
# 6. 通知用户
subscription = db.query(
"SELECT * FROM subscriptions WHERE id = ?",
(order[0]['subscription_id'],)
)[0]
notify_subscription_active(subscription)
app.logger.info(f"Payment success: {order_id}")
return 'success'
@app.route('/payment/wechat/callback', methods=['POST'])
def wechat_callback():
"""微信支付回调"""
xml_data = request.data
data = xml_to_dict(xml_data)
# 1. 验证签名
if not payment_service.verify_wechat_callback(data):
app.logger.error(f"Invalid signature: {data}")
return dict_to_xml({'return_code': 'FAIL', 'return_msg': '签名失败'})
# 2. 验证订单
if data.get('result_code') != 'SUCCESS':
return dict_to_xml({'return_code': 'SUCCESS', 'return_msg': 'OK'})
out_trade_no = data.get('out_trade_no')
# 3-6. 处理订单（同支付宝）
process_payment_success(out_trade_no, data.get('transaction_id'))
return dict_to_xml({'return_code': 'SUCCESS', 'return_msg': 'OK'})订阅管理 📋
用户取消订阅怎么办？
这是我当时最纠结的问题：
场景A：立即回收
- 优点：节省资源
- 缺点：用户体验差，已分享的链接会失效
场景B：保留到月底
- 优点：用户友好
- 缺点：浪费资源
我的方案：30天宽限期
class SubscriptionManager:
"""订阅管理系统"""
def __init__(self):
self.grace_period_days = 30
def cancel_subscription(self, subscription_id):
"""取消订阅"""
subscription = db.query(
"SELECT * FROM subscriptions WHERE id = ?",
(subscription_id,)
)[0]
# 检查是否在宽限期内
if subscription['status'] == 'active':
# 设置为取消状态，但保留到订阅期结束
db.execute("""
UPDATE subscriptions
SET status = 'cancelled',
cancelled_at = ?,
auto_renew = FALSE
WHERE id = ?
""", (datetime.now(), subscription_id))
# 计算实际失效时间
expiry_date = subscription['end_date']
grace_expiry = expiry_date + timedelta(days=self.grace_period_days)
return {
'success': True,
'message': f'您的别名将保留至 {grace_expiry.strftime("%Y-%m-%d")}'
}
return {'error': '订阅状态无效'}
def check_alias_valid(self, alias):
"""检查别名是否有效"""
# 1. 检查是否有活跃订阅
active = db.query("""
SELECT * FROM subscriptions s
JOIN custom_aliases a ON s.id = a.subscription_id
WHERE a.alias = ?
AND s.status = 'active'
AND s.end_date > ?
""", (alias, datetime.now()))
if active:
return {'valid': True, 'reason': 'active_subscription'}
# 2. 检查是否在宽限期内
in_grace = db.query("""
SELECT * FROM subscriptions s
JOIN custom_aliases a ON s.id = a.subscription_id
WHERE a.alias = ?
AND s.status = 'cancelled'
AND s.end_date > ?
""", (alias, datetime.now() - timedelta(days=self.grace_period_days)))
if in_grace:
subscription = in_grace[0]
grace_end = subscription['end_date'] + timedelta(days=self.grace_period_days)
return {
'valid': True,
'reason': 'grace_period',
'expires_at': grace_end,
'warning': f'此别名将于 {grace_end.strftime("%Y-%m-%d")} 失效'
}
# 3. 已失效
return {'valid': False, 'reason': 'expired'}
def auto_renew_subscription(self, subscription_id):
"""自动续费"""
subscription = db.query(
"SELECT * FROM subscriptions WHERE id = ?",
(subscription_id,)
)[0]
if not subscription['auto_renew']:
return None
# 创建续费订单
new_order = payment_service.create_payment(
subscription['user_id'],
subscription['tier'],
subscription['months'],
subscription['default_payment_method']
)
# 自动扣款（需要用户授权免密支付）
try:
payment_service.execute_payment(new_order['order_id'])
return {'success': True}
except Exception as e:
app.logger.error(f"Auto renew failed: {e}")
# 通知用户续费失败
notify_renewal_failed(subscription)
return {'error': str(e)}
def cleanup_expired_aliases(self):
"""清理过期别名（定时任务）"""
# 每天执行一次
expired = db.query("""
SELECT a.* FROM custom_aliases a
LEFT JOIN subscriptions s ON s.id = a.subscription_id
WHERE (s.status = 'expired' OR s.status = 'cancelled')
AND s.end_date < ?
""", (datetime.now() - timedelta(days=self.grace_period_days),))
for alias in expired:
# 标记为可回收
db.execute("""
UPDATE custom_aliases
SET status = 'recyclable',
recycled_at = ?
WHERE id = ?
""", (datetime.now(), alias['id']))
app.logger.info(f"Alias recycled: {alias['alias']}")
return {'recycled': len(expired)}数据库设计
-- 订阅表
CREATE TABLE subscriptions (
id VARCHAR(50) PRIMARY KEY,
user_id INTEGER NOT NULL,
tier VARCHAR(20) NOT NULL,
tier_name VARCHAR(50) NOT NULL,
months INTEGER NOT NULL,
monthly_price DECIMAL(10,2) NOT NULL,
total_price DECIMAL(10,2) NOT NULL,
start_date TIMESTAMP NOT NULL,
end_date TIMESTAMP NOT NULL,
status VARCHAR(20) NOT NULL, -- pending_payment, active, cancelled, expired
auto_renew BOOLEAN DEFAULT FALSE,
default_payment_method VARCHAR(20),
cancelled_at TIMESTAMP,
activated_at TIMESTAMP,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
INDEX idx_user_id (user_id),
INDEX idx_status (status),
INDEX idx_end_date (end_date)
);
-- 支付订单表
CREATE TABLE payment_orders (
id INTEGER PRIMARY KEY AUTOINCREMENT,
order_id VARCHAR(50) UNIQUE NOT NULL,
subscription_id VARCHAR(50) NOT NULL,
user_id INTEGER NOT NULL,
amount DECIMAL(10,2) NOT NULL,
payment_method VARCHAR(20) NOT NULL,
status VARCHAR(20) NOT NULL, -- pending, paid, failed, refunded
trade_no VARCHAR(100),
paid_at TIMESTAMP,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (subscription_id) REFERENCES subscriptions(id),
INDEX idx_order_id (order_id),
INDEX idx_status (status)
);
-- 自定义别名表
CREATE TABLE custom_aliases (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL,
subscription_id VARCHAR(50),
alias VARCHAR(50) UNIQUE NOT NULL,
target_url VARCHAR(500) NOT NULL,
status VARCHAR(20) NOT NULL, -- active, recyclable, deleted
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
recycled_at TIMESTAMP,
FOREIGN KEY (subscription_id) REFERENCES subscriptions(id),
INDEX idx_alias (alias),
INDEX idx_status (status),
INDEX idx_user_id (user_id)
);收入数据 📈
第一个月的惊喜
付费功能上线一个月后，我做了个统计：
import pandas as pd
import matplotlib.pyplot as plt
# 查询订阅数据
subscriptions = db.query("""
SELECT tier, COUNT(*) as count, total_price
FROM subscriptions
WHERE status IN ('active', 'cancelled')
AND created_at >= '2026-02-01'
AND created_at < '2026-03-01'
GROUP BY tier
""")
# 统计数据
stats = {
'总付费用户': sum(s['count'] for s in subscriptions),
'月收入': sum(s['count'] * s['total_price'] for s in subscriptions),
'客单价': sum(s['count'] * s['total_price'] for s in subscriptions) / sum(s['count'] for s in subscriptions),
'付费转化率': '2.0%', # 200付费 / 10000活跃
}
print("=== 2026年2月收入统计 ===\n")
for key, value in stats.items():
print(f"{key}：{value}")
print("\n=== 套餐分布 ===")
for sub in subscriptions:
print(f"{sub['tier']}：{sub['count']}人，收入 ¥{sub['count'] * sub['total_price']}")输出：
=== 2026年2月收入统计 ===
总付费用户：200
月收入：¥15000
客单价：¥75
付费转化率：2.0%
=== 套餐分布 ===
basic：150人，收入 ¥7800
pro：50人，收入 ¥7200收入分析
让我用代码做个详细分析：
class RevenueAnalyzer:
"""收入分析器"""
def __init__(self):
self.server_cost = 5000 # 月服务器成本
self.payment_fee_rate = 0.006 # 0.6%支付手续费
def analyze_month(self, year, month):
"""分析某月收入"""
# 获取当月所有订阅
subscriptions = db.query("""
SELECT * FROM subscriptions
WHERE strftime('%Y-%m', created_at) = ?
""", (f"{year}-{month:02d}",))
analysis = {
'month': f"{year}-{month:02d}",
'new_subscriptions': len(subscriptions),
'gross_revenue': 0,
'payment_fees': 0,
'net_revenue': 0,
'profit': 0,
'tier_breakdown': {}
}
for sub in subscriptions:
tier = sub['tier']
amount = sub['total_price']
# 套餐分组
if tier not in analysis['tier_breakdown']:
analysis['tier_breakdown'][tier] = {
'count': 0,
'revenue': 0
}
analysis['tier_breakdown'][tier]['count'] += 1
analysis['tier_breakdown'][tier]['revenue'] += amount
analysis['gross_revenue'] += amount
# 计算费用和利润
analysis['payment_fees'] = analysis['gross_revenue'] * self.payment_fee_rate
analysis['net_revenue'] = analysis['gross_revenue'] - analysis['payment_fees']
analysis['profit'] = analysis['net_revenue'] - self.server_cost
return analysis
def predict_mrr(self, current_month):
"""预测MRR（月度经常性收入）"""
# 获取所有活跃订阅
active = db.query("""
SELECT tier, total_price, months, end_date
FROM subscriptions
WHERE status = 'active'
AND end_date > ?
""", (datetime.now(),))
mrr = 0
for sub in active:
monthly_revenue = sub['total_price'] / sub['months']
mrr += monthly_revenue
return mrr
def calculate_ltv(self, user_id):
"""计算用户生命周期价值"""
user_subs = db.query("""
SELECT total_price, months, start_date, end_date
FROM subscriptions
WHERE user_id = ?
AND status IN ('active', 'cancelled', 'expired')
""", (user_id,))
if not user_subs:
return 0
total_revenue = sum(sub['total_price'] for sub in user_subs)
# 计算用户生命周期
first_sub = min(user_subs, key=lambda x: x['start_date'])
last_sub = max(user_subs, key=lambda x: x['end_date'])
lifespan = (last_sub['end_date'] - first_sub['start_date']).days / 30
return {
'total_revenue': total_revenue,
'lifespan_months': round(lifespan, 1),
'ltv': total_revenue
}
# 分析2月数据
analyzer = RevenueAnalyzer()
feb_analysis = analyzer.analyze_month(2026, 2)
print("=== 2026年2月收入分析 ===\n")
print(f"新订阅数：{feb_analysis['new_subscriptions']}")
print(f"总收入：¥{feb_analysis['gross_revenue']:.2f}")
print(f"支付手续费：¥{feb_analysis['payment_fees']:.2f}")
print(f"净收入：¥{feb_analysis['net_revenue']:.2f}")
print(f"服务器成本：¥{analyzer.server_cost}")
print(f"净利润：¥{feb_analysis['profit']:.2f}")
print(f"利润率：{feb_analysis['profit']/feb_analysis['gross_revenue']*100:.1f}%")
print("\n=== 套餐分布 ===")
for tier, data in feb_analysis['tier_breakdown'].items():
print(f"{tier}：{data['count']}人，收入 ¥{data['revenue']:.2f}")
# 预测MRR
mrr = analyzer.predict_mrr(2026)
print(f"\n当前MRR：¥{mrr:.2f}")
# 计算LTV
high_value_users = db.query("""
SELECT user_id, SUM(total_price) as total
FROM subscriptions
GROUP BY user_id
ORDER BY total DESC
LIMIT 5
""")
print("\n=== 高价值用户TOP5 ===")
for user in high_value_users:
ltv = analyzer.calculate_ltv(user['user_id'])
print(f"用户{user['user_id']}：累计消费 ¥{ltv['total_revenue']}，生命周期 {ltv['lifespan_months']}个月")输出：
=== 2026年2月收入分析 ===
新订阅数：200
总收入：¥15000.00
支付手续费：¥90.00
净收入：¥14910.00
服务器成本：¥5000
净利润：¥9910.00
利润率：66.1%
=== 套餐分布 ===
basic：150人，收入 ¥7800.00
pro：50人，收入 ¥7200.00
当前MRR：¥18000.00
=== 高价值用户TOP5 ===
用户1024：累计消费 ¥1188.00，生命周期 12个月
用户2048：累计消费 ¥594.00，生命周期 6个月
用户3072：累计消费 ¥594.00，生命周期 6个月
用户4096：累计消费 ¥351.00，生命周期 3个月
用户5120：累计消费 ¥351.00，生命周期 3个月关键发现：
- 首次盈利！净利润 ¥9910，终于覆盖了服务器成本
- 专业版用户ARPU更高：虽然人数少，但贡献了48%的收入
- MRR持续增长：当前MRR ¥18000 > 2月收入 ¥15000，说明续费情况良好
- 高价值用户明显：TOP5用户贡献了20%的收入
本节小结 📝
商业模式验证
| 指标 | 目标 | 实际 | 状态 |
|---|---|---|---|
| 付费转化率 | 1% | 2.0% | ✅ 超预期 |
| 月收入 | ¥10000 | ¥15000 | ✅ 超预期 |
| 客单价 | ¥50 | ¥75 | ✅ 超预期 |
| 利润率 | 50% | 66% | ✅ 超预期 |
结论： 付费模式跑通了！
经验总结
定价策略
- 免费版建立用户基础
- 基础版满足轻度需求
- 专业版捕获高价值用户
支付体验
- 支付宝+微信双通道
- 回调验证确保安全
- 30天宽限期提升体验
数据驱动
- MRR监控长期趋势
- LTV识别高价值用户
- 转化漏斗优化定价
下一步计划
赚钱了，但也有了新问题——
老链接堆积如山，需要管理……
练习题
练习1：防止退款滥用
如何防止用户通过退款来”免费试用”付费别名？
参考答案
class RefundProtection:
"""退款保护机制"""
def __init__(self):
self.trial_period_days = 7
self.trial_click_limit = 100
self.refund_reputation_penalty = 20
def check_refund_eligibility(self, subscription_id):
"""检查是否可以退款"""
subscription = db.query(
"SELECT * FROM subscriptions WHERE id = ?",
(subscription_id,)
)[0]
# 检查1：试用期已过
days_used = (datetime.now() - subscription['activated_at']).days
if days_used > self.trial_period_days:
return {
'eligible': False,
'reason': f'使用超过{self.trial_period_days}天，不支持退款'
}
# 检查2：点击量超限
alias_clicks = self._get_alias_clicks(subscription_id)
if alias_clicks > self.trial_click_limit:
# 部分退款
refund_ratio = max(0.5, 1 - (alias_clicks / self.trial_click_limit))
return {
'eligible': True,
'partial_refund': True,
'refund_ratio': refund_ratio,
'reason': f'使用{alias_clicks}次，可退{refund_ratio*100:.0f}%'
}
return {'eligible': True, 'partial_refund': False}
def process_refund(self, user_id, subscription_id):
"""处理退款"""
# 检查资格
eligibility = self.check_refund_eligibility(subscription_id)
if not eligibility['eligible']:
return {'error': eligibility['reason']}
# 执行退款
if eligibility.get('partial_refund'):
# 部分退款
subscription = db.query(
"SELECT * FROM subscriptions WHERE id = ?",
(subscription_id,)
)[0]
refund_amount = subscription['total_price'] * eligibility['refund_ratio']
else:
# 全额退款
refund_amount = subscription['total_price']
# 调用支付平台退款接口
payment_service.refund(subscription['order_id'], refund_amount)
# 更新订阅状态
db.execute("""
UPDATE subscriptions
SET status = 'refunded'
WHERE id = ?
""", (subscription_id,))
# 降低用户信誉
reputation_system.reduce_score(user_id, self.refund_reputation_penalty)
# 限制再次购买
reputation_system.ban_user_from_trial(user_id, days=30)
return {
'success': True,
'refund_amount': refund_amount,
'message': '退款成功，账户信誉受到影响'
}策略总结：
- 时间限制：7天试用期内可退款
- 使用量限制：超过100次点击，部分退款
- 信誉机制：退款影响信誉分数
- 购买限制：退款用户30天内不能购买
练习2：提升付费转化率
当前转化率2%，如何提升到5%？
参考答案
class ConversionOptimizer:
"""转化率优化器"""
def __init__(self):
self.strategies = {
'免费试用': {
'description': '新用户7天免费试用专业版',
'expected_lift': '1.5%'
},
'限时折扣': {
'description': '首月5折优惠',
'expected_lift': '1.0%'
},
'功能限制': {
'description': '免费版每日创建10个链接上限',
'expected_lift': '0.8%'
},
'智能推荐': {
'description': '根据使用行为推荐合适套餐',
'expected_lift': '0.5%'
},
'社交证明': {
'description': '展示"XX公司正在使用"',
'expected_lift': '0.3%'
}
}
def recommend_upgrade(self, user_id):
"""智能推荐套餐"""
user = db.query("""
SELECT * FROM users WHERE id = ?
""", (user_id,))[0]
# 分析用户行为
usage = db.query("""
SELECT COUNT(*) as link_count,
MAX(created_at) as last_active
FROM links
WHERE user_id = ?
""", (user_id,))[0]
# 推荐1：重度使用者
if usage['link_count'] > 50:
return {
'tier': 'pro',
'reason': '您已创建50+链接，专业版更划算',
'discount': '首月8折'
}
# 推荐2：活跃用户
if usage['link_count'] > 10:
return {
'tier': 'basic',
'reason': '升级基础版，解锁自定义别名',
'discount': '首月5折'
}
# 推荐3：新用户
return {
'tier': 'trial',
'reason': '7天免费试用专业版',
'discount': '免费'
}优化策略：
- 免费试用：降低决策门槛
- 限时折扣：制造紧迫感
- 功能限制：提升免费版痛点
- 智能推荐：个性化触达
- 社交证明：建立信任
（下一节：链接管理）