[스케쥴링] 특정 시간에 실행하기 - 휴장일 커버

파이썬에서 datetime.datetime object끼리는 크기 비교가 된다는 점을 이용해,

특정 시간을 기준으로 대기/실행 작업을 수행한다.

대기할 경우, datetime.datetime 객체와 datetime.timedelta를 더해 다음 기준 시간을 계산한다.

time.sleep(datetime.timedelta(days=1, seconds=10).total_seconds())로 적당한 휴지시간을 둘 수 있다.

+ 휴장일을 커버할 수 있도록 구성했다

---

 

[python] 모듈 정리하기 까지해서 한 번 보내는 사이클은 완료되었다.

이제 해당 작업을 장이 열리는 시간에 반복하면 된다.

 

목표

하루에 확인은 4번만 하기로 하고, 주간에만 실행되니까,

월요일부터 금요일까지 10시 20분, 12시 20분, 2시 50분, 3시 51분에 실행해야 한다.

(20분 지연 데이터인 점을 감안하면 실제로는 10시, 12시, 2시 30분, 3시 31분(종가) 데이터인 셈이다.)

 

혹시나 다음 날이 휴장일일 수 있으니, 매일 마지막 시간에는 다음 영업일이 휴장인지 여부를 체크하고,

필요한 경우 추가로 휴지시간을 두어야 한다.

 

시간 출력 포맷 관련해서는 이쪽에서 다루었다. [datetime] 시간포맷: datetime.strftime("%Y-%m-%d")

다음 영업일이 휴장인지 확인하는 단계는 이쪽: [데이터수집] 내일은 휴장일? 한국거래소 확인 (1) download

 

로직은 아래와 같다. 한 사이클마다 마지막 단계에서 아래 함수를 호출할 예정이다.

 

오늘 일자를 확인: 개장일인가?(월~금요일인가?)

-> O: 시간을 확인: 개장시간 전인가?

-> X: 가장 가까운 영업일의 10시 20분으로 예약 (종료)

    -> O: 오늘 10시 20분으로 예약 (종료)

    -> X: 장 종료 이후인가?

         -> O: 가장 가까운 영업일의 10시 20분으로 예약 (종료)

         -> X: 2시 50분이 지났는가? [여기부터 for문]

             -> O: 오늘 3시 51분으로 예약 (종료)

             -> X: 12시 20분이 지났는가?

                 -> O: 오늘 2시 50분으로 예약 (종료)

                 -> X: 10시 20분이 지났는가?

                     -> O: 오늘 12시 20분으로 예약 (종료)

                     -> X: 오늘 10시 20분으로 예약 (종료)

 

 

import time
from datetime import datetime

# 전송 조건 일자
wakeUp = {
    'day': [0, 1, 2, 3, 4],
    'time': sorted([(10, 20), (12, 20), (14, 50), (15, 51)]),
    'nextDay': { # 일~목요일은 다음날, 금~토요일은 3일 후, 2일 후
        0: 1,
        1: 1,
        2: 1,
        3: 1,
        4: 3,
        5: 2,
        6: 1
    }
}

def setAlarm(wakeUp, current=None):
    if current is None:
        now = datetime.datetime.now()
    else:
        now = current
    if now.weekday() in wakeUp['day']:  # 개장일인가?
        before = datetime.datetime(now.year, now.month, now.day,  # 장 시작
                                   wakeUp['time'][0][0], wakeUp['time'][0][1])
        after = datetime.datetime(now.year, now.month, now.day,   # 장 종료
                                   wakeUp['time'][-1][0], wakeUp['time'][-1][1])
        if now <= before:  # 개장 시간 전이면
            return before 
        elif now <= after:  # 장 중이면(장 종료 전이면)
            for t in sorted(wakeUp['time']):
                setted = datetime.datetime(now.year, now.month,
                                           now.day, t[0], t[1])
                if now < setted:
                    return setted

    # 주말 or 장 종료 이후
    se = datetime.datetime(now.year, now.month, now.day,  # 오늘 기준 첫 타임
                           wakeUp['time'][0][0], wakeUp['time'][0][1])
    plusDay = datetime.timedelta(days=wakeUp['nextDay'][now.weekday()])
    return se + plusDay

 

이쪽은 테스트용 함수

years = range(2020, 2025)
months = range(1, 13)
days = range(1, 32)
hours = range(0, 24)
mins = range(0, 60)

for year in years:
    for day in days:
        test = datetime.datetime(year, 12, day, 11, 12, 22)
        sample = setAlarm(wakeUp, test)
        print(test, test.strftime("%a"), sample, sample.strftime("%a"))
        
