홈으로 돌아가기

Bybit 공급 수요 존 검색 자동화

시스템은 ATR 필터가 적용된 order 및 rejection blocks를 통해 Bybit의 supply/demand zones 검색을 자동화합니다. 5m-TF pinbar로 진입 확인 및 차트 Telegram 전송. 중/시니어 개발자를 위한 전체 코드 분석.

Python 봇: demand/supply zones Bybit + Telegram
Advertisement 728x90

바이빗용 텔레그램 알림 자동 공급/수요 구역 스캐너

파이썬 기반 시스템이 바이빗의 모든 USDT 페어를 스캔하여 스마트 머니 개념을 활용해 공급 및 수요 구역을 식별하고, 핀바 형성과 함께 가격 재테스트를 추적하며, 차트 마크업과 함께 거래 신호를 텔레그램으로 즉시 전송합니다. 수백 개의 자산을 수동 분석으로 몇 시간 걸리던 것을 몇 분 만에 처리합니다.

로직은 다음과 같은 체인을 따릅니다: 구역 감지 → 가격 재테스트 → 가격 행동 반응 → 신호 생성. 구역은 ATR, 브레이크아웃, 중복을 기준으로 필터링하여 관련 없는 수준을 제외합니다.

공급 및 수요 구역 이론

유동성 구역은 포지션이 축적되고 이후 모멘텀이 발생하는 핵심 영역입니다. 수요 구역은 가격이 공격적으로 상승한 곳에서 형성되고, 공급 구역은 가격이 급격히 하락한 곳에서 형성됩니다. 신호는 가격이 핀바 확인(긴 심지를 가진 캔들)과 함께 구역으로 돌아올 때 트리거됩니다.

Google AdInline article slot

구역 유효성 기준:

  • 구역으로의 임펄스 > 2 ATR
  • 구역 밖으로의 임펄스 > 2 ATR
  • 구역 크기 1–6 ATR
  • 형성 후 브레이크아웃 없음
MAX_ZONE_ATR = 7
MIN_ZONE_ATR = 1
LOOKBACK_PRE = 8
LOOKAHEAD_POST = 8
MIN_PRE_MOVE_ATR = 2
MIN_POST_MOVE_ATR = 2
MIN_ZONE_SIZE_ATR = 1.0
MAX_ZONE_SIZE_ATR = 6.0

매개변수는 민감도를 조정합니다: 구역은 4H 타임프레임, 변동성 평가는 ATR 기간 100, 300초마다 스캔합니다.

바이빗 API 통합

시스템은 심볼 목록과 과거 캔들을 로드합니다:

Google AdInline article slot
def get_symbols():
    data = client.get_instruments_info(category="linear")
    symbols = []
    for item in data["result"]["list"]:
        if item["quoteCoin"] == "USDT":
            symbols.append(item["symbol"])
    return symbols

def get_klines(symbol, interval, limit):
    data = client.get_kline(
        category="linear",
        symbol=symbol,
        interval=interval,
        limit=limit
    )
    klines = data["result"]["list"]
    klines.reverse()  # 오래된 것부터 최신 순으로
    return klines

USDT 페어만 필터링됩니다. 데이터는 계산된 ATR과 함께 pandas DataFrame에서 처리됩니다.

오더 블록 감지

오더 블록은 구역의 기초 구조입니다. 수요 OB: 약세 캔들 뒤에 이전 고점 위로 강세 임펄스가 이어집니다.

def detect_order_blocks(df):
    for i in range(1, len(df)):
        o1 = df.open.iloc[i-1]
        c1 = df.close.iloc[i-1]
        h1 = df.high.iloc[i-1]
        l1 = df.low.iloc[i-1]
        o2 = df.open.iloc[i]
        c2 = df.close.iloc[i]
        atr = df.atr.iloc[i]

        # 수요 OB
        if c1 < o1 and c2 > o2 and c2 > h1:
            impulse = c2 - l1
            if impulse > atr * 1.5:
                zones.append({
                    "type": "demand",
                    "kind": "orderblock",
                    "low": l1,
                    "high": o1,
                    "start_bar": i-1
                })

