📌 Межбиржевой арбитраж.

import time
import ccxt

def get_price(exchange, symbol):
    try:
        ticker = exchange.fetch_ticker(symbol)
        return ticker['last']
    except Exception as e:
        print(f"Ошибка при получении цены с {exchange.id}: {e}")
        return None

def main():
    bybit = ccxt.bybit()
    binance = ccxt.binance()
    symbol = 'BTC/USDT'

    while True:
        price_bybit = get_price(bybit, symbol)
        price_binance = get_price(binance, symbol)

        if price_bybit and price_binance:
            spread = abs(price_bybit - price_binance)
            print(f"Цена Bybit: {price_bybit}, Цена Binance: {price_binance}, Спред: {spread}")
            
            if spread > 200:  # Порог арбитражной возможности
                print("Арбитражная возможность найдена!")
            else:
                print("Нет арбитражных возможностей на данный момент.")
        
        time.sleep(5)  # Пауза перед следующим запросом

if __name__ == "__main__":
    main()

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

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

💾 Peewee

📌 Библиотека Python, простая и эффективная ORM (Object-Relational Mapping) для SQLite, PostgreSQL и MySQL. Она удобна для небольших и средних проектов, где не нужен сложный SQLAlchemy.

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

- Простота использования, минимальный код.

- Поддержка транзакций, связей и миграций.

- Совместимость с SQLite, PostgreSQL и MySQL.

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

pip install peewee

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

#библиотеки

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

📌 Скрипт coinflip

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

📌 Скрипт простой Telegram бот coinflip со ставкой в BTC.

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

#telegram

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

📌 Поиск некоторых свечных паттернов и вывод результата в консоль.

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

import requests
import pandas as pd

# Функция для получения исторических данных OHLCV с Bybit
def get_ohlcv(symbol="BTCUSDT", interval="15"):
    url = f"https://api.bybit.com/v5/market/kline?category=linear&symbol={symbol}&interval={interval}&limit=50"
    response = requests.get(url).json()
    if "result" in response and "list" in response["result"]:
        data = response["result"]["list"]
        df = pd.DataFrame(data, columns=["timestamp", "open", "high", "low", "close", "volume", "turnover"])
        df[["open", "high", "low", "close"]] = df[["open", "high", "low", "close"]].astype(float)
        df["timestamp"] = pd.to_datetime(df["timestamp"], unit="ms")
        df = df.sort_values("timestamp").reset_index(drop=True)
        return df
    return None

# Функция для определения свечных паттернов
def detect_patterns(df):
    patterns = []
    for i in range(1, len(df)):
        prev_candle = df.iloc[i - 1]
        candle = df.iloc[i]

        open_price, close_price = candle["open"], candle["close"]
        high, low = candle["high"], candle["low"]

        # 1. Молот (Hammer)
        if (close_price > open_price) and ((open_price - low) > 2 * (close_price - open_price)) and ((high - close_price) < 0.2 * (close_price - open_price)):
            patterns.append((candle["timestamp"], "Hammer"))

        # 2. Перевернутый молот (Inverted Hammer)
        elif (close_price > open_price) and ((high - close_price) > 2 * (close_price - open_price)) and ((open_price - low) < 0.2 * (close_price - open_price)):
            patterns.append((candle["timestamp"], "Inverted Hammer"))

        # 3. Бычье поглощение (Bullish Engulfing)
        elif (prev_candle["close"] < prev_candle["open"]) and (close_price > open_price) and (close_price > prev_candle["open"]) and (open_price < prev_candle["close"]):
            patterns.append((candle["timestamp"], "Bullish Engulfing"))

        # 4. Медвежье поглощение (Bearish Engulfing)
        elif (prev_candle["close"] > prev_candle["open"]) and (close_price < open_price) and (close_price < prev_candle["open"]) and (open_price > prev_candle["close"]):
            patterns.append((candle["timestamp"], "Bearish Engulfing"))

        # 5. Доджи (Doji)
        elif abs(open_price - close_price) < 0.1 * (high - low):
            patterns.append((candle["timestamp"], "Doji"))

    return patterns

# Основная логика
df = get_ohlcv()
if df is not None:
    detected_patterns = detect_patterns(df)
    if detected_patterns:
        print("Обнаружены паттерны:")
        for timestamp, pattern in detected_patterns:
            print(f"{timestamp}: {pattern}")
    else:
        print("Свечных паттернов не найдено.")
else:
    print("Ошибка получения данных с Bybit.")

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

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

💾 Turso

📌 Библиотека Python, облачная распределённая база данных, построенная на основе libSQL (форк SQLite). Она идеально подходит для приложений с низкой задержкой, где требуется быстрый доступ к данным.

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

- Поддержка горизонтального масштабирования на основе SQLite.

- Работает как распределённая база данных с низкой задержкой.

