[Summary] 텔레그램봇: 채권모니터링 Review

언어

python3.7

 

이용한 python 라이브러리

os, time, datetime, requests, pandas, xlrd [, bs4, telegram]

 

상세 내용

증권시장 거래시간의 특정 시점에, 한국거래소에서 채권거래 호가 정보를 수집 후,

한국은행 API로 비교 기준인 최신의 채권 수익률을 확인(xml/json)하여,

전처리(null 데이터를 제외하고, 잔존기간을 일자로 환산,

투자등급/잔존기간/수익률 기준으로 필터링)한 다음,

텔레그램 메신저를 통해 적당한 메시지 포맷으로 전송한다.

여기서 증권시장 거래시간의 특정 시점이란 10시, 12시, 2시 30분, 3시 31분(종가) 처럼 특정 시간을 말한다.

또한 휴장일 정보를 확인(download/view)하여, 익일이 휴장일인 경우 건너뛸 수 있도록 한다.

 

결과 메시지 예시

 

어려웠던 점 / 해결책 / 이후 방향 리뷰

-데이터 수집 관련

    requests.post 방식으로 데이터를 넘겨주는 데이터수집이 처음이라 헤맸다. 개발자도구 - 네트워크를 뜯어보고, 필요한 데이터를 전송(form data 파라미터)하는 것으로 해결했다. 처음 시도했던 작업은 한국거래소 채권 호가 데이터수집이었고, 한국거래소 휴장일 정보 확인에서 한번 더 이용했다. 이 패턴은 이후에 진행한 한국부동산원 데이터수집시에 비교적 상세하게 기록해두었다. --데이터 전송방식 추가 학습 필요

 

-데이터 정리/포맷팅 관련

    (1) python pandas 메소드 자꾸 까먹어서 계속 검색을 해야했다. 이번에 주로 이용한 메소드는 여기 정리해둔다. 주로 인덱싱과 간단한 데이터프레임 조작법이다.

# NULL 값은 -로 대체해서 가져오고, 숫자에서 천의 자리 쉼표는 무시하고 int/float 데이터로 가져온다.
csv = pd.read_csv('./text.csv', na_values='-', thousands=r',')
# year2date 함수를 csv['잔존기간'] 컬럼에 적용 후, csv dataFrame에 'Y2D'라는 컬럼명으로 추가한다.
csv['Y2D'] = csv['잔존기간'].apply(year2date)

# 아래 conditions에 해당하는 조건으로 csv를 조회해서 csv_filtered에 저장한다.
conditions = "신용등급 not in ['B', 'B+', 'B-'] and Y2D <= 365 && 매수수익률 >= 2.0 or 매수수익률 == 0.0"
csv_filtered = csv.query(conditions)

# '거래량' 컬럼 기준으로 내림차순으로 정렬한다.
csv_sorted = csv_filtered.sort_values(by=['거래량'], ascending=False)

# '거래대금', '잔존기간' 컬럼을 삭제한다.
csv_dropped = csv_sorted.drop(['거래대금', '잔존기간'], axis=1)\

'''
# df
    신용등급  가격
1     A    10000
2     B    8000
3     C    4000
'''
# df에서 index 1번 row의 '신용등급' 컬럼 데이터를 가져온다.
print(df.loc[1, '신용등급']) # A

# df에서 row 1번째, column 0번째 데이터를 가져온다. iloc은 실제 로우 인덱스명과 관계없이 0번부터 시작한다.
print(df.iloc[1, 0]) # B

# df에서 index 1번 row의 '신용등급', '가격' 컬럼 데이터를 가져온다.
print(df.loc[1, '신용등급':'가격']) # .loc인 경우 범위의 시작/끝을 포함해서 다 가져온다.
'''
신용등급        A
가격      10000
Name: 1, dtype: object
'''
# df에서 row 1번째, column 0번째부터 1번째까지 데이터를 가져온다.
print(df.iloc[1, 0:1]) # .iloc인 경우 list인덱싱과 같이 시작 포함, 끝 미포함으로 처리한다.
'''
신용등급    B
Name: 2, dtype: object
'''

    (2) python string formatting 관련해서도 검색을 많이 했다. 해당 내용은 여기에 정리해두었다.

a = 98765.4321
print(f"{a}")       # 98765.4321
print(f"{a:>20}")   #           98765.4321
print(f"{a:<20}")   # 98765.4321          /
print(f"{a:^20}")   #     98765.4321      /
print(f"{a:*^20}")  # *****98765.4321*****
print(f"{438765434567654345:,}")  # 438,765,434,567,654,345
print(f"{438765434567654345:,>50,}")
#,,,,,,,,,,,,,,,,,,,,,,,,,,,438,765,434,567,654,345

    (3) python datetime formatting도 마찬가지로 검색을 많이 했다. 해당 내용은 여기에 있다.

import time
import datetime

now = datetime.datetime.now()
finalDay = datetime.datetime(2020, 12, 31)
timedelta = finalDay - now

print("오늘 날짜: " + now.strftime("%Y-%m-%d")) # 오늘 날짜: 2020-12-18
print("현재 시각: " + now.strftime("%H:%M:%S %p")) # 현재 시각: 12:41:10 PM

print("오늘은 " + now.strftime("%Y년도의 " + "%j" + "번째 날, " + "%W" + "번째 주입니다."))
# 오늘은 2020년도의 353번째 날, 50번째 주입니다.
print("올해 마지막 날까지 " + str(timedelta.days) + "일 남았습니다.")
# 올해 마지막 날까지 12일 남았습니다.
print("올해 마지막 날까지 총 " + str(timedelta.total_seconds()) + "초 남았습니다.")
# 올해 마지막 날까지 총 1077529.023712초 남았습니다.

 

 

