웹 개발

안녕하세요 오랜만에 이렇게 블로그를 쓰게되네요😀 요즘 제가 꾸준하게 작업하고 있는 바이낸스 알파 포인트 작업을 하면서 불편했던점을 해결해나가는 스토리를 포스팅 해보려고 합니다. 
일단, 시작하기에 앞서 바이낸스 알파 포인트부터 알아보도록하죠

---

바이낸스 Alpha Point 이벤트란?

바이내스에서 진행하는 Alpha Point 이벤트는 특정 토큰(Alpha 토큰)으로 거래하면서 거래량에 비례해 Alpha Point를 부여받는 이벤트입니다. 이 포인트를 모으면 알파마켓에 상장할 토큰을 에어드랍 및 IDO 참여가 가능해집니다. 문제는 거래량을 확인하는 과정이 번거롭다는 것입니다.

 

하루 거래량으로 16점을 억기 위해서는 넉넉잡아 $74,000의 거래량을 달성해야합니다. 이를 $800 단위로 스왑하면 약 93번의 걸래를 해야 하죠. 이 거래 내역을 일일이 확인하려면 거래소에서 스크롤을 내리며 하나씩 세어야 했습니다.

 

10일 정도 이 과정을 반복하다 보니, 너무 비효율적이라 생각이 들었습니다. 그래서 Alpha 토큰의 거래량만 자동으로 추출하는 웹사이트를 만들어 보았습니다.

---

제작

현재 제가 만들려는 웹사이트의 핵심은 바이낸스 체인 기반 토큰의 거래 데이터를 추출하는 것과 현시간의 토큰 가격을 불러와서 거래내역을 분석해 하루단위의 거래량을 측정하는 것입니다 이를 위해 두 가지 주요 API를 활용했습니다.

 

1. BscScan API:

•  BscScan은 BSC 블록체인 데이터를 탐색할 수 있는 도구입니다. 처음에는 API 키 없이 데이터를 가져오려 했지만, BscScan에서 봇 감지를 위해 Cloudflare 캡처를 사용하고 있어 결국 API 키를 발급 받아야 했습니다.
  

  

  
  
  
• 로그인 후 간단한 API키를 생성해 사용했습니다.
  

  

2. CoinGecko API:

• 거래된 토큰의 현재 가격을 알아야 거래량을 USD로 계산할 수 있습니다. CoinGecko는 무료로 제공되는 공용 API를 통해 토큰 가격 데이터를 쉽게 가져올 수 있었습니다. 이 데이터를 BscScan에서 가져온 거래 내역과 결합해 총 거래량을 계산했습니다.

*주의* 코인게코에서 무료로 제공하는 API는 성능면에서 영 별로더라고요 1분마다 50회 조회만 가능해서 사이트를 여러사람들에게 공유하고 사용할때는 CoinGecko API 오류가 자주 뜨더라고요.

---

개발 환경 설정

웹사이트를 만들기 위해 필요한 파이썬 도구들을 다음과 같습니다

• Flask - 웹 애플리케이션 프레임워크로, 사용자 인터페이스와 백엔드 로직을 구현합니다.
• aiohttp - 비동기 HTTP 요청을 처리해 API 호출을 효율적으로 관리합니다.
• asyncio - 비동기 프로그래밍을 지원해 다중 API 호출을 빠르게 처리합니다.
• pytz - 한국 시간(KST) 기준으로 데이터를 처리하기 위해 사용합니다.
• logging - 디버깅과 오류 추적을 위한 로그 설정입니다.
  

  

---

이제 본격적으로 동작 원리를 설명드리겠습니다. 이 웹 사이트는 사용자가 자신의 지갑 주소를 입력하면, 최근 2일간의 Alpha 토큰 거래량을 USD로 계산해 보여줍니다.

 

1. BscSacn API로 걸 데이터 가져오기:

• 사용자의 지갑 주소를 입력 받아 BscScan에서 최근 거래 내역을 가져옵니다. 특정 블록(2일 전부터 현재까지)을 기준으로 필터링해 데이터를 효율적으로 처리합니다.
  

  

2. CoinGecko API로 토큰 가격 조회:

• 화이트리스트에 포함된 토큰(USDT, BNB, ZKJ, MYX, AIOT, B2)의 현재 USD 가격을 가져옵니다.
  

  

3. 거래량 계산 및 필터링:

• BNB ↔ USDT 스왑은 이벤트에서 제외되므로 이를 필터링합니다.
• 거래 데이터를 한국 시간(KST) 기준로 일별로 그룹화합니다. (매일 08:59:59를 기준으로 날짜를 구분)
• 각 거래의 USD 가치를 계산하고, 2배 포인트 토큰은 두 배로 계산합니다. (현재 진행중인 이벤트 적용중)
• 동일한 거래가 중복 계산되지 않도록 해시를 추적합니다.
  

  

