手把手教你用Python实现私域流量自动激活系统

一、系统架构与核心逻辑设计

私域流量激活的核心在于“行为追踪”与“规则触发”。本系统采用 Python + Redis 构建，利用 Redis 的高性能读写能力处理用户行为埋点，使用 Python 编写灵活的规则引擎进行实时判断。

我们将实现一个具体的业务场景：“沉默用户召回”。即当用户在 3 天内有浏览行为但未下单时，自动触发优惠券推送；当高价值用户（累计消费 > 1000元）超过 7 天未登录时，触发短信唤醒。

技术选型理由：

• Redis：用于存储用户行为位掩码和最后活跃时间戳，支持毫秒级读写。
• Python：逻辑编写简单，易于集成各类短信或邮件 API。

二、环境准备与依赖安装

在开始编写代码前，必须确保本地环境已配置好 Redis 服务和 Python 运行环境。请严格按照以下步骤操作，不要跳过任何环节。

1. 安装并启动 Redis 服务

如果你本地没有 Redis，请根据你的操作系统执行以下命令安装：

MacOS (使用 Homebrew):

brew install redis
brew services start redis

Ubuntu (使用 apt):

sudo apt-get update
sudo apt-get install redis-server
sudo service redis-server start

Windows:

建议直接下载 Microsoft Store 版本或使用 WSL。下载地址：https://github.com/microsoftarchive/redis/releases，下载后解压并运行 redis-server.exe。

2. 安装 Python 依赖库

在项目根目录下创建 requirements.txt，或者直接执行以下 pip 命令安装必要的库：

pip install redis

我们只需要原生 redis 库即可，不需要额外的 web 框架，保持系统轻量化。

三、数据结构设计

为了高效存储和查询用户状态，我们在 Redis 中设计以下 Key 结构：

• 用户行为 Key：user:behavior:{user_id} (String 类型，存储位掩码)
• 最后活跃时间 Key：user:last_active:{user_id} (String 类型，存储 Unix 时间戳)
• 用户总消费 Key：user:total_spent:{user_id} (String 类型，存储金额)

位掩码定义（用于高效判断组合行为）：

• 1 (0001): 登录
• 2 (0010): 浏览商品
• 4 (0100): 加入购物车
• 8 (1000): 下单支付

例如，如果用户状态值为 3 (0011)，代表该用户既登录了也浏览了，但未加购和下单。

四、核心代码实现

为了方便直接运行，我们将所有逻辑封装在一个文件中。请在本地创建 activation_system.py 文件，并复制以下完整代码。

1. 配置与初始化模块

import redis
import time
import datetime
连接 Redis，默认 host 为 localhost，port 为 6379
如果有密码，请添加 password='你的密码' 参数
r = redis.StrictRedis(host='localhost', port=6379, db=0, decode_responses=True)
行为位掩码定义
ACTION_LOGIN = 1
ACTION_VIEW = 2
ACTION_CART = 4
ACTION_PAY = 8
激活规则配置
RULE_BROWSE_NO_PAY_DAYS = 3   浏览未下单超过3天
RULE_VIP_SILENT_DAYS = 7      VIP用户沉默超过7天
RULE_VIP_THRESHOLD = 1000     VIP定义门槛

2. 用户行为埋点服务

这是系统的入口，模拟用户在 App 或网页端产生的行为。你需要将此逻辑集成到你们现有的后端接口中。

class UserTracker:
@staticmethod
def track(user_id, action_type, total_spent_delta=0):
"""
记录用户行为
:param user_id: 用户ID
:param action_type: 行为类型 (位掩码)
:param total_spent_delta: 本次消费金额增量
"""
pipe = r.pipeline()
1. 更新行为位掩码 (使用按位或运算叠加状态)
behavior_key = f"user:behavior:{user_id}"
pipe.setbit(behavior_key, action_type, 1)
2. 更新最后活跃时间
time_key = f"user:last_active:{user_id}"
current_ts = int(time.time())
pipe.set(time_key, current_ts)
3. 如果有消费，更新总金额
if total_spent_delta > 0:
spent_key = f"user:total_spent:{user_id}"
pipe.incrbyfloat(spent_key, total_spent_delta)
pipe.execute()
print(f"[埋点成功] 用户ID: {user_id}, 行为: {action_type}, 时间: {current_ts}")

3. 激活规则引擎

这是核心逻辑，负责扫描用户数据并判断是否触发激活策略。为了演示，这里使用轮询方式，生产环境建议使用 Celery Beat 或定时任务 Crontab。

