📌  Поиск аномальной ликвидности (детектор "стенок")

Этот скрипт анализирует глубину стакана и ищет необычно крупные лимитные ордера ("стены"), которые могут повлиять на движение цены,используя публичное API Bybit👨‍💻

import requests
import time

# Константы
ORDERBOOK_URL = "https://api.bybit.com/v5/market/orderbook"
SYMBOL = "BTCUSDT"
CATEGORY = "spot"
DEPTH_LEVELS = 20  # Количество уровней стакана для анализа
VOLUME_THRESHOLD_RATIO = 0.1  # Порог "стен" (10% от общего объёма стакана)
REFRESH_INTERVAL = 5  # Интервал обновления (в секундах)

def get_orderbook():
    """Получает стакан заявок"""
    params = {"category": CATEGORY, "symbol": SYMBOL}
    response = requests.get(ORDERBOOK_URL, params=params)
    if response.status_code == 200:
        result = response.json().get("result", {})
        return result.get("b", []), result.get("a", [])
    else:
        raise Exception(f"Ошибка API Orderbook: {response.status_code}, {response.text}")

def detect_liquidity_walls():
    """Анализ ликвидности и поиск крупных заявок"""
    try:
        print(f"Мониторинг ликвидности {SYMBOL}...\n")
        while True:
            bids, asks = get_orderbook()

            # Берём только топ-N уровней стакана
            bids = [(float(price), float(size)) for price, size in bids[:DEPTH_LEVELS]]
            asks = [(float(price), float(size)) for price, size in asks[:DEPTH_LEVELS]]

            # Считаем общий объём заявок в стакане
            total_bid_volume = sum(size for _, size in bids)
            total_ask_volume = sum(size for _, size in asks)

            # Порог для "стен" ликвидности (10% от всего объёма стакана)
            bid_threshold = total_bid_volume * VOLUME_THRESHOLD_RATIO
            ask_threshold = total_ask_volume * VOLUME_THRESHOLD_RATIO

            # Ищем "стены" – заявки, превышающие порог
            big_bids = [(price, size) for price, size in bids if size >= bid_threshold]
            big_asks = [(price, size) for price, size in asks if size >= ask_threshold]

            # Выводим результаты
            if big_bids:
                print(f"🟢 Крупные заявки на покупку:")
                for price, size in big_bids:
                    print(f"   Цена: {price:.2f}, Объём: {size:.2f}")

            if big_asks:
                print(f"🔴 Крупные заявки на продажу:")
                for price, size in big_asks:
                    print(f"   Цена: {price:.2f}, Объём: {size:.2f}")

            print("-" * 40)
            time.sleep(REFRESH_INTERVAL)

    except Exception as e:
        print(f"Ошибка: {e}")

if __name__ == "__main__":
    detect_liquidity_walls()

📌 Подпишись  Python_Scripts❗️

🚀 Полезные библиотеки Python

💾 Uplink

📌 Библиотека Python, для создания HTTP-клиентов с декларативным API (аналог Retrofit в Java). Упрощает работу с REST API.

Особенности:

- Удобное объявление API через аннотации.

- Поддержка асинхронных запросов.

- Автоматическое управление сессиями и заголовками.

⚙️ Установка 👇👇👇

pip install uplink

📁 Документация

#библиотеки

📌 Подпишись Python_Scripts ❗️

📌 Скрипт telegram stars

💻  ЯП:  Python 3.11+🐍
💾  Модули: aiogram, aiohttp
📂  База данных: -

📌 Скрипт шаблон Telegram бота для приёма платежей/донатов в звёздах.

📌 Вся информация приведена исключительно в ознакомительных целях❗️❗️❗️

#telegram

📌 Подпишись  Python_Scripts❗️

📌 Индикатор накопления/распределения (A/D Line) для анализа тренда
Этот индикатор помогает определить, идёт ли цена вверх на реальном объёме, или движение искусственное и может скоро развернуться,используя публичное API Bybit👨‍💻

import requests
import time

# Константы
KLINE_URL = "https://api.bybit.com/v5/market/kline"
SYMBOL = "BTCUSDT"
CATEGORY = "spot"
INTERVAL = "15"  # 15-минутные свечи
LIMIT = 50  # Количество свечей для анализа

