จากประสบการณ์การเทรดคริปโตฯ ของผู้เขียนที่ใช้งาน Bybit มานานกว่า 3 ปี ปัญหาที่พบบ่อยที่สุดคือการจัดการหลายบัญชีพร้อมกัน วันหนึ่งผมเจอสถานการณ์ที่ราคา BTC ลดลง 15% ภายใน 1 ชั่วโมง แต่กลับไม่สามารถประเมินความเสียหายรวมได้ทันท่วงที เพราะต้องเปิดดูทีละบัญชี สุดท้ายพอร์ตโฟลิโอถูก Liquidation ไปทั้งหมด 3 บัญชี ขาดทุนรวมกว่า $50,000

บทความนี้จะสอนวิธีสร้างระบบ Bybit持仓监控 ที่สามารถรวมข้อมูลจากหลายบัญชี และคำนวณความเสี่ยงรวมแบบเรียลไทม์ พร้อมแนะนำเครื่องมือ AI ที่ช่วยวิเคราะห์ข้อมูลได้อย่างมีประสิทธิภาพ

ปัญหาหลักของการจัดการหลายบัญชี Bybit

การดึงข้อมูลพอร์ตโฟลิโอจากหลายบัญชี

เริ่มจากการสร้าง Client สำหรับเชื่อมต่อ Bybit API ทั้งหมดในครั้งเดียว

import requests
import time
from typing import Dict, List, Optional
from dataclasses import dataclass
from decimal import Decimal

@dataclass
class Position:
    symbol: str
    size: float
    entry_price: float
    mark_price: float
    unrealized_pnl: float
    leverage: int
    margin: float
    account_name: str

class BybitPortfolioMonitor:
    """ระบบติดตามพอร์ตโฟลิโอ Bybit หลายบัญชี"""
    
    def __init__(self):
        self.accounts: Dict[str, dict] = {}
        self.base_url = "https://api.bybit.com"
        
    def add_account(self, name: str, api_key: str, api_secret: str, 
                   testnet: bool = False) -> None:
        """เพิ่มบัญชี Bybit ที่ต้องการติดตาม"""
        self.accounts[name] = {
            'api_key': api_key,
            'api_secret': api_secret,
            'testnet': testnet,
            'base_url': "https://api-testnet.bybit.com" if testnet else "https://api.bybit.com"
        }
        
    def get_all_positions(self) -> List[Position]:
        """ดึงข้อมูลสถานะทั้งหมดจากทุกบัญชี"""
        all_positions = []
        
        for account_name, credentials in self.accounts.items():
            try:
                positions = self._fetch_positions(credentials, account_name)
                all_positions.extend(positions)
            except Exception as e:
                print(f"❌ ข้อผิดพลาดบัญชี {account_name}: {e}")
                
        return all_positions
        
    def _fetch_positions(self, credentials: dict, account_name: str) -> List[Position]:
        """ดึงข้อมูลสถานะจากบัญชีเดียว"""
        # ใช้ bybit-connector หรือ requests ตามที่ต้องการ
        endpoint = f"{credentials['base_url']}/v5/position/list"
        
        headers = {
            'X-BAPI-API-KEY': credentials['api_key'],
            'X-BAPI-SIGN': self._generate_signature(credentials),
            'X-BAPI-SIGN-TYPE': '2',
            'X-BAPI-TIMESTAMP': str(int(time.time() * 1000)),
            'Content-Type': 'application/json'
        }
        
        response = requests.get(endpoint, headers=headers, params={'category': 'linear'})
        response.raise_for_status()
        
        data = response.json()
        if data['retCode'] != 0:
            raise ConnectionError(f"API Error: {data['retMsg']}")
            
        positions = []
        for item in data['result']['list']:
            if float(item['size']) != 0:  # กรองเฉพาะสถานะที่เปิดอยู่
                positions.append(Position(
                    symbol=item['symbol'],
                    size=float(item['size']),
                    entry_price=float(item['avgPrice']),
                    mark_price=float(item['markPrice']),
                    unrealized_pnl=float(item['unrealizedPnl']),
                    leverage=int(item['leverage']),
                    margin=float(item['positionIM']),
                    account_name=account_name
                ))
                
        return positions

การคำนวณ Risk Exposure รวมทุกบัญชี

หลังจากได้ข้อมูลสถานะจากทุกบัญชีแล้ว ขั้นตอนสำคัญคือการคำนวณ Risk Exposure รวม