'''
2020-12-01 11:12:22 Tue  2020-12-01 12:20:00 Tue
2020-12-02 11:12:22 Wed  2020-12-02 12:20:00 Wed
2020-12-03 11:12:22 Thu  2020-12-03 12:20:00 Thu
2020-12-04 11:12:22 Fri  2020-12-04 12:20:00 Fri
2020-12-05 11:12:22 Sat  2020-12-07 10:20:00 Mon
2020-12-06 11:12:22 Sun  2020-12-07 10:20:00 Mon
2020-12-07 11:12:22 Mon  2020-12-07 12:20:00 Mon
2020-12-08 11:12:22 Tue  2020-12-08 12:20:00 Tue
2020-12-09 11:12:22 Wed  2020-12-09 12:20:00 Wed
...
2024-12-25 11:12:22 Wed 2024-12-25 12:20:00 Wed
2024-12-26 11:12:22 Thu 2024-12-26 12:20:00 Thu
2024-12-27 11:12:22 Fri 2024-12-27 12:20:00 Fri
2024-12-28 11:12:22 Sat 2024-12-30 10:20:00 Mon
2024-12-29 11:12:22 Sun 2024-12-30 10:20:00 Mon
2024-12-30 11:12:22 Mon 2024-12-30 12:20:00 Mon
2024-12-31 11:12:22 Tue 2024-12-31 12:20:00 Tue
'''

목표대로 잘 나온다.

 

이제 휴장일 여부를 포함해서 메인함수를 실행할 수 있도록 세팅한다.

def isHoliday():
    # 다음날 휴장일인지 확인하는 함수
    # 현재는 테스트용으로 True / False 번갈아가며 나오게만 구성
    while True:
        yield True
        yield False

def takeaNap(wakeUpTime):
    bedtime = (wakeUpTime - datetime.datetime.now()).total_seconds()
    print(bedtime, wakeUpTime)
    time.sleep(bedtime)

def scheduler(wakeUp, func, isHoliday):
    while True:
        wakeUpTime = setAlarm(wakeUp)
        if isHoliday():
            wakeUpTime = wakeUpTime + datetime.timedelta(days=1)
        takeaNap(wakeUpTime)
        func()

혹시 시간 계산하는 동안 지연이 있을까봐 마지막에 takeaNap()에서 최종 대기시간을 다시 구하도록 구성했다.

scheduler(wakeUp, main, isHoliday().__next__)

 

---

연휴 처리 관련 보충

이슈

2020-12-31에 실행했을 때, 2021-01-01로 지정되는 오류가 발생했다. 금요일이 휴일이기 때문에, 의도대로라면 다음 월요일인 20201-01-04가 출력되어야 한다.

 

원인

위에서 비워두었던 isHoliday() 이후 휴일처리가 단순 +1일인데서 발생했다. 위의 로직은 단순히 내일은 휴장일인가만 체크하는데, 그렇게 되면 연휴인 경우, 내일도 휴장, 모레도 휴장인 경우 및 연말~연초의 연도 변경 지점을 커버하지 못한다.

 

수정사항

1. isHoliday()는 다음날이 아니라 특정 일자를 입력받고 해당 일자가 휴장인지를 확인하는 함수가 되어야 한다. 그리고 입력받은 날짜 기준으로 연도를 확인해서, 년이 넘어가도 처리할 수 있도록 한다.

2. 휴장일 여부는 setAlarm() 안에서 진행되어야 한다. 금요일이 휴장인 경우, 단순히 하루를 추가하면 다음 일정이 토요일에 세팅되는 이슈가 생기기 때문에, 기존에     # 주말 or 장 종료 이후 케이스를 처리했던 부분에서 휴장일까지 한 번에 다루는 편이 깔끔하다. 따라서 isHoliday()는 setAlarm() 안으로 들어가야 한다.

3. scheduler()에서는 isHoliday() 핸들링이 빠져야한다.

 

# 전체 코드

scripts/holiday.py 연도 입력 받기 및 기본값 관련해서 이쪽의 코드를 약간 수정했다.

import requests
import time

_userAgent = 'Mozilla/5.0 (Windows Phone 10.0; Android 4.2.1; Microsoft; Lumia 950)'
# otp url 정보
_otpReferer = 'http://marketdata.krx.co.kr/contents/MKD/01/0110/01100305/MKD01100305.jsp'
_otpURL = 'http://marketdata.krx.co.kr/contents/COM/GenerateOTP.jspx?bld=MKD/01/0110/01100305/mkd01100305_01&name=form&_='
# view url 정보
_viewReferer = 'http://marketdata.krx.co.kr/mdi'
_viewURL = 'http://marketdata.krx.co.kr/contents/MKD/99/MKD99000001.jspx'

_otp_headers = {
                'User-Agent': _userAgent,
                'Referer': _otpReferer
              }
