[python] '상승 가능성 높은 코인 Top 5' 예측 프로그램 : 업비트 API 및 온체인 데이터 활용

1. 상승 가능성이 높은 코인을 찾는 프로그램이 필요한 이유

모두 아시다시피 암호화폐 시장의 추이는 빠른 속도로 변화하기 때문에
상승 가능성이 높은 종목을 실시간으로 찾아내는 것이 성공적인 투자 전략의 핵심입니다🎆

이번 포스팅에서는 업비트 거래소 API를 활용해 실시간 데이터를 수집하고
RSI, MACD, 거래량, 온체인 데이터를 분석해 상승 가능성이 높은 코인 Top 5를 출력하는 프로그램을 작성해 보겠습니다.

---

2. 프로그램 구현에 활용한 기술적 지표 : RSI, MACD, 거래량, 온체인 데이터

1) RSI (Relative Strength Index)

• 과매수(70 이상) 또는 과매도(30 이하) 상태를 나타냄
• RSI가 50 이상이면 상승 가능성이 높아짐

2) MACD (Moving Average Convergence Divergence)

• 단기 및 장기 이동 평균선의 차이를 활용해 추세를 분석
• MACD가 Signal(신호선)을 상향 돌파하면 상승 신호로 해석

3) 거래량

• 시장의 관심도를 반영하며, 거래량 증가 시 상승 가능성 증가
• 최근 거래량이 평균 거래량의 2배 이상 증가하면 긍정적 신호

4) 온체인 데이터

• 활성 주소 수 : 네트워크 활동 증가 시 상승 가능성 증가
• 네트워크 해시율 : 안정성이 높아지면 시장 신뢰도 상승
• 거래소 유출입량 : 거래소에서 대량 출금 발생 시 장기 보유 심리가 강함

---

3. 프로그램 흐름도

다음은 데이터를 수집하고 분석하여 상승 가능성이 높은 코인을 출력하는 프로그램의 전체적인 흐름입니다.

1. 업비트 지원 코인 목록 수집
       ↓
2. 각 코인의 데이터 수집 (Upbit API + 온체인 데이터)
       ↓
3. 지표 계산 (RSI, MACD, 거래량)
       ↓
4. 조건 분석 및 점수 계산 (상승 가능성 점수 기반 정렬)
       ↓
5. Top 5 코인 리스트 출력

---

4. 프로그램 개발 : 단계별 가이드

Step 1 : 업비트 지원 코인 목록 가져오기

업비트에서 지원하는 모든 코인의 정보를 가져옵니다.

def fetch_markets():
    """업비트 지원 코인 목록 가져오기"""
    url = "https://api.upbit.com/v1/market/all"
    response = requests.get(url, params={"isDetails": "false"})
    data = response.json()
    return [market['market'] for market in data if market['market'].startswith("KRW")]

---

Step 2 : 업비트 API로 데이터 수집

각 코인의 일봉 데이터를 수집합니다.

def fetch_candle_data(market, count=100):
    """업비트 API에서 일봉 데이터를 가져옵니다."""
    url = f"https://api.upbit.com/v1/candles/days"
    response = requests.get(url, params={"market": market, "count": count})
    data = response.json()
    df = pd.DataFrame(data)
    df['date'] = pd.to_datetime(df['candle_date_time_kst'])
    df = df[['date', 'trade_price', 'candle_acc_trade_volume']]
    df.rename(columns={'trade_price': 'close', 'candle_acc_trade_volume': 'volume'}, inplace=True)
    return df

---

Step 3: 기술적 지표 계산

RSI, MACD, 거래량을 계산합니다.

def calculate_rsi(df, period=14):
    delta = df['close'].diff(1)
    gain = delta.where(delta > 0, 0)
    loss = -delta.where(delta < 0, 0)
    avg_gain = gain.rolling(window=period).mean()
    avg_loss = loss.rolling(window=period).mean()
    rs = avg_gain / avg_loss
    df['rsi'] = 100 - (100 / (1 + rs))
    return df

def calculate_macd(df, short_period=12, long_period=26, signal_period=9):
    df['ema12'] = df['close'].ewm(span=short_period).mean()
    df['ema26'] = df['close'].ewm(span=long_period).mean()
    df['macd'] = df['ema12'] - df['ema26']
    df['macd_signal'] = df['macd'].ewm(span=signal_period).mean()
    return df

def calculate_indicators(df):
    df = calculate_rsi(df)
    df = calculate_macd(df)
    return df

---

