트레이딩뷰 Webhook Server 만들기(FastAPI라이브러리)

TradingView의 Webhook 서버 만들기

서론

Webhook은 실시간 정보 교환을 위한 강력한 도구로, 특정 이벤트가 발생했을 때 사전에 정의된 HTTP 콜백을 트리거합니다. TradingView의 Webhook을 활용하면 시장 동향, 가격 변동 등 중요한 신호를 즉각적으로 포착하여 자동화된 트레이딩 시스템에 통합할 수 있습니다. 본 포스팅에서는 Python의 FastAPI 라이브러리를 사용하여 안전하고 효율적인 Webhook 서버를 구축하는 방법을 소개합니다.

개발 환경 설정

먼저 Python과 FastAPI를 설치해야 합니다. Python은 공식 웹사이트에서 다운로드할 수 있으며, FastAPI는 pip를 통해 쉽게 설치할 수 있습니다.

pip install fastapi uvicorn

 

이 명령어는 FastAPI와 함께 ASGI 서버인 Uvicorn도 설치합니다.

 

HTTPS 통신을 위한 PEM 파일 생성 및 Web Hook 서버 실행

안전한 데이터 전송을 위해 HTTPS를 설정합니다. OpenSSL을 사용하여 PEM 파일(공개키 및 개인키 포함)을 생성합니다.

openssl req -x509 -newkey rsa:4096 -keyout key.pem -out cert.pem -days 365

 

FastAPI 서버를 시작할 때 uvicorn에 PEM 파일을 전달하여 HTTPS를 활성화합니다.

# webhook_server.py
from fastapi import FastAPI, Request

app = FastAPI()

@app.post("/webhook")
async def handle_webhook(request: Request):
    data = await request.json()
    # 처리 로직 구현
    return {"message": "Received"}

uvicorn main:app --reload --ssl-keyfile=key.pem --ssl-certfile=cert.pem

또는 python을 통해서 직접 실행 할 수도 있습니다.

# webhook_server.py
import uvicorn


from fastapi import FastAPI, Request

@app.post("/webhook")
async def handle_webhook(request: Request):
    data = await request.json()
    # 처리 로직 구현
    return {"message": "Received"}
    
if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=443, ssl_keyfile="key.pem", ssl_certfile="cert.pem")

sudo python webhook_server.py

 

Class를 사용한 메시지 받는 방법

웹 애플리케이션에서 데이터를 수신하고 처리하는 과정을 효율적으로 관리하기 위해, FastAPI는 Pydantic의 BaseModel을 활용한 클래스 기반의 요청 본문 처리를 지원합니다. 이 접근 방식을 통해, 개발자는 데이터의 유효성 검사, 직렬화 및 역직렬화 작업을 간편하게 수행할 수 있습니다.

WebhookMessage 클래스 정의

WebhookMessage 클래스는 TradingView Webhook으로부터 전송된 데이터의 구조를 정의합니다. 이 클래스는 다양한 필드를 포함하며, 각 필드는 거래 플랫폼, 시장, 통화, 심볼, 주문 유형 등의 정보를 나타냅니다.

from pydantic import BaseModel
from typing import Optional

class WebhookMessage(BaseModel):
    key_name: Optional[str]
    group_name: Optional[str]
    platform: str
    market: str
    currency: str
    symbol: str
    orderType: str
    side: str
    price: Optional[float]
    qty_per: str

 

Webhook 데이터 처리 엔드포인트 구현

FastAPI의 라우팅 데코레이터 @app.post("/webhook")를 사용하여, /webhook 경로에 POST 요청을 처리하는 엔드포인트를 구현합니다. 이 엔드포인트는 WebhookMessage 인스턴스를 매개변수로 받아, 전송된 데이터를 구조화된 형태로 처리합니다.

from fastapi import FastAPI
app = FastAPI()

@app.post("/webhook")
async def handle_webhook(message: WebhookMessage):
    key_name = message.key_name
    group_name = message.group_name
    platform = message.platform
    market = message.market
    currency = message.currency
    symbol = message.symbol
    order_type = message.orderType
    side = message.side.lower()
    price = message.price if type(message.price) is float else 0
    qty_per = message.qty_per
    return {"message": "Data processed successfully"}

 

이 방법을 사용하면, Webhook으로부터 수신된 데이터를 효과적으로 처리하고, 애플리케이션의 다른 부분과 쉽게 통합할 수 있습니다. 또한, Pydantic의 유효성 검사 기능을 활용하여 데이터의 정확성을 보장할 수 있습니다.