4. 웹 인터페이스 제공:

Flask를 사용하여 간단한 웹페이지르 구현했습니다. 사용자는 홈페이지에서 지갑 주소를 입력하고, 결과를 테이블 형태로 확인할 수 있습니다.

import asyncio
import aiohttp
from flask import Flask, request, render_template
from datetime import datetime, timedelta, time
import pytz
import logging

app = Flask(__name__)

# 로그 설정
logging.basicConfig(
    level=logging.INFO,
    format="%(levelname)s:%(name)s:%(message)s"
)
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

BSCSCAN_API_KEY = "사용자 API키 입력란"
COINGECKO_API_URL = "https://api.coingecko.com/api/v3"

WHITELIST_TOKENS = {
    "USDT": "0x55d398326f99059ff775485246999027b3197955",
    "BNB":  "0xbb4cdb9cbd36b01bd1cbaebf2de08d9173bc095c",
    "ZKJ":  "0xC71B5F631354BE6853eFe9C3Ab6b9590F8302e81",
    "MYX":  "0xd82544bf0dfe8385ef8fa34d67e6e4940cc63e16",
    "AIOT": "0x55ad16bd573b3365f43a9daeb0cc66a73821b4a5",
    "B2":   "0x783c3f003f172c6ac5ac700218a357d2d66ee2a2"
}

def is_double_volume_token(addr: str) -> bool:
    """MYX, AIOT, B2, BNB 주소인지 확인"""
    double_tokens = {"MYX", "AIOT", "B2", "BNB"}
    return addr.lower() in {WHITELIST_TOKENS[name].lower() for name in double_tokens}

async def get_block_by_timestamp(session, timestamp):
    url = (
        f"https://api.bscscan.com/api"
        f"?module=block&action=getblocknobytime"
        f"&timestamp={timestamp}&closest=before"
        f"&apikey={BSCSCAN_API_KEY}"
    )
    try:
        async with session.get(url, timeout=aiohttp.ClientTimeout(total=30)) as resp:
            resp.raise_for_status()
            data = await resp.json()
            if data["status"] == "1":
                return data["result"]
            else:
                logger.error(f"Block fetch failed: {data['message']}")
    except aiohttp.ClientError as e:
        logger.error(f"BscScan API error: {e}")
    return None

async def get_token_price(session, contract_address, retries=3):
    platform = "binance-smart-chain"
    url = (
        f"{COINGECKO_API_URL}/simple/token_price/{platform}"
        f"?contract_addresses={contract_address}&vs_currencies=usd"
    )
    for i in range(retries):
        try:
            async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp:
                resp.raise_for_status()
                data = await resp.json()
                key = contract_address.lower()
                if key in data and "usd" in data[key]:
                    return data[key]["usd"]
                logger.warning(f"No price for {contract_address}")
                return 0.0
        except aiohttp.ClientError as e:
            logger.error(f"Price fetch attempt {i+1} failed for {contract_address}: {e}")
            if i < retries - 1:
                await asyncio.sleep(2)
    return 0.0

async def get_whitelist_token_prices(session):
    prices = {}
    for name, addr in WHITELIST_TOKENS.items():
        price = await get_token_price(session, addr)
        prices[addr.lower()] = price
        logger.info(f"Fetched price for {name}: ${price:.2f}")
    return prices

async def get_bnb_transactions(session, wallet_address, start_block):
    """BNB 네이티브 전송 가져오기"""
    txs, seen = [], set()
    page, offset = 1, 1000
    while True:
        url = (
            f"https://api.bscscan.com/api"
            f"?module=account&action=txlist"
            f"&address={wallet_address}"
            f"&startblock={start_block}&endblock=99999999"
            f"&page={page}&offset={offset}"
            f"&sort=desc&apikey={BSCSCAN_API_KEY}"
        )
        try:
            async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp:
                resp.raise_for_status()
                data = await resp.json()
                if data["status"] != "1":
                    logger.error(f"BNB Tx fetch error: {data['message']}")
                    return []
                batch = data["result"]
                new = [tx for tx in batch if tx["hash"] not in seen]
                for tx in new: seen.add(tx["hash"])
                txs.extend(new)
                if len(batch) < offset: break
                page += 1
        except aiohttp.ClientError as e:
            logger.error(f"BNB Tx API error: {e}")
            return []
    # BNB 전송만 필터링 (value > 0, 토큰 거래 제외)
    bnb_txs = []
    for tx in txs:
        if int(tx["value"]) > 0 and not tx.get("contractAddress"):
            bnb_txs.append({
                "hash": tx["hash"],
                "timeStamp": tx["timeStamp"],
                "value": tx["value"],
                "tokenSymbol": "BNB",
                "tokenDecimal": "18",
                "contractAddress": WHITELIST_TOKENS["BNB"].lower()
            })
    return bnb_txs