- Лёгкость интеграции в Python-приложения.

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

curl -sSfL https://get.tur.so/install.sh | bash

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

#библиотеки

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

📌 Скрипт AI агент

💻  ЯП:  Python 3.11+🐍
💾  Модули: flask, openai, python-telegram-bot, tweepy, aiohttp
📂  База данных: -

📌 Скрипт бот — помощник на основе искусственного интеллекта, который автоматически реагирует на упоминания и ответы на твиты.

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

#twitter

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

📌 Анализ рыночного настроения и поиск потенциальных торговых возможностей.

✅ Что показывает Open Interest (OI)?

🔹 Высокий OI → Больше открытых позиций → Растёт волатильность.

🔹 Резкое падение OI → Закрываются позиции → Возможен разворот тренда.

✅ Что показывает Funding Rate?

🔹 Высокий Funding Rate (> 0.01%) → Лонгисты платят шортам → Возможно снижение цены.

🔹 Отрицательный Funding Rate (< 0%) → Шорты платят лонгам → Возможен рост цены.

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

import requests
import time

# Конфиг
SYMBOL = "BTCUSDT"
INTERVAL = 30  # Интервал обновления (секунды)
FUNDING_THRESHOLD = 0.0003  # Новый порог тревоги (0.03% -> 0.0003)

def get_open_interest_and_funding():
    url = "https://api.bybit.com/v5/market/tickers"
    params = {"category": "linear", "symbol": SYMBOL}

    try:
        response = requests.get(url, params=params)
        data = response.json()

        if response.status_code == 200 and "result" in data:
            tickers = data["result"].get("list", [])
            if tickers:
                ticker = tickers[0]
                open_interest = float(ticker.get("openInterest", 0))
                funding_rate = float(ticker.get("fundingRate", 0))  # Оставляем как есть, без *100

                return open_interest, funding_rate
            else:
                print("[WARN] Данные отсутствуют. Возможно, малый объем торговли.")
                return None, None
        else:
            print(f"[ERROR] API вернул некорректный ответ: {data}")
            return None, None

    except requests.exceptions.RequestException as e:
        print(f"[ERROR] Ошибка сети: {e}")
        return None, None

while True:
    open_interest, funding_rate = get_open_interest_and_funding()

    if open_interest is not None and funding_rate is not None:
        print(f"[INFO] {SYMBOL} | Open Interest: {open_interest:.2f} | Funding Rate: {funding_rate:.6f}")

        # Улучшенные алерты
        if funding_rate >= FUNDING_THRESHOLD:
            print("[🔥 ALERT] Фондинг высокий! Возможен разворот вниз.")
        elif funding_rate <= -FUNDING_THRESHOLD:
            print("[🧊 ALERT] Фондинг отрицательный! Возможен шорт-сквиз.")
    else:
        print("[ERROR] Не удалось получить данные, повторная попытка...")

    time.sleep(INTERVAL)

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

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

💾 HTTPretty

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

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

- Позволяет эмулировать API-ответы для тестирования.

- Перехватывает HTTP-запросы requests, urllib3 и других клиентов.

- Поддержка динамических ответов, заголовков и задержек.

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

pip install httpretty

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

#библиотеки

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

📌 Скрипт юзернейм

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

📌 Скрипт Telegram бот, который на основе контекста заданного пользователем генерирует уникальный/свободный юзернейм Telegram и сохраняет его в БД.

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

#telegram

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

📌 Автоматическое отслеживание крупных маркет-ордеров (Whale Tracker) на Bybit.

📈 В чем польза?

✔ Отслеживание китов – Видно, когда крупные игроки входят в рынок.

✔ Анализ трендов – Если много крупных покупок → возможно, цена пойдет вверх.

✔ Быстрая реакция – Можно сразу принимать решения о входе/выходе из сделки.

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

import asyncio
import websockets
import json

# Минимальный объем сделки для "кита" в долларах
MIN_WHALE_ORDER = 100000  # $100,000

async def track_whales():
    url = "wss://stream.bybit.com/v5/public/spot"  # WebSocket API для SPOT рынка

    async with websockets.connect(url) as ws:
        # Подписываемся на поток трейдов BTCUSDT
        subscribe_msg = {
            "op": "subscribe",
            "args": ["publicTrade.BTCUSDT"]
        }
        await ws.send(json.dumps(subscribe_msg))
        
        print("[INFO] Подключено к Bybit WebSocket. Ожидание крупных сделок...")
        
        while True:
            response = await ws.recv()
            data = json.loads(response)

            if "data" in data:
                for trade in data["data"]:
                    try:
                        price = float(trade["p"])   # Цена сделки
                        size = float(trade["v"])    # Объем в BTC
                        side = trade["S"]           # "Buy" или "Sell"
                        trade_value = price * size  # Стоимость сделки в $

                        if trade_value >= MIN_WHALE_ORDER:
                            print(f"[🚀 WHALE ALERT] {side} {size} BTC @ {price} USDT (≈ {trade_value:,.2f} $)")

                    except KeyError as e:
                        print(f"[ERROR] Отсутствует ключ {e} в данных сделки: {trade}")
                    except ValueError as e:
                        print(f"[ERROR] Ошибка преобразования данных: {e}")

