본문 바로가기
  • 테크에 관한 모든 것.
IT

[python] 비트코인 거래소 공지사항 크롤링 & 텔레그램 전송 프로그램

by @TA 2024. 10. 30.

1. 개요

이번 포스팅에서는 국내/외 주요 암호화폐 거래소 공지사항을 실시간으로 크롤링하여

텔레그램을 통해 알람 전송하는 프로그램을 작성해봤습니다.

 

바이낸스, 코인베이스, 업비트 같은 대형 거래소의 공지사항은 시세에 큰 영향을 미칠 수 있으므로

프로그램을 통해 중요한 발표나 업데이트를 빠르게 확인하고, 투자 전략에 반영할 수 있도록 하는 것이 주 목적입니다.


2. 환경 설정 : 패키지 설치

아래 코드를 실행하여 필요한 파이썬 패키지를 설치합니다.

# 필수 패키지 설치
!pip install requests beautifulsoup4 python-telegram-bot

 

  • requests: 웹 페이지에 HTTP 요청을 보내기 위해 사용합니다.
  • beautifulsoup4: HTML을 파싱하여 필요한 데이터를 쉽게 추출할 수 있습니다.
  • python-telegram-bot: 텔레그램 API와 통신할 수 있도록 도와주는 라이브러리입니다.

3. 텔레그램 봇 생성 및 설정

  1. 텔레그램에서 BotFather를 검색하여 시작합니다.
  2. /newbot 명령을 입력하여 새로운 봇을 생성하고 이름을 설정합니다.
  3. 생성된 봇의 **토큰(Token)**을 복사해 둡니다.
# 텔레그램 봇 설정
from telegram import Bot

# 텔레그램 봇 토큰
TELEGRAM_TOKEN = 'YOUR_TELEGRAM_BOT_TOKEN'  # BotFather에게 받은 토큰을 입력하세요
bot = Bot(token=TELEGRAM_TOKEN)

4. 거래소 공지사항 크롤링 코드 작성

(1) Binance 공지사항 크롤링

import requests
from bs4 import BeautifulSoup

def fetch_binance_announcements():
    url = "https://www.binance.com/en/support/announcement"
    response = requests.get(url)
    soup = BeautifulSoup(response.text, "html.parser")

    announcements = []
    for item in soup.find_all('a', class_='css-1ej4hfo'):
        title = item.get_text()
        link = f"https://www.binance.com{item['href']}"
        announcements.append((title, link))
    return announcements

 

(2) Coinbase 공지사항 크롤링

def fetch_coinbase_announcements():
    url = "https://blog.coinbase.com/latest"
    response = requests.get(url)
    soup = BeautifulSoup(response.text, "html.parser")

    announcements = []
    for item in soup.find_all('a', class_='ds-link'):
        title = item.get_text()
        link = item['href']
        announcements.append((title, link))
    return announcements

 

(3) Upbit 공지사항 크롤링

def fetch_upbit_announcements():
    url = "https://upbit.com/service_center/notice"
    response = requests.get(url)
    soup = BeautifulSoup(response.text, "html.parser")

    announcements = []
    for item in soup.find_all('a', class_='notice-link'):
        title = item.get_text().strip()
        link = f"https://upbit.com{item['href']}"
        announcements.append((title, link))
    return announcements

5. 크롤링 결과 실시간 전송 코드 작성

각 거래소의 공지사항을 텔레그램으로 전송하는 함수를 작성합니다.

def send_announcement_to_telegram(bot, chat_id, title, link):
    message = f"📢 {title}\n🔗 {link}"
    bot.send_message(chat_id=chat_id, text=message)

# 예시
chat_id = 'YOUR_TELEGRAM_CHAT_ID'
for title, link in fetch_binance_announcements():
    send_announcement_to_telegram(bot, chat_id, title, link)

6. 실행 및 테스트

이제 모든 코드를 통합하여 실시간으로 공지사항을 전송할 수 있도록 합니다.

이를 주기적으로 실행하거나 스케줄링하여 최신 공지사항을 자동으로 확인할 수 있습니다.

import time

chat_id = 'YOUR_TELEGRAM_CHAT_ID'

while True:
    for title, link in fetch_binance_announcements():
        send_announcement_to_telegram(bot, chat_id, title, link)
    time.sleep(60 * 5)  # 5분마다 실행

7. 에러 및 트러블슈팅

