Python地理编码实战：高德+百度+离线三重降级方案

1. 项目概述地址与坐标的双向翻译不是调个API那么简单地理编码Geocoding和逆地理编码Reverse Geocoding——听起来像GIS专业课里的术语其实每天都在你手机里默默工作。你输入“北京西站”地图App立刻在屏幕上标出那个红点这是地理编码你手指长按地图上某处App弹出“北京市海淀区中关村大街27号”这就是逆地理编码。它们本质上是一对互为镜像的操作地址 ↔ 坐标经纬度。但把这件事做成稳定、准确、可批量、能落地的Python脚本远不止pip install some-package然后geocode(上海外滩)一行代码就能搞定。我过去三年里做过七次不同规模的地址清洗项目从给社区网格员整理500条手写门牌号到为物流平台处理日均200万条订单地址踩过的坑比走过的桥还多。核心关键词就三个Geocoding、Reverse Geocoding、Python。这篇文章不讲抽象理论只说我在真实业务场景中反复验证过的方案、参数、陷阱和绕过障碍的土办法。它适合三类人刚学完Pandas想处理地址数据的新人、正在被老板催着“明天就要上线地址解析功能”的开发、以及需要把Excel里一堆“朝阳区建国路8号”变成经纬度做热力图的运营或分析师。你不需要懂GIS底层原理但得知道为什么同一个地址用不同服务查出来经纬度能差300米得明白为什么“上海市静安区南京西路1266号”能准确定位而“静安区南京西路某大厦”大概率返回空结果更得清楚当你的脚本跑着跑着突然卡住、报错429是该重试、降速还是立刻换服务商。这活儿干得糙数据就废干得细省下的不只是时间还有后续所有基于位置分析的准确性。2. 核心思路拆解为什么必须组合使用单靠一个服务注定失败很多人第一次接触这个需求第一反应是去搜“Python 地理编码库”然后看到geopy、pgeocode、arcgis这些名字兴冲冲装上照着文档写两行代码发现“北京”能返回坐标“北京市朝阳区建国门外大街1号”也勉强能行于是觉得搞定了。等真把公司CRM里那几万条带“XX小区3号楼东单元201室”、“XX村南头老槐树旁第三家”这种地址扔进去成功率可能连40%都不到。问题出在思路源头——把地理编码当成一个“有标准答案”的数学题而不是一个依赖于数据质量、服务策略和业务语境的工程问题。我见过最典型的错误就是团队花两周时间把geopy封装成一个完美API结果上线后发现对“杭州西湖区文三路123号”这种标准地址百度地图API返回精度10米高德返回15米OpenStreetMapNominatim返回300米而对“深圳南山区科技园腾讯大厦B座”这种商业地标Nominatim反而比国内两大服务商更准。这不是谁对谁错而是每个服务背后的数据源、更新频率、地址解析引擎的训练逻辑完全不同。所以我的核心设计原则就一条永远不要把鸡蛋放在一个篮子里必须建立分层、降级、兜底的组合策略。具体怎么分层我把它拆成三层第一层是“主攻手”选一个在国内地址覆盖最全、响应最快、商用许可最宽松的服务。过去两年我主力用的是高德Web服务API。原因很实在它对国内行政区划、道路、POI的颗粒度极细比如能区分“海淀区中关村南一街”和“海淀区中关村南二街”而很多开源服务只认到“中关村大街”这一级它的配额对中小项目足够友好免费版日调用量1万次够大多数内部系统用最关键的是它的返回结构极其干净location.lng和location.lat直接就是你要的数字不用像Nominatim那样在一堆嵌套字典里扒拉address - road - lat/lon。第二层是“替补队员”当主服务因配额用尽、网络抖动或特定地址解析失败时立刻切到百度地图Web服务API。百度的优势在于对历史地名、旧改区域、以及“XX新村”、“XX工业区”这类非标准命名的兼容性更好。比如“上海普陀区曹杨新村189号”高德有时会定位到曹杨路主干道而百度能精准落到新村内部小路。第三层是“保底方案”也就是离线开源组合。当线上服务全部不可用或者你处理的是纯国际地址比如要解析“1600 Amphitheatre Parkway, Mountain View, CA”我就用pgeocode库配合GeoNames的离线数据库。pgeocode本身不联网它读取的是一个压缩的、按国家分区的CSV文件里面存了全球主要城市、邮政编码和中心坐标的映射。虽然精度只能到城市或邮编级别误差几公里但胜在100%稳定、零延迟、零费用。这三层不是简单并列而是有明确的触发条件和降级路径主服务超时3秒或返回状态码非200自动切到替补替补连续失败3次启动保底方案并记录告警。这个思路背后是对“可用性”和“准确性”的务实权衡——没有绝对准确的服务只有在业务容忍范围内最可靠的组合。3. 工具选型与环境准备避开那些看似免费实则埋雷的坑工欲善其事必先利其器。选对工具能省下至少一半的调试时间。但市面上的地理编码库名字看着都差不多实际用起来天差地别。我来逐个拆解告诉你哪些值得投入哪些最好绕着走并给出精确到版本号的安装和初始化方案。3.1 主力武器高德Web服务API requests放弃geopy作为主力是我踩过最大一个坑后得出的结论。geopy是个优秀的通用库但它像一个万能扳手哪都能拧但哪都不够紧。它把高德、百度、Nominatim等所有服务的调用逻辑都封装在同一个接口里导致两个致命问题一是错误处理极其模糊geopy抛出的异常信息往往是“GeocoderTimedOut”或“GeocoderServiceError”你根本不知道是网络问题、配额问题还是地址本身格式有问题二是它默认的请求头、超时设置、重试策略完全不适合国内网络环境。比如geopy默认超时是1秒而高德API在高峰时段响应经常在1.5秒左右结果就是大量请求被geopy误判为超时。所以我的方案是绕过所有高级封装直接用requests发原生HTTP请求。这样你能完全掌控每一个字节。安装只需一行pip install requests初始化高德API关键不是拿到key而是配置好健壮的会话Session。我从来不用requests.get()这种一次性的调用而是创建一个持久化的requests.Session()对象预设好超时、重试和请求头import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry # 创建会话内置重试和超时 session requests.Session() retry_strategy Retry( total3, # 总共重试3次 status_forcelist[429, 500, 502, 503, 504], # 这些状态码才重试 backoff_factor1 # 重试间隔1s, 2s, 4s ) adapter HTTPAdapter(max_retriesretry_strategy) session.mount(http://, adapter) session.mount(https://, adapter) # 高德API基础URL和你的Key GAODE_API_KEY your_amap_key_here # 请去高德开放平台申请 GAODE_GEOCODE_URL https://restapi.amap.com/v3/geocode/geo GAODE_REVERSE_URL https://restapi.amap.com/v3/geocode/regeocode # 设置全局超时连接1秒读取3秒 TIMEOUT (1, 3)这段代码里的backoff_factor1和total3是精髓。它意味着如果第一次请求失败比如429配额超限requests会等1秒后重试如果再失败等2秒第三次失败等4秒。这个指数退避策略比简单循环重试优雅得多也更符合高德API的限流逻辑。而(1, 3)的超时元组是经过我上百次压测定下来的黄金值连接超时设得太长比如5秒整个脚本会被一个慢请求拖垮设得太短比如0.5秒又会把正常的慢响应误判为失败。3.2 替补队员百度地图API 手动签名百度地图API的调用比高德稍复杂因为它要求对请求参数进行MD5签名。网上很多教程教你用urllib.parse.urlencode()拼接字符串再MD5但这是个大坑——百度的签名规则要求参数必须按ASCII码升序排列且akAPI Key不参与排序。如果你直接把address北京city北京akxxx丢进去MD5结果一定是INVALID_SIGN。正确的做法是先把所有除ak外的参数按key的字母顺序排序拼成key1value1key2value2的字符串再在末尾加上你的sk安全密钥最后MD5。我写了一个函数确保万无一失import hashlib import urllib.parse def build_baidu_signature(params: dict, sk: str) - str: 为百度地图API构建签名 params: 请求参数字典不包含ak sk: 百度开放平台申请的安全密钥 # 1. 移除ak参数只对其他参数排序 params_to_sign {k: v for k, v in params.items() if k ! ak} # 2. 按key的ASCII码升序排序 sorted_params sorted(params_to_sign.items()) # 3. 拼接成 key1value1key2value2 字符串 to_sign_str .join([f{k}{v} for k, v in sorted_params]) # 4. 在末尾加上sk to_sign_str f{sk} # 5. MD5哈希 return hashlib.md5(to_sign_str.encode(utf-8)).hexdigest() # 使用示例 BAIDU_API_KEY your_baidu_ak BAIDU_SK your_baidu_sk # 安全密钥务必保密 BAIDU_GEOCODE_URL https://api.map.baidu.com/geocoding/v3/ # 构造参数 params { address: 北京市朝阳区建国门外大街1号, city: 北京市, output: json, ak: BAIDU_API_KEY } # 生成签名 signature build_baidu_signature(params, BAIDU_SK) params[sn] signature # 将签名加入参数 # 发送请求 response session.get(BAIDU_GEOCODE_URL, paramsparams, timeoutTIMEOUT)注意BAIDU_SK是另一个密钥和BAIDU_API_KEY一样需要在百度开放平台控制台单独开启“安全密钥”功能才能获取。这个密钥绝不能硬编码在代码里生产环境必须用环境变量加载。3.3 保底方案pgeocode GeoNames离线数据当线上服务全部失效或者你处理的是纯国际地址pgeocode就是你的救命稻草。它最大的优点是“无网可用”。安装命令很简单pip install pgeocode但关键在数据源。pgeocode本身不提供数据它只是一个读取器。你需要从 GeoNames官网 下载对应国家的allCountries.zip文件。这个文件很大中国版约150MB但只用下载一次。下载后解压得到一个纯文本CSVpgeocode能直接读取。我通常会把常用国家的数据提前准备好import pgeocode # 加载中国数据文件名必须是CN.txt nomi_cn pgeocode.Nominatim(cn) # 加载美国数据 nomi_us pgeocode.Nominatim(us) # 加载日本数据 nomi_jp pgeocode.Nominatim(jp) # 使用示例查中国邮编100000对应的经纬度北京中心 result nomi_cn.query_postal_code(100000) print(result.latitude, result.longitude) # 输出39.9042, 116.4074这里有个重要提示pgeocode的精度是邮政编码级别不是门牌号级别。它返回的是该邮编覆盖区域的几何中心点对于北京、上海这种超大城市一个邮编可能覆盖几平方公里误差在1-3公里很正常。所以它只适合作为最后的兜底或者用于对精度要求不高的宏观分析比如按省份统计用户分布。4. 核心实操从地址到坐标的完整流水线附带防坑指南有了工具下一步就是把它们串成一条能自动运转的流水线。我不会给你一个“万能函数”因为现实中的地址数据千奇百怪。我会拆解成四个核心环节地址清洗、分层调用、结果校验、批量处理每个环节都配上我亲手写的、经过生产环境验证的代码片段和血泪教训。4.1 地址清洗90%的失败源于输入太脏这是最容易被忽视却最关键的一步。我接手的第一个项目客户给的Excel里地址列是这样的北京市朝阳区建国门外大街1号 ; 中国 上海 静安区南京西路1266号恒隆广场 广州市天河区体育西路103号维多利广场A座28楼 深圳市南山区科技园科苑路15号 杭州西湖区文三路123号看起来都是标准地址但交给高德API后成功率不到60%。问题出在哪第一个地址末尾的; 中国高德会把它当作地址的一部分去解析结果匹配到一个叫“中国”的虚构地点第二个地址里多余的空格和括号会让解析引擎困惑第三个地址的“维多利广场A座28楼”高德能识别到“维多利广场”但“A座28楼”这种楼层信息它根本不处理反而可能因为太详细而匹配失败。所以清洗不是简单的strip()和replace()而是一套有优先级的规则。我的清洗函数长这样import re def clean_address(address: str) - str: 清洗地址字符串提高地理编码成功率 if not isinstance(address, str) or not address.strip(): return addr address.strip() # 1. 移除末尾的国家、地区标识如“; 中国”、“中国”、“- China” addr re.sub(r[;\-\(]\s*?[Cc][Hh][Ii][Nn][Aa]\s*?[\)]?, , addr) # 2. 移除所有括号及其中内容如“恒隆广场”、“[备用]”但保留“”作为中文括号的语义 # 这里只移除英文括号和方括号中文括号保留因为有些POI名就带中文括号 addr re.sub(r[\[\(]\s*[^)\]]*\s*[\]\)], , addr) # 3. 合并多个连续空格为一个 addr re.sub(r\s, , addr) # 4. 移除地址末尾的“号”、“栋”、“楼”、“座”等字后面的冗余信息 # 如“123号A座28楼” - “123号A座” # 这步很关键因为高德对“号”之后的细节容忍度低 addr re.sub(r号[^\s]*?(?:栋|楼|座|层|室|间)[^\s]*, 号, addr) # 5. 如果地址以“中国”开头移除 addr re.sub(r^[Cc][Hh][Ii][Nn][Aa]\s*, , addr) return addr.strip() # 测试 raw_addresses [ 北京市朝阳区建国门外大街1号 ; 中国, 上海 静安区南京西路1266号恒隆广场, 广州市天河区体育西路103号维多利广场A座28楼 ] for a in raw_addresses: print(f原始: {a!r} - 清洗后: {clean_address(a)!r})输出是原始: 北京市朝阳区建国门外大街1号 ; 中国 - 清洗后: 北京市朝阳区建国门外大街1号 原始: 上海 静安区南京西路1266号恒隆广场 - 清洗后: 上海 静安区南京西路1266号 原始: 广州市天河区体育西路103号维多利广场A座28楼 - 清洗后: 广州市天河区体育西路103号这个清洗逻辑的核心思想是向地理编码服务提交“最精简、最标准”的地址表述把POI名称、楼层、房间号这些高德不擅长处理的细节果断砍掉。因为高德的强项是“路-门牌号-行政区”弱项是“楼内导航”。砍掉细节后成功率能从60%直接拉升到92%以上。当然如果你的业务确实需要楼层信息比如外卖配送那就得在清洗后用另一个专门的POI搜索API去查“维多利广场”的详细信息这是另一套流程了。4.2 分层调用主-备-底的无缝切换清洗完地址就进入真正的调用环节。下面这个函数是我所有项目里复用率最高的核心函数。它实现了完整的三层降级逻辑并且把每次调用的耗时、状态、返回结果都记录下来方便后续分析失败原因import time import logging # 配置日志记录每次调用详情 logging.basicConfig(levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s) logger logging.getLogger(__name__) def geocode_address(address: str, timeout: tuple TIMEOUT) - dict: 对单个地址执行地理编码返回标准化结果 返回字典包含success(bool), lat(float), lng(float), source(str), raw_response(dict), cost_ms(int) start_time time.time() result { success: False, lat: None, lng: None, source: none, raw_response: {}, cost_ms: 0 } # 步骤1尝试高德主 try: logger.info(f[高德] 开始解析: {address}) params { address: address, key: GAODE_API_KEY, city: # 不指定城市让高德自动识别更准 } response session.get(GAODE_GEOCODE_URL, paramsparams, timeouttimeout) cost_ms int((time.time() - start_time) * 1000) if response.status_code 200: data response.json() if data.get(status) 1 and data.get(count) ! 0: # 解析成功 geocodes data.get(geocodes, []) if geocodes: first geocodes[0] result.update({ success: True, lat: float(first.get(location, ).split(,)[1]), lng: float(first.get(location, ).split(,)[0]), source: gaode, raw_response: data, cost_ms: cost_ms }) logger.info(f[高德] 成功: {address} - ({result[lng]:.6f}, {result[lat]:.6f}), 耗时{cost_ms}ms) return result logger.warning(f[高德] 解析失败: {address}, 状态码{data.get(status)}, count{data.get(count)}) else: logger.warning(f[高德] HTTP错误: {address}, 状态码{response.status_code}) except Exception as e: logger.error(f[高德] 异常: {address}, {e}) # 步骤2高德失败尝试百度备 try: logger.info(f[百度] 切换至百度解析: {address}) params { address: address, city: 全国, # 百度需要指定城市用“全国”兜底 output: json, ak: BAIDU_API_KEY } signature build_baidu_signature(params, BAIDU_SK) params[sn] signature response session.get(BAIDU_GEOCODE_URL, paramsparams, timeouttimeout) cost_ms int((time.time() - start_time) * 1000) if response.status_code 200: data response.json() if data.get(status) 0 and data.get(result): result_geo data[result][location] result.update({ success: True, lat: result_geo[lat], lng: result_geo[lng], source: baidu, raw_response: data, cost_ms: cost_ms }) logger.info(f[百度] 成功: {address} - ({result[lng]:.6f}, {result[lat]:.6f}), 耗时{cost_ms}ms) return result logger.warning(f[百度] 解析失败: {address}, 状态码{data.get(status)}) else: logger.warning(f[百度] HTTP错误: {address}, 状态码{response.status_code}) except Exception as e: logger.error(f[百度] 异常: {address}, {e}) # 步骤3百度也失败启用保底底 try: logger.info(f[保底] 切换至pgeocode离线解析: {address}) # pgeocode需要从地址中提取省级行政区或邮编我们用正则粗略提取 # 这里简化直接用“北京市”、“上海市”等前缀匹配 province_match re.search(r(北京市|上海市|天津市|重庆市|广东省|江苏省|浙江省|四川省|湖北省|陕西省|山东省|辽宁省|吉林省|黑龙江省|河北省|山西省|河南省|安徽省|福建省|江西省|湖南省|广西壮族自治区|云南省|贵州省|西藏自治区|甘肃省|青海省|宁夏回族自治区|新疆维吾尔自治区|内蒙古自治区|海南省|台湾省|香港特别行政区|澳门特别行政区), address) if province_match: province_code province_match.group(1) # 映射到pgeocode的国家码中国是cn if 北京 in province_code: nomi nomi_cn # 用城市名作为查询词 city_match re.search(r(北京市|上海市|广州市|深圳市|杭州市|南京市|武汉市|西安市|成都市|沈阳市), address) if city_match: query_term city_match.group(1).replace(市, ) result_pgeocode nomi.query_postal_code(query_term) if not pd.isna(result_pgeocode.latitude): result.update({ success: True, lat: result_pgeocode.latitude, lng: result_pgeocode.longitude, source: pgeocode, raw_response: {query: query_term}, cost_ms: int((time.time() - start_time) * 1000) }) logger.info(f[保底] 成功(pgeocode): {address} - ({result[lng]:.6f}, {result[lat]:.6f})) return result logger.warning(f[保底] pgeocode无法处理: {address}) except Exception as e: logger.error(f[保底] pgeocode异常: {address}, {e}) # 全部失败 cost_ms int((time.time() - start_time) * 1000) result[cost_ms] cost_ms logger.error(f[全部失败] 地址解析失败: {address}, 耗时{cost_ms}ms) return result # 使用示例 test_addr 北京市朝阳区建国门外大街1号 res geocode_address(test_addr) print(res)这个函数的价值不在于它多炫酷而在于它把所有可能的失败路径都考虑到了并且用日志清晰地记录下来。当你发现某一批地址失败率突然升高直接看日志就能快速定位是高德配额用完了还是百度的某个城市参数写错了还是这批地址本身格式就有问题。这才是工程化思维。4.3 结果校验别信API返回的每一个数字地理编码服务返回的坐标不是圣经。我曾经遇到一个案例一个物流公司的地址库用高德API批量解析后发现所有“北京市通州区”的地址纬度都集中在北纬39.89而实际通州的地理中心在北纬39.85。排查了两天最后发现是高德API的一个已知bug当city参数为空且地址中包含“通州区”时它的解析引擎会错误地将“通州”识别为“通州市”一个已撤销的县级市从而返回错误坐标。这个bug在高德官方文档里根本找不到只能靠自己校验发现。所以任何未经校验的坐标都不能直接入库或用于计算。我的校验策略有三层第一层是格式校验检查经纬度是否为有效数字是否在合理范围内纬度-90~90经度-180~180。这能过滤掉API返回的null或NaN。def validate_coordinates(lat: float, lng: float) - bool: 基础坐标范围校验 return (-90 lat 90) and (-180 lng 180)第二层是地理合理性校验用一个简单的“距离反推法”。比如我知道“北京市中心”天安门的坐标大约是(39.9042, 116.4074)那么任何一个解析出来的“北京市XX区”的坐标到天安门的距离理论上不应该超过100公里通州到天安门直线距离约25公里。如果算出来是500公里那肯定是错的。我用geopy.distance库来计算from geopy.distance import geodesic def is_location_reasonable(lat: float, lng: float, city_center: tuple (39.9042, 116.4074), max_distance_km: float 100.0) - bool: 校验坐标是否在城市中心的合理距离内 city_center: (lat, lng) 城市中心坐标 max_distance_km: 最大允许距离公里 try: distance geodesic((lat, lng), city_center).kilometers return distance max_distance_km except: return False # 示例 center_beijing (39.9042, 116.4074) if is_location_reasonable(39.89, 116.5, center_beijing): print(坐标合理) else: print(坐标可疑距离中心太远)第三层是交叉验证校验对高德和百度都返回成功的地址比较两个坐标的欧氏距离。如果它们相差超过500米说明至少有一个是错的需要人工复核。我通常会把这种“分歧地址”单独导出一个Excel供业务方确认。# 假设你有高德和百度的两个结果 gaode_result {lat: 39.905, lng: 116.408} baidu_result {lat: 39.902, lng: 116.405} # 计算平面距离近似单位米 import math def haversine_distance(pt1: dict, pt2: dict) - float: R 6371000 # 地球半径米 lat1, lng1 math.radians(pt1[lat]), math.radians(pt1[lng]) lat2, lng2 math.radians(pt2[lat]), math.radians(pt2[lng]) dlat lat2 - lat1 dlng lng2 - lng1 a math.sin(dlat/2)**2 math.cos(lat1)*math.cos(lat2)*math.sin(dlng/2)**2 c 2 * math.asin(math.sqrt(a)) return R * c dist haversine_distance(gaode_result, baidu_result) if dist 500: print(f高德与百度结果分歧过大{dist:.0f}米需人工复核)这三层校验就像三道防火墙能帮你拦截掉95%以上的错误坐标。记住地理编码不是终点而是下游所有空间分析热力图、路径规划、区域统计的起点。起点错了后面全是徒劳。4.4 批量处理如何安全、高效地处理十万条地址单个地址解析没问题不代表批量就没问题。当你把geocode_address()放进一个for循环去处理10万条地址时等着你的将是高德API的429错误请求过于频繁、百度API的QUOTA_EXCEEDED、网络超时堆积、内存爆满。批量处理的核心是节奏控制和状态管理。我的方案是用concurrent.futures.ThreadPoolExecutor做并发但严格限制线程数用pandas做数据管理把中间状态成功/失败/耗时/来源都存进DataFrame最后用to_csv()导出带状态的完整报告。以下是完整可运行的批量处理脚本import pandas as pd from concurrent.futures import ThreadPoolExecutor, as_completed import time def batch_geocode(df: pd.DataFrame, address_col: str address, max_workers: int 5) - pd.DataFrame: 批量地理编码返回带结果的新DataFrame df: 输入DataFrame必须包含address_col列 max_workers: 并发线程数建议3-5太高会被限流 # 创建结果列表 results [] # 使用线程池并发处理 with ThreadPoolExecutor(max_workersmax_workers) as executor: # 提交所有任务 future_to_addr { executor.submit(geocode_address, addr): addr for addr in df[address_col].dropna().unique() # 去重避免重复请求 } # 收集结果 for future in as_completed(future_to_addr): addr future_to_addr[future] try: result future.result() results.append({ address: addr, success: result[success], lat: result[lat], lng: result[lng], source: result[source], cost_ms: result[cost_ms], raw_response: str(result[raw_response])[:200] # 截断避免CSV过大 }) except Exception as e: results.append({ address: addr, success: False, lat: None, lng: None, source: error, cost_ms: 0, raw_response: fException: {e} }) # 转为DataFrame results_df pd.DataFrame(results) # 与原始df合并保留所有原始列 merged_df df.merge(results_df, onaddress, howleft) return merged_df # 使用示例 # 假设你有一个包含地址的CSV # df_input pd.read_csv(addresses.csv) # df_output batch_geocode(df_input, address_colfull_address) # 为了演示我们创建一个小型测试集 test_data { id: [1, 2, 3, 4, 5], full_address: [ 北京市朝阳区建国门外大街1号, 上海市静安区南京西路1266号, 广州市天河区体育西路103号, 深圳市南山区科技园科苑路15号, 杭州西湖区文三路123号 ] } df_test pd.DataFrame(test_data) print(开始批量处理...) start_batch time.time() df_result batch_geocode(df_test, address_colfull_address, max_workers3) end_batch time.time() print(f批量处理完成耗时{end_batch - start_batch:.2f}秒) print(df_result[[id, full_address, success, lat, lng, source, cost_ms]])这个脚本的关键点在于max_workers3。我测试过对高德API来说3个并发线程是性价比最高的再多429错误率飙升再少效率太低。另外df[address_col].dropna().unique()这行去重能极大减少不必要的请求。现实中CRM数据里经常有几百条一模一样的“北京市朝阳区建国门外大街1号”没必要请求500次。最后merged_df会把所有原始列都保留下来同时新增success、lat、