如何在自定义Telegram bot内部实现定时循环发送价格更新功能

解决方案 #

直接使用python-telegram-bot库内置的JobQueue调度组件实现定时推送，无需引入任何第三方依赖，也不会阻塞原有机器人的命令响应逻辑。

修改后的完整代码如下：

import telegram
from telegram.ext import Updater
from telegram.ext import CommandHandler
from tracker import get_prices

telegram_bot_token = "mybot"

updater = Updater(token=telegram_bot_token, use_context=True)
dispatcher = updater.dispatcher
job_queue = updater.job_queue

# 单独封装价格推送逻辑，每次触发生成全新消息
def send_price(context):
    job = context.job
    chat_id = job.context
    message = ""

    crypto_data = get_prices()
    for i in crypto_data:
        coin = crypto_data[i]["coin"]
        price = crypto_data[i]["price"]
        change_day = crypto_data[i]["change_day"]
        change_hour = crypto_data[i]["change_hour"]
        message += f" {coin}={price:,.5f}$ \nHour Change: {change_hour:.3f}%\nDay Change: {change_day:.3f}%\n\n"

    context.bot.send_message(chat_id=chat_id, text=message)

def start(update, context):
    chat_id = update.effective_chat.id
    # 先立即发送一次当前价格
    send_price(context)
    # 注册5分钟周期的定时推送任务，interval单位为秒，需要更长间隔直接调整该数值即可
    new_job = job_queue.run_repeating(send_price, interval=300, first=0, context=chat_id)
    # 把任务id存到聊天数据里，方便后续取消
    context.chat_data['price_job'] = new_job
    update.effective_message.reply_text("已开启价格推送，每5分钟更新一次，发送/stop可停止推送")

def stop(update, context):
    if 'price_job' in context.chat_data:
        context.chat_data['price_job'].schedule_removal()
        del context.chat_data['price_job']
        update.effective_message.reply_text("已停止价格推送")
    else:
        update.effective_message.reply_text("当前未开启价格推送")

dispatcher.add_handler(CommandHandler("start", start))
dispatcher.add_handler(CommandHandler("stop", stop))
updater.start_polling()
updater.idle()

关键逻辑说明 #

• 所有功能都基于现有库的原生能力实现，不需要引入额外第三方服务或工具
• send_price函数每次执行都会重新拉取价格、生成全新的消息内容，单独调用发送接口，不会和历史消息产生关联，满足每次单独发送新消息的需求
• 定时任务和原有命令响应逻辑并行运行，不会互相阻塞，机器人依然可以正常响应其他命令

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