_view_headers = {
                 'User-Agent': _userAgent,
                 'Referer': _viewReferer
               }


def getOTP(otpURL, otp_headers):
    nowTime = str(int(round(time.time() * 1000)))  # 함수 안에 넣으면서 람다를 지웠다
    otp = requests.get(url=otpURL + nowTime, headers=otp_headers)
    return otp.content


def getFormData(OTP, year):
    return {
               'gridTp': "KRX",
               'search_bas_yy': str(year),  # '2020' # 년도
               'code': OTP,  # 위에서 획득한 OTP를 여기 넣어주자
               'pagePath': '/contents/MKD/01/0110/01100305/MKD01100305.jsp'
             }


def getData(viewURL, FormData, view_headers):
    return requests.get(viewURL, FormData, headers=view_headers)


def getHolidayList(year=time.gmtime().tm_year, otp_headers=_otp_headers,
                   view_headers=_view_headers, otpURL=_otpURL,
                   viewURL=_viewURL):
    # 외부에서 import해서 쓸 때, 헤더 입력이 번거로울 경우를 대비해 디폴트 정보를 넣어두기로 했다.
    OTP = getOTP(otpURL, otp_headers)
    time.sleep(1)
    formData = getFormData(OTP, year)
    data = getData(viewURL, formData, view_headers).json()
    return [each['calnd_dd_dy'] for each in data['block1']]


def main():
    hList = getHolidayList(time.gmtime().tm_year, _otp_headers, _view_headers, _otpURL, _viewURL)
    return hList


if __name__ == '__main__':
    main()
    

 

scheduler.py

setAlarm(): 이슈 부분인 의 # 장 종료 or 주말 및 기타 휴장일 부분을 수정했다. 휴장이 아닐 때까지 휴장일 여부를 반복해서 검사하고, 휴장일인 경우, +1일이 아니라 wakeUp['nextDay']의 해당 날짜로 넘긴다.

isHoliday(): 연도 검색을 추가했다.

import time
import datetime
from scripts.holiday import getHolidayList

# 전송 조건 일자
wakeUp = {
    'day': [0, 1, 2, 3, 4],
    'time': sorted([(10, 20), (12, 20), (14, 50), (15, 51)]),
    'nextDay': {  # 일~목요일은 다음날, 금~토요일은 3일 후, 2일 후
        0: 1,
        1: 1,
        2: 1,
        3: 1,
        4: 3,
        5: 2,
        6: 1
    }
}


def setAlarm(wakeUp, current=None):
    if current is None:
        now = datetime.datetime.now()
    else:
        now = current
    if now.weekday() in wakeUp['day']:  # 개장일인가?
        before = datetime.datetime(now.year, now.month, now.day,  # 장 시작
                                   wakeUp['time'][0][0], wakeUp['time'][0][1])
        after = datetime.datetime(now.year, now.month, now.day,   # 장 종료
                                  wakeUp['time'][-1][0], wakeUp['time'][-1][1])
        if now <= before:  # 개장 시간 전이면
            return before
        elif now <= after:  # 장 중이면(장 종료 전이면)
            for t in sorted(wakeUp['time']):
                setted = datetime.datetime(now.year, now.month,
                                           now.day, t[0], t[1])
                if now < setted:
                    return setted

    # 장 종료 이후 or 주말 및 기타 휴장일
    firstTime = datetime.datetime(now.year, now.month, now.day,  # 오늘 기준 첫 타임
                                  wakeUp['time'][0][0], wakeUp['time'][0][1])

    def plusDay(x):  # 다음 영업일까지 남은날짜 구하기
        return datetime.timedelta(days=wakeUp['nextDay'][x.weekday()])
    nextTime = firstTime + plusDay(now)
    while isHoliday(nextTime):  # 해당 일자가 쉬는 날인 동안
        nextTime = nextTime + plusDay(nextTime)  # 다음날로 넘기기
    return nextTime


def takeaNap(wakeUpTime):
    bedtime = (wakeUpTime - datetime.datetime.now()).total_seconds()
    print(bedtime, wakeUpTime)
    time.sleep(bedtime)


def isHoliday(wakeUpTime):
    yymmdd = wakeUpTime.strftime("%Y-%m-%d")  # 리스트가 2020-01-01 형태라서 맞춰주기
    if yymmdd in getHolidayList(wakeUpTime.strftime("%Y")):
        return True
    return False
    

def scheduler(wakeUp, func):
    while True:
        wakeUpTime = setAlarm(wakeUp)
        print(wakeUpTime)
        takeaNap(wakeUpTime)
        func()


def main():
    def a():  # 테스트용 함수
        print('helloWorld')
    scheduler(wakeUp, a)


if __name__ == "__main__":
    main()