# Запуск асинхронного трекера
asyncio.run(track_whales())

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

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

💾 Xonsh

📌 Библиотека Python, мощный интерактивный командный интерпретатор, который объединяет возможности Bash и Python. Он позволяет выполнять команды оболочки (ls, cd, grep) наряду с Python-кодом без переключения между средами.

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

- Позволяет смешивать Bash и Python в одной среде.

- Поддерживает автодополнение, алиасы и расширения.

- Работает как интерактивный интерпретатор и скриптовый язык.

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

pip install xonsh

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

#библиотеки

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

📌 Скрипт шаблон магазина

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

📌 Скрипт Telegram-бот для цифрового интернет-магазина.
Бот использует веб-хуки и интегрирован с тремя платежными системами: ЮКасса, Telegram-Stars и Robocassa.
Включает админ-панель, профиль пользователя и каталог товаров с разбивкой на категории.

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

#telegram

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

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

import requests
import pandas as pd

# Конфигурация
SYMBOL = "BTCUSDT"
INTERVAL = "60"  # Таймфрейм (в минутах)
LIMIT = 200  # Количество свечей

# Функция загрузки данных
def get_historical_data(symbol, interval, limit):
    url = f"https://api.bybit.com/v5/market/kline?category=linear&symbol={symbol}&interval={interval}&limit={limit}"
    
    try:
        response = requests.get(url)
        response.raise_for_status()
        data = response.json()
        
        if "result" in data and "list" in data["result"]:
            df = pd.DataFrame(data["result"]["list"], columns=["open_time", "open", "high", "low", "close", "volume", "turnover"])
            df["open_time"] = pd.to_datetime(pd.to_numeric(df["open_time"]), unit="ms")  # Исправлено
            df[["open", "high", "low", "close", "volume"]] = df[["open", "high", "low", "close", "volume"]].astype(float)
            
            df = df.sort_values(by="open_time", ascending=True)  # Добавлено
            return df
        else:
            print("[ERROR] Пустой ответ API Bybit.")
            return None
    except requests.exceptions.RequestException as e:
        print(f"[ERROR] Ошибка запроса: {e}")
        return None

# Стратегия SMA (пересечение средних)
def apply_strategy(df):
    df["SMA_10"] = df["close"].rolling(window=10).mean()
    df["SMA_50"] = df["close"].rolling(window=50).mean()
    
    df["Signal"] = 0
    df.loc[df["SMA_10"] > df["SMA_50"], "Signal"] = 1  # Покупка
    df.loc[df["SMA_10"] < df["SMA_50"], "Signal"] = -1  # Продажа
    
    return df

# Основной код
print("[INFO] Загружаем данные...")
df = get_historical_data(SYMBOL, INTERVAL, LIMIT)

if df is not None:
    print("[INFO] Данные загружены. Применяем стратегию...")
    df = apply_strategy(df)
    print(df[["open_time", "close", "SMA_10", "SMA_50", "Signal"]].dropna().tail(20))
else:
    print("[ERROR] Не удалось получить данные.")

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

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

💾 Pypeln

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

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

- Простая многопоточная и многопроцессорная обработка данных.

- Интуитивный API, похожий на map и filter.

- Поддержка асинхронных задач (asyncio).

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

pip install pypeln

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

#библиотеки

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

📌 Скрипт загрузка видео в ваш YouTube канал

💻  ЯП:  Python 3.11+🐍
💾  Модули: pyrogram, google-api-python- client
📂  База данных: -

📌 Скрипт Telegram бот для автоматизации выгрузки видео в ваши YouTube канал.

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

#telegram #youtube

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

📌 Smart Order Routing для BTC/USDT👨‍💻

Функционал скрипта:

1. Запрашивает лучшие цены покупки (bid) и продажи (ask) на Bybit и Binance через REST API.

2. Сравнивает полученные данные и выбирает биржу с наилучшей ценой.

3. Выводит рекомендации, где лучше купить или продать.

import ccxt
import time

# Подключение к биржам
bybit = ccxt.bybit()
binance = ccxt.binance()

SYMBOL = "BTC/USDT"

