如何在Python中实现每月自动获取美股首个交易日并持续运行？

现有代码仅能单次计算下一个月的美股首个交易日并返回等待时长，无法在服务器上长期循环执行。要实现按月持续运行，需从以下几个方面调整：

• 添加无限循环逻辑：让程序完成当月操作后，自动重新计算下一个月的目标日期，进入下一轮等待，而非执行一次就退出。
• 修正开盘时间获取方式：原代码中nyse["market_open", f"{first_market_day_next_month}"]写法错误，应通过日历的schedule方法获取指定日期的开盘时间。
• 规范时区处理：全程保留纽约时区信息，避免时区转换导致的时间偏差，原代码中tz_localize(None)的操作易引发时间错误。
• 增加异常捕获与容错机制：对API调用、日期计算等环节添加异常捕获，避免单次错误导致整个程序终止，同时添加重试逻辑。
• 明确操作执行节点：在等待到目标开盘时间后执行业务逻辑，完成后再进入下一个周期的计算。

修改后的完整代码 #

import pandas as pd
import mcal
import time

def run_monthly_trading_task():
    nyse = mcal.get_calendar("XNYS")
    nyse_tz = "America/New_York"
    
    while True:
        try:
            # 获取当前纽约时区时间
            current_time = pd.Timestamp.now(tz=nyse_tz)
            
            # 计算下一个月的年份和月份
            if current_time.month == 12:
                next_year = current_time.year + 1
                next_month = 1
            else:
                next_year = current_time.year
                next_month = current_time.month + 1
            
            # 获取下一个月的首个有效交易日（无需逐天遍历，直接调用接口获取）
            start_of_next_month = pd.Timestamp(year=next_year, month=next_month, day=1, tz=nyse_tz)
            end_of_next_month = start_of_next_month + pd.DateOffset(months=1) - pd.DateOffset(days=1)
            valid_days = nyse.valid_days(start_date=start_of_next_month, end_date=end_of_next_month)
            first_market_day = valid_days[0]
            
            # 获取当日开盘时间
            daily_schedule = nyse.schedule(start_date=first_market_day, end_date=first_market_day)
            market_open_time = daily_schedule["market_open"].iloc[0]
            
            # 计算等待时长
            time_diff_seconds = (market_open_time - current_time).total_seconds()
            
            if time_diff_seconds > 0:
                print(f"距离下月首个美股交易日开盘还有 {int(time_diff_seconds)} 秒，等待中...")
                time.sleep(time_diff_seconds)
            else:
                # 处理程序重启时已过开盘时间的边界情况
                print("当前时间已过下月首个交易日开盘时间，立即执行操作")
            
            # --------------------------
            # 在这里插入你的业务逻辑代码
            print("开始执行当月操作...")
            # 示例操作：print("完成当月数据同步/交易指令发送")
            # --------------------------
            
            print("当月操作完成，准备进入下一个周期\n")
            
        except Exception as e:
            print(f"运行出错: {str(e)}，10分钟后重试...")
            time.sleep(600)  # 出错后等待10分钟再重试

if __name__ == "__main__":
    run_monthly_trading_task()

代码核心优化说明 #

1. 用while True实现无限循环，保证程序持续运行
2. 直接通过valid_days接口获取整月有效交易日，替代原逐天遍历的低效逻辑
3. 全程保留纽约时区信息，避免时区转换导致的时间偏差
4. 添加全局异常捕获，出错后自动重试，提升程序稳定性
5. 处理了程序重启时已过开盘时间的边界情况，避免任务遗漏

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