[python] 모듈 정리하기

전체 코드가 길진 않지만, 작업 단위별로 파일을 따로 두는 게 확인하기 수월할 것 같아 코드를 분리하기로 했다.

폴더에서 스크립트 임포트를 임포트하면서 약간 헤맸어서 여기에 정리해두었다:

hayjo.tistory.com/63

[import] 폴더 안의 스크립트 임포트하기

 

전체 작업은 이렇게 나뉜다.

 

데이터 받아오기1: requests로 한국거래소 사이트에서 OTP 발급받은 후 필요한 데이터 csv형태로 받아오기

데이터 받아오기2: requests로 한국은행 경제통계사이트에서 최신일자 데이터 받아오기

데이터 정리: csv 파일을 pandas로 열어서 필요한 부분만 추출 후, 전송할 포맷으로 변경

메시지 전송: 텔레그램 api를 이용해 실제 메시지 전송

 

 

코드도 작업을 따라 분리해서 scripts 폴더에 정리해두고, 여기에 main.py 파일을 추가해서 실제 실행은 main()에서 하려고 한다.

아래와 같은 형식으로 구성하는 것이 목표.

scripts 폴더 안에 있는 파일들의 main() 함수는 해당 파일 테스트용도다.

ProjectFolder
├── scripts
│   └── 데이터 받아오기1
│   └── 데이터 받아오기2
│   └── 데이터 정리
│   └── 메시지 전송
└── data
│   └── (data.csv) # 실행하면 생성
└── main.py

 

main.py

# -*- coding: utf-8 -*-
import os
import requests
import pandas as pd
import time
import datetime
from bs4 import BeautifulSoup
import telegram
from scripts import scrapyingData as sd, ecosClass as ecos, messageFormatting as mf

def main():
    Key = '인증키'
    now = datetime.datetime.now().strftime("%y%h%d-%H%M%S")
    maxY2D = 365*3
    minYield = 2.0

    filePath = 'data/data_' + str(now) + '.csv'
    if not os.path.exists(filePath): # 없을 경우만 생성
        otp = sd.getOTP()
        time.sleep(1)
        sd.writeFile(otp, filePath)
    
    corpBondAAm = ecos.Ecos(Key, STAT_CODE='060Y001', ITEM_CODE1='010300000', count=19)
    meanData = corpBondAAm.getLatest()
    meanYield = float(meanData['value'])

    conditions = mf.getCondition(meanYield, maxY2D, minYield)

    csv_sorted = mf.processCSV(filePath, conditions)
    msgForm = mf.sendingForm(csv_sorted)

    bot_token = 'token' # bot은 빼고 입력해야된다.
    bot = telegram.Bot(bot_token) # Bot 대문자임에 주의
    chat_id = '00000000'

    bot.sendMessage(chat_id, text=msgForm)


if __name__ == "__main__":
    main()

 

데이터 받아오기1

scrapyingData.py

한국거래소에서 데이터를 받아오는 스크립트

[데이터수집] 한국거래소 정보 가져오기

import requests
from datetime import datetime, timedelta
import time

headers1 = {'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_10_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/39.0.2171.95 Safari/537.36',
           'Referer': 'http://marketdata.krx.co.kr/contents/MKD/05/0502/05020201/MKD05020201.jsp'
           }

headers2 = {'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_10_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/39.0.2171.95 Safari/537.36',
           'Referer': 'http://marketdata.krx.co.kr/'
           }

#def nowMS():
#    return int(float(datetime.now().strftime("%s.%f"))*1000)
# 데이터를 조회할 때 현재 시간을 넘겨줘야 해서 작성했으나, 파일로 다운받으면서 필요 없게된 함수.
# 혹시 이후에 조회시에 사용할 수 있을까 싶어 남겨둠

def getOTP(): # OTP를 발급받기 위한 함수
    #_id = str(nowMS()
    get_req_url = 'http://marketdata.krx.co.kr/contents/COM/GenerateOTP.jspx?name=fileDown&filetype=csv&url=MKD/05/0502/05020201/mkd05020201&bnd_clss_cd=on&mult_byinst_credit_valu_grd=on&remain_pd=on&sort_methd_tp_cd=O&pagePath=%2Fcontents%2FMKD%2F05%2F0502%2F05020201%2FMKD05020201.jsp='
    html = requests.get(url=get_req_url, headers=headers1)
    
    return html.text

def writeFile(code, path): # 필요한 데이터 다운로드 받아서 저장하는 함수
    form_data = {
        'code': code
        }
    _url = 'http://file.krx.co.kr/download.jspx'
    data = requests.get(_url, form_data, headers=headers2).text
    with open(path, 'a+', encoding='utf8') as f:
        f.write(data)
    
def main():
    path = './test.csv'
    code = getOTP()
    writeFile(code, path)

if __name__ == "__main__":
    main()

 

데이터 받아오기2

ecosClass.py

한국은행 경제통계사이트에서 최신 금리 정보 받아오는 스크립트, 여러 항목을 받아올 수도 있을 것 같아 class로 구성함

[ECOS API] 한국은행 경제통계 API 이용 (1) xml

