Anvil 로 웹 앱 만들어서 배포하기 (기능편) - <2> Timer 로 주기적 Background 작업 생성하기

TL;DR : 완성본 체험 링크 (아래 과정만 따라하면 누구나 동일하게 작성 및 배포가 가능하다!)

웹 앱에는 종종 주기적으로 실행해야하는 기능들이 생기는데, 예를 들어서 현재 작업 진행 상태를 자동으로 지속 모니터링을 한다든지 정해진 시간마다 메세지를 생성한다든지 하는 상황이 바로 그렇다.

Anvil에서도 해당 기능을 제공하는데, Timer를 이용하면 퍽 쉽게 구현할 수 있다. 다른 언어로 이런 백그라운드 프로세스를 별도 관리하는게 꽤나 까다로움을 생각하면 매우 편리한 기능이라고 할 수 있겠다.

본 포스팅에서는 이러한 Timer 기능을 간단한 예제와 함께 알아보겠다.

본 포스팅을 따라오기 위해서는 기본적으로 Anvil 로 웹 앱을 배포하는 루틴에 대해 알아야하므로 이전 발행글을 읽고 오길 권장한다.

2023.03.20 - [파이썬으로 웹 App 작성하기] - Anvil 로 웹 앱 만들어서 배포하기 (1) - Getting Started

Anvil 로 웹 앱 만들어서 배포하기 (1) - Getting Started

코인 차트 읽어오기

1) UI

현재 상태를 지속적으로 읽어와야하는 작업의 대표적인 사례라고 한다면 주식 및 코인 차트를 들 수 있겠다. 새로 Blank Project를 만들어서 선택한 코인 마켓에 대해 실시간 캔들차트를 보여주는걸로 예제를 준비해본다. 본 포스팅에서는 Theme 중 Rally Theme을 사용해보도록 하겠다.

Rally Theme 기반의 새 블랭크 프로젝트

제일 먼저 빈 디자인 화면에, 베이스가 될 Card를 놔주고

Card 추가

그 다음으로는 드롭박스와 플롯 툴을 추가해서

드롭다운과 플롯 툴 추가

아래와 같이 기본적인 UI 틀을 완성한다.

완성된 예제 UI

2) 코인 차트 읽어서 표시하기

마켓 리스트를 읽어오고 이를 plot 하는 기능은 아래의 코드블럭으로도 확인 가능하지만, 필자의 다른 포스팅 시리즈 중 

2023.04.17 - [파이썬으로 웹 App 작성하기/파이썬으로 Upbit 자동거래하기] - Anvil과 파이썬 Upbit API로 커스텀 코인 트레이더 어플 만들기 - 1 : 기본 API

Anvil과 파이썬 Upbit API로 커스텀 코인 트레이더 어플 만들기 - 1 : 기본 API

에서 확인할 수 있다.

1. 마켓과 드롭다운

먼저, 현재 거래 가능한 코인 market의 리스트를 받아오는 코드는 아래와 같은데,

def get_market_list(self):
    url = "https://api.upbit.com/v1/market/all?isDetails=true"
    headers = {"accept": "application/json"}
    res = http.request(url=url, method='GET', headers =headers)
    data = json.loads(res.get_bytes().decode('utf-8'))
    return [d['market'] for d in data if 'KRW' in d['market']]

업비트 API를 이용해서 얻은 전체 market 리스트에서 KRW(원화) 거래가능 시장만 남긴 리스트를 반환해주는 함수이다.

이제 이 함수를 Form 코드 아래에 선언해주고, 아래와 같이 초기화 과정에서 시장 리스트를 긁어오게 만든다.

초기화 하면서 마켓 정보 긁어오도록 하기

이제 이 긁어온 self.markets 정보를 아까 UI에 추가해둔 drop down과 연결하기 위해서 

1. 드롭다운 UI 를 선택
2. 우측 사이드바를 스크롤해서 Properties로 이동하여 DATA BINDINGS를 찾은 다음 +ADD 누르기
3. items 항목을 드롭다운 메뉴에서 찾아서 이걸 위에서 정의한 self.markets랑 연결해주기

데이터 바인딩 방법

2. 분봉 차트 읽어오기

위와 같이 마켓 정보 읽어 오기가 끝나면 사용자가 선정한 마켓에 따라서 분봉 차트를 조회하는 부분을 짜본다.

Anvil에서 가장 기본으로 제공하는 차트 그리기 엔진은 Plotly 이므로 일반적으로 차트를 그리는데 잘 쓰이는 matplotlib이 아니라 plotly를 기준으로 작성한 차트 그리기 코드는 아래와 같다.

def plot(self, market):
    ############## 분봉 읽어오기 #####################
    url = f"https://api.upbit.com/v1/candles/minutes/1"
    querystring = {"market": market, "count": 20}
    headers = {"accept": "application/json"}
    res = http.request(url=url, method='GET', headers =headers, data=querystring)
    data = json.loads(res.get_bytes().decode('utf-8'))
    ##################################################
    
    ############## 분봉 정보 처리하기 #####################
    timestamps = [':'.join(d['candle_date_time_kst'].split('T')[1].split(':')[:-1]) for d in data][::-1]
    opens = list(map(float, [d['opening_price'] for d in data]))[::-1]
    highs = list(map(float, [d['high_price'] for d in data]))[::-1]
    lows = list(map(float, [d['low_price'] for d in data]))[::-1]
    closes = list(map(float, [d['trade_price'] for d in data]))[::-1]
    colors = ['green' if opens[i] < closes[i] else 'red' for i in range(len(data))]
    #######################################################
    
    ############## 분봉차트 그리기 #####################
    self.plot_1.data = [go.Candlestick(x=timestamps, open=opens, high=highs, low=lows, close=closes, increasing=dict(line=dict(color='green')), decreasing=dict(line=dict(color='red')))]
    ####################################################

