一、环境准备与依赖安装
在开始构建A/B测试系统之前，必须先配置好基础的运行环境。本文采用Python作为后端语言，Redis作为数据存储，Flask作为Web框架。这种组合轻量且高效，非常适合快速落地。
确保你的系统中已经安装了Python 3.8或更高版本。接着，我们需要安装Flask框架和Redis的Python客户端。打开终端，直接执行以下命令进行安装：
```bash
pip install flask redis
```
为了运行Redis数据库，最简单的方式是使用Docker。如果你本地已经安装了Docker，直接运行以下命令启动一个Redis实例：
```bash
docker run -d -p 6379:6379 --name ab-test-redis redis:alpine
```
执行完毕后，Redis将在本机的6379端口运行。如果没有Docker，请自行下载并安装Redis Server，确保服务启动并监听6379端口。
二、项目目录结构规划
为了保证代码的清晰度，我们创建一个名为ab_test_project的文件夹，并在其中建立以下文件结构：
- ab_test_project/
- app.py (主程序入口)
- ab_core.py (A/B测试核心逻辑)
- templates/ (存放HTML文件)
三、核心A/B测试逻辑实现
在ab_core.py文件中，我们将编写处理流量分组和数据统计的核心类。这个类负责将用户分配到不同的实验组（A组或B组），并记录用户的浏览和转化行为。
创建并打开ab_core.py，输入以下完整代码：
```python
import redis
import random
import hashlib
class ABTestManager:
def __init__(self, redis_host='localhost', redis_port=6379):
self.r = redis.StrictRedis(host=redis_host, port=redis_port, decode_responses=True)
self.experiment_name = "landing_page_conversion"
def get_group(self, user_id):
"""
根据user_id确定用户所属分组，确保同一用户每次访问看到的是同一版本
使用哈希算法保证分配的一致性
"""
检查Redis中是否已有该用户的分组记录
existing_group = self.r.hget(f"user:{self.experiment_name}", user_id)
if existing_group:
return existing_group
如果没有记录，使用哈希算法进行分配
这里使用简单的哈希取模，50%概率去A组，50%概率去B组
hash_val = int(hashlib.md5(user_id.encode()).hexdigest(), 16)
group = 'A' if hash_val % 2 == 0 else 'B'
将分组结果存入Redis，永久保存（或设置过期时间）
self.r.hset(f"user:{self.experiment_name}", user_id, group)
return group
def track_event(self, group, event_type):
"""
记录事件数据到Redis
event_type: 'view' (浏览) 或 'conversion' (转化)
"""
key = f"stats:{self.experiment_name}:{group}:{event_type}"
self.r.incr(key)
def get_stats(self):
"""
获取当前的统计数据
"""
stats = {}
for group in ['A', 'B']:
views = self.r.get(f"stats:{self.experiment_name}:{group}:view") or 0
conversions = self.r.get(f"stats:{self.experiment_name}:{group}:conversion") or 0
stats[group] = {
'views': int(views),
'conversions': int(conversions),
'rate': round(int(conversions) / int(views) 100, 2) if int(views) > 0 else 0
}
return stats
```
这段代码实现了三个关键功能：get_group利用用户ID的哈希值进行分组，保证用户粘性；track_event利用Redis的原子递增操作记录数据，保证高并发下的数据准确性；get_stats用于计算各组的转化率。
四、Web服务搭建与路由配置
接下来，在app.py中编写Flask应用，将核心逻辑与HTTP请求连接起来。我们需要两个主要路由：一个用于展示页面（根据分组展示不同内容），一个用于接收前端传回的转化事件。
创建并打开app.py，输入以下完整代码：
```python
from flask import Flask, render_template, request, jsonify
from ab_core import ABTestManager
import uuid
app = Flask(__name__)
ab_manager = ABTestManager()
@app.route('/')
def index():
获取用户ID，这里为了演示，从Cookie中获取，如果没有则生成一个新的
user_id = request.cookies.get('ab_test_user_id')
如果Cookie中没有用户ID，生成一个UUID并设置到响应中（这里简化处理，实际需在response中设置）
注意：在Flask中设置Cookie需要通过make_response，这里为了逻辑清晰，
我们假设前端JS会协助处理Cookie，或者我们在模板渲染时处理。
为保证零门槛，我们在后端直接处理。
if not user_id:
user_id = str(uuid.uuid4())
获取分组
group = ab_manager.get_group(user_id)
记录浏览事件
ab_manager.track_event(group, 'view')
渲染模板，传入group和user_id
resp = render_template('index.html', group=group, user_id=user_id)
如果是刚生成的ID，需要设置Cookie
if request.cookies.get('ab_test_user_id') != user_id:
resp.set_cookie('ab_test_user_id', user_id, max_age=60602430) 保存30天
return resp
@app.route('/track_conversion', methods=['POST'])
def track_conversion():
user_id = request.cookies.get('ab_test_user_id')
if not user_id:
return jsonify({'status': 'error', 'message': 'User ID not found'}), 400
重新计算该用户的分组（因为需要知道是哪个组转化了）
group = ab_manager.get_group(user_id)
记录转化事件
ab_manager.track_event(group, 'conversion')
return jsonify({'status': 'success', 'group': group})
@app.route('/stats')
def stats():
data = ab_manager.get_stats()
return jsonify(data)
if __name__ == '__main__':
app.run(debug=True, port=5000)
```
五、前端页面与埋点代码
前端部分需要根据后端传入的group变量展示不同的UI设计，并在用户发生关键行为（如点击“立即购买”按钮）时，调用后端的/track_conversion接口。
在templates文件夹下创建index.html，输入以下完整代码：
```html
高转化落地页测试
专业数据分析工具
提升您的业务效率，洞察数据背后的价值。
{% if group == 'A' %}
版本 A：标准引导
{% else %}
版本 B：强调紧迫感引导
{% endif %}
```
六、运行系统与验证效果
代码编写完成后，即可启动系统进行验证。在终端中进入项目根目录，执行以下命令启动Flask服务：
```bash
python app.py
```
启动成功后，打开浏览器访问 http://127.0.0.1:5000。
验证步骤：
- 首次访问，你会看到页面显示“版本 A”或“版本 B”。按F12打开开发者工具，查看Application/Cookies，会发现
ab_test_user_id已被写入。
- 刷新页面，页面展示的版本应保持不变（因为Cookie中的用户ID未变，后端哈希计算结果一致）。
- 点击页面上的按钮，浏览器会弹出“转化成功”的提示，表明数据已发送至后端。
- 为了模拟多用户效果，你可以使用浏览器的无痕模式访问多次，或者清除Cookie后重新访问，观察不同分组的出现情况。
七、数据看板与结果分析
当积累了一定的流量数据后，我们需要查看A/B两组的转化率对比。我们已经在app.py中预留了/stats接口。
在浏览器新标签页中访问 http://127.0.0.1:5000/stats。你将看到类似以下的JSON数据：
```json
{
"A": {
"views": 50,
"conversions": 5,
"rate": 10.0
},
"B": {
"views": 52,
"conversions": 10,
"rate": 19.23
}
}
```
通过观察rate字段，你可以直观地对比不同版本的转化效果。在上述示例数据中，B组的转化率（19.23%）明显高于A组（10.0%），这意味着B组的文案和按钮设计更能促进流量转化。基于此数据，你可以决定将B组的版本全量上线。
至此，一套完整的、基于Python和Redis的流量转化A/B测试系统已完全落地。这套系统没有依赖任何复杂的第三方SaaS平台，所有数据均掌握在自己手中，且具备极高的扩展性，可根据实际业务需求增加更多的实验维度。
