#!/usr/bin/env python3 """ Position Monitor - Sources de données fiables Vérifie l'état réel des positions via 2 sources: 1. The Graph (Arbitrum) - Source principale 2. Etherscan V2 (BSC) - Fallback """ import requests import json import os from datetime import datetime from typing import Dict, List, Optional from dotenv import load_dotenv from web3 import Web3 import logging load_dotenv() logger = logging.getLogger(__name__) # Configuration WALLET_ADDRESS = os.getenv('WALLET_ADDRESS', '0x7476737C49d55bea691e54c2c96fD39a6bAf3049').lower() BSCSCAN_API_KEY = os.getenv('BSCSCAN_API_KEY') # The Graph sur Arbitrum (source principale) THEGRAPH_URL = "https://gateway-arbitrum.network.thegraph.com/api/6edc89d15e0bffdb56057765735dfdc8/subgraphs/id/HxvztN7GmfYYDPb89K4FFYKFkRvBtscxYQqvX79D4XSQ" # Etherscan V2 (fallback BSC) ETHERSCAN_V2_URL = "https://api.etherscan.io/v2/api" APOLLOX_CONTRACT_BSC = "0x1b6F2d3844C6ae7D56ceb3C3643b9060ba28FEb0" # BSC RPC BSC_RPC = "https://bsc-dataseed.binance.org/" w3 = Web3(Web3.HTTPProvider(BSC_RPC)) class PositionMonitor: """ Monitor l'état réel des positions on-chain """ def __init__(self): self.wallet = WALLET_ADDRESS # ==================== THE GRAPH (ARBITRUM) ==================== def fetch_from_thegraph(self, limit: int = 100) -> List[Dict]: """ Récupère les positions depuis The Graph (Arbitrum) Source principale la plus fiable """ 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 limitOrder { pair { name } isLong token { symbol } amountIn limitPrice qty takeProfit stopLoss } trade { pair { name } isLong token { symbol } margin entryPrice qty openFee executionFee closeType closePrice closeFee fundingFee holdingFee pnl actualPnl } } } """ variables = { "user": self.wallet, "skip": 0, "first": limit } try: logger.info("📡 Interrogation The Graph (Arbitrum)...") response = requests.post( THEGRAPH_URL, json={'query': graphql_query, 'variables': variables}, headers={ 'Content-Type': 'application/json', 'Accept': '*/*', 'Origin': 'https://www.apollox.finance', 'User-Agent': 'Mozilla/5.0' }, timeout=30 ) response.raise_for_status() data = response.json() if 'errors' in data: logger.error(f"GraphQL errors: {data['errors']}") return [] histories = data.get('data', {}).get('orderAndTradeHistories', []) logger.info(f"✅ {len(histories)} entrées récupérées depuis The Graph") return histories except Exception as e: logger.error(f"❌ Erreur The Graph: {e}") return [] def parse_thegraph_positions(self, histories: List[Dict]) -> Dict[str, Dict]: """ Parse les données The Graph en positions standardisées """ positions = {} for entry in histories: trade = entry.get('trade') limit_order = entry.get('limitOrder') if not trade and not limit_order: continue action_type = entry.get('actionType', 'UNKNOWN') timestamp = int(entry.get('timestamp', 0)) tx_hash = entry.get('txHash', '') order_trade_hash = entry.get('orderOrTradeHash', '') # Trade hash unique trade_hash = order_trade_hash if order_trade_hash else tx_hash # Données du trade ou limit order data = trade if trade else limit_order pair_name = data.get('pair', {}).get('name', 'BTCUSD') is_long = data.get('isLong', False) # Parse trade exécuté if trade: 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 # 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' else: status = 'OPEN' # Fees open_fee = float(trade.get('openFee', 0)) close_fee = float(trade.get('closeFee', 0)) funding_fee = float(trade.get('fundingFee', 0)) else: # Limit order margin = float(limit_order.get('amountIn', 0)) entry_price = float(limit_order.get('limitPrice', 0)) qty = float(limit_order.get('qty', 0)) leverage = 0 status = 'PENDING' if action_type == 'OPEN_LIMIT' else 'CANCELLED' close_price = None pnl = None open_fee = 0 close_fee = 0 funding_fee = 0 # Créer position standardisée position = { 'trade_hash': trade_hash, 'tx_hash': tx_hash, 'symbol': f"{pair_name}_DEGEN", 'is_long': is_long, 'margin': margin, 'leverage': leverage, 'entry_price': entry_price, 'qty': qty, 'status': status, 'opened_at': datetime.fromtimestamp(timestamp).isoformat() if timestamp else None, 'close_price': close_price, 'pnl': pnl, 'open_fee': open_fee, 'close_fee': close_fee, 'funding_fee': funding_fee, 'action_type': action_type, 'source': 'thegraph_arbitrum', 'last_checked': datetime.now().isoformat() } positions[trade_hash] = position return positions # ==================== ETHERSCAN V2 (BSC FALLBACK) ==================== def fetch_from_etherscan(self, startblock: int = 0, limit: int = 100) -> List[Dict]: """ Récupère les TX depuis Etherscan V2 (BSC) Fallback si The Graph échoue """ if not BSCSCAN_API_KEY: logger.warning("⚠️ BSCSCAN_API_KEY manquante") return [] params = { 'chainid': '56', # BSC 'module': 'account', 'action': 'txlist', 'address': self.wallet, 'startblock': startblock, 'endblock': 99999999, 'page': 1, 'offset': limit, 'sort': 'desc', 'apikey': BSCSCAN_API_KEY } try: logger.info("📡 Interrogation Etherscan V2 (BSC)...") response = requests.get(ETHERSCAN_V2_URL, params=params, timeout=30) data = response.json() if data.get('status') == '1' and isinstance(data.get('result'), list): txs = data['result'] # Filtrer TX ApolloX apollox_txs = [ tx for tx in txs if tx.get('to', '').lower() == APOLLOX_CONTRACT_BSC.lower() ] logger.info(f"✅ {len(apollox_txs)} TX ApolloX récupérées (sur {len(txs)} total)") return apollox_txs else: logger.warning(f"⚠️ Etherscan V2: {data.get('message')} - {data.get('result')}") return [] except Exception as e: logger.error(f"❌ Erreur Etherscan V2: {e}") return [] def extract_trade_hash_from_tx(self, tx_hash: str) -> Optional[str]: """ Extrait le trade hash depuis les logs d'une TX """ try: receipt = w3.eth.get_transaction_receipt(tx_hash) # Chercher dans les logs for log in receipt['logs']: if log['address'].lower() == APOLLOX_CONTRACT_BSC.lower(): if len(log['topics']) >= 2: # Trade hash souvent dans topics[1] trade_hash = log['topics'][1].hex() return trade_hash return None except Exception as e: logger.debug(f"Impossible d'extraire trade hash de {tx_hash}: {e}") return None # ==================== API PUBLIQUE ==================== def get_active_positions(self) -> Dict[str, Dict]: """ Récupère les positions ACTIVES (OPEN) """ # 1. Essayer The Graph d'abord histories = self.fetch_from_thegraph(limit=50) if histories: all_positions = self.parse_thegraph_positions(histories) active_positions = { th: pos for th, pos in all_positions.items() if pos['status'] == 'OPEN' } logger.info(f"📊 {len(active_positions)} positions actives trouvées (The Graph)") return active_positions # 2. Fallback Etherscan V2 logger.warning("⚠️ Fallback sur Etherscan V2...") txs = self.fetch_from_etherscan(limit=50) if txs: # Filtrer les openMarketTrade récentes open_txs = [ tx for tx in txs if 'openMarketTrade' in tx.get('functionName', '').lower() ] logger.info(f"📊 {len(open_txs)} ouvertures récentes trouvées (Etherscan)") # Note: Etherscan ne donne pas l'état (OPEN/CLOSED) # Il faudrait interroger le contrat pour chaque position # Pour l'instant on retourne vide et log warning logger.warning("⚠️ Etherscan ne peut pas déterminer si positions sont ouvertes") return {} logger.error("❌ Aucune source disponible") return {} def get_all_positions(self, limit: int = 100) -> Dict[str, Dict]: """ Récupère TOUTES les positions (historique complet) """ histories = self.fetch_from_thegraph(limit=limit) if histories: all_positions = self.parse_thegraph_positions(histories) logger.info(f"📊 {len(all_positions)} positions totales trouvées") return all_positions logger.error("❌ Impossible de récupérer l'historique") return {} def check_position_status(self, trade_hash: str) -> Optional[str]: """ Vérifie le statut d'une position spécifique Returns: 'OPEN', 'CLOSED', 'LIQUIDATED', 'PENDING', ou None si non trouvée """ all_positions = self.get_all_positions(limit=200) if trade_hash in all_positions: status = all_positions[trade_hash]['status'] logger.info(f"Position {trade_hash[:16]}... = {status}") return status logger.warning(f"⚠️ Position {trade_hash[:16]}... non trouvée") return None if __name__ == '__main__': # Test du monitor logging.basicConfig(level=logging.INFO, format='%(message)s') monitor = PositionMonitor() print("=" * 80) print("🔍 TEST POSITION MONITOR") print("=" * 80) print() # Test 1: Positions actives print("1️⃣ POSITIONS ACTIVES:") print("-" * 80) active = monitor.get_active_positions() if active: for trade_hash, pos in active.items(): side = "LONG" if pos['is_long'] else "SHORT" print(f"\n📊 {pos['symbol']} {side} {pos['leverage']}x") print(f" Trade Hash: {trade_hash[:20]}...") print(f" Margin: ${pos['margin']:.2f}") print(f" Entry: ${pos['entry_price']:,.2f}") print(f" Status: {pos['status']}") print(f" Opened: {pos['opened_at']}") else: print(" ⚠️ Aucune position active") print() print("=" * 80)