def get_custom_day_key(kst_dt):
    if kst_dt.time() < time(8, 59, 59):
        key = (kst_dt - timedelta(days=1)).date()
    else:
        key = kst_dt.date()
    return key.strftime("%m.%d")

async def get_daily_volume(wallet_address):
    KST = pytz.timezone("Asia/Seoul")
    now = datetime.now(KST)
    two_days_ago_ts = int((now - timedelta(days=2)).timestamp())

    async with aiohttp.ClientSession() as session:
        # 1) 시작 블록 가져오기
        start_block = await get_block_by_timestamp(session, two_days_ago_ts)
        if not start_block:
            return {}

        # 2) ERC-20 토큰 거래 가져오기
        erc20_txs, seen = [], set()
        page, offset = 1, 1000
        while True:
            url = (
                f"https://api.bscscan.com/api"
                f"?module=account&action=tokentx"
                f"&address={wallet_address}"
                f"&startblock={start_block}&endblock=99999999"
                f"&page={page}&offset={offset}"
                f"&sort=desc&apikey={BSCSCAN_API_KEY}"
            )
            try:
                async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp:
                    resp.raise_for_status()
                    data = await resp.json()
                    if data["status"] != "1":
                        logger.error(f"Tx fetch error: {data['message']}")
                        return {}
                    batch = data["result"]
                    new = [tx for tx in batch if tx["hash"] not in seen]
                    for tx in new: seen.add(tx["hash"])
                    erc20_txs.extend(new)
                    if len(batch) < offset: break
                    page += 1
            except aiohttp.ClientError as e:
                logger.error(f"Tx API error: {e}")
                return {}

        # 3) BNB 거래 가져오기
        bnb_txs = await get_bnb_transactions(session, wallet_address, start_block)

        # 4) 화이트리스트 필터링
        wl = {addr.lower() for addr in WHITELIST_TOKENS.values()}
        ft = [tx for tx in erc20_txs if tx["contractAddress"].lower() in wl]
        ft.extend(bnb_txs)  # BNB 거래 추가

        # 5) 토큰 가격 가져오기
        prices = await get_whitelist_token_prices(session)
        grouped = {}
        for tx in ft:
            grouped.setdefault(tx["hash"], []).append(tx)

        # 6) BNB<->USDT 스왑 제외
        bnb, usdt = WHITELIST_TOKENS["BNB"].lower(), WHITELIST_TOKENS["USDT"].lower()
        swap_hashes = {
            h for h, lst in grouped.items()
            if len(lst) == 2
            and sorted([lst[0]["contractAddress"].lower(), lst[1]["contractAddress"].lower()]) == sorted([bnb, usdt])
        }
        for h in swap_hashes:
            logger.info(f"❌ Excluded BNB <-> USDT swap: {h}")

        # 7) 거래량 처리
        daily_volume_usd = {}
        today_key = get_custom_day_key(now)
        yesterday_key = get_custom_day_key(now - timedelta(days=1))
        seen_hashes_processed = set()
        usdt_addr = WHITELIST_TOKENS["USDT"].lower()

        for tx_hash, tx_list in grouped.items():
            if tx_hash in swap_hashes:
                continue

            try:
                if len(tx_list) == 2:
                    tx1, tx2 = tx_list
                    addr1, addr2 = tx1["contractAddress"].lower(), tx2["contractAddress"].lower()
                    if addr1 == usdt_addr and addr2 == usdt_addr:
                        logger.info(f"❌ Excluded USDT-only transaction: {tx_hash}")
                        continue
                    price1, price2 = prices.get(addr1, 0.0), prices.get(addr2, 0.0)
                    if price1 == 0.0 or price2 == 0.0:
                        continue

                    def get_amount(tx):
                        return float(tx["value"]) / (10 ** int(tx["tokenDecimal"]))

                    amt1, amt2 = get_amount(tx1), get_amount(tx2)
                    usd1, usd2 = amt1 * price1, amt2 * price2
                    kst_time = datetime.fromtimestamp(int(tx1["timeStamp"]), tz=pytz.UTC).astimezone(KST)
                    day_key = get_custom_day_key(kst_time)
                    if day_key not in [today_key, yesterday_key]:
                        continue

                    logger.info(f"✅ Included: {tx1['tokenSymbol']} {amt1:.2f} → {tx2['tokenSymbol']} {amt2:.2f} on {day_key}")

                    for addr, usd in zip([addr1, addr2], [usd1, usd2]):
                        symbol = tx1["tokenSymbol"] if addr == addr1 else tx2["tokenSymbol"]
                        multiplier = 2 if is_double_volume_token(addr) else 1
                        daily_volume_usd.setdefault(day_key, {"total": 0.0, "tokens": {}})
                        daily_volume_usd[day_key]["total"] += usd * multiplier
                        daily_volume_usd[day_key]["tokens"][symbol] = daily_volume_usd[day_key]["tokens"].get(symbol, 0.0) + usd * multiplier

                    seen_hashes_processed.add(tx_hash)
                else:
                    tx = tx_list[0]
                    if tx["hash"] in seen_hashes_processed:
                        continue
                    addr = tx["contractAddress"].lower()
                    if addr == usdt_addr:
                        logger.info(f"❌ Excluded USDT-only transaction: {tx['hash']}")
                        continue
                    price = prices.get(addr, 0.0)
                    if price == 0.0:
                        continue
                    amount = float(tx["value"]) / (10 ** int(tx["tokenDecimal"]))
                    usd = amount * price
                    kst_time = datetime.fromtimestamp(int(tx["timeStamp"]), tz=pytz.UTC).astimezone(KST)
                    day_key = get_custom_day_key(kst_time)
                    if day_key not in [today_key, yesterday_key]:
                        continue

                    logger.info(f"✅ Included: {tx['tokenSymbol']} {amount:.2f} → ${usd:.2f} on {day_key}")
                    multiplier = 2 if is_double_volume_token(addr) else 1
                    daily_volume_usd.setdefault(day_key, {"total": 0.0, "tokens": {}})
                    daily_volume_usd[day_key]["total"] += usd * multiplier
                    daily_volume_usd[day_key]["tokens"][tx["tokenSymbol"]] = daily_volume_usd[day_key]["tokens"].get(tx["tokenSymbol"], 0.0) + usd * multiplier
                    seen_hashes_processed.add(tx["hash"])
            except Exception as e:
                logger.error(f"Error processing transaction {tx_hash}: {e}")
                continue

        # 8) 이중 계산 보정
        for day in daily_volume_usd:
            daily_volume_usd[day]["total"] /= 2
            for sym in daily_volume_usd[day]["tokens"]:
                daily_volume_usd[day]["tokens"][sym] /= 2

        return daily_volume_usd

