Đây là bài viết thứ 47 trong chuỗi hướng dẫn Trading Bot của HolySheep AI. Trong bài viết này, mình sẽ hướng dẫn bạn cách xây dựng một hệ thống giám sát và cảnh báo real-time cho dữ liệu liquidation (thanh lý) trên sàn OKX futures, sử dụng Tardis API và xử lý dữ liệu bằng HolySheep AI.

Từ kinh nghiệm thực chiến của mình, việc nắm bắt được dữ liệu liquidation sớm có thể giúp bạn:

Giới thiệu hệ thống

Hệ thống này bao gồm 3 thành phần chính:

Kiến thức cần chuẩn bị

Nếu bạn là người mới hoàn toàn, đừng lo lắng. Mình sẽ giải thích mọi thứ từ đầu. Bạn chỉ cần:

Bước 1: Đăng ký các dịch vụ cần thiết

1.1. Đăng ký Tardis API

Tardis là dịch vụ cung cấp dữ liệu thị trường crypto chất lượng cao, bao gồm dữ liệu liquidation. Truy cập tardis.ai và đăng ký tài khoản.

Gợi ý ảnh chụp màn hình: Giao diện đăng ký Tardis với các trường Email, Password

Sau khi đăng ký, bạn sẽ nhận được API key dạng:

tardis_api_key = "your_tardis_api_key_here"

1.2. Đăng ký HolySheep AI

HolySheep AI là nền tảng API AI với tỷ giá ¥1=$1 (tiết kiệm 85%+ so với các provider khác), hỗ trợ WeChat/Alipay, độ trễ <50ms, và tín dụng miễn phí khi đăng ký.

Gợi ý ảnh chụp màn hình: Trang đăng ký HolySheep với gói tín dụng miễn phí

👉 Đăng ký tại đây để nhận tín dụng miễn phí và bắt đầu sử dụng ngay hôm nay.

Bước 2: Cài đặt môi trường Python

2.1. Kiểm tra Python

Mở terminal (Command Prompt trên Windows, Terminal trên Mac) và gõ:

python3 --version

Nếu thấy thông báo dạng Python 3.9.0 hoặc cao hơn, bạn đã có Python. Nếu chưa, hãy tải và cài đặt từ python.org.

2.2. Tạo thư mục dự án

mkdir liquidation-monitor
cd liquidation-monitor

2.3. Tạo virtual environment (môi trường ảo)

Virtual environment giúp project của bạn không bị xung đột với các project khác:

python3 -m venv venv


Trên Windows:

venv\Scripts\activate


Trên Mac/Linux:

source venv/bin/activate

2.4. Cài đặt các thư viện cần thiết

pip install requests websockets pandas holy-sheep-sdk python-dotenv

Trong đó:

Bước 3: Thiết lập cấu trúc dự án

3.1. Tạo file cấu hình .env

Tạo file .env trong thư mục dự án để lưu trữ các API keys an toàn:

# Tardis API Configuration
TARDIS_API_KEY=your_tardis_api_key_here


HolySheep AI Configuration  

HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY


Alert Settings

TELEGRAM_BOT_TOKEN=your_telegram_bot_token
TELEGRAM_CHAT_ID=your_chat_id
DISCORD_WEBHOOK=your_discord_webhook

⚠️ Lưu ý bảo mật: Không bao giờ chia sẻ file .env với người khác hoặc đẩy lên GitHub. Thêm .env vào file .gitignore.

3.2. Cấu trúc thư mục hoàn chỉnh

liquidation-monitor/
├── .env                 # API keys (KHÔNG đẩy lên Git)
├── .gitignore
├── config.py            # Đọc cấu hình từ .env
├── tardis_client.py      # Kết nối Tardis WebSocket
├── data_processor.py     # Xử lý dữ liệu liquidation
├── alert_system.py       # Gửi cảnh báo
├── main.py              # Chương trình chính
└── requirements.txt

Bước 4: Viết code kết nối Tardis API