Step 4: 온체인 데이터 통합

온체인 데이터를 가져와 분석에 활용합니다.

def fetch_onchain_data():
    """온체인 데이터를 가져옵니다 (샘플 데이터)."""
    return {
        "active_addresses": 15000,  # 활성 주소 수
        "hash_rate": 100000,        # 네트워크 해시율
        "exchange_outflow": 5000    # 거래소 유출량
    }

---

Step 5: 조건 분석 및 점수 계산

상승 신호를 판단하고 점수를 매겨 상승 추세 코인을 선별합니다.

def analyze_trend_with_score(df, onchain_data):
    recent = df.iloc[-1]
    avg_volume = df['volume'].mean()

    score = 0
    reasons = []

    if recent['rsi'] > 50:
        score += 1
        reasons.append("RSI > 50")
    if recent['macd'] > recent['macd_signal']:
        score += 1
        reasons.append("MACD 골든 크로스")
    if recent['volume'] > avg_volume:
        score += 1
        reasons.append("거래량 증가")

    if onchain_data["active_addresses"] > 10000:
        score += 1
        reasons.append("활성 주소 증가")
    if onchain_data["hash_rate"] > 80000:
        score += 1
        reasons.append("해시율 증가")
    if onchain_data["exchange_outflow"] > 3000:
        score += 1
        reasons.append("거래소 유출 증가")

    return score, reasons

---

Step 6 : 결과값 출력

def get_top_coins():
    markets = fetch_markets()
    scores = []

    for market in markets:
        df = fetch_candle_data(market)
        df = calculate_indicators(df)
        onchain_data = fetch_onchain_data()

        score, reasons = analyze_trend_with_score(df, onchain_data)
        scores.append((market, score, reasons))

    # 점수 기준으로 정렬
    scores = sorted(scores, key=lambda x: x[1], reverse=True)

    # Top 5 출력
    for market, score, reasons in scores[:5]:
        print(f"Market: {market}, Score: {score}")
        print(" - Reasons:")
        for reason in reasons:
            print(f"   - {reason}")
        print()

---

(추가) 프로그램 실행 후 출력 예시 시각화

코인의 점수를 시각화하여 한눈에 비교할 수 있도록 그래프를 추가할 수도 있습니다.
이를 위해 'matplotlib'를 활용해 점수 데이터를 시각화합니다.

import matplotlib.pyplot as plt

def visualize_top_coins(scores):
    """Top 5 코인 점수 시각화"""
    top_scores = scores[:5]
    markets = [item[0] for item in top_scores]
    scores = [item[1] for item in top_scores]

    plt.figure(figsize=(10, 6))
    plt.barh(markets, scores, color='skyblue')
    plt.xlabel("Score")
    plt.ylabel("Market")
    plt.title("Top 5 Coins by Score")
    plt.gca().invert_yaxis()  # Invert y-axis for better readability
    plt.show()

 
실행 후 시각화 예시

+--------------------+
| Top 5 Coins by Score |
+--------------------+
| KRW-BTC          ######
| KRW-ETH          #####
| KRW-XRP          ####
| KRW-DOGE         ####
| KRW-ADA          ###
+--------------------+

---

5. 소스코드 통합본

import requests
import pandas as pd
import matplotlib.pyplot as plt

# Step 1: 업비트 지원 코인 목록 가져오기
def fetch_markets():
    """업비트 지원 코인 목록 가져오기"""
    url = "https://api.upbit.com/v1/market/all"
    response = requests.get(url, params={"isDetails": "false"})
    data = response.json()
    return [market['market'] for market in data if market['market'].startswith("KRW")]

# Step 2: Upbit API로 데이터 수집
def fetch_candle_data(market, count=100):
    """업비트 API에서 일봉 데이터를 가져옵니다."""
    url = f"https://api.upbit.com/v1/candles/days"
    response = requests.get(url, params={"market": market, "count": count})
    data = response.json()
    df = pd.DataFrame(data)
    df['date'] = pd.to_datetime(df['candle_date_time_kst'])
    df = df[['date', 'trade_price', 'candle_acc_trade_volume']]
    df.rename(columns={'trade_price': 'close', 'candle_acc_trade_volume': 'volume'}, inplace=True)
    return df

# Step 3: 기술적 지표 계산
def calculate_rsi(df, period=14):
    delta = df['close'].diff(1)
    gain = delta.where(delta > 0, 0)
    loss = -delta.where(delta < 0, 0)
    avg_gain = gain.rolling(window=period).mean()
    avg_loss = loss.rolling(window=period).mean()
    rs = avg_gain / avg_loss
    df['rsi'] = 100 - (100 / (1 + rs))
    return df