def get_kline_data():
    """Получает исторические свечи (open, high, low, close, volume)"""
    params = {"category": CATEGORY, "symbol": SYMBOL, "interval": INTERVAL, "limit": LIMIT}
    response = requests.get(KLINE_URL, params=params)
    
    if response.status_code == 200:
        result = response.json().get("result", {}).get("list", [])
        
        # Проверяем структуру ответа и корректно парсим данные
        candles = []
        for c in result:
            try:
                open_price = float(c[1])
                high = float(c[2])
                low = float(c[3])
                close = float(c[4])
                volume = float(c[5])
                candles.append((open_price, high, low, close, volume))
            except (ValueError, IndexError):
                print(f"Ошибка обработки свечи: {c}")
        
        return candles
    else:
        raise Exception(f"Ошибка API Kline: {response.status_code}, {response.text}")

def calculate_ad_line(candles):
    """Вычисляет линию накопления/распределения (A/D)"""
    ad_line = 0
    ad_values = []
    
    for open_price, high, low, close, volume in candles:
        if high != low:
            money_flow_multiplier = ((close - low) - (high - close)) / (high - low)
        else:
            money_flow_multiplier = 0  # Защита от деления на ноль
            
        money_flow_volume = money_flow_multiplier * volume
        ad_line += money_flow_volume
        ad_values.append(ad_line)

    return ad_values

def detect_trend():
    """Анализ тренда на основе A/D Line"""
    try:
        print(f"Начинаем анализ {SYMBOL}...\n")
        while True:
            candles = get_kline_data()
            if not candles:
                print("Не удалось получить свечи, пробуем снова...")
                time.sleep(30)
                continue

            ad_values = calculate_ad_line(candles)
            current_price = candles[-1][3]  # Цена закрытия последней свечи
            
            ad_trend = "⬆️ Восходящий" if ad_values[-1] > ad_values[-2] else "⬇️ Нисходящий"
            
            print(f"Цена: {current_price:.2f}, A/D Line: {ad_values[-1]:.2f}, Тренд: {ad_trend}")
            print("-" * 40)
            
            time.sleep(30)  # Обновление каждые 30 секунд
    except Exception as e:
        print(f"Ошибка: {e}")

if __name__ == "__main__":
    detect_trend()

📌 Подпишись  Python_Scripts❗️

🚀 Полезные библиотеки Python

💾 Perfplot

📌 Библиотека Python, для удобного сравнения производительности различных реализаций кода.

Особенности:

- Визуализация результатов в виде графиков.

- Простая интеграция с NumPy.

- Хорошая альтернатива timeit с графическим выводом.

⚙️ Установка 👇👇👇

pip install perfplot

📁 Документация

#библиотеки

📌 Подпишись Python_Scripts ❗️

📌 Скрипт парсер DexScreener

💻  ЯП:  Python 3.11+🐍
💾  Модули: bs4, requests, lxml, seleniumbase
📂  База данных: -

📌 Скрипт для парсинга данных с сайта DexScreener.com, по умолчанию собирает данные с первых двадцати страниц выдачи

📌 Вся информация приведена исключительно в ознакомительных целях❗️❗️❗️

#other

📌 Подпишись  Python_Scripts❗️

📌Определяем консолидацию цены — моменты, когда рынок сжимается перед возможным резким движением,используя публичное API Bybit👨‍💻

import requests

# Настройки
SYMBOL = "BTCUSDT"
KLINE_URL = "https://api.bybit.com/v5/market/kline"
INTERVAL = "15"  # 15-минутные свечи
LIMIT = 10  # Количество свечей для анализа
THRESHOLD = 0.2  # Порог консолидации (% от средней цены)

def get_kline_data():
    """Получает исторические свечи"""
    params = {"category": "spot", "symbol": SYMBOL, "interval": INTERVAL, "limit": LIMIT}
    response = requests.get(KLINE_URL, params=params)

    if response.status_code == 200:
        return response.json().get("result", {}).get("list", [])
    else:
        print(f"Ошибка API: {response.status_code}")
        return None

def analyze_consolidation():
    """Определяет узкие диапазоны консолидации"""
    candles = get_kline_data()
    if not candles:
        return
    
    prices = [float(c[4]) for c in candles]  # Цены закрытия
    high = max(prices)
    low = min(prices)
    avg_price = sum(prices) / len(prices)
    
    range_pct = ((high - low) / avg_price) * 100

    if range_pct < THRESHOLD:
        print(f"🔎 Консолидация! Диапазон = {range_pct:.2f}% (Цена: {avg_price:.2f})")
    else:
        print(f"⚡ Нет консолидации. Диапазон = {range_pct:.2f}%")

if __name__ == "__main__":
    analyze_consolidation()

📌 Подпишись  Python_Scripts❗️

🚀 Полезные библиотеки Python

💾 Pydantic-extra-types

📌 Библиотека Python, расширение для Pydantic, добавляющее поддержку сложных типов данных, таких как IP-адреса, UUID, платежные карты, Bitcoin-адреса и др.