4.1. File config.py - Đọc cấu hình

# config.py
import os
from dotenv import load_dotenv

load_dotenv()


Tardis API

TARDIS_API_KEY = os.getenv("TARDIS_API_KEY")


HolySheep AI

HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY")
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"


Alert Channels

TELEGRAM_BOT_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN")
TELEGRAM_CHAT_ID = os.getenv("TELEGRAM_CHAT_ID")
DISCORD_WEBHOOK = os.getenv("DISCORD_WEBHOOK")


Trading Settings

LIQUIDATION_THRESHOLD_LONG = 100000  # $100K cho long liquidation
LIQUIDATION_THRESHOLD_SHORT = 100000  # $100K cho short liquidation

4.2. File tardis_client.py - Kết nối WebSocket

Tardis cung cấp dữ liệu liquidation thông qua WebSocket. WebSocket là kết nối liên tục, cho phép nhận dữ liệu tức thì khi có sự kiện xảy ra.

# tardis_client.py
import json
import asyncio
import websockets
from websockets.exceptions import ConnectionClosed
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class TardisClient:
    """Kết nối WebSocket với Tardis API để nhận dữ liệu liquidation"""
    
    TARDIS_WS_URL = "wss://ws.tardis.dev/v1/stream"
    
    def __init__(self, api_key, exchange="okx", channel="liquidations"):
        self.api_key = api_key
        self.exchange = exchange
        self.channel = channel
        self.connection = None
        self.message_callback = None
    
    async def connect(self):
        """Thiết lập kết nối WebSocket"""
        params = {
            "exchange": self.exchange,
            "channel": self.channel,
            "key": self.api_key
        }
        
        query_string = "&".join([f"{k}={v}" for k, v in params.items()])
        ws_url = f"{self.TARDIS_WS_URL}?{query_string}"
        
        logger.info(f"Đang kết nối Tardis WebSocket...")
        self.connection = await websockets.connect(ws_url)
        logger.info("✅ Kết nối Tardis thành công!")
    
    def set_message_callback(self, callback):
        """Đặt function xử lý khi nhận được message"""
        self.message_callback = callback
    
    async def listen(self):
        """Lắng nghe và xử lý messages liên tục"""
        try:
            async for message in self.connection:
                try:
                    data = json.loads(message)
                    
                    if self.message_callback:
                        await self.message_callback(data)
                        
                except json.JSONDecodeError as e:
                    logger.error(f"Lỗi parse JSON: {e}")
                    
        except ConnectionClosed as e:
            logger.error(f"Kết nối bị đóng: {e.code} - {e.reason}")
            await self.reconnect()
    
    async def reconnect(self):
        """Tự động kết nối lại khi bị mất kết nối"""
        logger.info("Đang thử kết nối lại sau 5 giây...")
        await asyncio.sleep(5)
        await self.connect()
        await self.listen()
    
    async def close(self):
        """Đóng kết nối"""
        if self.connection:
            await self.connection.close()
            logger.info("Đã đóng kết nối Tardis")

Gợi ý ảnh chụp màn hình: Terminal hiển thị log kết nối thành công với Tardis

Bước 5: Xử lý dữ liệu với HolySheep AI

5.1. File data_processor.py - Xử lý và phân tích

Sau khi nhận dữ liệu liquidation từ Tardis, chúng ta cần xử lý và phân tích để đưa ra cảnh báo có ý nghĩa. Đây là lúc HolySheep AI phát huy sức mạnh.

# data_processor.py
import pandas as pd
import json
from datetime import datetime
import config
import requests