from collections import defaultdict
import numpy as np

class RiskAnalyzer:
    """เครื่องมือวิเคราะห์ความเสี่ยงรวมจากหลายบัญชี"""
    
    def calculate_total_exposure(self, positions: List[Position]) -> Dict[str, float]:
        """คำนวณ exposure รวมแยกตามสินทรัพย์"""
        exposure = defaultdict(float)
        
        for pos in positions:
            # กำหนด direction: Long=+, Short=-
            direction = 1 if pos.size > 0 else -1
            # exposure = size * price (ใช้ mark_price ปัจจุบัน)
            exposure[pos.symbol] += abs(pos.size) * pos.mark_price * direction
            
        return dict(exposure)
        
    def detect_hedge_ratio(self, positions: List[Position], 
                          target_symbol: str) -> float:
        """ตรวจจับสัดส่วน Hedge ของสินทรัพย์เดียวกันในหลายบัญชี"""
        longs = 0
        shorts = 0
        
        for pos in positions:
            if pos.symbol != target_symbol:
                continue
            if pos.size > 0:
                longs += pos.size * pos.mark_price
            else:
                shorts += abs(pos.size) * pos.mark_price
                
        if longs + shorts == 0:
            return 0.0
            
        # hedge_ratio = min(longs, shorts) / max(longs, shorts)
        hedge_ratio = min(longs, shorts) / max(longs, shorts) if max(longs, shorts) > 0 else 0
        return hedge_ratio
        
    def calculate_portfolio_var(self, positions: List[Position], 
                               confidence: float = 0.95) -> float:
        """คำนวณ Value at Risk (VaR) ของพอร์ตรวม"""
        total_pnl = sum(p.unrealized_pnl for p in positions)
        
        # ประมาณ VaR จาก volatility ของแต่ละสถานะ
        # ใช้ assumptions ที่ conservative สำหรับ crypto
        volatilities = {
            'BTC': 0.03,      # 3% daily volatility
            'ETH': 0.04,
            'SOL': 0.06,
            'DEFAULT': 0.05
        }
        
        portfolio_value = sum(abs(p.size) * p.mark_price for p in positions)
        
        # VaR = Portfolio Value * Volatility * Z-score
        z_scores = {0.90: 1.28, 0.95: 1.65, 0.99: 2.33}
        z = z_scores.get(confidence, 1.65)
        
        # Weighted volatility
        weighted_vol = 0
        for pos in positions:
            vol = volatilities.get(pos.symbol.replace('USDT', ''), volatilities['DEFAULT'])
            weight = abs(pos.size) * pos.mark_price / portfolio_value if portfolio_value > 0 else 0
            weighted_vol += weight * vol
            
        var = portfolio_value * weighted_vol * z
        return var
        
    def generate_risk_report(self, positions: List[Position]) -> dict:
        """สร้างรายงานความเสี่ยงแบบครบวงจร"""
        report = {
            'total_accounts': len(set(p.account_name for p in positions)),
            'total_positions': len(positions),
            'total_exposure': sum(abs(p.size) * p.mark_price for p in positions),
            'total_unrealized_pnl': sum(p.unrealized_pnl for p in positions),
            'exposure_by_symbol': self.calculate_total_exposure(positions),
            'var_95': self.calculate_portfolio_var(positions, 0.95),
            'liquidation_risk': self.assess_liquidation_risk(positions)
        }
        
        # ตรวจจับ over-hedged หรือ under-hedged
        symbols = set(p.symbol for p in positions)
        for symbol in symbols:
            hedge = self.detect_hedge_ratio(positions, symbol)
            if hedge < 0.3:
                report[f'{symbol}_warning'] = 'Low hedge protection'
            elif hedge > 0.9:
                report[f'{symbol}_warning'] = 'Near-perfect hedge - check if intentional'
                
        return report
        
    def assess_liquidation_risk(self, positions: List[Position]) -> List[dict]:
        """ประเมินความเสี่ยง Liquidation ของแต่ละสถานะ"""
        risks = []
        
        for pos in positions:
            # คำนวณระยะห่างจากราคา Liquidation
            # สำหรับ Long: (entry - liq_price) / entry
            # สำหรับ Short: (liq_price - entry) / entry
            
            if pos.size > 0:  # Long
                risk_pct = abs(pos.unrealized_pnl) / pos.margin if pos.margin > 0 else 0
            else:  # Short
                risk_pct = abs(pos.unrealized_pnl) / pos.margin if pos.margin > 0 else 0
                
            # ถ้า unrealized loss > 50% ของ margin = high risk
            if risk_pct > 0.5:
                risks.append({
                    'symbol': pos.symbol,
                    'account': pos.account_name,
                    'risk_level': 'HIGH',
                    'loss_ratio': risk_pct
                })
                
        return risks

