Scheduler¶

! pip install APScheduler

Crontab 작업을 활용 (리눅스 기본)
class sched.scheduler() 파이썬 기본모듈의 활용
https://pypi.org/project/schedule/ 의 활용
https://pypi.org/project/APScheduler/ 의 활용

참고 블로그

In [4]:

# %reset
holiday_2020 = """2020-01-01,2020-01.24,2020-01-27,2020-04-30,2020-05-01,
2020-05-05,2020-09-30,2020-10-01,2020-10-02,2020-11-19,2020-12-25"""

1 APScheduler¶

! pip install APScheduler

01 실행 Interval 을 정의하기¶

실행 간격을 지정 (상대적 TimeSet)

In [8]:

def exec_interval(): 
    print('인터벌 함수의 실행 : exec interval')

# 예약방식 interval로 설정, 10초마다 한번 실행    
from apscheduler.schedulers.blocking import BlockingScheduler
sched = BlockingScheduler()
sched.add_job(exec_interval, 'interval', seconds=1, id="test_2")
sched.add_job(exec_interval, 'interval', seconds=3, id="test_2")
# sched.start()

Out[8]:

<Job (id=test_2 name=exec_interval)>

02 실행 조건을 정의하기¶

절대적 Time Set

In [9]:

def exec_cron(): 
    print('Cron 함수의 실행 : exec cron')

# cron 사용
# id는 고유번호로 겹치면 => 'Job identifier (test_1) conflicts with an existing job'
sched.add_job(exec_cron, 'cron', second='*/5', id="test_1")
sched.add_job(exec_cron, 'cron', minute='*/5', second='10,30')
sched.add_job(exec_cron, 'cron', minute="59",  second='10', id="test_2")
sched.add_job(exec_cron, 'cron', day_of_week="0-5", hour='23', minute='30')
# ex) (5분)10,30초, (10분)10,30초, (15분)10,30초 : 스케줄링 시작
sched.add_job(exec_cron, 'cron', day_of_week="0-6", minute='*/5', second='10,30')
# sched.start()

Out[9]:

<Job (id=ee70736de68d4bc39b77f59e03f55cb0 name=exec_cron)>

03 실행 조건의 정의 2¶

파라미터로 실행조건 추가하기

In [4]:

# 파라미터를 받아서 실행하는 Job
def test(test_str, test_str2):
    print("Hello: %s, %s " % (test_str, test_str2))

# args 에 값은 반드시 배열로 입력, 넣지 않으면 다음의 오류를 출력합니다
# ValueError: The list of positional arguments is longer than the target callable can handle (allowed: 0, given in args: 2)
sched.add_job(test, 'cron', minute="1", second='*/5', id="test_11", args=["테스트", "반가워요"])

Out[4]:

<Job (id=test_11 name=test)>

In [2]:

! pip install requests-cache

Collecting requests-cache
  Downloading https://files.pythonhosted.org/packages/7f/55/9b1c40eb83c16d8fc79c5f6c2ffade04208b080670fbfc35e0a5effb5a92/requests_cache-0.5.2-py2.py3-none-any.whl
Requirement already satisfied: requests>=1.1.0 in /home/momukji/Python/python/lib/python3.6/site-packages (from requests-cache)
Requirement already satisfied: certifi>=2017.4.17 in /home/momukji/Python/python/lib/python3.6/site-packages (from requests>=1.1.0->requests-cache)
Requirement already satisfied: chardet<3.1.0,>=3.0.2 in /home/momukji/Python/python/lib/python3.6/site-packages (from requests>=1.1.0->requests-cache)
Requirement already satisfied: urllib3!=1.25.0,!=1.25.1,<1.26,>=1.21.1 in /home/momukji/Python/python/lib/python3.6/site-packages (from requests>=1.1.0->requests-cache)
Requirement already satisfied: idna<2.9,>=2.5 in /home/momukji/Python/python/lib/python3.6/site-packages (from requests>=1.1.0->requests-cache)
Installing collected packages: requests-cache
Successfully installed requests-cache-0.5.2

2 Running Python¶

! pip install APScheduler

In [8]:

from muyong.stock import Krx, Think, Shinhan
from muyong.util  import Telegram
t = Telegram()

import requests_cache, os # requests Cache 로 속도 높이기
from datetime import date

def krx_crawler():
    requests_cache.install_cache('demo_cache')
    now      = date.today()
    date_txt = now.strftime('%m%d')
    
    if now.weekday() not in [0,1,2,3,4]:
        print("today is not a Business day!")
        return None
    
    else: # Business Day 인 경우
        codes = Krx().get_code()
        t.msg('-- 장 마감 후 거래원 수집시작 --')
        df    = Think().get_trader_sql(codes) #, db_memory=True
        df.to_csv('csv/'+ date_txt + '_top_trader.csv', index=None)
        t.msg('상위 거래원 수집완료, 수집끝!')
        
        # 오늘이 금요일인 경우
        if now.weekday() == 4: 
            df = Shinhan().get_trader_sql(codes) #, db_memory=True
            df.to_csv('csv/' + date_txt + '_all_trader.csv', index=None)
            t.msg('거래원별 거래량 수집완료')
    os.remove("demo_cache.sqlite")

In [11]:

from apscheduler.schedulers.blocking import BlockingScheduler
sched = BlockingScheduler()
# sched.add_job(krx_crawler, 'interval', seconds=1, id="test_2")
sched.add_job(krx_crawler, 'cron', day_of_week="0-5", hour='15', minute='50', id="krx_1")
# sched.start()

Out[11]:

<Job (id=krx_1 name=krx_crawler)>

In [18]:

date.today().strftime('%m-%d')

Out[18]:

'09-30-00-09-1569769200'

In [10]:

def get_name():
    return 'games'

type(get_name())

Out[10]:

str

In [5]:

app_name = get_name()
name = f'{app_name}:user-list'
name

Out[5]:

'games:user-list'