#!/usr/bin/env python3 """ Position Sync - Synchronisation positions.json avec sources on-chain Réconcilie les positions locales avec les données réelles (The Graph + Etherscan) """ import json import os from datetime import datetime from typing import Dict, List import logging from position_monitor import PositionMonitor logger = logging.getLogger(__name__) POSITIONS_FILE = 'positions.json' class PositionSync: """ Synchronise positions.json avec les sources on-chain """ def __init__(self, positions_file: str = POSITIONS_FILE): self.positions_file = positions_file self.monitor = PositionMonitor() def load_local_positions(self) -> Dict: """ Charge positions.json """ if not os.path.exists(self.positions_file): logger.warning(f"⚠️ {self.positions_file} n'existe pas, création...") return { 'wallet': self.monitor.wallet, 'last_updated': datetime.now().isoformat(), 'positions': {} } try: with open(self.positions_file, 'r') as f: data = json.load(f) positions = data.get('positions', {}) logger.info(f"📂 {len(positions)} positions locales chargées") return data except Exception as e: logger.error(f"❌ Erreur chargement {self.positions_file}: {e}") return {'wallet': self.monitor.wallet, 'positions': {}} def save_local_positions(self, data: Dict): """ Sauvegarde positions.json """ data['last_updated'] = datetime.now().isoformat() try: with open(self.positions_file, 'w') as f: json.dump(data, f, indent=2) logger.info(f"💾 {len(data.get('positions', {}))} positions sauvegardées") except Exception as e: logger.error(f"❌ Erreur sauvegarde {self.positions_file}: {e}") def sync_positions(self) -> Dict: """ Synchronise positions locales avec sources on-chain Process: 1. Charge positions.json locales 2. Récupère état réel depuis The Graph (Arbitrum) 3. Met à jour statut des positions locales 4. Ajoute nouvelles positions on-chain manquantes 5. Sauvegarde """ logger.info("=" * 80) logger.info("🔄 SYNCHRONISATION POSITIONS") logger.info("=" * 80) # 1. Charger local local_data = self.load_local_positions() local_positions = local_data.get('positions', {}) logger.info(f"📂 Positions locales: {len(local_positions)}") # 2. Récupérer on-chain onchain_positions = self.monitor.get_all_positions(limit=200) logger.info(f"🔗 Positions on-chain: {len(onchain_positions)}") # 3. Sync updated_count = 0 added_count = 0 local_open_count = 0 # Mettre à jour positions locales avec état on-chain for trade_hash, local_pos in list(local_positions.items()): local_status = local_pos.get('status') # DEBUG: Log toutes les positions locales OPEN if local_status == 'OPEN': in_onchain = trade_hash in onchain_positions logger.info(f"🔍 DEBUG: Position locale OPEN: {trade_hash[:16]} | Dans onchain: {in_onchain}") if trade_hash in onchain_positions: onchain_pos = onchain_positions[trade_hash] onchain_status = onchain_pos['status'] # Vérifier si status a changé if local_status != onchain_status: logger.info(f"🔄 Position {trade_hash[:16]}...") logger.info(f" Status: {local_status} → {onchain_status}") # Mettre à jour avec données on-chain local_positions[trade_hash].update({ 'status': onchain_status, 'close_price': onchain_pos.get('close_price'), 'pnl': onchain_pos.get('pnl'), 'closed_at': datetime.now().isoformat() if onchain_status != 'OPEN' else None, 'synced_at': datetime.now().isoformat() }) updated_count += 1 else: # Position locale n'existe plus on-chain → fermée if local_status == 'OPEN': logger.info(f"🔄 Position {trade_hash[:16]}... (PAS dans onchain)") logger.info(f" Status: OPEN → CLOSED (n'existe plus on-chain)") local_positions[trade_hash].update({ 'status': 'CLOSED', 'closed_at': datetime.now().isoformat(), 'synced_at': datetime.now().isoformat() }) updated_count += 1 # Recompter positions actives après mises à jour local_open_count = sum(1 for p in local_positions.values() if p.get('status') == 'OPEN') # Ajouter positions on-chain manquantes localement for trade_hash, onchain_pos in onchain_positions.items(): if trade_hash not in local_positions: logger.info(f"➕ Nouvelle position découverte: {trade_hash[:16]}...") logger.info(f" {onchain_pos['symbol']} {'LONG' if onchain_pos['is_long'] else 'SHORT'} | Status: {onchain_pos['status']}") local_positions[trade_hash] = onchain_pos added_count += 1 # 4. Sauvegarder local_data['positions'] = local_positions self.save_local_positions(local_data) # 5. Résumé logger.info("=" * 80) logger.info("📊 RÉSUMÉ SYNC") logger.info("=" * 80) logger.info(f"Positions mises à jour: {updated_count}") logger.info(f"Positions ajoutées: {added_count}") logger.info(f"Positions actives locales: {local_open_count}") # Positions actives on-chain onchain_open = [p for p in onchain_positions.values() if p['status'] == 'OPEN'] logger.info(f"✅ Positions actives on-chain: {len(onchain_open)}") if onchain_open: for pos in onchain_open: side = "LONG" if pos['is_long'] else "SHORT" logger.info(f" {pos['symbol']} {side} {pos['leverage']}x @ ${pos['entry_price']:,.2f}") logger.info("=" * 80) return local_data def verify_active_positions(self) -> List[Dict]: """ Vérifie uniquement les positions actives et retourne celles réellement OPEN """ local_data = self.load_local_positions() local_positions = local_data.get('positions', {}) # Positions marquées OPEN localement local_open = { th: pos for th, pos in local_positions.items() if pos.get('status') == 'OPEN' } if not local_open: logger.info("📊 Aucune position active locale") return [] logger.info(f"🔍 Vérification de {len(local_open)} positions actives...") # Vérifier état réel on-chain verified_open = [] for trade_hash, local_pos in local_open.items(): status = self.monitor.check_position_status(trade_hash) if status == 'OPEN': verified_open.append(local_pos) logger.info(f" ✅ {trade_hash[:16]}... toujours OPEN") elif status: logger.warning(f" ⚠️ {trade_hash[:16]}... fermée (status: {status})") # Mettre à jour local local_positions[trade_hash]['status'] = status local_positions[trade_hash]['synced_at'] = datetime.now().isoformat() else: logger.warning(f" ❓ {trade_hash[:16]}... non trouvée on-chain") # Sauvegarder mises à jour if len(verified_open) != len(local_open): local_data['positions'] = local_positions self.save_local_positions(local_data) logger.info(f"✅ {len(verified_open)} positions réellement actives") return verified_open if __name__ == '__main__': # Test du sync logging.basicConfig(level=logging.INFO, format='%(message)s') sync = PositionSync() print() print("🔄 Test 1: Synchronisation complète") print() sync.sync_positions() print() print("🔍 Test 2: Vérification positions actives") print() active = sync.verify_active_positions() if active: print(f"\n✅ {len(active)} position(s) active(s) vérifiée(s)") for pos in active: side = "LONG" if pos.get('is_long') else "SHORT" print(f" {pos.get('symbol')} {side} @ ${pos.get('entry_price', 0):,.2f}") else: print("\n✅ Aucune position active") print()