첫 번째 블럭에서 분봉차트 정보를 업비트에서 가져오고, 가져온 정보를 두번째 블럭에서 파싱해서 세번째 블럭에서는 plotly를 이용해서 아까 UI에서 추가해둔 plot 툴(self.plot_1)에 캔들 데이터를 할당해준다.

이제 이 plot 함수를 유저가 드롭다운 선택을 바꿀때마다 수행하게 해주는 코드를 아래와 같이 작성해준다.

드롭다운에 change event 핸들러 추가하기

def drop_down_1_change(self, **event_args):
    """This method is called when an item is selected"""
    self.market = self.drop_down_1.selected_value
    self.plot(self.market)

3. 중간 결과물

이렇게 해놓고 중간 결과물을 확인하기 위해 우상단에 있는 Run 버튼을 눌러보면 아래와 같이 실행이 되는걸 확인할 수 있다. 드롭다운에서 원하는 마켓을 선택하면 최근 20분에 해당하는 분봉차트가 성공적으로 그려지는걸 확인할 수 있다.

20분 분봉차트

이때까지의 코드는 아래와 같다.

from ._anvil_designer import Form1Template
from anvil import *
import json
import plotly.graph_objects as go
Plot.templates.default = "rally"

class Form1(Form1Template):
  def __init__(self, **properties):
    # Set Form properties and Data Bindings.
    self.markets = self.get_market_list()
    self.init_components(**properties)
    self.market = self.markets[0]
    self.plot(self.market)
    # Any code you write here will run before the form opens.
  
  def get_market_list(self):
    url = "https://api.upbit.com/v1/market/all?isDetails=true"
    headers = {"accept": "application/json"}
    res = http.request(url=url, method='GET', headers =headers)
    data = json.loads(res.get_bytes().decode('utf-8'))
    return [d['market'] for d in data if 'KRW' in d['market']]

  def plot(self, market):
    url = f"https://api.upbit.com/v1/candles/minutes/1"
    querystring = {"market": market, "count": 20}
    headers = {"accept": "application/json"}
    res = http.request(url=url, method='GET', headers =headers, data=querystring)
    data = json.loads(res.get_bytes().decode('utf-8'))

    timestamps = [':'.join(d['candle_date_time_kst'].split('T')[1].split(':')[:-1]) for d in data][::-1]
    opens = list(map(float, [d['opening_price'] for d in data]))[::-1]
    highs = list(map(float, [d['high_price'] for d in data]))[::-1]
    lows = list(map(float, [d['low_price'] for d in data]))[::-1]
    closes = list(map(float, [d['trade_price'] for d in data]))[::-1]
    colors = ['green' if opens[i] < closes[i] else 'red' for i in range(len(data))]

    self.plot_1.data = [go.Candlestick(x=timestamps, open=opens, high=highs, low=lows, close=closes, increasing=dict(line=dict(color='green')), decreasing=dict(line=dict(color='red')))]

  def drop_down_1_change(self, **event_args):
    """This method is called when an item is selected"""
    self.market = self.drop_down_1.selected_value
    self.plot(self.market)

3) Timer를 이용해 실시간 차트 업데이트 구현하기

분봉차트의 진면목은 실시간으로 업데이트가 될 때 비로소 드러나게 될 텐데, 지금은 수동으로 다른 마켓으로 갔다가 돌아와야만 업데이트가 되는 반쪽짜리 차트 플롯이라고 할 수 있으므로, 아래와 같이 툴박스에서 Timer를 찾아서 UI에 드래그 해준다. 아무곳이나 드래그 해주면 추가된다.

Invisible Components에 추가된 타이머

이때 이 timer_1을 클릭해서 선택해주고, properties 에서 interval(업데이트를 수행할 주기, 초) 세팅 및 맨 아래의 tick 함수를 할당해준다.

타이머 설정

이때 timer_1_tick 함수를 아래와 같이 작성해주면

def timer_1_tick(self, **event_args):
    """This method is called Every [interval] seconds. Does not trigger if [interval] is 0."""
    self.plot(self.market)

매우 간단하게 실시간 플롯이 완성된다.

Run을 통해서 이를 확인해볼 수 있는데

최종 결과물

미세....하긴 하지만 이건 실제 마켓이 변동이 잘 안되고 있어서니까 넘어가면 실시간으로 현재 거래상황을 반영하는 분봉이 잘 그려짐을 볼 수 있다.

상기 코드를 그대로 필자가 Anvil을 통해 배포하고 있는 링크는 아래와 같다.

https://superior-even-ideal.anvil.app

Rally App 1

무료 계정 누구나 우상단의 publish 버튼을 통해 공개 URL로 본인의 app을 배포할 수 있으므로 관심있는 독자는 언제든 시도해보길 권장한다.