Особенности:

- Валидация номеров банковских карт и криптокошельков.

- Поддержка MAC-адресов, доменных имен и пр.

- Интеграция с Pydantic 2.0.

⚙️ Установка 👇👇👇

pip install pydantic-extra-types

📁 Документация

#библиотеки

📌 Подпишись Python_Scripts ❗️

📌 Скрипт предоставляет информацию об устройстве по его imei

💻  ЯП:  Python 3.11+🐍
💾  Модули: python-telegram-bot, requests, sniffio
📂  База данных: sqlalchemy

📌 Скрипт Telegram-bot который по запросу пользователя выводит информацию об устройстве по его imei в формате json.

📌 Вся информация приведена исключительно в ознакомительных целях❗️❗️❗️

#telegram

📌 Подпишись  Python_Scripts❗️

📌Анализ импульсных движений — отслеживаеи сильные свечи, которые могут сигнализировать о начале тренда,используя публичное API Bybit👨‍💻

import requests

# Настройки
SYMBOL = "BTCUSDT"
KLINE_URL = "https://api.bybit.com/v5/market/kline"
INTERVAL = "15"  # 15-минутные свечи
LIMIT = 20  # Количество свечей для анализа
IMPULSE_THRESHOLD = 1.5  # Порог (x раз больше среднего тела свечи)

def get_kline_data():
    """Получает исторические свечи"""
    params = {"category": "spot", "symbol": SYMBOL, "interval": INTERVAL, "limit": LIMIT}
    response = requests.get(KLINE_URL, params=params)

    if response.status_code == 200:
        return response.json().get("result", {}).get("list", [])
    else:
        print(f"Ошибка API: {response.status_code}")
        return None

def analyze_impulse():
    """Определяет импульсные свечи"""
    candles = get_kline_data()
    if not candles:
        return

    # Вычисляем средний размер тела свечи
    body_sizes = [abs(float(c[4]) - float(c[1])) for c in candles]  # |Close - Open|
    avg_body = sum(body_sizes) / len(body_sizes)

    # Проверяем последнюю свечу
    last_open = float(candles[-1][1])
    last_close = float(candles[-1][4])
    last_body = abs(last_close - last_open)

    if last_body > avg_body * IMPULSE_THRESHOLD:
        direction = "⬆️ Вверх" if last_close > last_open else "⬇️ Вниз"
        print(f"🚀 Импульсное движение! {direction} (Тело свечи: {last_body:.2f}, Среднее: {avg_body:.2f})")
    else:
        print("📉 Нет сильного импульса.")

if __name__ == "__main__":
    analyze_impulse()

📌 Подпишись  Python_Scripts❗️

🚀 Полезные библиотеки Python

💾 Hatch

📌 Библиотека Python, cовременный инструмент для управления пакетами, созданный как альтернатива Poetry и Pipenv.

Особенности:

- Упрощённое управление зависимостями и виртуальными окружениями.

- Встроенная поддержка PEP 621 (нового формата pyproject.toml).

- Автоматизация релизов и CI/CD.

⚙️ Установка 👇👇👇

pip install hatch

📁 Документация

#библиотеки

📌 Подпишись Python_Scripts ❗️

📌 Скрипт оплата кейсов в PUBG

💻  ЯП:  Python 3.11+🐍
💾  Модули: fastapi, aiohttp
📂  База данных: postgresql

📌 Скрипт Telegram бот с мини приложением для оплаты кейсов в PUBG.

📌 Вся информация приведена исключительно в ознакомительных целях❗️❗️❗️

#telegram

📌 Подпишись  Python_Scripts❗️

📌Автоматическое определение разворотов рынка с помощью индикатора RSI (Relative Strength Index),используя публичное API Bybit👨‍💻

import requests
import numpy as np

# Настройки
SYMBOL = "BTCUSDT"
KLINE_URL = "https://api.bybit.com/v5/market/kline"
INTERVAL = "15"  # 15-минутные свечи
LIMIT = 100  # Количество свечей для расчета
RSI_PERIOD = 14  # Длина RSI
OVERBOUGHT = 70  # Уровень перекупленности
OVERSOLD = 30  # Уровень перепроданности

def get_kline_data():
    """Получает исторические свечи"""
    params = {"category": "spot", "symbol": SYMBOL, "interval": INTERVAL, "limit": LIMIT}
    response = requests.get(KLINE_URL, params=params)

    if response.status_code == 200:
        return response.json().get("result", {}).get("list", [])
    else:
        print(f"Ошибка API: {response.status_code}")
        return None