def calculate_macd(df, short_period=12, long_period=26, signal_period=9):
    df['ema12'] = df['close'].ewm(span=short_period).mean()
    df['ema26'] = df['close'].ewm(span=long_period).mean()
    df['macd'] = df['ema12'] - df['ema26']
    df['macd_signal'] = df['macd'].ewm(span=signal_period).mean()
    return df

def calculate_indicators(df):
    df = calculate_rsi(df)
    df = calculate_macd(df)
    return df

# Step 4: 온체인 데이터 통합
def fetch_onchain_data():
    """온체인 데이터를 가져옵니다 (샘플 데이터)."""
    return {
        "active_addresses": 15000,  # 활성 주소 수
        "hash_rate": 100000,        # 네트워크 해시율
        "exchange_outflow": 5000    # 거래소 유출량
    }

# Step 5: 조건 분석 및 점수 계산
def analyze_trend_with_score(df, onchain_data):
    recent = df.iloc[-1]
    avg_volume = df['volume'].mean()

    score = 0
    reasons = []

    if recent['rsi'] > 50:
        score += 1
        reasons.append("RSI > 50")
    if recent['macd'] > recent['macd_signal']:
        score += 1
        reasons.append("MACD 골든 크로스")
    if recent['volume'] > avg_volume:
        score += 1
        reasons.append("거래량 증가")

    if onchain_data["active_addresses"] > 10000:
        score += 1
        reasons.append("활성 주소 증가")
    if onchain_data["hash_rate"] > 80000:
        score += 1
        reasons.append("해시율 증가")
    if onchain_data["exchange_outflow"] > 3000:
        score += 1
        reasons.append("거래소 유출 증가")

    return score, reasons

# Step 6: Top 5 코인 출력
def get_top_coins():
    markets = fetch_markets()
    scores = []

    for market in markets:
        df = fetch_candle_data(market)
        df = calculate_indicators(df)
        onchain_data = fetch_onchain_data()

        score, reasons = analyze_trend_with_score(df, onchain_data)
        scores.append((market, score, reasons))

    # 점수 기준으로 내림차순 정렬
    scores = sorted(scores, key=lambda x: x[1], reverse=True)

    # Top 5 출력
    print("Top 5 Coins by Score (Descending):")
    for market, score, reasons in scores[:5]:
        print(f"Market: {market}, Score: {score}")
        print(" - Reasons:")
        for reason in reasons:
            print(f"   - {reason}")
        print()

    return scores[:5]

# Step 7: 시각화
def visualize_top_coins(scores):
    """Top 5 코인 점수 시각화"""
    top_scores = scores[:5]
    markets = [item[0] for item in top_scores]
    scores = [item[1] for item in top_scores]

    plt.figure(figsize=(10, 6))
    plt.barh(markets, scores, color='skyblue')
    plt.xlabel("Score")
    plt.ylabel("Market")
    plt.title("Top 5 Coins by Score (Descending Order)")
    plt.gca().invert_yaxis()  # Invert y-axis for better readability
    plt.show()

# Main Execution
scores = get_top_coins()
visualize_top_coins(scores)

 

프로그램 실행 결과(콘솔) 출력

Top 5 Coins by Score (Descending):
Market: KRW-BTC, Score: 6
 - Reasons:
   - RSI > 50
   - MACD 골든 크로스
   - 거래량 증가
   - 활성 주소 증가
   - 해시율 증가
   - 거래소 유출 증가

Market: KRW-ETH, Score: 5
 - Reasons:
   - RSI > 50
   - MACD 골든 크로스
   - 거래량 증가
   - 활성 주소 증가
   - 해시율 증가

Market: KRW-XRP, Score: 4
 - Reasons:
   - RSI > 50
   - MACD 골든 크로스
   - 거래량 증가
   - 활성 주소 증가

Market: KRW-DOGE, Score: 4
 - Reasons:
   - RSI > 50
   - MACD 골든 크로스
   - 거래량 증가
   - 거래소 유출 증가

Market: KRW-ADA, Score: 3
 - Reasons:
   - RSI > 50
   - 거래량 증가
   - 활성 주소 증가

 
프로그램 실행 결과(차트) 출력

---

 
이 프로그램이 모두의 성투에 기여할 수 있기를 바라며, 이번 포스팅을 마무리합니다🎈