#!/usr/bin/env python3 """ Client The Graph pour récupérer les positions en temps réel """ import requests from typing import List, Dict from datetime import datetime class TheGraphClient: """Client pour interroger The Graph API (ApolloX BSC)""" def __init__(self, wallet_address: str): self.wallet_address = wallet_address.lower() self.thegraph_url = "https://gateway-arbitrum.network.thegraph.com/api/6edc89d15e0bffdb56057765735dfdc8/subgraphs/id/HxvztN7GmfYYDPb89K4FFYKFkRvBtscxYQqvX79D4XSQ" self.graphql_query = """ query OrderAndTradeHistorys($skip: Int, $first: Int, $user: String) { orderAndTradeHistories( skip: $skip first: $first orderBy: timestamp orderDirection: desc where: { user: $user actionType_not: ADD_MARGIN } ) { id orderOrTradeHash actionType timestamp txHash trade { pair { name } isLong token { symbol } margin entryPrice qty openFee executionFee closeType closePrice closeFee fundingFee holdingFee pnl actualPnl } } } """ def fetch_positions(self, limit: int = 1000) -> List[Dict]: """Récupère toutes les positions depuis The Graph""" all_histories = [] skip = 0 batch_size = min(limit, 1000) while True: variables = { "user": self.wallet_address, "skip": skip, "first": batch_size } try: response = requests.post( self.thegraph_url, json={ 'query': self.graphql_query, 'variables': variables }, headers={ 'Content-Type': 'application/json', 'Accept': '*/*', 'Origin': 'https://www.apollox.finance', 'Referer': 'https://www.apollox.finance/', 'User-Agent': 'Mozilla/5.0' }, timeout=10 ) response.raise_for_status() data = response.json() if 'errors' in data: print(f"⚠️ The Graph error: {data['errors']}") break histories = data.get('data', {}).get('orderAndTradeHistories', []) if not histories: break all_histories.extend(histories) if len(histories) < batch_size: break skip += batch_size except Exception as e: print(f"⚠️ Error fetching from The Graph: {e}") break return self._parse_positions(all_histories) def _parse_positions(self, histories: List[Dict]) -> List[Dict]: """Parse l'historique et crée les positions""" positions = {} for entry in histories: trade = entry.get('trade') if not trade: continue action_type = entry.get('actionType', 'UNKNOWN') timestamp = int(entry.get('timestamp', 0)) tx_hash = entry.get('txHash', '') order_trade_hash = entry.get('orderOrTradeHash', '') opened_at = datetime.fromtimestamp(timestamp).isoformat() if timestamp else None # Données du trade pair_name = trade.get('pair', {}).get('name', 'BTCUSD') is_long = trade.get('isLong', False) token_symbol = trade.get('token', {}).get('symbol', 'USDT') # Valeurs numériques margin = float(trade.get('margin', 0)) entry_price = float(trade.get('entryPrice', 0)) qty = float(trade.get('qty', 0)) close_type = trade.get('closeType') close_price = float(trade.get('closePrice', 0)) if trade.get('closePrice') else None pnl = float(trade.get('actualPnl', 0)) if trade.get('actualPnl') else None # Fees open_fee = float(trade.get('openFee', 0)) close_fee = float(trade.get('closeFee', 0)) funding_fee = float(trade.get('fundingFee', 0)) holding_fee = float(trade.get('holdingFee', 0)) execution_fee = float(trade.get('executionFee', 0)) # Leverage leverage = int((qty * entry_price) / margin) if margin > 0 and entry_price > 0 else 0 # Status if close_type: if 'LIQUIDAT' in close_type.upper(): status = 'LIQUIDATED' else: status = 'CLOSED' closed_at = opened_at else: status = 'OPEN' closed_at = None # Total fees total_fees = open_fee + close_fee + abs(funding_fee) + holding_fee + execution_fee # PnL net (après fees) pnl_net = (pnl - total_fees) if pnl is not None else None # PnL% pnl_pct = (pnl / margin) * 100 if pnl is not None and margin > 0 else None pnl_net_pct = (pnl_net / margin) * 100 if pnl_net is not None and margin > 0 else None # Trade hash unique trade_hash = order_trade_hash if order_trade_hash else tx_hash # Créer ou mettre à jour position position = { 'trade_hash': trade_hash, 'tx_hash': tx_hash, 'symbol': f"{pair_name}_DEGEN", 'pair_address': '0xD89aE81e55b1Bd46591CB5555209fdF8fFb43b76', 'is_long': is_long, 'margin': margin, 'leverage': leverage, 'entry_price': entry_price, 'qty': qty, 'stop_loss': 0, 'take_profit': 0, 'opened_at': opened_at, 'status': status, 'closed_at': closed_at, 'close_tx_hash': tx_hash if status != 'OPEN' else None, 'exit_price': close_price, 'pnl': pnl, 'pnl_pct': pnl_pct, 'pnl_net': pnl_net, 'pnl_net_pct': pnl_net_pct, 'total_fees': total_fees, 'close_reason': close_type, 'open_fee': open_fee, 'close_fee': close_fee, 'funding_fee': funding_fee, 'holding_fee': holding_fee, 'execution_fee': execution_fee, 'action_type': action_type, 'token_symbol': token_symbol, } positions[trade_hash] = position return list(positions.values()) def get_stats(self, positions: List[Dict]) -> Dict: """Calcule les stats depuis les positions""" if not positions: return { 'total_positions': 0, 'open_positions': 0, 'closed_positions': 0, 'liquidated_positions': 0, 'total_pnl': 0, 'total_fees': 0, 'net_pnl': 0, 'win_rate': 0, 'avg_leverage': 0, } open_pos = [p for p in positions if p['status'] == 'OPEN'] closed_pos = [p for p in positions if p['status'] == 'CLOSED'] liquidated = [p for p in positions if p['status'] == 'LIQUIDATED'] # PnL total_pnl = sum(p.get('pnl', 0) or 0 for p in positions if p.get('pnl')) # Fees total_fees = sum( (p.get('open_fee', 0) or 0) + (p.get('close_fee', 0) or 0) + abs(p.get('funding_fee', 0) or 0) + (p.get('holding_fee', 0) or 0) + (p.get('execution_fee', 0) or 0) for p in positions ) # Net PnL net_pnl = total_pnl - total_fees # Win rate winning = [p for p in closed_pos if (p.get('pnl') or 0) > 0] win_rate = (len(winning) / len(closed_pos) * 100) if closed_pos else 0 # Leverage moyen leverages = [p['leverage'] for p in positions if p.get('leverage', 0) > 0] avg_leverage = sum(leverages) / len(leverages) if leverages else 0 return { 'total_positions': len(positions), 'open_positions': len(open_pos), 'closed_positions': len(closed_pos), 'liquidated_positions': len(liquidated), 'total_pnl': total_pnl, 'total_fees': total_fees, 'net_pnl': net_pnl, 'win_rate': win_rate, 'avg_leverage': avg_leverage, }