공급도 유사하게 처리됩니다. 임펄스는 범위 내 움직임을 제외하기 위해 >1.5 ATR로 필터링됩니다.

Google AdInline article slot

리젝션 블록 및 고급 구역

리젝션 블록은 캔들 비율로 감지됩니다: 하단 심지 >55% 범위, 바디 <35%.

def detect_rejection_blocks(df):
    for i in range(2, len(df)-1):
        o = df.open.iloc[i]
        c = df.close.iloc[i]
        h = df.high.iloc[i]
        l = df.low.iloc[i]
        atr = df.atr.iloc[i]
        body = abs(c - o)
        full = h - l
        upper_wick = h - max(o,c)
        lower_wick = min(o,c) - l
        body_ratio = body / full
        lower_ratio = lower_wick / full

        # 수요 리젝션
        if lower_ratio > 0.55 and body_ratio < 0.35:
            zones.append({
                "type": "demand",
                "kind": "rejection",
                "low": l,
                "high": min(o,c),
                "start_bar": i
            })

고급 필터는 ATR에 대한 구역 전/후 임펄스를 확인합니다.

무효화 함수:

def is_zone_broken(df, zone):
    start_idx = zone['start_bar'] + 1
    slice_df = df.iloc[start_idx:]
    if zone['type'] == 'supply':
        break_level = slice_df['high'].max() if INVALIDATION_METHOD == "wick" else slice_df['close'].max()
        return break_level > zone['high']
    else:
        break_level = slice_df['low'].min() if INVALIDATION_METHOD == "wick" else slice_df['close'].min()
        return break_level < zone['low']

중복 구역은 신선도로 정렬하고 중복을 제외하여 제거됩니다.

가장 가까운 구역 선택 및 핀바 트리거

현재 가격에 가장 가까운 두 구역이 선택됩니다:

def get_nearest_zones(price, zones, n=2):
    def zone_distance(z):
        if price < z["low"]:
            return z["low"] - price
        elif price > z["high"]:
            return price - z["high"]
        return 0
    return sorted(zones, key=zone_distance)[:n]

5m 타임프레임의 핀바:

def is_bullish_pinbar(candle):
    o, h, l, c = float(candle[1]), float(candle[2]), float(candle[3]), float(candle[4])
    body = abs(c - o)
    total_range = h - l
    if body > 0.3 * total_range or c <= o or lower_wick < body * 2 or upper_wick > body * 0.5:
        return False
    return True

신호 생성 및 전송

롱 신호 확인:

def check_long_signal(symbol, klines_5m, zones):
    last = klines_5m[-2]  # 닫힌 캔들
    price = float(last[4])
    if not is_bullish_pinbar(last):
        return None
    for zone in zones:
        if price_in_zone(price, zone):
            key = f"{symbol}_{last[0]}_long"
            if key in sent_signals:
                return None
            sent_signals[key] = True
            return zone
    return None

신호는 구역 마크업이 있는 차트에 시각화되고 봇 API를 통해 텔레그램으로 전송됩니다.

핵심 포인트

  • 시스템은 구역은 4H, 트리거는 5m을 사용하여 낮은 타임프레임의 노이즈를 필터링합니다.
  • ATR 정규화는 BTC에 묶이지 않고 자산 변동성에 맞춰 구역을 조정합니다.
  • sent_signals를 통한 중복 방지 및 무효화된 구역 제거.
  • 모든 바이빗 USDT 페어를 몇 분 만에 처리하며 가장 가까운 구역에 집중합니다.
  • 구조적 확인 후 최종 트리거로 가격 행동(핀바) 사용.

— Editorial Team

Advertisement 728x90

다음 읽기