class LiquidationProcessor:
    """Xử lý và phân tích dữ liệu liquidation"""
    
    def __init__(self):
        self.liquidation_history = []
        self.holy_sheep_api_key = config.HOLYSHEEP_API_KEY
        self.holy_sheep_base_url = config.HOLYSHEEP_BASE_URL
    
    def process_liquidation(self, data):
        """Xử lý một record liquidation từ Tardis"""
        try:
            # Tardis trả về dữ liệu theo định dạng của họ
            liquidation = {
                "timestamp": data.get("timestamp", datetime.now().isoformat()),
                "symbol": data.get("symbol", "UNKNOWN"),
                "side": data.get("side", "unknown"),  # "buy" = short liquidation, "sell" = long liquidation
                "price": float(data.get("price", 0)),
                "size": float(data.get("size", 0)),  # Số lượng
                "unit": data.get("unit", "USDT"),
                "contract_type": data.get("contractType", "perpetual")
            }
            
            # Tính giá trị USD của liquidation
            liquidation["value_usd"] = liquidation["price"] * liquidation["size"]
            
            # Xác định loại liquidation
            if liquidation["side"] == "buy":
                liquidation["type"] = "SHORT_LIQUIDATION"
                liquidation["description"] = "Short bị thanh lý (giá tăng)"
            else:
                liquidation["type"] = "LONG_LIQUIDATION"
                liquidation["description"] = "Long bị thanh lý (giá giảm)"
            
            # Thêm vào lịch sử
            self.liquidation_history.append(liquidation)
            
            # Chỉ giữ 1000 records gần nhất
            if len(self.liquidation_history) > 1000:
                self.liquidation_history = self.liquidation_history[-1000:]
            
            return liquidation
            
        except Exception as e:
            print(f"Lỗi xử lý liquidation: {e}")
            return None
    
    def get_statistics(self, minutes=60):
        """Lấy thống kê liquidation trong N phút gần nhất"""
        now = datetime.now()
        recent = [
            liq for liq in self.liquidation_history
            if self._is_within_minutes(liq["timestamp"], minutes)
        ]
        
        if not recent:
            return {
                "total_long_liquidations": 0,
                "total_short_liquidations": 0,
                "long_value_usd": 0,
                "short_value_usd": 0,
                "total_value_usd": 0
            }
        
        df = pd.DataFrame(recent)
        
        stats = {
            "total_long_liquidations": len(df[df["type"] == "LONG_LIQUIDATION"]),
            "total_short_liquidations": len(df[df["type"] == "SHORT_LIQUIDATION"]),
            "long_value_usd": df[df["type"] == "LONG_LIQUIDATION"]["value_usd"].sum(),
            "short_value_usd": df[df["type"] == "SHORT_LIQUIDATION"]["value_usd"].sum(),
            "total_value_usd": df["value_usd"].sum(),
            "largest_single": df["value_usd"].max() if len(df) > 0 else 0,
            "count": len(df)
        }
        
        return stats
    
    def _is_within_minutes(self, timestamp_str, minutes):
        """Kiểm tra timestamp có trong N phút gần đây không"""
        try:
            timestamp = datetime.fromisoformat(timestamp_str.replace("Z", "+00:00"))
            now = datetime.now()
            diff = (now - timestamp).total_seconds() / 60
            return diff <= minutes
        except:
            return False
    
    def analyze_with_ai(self, stats):
        """Sử dụng HolySheep AI để phân tích dữ liệu liquidation"""
        prompt = f"""Phân tích dữ liệu liquidation OKX futures:

Thống kê 60 phút gần nhất:
- Tổng Long bị thanh lý: {stats['total_long_liquidations']} lần, trị giá ${stats['long_value_usd']:,.0f}
- Tổng Short bị thanh lý: {stats['total_short_liquidations']} lần, trị giá ${stats['short_value_usd']:,.0f}
- Tổng giá trị: ${stats['total_value_usd']:,.0f}
- Liquidation lớn nhất: ${stats['largest_single']:,.0f}

Hãy phân tích:
1. Xu hướng thị trường hiện tại (bullish/bearish)?
2. Có dấu hiệu panic selling/buying không?
3. Khuyến nghị cho traders ngắn hạn?

Trả lời ngắn gọn, dễ hiểu, có emoji."""
        
        try:
            response = requests.post(
                f"{self.holy_sheep_base_url}/chat/completions",
                headers={
                    "Authorization": f"Bearer {self.holy_sheep_api_key}",
                    "Content-Type": "application/json"
                },
                json={
                    "model": "gpt-4o",
                    "messages": [{"role": "user", "content": prompt}],
                    "temperature": 0.7,
                    "max_tokens": 500
                },
                timeout=10
            )
            
            if response.status_code == 200:
                result = response.json()
                return result["choices"][0]["message"]["content"]
            else:
                return f"Không thể phân tích AI (Error: {response.status_code})"
                
        except Exception as e:
            return f"Lỗi kết nối HolySheep AI: {str(e)}"