การส่ง Alert เมื่อความเสี่ยงสูง

ระบบที่ดีต้องสามารถแจ้งเตือนได้ทันท่วงที โดยเฉพาะเมื่อความเสี่ยงเกินเกณฑ์ที่กำหนด

import json
from datetime import datetime

class RiskAlertSystem:
    """ระบบแจ้งเตือนความเสี่ยงแบบ Real-time"""
    
    def __init__(self, telegram_bot_token: str = None, 
                 line_notify_token: str = None):
        self.telegram_token = telegram_bot_token
        self.line_token = line_notify_token
        self.alert_thresholds = {
            'max_total_exposure': 100000,      # $100,000
            'max_var_pct': 0.15,               # 15% of portfolio
            'max_single_position_pct': 0.30,   # 30% of portfolio
            'liquidation_warning_pct': 0.5     # 50% margin consumed
        }
        
    def check_and_alert(self, risk_report: dict, positions: List[Position]) -> None:
        """ตรวจสอบเกณฑ์และส่ง Alert หากเกิน"""
        alerts = []
        
        # 1. ตรวจสอบ Total Exposure
        if risk_report['total_exposure'] > self.alert_thresholds['max_total_exposure']:
            alerts.append({
                'type': 'EXPOSURE_LIMIT',
                'severity': 'HIGH',
                'message': f"⚠️ ความเสี่ยงรวมสูงเกิน ${risk_report['total_exposure']:,.2f}"
            })
            
        # 2. ตรวจสอบ VaR
        var_pct = risk_report['var_95'] / risk_report['total_exposure'] if risk_report['total_exposure'] > 0 else 0
        if var_pct > self.alert_thresholds['max_var_pct']:
            alerts.append({
                'type': 'VAR_WARNING',
                'severity': 'MEDIUM',
                'message': f"📊 VaR (95%) สูง: {var_pct:.1%} ของพอร์ต"
            })
            
        # 3. ตรวจสอบ Liquidation Risk
        for risk in risk_report.get('liquidation_risk', []):
            alerts.append({
                'type': 'LIQUIDATION_RISK',
                'severity': 'CRITICAL',
                'message': f"🚨 {risk['symbol']} ใน {risk['account']}: "
                          f"Loss {risk['loss_ratio']:.1%} ของ Margin"
            })
            
        # ส่ง Alert ทั้งหมด
        for alert in alerts:
            self._send_alert(alert)
            
    def _send_alert(self, alert: dict) -> None:
        """ส่ง Alert ผ่านช่องทางที่กำหนด"""
        message = f"[{datetime.now().strftime('%H:%M:%S')}] {alert['message']}"
        
        if self.telegram_token:
            self._send_telegram(message)
            
        if self.line_token:
            self._send_line(message)
            
        print(message)
        
    def _send_telegram(self, message: str, chat_id: str) -> None:
        """ส่งข้อความผ่าน Telegram Bot"""
        url = f"https://api.telegram.org/bot{self.telegram_token}/sendMessage"
        payload = {
            'chat_id': chat_id,
            'text': message,
            'parse_mode': 'HTML'
        }
        requests.post(url, json=payload)
        
    def _send_line(self, message: str) -> None:
        """ส่งข้อความผ่าน LINE Notify"""
        url = "https://notify-api.line.me/api/notify"
        headers = {'Authorization': f'Bearer {self.line_token}'}
        payload = {'message': message}
        requests.post(url, headers=headers, data=payload)

การใช้ AI วิเคราะห์ความเสี่ยงเชิงลึก

นอกจากการคำนวณเชิงตัวเลขแล้ว การใช้ AI ช่วยวิเคราะห์สามารถให้ข้อมูลเชิงลึกที่ซับซ้อนกว่า เช่น การวิเคราะห์ Correlation ระหว่างสินทรัพย์ หรือการคาดการณ์ความเสี่ยงในอนาคต