@app.route("/", methods=["GET", "POST"])
def index():
    if request.method == "POST":
        addr = request.form["wallet_address"]
        data = asyncio.run(get_daily_volume(addr))
        if not data:
            return render_template("result.html", result={}, error="거래 내역이 없거나 가져오기 실패")
        ordered = dict(sorted(data.items(), reverse=True))
        return render_template("result.html", result=ordered)
    return render_template("index.html")

if __name__ == "__main__":
    app.run(debug=True)

 

---

접속하기

1. http://127.0.0.1:5000/에 접속합니다.

 

 

2. 지갑주소 입력: 홈페이지에서 BSC 지갑 주소를 입력합니다.

 

 

3. 최근 2일간의 거래량이 날짜별로 정리된 테이블로 표시됩니다. 각 토큰의 거래량과 총 USD 가치가 포함됩니다.

 

 

4. 포인트 계산: 표시괸 거래량을 바탕으로 Alpha Point를 추청할 수 있습니다.

 

끝으로 이번 포스트에서는 바이낸스 Alpha Point 이벤트의 번거로운 거래량 확인 과정을 해결하기 위해 웹사이트를 제작한 과정을 담아냈습니다. BscScan과 CoinGecko API를 활용해 거래 데이터를 효율적으로 추출하고, Flask 기반의 간단한 웹 인터페이스를 구현하여 이제는 손으로 스크롤을 하지않고 주소만 입력하면 간편하게 거래량을 조회할 수 있게되었습니다. (●'◡'●)

😊다음 포스트에서는 VPS를 이용해 웹서버를 구축하고 이 사이트를 실제로 운영하는 과정을 자세히 다뤄보겠습니다.

 

https://www.flaticon.com/kr/free-icons/

파일 및 폴더 아이콘 제작자: smashingstocks
거래 내역 아이콘 제작자: Nuaba 
주문 내역 아이콘 제작자: HAJICON 
비즈니스 및 금융 아이콘 제작자: Talha Dogar
화이트리스트 아이콘 제작자: Hopstarter
지구본 위치 아이콘 제작자: Ehtisham Abid
진상 아이콘 제작자: Uniconlabs
X2 아이콘 제작자: Freepik