def calculate_rsi(prices, period=14):
    """Рассчитывает RSI"""
    deltas = np.diff(prices)
    gains = np.maximum(deltas, 0)
    losses = np.abs(np.minimum(deltas, 0))

    avg_gain = np.convolve(gains, np.ones(period) / period, mode='valid')
    avg_loss = np.convolve(losses, np.ones(period) / period, mode='valid')

    rs = avg_gain / (avg_loss + 1e-10)
    rsi = 100 - (100 / (1 + rs))
    return rsi

def analyze_rsi():
    """Анализирует рынок по RSI"""
    candles = get_kline_data()
    if not candles:
        return

    close_prices = np.array([float(c[4]) for c in candles])
    rsi = calculate_rsi(close_prices, RSI_PERIOD)

    if rsi[-1] > OVERBOUGHT:
        print(f"🔥 RSI = {rsi[-1]:.2f} → Возможен разворот вниз (перекупленность)")
    elif rsi[-1] < OVERSOLD:
        print(f"🟢 RSI = {rsi[-1]:.2f} → Возможен разворот вверх (перепроданность)")
    else:
        print(f"📊 RSI = {rsi[-1]:.2f} → Нейтральная зона")

if __name__ == "__main__":
    analyze_rsi()

📌 Подпишись  Python_Scripts❗️

🚀 Полезные библиотеки Python

💾 Snoop

📌 Библиотека Python,для отладки кода. Автоматически логирует выполнение строк кода, их аргументы и значения переменных.
Вывод в консоли будет включать информацию о вызовах функций и изменениях переменных.

Особенности:

- Подсвечивает код при выполнении.

- Показывает изменения переменных в реальном времени.

- Легко интегрируется в большие проекты.

⚙️ Установка 👇👇👇

pip install snoop

📁 Документация

#библиотеки

📌 Подпишись Python_Scripts ❗️

📌 Скрипт парсер Steam

💻  ЯП:  Python 3.11+🐍
💾  Модули: bs4, lxml, requests, pyTelegramBotAPI
📂  База данных: -

📌 Скрипт Telegram бот находит самые дешевые ключи для Steam региона Россия, показывает раздачи бесплатных игр, выводит информацию о самых ожидаемых играх.

📌 Вся информация приведена исключительно в ознакомительных целях❗️❗️❗️

#telegram

📌 Подпишись  Python_Scripts❗️

📌Автоматическое обнаружение пробоя уровней поддержки и сопротивления!

Этот скрипт:
✅ Получает исторические свечи BTCUSDT
✅ Определяет ключевые уровни поддержки и сопротивления
✅ Выдает сигнал, если цена пробивает уровень

Используем публичное API Bybit👨‍💻

import requests
import numpy as np

# Настройки
SYMBOL = "BTCUSDT"
KLINE_URL = "https://api.bybit.com/v5/market/kline"
INTERVAL = "15"  # 15-минутные свечи
LIMIT = 50  # Количество свечей для анализа
TOLERANCE = 0.002  # Допустимый разрыв (0.2%)

def get_kline_data():
    """Получает исторические свечи"""
    params = {"category": "spot", "symbol": SYMBOL, "interval": INTERVAL, "limit": LIMIT}
    response = requests.get(KLINE_URL, params=params)

    if response.status_code == 200:
        return response.json().get("result", {}).get("list", [])
    else:
        print(f"Ошибка API: {response.status_code}")
        return None

def find_support_resistance(candles):
    """Определяет уровни поддержки и сопротивления"""
    closes = np.array([float(c[4]) for c in candles])
    highs = np.array([float(c[2]) for c in candles])
    lows = np.array([float(c[3]) for c in candles])

    support = np.min(lows)
    resistance = np.max(highs)

    return support, resistance

def detect_breakout():
    """Проверяет пробой уровней"""
    candles = get_kline_data()
    if not candles:
        return

    support, resistance = find_support_resistance(candles)
    last_close = float(candles[-1][4])

    if last_close >= resistance * (1 + TOLERANCE):
        print(f"🚀 Пробой вверх! Цена {last_close:.2f} выше сопротивления {resistance:.2f}")
    elif last_close <= support * (1 - TOLERANCE):
        print(f"⚠️ Пробой вниз! Цена {last_close:.2f} ниже поддержки {support:.2f}")
    else:
        print(f"📊 Цена {last_close:.2f}, диапазон: {support:.2f} - {resistance:.2f} (без пробоя)")

if __name__ == "__main__":
    detect_breakout()

📌 Подпишись  Python_Scripts❗️

🚀 Полезные библиотеки Python