ผู้เขียนแนะนำให้ใช้ HolySheep AI ซึ่งมีความเร็วในการตอบสนองต่ำกว่า 50ms และรองรับโมเดลหลากหลาย เหมาะสำหรับการประมวลผลข้อมูลพอร์ตโฟลิโอจำนวนมากแบบเรียลไทม์

import json

class AIPortfolioAnalyzer:
    """ใช้ AI วิเคราะห์พอร์ตโฟลิโอแบบลึก"""
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://api.holysheep.ai/v1"  # ใช้ HolySheep API
        
    def analyze_portfolio_risk(self, risk_report: dict, 
                              positions: List[Position]) -> str:
        """ใช้ AI วิเคราะห์และให้คำแนะนำ"""
        
        # สร้าง Summary ของพอร์ต
        portfolio_summary = self._create_portfolio_summary(risk_report, positions)
        
        prompt = f"""คุณเป็นผู้เชี่ยวชาญด้าน Risk Management สำหรับ Cryptocurrency
        วิเคราะห์พอร์ตโฟลิโอต่อไปนี้และให้คำแนะนำ:
        
        {portfolio_summary}
        
        ระบุ:
        1. ความเสี่ยงหลักที่ต้องจัดการ
        2. การกระจายความเสี่ยงที่เหมาะสม
        3. ข้อแนะนำเฉพาะบุคคล
        """
        
        response = self._call_ai(prompt)
        return response
        
    def generate_hedge_strategy(self, positions: List[Position]) -> dict:
        """สร้างกลยุทธ์ Hedging อัตโนมัติ"""
        
        exposure_summary = {}
        for pos in positions:
            symbol = pos.symbol.replace('USDT', '')
            if symbol not in exposure_summary:
                exposure_summary[symbol] = {'long': 0, 'short': 0}
            if pos.size > 0:
                exposure_summary[symbol]['long'] += abs(pos.size) * pos.mark_price
            else:
                exposure_summary[symbol]['short'] += abs(pos.size) * pos.mark_price
                
        prompt = f"""จากข้อมูล Exposure ต่อไปนี้:
        {json.dumps(exposure_summary, indent=2)}
        
        ออกแบบกลยุทธ์ Hedging ที่เหมาะสม โดยระบุ:
        - สัดส่วนที่ควร Hedge
        - ราคาเป้าหมายสำหรับ Orders ใหม่
        - Stop Loss ที่แนะนำ
        """
        
        response = self._call_ai(prompt)
        return json.loads(response)
        
    def _create_portfolio_summary(self, risk_report: dict, 
                                  positions: List[Position]) -> str:
        """สร้าง Summary สำหรับส่งให้ AI"""
        summary_parts = [
            f"จำนวนบัญชี: {risk_report['total_accounts']}",
            f"จำนวนสถานะ: {risk_report['total_positions']}",
            f"Exposure รวม: ${risk_report['total_exposure']:,.2f}",
            f"Unrealized PnL: ${risk_report['total_unrealized_pnl']:,.2f}",
            f"VaR (95%): ${risk_report['var_95']:,.2f}",
            "\nรายละเอียดต่อสินทรัพย์:",
        ]
        
        for symbol, exposure in risk_report['exposure_by_symbol'].items():
            summary_parts.append(f"  - {symbol}: ${exposure:,.2f}")
            
        return "\n".join(summary_parts)
        
    def _call_ai(self, prompt: str) -> str:
        """เรียก HolySheep AI API"""
        url = f"{self.base_url}/chat/completions"
        
        headers = {
            'Authorization': f'Bearer {self.api_key}',
            'Content-Type': 'application/json'
        }
        
        payload = {
            'model': 'gpt-4.1',  # ใช้โมเดลที่เหมาะสม
            'messages': [
                {'role': 'system', 'content': 'คุณเป็นผู้เชี่ยวชาญด้าน Crypto Risk Management'},
                {'role': 'user', 'content': prompt}
            ],
            'temperature': 0.3,
            'max_tokens': 2000
        }
        
        response = requests.post(url, headers=headers, json=payload)
        response.raise_for_status()
        
        data = response.json()
        return data['choices'][0]['message']['content']

การรวมทุกอย่างเข้าด้วยกัน

import schedule
import time
import threading