def get_prices():
    prices = {}

    try:
        bybit_ticker = bybit.fetch_ticker(SYMBOL)
        prices["Bybit"] = {
            "bid": bybit_ticker["bid"],
            "ask": bybit_ticker["ask"]
        }
    except Exception as e:
        print(f"[ERROR] Ошибка получения данных с Bybit: {e}")

    try:
        binance_ticker = binance.fetch_ticker(SYMBOL)
        prices["Binance"] = {
            "bid": binance_ticker["bid"],
            "ask": binance_ticker["ask"]
        }
    except Exception as e:
        print(f"[ERROR] Ошибка получения данных с Binance: {e}")

    return prices

def find_best_route():
    prices = get_prices()

    if not prices or len(prices) < 2:
        print("[WARN] Недостаточно данных для расчета арбитража.")
        return
    
    best_buy = min(prices, key=lambda x: prices[x]["ask"])
    best_sell = max(prices, key=lambda x: prices[x]["bid"])

    buy_price = prices[best_buy]["ask"]
    sell_price = prices[best_sell]["bid"]

    print("\n📊 **Текущие цены на биржах:**")
    for exchange, data in prices.items():
        print(f" - {exchange}: Bid = {data['bid']} USDT | Ask = {data['ask']} USDT")

    print("\n📌 **Рекомендация:**")
    if best_buy != best_sell:
        profit = sell_price - buy_price
        print(f"✅ **Купить на {best_buy} за {buy_price:.2f} USDT**")
        print(f"✅ **Продать на {best_sell} за {sell_price:.2f} USDT**")
        print(f"📈 **Возможная прибыль: {profit:.2f} USDT**")
    else:
        print("⚠ Разница цен недостаточна для арбитража.")

if __name__ == "__main__":
    while True:
        print("\n[INFO] Поиск лучшей цены...")
        find_best_route()
        time.sleep(10)  # Обновление каждые 10 секунд

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

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

💾 CCXT

📌 Библиотека Python, для работы с API более 100 криптовалютных бирж. Она позволяет получать рыночные данные, управлять ордерами и автоматизировать торговые стратегии.

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

- Поддержка 100+ бирж, включая Binance, Kraken, Coinbase, Bybit, KuCoin.

- Унифицированный API — одинаковые методы для всех бирж.

- Работа с публичными и приватными API (торговля, управление балансом, исторические данные).

- Поддержка WebSocket для потокового получения данных (через ccxt.pro).

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

pip install ccxt

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

#библиотеки

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

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

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

📌 Скрипт Telegram бот, который который автоматически извлекает данные токена, ссылки на соцсети - Twitter и Telegram из DexScreener.

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

#telegram

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

📌 Автоматический анализ волатильности криптовалют с использованием ccxt

Этот скрипт анализирует волатильность криптовалют на бирже Binance (или другой, поддерживаемой ccxt). Он помогает определить самые волатильные активы, что полезно для скальпинга, арбитража и высокочастотной торговли.

🔥 Как это поможет трейдеру?

✅ Скальпинг – находите монеты с высокой волатильностью.

✅ Адаптация – можно встроить в бота для торговли на пробой или контртренд.

✅ Риск-менеджмент – понимание волатильности помогает определять размер стоп-лосса.

import ccxt
import pandas as pd
import time

# Инициализация Binance API
exchange = ccxt.binance()

# Параметры анализа
SYMBOLS = ["BTC/USDT", "ETH/USDT", "BNB/USDT", "SOL/USDT", "XRP/USDT"]
TIMEFRAME = "1h"  # Таймфрейм (можно поменять)
LIMIT = 50  # Количество свечей

def fetch_data(symbol, timeframe, limit):
    """ Получает исторические данные """
    try:
        ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
        df = pd.DataFrame(ohlcv, columns=["timestamp", "open", "high", "low", "close", "volume"])
        df["timestamp"] = pd.to_datetime(df["timestamp"], unit="ms")
        return df
    except Exception as e:
        print(f"[ERROR] Ошибка загрузки данных {symbol}: {e}")
        return None

def calculate_volatility(df):
    """ Вычисляет волатильность """
    df["range"] = df["high"] - df["low"]  # Размах свечи
    df["atr"] = df["range"].rolling(window=14).mean()  # ATR (средний размах)
    df["std_dev"] = df["close"].rolling(window=14).std()  # Стандартное отклонение
    return df

def main():
    print("[INFO] Анализ волатильности активов...")
    
    results = []
    
    for symbol in SYMBOLS:
        df = fetch_data(symbol, TIMEFRAME, LIMIT)
        if df is not None:
            df = calculate_volatility(df)
            latest = df.iloc[-1]
            results.append({
                "symbol": symbol,
                "last_price": latest["close"],
                "range": latest["range"],
                "atr": latest["atr"],
                "std_dev": latest["std_dev"]
            })
    
    # Сортировка по ATR (самые волатильные активы в топе)
    df_results = pd.DataFrame(results).sort_values(by="atr", ascending=False)
    print("\n[RESULTS] Топ волатильных активов:\n")
    print(df_results)

if __name__ == "__main__":
    main()

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