💾 Pint

📌 Библиотека Python, для работы с физическими величинами и единицами измерения. Полезна для научных и инженерных вычислений.

Особенности:

- Автоматическое преобразование единиц измерения.

- Поддержка сложных вычислений с единицами.

- Интеграция с NumPy и Pandas.

⚙️ Установка 👇👇👇

pip install pint

📁 Документация

#библиотеки

📌 Подпишись Python_Scripts ❗️

📌 Скрипт авто продажа в Telegram

💻  ЯП:  Python 3.11+🐍
💾  Модули: aiogram
📂  База данных: aiosqlite

📌 Скрипт простой Telegram-бот, который позволяет пользователям покупать любые уникальные ключи, используя виртуальную валюту Telegram ⭐️.

📌 Вся информация приведена исключительно в ознакомительных целях❗️❗️❗️

#telegram

📌 Подпишись  Python_Scripts❗️

📌Автоматическое обнаружение пробоя уровней поддержки и сопротивления.
Доработка:
Фильтрация ложных пробоев с помощью индекса волатильности ATR (Average True Range).

Используем публичное API Bybit👨‍💻

import requests
import numpy as np

# Настройки
SYMBOL = "BTCUSDT"
KLINE_URL = "https://api.bybit.com/v5/market/kline"
INTERVAL = "15"  # 15-минутные свечи
LIMIT = 100  # Количество свечей для анализа (для ATR)
ATR_PERIOD = 14  # Период ATR
BREAKOUT_MULTIPLIER = 1.5  # Насколько цена должна превысить ATR для подтвержденного пробоя

def get_kline_data():
    """Получает исторические свечи"""
    params = {"category": "spot", "symbol": SYMBOL, "interval": INTERVAL, "limit": LIMIT}
    response = requests.get(KLINE_URL, params=params)

    if response.status_code == 200:
        return response.json().get("result", {}).get("list", [])
    else:
        print(f"Ошибка API: {response.status_code}")
        return None

def calculate_atr(candles, period=ATR_PERIOD):
    """Рассчитывает ATR (средний истинный диапазон)"""
    highs = np.array([float(c[2]) for c in candles])
    lows = np.array([float(c[3]) for c in candles])
    closes = np.array([float(c[4]) for c in candles])

    prev_closes = np.roll(closes, shift=1)  # Смещаем массив на 1 (добавится дубликат первого элемента)
    prev_closes[0] = closes[0]  # Убираем лишнее смещение

    tr = np.maximum(highs - lows, np.maximum(abs(highs - prev_closes), abs(lows - prev_closes)))
    atr = np.convolve(tr, np.ones(period) / period, mode='valid')

    return atr[-1]  # Берем последнее значение ATR

def find_support_resistance(candles):
    """Определяет уровни поддержки и сопротивления"""
    highs = np.array([float(c[2]) for c in candles])
    lows = np.array([float(c[3]) for c in candles])

    support = np.min(lows)
    resistance = np.max(highs)

    return support, resistance

def detect_breakout():
    """Проверяет пробой уровней с фильтрацией по ATR"""
    candles = get_kline_data()
    if not candles:
        return

    support, resistance = find_support_resistance(candles[-50:])  # Берем последние 50 свечей
    last_close = float(candles[-1][4])
    atr = calculate_atr(candles)

    breakout_threshold = atr * BREAKOUT_MULTIPLIER  # Минимальный разрыв для подтверждения пробоя

    if last_close >= resistance + breakout_threshold:
        print(f"🚀 Подтвержденный пробой вверх! Цена {last_close:.2f} выше сопротивления {resistance:.2f} (ATR {atr:.2f})")
    elif last_close <= support - breakout_threshold:
        print(f"⚠️ Подтвержденный пробой вниз! Цена {last_close:.2f} ниже поддержки {support:.2f} (ATR {atr:.2f})")
    else:
        print(f"📊 Цена {last_close:.2f}, диапазон: {support:.2f} - {resistance:.2f} (без пробоя)")

if __name__ == "__main__":
    detect_breakout()

📌 Подпишись  Python_Scripts❗️

🚀 Полезные библиотеки Python

💾 PyInfra

📌 Библиотека Python, для управления конфигурацией серверов и развертывания приложений с использованием простого Python-кода вместо громоздких YAML-файлов.

Особенности:

- Позволяет управлять серверами с использованием чистого Python.

- Работает без агентов на удалённых машинах.

- Поддерживает параллельное выполнение команд.

⚙️ Установка 👇👇👇

pip install pyinfra

📁 Документация

#библиотеки

📌 Подпишись Python_Scripts ❗️