5.2. Test kết nối HolySheep AI

Trước khi tích hợp, hãy test xem HolySheep AI hoạt động tốt không:

# test_holy_sheep.py
import requests
import config

def test_holy_sheep_connection():
    """Test kết nối HolySheep AI"""
    
    headers = {
        "Authorization": f"Bearer {config.HOLYSHEEP_API_KEY}",
        "Content-Type": "application/json"
    }
    
    payload = {
        "model": "gpt-4o",
        "messages": [
            {"role": "user", "content": "Trả lời ngắn: OKX liquidation là gì?"}
        ],
        "max_tokens": 50
    }
    
    try:
        response = requests.post(
            f"{config.HOLYSHEEP_BASE_URL}/chat/completions",
            headers=headers,
            json=payload,
            timeout=10
        )
        
        if response.status_code == 200:
            result = response.json()
            print("✅ Kết nối HolySheep AI thành công!")
            print(f"Response: {result['choices'][0]['message']['content']}")
            print(f"Usage: {result.get('usage', {})}")
            return True
        else:
            print(f"❌ Lỗi: {response.status_code}")
            print(response.text)
            return False
            
    except Exception as e:
        print(f"❌ Lỗi kết nối: {e}")
        return False

if __name__ == "__main__":
    test_holy_sheep_connection()

Gợi ý ảnh chụp màn hình: Output terminal cho thấy kết nối thành công với thời gian phản hồi

Bước 6: Hệ thống cảnh báo

6.1. File alert_system.py - Gửi thông báo

# alert_system.py
import requests
import logging
from datetime import datetime

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class AlertSystem:
    """Hệ thống gửi cảnh báo qua nhiều kênh"""
    
    def __init__(self, config):
        self.telegram_token = config.TELEGRAM_BOT_TOKEN
        self.telegram_chat_id = config.TELEGRAM_CHAT_ID
        self.discord_webhook = config.DISCORD_WEBHOOK
    
    def format_liquidation_alert(self, liquidation, stats=None):
        """Format message cảnh báo liquidation"""
        
        emoji = "🔴" if liquidation["type"] == "LONG_LIQUIDATION" else "🔵"
        side_text = "LONG" if liquidation["type"] == "LONG_LIQUIDATION" else "SHORT"
        
        message = f"""
{emoji} *LIQUIDATION ALERT*

📊 Symbol: {liquidation['symbol']}
📉 Side: *{side_text}*
💰 Giá: ${liquidation['price']:,.2f}
📦 Size: {liquidation['size']:,.4f}
💵 Value: ${liquidation['value_usd']:,.0f}

⏰ Time: {liquidation['timestamp']}
"""
        
        if stats:
            message += f"""
📈 Thống kê 60 phút:
   • Long liq: {stats['total_long_liquidations']} lần (${stats['long_value_usd']:,.0f})
   • Short liq: {stats['total_short_liquidations']} lần (${stats['short_value_usd']:,.0f})
   • Tổng: ${stats['total_value_usd']:,.0f}
"""
        
        return message.strip()
    
    def send_telegram(self, message):
        """Gửi thông báo qua Telegram"""
        if not self.telegram_token or not self.telegram_chat_id:
            logger.warning("Telegram config chưa được thiết lập")
            return False
        
        url = f"https://api.telegram.org/bot{self.telegram_token}/sendMessage"
        payload = {
            "chat_id": self.telegram_chat_id,
            "text": message,
            "parse_mode": "Markdown"
        }
        
        try:
            response = requests.post(url, json=payload, timeout=10)
            if response.status_code == 200:
                logger.info("✅ Telegram alert sent")
                return True
            else:
                logger.error(f"❌ Telegram error: {response.text}")
                return False
        except Exception as e:
            logger.error(f"❌ Telegram exception: {e}")
            return False
    
    def send_discord(self, message):
        """Gửi thông báo qua Discord Webhook"""
        if not self.discord_webhook:
            logger.warning("Discord webhook chưa được thiết lập")
            return False
        
        payload = {
            "content": message,
            "username": "Liquidation Bot"
        }
        
        try:
            response = requests.post(
                self.discord_webhook, 
                json=payload, 
                timeout=10
            )
            if response.status_code in [200, 204]:
                logger.info("✅ Discord alert sent")
                return True
            else:
                logger.error(f"❌ Discord error: {response.text}")
                return False
        except Exception as e:
            logger.error(f"❌ Discord exception: {e}")
            return False
    
    def send_alert(self, liquidation, stats=None):
        """Gửi cảnh báo qua tất cả các kênh đã cấu hình"""
        message = self.format_liquidation_alert(liquidation, stats)
        
        # Gửi đồng thời qua nhiều kênh
        self.send_telegram(message)
        self.send_discord(message)
        
        # In ra console
        print(message)
    
    def send_summary(self, ai_analysis):
        """Gửi tổng hợp phân tích AI qua Telegram"""
        if not self.telegram_token or not self.telegram_chat_id:
            return
        
        message = f"""
📊 *LIQUIDATION SUMMARY*

{ai_analysis}
"""
        
        self.send_telegram(message)

