如何使用Python每5秒持续轮询JSON URL检测数据变化

要实现单次启动脚本持续轮询检测变动，核心改两个点就行：

• 不要每次检测都重置对比基准：脚本启动时拉取第一次全量数据作为初始基准，每次检测完成后把最新拉取的数据更新为新基准，避免重复上报
• 用无限循环包裹检测逻辑：每轮完成所有URL的拉取、对比、上报后，休眠5秒再进入下一轮检测，不需要手动重复运行脚本

原代码还有几个显性问题需要修复：

1. 函数参数传递不规范：check_new_coins没有接收URL参数，依赖全局变量容易串数据
2. 变量覆盖：get_data里把传入的URL字符串变量赋值为请求响应对象，存在隐性bug
3. 无异常处理：网络波动、接口报错会直接导致脚本崩溃退出
4. 多URL检测时没有独立维护基准，两个接口的数据会混在一起对比

import time
import requests
from datetime import datetime
from requests.exceptions import RequestException

URLS = [
    "https://api.coingecko.com/api/v3/coins/markets?vs_currency=usd&order=market_cap_desc&per_page=250&page=1&sparkline=false",
    "https://api.coingecko.com/api/v3/coins/markets?vs_currency=usd&order=market_cap_desc&per_page=250&page=2&sparkline=false"
]
POLL_INTERVAL = 5  # 轮询间隔，单位秒

def get_coin_ids(url: str) -> set | None:
    """拉取指定接口返回的币种ID集合，带异常捕获"""
    try:
        resp = requests.get(url, timeout=10)
        resp.raise_for_status()  # 接口返回4xx/5xx状态码时直接抛出异常
        data = resp.json()
        return {coin['id'] for coin in data}
    except RequestException as e:
        print(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] 请求失败 {url}: {str(e)}")
        return None
    except KeyError as e:
        print(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] 接口返回格式异常，缺失字段 {str(e)}")
        return None

if __name__ == "__main__":
    # 初始化每个URL对应的基准数据集
    baseline_map = {}
    print("初始化基准数据中...")
    for url in URLS:
        coin_set = get_coin_ids(url)
        while not coin_set:
            print(f"初始化{url}失败，{POLL_INTERVAL}秒后重试")
            time.sleep(POLL_INTERVAL)
            coin_set = get_coin_ids(url)
        baseline_map[url] = coin_set
        print(f"接口 {url} 初始化完成，共加载{len(coin_set)}个币种")
    
    print(f"启动轮询，每{POLL_INTERVAL}秒检测一次变动")
    while True:
        for url in URLS:
            current_coins = get_coin_ids(url)
            if not current_coins:
                # 本次拉取失败，跳过等下一轮重试
                continue
            # 计算新增币种
            new_coins = current_coins - baseline_map[url]
            if new_coins:
                ts = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
                for coin in new_coins:
                    print(f"[{ts}] 检测到新增币种: {coin}  来源接口: {url}")
                # 更新基准为最新数据
                baseline_map[url] = current_coins
        # 本轮所有接口检测完成，等待间隔后进入下一轮
        time.sleep(POLL_INTERVAL)

• 直接用集合（set）存储币种ID，差集计算比列表转set对比效率更高，也自动去重
• 给请求加了10秒超时和状态码校验，遇到网络问题不会直接崩溃，会在下一轮自动重试
• 每个URL独立维护基准数据，不会出现跨接口对比的逻辑错误
• 输出带时间戳，方便定位变动发生的时间
• 如果需要同时检测被移除的币种，只要在对比时加一行removed_coins = baseline_map[url] - current_coins即可拿到下架的币种列表
• 如果需要脚本重启后不丢失基准数据，可以在每次更新baseline_map时把数据序列化存到本地json文件，脚本启动时优先读取本地文件作为初始基准即可
• 公开接口有频率限制，不要把轮询间隔设到3秒以内，避免被封禁IP

内容的提问来源于stack exchange，提问作者Tendekai Muchenje