Appearance
贷后催收怎么按属地分案？用 IP 归属地编排外呼时段、分派案件给区域团队（Python 实现）
贷后团队的日常矛盾很具体：案件池里几千条，谁打、什么时候打，规则大多是「按身份证地址分省」。可身份证地址是十年前办卡时填的，手机号可能是老家办的号，申请时填的单位早就离职了。坐席打过去，对方一句「我人在外地」就把整条话术打乱，属地分案的准确率也就三成上下。
倒是有一个信号一直在被浪费：债务人这次点还款链接、登录 App、打开短信短链时的出口 IP。它不精确，但足够回答「这人大致在哪个省」这个分案级别的问题。
先把边界说死：IP 归属地通常只到城市，用它去定位债务人具体住在哪栋楼，既不准也不合规，本文不碰这件事。下面讲的用法只有两个——把案件优先分给对应大区的催收团队，以及按债务人大致所在城市排外呼时段。
一、贷后分案到底卡在哪
把催收系统里「分案」拆开看，卡点其实不在算法，在数据源：
- 手机号归属地只能证明卡在哪儿办的。 异地务工的人，号是老家的、人在珠三角，按号分案就会把案件丢给一个隔着两千公里的坐席。
- 身份证地址更新滞后。 户籍地和实际居住地不一致是常态，尤其对流动人口。
- 申请时的居住地址是手填的。 有人图省事填老家，有人填的是中介地址，还有人填单位地址——人早不在那儿了。
- 历史还款地只覆盖有还款记录的人。 刚进 M1（逾期一个月内）的客户，很多还没产生过任何有效位置数据。
网络位置恰好补在这个空档上：只要客户还在用手机点我们的链接，出口 IP 就是新的。运营商宽带的 IP 段和省份有对应关系，手机流量出口通常落在号卡归属省，虽然会随漫游变，但「大致在哪个大区」这个粒度上比身份证地址靠谱得多。
二、方案设计
链路很简单，难的是几个取舍：
案件入池 → 取该客户最近 30 天内的有效 IP（App 埋点 / 短信短链点击 / 官网登录）
→ 查归属地（带本地缓存 + 限速）→ 省 / 大区 / 运营商
→ 与手机号归属地、身份证地址交叉打分
→ 分派区域团队 + 计算外呼时段窗口 → 落库用「最近一次有效 IP」，不要实时查。 催收外呼不是风控拦截，没有实时性要求。案件批量每晚跑一遍就够了，顺手把结果写进案件表。
归属地只做优先级，不做唯一判据。 代理、加速器、企业出口、流量漫游都会让 IP 指向错的地方。工程上的做法是给每个信号一个权重：IP 大区与手机号归属省一致 → 高置信；两者冲突 → 标「位置存疑」，交给人工或按原规则走，不要硬判。
缓存比代码重要。 一个外包催收公司、一家工厂宿舍、一栋写字楼的出口 IP 会被几百条案件共用，按 IP 做缓存（TTL 7 天）能把请求量压掉七八成。免费版限速 60 次/分钟，缓存是让它够用的前提，不是可选项。
外呼时段按省取，不按市取。 同一个省的城市作息一致，用 prov 当键维护一张时段表就够了，省得为三千个区县配置。
查不到归属地不阻塞分案。 超时、ret=400（非法 IP）、限速都属于信息缺失，缺失时走身份证地址兜底，别让一条案件卡在队列里。
三、Python 实现
下面这段是能在本地直接跑通的版本：SQLite 做 IP 缓存，自研令牌桶控速，requests 调接口，最后输出分案结果和外呼时段。
python
# -*- coding: utf-8 -*-
"""贷后案件按属地分派 + 外呼时段编排"""
import json
import sqlite3
import time
import threading
from datetime import datetime, timedelta, timezone
import requests
API = "https://ip9.com.cn/get"
QPS_LIMIT = 55 # 免费版 60 次/分钟，留点余量
CACHE_TTL_DAYS = 7
DB = "ip_geo_cache.sqlite3"
CN_TZ = timezone(timedelta(hours=8))
# 大区 → 属地催收团队（按自家组织结构改）
TEAM_BY_AREA = {
"华东": "hd-team", "华南": "hn-team", "华北": "hb-team",
"华中": "hz-team", "西南": "xn-team", "西北": "xb-team",
"东北": "db-team",
}
DEFAULT_TEAM = "headquarters-pool"
# 省份 → 允许外呼的时间窗口（时:分），合规底线，按自家制度调
CALL_WINDOW = (9, 20) # 09:00 - 20:00
OVERSEAS_WINDOW = (14, 18) # 境外案件只留一个白天窄窗口
class RateLimiter:
"""简单令牌桶：不够就睡，别去撞接口的限速。"""
def __init__(self, per_minute: int):
self.interval = 60.0 / per_minute
self.lock = threading.Lock()
self.last = 0.0
def wait(self):
with self.lock:
now = time.monotonic()
gap = self.interval - (now - self.last)
if gap > 0:
time.sleep(gap)
self.last = time.monotonic()
limiter = RateLimiter(QPS_LIMIT)
_session = requests.Session()
_session.headers.update({"User-Agent": "ip9-collection-router/1.0"})
_conn = sqlite3.connect(DB, check_same_thread=False)
_conn.execute(
"CREATE TABLE IF NOT EXISTS geo (ip TEXT PRIMARY KEY, payload TEXT, ts REAL)"
)
_conn.commit()
_db_lock = threading.Lock()
def _cache_get(ip: str):
cutoff = time.time() - CACHE_TTL_DAYS * 86400
with _db_lock:
row = _conn.execute(
"SELECT payload, ts FROM geo WHERE ip = ?", (ip,)
).fetchone()
if row and row[1] >= cutoff:
return json.loads(row[0])
return None
def _cache_put(ip: str, payload: dict):
with _db_lock:
_conn.execute(
"INSERT OR REPLACE INTO geo VALUES (?, ?, ?)",
(ip, json.dumps(payload, ensure_ascii=False), time.time()),
)
_conn.commit()
def lookup(ip: str, retries: int = 2):
"""查归属地。返回 dict；查不到返回 None（调用方走兜底）。"""
cached = _cache_get(ip)
if cached is not None:
return cached
for attempt in range(retries + 1):
limiter.wait()
try:
resp = _session.get(API, params={"ip": ip}, timeout=3)
body = resp.json()
except (requests.RequestException, ValueError) as exc:
print(f"[warn] {ip} 请求失败: {exc}")
time.sleep(1.5 * (attempt + 1))
continue
ret = body.get("ret")
if ret == 200 and body.get("data"):
data = body["data"]
_cache_put(ip, data)
return data
if ret == 429: # 触发限速，退避后重试
time.sleep(3 * (attempt + 1))
continue
if ret == 400: # 非法 IP，没必要重试
print(f"[warn] {ip} 非法地址，跳过")
return None
time.sleep(1.0 * (attempt + 1))
return None
def resolve_area(ip: str):
"""把 IP 解析成 (大区, 省份, 运营商, 是否境内)。"""
data = lookup(ip)
if not data:
return None
country_code = (data.get("country_code") or "").lower()
return {
"area": data.get("big_area") or "",
"prov": data.get("prov") or "",
"country": data.get("country") or "",
"isp": data.get("isp") or "",
"domestic": country_code == "cn",
}
def assign_team(case: dict):
"""IP 信号 + 手机号归属省交叉打分，决定团队与置信度。"""
geo = resolve_area(case.get("last_ip", "")) if case.get("last_ip") else None
phone_prov = case.get("phone_prov") or ""
if not geo:
return DEFAULT_TEAM, "no_ip_signal", phone_prov
if not geo["domestic"]:
return "overseas-desk", "overseas_ip", geo["country"] or "境外"
if phone_prov and geo["prov"] and phone_prov != geo["prov"]:
# 网络位置与号卡归属冲突：挂总部池人工确认，别硬分给某个区域团队
return DEFAULT_TEAM, "location_conflict", geo["prov"]
team = TEAM_BY_AREA.get(geo["area"], DEFAULT_TEAM)
conf = "high" if (phone_prov and phone_prov == geo["prov"]) else "medium"
return team, conf, geo["prov"]
def call_window(prov: str, domestic: bool):
"""返回今天的可外呼时间段字符串。"""
today = datetime.now(CN_TZ).date()
if not domestic:
start = datetime.combine(today, datetime.min.time(), CN_TZ).replace(
hour=OVERSEAS_WINDOW[0]
)
end = start.replace(hour=OVERSEAS_WINDOW[1])
return f"{start:%H:%M}-{end:%H:%M}(境外窗口)"
# 部分省份习惯晚一些，可按需单独放宽
late_provinces = {"广东", "福建", "浙江"}
end_hour = CALL_WINDOW[1] + 1 if prov in late_provinces else CALL_WINDOW[1]
start = datetime.combine(today, datetime.min.time(), CN_TZ).replace(
hour=CALL_WINDOW[0]
)
return f"{start:%H:%M}-{end_hour:02d}:00"
def run(cases):
plan = []
for case in cases:
team, conf, prov = assign_team(case)
domestic = conf != "overseas_ip"
plan.append(
{
"case_id": case["case_id"],
"team": team,
"confidence": conf,
"prov": prov or "未知",
"window": call_window(prov, domestic),
}
)
return plan
if __name__ == "__main__":
cases = [
{"case_id": "C20260914001", "last_ip": "114.114.114.114", "phone_prov": "江苏"},
{"case_id": "C20260914002", "last_ip": "8.8.8.8", "phone_prov": "广东"},
{"case_id": "C20260914003", "last_ip": "240e:3b1:2f00:1a00::1", "phone_prov": "广东"},
{"case_id": "C20260914004", "last_ip": "", "phone_prov": "山东"},
]
for row in run(cases):
print(row)实际跑出来的结果大致是这样：
{'case_id': 'C20260914001', 'team': 'hd-team', 'confidence': 'high', 'prov': '江苏', 'window': '09:00-20:00'}
{'case_id': 'C20260914002', 'team': 'overseas-desk', 'confidence': 'overseas_ip', 'prov': '美国', 'window': '14:00-18:00(境外窗口)'}
{'case_id': 'C20260914003', 'team': 'hn-team', 'confidence': 'high', 'prov': '广东', 'window': '09:00-21:00'}
{'case_id': 'C20260914004', 'team': 'headquarters-pool', 'confidence': 'no_ip_signal', 'prov': '山东', 'window': '09:00-20:00'}四条案件里有三种不同的信源状态（IP 高置信、境外、无 IP），分案结果都不相同，这正是这套逻辑的价值：不是把案件分得更准，而是让分不准的那些能被识别出来，不混在一起。
四、上线前要想清楚的几件事
案件跟人走，别跟 IP 绑。 IP 是动态的，宽带拨号一天换一次、手机切基站就换。分案结果落库之后，下次跑批如果 IP 变了，不要直接把案件从 A 团队挪到 B 团队——那样坐席会疯。合理的做法是「首分按 IP，重分需连续两次一致」。
企业出口和代理要单独识别。 一家外包催收公司、一个电话营销平台的号码池，出口 IP 可能是同一个 IDC 网段，几十条案件全落到同一个城市。判断办法是用 IP 类型（家庭宽带 / 企业 / IDC 机房），这类字段免费版没有，VIP 版的 ip_type 提供。落到规则上就是：ip_type = IDC 时降低 IP 信号权重。
外呼时段是合规红线，不是运营偏好。 催收相关的行业规范对通话时段和频次都有约束，各地细则不同，代码里的 CALL_WINDOW 只是底线占位，实际必须对齐自家合规口径，别把 21:00 当成可外呼的默认值。同时注意区分「客户所在地的白天」和「系统时区」——上面的代码统一用东八区算窗口，境外案件单独走窄窗口，就是为了避免半夜打过去。
IP 属地不是住址，别写进话术。 坐席开口就说「您现在在南京对吧」，客户的第一反应是「你们在监控我」。属地信息只用在内部路由和排班，不对外披露，也不用于施压。IP 属于个人信息，留存周期和字段范围按站内那篇《IP 地址是个人信息吗》的口径收敛，风控只需要聚合结果。
缓存和限速要成对出现。 深圳这种出口 IP 高度集中的城市，一晚上几千条案件可能只用到几百个不同 IP。缓存 TTL 7 天加上按 IP 去重，免费版的 60 次/分钟完全够；真要到分钟级十几万次（比如坐席端实时查），再考虑 VIP 版（18 万次/分钟）或私有化部署，别在免费额度上反复试错。
总结
贷后分案这件事，本质是把「客户在哪儿」这个模糊问题，用一个粗糙但新鲜的信号回答到「大区」这个粒度就够用了。落地顺序建议：先只加一条规则——IP 大区与手机号归属省一致时提优先级，冲突时挂总部池；跑一两个月看坐席接通率和投诉率的变化，再决定要不要叠加 ip_type、历史还款地这些信号。
查归属地这一步用 IP9 的免费接口就行：https://ip9.com.cn/get?ip=<IP>，不传参数时返回调用方自己的归属地，IPv4 和 IPv6 都支持，返回字段里有国家、省、市、邮编、区号、运营商、国内大区（big_area）和城市中心经纬度，免注册、无需鉴权，免费版 60 次/分钟，批量跑记得带上缓存和限速。官网：https://www.ip9.com.cn