创建比特币价格通知Telegram聊天机器人时遭遇PytzUsageWarning错误

警告原因 #

这个警告来自APScheduler库，它内部使用了pytz库的zone属性，而pytz已被标记为弃用，官方推荐迁移到标准库的zoneinfo（Python 3.9+）或backports.zoneinfo（Python 3.8及以下）作为时区处理工具。

解决步骤 #

1. 升级APScheduler到最新版本 #

确保使用APScheduler的最新版本，新版本已支持zoneinfo，不再依赖pytz的旧接口：

pip install --upgrade apscheduler

2. 显式配置APScheduler使用zoneinfo #

初始化BlockingScheduler时，指定时区使用zoneinfo格式，比如UTC或本地时区：

from zoneinfo import ZoneInfo  # Python 3.9+
# Python 3.8及以下先安装依赖：pip install backports.zoneinfo，再导入
# from backports.zoneinfo import ZoneInfo

aps = BlockingScheduler(timezone=ZoneInfo("UTC"))
# 也可使用本地时区，例如"Asia/Shanghai"
# aps = BlockingScheduler(timezone=ZoneInfo("Asia/Shanghai"))

3. 修正代码中的功能异常问题 #

你的代码存在几处会导致机器人无法正常运行的错误，一并修正：

• telegram.Bot的消息发送方法是send_message，而非sendmsg
• Alert类的方法存在笔误和错误调用
• 创建DataFrame时集合语法错误
• 避免每次任务执行重复创建机器人实例

修正后的完整代码：

import requests
import telegram
import pandas as pd
from apscheduler.schedulers.background import BlockingScheduler
from zoneinfo import ZoneInfo  # Python 3.9+; 3.8及以下用backports.zoneinfo

# 全局初始化机器人实例，避免重复创建连接
bot_token = "5668522544:AAFqNFcgd5wDBtQbJBhRayfPx9VpVPVjcyQ"
chat_id = 5734902861
core_bot = telegram.Bot(token=bot_token)

# 测试API连接
dt = requests.get('https://min-api.cryptocompare.com/data/price?fsym=BTC&tsyms=USD,EUR')
print(dt.text)

# 获取更新测试
updates = core_bot.getUpdates()
for i in updates:
    print(i.message)


class Chatbot:
    def __init__(self, token, chat_id):
        self.core = telegram.Bot(token)
        self.updater = Updater(token)
        self.chat_id = chat_id

    def sendmsg(self, text):
        self.core.send_message(chat_id=self.chat_id, text=text)

    def stop(self):
        self.updater.stop()


class Alert(Chatbot):
    def __init__(self, token, chat_id):
        super().__init__(token, chat_id)
        self.updater.stop()

    def add_command_handler(self, cmd, func):
        self.updater.dispatcher.add_handler(CommandHandler(cmd, func))

    def start_bot(self):
        self.sendmsg("比特币价格提醒机器人已启动！")


# 初始化调度器并指定时区
aps = BlockingScheduler(timezone=ZoneInfo("UTC"))


def push():
    dt = requests.get("https://min-api.cryptocompare.com/data/price?fsym=BTC&tsyms=USD,EUR")
    # 使用全局机器人实例发送消息
    core_bot.send_message(chat_id=chat_id, text=dt.text)

    # 修正DataFrame创建逻辑
    price_data = dt.json()
    price_now = pd.DataFrame({
        "USD": [price_data["USD"]],
        "EUR": [price_data["EUR"]]
    })

    # 处理CSV存储，兼容文件不存在的情况
    try:
        data = pd.read_csv("ALERTBOT.csv")
    except FileNotFoundError:
        data = price_now
    else:
        data = pd.concat([data, price_now], ignore_index=True)
    
    data = data.loc[:, ['USD', 'EUR']]
    data.to_csv("ALERTBOT.csv", index=False)


aps.add_job(push, 'interval', seconds=60)
aps.start()

额外说明 #

• Python 3.8及以下版本需先安装backports.zoneinfo：pip install backports.zoneinfo，再替换导入语句为from backports.zoneinfo import ZoneInfo
• 全局初始化机器人实例可避免重复创建连接，提升运行效率
• 新增CSV文件不存在的异常处理，解决首次运行报错问题

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