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
|
