import json import os.path import constant from utils import tool, hx_qc_value_util @tool.singleton class TradingDatesManager: def __init__(self): self.__trading_dates = [] self.load_data() def get_trading_dates(self): return self.__trading_dates def load_data(self): """ 从文件中加载交易日期 :return: """ if not os.path.exists(constant.TRADING_DATES_PATH): return with open(constant.TRADING_DATES_PATH, mode='r', encoding='utf-8') as f: datas = json.load(f) self.__trading_dates = datas def get_pre_trading_day(self, now_day): """ 获取前一个交易日 :return: """ for day in reversed(self.__trading_dates): if day < now_day: return day return None def get_next_trading_day(self, now_day): """ 获取后一个交易日 :return: """ for day in self.__trading_dates: if day > now_day: return day return None # 获取多个交易日 def get_pres_trading_days(self, now_day, count): """ 获取当前日期的前 count 个交易日(从近到远排序)。 :param now_day: 当前日期(datetime.date 或 datetime.datetime) :param count: 需要获取的交易日数量(正整数) :return: 日期列表,例如 [前1个交易日, 前2个交易日, ...] :raises: ValueError 如果 get_pre_trading_day() 返回无效值或 count 非法 """ days = [] temp = now_day for _ in range(count): temp = self.get_pre_trading_day(temp) if temp is None: print(f"get_pre_trading_day() 返回无效值") days.append(temp) return days @classmethod def update_trading_dates(cls): """ 更新交易日历 :return: 日期列表 """ # 获取前后10个交易日的数据 dates = [] now_day = tool.get_now_date_str() for i in range(100): day = hx_qc_value_util.get_previous_trading_date(now_day) dates.append(day) now_day = day dates.sort() now_day = dates[-1] for i in range(10): day = hx_qc_value_util.get_next_trading_date(now_day) dates.append(day) now_day = day dates.sort() if dates: with open(constant.TRADING_DATES_PATH, mode='w', encoding='utf-8') as f: f.write(json.dumps(dates)) return dates