PK!qD=busiday/__init__.pyimport pendulum as pm from typing import List, Optional from enum import Enum from busiday.stockmarketclock import ( Schedule, StockmarketClock, StockmarketClockDataRow, make_stockmarket_clock_data_from_url, ) DATA: StockmarketClock = make_stockmarket_clock_data_from_url( "https://s3-ap-northeast-1.amazonaws.com/busiday-dev-result-storage/busiday_data.json" ) class WorkTimeState(Enum): WORKING = 0 MORNING = 1 NIGHT = 2 HOLIDAY = 3 LUNCH = 4 def now(exchange: str, the_dt: Optional[pm.DateTime] = None) -> pm.DateTime: """現在の時間を取引所タイムゾーンで返す。 """ data_row: StockmarketClockDataRow = DATA.data[exchange] now_ = the_dt or pm.now("UTC") return now_.in_tz(data_row.tz) def _schedule_on_date(data_row: StockmarketClockDataRow, the_date: pm.Date) -> Optional[Schedule]: if _not_workday_p(data_row, the_date): return None elif the_date in data_row.halfdays: idx = data_row.halfdays.index(the_date) return data_row.halfdaySchedules[idx] else: return data_row.regularSchedule def worktime_state(exchange: str, the_dt: Optional[pm.DateTime] = None) -> WorkTimeState: """時刻the_dtでのステート文字列を返す。 """ data_row: StockmarketClockDataRow = DATA.data[exchange] now_ = now(exchange, the_dt) date_ = now_.date() time_ = now_.time() sch = _schedule_on_date(data_row, date_) if sch is None: return WorkTimeState.HOLIDAY elif time_ < sch.openTime: return WorkTimeState.MORNING elif sch.lunchBeginTime and sch.lunchBeginTime < time_ < sch.lunchEndTime: return WorkTimeState.LUNCH elif sch.closeTime < time_: return WorkTimeState.NIGHT else: return WorkTimeState.WORKING def _workday_p(data_row: StockmarketClockDataRow, the_date: pm.Date) -> bool: return the_date.day_of_week not in {pm.SATURDAY, pm.SUNDAY} and the_date not in data_row.holidays def _not_workday_p(data_row: StockmarketClockDataRow, the_date: pm.Date) -> bool: return the_date.day_of_week in {pm.SATURDAY, pm.SUNDAY} or the_date in data_row.holidays def workday_p(exchange: str, the_date: pm.Date) -> bool: """the_dateが営業日か返す述語。 """ data_row: StockmarketClockDataRow = DATA.data[exchange] return _workday_p(data_row, the_date) def worktime_p(exchange: str, the_dt: Optional[pm.DateTime] = None) -> bool: """現在時間が営業時間かを返す述語。 """ state = worktime_state(exchange, the_dt) return state == WorkTimeState.WORKING def workday(exchange: str, the_dt: Optional[pm.DateTime] = None, workday_offset: int = 0) -> pm.Date: """th_date時点について、次のような定義のworkday_offsetで示される日付を返す。 - workday_offset=0: その時点で最も最近に訪れた営業日 (Day0) - workday_offset=-1: Day0の前営業日 - workday_offset=+1: Day0の翌営業日 """ data_row: StockmarketClockDataRow = DATA.data[exchange] # 1. workday_offset=0 の指す日付をまず計算する now_ = now(exchange, the_dt) date_ = now_.date() time_ = now_.time() sch: Optional[Schedule] = _schedule_on_date(data_row, date_) if sch is None: # 休日 tmp_date = date_ # その日以降の平日を得る while _not_workday_p(data_row, tmp_date): tmp_date = tmp_date.subtract(days=1) zero_date = tmp_date elif time_ < sch.openTime: # 営業日で開始時間より前ならそれより前の日がzeroになる tmp_date = date_.subtract(days=1) while _not_workday_p(data_row, tmp_date): tmp_date = tmp_date.subtract(days=1) zero_date = tmp_date else: # 営業日で開始時間を過ぎている zero_date = date_ # 2. 引数 workday_offset=N の指示通りオフセットを加える offset_date = zero_date step_days = 1 if workday_offset >= 0 else -1 for _ in range(abs(workday_offset)): offset_date = offset_date.add(days=step_days) while _not_workday_p(data_row, offset_date): offset_date = offset_date.add(days=step_days) return offset_date def workdays(exchange: str, the_dt: pm.DateTime, workday_offset_list: List[int]) -> List[pm.Date]: return [workday(exchange, the_dt, offset) for offset in workday_offset_list] def _timeevent_at_the_dt(exchange: str, the_dt: pm.DateTime, workday_offset: int, event: str) -> pm.DateTime: data_row: StockmarketClockDataRow = DATA.data[exchange] date_ = workday(exchange, the_dt, workday_offset) sch = _schedule_on_date(data_row, date_) time_ = getattr(sch, event) if time_ is None: return None dt = pm.datetime(date_.year, date_.month, date_.day, time_.hour, time_.minute, time_.second, tz=data_row.tz) return dt def open_time_at_the_dt(exchange: str, the_dt: Optional[pm.DateTime] = None, workday_offset: int = 0) -> pm.DateTime: return _timeevent_at_the_dt(exchange, the_dt, workday_offset, "openTime") def lunch_begin_time_at_the_dt( exchange: str, the_dt: Optional[pm.DateTime] = None, workday_offset: int = 0 ) -> pm.DateTime: return _timeevent_at_the_dt(exchange, the_dt, workday_offset, "lunchBeginTime") def lunch_end_time_at_the_dt( exchange: str, the_dt: Optional[pm.DateTime] = None, workday_offset: int = 0 ) -> pm.DateTime: return _timeevent_at_the_dt(exchange, the_dt, workday_offset, "lunchEndTime") def close_time_at_the_dt(exchange: str, the_dt: Optional[pm.DateTime] = None, workday_offset: int = 0) -> pm.DateTime: return _timeevent_at_the_dt(exchange, the_dt, workday_offset, "closeTime") def tz(exchange: str) -> str: data_row: StockmarketClockDataRow = DATA.data[exchange] return data_row.tz def _timeevent_on_the_date(exchange: str, the_date: pm.Date, event: str) -> pm.DateTime: data_row: StockmarketClockDataRow = DATA.data[exchange] sch = _schedule_on_date(data_row, the_date) if sch is None: return None time_ = getattr(sch, event) if time_ is None: return None dt = pm.datetime( the_date.year, the_date.month, the_date.day, time_.hour, time_.minute, time_.second, tz=data_row.tz ) return dt def open_time_on_the_date(exchange: str, the_date: pm.Date): return _timeevent_on_the_date(exchange, the_date, "openTime") def lunch_begin_time_on_the_date(exchange: str, the_date: pm.Date): return _timeevent_on_the_date(exchange, the_date, "lunchBeginTime") def lunch_end_time_on_the_date(exchange: str, the_date: pm.Date): return _timeevent_on_the_date(exchange, the_date, "lunchEndTime") def close_time_on_the_date(exchange: str, the_date: pm.Date): return _timeevent_on_the_date(exchange, the_date, "closeTime") PK!B B busiday/stockmarketclock.pyimport pendulum as pm import requests from dataclasses import dataclass from typing import List, Dict, Optional @dataclass class Schedule(object): openTime: pm.Time = None closeTime: pm.Time = None lunchBeginTime: Optional[pm.Time] = None lunchEndTime: Optional[pm.Time] = None def asjson(self): return { "openTime": str(self.openTime), "closeTime": str(self.closeTime), "lunchBeginTime": str(self.lunchBeginTime) if self.lunchBeginTime else None, "lunchEndTime": str(self.lunchEndTime) if self.lunchEndTime else None, } @classmethod def loadjson(cls, obj): return cls( **{ "openTime": pm.parse(obj["openTime"]).time(), "closeTime": pm.parse(obj["closeTime"]).time(), "lunchBeginTime": ( pm.parse(obj["lunchBeginTime"]).time() if obj["lunchBeginTime"] else None ), "lunchEndTime": ( pm.parse(obj["lunchEndTime"]).time() if obj["lunchEndTime"] else None ), } ) @dataclass class StockmarketClockDataRow(object): tz: str regularSchedule: Schedule halfdays: List[pm.Date] halfdaySchedules: List[Schedule] holidays: List[pm.Date] def asjson(self): return { "tz": self.tz, "regularSchedule": self.regularSchedule.asjson(), "halfdays": [each.to_date_string() for each in self.halfdays], "halfdaySchedules": [each.asjson() for each in self.halfdaySchedules], "holidays": [each.to_date_string() for each in self.holidays], } @classmethod def loadjson(cls, obj): return cls( **{ "tz": obj["tz"], "regularSchedule": Schedule.loadjson(obj["regularSchedule"]), "halfdays": [pm.parse(each).date() for each in obj["halfdays"]], "halfdaySchedules": [ Schedule.loadjson(each) for each in obj["halfdaySchedules"] ], "holidays": [pm.parse(each).date() for each in obj["holidays"]], } ) @dataclass class StockmarketClock(object): syncAt: List[pm.date] data: Dict[str, StockmarketClockDataRow] def asjson(self): return { "syncAt": self.syncAt.to_datetime_string(), "data": {key: val.asjson() for key, val in self.data.items()}, } @classmethod def loadjson(cls, obj): return cls( **{ "syncAt": pm.parse(obj["syncAt"]), "data": { key: StockmarketClockDataRow.loadjson(val) for key, val in obj["data"].items() }, } ) def make_stockmarket_clock_data_from_url(url: str) -> StockmarketClock: res = requests.get(url) if res.status_code != 200: raise RuntimeError() result = StockmarketClock.loadjson(res.json()) return result PK!HnHTUbusiday-0.3.2.dist-info/WHEEL A н#Z;/"d&F[xzw@Zpy3Fv]\fi4WZ^EgM_-]#0(q7PK!H-4X#! busiday-0.3.2.dist-info/METADATAQK0+򨰆n(0p>_Bt*%|p*pe2:G%(Żg~Df26^)}.:r,j#ZUG942x?wmQ ÷cVF+ڡ?w tRcT@].xd{nduPy1tTLdwL5aPxIA?ĆSv)/'f}]8wTOb>Qq#iὰ%r*QvWXv PK!Hhmbusiday-0.3.2.dist-info/RECORDu̻v0oIPãaP" o׷C9۝n4YL2 *knjr{{*é|ǫoxjof@"ZDW?Դ(IW$ߩO Fp2-?F>+h)fO_x>9Y?KoƼZ,}W*ºw@eo!.\ 4)Gen~Z*u[6|5R$RD(~] PK!qD=busiday/__init__.pyPK!B B busiday/stockmarketclock.pyPK!HnHTUp'busiday-0.3.2.dist-info/WHEELPK!H-4X#! 'busiday-0.3.2.dist-info/METADATAPK!Hhm`)busiday-0.3.2.dist-info/RECORDPKo*