주요 오류 해결

  1. 텔레그램 API 오류 : bot.send_message 호출 시 발생하는 오류는 주로 chat_id가 잘못 설정되었을 때 발생합니다. 올바른 채팅 ID를 사용했는지 확인해주세요.
  2. HTTP 응답 오류 : 일부 거래소에서 403 Forbidden 오류가 발생할 수 있습니다. 이 경우 IP 주소가 차단된 것일 수 있으므로, VPN 또는 프록시 서버를 사용하여 접근해보세요.

8. 전체 소스코드

위에서 설명한 코드 전체 내용을 합친 통합 코드입니다.

이 코드를 실행하면 Binance, Coinbase, Upbit의 공지사항을 주기적으로 확인하여 텔레그램으로 실시간 전송하게 됩니다.

 

필요한 패키지를 설치한 후, YOUR_TELEGRAM_BOT_TOKEN과 YOUR_TELEGRAM_CHAT_ID를 입력하면 바로 실행할 수 있도록 작성되었습니다.

# 필요한 패키지 설치
!pip install requests beautifulsoup4 python-telegram-bot

# 모듈 임포트
import requests
from bs4 import BeautifulSoup
from telegram import Bot
import time

# 텔레그램 봇 설정
TELEGRAM_TOKEN = 'YOUR_TELEGRAM_BOT_TOKEN'  # BotFather에게 받은 토큰을 입력하세요
chat_id = 'YOUR_TELEGRAM_CHAT_ID'  # 텔레그램에서 자신의 채팅 ID를 입력하세요
bot = Bot(token=TELEGRAM_TOKEN)

# Binance 공지사항 크롤링 함수
def fetch_binance_announcements():
    url = "https://www.binance.com/en/support/announcement"
    response = requests.get(url)
    soup = BeautifulSoup(response.text, "html.parser")

    announcements = []
    for item in soup.find_all('a', class_='css-1ej4hfo'):
        title = item.get_text()
        link = f"https://www.binance.com{item['href']}"
        announcements.append((title, link))
    return announcements

# Coinbase 공지사항 크롤링 함수
def fetch_coinbase_announcements():
    url = "https://blog.coinbase.com/latest"
    response = requests.get(url)
    soup = BeautifulSoup(response.text, "html.parser")

    announcements = []
    for item in soup.find_all('a', class_='ds-link'):
        title = item.get_text()
        link = item['href']
        announcements.append((title, link))
    return announcements

# Upbit 공지사항 크롤링 함수
def fetch_upbit_announcements():
    url = "https://upbit.com/service_center/notice"
    response = requests.get(url)
    soup = BeautifulSoup(response.text, "html.parser")

    announcements = []
    for item in soup.find_all('a', class_='notice-link'):
        title = item.get_text().strip()
        link = f"https://upbit.com{item['href']}"
        announcements.append((title, link))
    return announcements

# 텔레그램으로 공지사항 전송 함수
def send_announcement_to_telegram(bot, chat_id, title, link):
    message = f"📢 {title}\n🔗 {link}"
    bot.send_message(chat_id=chat_id, text=message)

# 프로그램 메인 함수
def main():
    sent_announcements = set()  # 중복 전송 방지용 set

    while True:
        # Binance 공지사항 전송
        for title, link in fetch_binance_announcements():
            if link not in sent_announcements:
                send_announcement_to_telegram(bot, chat_id, title, link)
                sent_announcements.add(link)

        # Coinbase 공지사항 전송
        for title, link in fetch_coinbase_announcements():
            if link not in sent_announcements:
                send_announcement_to_telegram(bot, chat_id, title, link)
                sent_announcements.add(link)

        # Upbit 공지사항 전송
        for title, link in fetch_upbit_announcements():
            if link not in sent_announcements:
                send_announcement_to_telegram(bot, chat_id, title, link)
                sent_announcements.add(link)

        time.sleep(60 * 5)  # 5분마다 실행

# 실행
if __name__ == "__main__":
    main()

설명

  1. fetch_binance_announcements(), fetch_coinbase_announcements(), fetch_upbit_announcements() 함수는 각 거래소의 공지사항을 크롤링하여 제목과 링크를 추출합니다.
  2. send_announcement_to_telegram() 함수는 특정 공지사항을 텔레그램으로 전송합니다.
  3. main() 함수는 주기적으로 각 거래소의 공지사항을 크롤링하고, 새로운 공지사항이 있을 경우에만 텔레그램으로 전송합니다. sent_announcements 집합을 사용하여 중복 전송을 방지합니다.
  4. 이 코드는 매 5분마다 실행되어 최신 공지사항을 확인하고 알림을 전송하도록 설정되어 있습니다.

이 코드로 암호화폐 거래소의 공지사항을 실시간으로 받아볼 수 있습니다.

반응형