Automatyzacja wyszukiwania stref supply demand na Bybit

Zautomatyzowany system wykrywania stref podaży i popytu z powiadomieniami na Telegramie dla Bybit

System w Pythonie skanuje wszystkie pary USDT na Bybit, identyfikuje strefy popytu i podaży według koncepcji Smart Money, śledzi powroty ceny z formowaniem się pinbar i natychmiast wysyła sygnały handlowe na Telegram z oznaczeniami na wykresie. Przetwarzanie setek aktywów zajmuje minuty zamiast godzin ręcznej analizy.

Logika opiera się na łańcuchu: wykrycie strefy → powrót ceny → reakcja w postaci price action → generacja sygnału. Strefy są filtrowane według ATR, przebić i przecięć, aby wykluczyć nieaktualne poziomy.

Teoria stref podaży i popytu

Strefy płynności to kluczowe obszary, gdzie następuje akumulacja pozycji i późniejszy impuls. Strefy popytu formują się tam, skąd cena agresywnie rosła, podaży – skąd gwałtownie spadała. Sygnał pojawia się przy powrocie ceny do strefy z potwierdzeniem pinbar (świeca z długim knotem).

Kryteria ważności stref:

• Impuls przed strefą > 2 ATR
• Impuls po strefie > 2 ATR
• Rozmiar strefy 1–6 ATR
• Brak przebicia po uformowaniu

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

Parametry dostosowują czułość: rama czasowa 4H dla stref, okres ATR 100 do oceny zmienności, skanowanie co 300 sekund.

Integracja z API Bybit

System ładuje listę symboli i historyczne świece:

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()  # od starych do nowych
    return klines

Filtrowane są tylko pary USDT. Dane przetwarzane są w pandas DataFrame z obliczonym ATR.

Wykrywanie order blocks

Order blocks to podstawowe struktury dla stref. Demand OB: niedźwiedzia świeca, po której następuje byczy impuls powyżej high poprzedniej.

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]

        # DEMAND 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
                })

Analogicznie dla podaży. Impuls filtrowany >1.5 ATR, aby wykluczyć ruchy zakresowe.

Rejection blocks i zaawansowane strefy

Rejection blocks wykrywane według proporcji świecy: dolny knot >55% zakresu, korpus <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

        # DEMAND rejection
        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
            })

Zaawansowany filtr sprawdza impulsy przed/po strefie względem ATR.

Funkcja unieważnienia:

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']

Usunięcie przecinających się stref sortuje według świeżości i wyklucza nakładki.

Wybór najbliższych stref i trigger pinbar

Wybrane są 2 najbliższe strefy do aktualnej ceny:

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]

Pinbar na TF 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

Generowanie i wysyłanie sygnałów

Sprawdzenie sygnału long:

def check_long_signal(symbol, klines_5m, zones):
    last = klines_5m[-2]  # zamknięta świeca
    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

Sygnał wizualizowany na wykresie z oznaczeniami stref i wysyłany na Telegram przez API bota.

Co jest ważne

• System używa 4H dla stref i 5m dla triggerów, filtrując szum niższych TF.
• Normalizacja ATR dostosowuje strefy do zmienności aktywa bez powiązania z BTC.
• Ochrona przed duplikatami przez sent_signals i unieważnienie przebitych stref.
• Przetwarzanie wszystkich par USDT Bybit w ciągu minut z naciskiem na najbliższe strefy.
• Price action (pinbar) jako finalny trigger po strukturalnym potwierdzeniu.

— Editorial Team