def run_daily_risk_check():
    """รันการตรวจสอบความเสี่ยงทุกวัน"""
    
    # 1. ตั้งค่า Monitor
    monitor = BybitPortfolioMonitor()
    
    # เพิ่มบัญชีทั้งหมด
    monitor.add_account(
        name='Main Account',
        api_key='YOUR_BYBIT_API_KEY_1',
        api_secret='YOUR_BYBIT_SECRET_1',
        testnet=False
    )
    monitor.add_account(
        name='Hedge Account',
        api_key='YOUR_BYBIT_API_KEY_2',
        api_secret='YOUR_BYBIT_SECRET_2',
        testnet=False
    )
    
    # 2. ตั้งค่า AI Analyzer
    ai_analyzer = AIPortfolioAnalyzer(
        api_key='YOUR_HOLYSHEEP_API_KEY'  # ใช้ HolySheep สำหรับ AI
    )
    
    # 3. ตั้งค่า Alert
    alerts = RiskAlertSystem(
        telegram_bot_token='YOUR_TELEGRAM_TOKEN',
        line_notify_token='YOUR_LINE_TOKEN'
    )
    
    # 4. ดึงข้อมูลและวิเคราะห์
    print("📊 กำลังดึงข้อมูลพอร์ตโฟลิโอ...")
    positions = monitor.get_all_positions()
    
    print("🔍 กำลังวิเคราะห์ความเสี่ยง...")
    analyzer = RiskAnalyzer()
    risk_report = analyzer.generate_risk_report(positions)
    
    print("📱 กำลังส่ง Alert...")
    alerts.check_and_alert(risk_report, positions)
    
    print("🤖 กำลังขอคำแนะนำจาก AI...")
    ai_advice = ai_analyzer.analyze_portfolio_risk(risk_report, positions)
    print(f"\n💡 คำแนะนำจาก AI:\n{ai_advice}")
    
    return risk_report


รันทุก 15 นาที

schedule.every(15).minutes.do(run_daily_risk_check)


รันทุกวันตอน 09:00 และ 21:00

schedule.every().day.at("09:00").do(run_daily_risk_check)
schedule.every().day.at("21:00").do(run_daily_risk_check)

if __name__ == '__main__':
    print("🚀 เริ่มต้นระบบติดตามความเสี่ยง Bybit")
    run_daily_risk_check()  # รันทันที
    
    while True:
        schedule.run_pending()
        time.sleep(60)

ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข

ข้อผิดพลาด สาเหตุ วิธีแก้ไข
ConnectionError: timeout Bybit API ตอบสนองช้าเกินไป หรือ Rate Limit เพิ่ม Retry Logic และ Exponential Backoff:
for attempt in range(3):
    try:
        response = requests.get(url, timeout=30)
        response.raise_for_status()
        break
    except requests.exceptions.Timeout:
        if attempt < 2:
            time.sleep(2 ** attempt)  # 1s, 2s, 4s
        else:
            raise
401 Unauthorized API Key หมดอายุ หรือ Signature ไม่ถูกต้อง ตรวจสอบ Timestamp และ Signature Algorithm:
import hmac
import hashlib

def generate_signature(secret, timestamp, recv_window, query_string):
    param_str = f'{timestamp}{api_key}{recv_window}{query_string}'
    return hmac.new(
        secret.encode('utf-8'),
        param_str.encode('utf-8'),
        hashlib.sha256
    ).hexdigest()
RateLimitError: 10006 เรียก API บ่อยเกินไป (เกิน 10 requests/second) ใช้ Rate Limiter และ Cache:
import time
from functools import wraps

def rate_limit(calls=10, period=1):
    def decorator(func):
        last_called = [0]
        @wraps(func)
        def wrapper(*args, **kwargs):
            elapsed = time.time() - last_called[0]
            if elapsed < period / calls:
                time.sleep(period / calls - elapsed)
            result = func(*args, **kwargs)
            last_called[0] = time.time()
            return result
        return wrapper
    return decorator
Position data mismatch ข้อมูลระหว่าง Linear แลา Inverse contracts ไม่ตรงกัน ใช้ unified margin account หรือ filter ตาม category:
# Linear: USDT Perpetual

Inverse: BTCUSD Perpetual

categories = {
    'linear': ['BTCUSDT', 'ETHUSDT'],
    'inverse': ['BTCUSD', 'ETHUSD']
}

for category, symbols in categories.items():
    params = {'category': category, 'symbol': symbols[0]}
    response = requests.get(endpoint, params=params, headers=headers)

เหมาะกับใคร / ไม่เหมาะกับใคร

แหล่งข้อมูลที่เกี่ยวข้อง

บทความที่เกี่ยวข้อง