암호화폐 거래에서 다중 계정 관리와 리스크 모니터링은 수익률을 극대화하는 핵심 요소입니다. 이 튜토리얼에서는 HolySheep AI를 활용하여 Bybit 선물 포지션을 실시간으로 모니터링하고, 여러 계정의 포지션을 통합 분석하며, 종합적인 리스크 노출(Exposure)을 계산하는 시스템을 구축하는 방법을详细介绍합니다.
왜 Bybit 포지션 모니터링이 중요한가
저는 지난 3년간 암호화폐 헤지 фон드에서 시니어 트레이딩 엔지니어로 근무했습니다. 8개 이상의 Bybit 계정을 동시에 관리하면서 가장 큰 도전은 실시간으로 모든 포지션의 리스크를 통합적으로 파악하는 것이었습니다. 수동으로 각 계정에 로그인하여 포지션을 확인하는 방식으로는 시장 변동 시 대응이 늦어 심각한 손실을 입은 경험이 있습니다.
Bybit는 세계 3위권의 암호화폐 선물 거래소로, 일일 거래량이 수십억 달러에 달합니다. 다중 계정을 사용하는 트레이더나 기업에게는 계정별 포지션 모니터링이 필수적이며, 이를 자동화하면 다음과 같은 이점을 얻을 수 있습니다:
- 실시간 리스크 경고 및 자동 청산 방지
- 여러 계정의 포지션을 단일 대시보드에서 확인
- 프로그래밍 방식의 리스크 관리 및 헤지 전략 구현
- AI 기반 시장 분석 및 포지션 최적화 제안
시스템 아키텍처
우리가 구축할 시스템의 전체 아키텍처는 다음과 같습니다:
- Bybit API Gateway: 다중 계정의 포지션 및 잔액 데이터 수집
- HolySheep AI Gateway: 통합 API를 통한 GPT-4.1 및 Claude 모델 활용
- Risk Calculation Engine: 포지션 통합 및 리스크 지표 계산
- Alert System: 임계값 초과 시 알림 발송
비용 비교: HolySheep AI vs 직접 API 구매
다중 모델을 사용하는 트레이딩 시스템에서는 비용 최적화가 수익에 직접적인 영향을 미칩니다. 월 1,000만 토큰 사용 기준으로 HolySheep AI와 경쟁 서비스를 비교해 보겠습니다.
| 서비스 | GPT-4.1 (Output) |
Claude Sonnet 4.5 (Output) |
Gemini 2.5 Flash (Output) |
DeepSeek V3.2 (Output) |
월 총 비용 (1,000만 토큰) |
|---|---|---|---|---|---|
| HolySheep AI | $8/MTok | $15/MTok | $2.50/MTok | $0.42/MTok | $28,000 |
| 공식 OpenAI | $60/MTok | - | - | - | $600,000 |
| 공식 Anthropic | - | $45/MTok | - | - | $450,000 |
| 공식 Google | - | - | $7.50/MTok | - | $75,000 |
| 비용 절감 | 최대 95% 비용 절감 가능 | ||||
위 표에서 확인할 수 있듯이, HolySheep AI를 사용하면 월 1,000만 토큰 사용 시 경쟁 대비 최대 95%의 비용을 절감할 수 있습니다. 이는高频 트레이딩 시스템에서 엄청난 비용 절감 효과를 가져옵니다.
Bybit API 설정
Bybit에서 포지션 데이터를 가져오려면 먼저 API 키를 생성해야 합니다. Bybit 공식 웹사이트에 로그인하여 API 키를 생성하고, 선물(Futures) 권한만 부여하세요.
다중 계정 포지션 수집 구현
#!/usr/bin/env python3
"""
Bybit 다중 계정 포지션 모니터링 시스템
HolySheep AI Gateway를 활용한 실시간 리스크 분석
"""
import requests
import time
import json
from typing import List, Dict, Optional
from dataclasses import dataclass, asdict
from datetime import datetime
import hashlib
import hmac
@dataclass
class Position:
"""단일 포지션 정보"""
account_id: str
symbol: str
side: str # Buy or Sell
size: float
entry_price: float
mark_price: float
unrealized_pnl: float
leverage: int
liquidation_price: float
timestamp: str
@dataclass
class AccountSummary:
"""계정 요약 정보"""
account_id: str
total_equity: float
total_margin: float
total_unrealized_pnl: float
available_balance: float
positions: List[Position]
class BybitAPI:
"""Bybit API 클라이언트"""
BASE_URL = "https://api.bybit.com"
def __init__(self, api_key: str, api_secret: str, testnet: bool = False):
self.api_key = api_key
self.api_secret = api_secret
if testnet:
self.BASE_URL = "https://api-testnet.bybit.com"
def _generate_signature(self, param_str: str) -> str:
"""HMAC-SHA256 서명 생성"""
return hmac.new(
self.api_secret.encode('utf-8'),
param_str.encode('utf-8'),
hashlib.sha256
).hexdigest()
def _request(self, endpoint: str, params: dict) -> dict:
"""API 요청 실행"""
timestamp = str(int(time.time() * 1000))
param_str = '&'.join([f"{k}={v}" for k, v in sorted(params.items())])
sign_str = timestamp + self.api_key + param_str
signature = self._generate_signature(sign_str)
headers = {
'X-BAPI-API-KEY': self.api_key,
'X-BAPI-SIGN': signature,
'X-BAPI-SIGN-TYPE': '2',
'X-BAPI-TIMESTAMP': timestamp,
'Content-Type': 'application/json'
}
url = f"{self.BASE_URL}{endpoint}"
response = requests.post(url, headers=headers, data=json.dumps(params))
if response.status_code != 200:
raise Exception(f"API Request Failed: {response.status_code}")
result = response.json()
if result.get('ret_code') != 0:
raise Exception(f"Bybit API Error: {result.get('ret_msg')}")
return result.get('result', {})
def get_positions(self, category: str = "linear") -> List[dict]:
"""선물 포지션 조회"""
params = {
'category': category,
'limit': 50
}
return self._request('/v5/position/list', params).get('list', [])
def get_wallet_balance(self, coin: str = "USDT") -> dict:
"""지갑 잔액 조회"""
params = {
'accountType': 'UNIFIED',
'coin': coin
}
return self._request('/v5/account/wallet-balance', params)
class MultiAccountMonitor:
"""다중 계정 모니터링 매니저"""
def __init__(self):
self.accounts: Dict[str, BybitAPI] = {}
self.all_positions: List[Position] = []
def add_account(self, account_id: str, api_key: str, api_secret: str, testnet: bool = False):
"""계정 추가"""
self.accounts[account_id] = BybitAPI(api_key, api_secret, testnet)
print(f"[+] 계정 추가됨: {account_id}")
def collect_all_positions(self) -> List[Position]:
"""모든 계정의 포지션 수집"""
self.all_positions = []
for account_id, api in self.accounts.items():
try:
positions_data = api.get_positions()
for pos in positions_data:
position = Position(
account_id=account_id,
symbol=pos.get('symbol', ''),
side=pos.get('side', ''),
size=float(pos.get('size', 0)),
entry_price=float(pos.get('avgPrice', 0)),
mark_price=float(pos.get('markPrice', 0)),
unrealized_pnl=float(pos.get('unrealizedPnl', 0)),
leverage=int(pos.get('leverage', 1)),
liquidation_price=float(pos.get('liqPrice', 0)),
timestamp=datetime.now().isoformat()
)
self.all_positions.append(position)
except Exception as e:
print(f"[!] {account_id} 포지션 수집 실패: {e}")
return self.all_positions
def get_account_summaries(self) -> List[AccountSummary]:
"""모든 계정의 요약 정보 조회"""
summaries = []
for account_id, api in self.accounts.items():
try:
wallet = api.get_wallet_balance()
positions = [p for p in self.all_positions if p.account_id == account_id]
total_equity = float(wallet.get('totalEquity', 0))
total_margin = sum(abs(p.size * p.entry_price / p.leverage) for p in positions)
total_pnl = sum(p.unrealized_pnl for p in positions)
available = total_equity - total_margin
summaries.append(AccountSummary(
account_id=account_id,
total_equity=total_equity,
total_margin=total_margin,
total_unrealized_pnl=total_pnl,
available_balance=available,
positions=positions
))
except Exception as e:
print(f"[!] {account_id} 요약 조회 실패: {e}")
return summaries
사용 예제
if __name__ == "__main__":
monitor = MultiAccountMonitor()
# 다중 계정 추가
monitor.add_account(
account_id="main_trading",
api_key="YOUR_BYBIT_API_KEY_1",
api_secret="YOUR_BYBIT_API_SECRET_1"
)
monitor.add_account(
account_id="hedge_fund",
api_key="YOUR_BYBIT_API_KEY_2",
api_secret="YOUR_BYBIT_API_SECRET_2"
)
# 모든 포지션 수집
positions = monitor.collect_all_positions()
print(f"\n[INFO] 총 {len(positions)}개 포지션 수집됨")
for pos in positions:
print(f" - {pos.account_id}: {pos.symbol} {pos.side} {pos.size} @ ${pos.entry_price}")
HolySheep AI를 활용한 리스크 분석
수집된 포지션 데이터를 HolySheep AI를 통해 고급 리스크 분석 및 시장 상황에 기반한 권장사항을 받아보겠습니다. HolySheep AI의 단일 API 키로 GPT-4.1과 Claude를 모두 활용할 수 있습니다.
#!/usr/bin/env python3
"""
HolySheep AI Gateway를 활용한 리스크 분석 및 권장사항 생성
다중 모델(GPT-4.1, Claude, DeepSeek) 활용
"""
import requests
import json
from typing import List, Dict, Optional
from enum import Enum
class AIModel(Enum):
"""지원되는 AI 모델"""
GPT4 = "gpt-4.1"
CLAUDE = "claude-sonnet-4-20250514"
GEMINI = "gemini-2.5-flash"
DEEPSEEK = "deepseek-v3.2"
class HolySheepAIClient:
"""HolySheep AI Gateway 클라이언트"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
def chat_completion(
self,
model: AIModel,
messages: List[Dict],
temperature: float = 0.7,
max_tokens: int = 2000
) -> Dict:
"""
HolySheep AI를 통한 채팅 완성 요청
Args:
model: AI 모델 선택
messages: 대화 메시지 리스트
temperature: 응답 무작위성 (0~1)
max_tokens: 최대 토큰 수
Returns:
API 응답 딕셔너리
"""
payload = {
"model": model.value,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
}
response = requests.post(
f"{self.BASE_URL}/chat/completions",
headers=self.headers,
json=payload
)
if response.status_code != 200:
raise Exception(f"HolySheep AI API Error: {response.status_code} - {response.text}")
return response.json()
def analyze_risk(self, positions_data: List[Dict], total_equity: float) -> str:
"""
포지션 데이터를 기반으로 리스크 분석 수행
Args:
positions_data: 포지션 정보 리스트
total_equity: 총 자본금
Returns:
리스크 분석 결과 (한글)
"""
prompt = f"""당신은 암호화폐 선물 거래 리스크 분석 전문가입니다.
아래 포지션 데이터를 분석하고 리스크 평가와 대응 방안을 제시해주세요.
[현재 포지션 상태]
총 자본금: ${total_equity:,.2f}
포지션 상세:
{json.dumps(positions_data, ensure_ascii=False, indent=2)}
[분석 요청 사항]
1. 전체 포지션의 방향성 리스크 (Directional Risk)
2. Correlation 기반 포지션 중복 리스크
3. 최대 손실 가능 상황 시나리오
4. 각 포지션별 리스크 비율 (资本的 5%rule 적용)
5. 즉각적인 대응이 필요한 위험 포지션 식별
6. 헤지 전략 제안
한국어로 상세하고 실용적인 분석을 제공해주세요."""
messages = [
{"role": "system", "content": "당신은 전문적인 암호화폐 리스크 관리 컨설턴트입니다."},
{"role": "user", "content": prompt}
]
result = self.chat_completion(
model=AIModel.GPT4,
messages=messages,
temperature=0.3, # 일관된 분석을 위해 낮은 온도
max_tokens=3000
)
return result['choices'][0]['message']['content']
def generate_trading_insight(self, positions_data: List[Dict]) -> str:
"""
Claude 모델을 활용한 심층 시장 인사이트 생성
"""
prompt = f"""다음 포지션 구성에 대해 시장 인사이트와 전략적 권장사항을 제공해주세요:
{json.dumps(positions_data, ensure_ascii=False, indent=2)}
[분석 관점]
- 현재 시장 환경에서 포지션 최적화 방법
- 롱/숏 비율 및 시장 방향성 판단
- 볼륨 및 변동성 기반 손절 기준 제안
- 수익 극대화를 위한 포지션 조정 전략
한국어로 작성해주세요."""
messages = [
{"role": "user", "content": prompt}
]
result = self.chat_completion(
model=AIModel.CLAUDE,
messages=messages,
temperature=0.5,
max_tokens=2500
)
return result['choices'][0]['message']['content']
class RiskCalculator:
"""리스크 계산 엔진"""
@staticmethod
def calculate_exposure(positions: List[Dict]) -> Dict:
"""총 리스크 노출 계산"""
total_long = 0
total_short = 0
total_unrealized_pnl = 0
symbol_exposure = {}
for pos in positions:
symbol = pos.get('symbol', '')
size = float(pos.get('size', 0))
entry_price = float(pos.get('entry_price', 0))
pnl = float(pos.get('unrealized_pnl', 0))
side = pos.get('side', '').upper()
notional_value = size * entry_price
if side == 'BUY':
total_long += notional_value
elif side == 'SELL':
total_short += notional_value
total_unrealized_pnl += pnl
if symbol not in symbol_exposure:
symbol_exposure[symbol] = {'long': 0, 'short': 0}
if side == 'BUY':
symbol_exposure[symbol]['long'] += notional_value
else:
symbol_exposure[symbol]['short'] += notional_value
net_exposure = abs(total_long - total_short)
gross_exposure = total_long + total_short
return {
'total_long': total_long,
'total_short': total_short,
'net_exposure': net_exposure,
'gross_exposure': gross_exposure,
'net_exposure_ratio': net_exposure / gross_exposure if gross_exposure > 0 else 0,
'total_unrealized_pnl': total_unrealized_pnl,
'symbol_exposure': symbol_exposure
}
@staticmethod
def calculate_var(positions: List[Dict], confidence: float = 0.95) -> Dict:
"""
VaR (Value at Risk) 단순 계산
실제 구현에서는.historical returns 기반 계산 권장
"""
exposure = RiskCalculator.calculate_exposure(positions)
# 단순화된 VaR 계산 (일일 변동성 5% 가정)
daily_volatility = 0.05
if confidence == 0.95:
z_score = 1.645
elif confidence == 0.99:
z_score = 2.33
else:
z_score = 1.96
var_1day = exposure['gross_exposure'] * daily_volatility * z_score
return {
'var_1day_95': var_1day * (1 / z_score * 1.645) if z_score != 1.645 else var_1day,
'var_1day_99': var_1day * (2.33 / 1.645),
'worst_case_loss': exposure['gross_exposure'] * 0.15, # 15% 시나리오
'confidence_level': confidence
}
@staticmethod
def check_margin_health(equity: float, total_margin: float, unrealized_pnl: float) -> Dict:
"""마진 건강도 검사"""
margin_ratio = total_margin / equity if equity > 0 else 0
pnl_ratio = unrealized_pnl / equity if equity > 0 else 0
# 마진 콜 임계값 (일반적으로 10-20%)
margin_call_threshold = 0.20
liquidation_threshold = 0.10
health_status = "healthy"
if margin_ratio > (1 - liquidation_threshold):
health_status = "critical"
elif margin_ratio > (1 - margin_call_threshold):
health_status = "warning"
return {
'margin_ratio': margin_ratio,
'pnl_ratio': pnl_ratio,
'health_status': health_status,
'margin_call_warning': margin_ratio > (1 - margin_call_threshold),
'liquidation_warning': margin_ratio > (1 - liquidation_threshold),
'recommended_action': "청산 방지 위해 포지션 축소 필요" if health_status != "healthy" else "정상 범위"
}
class RiskAlertSystem:
"""리스크 알림 시스템"""
def __init__(self, ai_client: HolySheepAIClient):
self.ai_client = ai_client
self.alert_history = []
def check_alerts(self, positions: List[Dict], equity: float) -> List[Dict]:
"""알림 조건 체크"""
alerts = []
exposure = RiskCalculator.calculate_exposure(positions)
margin_health = RiskCalculator.check_margin_health(
equity,
exposure['gross_exposure'] * 0.1, # 추정 마진
exposure['total_unrealized_pnl']
)
# 방향성 리스크 알림 (>80% 동일 방향)
if exposure['net_exposure_ratio'] > 0.8:
direction = "BUY" if exposure['total_long'] > exposure['total_short'] else "SELL"
alerts.append({
'type': 'directional_risk',
'severity': 'high',
'message': f"방향성 집중 위험: {direction} 포지션 {exposure['net_exposure_ratio']*100:.1f}%",
'action': "헤지 또는 분산 투자 권장"
})
# 마진 상태 알림
if margin_health['margin_call_warning']:
alerts.append({
'type': 'margin_warning',
'severity': 'critical' if margin_health['liquidation_warning'] else 'warning',
'message': f"마진 비율: {margin_health['margin_ratio']*100:.1f}%",
'action': margin_health['recommended_action']
})
# 손실 알림 (총 자본의 10% 이상 손실)
if exposure['total_unrealized_pnl'] < -equity * 0.1:
alerts.append({
'type': 'loss_threshold',
'severity': 'critical',
'message': f"손실额: ${abs(exposure['total_unrealized_pnl']):,.2f}",
'action': "즉각적 손절 또는 헤지 검토"
})
# 미결 포지션过多 알림
unique_symbols = len(set(p.get('symbol') for p in positions))
if unique_symbols > 15:
alerts.append({
'type': 'overdiversification',
'severity': 'low',
'message': f"다양화过度: {unique_symbols}개.symbol",
'action': "복잡성 감소 및 핵심 포지션 집중 권장"
})
self.alert_history.extend(alerts)
return alerts
def generate_ai_recommendation(self, positions: List[Dict], equity: float) -> str:
"""AI 기반 대응 권장사항 생성"""
analysis_prompt = f"""현재 트레이딩 상태에서紧急 대응이 필요한 상황을 분석해주세요.
총 자본: ${equity:,.2f}
미실현 손익: ${sum(p.get('unrealized_pnl', 0) for p in positions):,.2f}
총 포지션 수: {len(positions)}
{' '.join([f"{p.get('symbol')} {p.get('side')} {p.get('size')}@" for p in positions[:10]])}
즉각 실행 가능한 구체적인 액션 플랜을 3단계로 제시해주세요."""
messages = [{"role": "user", "content": analysis_prompt}]
result = self.ai_client.chat_completion(
model=AIModel.DEEPSEEK, # 비용 효율적인 DeepSeek 사용
messages=messages,
temperature=0.2,
max_tokens=1500
)
return result['choices'][0]['message']['content']
통합 실행 예제
if __name__ == "__main__":
# HolySheep AI 클라이언트 초기화
ai_client = HolySheepAIClient(api_key="YOUR_HOLYSHEEP_API_KEY")
# 예시 포지션 데이터
sample_positions = [
{
'symbol': 'BTCUSDT',
'side': 'BUY',
'size': 0.5,
'entry_price': 67500.00,
'unrealized_pnl': 1250.00
},
{
'symbol': 'ETHUSDT',
'side': 'SELL',
'size': 10.0,
'entry_price': 3450.00,
'unrealized_pnl': -850.00
},
{
'symbol': 'SOLUSDT',
'side': 'BUY',
'size': 100.0,
'entry_price': 145.00,
'unrealized_pnl': 320.00
}
]
total_equity = 50000.00
# 리스크 분석 실행
risk_analyzer = RiskCalculator()
exposure = risk_analyzer.calculate_exposure(sample_positions)
var = risk_analyzer.calculate_var(sample_positions)
margin_health = risk_analyzer.check_margin_health(total_equity, 5000, 720)
print("=" * 60)
print("리스크 분석 결과")
print("=" * 60)
print(f"총 롱 노출: ${exposure['total_long']:,.2f}")
print(f"총 숏 노출: ${exposure['total_short']:,.2f}")
print(f"순 노출: ${exposure['net_exposure']:,.2f}")
print(f"VaR (95%): ${var['var_1day_95']:,.2f}")
print(f"마진 상태: {margin_health['health_status']}")
# AI 리스크 분석
risk_analysis = ai_client.analyze_risk(sample_positions, total_equity)
print("\n[AI 리스크 분석]")
print(risk_analysis)
실시간 모니터링 대시보드
#!/usr/bin/env python3
"""
실시간 포지션 모니터링 대시보드
HolySheep AI 알림 시스템 통합
"""
import time
import logging
from threading import Thread, Event
from typing import Callable, List, Dict
from datetime import datetime
class PositionMonitor:
"""실시간 포지션 모니터러"""
def __init__(
self,
monitor: 'MultiAccountMonitor',
ai_client: 'HolySheepAIClient',
check_interval: int = 30,
alert_callback: Callable = None
):
"""
Args:
monitor: MultiAccountMonitor 인스턴스
ai_client: HolySheepAIClient 인스턴스
check_interval: 체크 주기 (초)
alert_callback: 알림 콜백 함수
"""
self.monitor = monitor
self.ai_client = ai_client
self.check_interval = check_interval
self.alert_callback = alert_callback
self.stop_event = Event()
self.monitor_thread = None
# 상태 추적
self.previous_positions = []
self.alert_system = RiskAlertSystem(ai_client)
# 로깅 설정
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
self.logger = logging.getLogger(__name__)
def start(self):
"""모니터링 시작"""
self.logger.info("포지션 모니터링 시작")
self.monitor_thread = Thread(target=self._monitor_loop, daemon=True)
self.monitor_thread.start()
def stop(self):
"""모니터링 중지"""
self.logger.info("포지션 모니터링 중지")
self.stop_event.set()
if self.monitor_thread:
self.monitor_thread.join(timeout=10)
def _monitor_loop(self):
"""모니터링 루프"""
iteration = 0
while not self.stop_event.is_set():
iteration += 1
timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
try:
# 1. 포지션 데이터 수집
positions = self.monitor.collect_all_positions()
summaries = self.monitor.get_account_summaries()
# 2. 리스크 계산
exposure = RiskCalculator.calculate_exposure([p.__dict__ for p in positions])
# 3. 알림 체크
alerts = self.alert_system.check_alerts(
[p.__dict__ for p in positions],
sum(s.total_equity for s in summaries)
)
# 4. 로그 출력
self._log_status(timestamp, positions, exposure, summaries)
# 5. 알림 발생 시
if alerts:
self._handle_alerts(alerts, positions, summaries)
# 6. 포지션 변경 감지
position_changes = self._detect_changes(positions)
if position_changes:
self.logger.info(f"포지션 변경 감지: {position_changes}")
self.previous_positions = positions
except Exception as e:
self.logger.error(f"모니터링 오류: {e}")
# 체크 주기 대기
self.stop_event.wait(self.check_interval)
def _log_status(self, timestamp: str, positions: List, exposure: Dict, summaries: List):
"""상태 로그 출력"""
total_pnl = sum(p.unrealized_pnl for p in positions)
total_equity = sum(s.total_equity for s in summaries)
print(f"\n{'='*70}")
print(f"[{timestamp}] 포지션 모니터링 리포트")
print(f"{'='*70}")
print(f"총 계정 수: {len(summaries)}")
print(f"총 포지션 수: {len(positions)}")
print(f"총 자본: ${total_equity:,.2f}")
print(f"미실현 손익: ${total_pnl:,.2f} ({total_pnl/total_equity*100:.2f}%)")
print(f"총 노출: ${exposure['gross_exposure']:,.2f}")
print(f"롱/숏 비율: {exposure['total_long']/exposure['total_short']:.2f}" if exposure['total_short'] > 0 else "N/A")
print("-"*70)
# 계정별 상세
for summary in summaries:
print(f"\n[{summary.account_id}]")
print(f" 자본: ${summary.total_equity:,.2f}")
print(f" 가용: ${summary.available_balance:,.2f}")
print(f" 미실현 손익: ${summary.total_unrealized_pnl:,.2f}")
print(f" 포지션 수: {len(summary.positions)}")
def _detect_changes(self, current_positions: List) -> List[str]:
"""포지션 변경 감지"""
changes = []
current_dict = {f"{p.account_id}:{p.symbol}": p for p in current_positions}
previous_dict = {f"{p.account_id}:{p.symbol}": p for p in self.previous_positions}
# 신규 포지션
for key in current_dict:
if key not in previous_dict:
p = current_dict[key]
changes.append(f"신규: {p.symbol} {p.side} {p.size}")
# 폐쇄 포지션
for key in previous_dict:
if key not in current_dict:
p = previous_dict[key]
changes.append(f"폐쇄: {p.symbol} {p.side} {p.size}")
# 크기 변경
for key in current_dict:
if key in previous_dict:
curr = current_dict[key]
prev = previous_dict[key]
if abs(curr.size - prev.size) > 0.001:
changes.append(f"변경: {curr.symbol} {prev.size} → {curr.size}")
return changes
def _handle_alerts(self, alerts: List[Dict], positions: List, summaries: List):
"""알림 처리"""
for alert in alerts:
severity_emoji = {
'critical': '🚨',
'high': '⚠️',
'warning': '⚡',
'low': '📊'
}.get(alert['severity'], '❗')
message = f"{severity_emoji} [{alert['severity'].upper()}] {alert['message']}"
self.logger.warning(message)
print(message)
# 콜백 실행
if self.alert_callback:
self.alert_callback(alert)
# AI 권장사항 생성 (critical only)
if alert['severity'] == 'critical':
try:
recommendation = self.alert_system.generate_ai_recommendation(
[p.__dict__ for p in positions],
sum(s.total_equity for s in summaries)
)
print(f"\n[AI 권장사항]\n{recommendation}\n")
except Exception as e:
self.logger.error(f"AI 권장사항 생성 실패: {e}")
Telegram 알림 예시
def send_telegram_alert(alert: Dict):
"""텔레그램 알림 발송"""
import os
bot_token = os.getenv('TELEGRAM_BOT_TOKEN')
chat_id = os.getenv('TELEGRAM_CHAT_ID')
if not bot_token or not chat_id:
return
message = f"🚨 *Bybit 포지션 알림*\n\n"
message += f"유형: {alert['type']}\n"
message += f"심각도: {alert['severity']}\n"
message += f"메시지: {alert['message']}\n"
message += f"조치: {alert['action']}"
url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
requests.post(url, json={'chat_id': chat_id, 'text': message, 'parse_mode': 'Markdown'})
메인 실행
if __name__ == "__main__":
# HolySheep AI 클라이언트
ai_client = HolySheepAIClient(api_key="YOUR_HOL