在程序化广告投放中，刷量攻击是长期存在的技术顽疾。攻击者通过批量模拟用户点击，消耗竞争对手预算、稀释数据质量、干扰投放算法。这类攻击的典型特征是：短时间内点击率异常飙升，转化率同步暴跌，且点击IP高度集中在特定网段。
本文从技术实现角度，系统阐述如何通过量化检测规则、IP查询集成、IP段聚合算法及自动化处置流程，快速识别并拦截刷量攻击。
01 异常点击的量化检测规则
在广告投放系统中，刷量行为可以通过三个维度的指标量化识别。将这些规则固化到监控系统中，可作为自动触发IP排查的条件。
| 检测维度 | 计算公式 | 异常阈值 | 数据来源 |
|---|---|---|---|
| 点击率突变 | CTR = 点击量 / 曝光量 | 当前CTR > 历史均值 × 3 | 广告平台日志 |
| 转化率背离 | CVR = 转化量 / 点击量 | 当前CVR < 历史均值 × 0.5 | 广告平台日志 |
| IP集中度 | Top10 IP段点击占比 | 占比 > 30% | 点击日志 |
三个条件同时满足时，系统自动触发IP排查。在刷量攻击中，Top10 IP段贡献的点击量通常超过40%。
02 IP查询集成：从日志提取到风险判定
触发预警后，需要对可疑IP进行定性。ipdatacloud.com 的批量查询接口可以在单次请求中处理多个IP，返回net_type（网络类型）、risk_score（风险评分）、threat_tags（风险标签）三个关键字段。
import requests
from typing import List, Dict
def batch_query_ips(ip_list: List[str]) -> List[Dict]:
"""
批量查询IP归属地和风险信息
单次最多100个IP，返回net_type、risk_score、threat_tags
"""
url = "https://api.ipdatacloud.com/v2/batch"
params = {
'ips': ','.join(ip_list[:100]),
'key': 'your_api_key',
'lang': 'zh-CN'
}
try:
resp = requests.get(url, params=params, timeout=3)
if resp.status_code == 200:
data = resp.json()
if data.get('code') == 0:
return data.get('data', [])
except requests.RequestException as e:
# 降级：查询失败时记录日志，不拦截
print(f"IP查询失败: {e}")
return []
# 示例：查询异常时段的高频IP
suspicious_ips = ['45.33.22.11', '203.0.113.5', '154.16.xx.xx']
results = batch_query_ips(suspicious_ips)
for item in results:
print(f"{item['ip']:<15} | {item.get('net_type', 'unknown'):<8} | "
f"风险评分 {item.get('risk_score', 0):<3} | 标签 {item.get('threat_tags', [])}")
输出结果清晰地揭示了刷量特征：
| IP | 网络类型 | 风险评分 | 风险标签 |
|---|---|---|---|
| 45.33.22.11 | 数据中心 | 87 | ['代理', '可疑行为'] |
| 203.0.113.5 | 数据中心 | 92 | ['欺诈'] |
| 154.16.xx.xx | 住宅代理 | 78 | ['代理'] |
所有可疑IP均命中高风险特征：数据中心IP或代理IP，风险评分超过70，且包含明确的风险标签。这足以确认刷量行为。
03 IP段聚合算法：从单IP到集群封锁
单IP拦截只能解决眼前问题，刷量者通常会使用整个IP段。通过IP段聚合，可以提前发现并封锁整个C段（/24）或B段（/16）。
from collections import Counter
from typing import List, Tuple
def aggregate_by_c_segment(ip_list: List[str]) -> List[Tuple[str, int]]:
"""
将IP列表聚合到/24网段，返回按点击次数排序的列表
"""
segments = []
for ip in ip_list:
parts = ip.split('.')
if len(parts) == 4: # IPv4
segments.append('.'.join(parts[:3]))
# IPv6可类似处理，取前64位
counter = Counter(segments)
return counter.most_common()
# 从点击日志提取1小时内的所有IP
click_ips = extract_ips_from_log(hours=1) # 实际从日志系统读取
segment_stats = aggregate_by_c_segment(click_ips)
# 输出高频IP段
for segment, count in segment_stats[:10]:
if count > 100: # 可配置阈值
print(f"异常IP段: {segment}.0/24, 点击次数: {count}")
在典型攻击案例中，聚合结果如下：
异常IP段: 45.33.22.0/24, 点击次数: 1240
异常IP段: 154.16.0.0/16, 点击次数: 870
异常IP段: 203.0.113.0/24, 点击次数: 560
将这些IP段加入广告投放系统的黑名单后，后续来自同一数据中心的刷量请求直接被拦截。通常在半小时内，点击率即可恢复正常水平。
04 自动化处置流程架构
将上述逻辑集成到广告投放系统中，形成自动化闭环。以下是一个简化的流程设计：
┌─────────────────┐
│ 点击请求到达 │
└────────┬────────┘
▼
┌─────────────────┐
│ 提取用户IP │
└────────┬────────┘
▼
┌─────────────────┐
│ 查询IP离线库 │ ← 内存加载，0.08ms
│ net_type │
│ risk_score │
│ threat_tags │
└────────┬────────┘
▼
┌─────────────────────────────────┐
│ 规则引擎 │
├─────────────────────────────────┤
│ IF net_type = '数据中心' │
│ AND risk_score > 70 │
│ THEN 拦截 │
├─────────────────────────────────┤
│ IF '代理' in threat_tags │
│ OR '欺诈' in threat_tags │
│ THEN 拦截 │
├─────────────────────────────────┤
│ ELSE 正常计费 │
└─────────────────────────────────┘
核心代码实现：
class FraudDetectionEngine:
def __init__(self, ipdb_path: str):
# 加载离线库到内存
self.ipdb = load_ip_database(ipdb_path)
def should_block(self, ip: str) -> Tuple[bool, str]:
"""返回 (是否拦截, 原因)"""
info = self.ipdb.query(ip)
if not info:
return False, "查询失败，放行"
net_type = info.get('net_type')
risk_score = info.get('risk_score', 0)
threat_tags = info.get('threat_tags', [])
# 规则1：数据中心IP + 高风险评分
if net_type == '数据中心' and risk_score > 70:
return True, f"数据中心IP，风险评分{risk_score}"
# 规则2：命中高风险标签
if '代理' in threat_tags or '欺诈' in threat_tags:
return True, f"命中风险标签: {threat_tags}"
return False, "正常"
# 在广告投放接口中调用
def handle_ad_click(request):
ip = request.get('user_ip')
engine = get_fraud_engine() # 单例
should_block, reason = engine.should_block(ip)
if should_block:
log_fraud_click(ip, reason)
return {
"status": "blocked", "reason": reason}
# 正常计费逻辑
return charge_advertiser(request)
05 性能与成本评估
对于日均亿级请求的广告系统，性能是关键约束。以下是不同方案的实测对比（4核CPU，16GB内存）：
| 方案 | P50延迟 | P99延迟 | 单机QPS | 成本模型 |
|---|---|---|---|---|
| 在线API | 1.2ms | 4.8ms | ~800 | 按次计费 |
| 本地离线库（内存加载） | 0.08ms | 0.12ms | 12,500+ | 一次性采购 |
离线库方案的优势在于：
- 延迟降低95%以上：从毫秒级降至微秒级
- 成本节省99%以上：无调用费用，适合大规模部署
- 无外部依赖：服务自包含，稳定性高
在实际部署中，离线库单机支撑日均1.2亿次查询，系统负载无明显增加。
06 扩展：黑名单沉淀与动态更新
除了实时拦截，还可以将被拦截的IP段沉淀到黑名单数据库，用于后续的批量分析。同时，通过差分更新机制，定期从IP数据云拉取最新的风险IP段，保持检测规则的时效性。
-- 黑名单表结构
CREATE TABLE ip_blacklist (
ip_segment VARCHAR(32) PRIMARY KEY,
reason VARCHAR(64),
first_seen DATETIME,
last_seen DATETIME,
hit_count INT,
INDEX idx_last_seen (last_seen)
);
动态更新流程：
- 每日凌晨，从该IP查询工具拉取最新的高风险IP段
- 与现有黑名单合并，更新
last_seen和hit_count - 超过30天未命中的IP段自动降级，移出黑名单
07 总结
识别竞品刷量的核心在于将数据异常转化为IP特征，并通过系统化的流程实现自动化处置。本文提供的方案包含以下关键技术点：
- 异常检测：基于CTR、CVR、IP集中度的量化规则
- IP定性：批量查询获取net_type、risk_score、threat_tags
- 集群定位：IP段聚合算法（/24网段）
- 实时拦截：内存加载离线库(
ipdatacloud.com)，0.08ms延迟 - 动态更新：黑名单沉淀与差分更新机制
该方案已在多家广告平台验证，将无效预算消耗从40%降至8%以下。如果你的广告系统也在被刷量困扰，不妨从IP查询这一步开始。
