#!/usr/bin/env python3 """ Binance P2P Liquidity Monitor Real-time P2P order book monitoring and liquidity event detection """ import asyncio import json from datetime import datetime, timedelta from pathlib import Path from typing import Dict, List, Optional import time from loguru import logger from src.api.binance_p2p_client import BinanceP2PClient from src.analyzers.advanced_transaction_analyzer import AdvancedTransactionAnalyzer from src.models.order import OrderBookSnapshot from config import Config, DEFAULT_CONFIG import os class LiquidityMonitor: """Main liquidity monitoring coordinator""" def __init__(self, config: Config): self.config = config # Initialize client with proxy settings from config self.client = BinanceP2PClient( proxy_url=config.PROXY_URL, proxy_type=config.PROXY_TYPE ) # Увеличиваем timeout D до 60 минут для 4-й стадии подтверждения self.analyzer = AdvancedTransactionAnalyzer(d_variable_timeout=max(config.D_VARIABLE_TIMEOUT, 3600)) self.is_running = False # Paths self.project_dir = Path(__file__).parent self.stats_path = self.project_dir / "liquidity_stats.json" async def start_monitoring(self): """Start the liquidity monitoring process""" logger.info("🚀 Starting Binance P2P Liquidity Monitor") logger.info(f"📊 Monitoring assets: {', '.join(self.config.ASSETS)}") logger.info(f"💱 Fiat currency: {self.config.FIAT}") logger.info(f"⏱️ Polling interval: {self.config.POLL_INTERVAL}s") self.is_running = True try: while self.is_running: await self._monitoring_cycle() except KeyboardInterrupt: logger.info("🛑 Monitoring stopped by user") except Exception as e: logger.error(f"❌ Fatal error in monitoring: {e}") raise finally: self.is_running = False async def _monitoring_cycle(self): """Single monitoring cycle for all assets and markets""" cycle_start = time.time() logger.info(f"🔄 Starting monitoring cycle at {datetime.now().strftime('%H:%M:%S')}") total_events = 0 total_snapshots = 0 # Process each fiat market fiat = self.config.FIAT logger.info(f"🌍 Processing {fiat} markets...") # Monitor both SELL and BUY orders for each asset for trade_type in self.config.TRADE_TYPES: logger.debug(f"📊 Processing {trade_type} orders...") # Get order books for all assets in this fiat and trade type order_books = await self.client.get_multi_asset_orders( assets=self.config.ASSETS, trade_type=trade_type, fiat=fiat ) # Analyze each order book for asset, order_book in order_books.items(): if order_book is None: logger.warning(f"⚠️ Failed to get {asset}/{fiat} {trade_type} order book") continue total_snapshots += 1 # Analyze changes events = self.analyzer.analyze_orderbook_changes(order_book) # Process events if events: await self._process_events(events, asset, fiat) total_events += len(events) # Log summary self._log_orderbook_summary(order_book) # Export analytics every cycle analytics = self.analyzer.export_stats_to_json() try: with open(self.stats_path, "w", encoding="utf-8") as f: json.dump(analytics, f, indent=2, ensure_ascii=False) except Exception as e: logger.error(f"Failed to write stats file {self.stats_path}: {e}") logger.info(f"📈 Analytics: {analytics['total_merchants']} merchants tracked, {analytics['active_d_variables']} active D-variables") cycle_time = time.time() - cycle_start logger.info(f"✅ Cycle completed in {cycle_time:.1f}s: {total_snapshots} snapshots, {total_events} events detected") # Wait for next cycle await asyncio.sleep(self.config.POLL_INTERVAL) async def _process_events(self, events: List, asset: str, fiat: str): """Process detected liquidity events""" for event in events: # Determine emoji based on event type emoji_map = { "real_trade": "💱", "top_up": "💰", "order_cancellation": "🚫", "VOLUME_DECREASE": "📉", "ORDER_CANCELLATION": "🚫", "MERCHANT_TOPUP": "💰", "NEW_MERCHANT": "🆕", "MERCHANT_DISAPPEARED": "👻" } # Обработка новых IndividualTransaction объектов if hasattr(event, 'transaction_type') and hasattr(event, 'amount'): emoji = emoji_map.get(event.transaction_type, "📊") amount_change = event.amount if event.transaction_type == "top_up" else -event.amount logger.info(f"{emoji} Event: {event.merchant_name} {amount_change:+.2f} {event.asset}") # Log detailed info for significant events if event.transaction_type in ["order_cancellation", "top_up", "real_trade"]: logger.info( f" 💱 {asset}/{fiat} | " f"Amount: {event.amount:.2f} | " f"Type: {event.transaction_type} | " f"Merchant: {event.merchant_name}" ) else: # Fallback для старых объектов emoji = emoji_map.get(str(getattr(event, 'transaction_type', 'UNKNOWN')), "📊") logger.info(f"{emoji} Event: {getattr(event, 'merchant_id', 'Unknown')} {getattr(event, 'amount_change', 0)} {getattr(event, 'asset', asset)}") def _log_orderbook_summary(self, order_book: OrderBookSnapshot): """Log order book summary""" logger.debug( f"📋 {order_book.asset}/{order_book.fiat} {order_book.trade_type}: " f"{len(order_book.orders)} orders, " f"{order_book.merchant_count} merchants, " f"total volume: {order_book.total_volume:.2f}, " f"best price: {order_book.best_price:.4f}" ) def stop_monitoring(self): """Stop the monitoring process""" self.is_running = False logger.info("🛑 Stopping liquidity monitor...") async def main(): """Main entry point""" # Setup logging project_dir = Path(__file__).parent logs_dir = project_dir / "logs" try: logs_dir.mkdir(parents=True, exist_ok=True) except Exception: pass logger.add( str(logs_dir / "liquidity_monitor.log"), rotation="1 day", retention="7 days", level="INFO", format="{time:YYYY-MM-DD HH:mm:ss} | {level} | {message}" ) # Load configuration # Priority: 1) defaults, 2) environment variables, 3) scanner_config.json config_path = project_dir / "scanner_config.json" config = DEFAULT_CONFIG # Check environment variables for proxy settings env_proxy_url = os.getenv("PROXY_URL") env_proxy_type = os.getenv("PROXY_TYPE", "http") try: if config_path.exists(): with open(config_path, "r", encoding="utf-8") as f: ui_cfg = json.load(f) # Map UI config keys to internal Config fiat = ui_cfg.get("fiat_currency") or DEFAULT_CONFIG.FIAT assets = ui_cfg.get("assets_to_track") or DEFAULT_CONFIG.ASSETS refresh = int(ui_cfg.get("refresh_interval") or DEFAULT_CONFIG.POLL_INTERVAL) # Extract proxy settings with priority: file config > env vars > defaults proxy_url = ui_cfg.get("proxy_url") or env_proxy_url or DEFAULT_CONFIG.PROXY_URL proxy_type = ui_cfg.get("proxy_type") or env_proxy_type or DEFAULT_CONFIG.PROXY_TYPE # Create a new Config with overrides config = Config( ASSETS=assets, FIAT=str(fiat).upper(), POLL_INTERVAL=refresh, D_VARIABLE_EXPIRY_MINUTES=DEFAULT_CONFIG.D_VARIABLE_EXPIRY_MINUTES, MIN_AMOUNT_CHANGE=DEFAULT_CONFIG.MIN_AMOUNT_CHANGE, LOG_LEVEL=DEFAULT_CONFIG.LOG_LEVEL, PROXY_URL=proxy_url, PROXY_TYPE=proxy_type ) proxy_info = f", proxy={proxy_type}://{proxy_url}" if proxy_url else "" logger.info( f"🛠️ Loaded scanner config: fiat={config.FIAT}, assets={','.join(config.ASSETS)}, interval={config.POLL_INTERVAL}s{proxy_info}" ) else: # No config file, but might have env vars if env_proxy_url: config = Config( ASSETS=DEFAULT_CONFIG.ASSETS, FIAT=DEFAULT_CONFIG.FIAT, POLL_INTERVAL=DEFAULT_CONFIG.POLL_INTERVAL, D_VARIABLE_EXPIRY_MINUTES=DEFAULT_CONFIG.D_VARIABLE_EXPIRY_MINUTES, MIN_AMOUNT_CHANGE=DEFAULT_CONFIG.MIN_AMOUNT_CHANGE, LOG_LEVEL=DEFAULT_CONFIG.LOG_LEVEL, PROXY_URL=env_proxy_url, PROXY_TYPE=env_proxy_type ) logger.info(f"ℹ️ scanner_config.json not found, using defaults with proxy from environment: {env_proxy_type}://{env_proxy_url}") else: logger.info("ℹ️ scanner_config.json not found, using default configuration") except Exception as e: logger.error(f"Failed to load scanner_config.json, using defaults: {e}") # Create and start monitor monitor = LiquidityMonitor(config) try: await monitor.start_monitoring() except KeyboardInterrupt: logger.info("👋 Graceful shutdown initiated") except Exception as e: logger.error(f"💥 Application crashed: {e}") raise finally: logger.info("🏁 Binance P2P Liquidity Monitor stopped") if __name__ == "__main__": asyncio.run(main())