Appearance
广告后台的地域报表能信吗？用后端日志核验投放地域和无效流量（Python 实现）
投放团队月底复盘时常遇到一个尴尬场面：媒体后台的地域报表写着「一线城市占比 58%，符合投放策略」，可自己 CRM 里这个月新签的客户，七成在三四线。两边数据都摆在桌上，谁也说服不了谁，最后结论往往是「IP 定位不准吧」——这个结论八成是错的。
地域数据对不上，原因通常不在定位精度的零点几个百分点，而在口径和流量质量这两件事上。这篇用 Python 写一套核验脚本：把后端的转化日志按渠道、按地域摊开，再和自己埋点里的 IP 归属地交叉比对，哪些渠道的地域数据有水分，一眼能看出来。
一、先把三件「对不上」的事分清
第一件：两边测的根本不是同一时刻。
媒体后台的地域，来自它收到广告请求（或点击上报）时的 IP；你后端日志里的地域，来自用户注册、下单那一刻的 IP。中间隔着的这段时间里，用户可能从办公室 WiFi 切到手机流量，出口从北京联通变成了河北某地市的省网出口。点击和转化 IP 完全不同，是很常见的事。所以对比时要比同一批转化的来源渠道，而不是拿媒体的「曝光地域」直接对后端的「下单地域」。
第二件：机器流量会挑「好看」的 IP。
刷量的工作室早就过了用单一机房 IP 硬刷的阶段。现在常见的是代理池：一批出口 IP 分散在多个省份，地域分布做得比真流量还均匀。但它们有几个擦不掉的特征——出口集中在少数几个 C 段，以及 ISP 字段带机房/云厂商字样。这两点不需要风控评分接口，用免费版 IP9 接口返回的 isp 字段加上自己的聚合就能看出来。要更准的话，VIP 版直接返回 ip_type（ISP 家庭 / BUS 企业 / IDC 机房），不用靠关键词猜。
第三件：移动端 IP 本来就不适合做城市级归因。
手机流量的出口常常是省级或大区汇聚点，查出来可能是手机卡注册地，和用户此刻所在城市差几个省都正常。拿这类 IP 去做「北京市朝阳区用户」这种颗粒度的核验，纯属自找麻烦。核验脚本里得把移动出口单独标出来，别混进城市分布。
二、方案设计：采集 → 打标 → 三张表
思路很土但管用：
- 采集层：把转化事件（注册/下单/留资）按「时间、渠道、IP、事件类型」落成一张表。这份数据你本来就有，只是平时没人拿它做地域核验。
- 打标层：批量查归属地，返回
prov、city、isp。免费版 60 次/分钟，所以必须带本地缓存和出站限速——缓存按 IP 落盘，成功结果缓存 7 天，失败结果只缓存 60 秒，避免一次接口抖动把一天的统计毁掉。 - 核验层：出三张对比表。
- 每个渠道的地域分布（按
city聚合），和媒体后台同地区数据并排看，偏差超过 20 个百分点且样本量过百的才值得深挖； - 每个渠道的C 段聚集度（同一
/24段占该渠道事件的比例），超过 40% 基本可判定出口集中； - 机房特征 IP 占比和海外 IP 占比，这两项和投放地区描述直接冲突时最说明问题。
- 每个渠道的地域分布（按
脚本不做封禁、不写结论，只把证据摆出来，判断交给人。风控口径一旦自动执行，误伤的代价比多花点预算大得多。
三、Python 实现
一个文件跑完，不依赖第三方库。下面的代码在我本机实测过，直接 python3 ad_geo_audit.py 会用内置的示例数据跑出一张报告，也可以用 python3 ad_geo_audit.py events.csv 换成自己的日志（列名固定为 ts,channel,ip,event）。
python
#!/usr/bin/env python3
"""广告地域数据核验：用后端转化日志核验媒体地域报表，并做无效流量初筛。"""
import csv
import io
import json
import os
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from collections import Counter, defaultdict
API = "https://ip9.com.cn/get"
CACHE_FILE = "ip_geo_cache.json"
TTL_OK, TTL_FAIL, MIN_INTERVAL = 7 * 86400, 60, 1.0 # 7天/60秒/1秒（免费版60次/分钟）
# 不带 User-Agent 的裸请求会被前置防护挡回来（实测 urllib 默认 UA 直接 403），
# requests/curl 有默认 UA 所以不容易踩到，写 urllib 要自己加。
HEADERS = {"User-Agent": "Mozilla/5.0 (compatible; ad-geo-audit/1.0)", "Accept": "application/json"}
# 机房/云厂商特征词：免费版没 ip_type 字段，先用 isp 文本 + IP 段聚集度做粗判；
# VIP 版有 ip_type（ISP/BUS/IDC），可以直接用字段判定，误判更少。
IDC_HINTS = ("云", "IDC", "数据中心", "机房", "BGP", "CDN", "服务器", "科技", "网络科技", "主机")
SAMPLE = """ts,channel,ip,event
2026-09-23T09:12:03,feed_ads,114.114.114.114,click
2026-09-23T09:12:07,feed_ads,121.40.16.8,click
2026-09-23T09:13:11,feed_ads,47.98.32.19,click
2026-09-23T09:15:40,feed_ads,223.5.5.5,click
2026-09-23T09:20:02,feed_ads,106.11.34.7,click
2026-09-23T10:01:22,search_ads,58.246.11.9,register
2026-09-23T10:03:51,search_ads,114.114.114.114,register
2026-09-23T10:08:10,search_ads,182.61.200.6,register
2026-09-23T10:30:45,search_ads,120.55.12.33,register
2026-09-23T11:00:00,search_ads,110.87.30.12,register
2026-09-23T11:21:19,kol_link,223.5.5.5,order
2026-09-23T11:40:02,kol_link,36.110.20.8,order
2026-09-23T12:02:33,kol_link,101.226.4.6,order
2026-09-23T12:30:11,kol_link,124.90.55.3,order
2026-09-23T13:05:47,kol_link,8.8.8.8,order
"""
class GeoIndex:
"""带本地缓存的归属地查询：缓存 + 限速 + 错误内部消化。"""
def __init__(self, path=CACHE_FILE):
self.path = path
self.cache = {}
self.last_call = 0.0
if os.path.exists(path):
try:
self.cache = json.load(open(path, encoding="utf-8"))
except (ValueError, OSError):
self.cache = {}
self.hit = self.miss = self.fail = 0
def _cached(self, ip):
rec = self.cache.get(ip)
if not rec:
return None
ttl = TTL_OK if rec.get("ret") == 200 else TTL_FAIL
if time.time() - rec.get("ts", 0) > ttl:
return None
return rec
def lookup(self, ip):
ip = (ip or "").strip()
if not ip:
return {"ret": 0, "err": "empty_ip"}
rec = self._cached(ip)
if rec:
self.hit += 1
return rec
gap = time.time() - self.last_call
if gap < MIN_INTERVAL:
time.sleep(MIN_INTERVAL - gap)
self.last_call = time.time()
url = API + "?" + urllib.parse.urlencode({"ip": ip})
out = {"ip": ip, "ret": 0, "ts": time.time()}
try:
req = urllib.request.Request(url, headers=HEADERS)
with urllib.request.urlopen(req, timeout=3) as resp:
body = json.loads(resp.read().decode("utf-8"))
out["ret"] = int(body.get("ret", 0))
data = body.get("data") or {}
out.update({
"country": (data.get("country") or "").strip(),
"prov": (data.get("prov") or "").strip(),
"city": (data.get("city") or "").strip(),
"isp": (data.get("isp") or "").strip(),
"big_area": (data.get("big_area") or "").strip(),
})
except urllib.error.HTTPError as e:
out["err"] = "http_%d" % e.code # 400 非法IP / 403 缺 UA / 429 限速
except Exception as e: # 超时、DNS、被中间代理改写
out["err"] = type(e).__name__
if out["ret"] != 200:
self.fail += 1
else:
self.miss += 1
self.cache[ip] = out
return out
def save(self):
json.dump(self.cache, open(self.path, "w", encoding="utf-8"), ensure_ascii=False)
def load_events(path=None):
raw = open(path, encoding="utf-8").read() if path else SAMPLE
return list(csv.DictReader(io.StringIO(raw)))
def c_segment(ip):
"""/24 段：同一机房出口常落在同一 C 段，比单个 IP 更能说明聚集。"""
if ":" in ip:
return ip.rsplit(":", 2)[0]
parts = ip.split(".")
return ".".join(parts[:3]) + ".0/24" if len(parts) == 4 else ip
def audit(rows, geo):
stat = defaultdict(lambda: {
"n": 0, "geo": Counter(), "seg": Counter(), "idc": 0, "overseas": 0,
"unknown": 0, "first": None, "last": None,
})
region_of = {}
for row in rows:
ch = row["channel"]
ip = row["ip"].strip()
rec = geo.lookup(ip)
s = stat[ch]
s["n"] += 1
s["seg"][c_segment(ip)] += 1
s["first"] = s["first"] or row["ts"]
s["last"] = row["ts"]
if rec.get("ret") != 200:
s["unknown"] += 1
region_of[ip] = "未知"
continue
country, prov, city = rec["country"], rec["prov"], rec["city"]
if country == "中国":
region = city or prov or "未知城市"
elif country == "保留":
region = "内网/保留地址"
else:
region = country or "海外"
s["overseas"] += 1
region_of[ip] = region
s["geo"][region] += 1
if any(h in rec["isp"] for h in IDC_HINTS):
s["idc"] += 1
return stat, region_of
def report(stat):
print("渠道 事件数 主要地域(占比) C段聚集度 机房特征 海外 查询失败")
print("-" * 104)
for ch, s in sorted(stat.items(), key=lambda kv: -kv[1]["n"]):
top = "、".join("%s %.0f%%" % (r, c * 100.0 / s["n"]) for r, c in s["geo"].most_common(3))
seg_top = s["seg"].most_common(1)[0][1] * 100.0 / s["n"]
print("%-10s %5d %-34s %8.0f%% %8.0f%% %4.0f%% %d"
% (ch, s["n"], top[:34], seg_top, s["idc"] * 100.0 / s["n"],
s["overseas"] * 100.0 / s["n"], s["unknown"]))
print("\n核验提示（人工判断，脚本只给证据）：")
for ch, s in sorted(stat.items(), key=lambda kv: -kv[1]["n"]):
seg_top = s["seg"].most_common(1)[0][1] * 100.0 / s["n"]
flags = []
if seg_top >= 40:
flags.append("同一 /24 段占 %.0f%%，疑似单机房出口" % seg_top)
if s["idc"] * 100.0 / s["n"] >= 50:
flags.append("机房/云厂商特征 IP 占 %.0f%%" % (s["idc"] * 100.0 / s["n"]))
if s["overseas"] * 100.0 / s["n"] >= 20:
flags.append("海外 IP 占 %.0f%%，与投放地区描述不符时需核对" % (s["overseas"] * 100.0 / s["n"]))
if s["unknown"] * 100.0 / s["n"] >= 10:
flags.append("查询失败 %.0f%%，先修数据入口再谈核验" % (s["unknown"] * 100.0 / s["n"]))
print(" [%s] %s" % (ch, "；".join(flags) if flags else "未发现明显异常"))
print("\n与媒体报表对比：把上表各地的转化占比，和广告后台同地区的点击占比并排放，")
print("偏差超过 20 个百分点且样本量 >100 的渠道，再去看 C 段聚集度和机房特征。")
def main():
path = sys.argv[1] if len(sys.argv) > 1 else None
rows = load_events(path)
geo = GeoIndex()
stat, _ = audit(rows, geo)
report(stat)
geo.save()
print("\n本次查询：缓存命中 %d，实调 %d，失败 %d；缓存已写入 %s"
% (geo.hit, geo.miss, geo.fail, CACHE_FILE))
if __name__ == "__main__":
main()示例数据跑出来的报告是这样的（真实输出，未修饰）：
渠道 事件数 主要地域(占比) C段聚集度 机房特征 海外 查询失败
feed_ads 5 杭州 60%、南京 20%、北京 20% 20% 80% 0% 0
search_ads 5 上海 20%、南京 20%、北京 20% 20% 20% 0% 0
kol_link 5 杭州 40%、北京 20%、上海 20% 20% 20% 20% 0feed_ads 这一档，机房特征 IP 占 80%，配合杭州那 60% 的集中度，基本可以拿去找媒体要明细了。
四、落地时容易踩的坑
阈值别硬抄。 40% 的 C 段聚集度对日投放量几千次的账户是常态，对几十万次的账户就是异常。先用自己 3 个月的正常数据跑一遍，看正常波动区间在哪，再定线。
机房 IP 不等于作弊。 很多正规企业的办公网出口、SaaS 工具回传、测试环境都落在云上，混进来几十个点很正常。真正要盯的是「机房特征 + 极端地域集中 + 高频重复事件」三者同时出现。
查询失败要单独统计。 免费版打满 60 次/分钟之后返回的是限速错误，如果脚本把这类失败直接当成「地域未知」混进分布里，你会得到一份看起来「很分散」的假报告。这也是代码里把 unknown 单独计数的原因。
只做聚合，不做个体画像。 地域统计看的是分布，没必要把「哪个 IP 干了什么」长期留下来。日志保留期限、字段脱敏按公司合规要求走，做地域核验这件事本身不需要精确到个人。
接口选择上，先小后大。 核验脚本每天要跑的 IP 数量级通常是几千到几万，免费版 https://ip9.com.cn/get?ip=<ip>（IP9 官网 https://www.ip9.com.cn，免注册、60 次/分钟、支持 IPv4/IPv6）配缓存完全够用；如果要做实时核验、或者需要 ip_type 区分家庭宽带和机房 IP，再考虑 VIP 版（提供区县、ip_type、ip_asn 字段）。
总结
广告地域数据对不上，先别怪定位不准。把后端日志按渠道摊开、打上归属地标签、和媒体报表并排一比，多数问题当场就有答案：口径不同的，对齐时间窗口；动手脚的，看 C 段聚集度和机房特征；移动端混进来的，把省级出口单独标出来。上面这份脚本可以直接拿来改字段跑，缓存和限速都做好了，跑几千个 IP 只要几分钟。