-로직 관련

    휴장일 관련 로직을 어떻게 만들 것인가가 가장 혼란스러웠다. 평일의 특정시간에만 실행하되, 공휴일 등 휴장일인 경우 건너뛰어야 한다. 실행하지 않는 동안에는 time.sleep()을 주는 것이 가장 우선이었기 때문에, 알람을 맞추는 느낌으로 실행시에 현재 시간을 확인해서 가장 가까운 다음 실행시점을 리턴할 수 있도록 했다. 요일 확인 -> 시간대 확인 -> 다음 시간 리턴, 만약 해당 년도의 휴장일 목록에 포함되면 여기에 +1일으로 구성했다.(여기)

    처음에는 휴장일 확인을 1회만 해서 연휴를 커버하지 못했다. 휴장일이 아닐 때까지 while문으로 휴장일인가?를 확인하는 방식으로 변경하고, 연말 휴장일과 연초 휴장일이 겹치는 2020-12-31 ~ 2021-01-01 사이의 휴장시기를 커버를 위해서 휴장일 여부를 확인할 때 기준 년도를 입력받게 수정했다. 그러고 나서 확인해보니 실제 주기적인 실행은 코드 내에 삽입하는 것보다 외적으로 관리하는 게 더 나을 것 같다. --리눅스 crontab/기타 스케쥴링 방법들

 

-코드 정리 관련

    (1) 코드를 여러 파일에 나누어서 작성 후, import 해서 사용하면서, 다른 폴더에 있는 스크립트를 가져올 때 혼선이 있었다. 해당 이슈는 여기에 정리해두었다. 현재 폴더에서 접근 가능한 폴더.하위폴더.하하위폴더 형태로 마침표로 연결해주면 된다.

# 현재 스크립트의 워킹디렉토리에서 f1/f2/m2.py로 접근가능한 파일을 import 하는 경우

from f1.f2 import m2
import f3.f2.m2 as m2
from f1.f2 import m1, m2, m3 # 여러 스크립트 임포트도 가능

    (2) import하는 경우에, 호출된 스크립트에서 필요한 모듈은 호출된 스크립트에 import 되어있어야 한다. 그 안에서 처리가 잘 끝난다면 굳이 호출하는 스크립트에서 또 호출할 필요는 없다. 설치만 잘 되어있다면 dependency는 알아서 해결된다. --import에 대한 상세 내용이 정리된 페이지를 찾아두었으니(realpython.com/python-import/) 이후 확인하고 다시 정리해두기로 한다.

# A.py
import requests
funcA = requests.get('https://www.naver.com/')
funcB = requests.get

# B.py
import A
print(A.funcA) # <Response [200]>
print(A.funcB('https://www.naver.com/')) # <Response [200]>

    (3) 전체 구성이 데이터 받아오기1,2, 데이터 정리, 데이터 전송의 4단계로 이루어져 있어서, 각각의 기능마다 스크립트를 분리하고, main.py에서 전부 import해서 사용하는 방식을 택했다.(여기) class 연습 차원에서 클래스로 정의해보기도 했는데, 기능을 나누는 단계나 분량이 어느 정도가 적절한지 아직 감이 안 잡힌다. --우수하게 설계된 다른 라이브러리들을 뜯어보는 게 좋을 듯하다.

    (4) 상대/절대 경로 이슈. 해당 파일이 없다고 에러가 나서 무슨 일인가 싶었는데, 데이터를 저장하는 위치가 아래처럼 돼있던 것이다.

filePath = 'data/data_' + str(now) + '.csv'

        테스트시의 working directory는 해당 스크립트가 포함된 디렉토리였는데, 이후에 다른 경로에서 실행했더니 data라는 디렉토리가 없다고 에러가 난 것. 해당 파일 위치가 바뀌지 않는다면 절대경로로 지정하거나, 이슈가 없도록 data 폴더 생성까지 스크립트에 포함시키는 편이 좋겠다.

 

-그외 돌발 이슈

    (1) 채권 수익률 수집과정에서 예상치 못했던 업데이트 일자 관련 이슈가 있었다. 원하는 종목의 최신일자 데이터를 확인하기 위해서는 api를 2번 호출해 1) 종목의 최신일자 조회 2) 1에서 획득한 일자 기준 데이터 조회 순으로 진행해야 했는데, 1)번 데이터가 업데이트 되고 2)번에 해당 데이터가 업데이트되기까지 지연이 있는 경우(10분 정도?)가 있었다. 이때 ValueError가 떴다. 이후에 수정하면서 조회 성공시 해당 일자를 저장해두고, 최신일자 데이터 조회에 실패할 경우, 저장해둔 최신일자를 불러와서 이용하는 방식으로 바꾸었다. 처음에는 스크립트 내부에 변수로 저장했으나, 이후 파일로 분리했다. 디폴트값 설정은 꼭 필요하다.

    (2) 라이브러리 설치과정에서 pip search error 이슈가 있었다. 라이브러리 검색이 안 돼서 확인해보니 과도하게 트래픽이 발생해 현재 비활성화 상태라고 한다. 해당 포스트 작성 이후 2주가 지난 현재 시점에도 여전히 비활성화 상태. 단순한 모듈 검색이라면 pypi 홈페이지 검색으로 대체할 수 있다.

    (3) 2021년 1월 16일 기준으로 작동을 중단했다. Access Denied로 봐서 아이피 차단인 듯한데, 이유를 파악하지 못했다. 이후에 파악되면 추가하기로 한다.

 

 

*덧붙임

구름ide 터미널에서 항상 실행시킬 경우 [이쪽 참조]

로그파일은 nohup.log에 자동으로 생성된다.

# nohup python scheduler.py &

 

종료할 경우, 프로세스 번호 확인해서 종료하면 된다.

# ps -ef
# kill pid