Gợi ý ảnh chụp màn hình: Telegram nhận được thông báo liquidation với format đẹp

Bước 7: Chương trình chính

7.1. File main.py - Điều phối toàn bộ hệ thống

# main.py
import asyncio
import logging
import signal
import sys
from datetime import datetime, timedelta

import config
from tardis_client import TardisClient
from data_processor import LiquidationProcessor
from alert_system import AlertSystem

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

class LiquidationMonitor:
    """Hệ thống giám sát liquidation chính"""
    
    def __init__(self):
        self.tardis = TardisClient(
            api_key=config.TARDIS_API_KEY,
            exchange="okx",
            channel="liquidations"
        )
        self.processor = LiquidationProcessor()
        self.alert_system = AlertSystem(config)
        
        self.last_summary_time = datetime.now()
        self.summary_interval = timedelta(minutes=30)
        
        # Flags
        self.running = True
        
        # Đăng ký signal handler
        signal.signal(signal.SIGINT, self.signal_handler)
        signal.signal(signal.SIGTERM, self.signal_handler)
    
    def signal_handler(self, signum, frame):
        """Xử lý khi nhận được signal dừng"""
        logger.info("Nhận signal dừng, đang shutdown...")
        self.running = False
    
    async def on_liquidation(self, data):
        """Xử lý khi nhận được dữ liệu liquidation"""
        
        # Xử lý dữ liệu
        liquidation = self.processor.process_liquidation(data)
        
        if not liquidation:
            return
        
        # Kiểm tra ngưỡng cảnh báo
        should_alert = self.should_send_alert(liquidation)
        
        if should_alert:
            stats = self.processor.get_statistics(minutes=60)
            self.alert_system.send_alert(liquidation, stats)
        
        # Gửi tổng hợp định kỳ
        self.check_periodic_summary()
    
    def should_send_alert(self, liquidation):
        """Quyết định có gửi cảnh báo không"""
        
        # Luôn alert liquidation lớn hơn ngưỡng
        value = liquidation["value_usd"]
        
        if liquidation["type"] == "LONG_LIQUIDATION":
            return value >= config.LIQUIDATION_THRESHOLD_LONG
        else:
            return value >= config.LIQUIDATION_THRESHOLD_SHORT
    
    def check_periodic_summary(self):
        """Gửi tổng hợp định kỳ"""
        now = datetime.now()
        
        if now - self.last_summary_time >= self.summary_interval:
            self.last_summary_time = now
            
            stats = self.processor.get_statistics(minutes=60)
            
            if stats["count"] > 0:
                logger.info(f"Gửi tổng hợp định kỳ: {stats['count']} liquidations")
                
                # Phân tích bằng AI
                ai_analysis = self.processor.analyze_with_ai(stats)
                
                # Gửi tổng hợp
                self.alert_system.send_summary(ai_analysis)
    
    async def start(self):
        """Bắt đầu hệ thống giám sát"""
        logger.info("=" * 50)
        logger.info("🚀 KHỞI ĐỘNG HỆ THỐNG GIÁM SÁT LIQUIDATION")
        logger.info("=" * 50)
        
        # Thiết lập callback
        self.tardis.set_message_callback(self.on_liquidation)
        
        try:
            # Kết nối và lắng nghe
            await self.tardis.connect()
            await self.tardis.listen()
            
        except KeyboardInterrupt:
            logger.info("Nhận Ctrl+C, đang dừng...")
        finally:
            await self.stop()
    
    async def stop(self):
        """Dừng hệ thống"""
        logger.info("Đang dừng hệ thống...")
        await self.tardis.close()
        logger.info("Đã dừng thành công!")


