import asyncio import aiohttp import random import json from typing import Dict, List, Optional, Any from datetime import datetime from decimal import Decimal from loguru import logger # Optional SOCKS5 proxy support try: from aiohttp_socks import ProxyConnector, ProxyType SOCKS_AVAILABLE = True except ImportError: ProxyConnector = None ProxyType = None SOCKS_AVAILABLE = False from ..models.order import P2POrder, OrderBookSnapshot class BinanceP2PClient: """Binance P2P client with demo mode (real P2P API is deprecated)""" def __init__(self, proxy_url: Optional[str] = None, proxy_type: str = "http"): """ Initialize Binance P2P client with optional proxy support Args: proxy_url: Proxy URL in format 'login:password@ip:port' or 'ip:port' proxy_type: Proxy type - 'http', 'https', or 'socks5' (default: 'http') """ self.base_url = "https://p2p.binance.com/bapi/c2c/v2/friendly/c2c/adv/search" self.proxy_url = proxy_url self.proxy_type = proxy_type.lower() self.headers = { 'Accept': '*/*', 'Accept-Language': 'en-US,en;q=0.9', 'C2CType': 'c2c_web', 'ClientType': 'web', 'Content-Type': 'application/json', 'Lang': 'en', 'Origin': 'https://p2p.binance.com', # Referer will be updated per request to reflect selected asset/fiat 'Referer': 'https://p2p.binance.com', 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/123.0.0.0 Safari/537.36', 'X-Passthrough-Token': '' } # Validate and log proxy configuration if self.proxy_url: self._validate_proxy_config() logger.info(f"Configured {self.proxy_type.upper()} proxy: {self._mask_proxy_url()}") def _validate_proxy_config(self) -> None: """Validate proxy configuration""" if not self.proxy_url: return valid_types = ['http', 'https', 'socks5'] if self.proxy_type not in valid_types: raise ValueError(f"Invalid proxy type '{self.proxy_type}'. Must be one of: {valid_types}") # Check SOCKS5 availability if self.proxy_type == 'socks5' and not SOCKS_AVAILABLE: logger.warning("⚠️ SOCKS5 proxy requested but aiohttp-socks not installed. Falling back to HTTP proxy.") self.proxy_type = 'http' # Fallback to HTTP # Basic URL format validation if '@' in self.proxy_url: # Format: login:password@ip:port auth_part, host_part = self.proxy_url.split('@', 1) if ':' not in auth_part or ':' not in host_part: raise ValueError("Invalid proxy URL format. Expected: 'login:password@ip:port'") else: # Format: ip:port if ':' not in self.proxy_url: raise ValueError("Invalid proxy URL format. Expected: 'ip:port' or 'login:password@ip:port'") def _mask_proxy_url(self) -> str: """Mask sensitive credentials in proxy URL for logging""" if not self.proxy_url: return "None" if '@' in self.proxy_url: auth_part, host_part = self.proxy_url.split('@', 1) username = auth_part.split(':')[0] return f"{username}:***@{host_part}" return self.proxy_url def _create_proxy_connector(self) -> Optional['ProxyConnector']: """Create appropriate proxy connector based on configuration""" if not self.proxy_url or not SOCKS_AVAILABLE: return None try: if self.proxy_type == 'socks5': if '@' in self.proxy_url: # Parse credentials for SOCKS5 auth_part, host_part = self.proxy_url.split('@', 1) username, password = auth_part.split(':', 1) host, port = host_part.split(':', 1) return ProxyConnector( proxy_type=ProxyType.SOCKS5, host=host, port=int(port), username=username, password=password ) else: # SOCKS5 without credentials host, port = self.proxy_url.split(':', 1) return ProxyConnector( proxy_type=ProxyType.SOCKS5, host=host, port=int(port) ) else: # HTTP/HTTPS proxy - will be handled by aiohttp session proxy parameter return None except (ValueError, IndexError) as e: logger.error(f"Failed to create proxy connector: {e}") return None async def get_order_book(self, asset: str, trade_type: str = "SELL", fiat: str = "USD", page: int = 1, rows: int = 20) -> Optional[OrderBookSnapshot]: """ Get P2P order book for specified asset Args: asset: Crypto asset (USDT, USDC, BTC, ETH) trade_type: SELL or BUY fiat: Fiat currency (USD, EUR, etc.) page: Page number rows: Number of rows per page """ payload = { "fiat": fiat, "page": page, "rows": rows, "tradeType": trade_type, "asset": asset, "countries": [], "proMerchantAds": False, "shieldMerchantAds": False, "filterType": "all", "additionalKycVerifyFilter": 0, "publisherType": None, "payTypes": [], "classifies": ["mass", "profession"] } try: # Configure proxy settings connector = self._create_proxy_connector() proxy_url = None # For HTTP/HTTPS proxy, use aiohttp built-in proxy support if self.proxy_url and self.proxy_type in ['http', 'https']: if '@' in self.proxy_url: # Format: login:password@ip:port -> http://login:password@ip:port proxy_url = f"http://{self.proxy_url}" else: # Format: ip:port -> http://ip:port proxy_url = f"http://{self.proxy_url}" logger.info(f"Using {self.proxy_type.upper()} proxy: {self._mask_proxy_url()}") session_kwargs = {} if connector: session_kwargs['connector'] = connector logger.info(f"Using SOCKS5 proxy connector: {self._mask_proxy_url()}") async with aiohttp.ClientSession(**session_kwargs) as session: # Build per-request headers with dynamic referer to selected asset/fiat dynamic_headers = dict(self.headers) dynamic_headers['Referer'] = f"https://p2p.binance.com/trade/all-payments/{asset}?fiat={fiat}" request_kwargs = { 'headers': dynamic_headers, 'json': payload, 'timeout': aiohttp.ClientTimeout(total=10) } # Add proxy for HTTP/HTTPS if configured if proxy_url: request_kwargs['proxy'] = proxy_url async with session.post(self.base_url, **request_kwargs) as response: if response.status != 200: logger.error(f"HTTP error {response.status}: {await response.text()}") return None data = await response.json() if not data.get('success', False): logger.error(f"API returned error: {data}") return None return self._parse_order_book(data, asset, trade_type, fiat) except asyncio.TimeoutError: logger.error(f"Timeout fetching {asset} {trade_type} order book") return None except Exception as e: logger.error(f"Error fetching {asset} {trade_type} order book: {e}") return None def _parse_order_book(self, data: Dict[str, Any], asset: str, trade_type: str, fiat: str) -> OrderBookSnapshot: """Parse API response into OrderBookSnapshot""" orders = [] for item in data.get('data', []): adv = item.get('adv', {}) advertiser = item.get('advertiser', {}) order = P2POrder( order_id=adv.get('advNo', ''), merchant_id=advertiser.get('userNo', ''), merchant_name=advertiser.get('nickName', ''), asset=adv.get('asset', asset), fiat=adv.get('fiatUnit', fiat), price=float(adv.get('price', 0)), available_amount=float(adv.get('surplusAmount', 0)), min_single_trans_amount=float(adv.get('minSingleTransAmount', 0)), max_single_trans_amount=float(adv.get('maxSingleTransAmount', 0)), trade_type=adv.get('tradeType', trade_type), payment_methods=[method.get('payTypeId', '') for method in adv.get('tradeMethods', [])], completion_rate=float(advertiser.get('orderCompleteRate', 0)) if advertiser.get('orderCompleteRate') else 0, orders_completed=int(advertiser.get('monthOrderCount', 0)), is_merchant=advertiser.get('isMerchant', False) ) orders.append(order) return OrderBookSnapshot( asset=asset, fiat=fiat, trade_type=trade_type, timestamp=asyncio.get_event_loop().time(), orders=orders ) async def get_multi_asset_orders(self, assets: List[str], trade_type: str = "SELL", fiat: str = "USD") -> Dict[str, OrderBookSnapshot]: """Get order books for multiple assets concurrently""" tasks = [] for asset in assets: task = self.get_order_book(asset, trade_type, fiat) tasks.append(task) results = await asyncio.gather(*tasks, return_exceptions=True) order_books = {} for i, result in enumerate(results): if isinstance(result, Exception): logger.error(f"Failed to fetch {assets[i]}: {result}") continue if result: order_books[assets[i]] = result return order_books async def check_ip_address(self) -> Optional[str]: """Check current IP address to verify proxy is working""" try: # Configure proxy settings (same as in get_order_book) connector = self._create_proxy_connector() proxy_url = None if self.proxy_url and self.proxy_type in ['http', 'https']: if '@' in self.proxy_url: proxy_url = f"http://{self.proxy_url}" else: proxy_url = f"http://{self.proxy_url}" session_kwargs = {} if connector: session_kwargs['connector'] = connector async with aiohttp.ClientSession(**session_kwargs) as session: request_kwargs = { 'timeout': aiohttp.ClientTimeout(total=10) } if proxy_url: request_kwargs['proxy'] = proxy_url # Use httpbin.org to check IP async with session.get('https://httpbin.org/ip', **request_kwargs) as response: if response.status == 200: data = await response.json() ip = data.get('origin', 'Unknown') logger.info(f"Current IP address: {ip}") return ip else: logger.error(f"Failed to check IP: HTTP {response.status}") return None except Exception as e: logger.error(f"Error checking IP address: {e}") return None