[ECOS API] 한국은행 경제통계 API 이용 (2) json

[Pandas] 데이터 정리 (1)  read_csv(na_values) & apply

[Pandas] 데이터 정리 (2) 필터링(query) 및 정렬

import requests
class Ecos:
    def __init__(self, Key, STAT_CODE, ITEM_CODE1, count):
        ''' count: SStatisticItemList에서 STAT_CODE 조회했을 때 ITEM_CODE1의 위치'''
        self.Key = Key
        self.STAT_CODE = STAT_CODE
        self.ITEM_CODE1 = ITEM_CODE1
        self.base = 'http://ecos.bok.or.kr/api/'
        self.count = count

    def getJSON(self, url):
        return requests.get(url).json()

    def getDate(self): # 최신 일자 받아오기
        url = self.base + 'StatisticItemList/' + self.Key \
              + '/json/kr/' + str(self.count) + '/' + str(self.count) \
              + '/' + self.STAT_CODE
        data = self.getJSON(url)
        return (data['StatisticItemList']['row'][0]['END_TIME'], 
                data['StatisticItemList']['row'][0]['CYCLE'])

    def getLatest(self): # 최신 데이터(기준 금리) 받아오기
        date, cycle = self.getDate()
        url = self.base + 'StatisticSearch/' + self.Key \
              + '/json/kr/1/1/' + self.STAT_CODE + '/' + cycle + '/' \
              + date + '/' + date + '/' + self.ITEM_CODE1
        data = self.getJSON(url)
        if "RESULT" in data:
            url = url.replace(date, self.validDate, 2)
            data = self.getJSON(url)
        returnDict = {
            'date': date,
            'value': float(data['StatisticSearch']['row'][0]['DATA_VALUE'])
        }
        self.validDate = date
        return returnDict
                           

def main():
    Key = 'sample'
    callTotal = Ecos(Key, STAT_CODE='060Y001', ITEM_CODE1='010101000', count=1)
    print(callTotal.getLatest()) # {'date': '20201216', 'value': 0.49}
    callBrokered = Ecos(Key, STAT_CODE='060Y001', ITEM_CODE1='010102000', count=2)
    print(callBrokered.getLatest()) # {'date': '20201216', 'value': 0.49}
    
if __name__ == '__main__':
    main()

 

데이터 정리

messageFormatting.py

csv 파일을 pandas로 열어서 필요한 부분만 추출 후, 전송할 포맷으로 변경

[python f'{formating}'] 봇 메시지 포맷팅

import pandas as pd
import scripts.ecosClass as ecos

def year2date(n):
    y, m, d = 0, 0, 0
    try:
        y, m, d = list(map(int, n.split(".")))
    except:
        pass
    return y*365 + m*30 + d

def processCSV(filePath, conditions):
    csv = pd.read_csv(filePath, na_values='-', thousands=r',')
    csv['Y2D'] = csv['잔존기간'].apply(year2date)
    string = " and ".join(conditions.values())
    csv_filtered = csv.query(string)
    csv_sorted = csv_filtered.sort_values(by=['거래량'], ascending=False)
    return csv_sorted

def getCondition(meanYield, maxY2D, minYield):
    return {
    'creditRate': "신용등급 not in ['BBB', 'BB', 'B', 'BBB+', 'BB+', " \
                  + "'B+', 'BBB-', 'BB-', 'B-']",
    'yield': "`매도최우선호가 수익률` >= (" + str(meanYield) + " * 0.9)",
    'Y2D': "Y2D <= " + str(maxY2D),
    'minYield': "(`매수최우선호가 수익률` >= " + str(minYield) + " or `매수최우선호가 수익률` == 0.0)"
    }

def sendingForm(df):
    message = ''
    for idx in df.index:
        message = message + f"{df.loc[idx, '종목명'][:22]:<22}\n" \
            + f" D-{df.loc[idx, 'Y2D']:>4}" \
            + f"    {df.loc[idx, '신용등급']:^4}" \
            + f"    {df.loc[idx, '가격']:>8,}({df.loc[idx, '수익률']:>1.3f}%)\n" \
            + f" {df.loc[idx, '매도최우선호가 가격']:>8,}({df.loc[idx, '매도최우선호가 수익률']:>1.3f}%)" \
            + f"  {df.loc[idx, '매수최우선호가 가격']:>8,}({df.loc[idx, '매수최우선호가 수익률']:>1.3f}%)"\
            + "\n\n"
    return message

def main():
    minYield = meanYield = 2.215
    maxY2D = 365*3
    conditions = getCondition(meanYield, maxY2D, minYield)
    filePath = '/data/text.csv'
    csv_sorted = processCSV(filePath, conditions)
    print(sendingForm(csv_sorted))

if __name__ == "__main__":
    main()

 

메시지 전송

api를 사용한 덕에 코드가 짧아져서 굳이 분리할 필요가 없을 것 같다.

[TelegramBot] 봇 생성과 메시지전송

import telegram

bot_token = 'token' # bot은 빼고 입력해야된다.
bot = telegram.Bot(bot_token) # Bot 대문자임에 주의
chat_id = '00000000'

bot.sendMessage(chat_id, text="데이터")