如何在Python交易机器人中定期更新变量并分时执行不同函数？

要实现两个定时任务的互斥执行，核心是用线程锁保证同一时间只有一个函数在运行，避免list_prices被并发修改或读取导致异常。下面提供两种实用方案：

方案一：schedule库 + 线程锁 #

schedule是轻量易用的定时任务库，配合threading.Lock可以快速实现需求：

import schedule
import time
import threading

list_prices = [...]
# 定义线程锁，保证函数互斥执行
task_lock = threading.Lock()

def function1():
    global list_prices
    # 加锁，确保function2不会同时运行
    with task_lock:
        print("开始执行function1更新价格列表")
        # 模拟更新逻辑
        list_prices = [price + 1 for price in list_prices]
        print("function1执行完成")

def function2():
    global list_prices
    # 加锁，确保function1不会同时运行
    with task_lock:
        print("开始执行function2处理价格数据")
        # 模拟处理逻辑
        current_avg = sum(list_prices) / len(list_prices)
        print(f"当前价格平均值：{current_avg}")
        print("function2执行完成")

# 设置定时任务
# 每日固定时间执行（比如凌晨2点）
schedule.every().day.at("02:00").do(function1)
# 每分钟执行一次
schedule.every(1).minutes.do(function2)

# 启动调度循环
while True:
    schedule.run_pending()
    time.sleep(1)

说明：

• with task_lock会自动获取和释放锁，确保两个函数不会同时进入执行逻辑
• schedule.run_pending()需要在循环中持续调用，检查是否有任务需要执行

方案二：APScheduler框架（更适合复杂场景） #

如果需要更灵活的定时配置（比如时区、任务持久化），可以用APScheduler，同样配合线程锁实现互斥：

from apscheduler.schedulers.blocking import BlockingScheduler
import threading

list_prices = [...]
task_lock = threading.Lock()

def function1():
    global list_prices
    with task_lock:
        print("开始执行function1更新价格列表")
        list_prices = [price + 1 for price in list_prices]
        print("function1执行完成")

def function2():
    global list_prices
    with task_lock:
        print("开始执行function2处理价格数据")
        current_avg = sum(list_prices) / len(list_prices)
        print(f"当前价格平均值：{current_avg}")
        print("function2执行完成")

# 初始化调度器
scheduler = BlockingScheduler()

# 添加每日任务（凌晨2点）
scheduler.add_job(function1, 'cron', hour=2)
# 添加每分钟任务
scheduler.add_job(function2, 'interval', minutes=1)

# 启动调度器
scheduler.start()

说明：

• APScheduler的BlockingScheduler会阻塞主线程运行调度逻辑，无需手动写循环
• 可以通过cron表达式设置更精细的定时规则，比如每周一执行等

额外建议 #

尽量避免使用全局变量list_prices，可以用类封装状态和方法，代码更易维护：

import schedule
import time
import threading

class TradingBot:
    def __init__(self):
        self.list_prices = [...]
        self.task_lock = threading.Lock()
    
    def update_prices(self):
        with self.task_lock:
            print("更新价格列表")
            self.list_prices = [p + 1 for p in self.list_prices]
    
    def process_prices(self):
        with self.task_lock:
            print("处理价格数据")
            avg = sum(self.list_prices) / len(self.list_prices)
            print(f"价格平均值：{avg}")

if __name__ == "__main__":
    bot = TradingBot()
    schedule.every().day.at("02:00").do(bot.update_prices)
    schedule.every(1).minutes.do(bot.process_prices)
    
    while True:
        schedule.run_pending()
        time.sleep(1)

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