class ActivationEngine:
@staticmethod
def send_coupon(user_id, reason):
"""模拟发送优惠券接口"""
print(f"🎉 [触发激活] 用户: {user_id}, 策略: 发送优惠券, 原因: {reason}")
实际代码中调用：requests.post("https://api.yourservice.com/send_coupon", ...)
@staticmethod
def send_sms(user_id, reason):
"""模拟发送短信接口"""
print(f"📩 [触发激活] 用户: {user_id}, 策略: 发送短信, 原因: {reason}")
@staticmethod
def check_user(user_id):
"""
检查单个用户是否满足激活条件
"""
behavior_key = f"user:behavior:{user_id}"
time_key = f"user:last_active:{user_id}"
spent_key = f"user:total_spent:{user_id}"
获取 Redis 数据
last_active_ts = r.get(time_key)
if not last_active_ts:
return  新用户或无数据
last_active_ts = int(last_active_ts)
current_ts = int(time.time())
days_inactive = (current_ts - last_active_ts) / 86400
获取行为状态 (Redis getbit 返回 0/1，这里我们需要获取整体掩码进行逻辑判断)
注意：Redis 的 bit 操作对于 offset 较大时性能略有损耗，但在小规模位运算中极快
has_viewed = r.getbit(behavior_key, ACTION_VIEW)
has_paid = r.getbit(behavior_key, ACTION_PAY)
规则 1: 浏览未下单召回
逻辑：有浏览行为(1) 且 无支付行为(0) 且 沉默天数 > 3
if has_viewed and not has_paid:
if days_inactive > RULE_BROWSE_NO_PAY_DAYS:
为了防止重复发送，可以在这里增加一个 "sent_flag" 的判断
此处演示直接触发
ActivationEngine.send_coupon(user_id, f"浏览未下单超过 {int(days_inactive)} 天")
规则 2: 高价值用户沉默唤醒
逻辑：总金额 > 1000 且 沉默天数 > 7
total_spent = r.get(spent_key)
if total_spent and float(total_spent) >= RULE_VIP_THRESHOLD:
if days_inactive > RULE_VIP_SILENT_DAYS:
ActivationEngine.send_sms(user_id, f"VIP用户沉默超过 {int(days_inactive)} 天，请回来")

五、全流程模拟测试

将以下代码追加到 activation_system.py 文件末尾。这段代码模拟了三个不同类型的用户，并演示了系统如何精准识别需要激活的对象。

def run_simulation():
print(" 开始初始化测试数据 ")
场景 A: 普通浏览用户，5天前活跃，未下单 -> 应触发规则1
user_a = 1001
UserTracker.track(user_a, ACTION_VIEW)  模拟浏览
修改时间戳模拟5天前
r.set(f"user:last_active:{user_a}", int(time.time()) - (5  86400))
场景 B: 高价值VIP，10天前活跃，已消费2000 -> 应触发规则2
user_b = 1002
UserTracker.track(user_b, ACTION_PAY, total_spent_delta=2000)  模拟支付
r.set(f"user:last_active:{user_b}", int(time.time()) - (10  86400))
场景 C: 新活跃用户，刚下单 -> 不应触发任何规则
user_c = 1003
UserTracker.track(user_c, ACTION_PAY, total_spent_delta=100)
print("\n 开始执行激活扫描 ")
扫描所有用户 (实际业务中请使用 Redis Scan 命令遍历，避免 keys )
test_users = [user_a, user_b, user_c]
for uid in test_users:
ActivationEngine.check_user(uid)
if __name__ == "__main__":
清理旧数据（可选）
r.flushdb()
run_simulation()

六、运行与验证结果

在终端中执行以下命令运行系统：

python activation_system.py

预期输出结果：

 开始初始化测试数据
[埋点成功] 用户ID: 1001, 行为: 2, 时间: 1715423100 (示例时间)
[埋点成功] 用户ID: 1002, 行为: 8, 时间: 1715077500 (示例时间)
[埋点成功] 用户ID: 1003, 行为: 8, 时间: 1715478300 (示例时间)
开始执行激活扫描
🎉 [触发激活] 用户: 1001, 策略: 发送优惠券, 原因: 浏览未下单超过 5 天
📩 [触发激活] 用户: 1002, 策略: 发送短信, 原因: VIP用户沉默超过 10 天，请回来

如果看到上述输出，说明系统逻辑运行正确。用户 1001 被识别为潜客并发放了优惠券，用户 1002 被识别为流失 VIP 并发送了短信，用户 1003 因为活跃而没有被打扰。

七、生产环境落地建议

上述代码是核心逻辑的精简版，要在生产环境直接落地，还需要注意以下三个关键细节，否则会导致性能瓶颈或重复发送。

1. 防重机制（去重） 目前的代码每次扫描都会触发。必须在 Redis 中增加 user:sent:{rule_id}:{user_id} 的 Key，设置过期时间为 30 天。触发前先检查 Key 是否存在，触发后设置 Key。

 伪代码示例
if not r.exists(f"user:sent:coupon_browse:{user_id}"):
send_coupon(...)
r.setex(f"user:sent:coupon_browse:{user_id}", 86400  30, "1")

2. 大规模数据扫描优化 代码中使用了简单的列表遍历。当用户量达到百万级时，严禁使用 keys 。请使用 Redis 的 SCAN 命令进行增量式遍历，或者维护一个“活跃用户集合”来缩小扫描范围。

 使用 SCAN 命令示例
for key in r.scan_iter(match="user:last_active:"):
user_id = key.split(":")[-1]
ActivationEngine.check_user(user_id)

3. 异步执行 埋点操作（写入 Redis）要同步完成以保证数据准确，但规则引擎的扫描和发送动作（发短信、发券）必须放入异步队列（如 Redis Stream + RQ，或 Celery）中执行，避免阻塞主业务线程。