转化漏斗与用户分层

不看数据做电商，就像闭着眼睛开车。

数据驱动增长架构

核心电商指标

转化漏斗分析

"""
电商转化漏斗
"""
from dataclasses import dataclass
@dataclass
class FunnelStage:
name: str
users: int
class FunnelAnalyzer:
"""转化漏斗分析"""
def __init__(self, stages: list[FunnelStage]):
self.stages = stages
def analyze(self) -> list[dict]:
"""漏斗分析"""
results = []
for i, stage in enumerate(self.stages):
if i == 0:
conversion = 1.0
drop_off = 0
else:
prev = self.stages[i - 1].users
conversion = stage.users / prev if prev else 0
drop_off = prev - stage.users
results.append({
"阶段": stage.name,
"人数": f"{stage.users:,}",
"转化率": f"{conversion*100:.1f}%",
"流失": f"{drop_off:,}",
})
return results
def bottleneck(self) -> str:
"""找到最大瓶颈"""
worst_stage = ""
worst_rate = 1.0
for i in range(1, len(self.stages)):
rate = (
self.stages[i].users / self.stages[i - 1].users
)
if rate < worst_rate:
worst_rate = rate
worst_stage = self.stages[i].name
return (
f"最大瓶颈: {worst_stage}"
f" (转化率 {worst_rate*100:.1f}%)"
)
# 演示
funnel = FunnelAnalyzer([
FunnelStage("访问首页", 100000),
FunnelStage("浏览商品", 45000),
FunnelStage("加入购物车", 12000),
FunnelStage("开始结账", 5000),
FunnelStage("完成支付", 3200),
])
print("=== 转化漏斗 ===")
for stage in funnel.analyze():
print(f"  {stage['阶段']}: {stage['人数']} (转化 {stage['转化率']}, 流失 {stage['流失']})")
print(f"\n{funnel.bottleneck()}")

RFM 用户分层

"""
RFM 用户分层分析
"""
from dataclasses import dataclass
@dataclass
class Customer:
id: str
name: str
recency_days: int     # 最近一次购买距今天数
frequency: int        # 购买次数
monetary: float       # 累计消费金额
class RFMAnalyzer:
"""RFM 分层"""
SEGMENTS = {
(True, True, True): "VIP 客户",
(True, True, False): "活跃忠诚",
(True, False, True): "高价值新客",
(True, False, False): "新客户",
(False, True, True): "沉睡高价值",
(False, True, False): "沉睡忠诚",
(False, False, True): "流失高价值",
(False, False, False): "流失客户",
}
def __init__(
self,
customers: list[Customer],
recency_threshold: int = 90,
frequency_threshold: int = 3,
monetary_threshold: float = 1000,
):
self.customers = customers
self.r_t = recency_threshold
self.f_t = frequency_threshold
self.m_t = monetary_threshold
def segment(self, c: Customer) -> str:
"""给单个客户分层"""
r = c.recency_days <= self.r_t
f = c.frequency >= self.f_t
m = c.monetary >= self.m_t
return self.SEGMENTS.get((r, f, m), "未分类")
def summary(self) -> dict:
"""分层汇总"""
groups: dict[str, list] = {}
for c in self.customers:
seg = self.segment(c)
if seg not in groups:
groups[seg] = []
groups[seg].append(c)
result = {}
for seg, members in groups.items():
total_rev = sum(m.monetary for m in members)
result[seg] = {
"人数": len(members),
"总消费": f"¥{total_rev:,.0f}",
"人均消费": f"¥{total_rev / len(members):,.0f}",
}
return result
def action_plan(self) -> dict:
"""运营策略"""
return {
"VIP 客户": "专属折扣、新品优先、1v1服务",
"活跃忠诚": "积分加倍、会员升级",
"高价值新客": "引导复购、推荐套装",
"新客户": "首购优惠券、引导建档",
"沉睡高价值": "召回优惠、限时折扣",
"流失客户": "大力度优惠券、短信触达",
}
# 演示
customers = [
Customer("C001", "张三", 15, 8, 5800),
Customer("C002", "李四", 30, 5, 2100),
Customer("C003", "王五", 200, 1, 300),
Customer("C004", "赵六", 5, 1, 3500),
Customer("C005", "钱七", 120, 6, 4200),
Customer("C006", "孙八", 300, 2, 150),
]
rfm = RFMAnalyzer(customers)
print("=== 客户分层 ===")
for c in customers:
print(f"  {c.name}: {rfm.segment(c)}")
print("\n=== 分层汇总 ===")
for seg, info in rfm.summary().items():
print(f"  {seg}: {info}")
print("\n=== 运营策略 ===")
for seg, action in rfm.action_plan().items():
print(f"  {seg}: {action}")

AOV 提升策略

下一章：AI 赋能电商——用人工智能释放增长潜力。