#!/usr/bin/env python3 """ Import positions depuis The Graph API (ApolloX subgraph) Récupère les VRAIES données: margin, leverage, PnL, etc. """ import requests import json from datetime import datetime from typing import List, Dict import os from dotenv import load_dotenv load_dotenv() WALLET_ADDRESS = os.getenv('WALLET_ADDRESS', '0x7476737C49d55bea691e54c2c96fD39a6bAf3049').lower() THEGRAPH_URL = "https://api.thegraph.com/subgraphs/name/apollx-apx/apollox-perp-bnb" def fetch_position_history() -> List[Dict]: """Récupère l'historique depuis The Graph""" query = """ query OrderAndTradeHistorys { orderAndTradeHistories( skip: 0 first: 1000 orderBy: timestamp orderDirection: desc where: { user: "%s" } ) { id orderOrTradeHash actionType timestamp marginIncrement txHash trade { id pair { name } isLong token { symbol } margin entryPrice qty openFee executionFee closeType closePrice closeFee fundingFee holdingFee pnl actualPnl } user { id } } } """ % WALLET_ADDRESS print(f"🔍 Récupération historique depuis The Graph...") print(f" Wallet: {WALLET_ADDRESS}\n") try: response = requests.post( THEGRAPH_URL, json={'query': query}, timeout=30 ) response.raise_for_status() data = response.json() if 'errors' in data: print(f"❌ Erreur GraphQL: {data['errors']}") return [] histories = data.get('data', {}).get('orderAndTradeHistories', []) print(f"✅ {len(histories)} entrées récupérées\n") return histories except Exception as e: print(f"❌ Erreur requête: {e}") return [] def parse_positions(histories: List[Dict]) -> Dict: """Parse l'historique et crée les positions""" positions = {} print("📊 Parsing des positions...\n") for entry in histories: trade = entry.get('trade') if not trade: continue trade_id = trade.get('id') if not trade_id: continue action_type = entry.get('actionType', 'UNKNOWN') timestamp = int(entry.get('timestamp', 0)) tx_hash = entry.get('txHash', '') # Convertir timestamp en ISO opened_at = datetime.fromtimestamp(timestamp).isoformat() if timestamp else None # Extraire données 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 (convertir depuis Wei/decimals) margin = float(trade.get('margin', 0)) / 1e18 entry_price = float(trade.get('entryPrice', 0)) / 1e8 qty = float(trade.get('qty', 0)) / 1e10 close_type = trade.get('closeType') close_price = float(trade.get('closePrice', 0)) / 1e8 if trade.get('closePrice') else None # PnL (déjà en USDT) pnl = float(trade.get('actualPnl', 0)) / 1e18 if trade.get('actualPnl') else None # Fees open_fee = float(trade.get('openFee', 0)) / 1e18 close_fee = float(trade.get('closeFee', 0)) / 1e18 funding_fee = float(trade.get('fundingFee', 0)) / 1e18 holding_fee = float(trade.get('holdingFee', 0)) / 1e18 # Calculer leverage leverage = int((qty * entry_price) / margin) if margin > 0 and entry_price > 0 else 0 # Déterminer 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 # Calculer PnL% si fermé pnl_pct = None if pnl is not None and margin > 0: pnl_pct = (pnl / margin) * 100 # Créer position position = { 'trade_hash': trade_id, 'tx_hash': tx_hash, 'symbol': f"{pair_name}_DEGEN", 'pair_address': '0xD89aE81e55b1Bd46591CB5555209fdF8fFb43b76', # BTC pair '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, 'close_reason': close_type, 'open_fee': open_fee, 'close_fee': close_fee, 'funding_fee': funding_fee, 'holding_fee': holding_fee, 'action_type': action_type } # Afficher side = "LONG" if is_long else "SHORT" print(f" {pair_name} {side} {leverage}x - {status}") print(f" Margin: ${margin:.2f} | Entry: ${entry_price:,.2f}") if pnl is not None: print(f" PnL: ${pnl:+.2f} ({pnl_pct:+.2f}%)") print() positions[trade_id] = position return positions def save_positions(positions: Dict): """Sauvegarde dans positions.json""" data = { 'positions': positions, 'last_updated': datetime.now().isoformat(), 'source': 'thegraph_apollox_subgraph' } with open('positions.json', 'w') as f: json.dump(data, f, indent=2) print(f"✅ {len(positions)} positions sauvegardées dans positions.json") def main(): """Main""" print("=" * 80) print("📥 IMPORT POSITIONS DEPUIS THE GRAPH") print("=" * 80) print() # 1. Fetch depuis The Graph histories = fetch_position_history() if not histories: print("❌ Aucune donnée récupérée") return # 2. Parse positions = parse_positions(histories) if not positions: print("❌ Aucune position parsée") return # 3. Save save_positions(positions) # 4. Stats print() print("=" * 80) print("📊 RÉSUMÉ") print("=" * 80) open_positions = [p for p in positions.values() if p['status'] == 'OPEN'] closed_positions = [p for p in positions.values() if p['status'] == 'CLOSED'] liquidated = [p for p in positions.values() if p['status'] == 'LIQUIDATED'] print(f"Total positions: {len(positions)}") print(f" OPEN: {len(open_positions)}") print(f" CLOSED: {len(closed_positions)}") print(f" LIQUIDATED: {len(liquidated)}") # Total PnL total_pnl = sum(p.get('pnl', 0) or 0 for p in positions.values() if p.get('pnl')) print(f"\n💰 Total PnL: ${total_pnl:+.2f}") # Total fees total_fees = sum( (p.get('open_fee', 0) or 0) + (p.get('close_fee', 0) or 0) + (p.get('funding_fee', 0) or 0) + (p.get('holding_fee', 0) or 0) for p in positions.values() ) print(f"💸 Total Fees: ${total_fees:.2f}") print() print("✅ Dashboard: http://localhost:8889") if __name__ == '__main__': main()