async def main():
    """Entry point"""
    monitor = LiquidationMonitor()
    await monitor.start()


if __name__ == "__main__":
    asyncio.run(main())

7.2. Chạy chương trình

# Chạy với virtual environment đã kích hoạt
python main.py

Gợi ý ảnh chụp màn hình: Terminal hiển thị hệ thống đang chạy và log các liquidation events

Bước 8: Tối ưu hóa và best practices

8.1. Thêm tính năng lọc theo symbol

Nếu bạn chỉ muốn theo dõi một số cặp nhất định:

# Thêm vào class LiquidationProcessor
def __init__(self, symbols=None):
    # ... existing init code ...
    self.symbols = symbols or ["BTC-USDT-SWAP", "ETH-USDT-SWAP"]  # Mặc định

def filter_by_symbol(self, liquidation):
    """Lọc liquidation theo symbol"""
    if not self.symbols:
        return True
    return liquidation["symbol"] in self.symbols

8.2. Thêm database để lưu trữ

Để phân tích lâu dài, bạn nên lưu trữ dữ liệu:

# Thêm vào data_processor.py
import sqlite3

def save_to_database(self, liquidation):
    """Lưu liquidation vào SQLite"""
    conn = sqlite3.connect('liquidations.db')
    cursor = conn.cursor()
    
    cursor.execute("""
        INSERT INTO liquidations 
        (timestamp, symbol, side, price, size, value_usd, type)
        VALUES (?, ?, ?, ?, ?, ?, ?)
    """, (
        liquidation["timestamp"],
        liquidation["symbol"],
        liquidation["side"],
        liquidation["price"],
        liquidation["size"],
        liquidation["value_usd"],
        liquidation["type"]
    ))
    
    conn.commit()
    conn.close()

8.3. Tạo dashboard đơn giản

# dashboard.py
from flask import Flask, render_template
import sqlite3

app = Flask(__name__)

@app.route('/')
def dashboard():
    conn = sqlite3.connect('liquidations.db')
    cursor = conn.cursor()
    
    # Lấy 50 liquidation gần nhất
    cursor.execute("""
        SELECT * FROM liquidations 
        ORDER BY timestamp DESC 
        LIMIT 50
    """)
    recent = cursor.fetchall()
    
    # Thống kê 24h
    cursor.execute("""
        SELECT 
            type,
            COUNT(*) as count,
            SUM(value_usd) as total
        FROM liquidations
        WHERE timestamp > datetime('now', '-24 hours')
        GROUP BY type
    """)
    stats = cursor

Tài nguyên liên quan
Bài viết liên quan