# Table of Contents - [API Reference - GmGnAPI Documentation](#api-reference-gmgnapi-documentation) - [GmGnAPI - Professional Python Client for GMGN.ai](#gmgnapi-professional-python-client-for-gmgn-ai) - [Getting Started - GmGnAPI Documentation](#getting-started-gmgnapi-documentation) - [Examples - GmGnAPI Documentation](#examples-gmgnapi-documentation) - [Advanced Usage - GmGnAPI Documentation](#advanced-usage-gmgnapi-documentation) --- # API Reference - GmGnAPI Documentation GmGnClient ---------- Basic WebSocket client for connecting to GMGN.ai's real-time data streams. Provides core functionality with automatic reconnection and event handling. class GmGnClient( websocket_url: str = "wss://gmgn.ai/ws", access_token: Optional[str] = None, max_reconnect_attempts: int = 10, reconnect_delay: float = 1.0, heartbeat_interval: float = 30.0 ) ### Constructor Parameters | Parameter | Type | Default | Description | | --- | --- | --- | --- | | `websocket_url` | `str` | `"wss://gmgn.ai/ws"` | WebSocket endpoint URL | | `access_token` | `Optional[str]` | `None` | Authentication token for protected channels | | `max_reconnect_attempts` | `int` | `10` | Maximum number of reconnection attempts | | `reconnect_delay` | `float` | `1.0` | Initial delay between reconnection attempts (seconds) | | `heartbeat_interval` | `float` | `30.0` | Heartbeat ping interval (seconds) | ### Methods #### `async subscribe_new_pools(chain: str = "sol") -> None` Subscribe to real-time new liquidity pool notifications. await client.subscribe_new_pools(chain="sol") #### `async subscribe_token_updates(tokens: List[str], chain: str = "sol") -> None` Subscribe to price and volume updates for specific tokens. tokens = ["EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v"] await client.subscribe_token_updates(tokens, chain="sol") #### `async subscribe_wallet_trades(wallets: List[str], chain: str = "sol") -> None` Subscribe to trading activity from specific wallet addresses. wallets = ["wallet_address_here"] await client.subscribe_wallet_trades(wallets, chain="sol") #### `async listen() -> None` Start listening for WebSocket messages and dispatch events. await client.listen() # Runs until connection closes #### `async connect() -> None` Establish WebSocket connection manually. #### `async disconnect() -> None` Close WebSocket connection gracefully. ### Event Decorators #### `@client.on_new_pool` Decorator for handling new pool notifications. @client.on_new_pool async def handle_new_pool(pool_data: NewPoolInfo): print(f"New pool: {pool_data.chain}") #### `@client.on_token_update` Decorator for handling token price/volume updates. @client.on_token_update async def handle_update(update: PairUpdateData): print(f"Price: ${update.price_usd}") #### `@client.on_wallet_trade` Decorator for handling wallet trading activity. #### `@client.on_error` Decorator for handling connection and data errors. #### `@client.on_connect` Decorator for handling successful connections. #### `@client.on_disconnect` Decorator for handling disconnections. GmGnEnhancedClient ------------------ Advanced WebSocket client with filtering, monitoring, data export, and alerting capabilities. Extends GmGnClient with enterprise-grade features for production use. class GmGnEnhancedClient(GmGnClient): def __init__( self, token_filter: Optional[TokenFilter] = None, export_config: Optional[DataExportConfig] = None, alert_config: Optional[AlertConfig] = None, enable_statistics: bool = True, **kwargs ) ### Additional Parameters | Parameter | Type | Description | | --- | --- | --- | | `token_filter` | `Optional[TokenFilter]` | Filter configuration for token screening | | `export_config` | `Optional[DataExportConfig]` | Data export settings (JSON, CSV, database) | | `alert_config` | `Optional[AlertConfig]` | Alert and notification configuration | | `enable_statistics` | `bool` | Enable real-time statistics collection | ### Enhanced Methods #### `get_statistics() -> MonitoringStats` Get current monitoring statistics and metrics. stats = client.get_statistics() print(f"Messages: {stats.total_messages}") print(f"Uptime: {stats.connection_uptime}s") #### `async export_data(format: str = "json", file_path: Optional[str] = None) -> str` Export collected data to specified format. file_path = await client.export_data("csv", "data.csv") print(f"Data exported to: {file_path}") #### `update_filter(new_filter: TokenFilter) -> None` Update token filtering criteria dynamically. #### `add_alert_condition(condition: Dict[str, Any]) -> None` Add new alert condition at runtime. ### Enhanced Events #### `@client.on_filtered_pool` Triggered when a pool passes all filter criteria. #### `@client.on_alert_triggered` Triggered when alert conditions are met. #### `@client.on_statistics_update` Periodic statistics updates (every 60 seconds). Data Models ----------- Pydantic data models for type-safe data handling and validation. #### NewPoolInfo Information about new liquidity pools * `c: str` - Chain identifier * `p: List[PoolData]` - Pool data list * `rg: Optional[str]` - Region #### PoolData Individual pool information * `a: str` - Pool address * `ex: str` - Exchange name * `ba: str` - Base token address * `qa: str` - Quote token address * `bti: Optional[TokenInfo]` - Base token info #### TokenInfo Detailed token information * `s: Optional[str]` - Symbol * `n: Optional[str]` - Name * `mc: Optional[int]` - Market cap * `v24h: Optional[float]` - 24h volume * `p: Optional[float]` - Price #### PairUpdateData Trading pair price/volume updates * `pair_address: str` * `token_address: str` * `price_usd: Optional[Decimal]` * `volume_24h_usd: Optional[Decimal]` * `chain: str` #### WalletTradeData Wallet trading activity * `wallet_address: str` * `chain: str` * `trades: List[TradeData]` * `total_volume_24h_usd: Optional[Decimal]` #### TradeData Individual trade information * `transaction_hash: str` * `trade_type: Literal["buy", "sell"]` * `amount_usd: Decimal` * `timestamp: datetime` Configuration Models -------------------- #### TokenFilter Configure token filtering criteria from decimal import Decimal from gmgnapi import TokenFilter filter_config = TokenFilter( min_market_cap=Decimal("100000"), # $100K minimum max_market_cap=Decimal("10000000"), # $10M maximum min_liquidity=Decimal("50000"), # $50K minimum liquidity min_volume_24h=Decimal("10000"), # $10K 24h volume exchanges=["raydium", "orca"], # Specific exchanges max_risk_score=0.3 # Low risk only ) #### DataExportConfig Configure data export settings from gmgnapi import DataExportConfig export_config = DataExportConfig( enabled=True, format="json", # "json", "csv", "parquet" file_path="./data/pools.json", max_file_size_mb=100, rotation_interval_hours=24, compress=True, include_metadata=True ) #### AlertConfig Configure alerts and notifications from gmgnapi import AlertConfig alert_config = AlertConfig( enabled=True, webhook_url="https://hooks.slack.com/...", email="alerts@example.com", conditions=[\ {\ "field": "market_cap",\ "operator": ">",\ "value": 1000000\ },\ {\ "field": "volume_24h",\ "operator": ">",\ "value": 500000\ }\ ], rate_limit_seconds=60 ) Exceptions ---------- #### GmGnAPIError Base exception class for all GmGnAPI errors #### ConnectionError Raised when WebSocket connection fails #### AuthenticationError Raised when authentication fails for protected channels #### ValidationError Raised when received data fails validation #### SubscriptionError Raised when subscription to a channel fails #### Exception Handling Example from gmgnapi import GmGnClient, GmGnAPIError, AuthenticationError try: async with GmGnClient() as client: await client.subscribe_wallet_trades(["wallet_address"]) await client.listen() except AuthenticationError as e: print(f"Authentication failed: {e}") except GmGnAPIError as e: print(f"API error: {e}") except Exception as e: print(f"Unexpected error: {e}") --- # GmGnAPI - Professional Python Client for GMGN.ai GmGnAPI Professional Python Client ================================== Connect to GMGN.ai's WebSocket API for real-time Solana blockchain data streams. Built with modern Python, async/await patterns, and enterprise-grade features. [Get Started](https://chipadevteam.github.io/GmGnAPI/getting-started.html) [View Examples](https://chipadevteam.github.io/GmGnAPI/examples.html) [Join Discord](https://discord.gg/ub46R9Dk) 100% Type Hinted Async First Python 3.8+ Compatible quick\_start.py import asyncio from gmgnapi import GmGnClient async def main(): async with GmGnClient() as client: # Subscribe to new token pools await client.subscribe_new_pools() # Handle real-time data @client.on_new_pool async def handle_pool(pool_data): print(f"New pool: {pool_data.chain}") # Start receiving data await client.listen() asyncio.run(main()) Key Features ------------ ### Real-time WebSocket Persistent connection with automatic reconnection and exponential backoff for reliable data streaming. ### Type Safety Full type hints with Pydantic v2 models for robust data validation and IDE support. ### Advanced Filtering Sophisticated filtering system for market cap, volume, liquidity, and custom conditions. ### Data Export Export to JSON, CSV, or SQLite with automatic rotation and compression options. ### Monitoring & Stats Real-time monitoring with detailed statistics and performance metrics. ### Smart Alerts Intelligent alerting system with webhook and email notifications for important events. Quick Installation ------------------ ### PyPI (Recommended) pip install gmgnapi ### Development git clone https://github.com/theshadow76/GmGnAPI.git cd GmGnAPI pip install -e . See It In Action ---------------- Basic Usage Advanced Filtering Monitoring import asyncio from gmgnapi import GmGnClient async def monitor_new_pools(): """Monitor new liquidity pools in real-time.""" async with GmGnClient() as client: await client.subscribe_new_pools(chain="sol") @client.on_new_pool async def handle_new_pool(pool_data): for pool in pool_data.pools: if pool.bti: # Base token info available print(f"šŸ†• New Pool: {pool.bti.s}") print(f" šŸ’° Market Cap: ${pool.bti.mc:,.2f}") print(f" šŸ“Š Liquidity: ${pool.il or 0:,.2f}") print(f" šŸŖ Exchange: {pool.ex}") print(" " + "="*40) await client.listen() asyncio.run(monitor_new_pools()) from gmgnapi import GmGnEnhancedClient, TokenFilter from decimal import Decimal async def monitor_high_value_pools(): """Monitor only high-value pools with custom filters.""" # Define filter for quality tokens token_filter = TokenFilter( min_market_cap=Decimal("100000"), # Min $100K market cap min_liquidity=Decimal("50000"), # Min $50K liquidity min_volume_24h=Decimal("10000"), # Min $10K 24h volume max_risk_score=0.3, # Low risk only exchanges=["raydium", "orca"] # Major DEXes only ) async with GmGnEnhancedClient(token_filter=token_filter) as client: await client.subscribe_new_pools() @client.on_filtered_pool async def handle_quality_pool(pool_data): print(f"šŸŽÆ Quality Pool Found!") print(f" Symbol: {pool_data.pools[0].bti.s}") print(f" Market Cap: ${pool_data.pools[0].bti.mc:,.2f}") await client.listen() from gmgnapi import GmGnEnhancedClient, AlertConfig, DataExportConfig async def enterprise_monitoring(): """Enterprise-grade monitoring with alerts and data export.""" # Configure alerts alert_config = AlertConfig( enabled=True, webhook_url="https://hooks.slack.com/...", conditions=[\ {"field": "market_cap", "operator": ">", "value": 1000000},\ {"field": "volume_24h", "operator": ">", "value": 500000}\ ] ) # Configure data export export_config = DataExportConfig( enabled=True, format="json", file_path="./data/pools.json", rotation_interval_hours=6 ) async with GmGnEnhancedClient( alert_config=alert_config, export_config=export_config ) as client: await client.subscribe_new_pools() # Monitor statistics @client.on_statistics_update async def log_stats(stats): print(f"šŸ“Š Stats: {stats.total_messages} messages") print(f" Uptime: {stats.connection_uptime:.1f}s") print(f" Tokens: {stats.unique_tokens_seen}") await client.listen() Complete Documentation ---------------------- [### Getting Started\ \ Installation, basic setup, and your first WebSocket connection.](https://chipadevteam.github.io/GmGnAPI/getting-started.html) [### API Reference\ \ Complete reference for all classes, methods, and data models.](https://chipadevteam.github.io/GmGnAPI/api-reference.html) [### Examples\ \ Practical examples from basic usage to advanced enterprise patterns.](https://chipadevteam.github.io/GmGnAPI/examples.html) [### Advanced Usage\ \ Filtering, monitoring, alerts, data export, and production deployment.](https://chipadevteam.github.io/GmGnAPI/advanced.html) --- # Getting Started - GmGnAPI Documentation Installation ------------ GmGnAPI is available on PyPI and can be installed using pip: pip install gmgnapi #### Development Installation For development or to get the latest features: git clone https://github.com/theshadow76/GmGnAPI.git cd GmGnAPI pip install -e . GMGN Account Setup ------------------ #### Create Your GMGN Account To use GmGnAPI, you'll need a GMGN account. Create your account using our referral link to support the project: [Create GMGN Account](https://gmgn.ai/?ref=9dLKvFyE&chain=sol) Using this referral link helps support GmGnAPI development #### Get Your API Token 1. Log in to your GMGN account 2. Navigate to Account Settings 3. Generate an API token 4. Copy and securely store your token **Security Note:** Never share your API token or commit it to version control. #### Join Our Community Get help, share experiences, and stay updated with the latest features: [Join Discord Server](https://discord.gg/ub46R9Dk) Requirements ------------ #### Python Version Python 3.8 or higher `python --version` #### Dependencies * websockets >= 12.0 * pydantic >= 2.0 * aiofiles >= 23.0 #### Optional * pandas (for data analysis) * numpy (for numerical operations) * aiohttp (for webhook alerts) # Install with optional dependencies pip install gmgnapi[all] # Or install specific optional dependencies pip install gmgnapi[pandas,analysis] Quick Start ----------- Here's the fastest way to start receiving real-time data: import asyncio from gmgnapi import GmGnClient async def main(): # Create client instance async with GmGnClient() as client: # Subscribe to new liquidity pools await client.subscribe_new_pools(chain="sol") # Define event handler @client.on_new_pool async def handle_new_pool(pool_data): print(f"šŸ†• New pool detected!") for pool in pool_data.pools: if pool.bti: # Base token info available print(f" Token: {pool.bti.s} ({pool.bti.n})") print(f" Market Cap: ${pool.bti.mc:,.2f}") print(f" Exchange: {pool.ex}") print("-" * 40) # Start listening for messages print("šŸ”„ Starting GMGN data stream...") await client.listen() # Run the async function if __name__ == "__main__": asyncio.run(main()) #### That's it! Save this as `quick_start.py` and run it with `python quick_start.py`. You should start seeing real-time pool data! Authentication -------------- Some GMGN.ai channels require authentication. You can provide your access token in several ways: ### 1\. Environment Variable (Recommended) # Set environment variable export GMGN_ACCESS_TOKEN="your_access_token_here" # Client will automatically use the environment variable async with GmGnClient() as client: await client.subscribe_wallet_trades() # Requires auth ### 2\. Direct Parameter async with GmGnClient(access_token="your_token") as client: await client.subscribe_wallet_trades() ### 3\. Configuration File # config.json { "access_token": "your_access_token_here", "default_chain": "sol" } # In your code import json with open('config.json') as f: config = json.load(f) async with GmGnClient(access_token=config['access_token']) as client: # Your code here #### Security Note Never commit access tokens to version control. Use environment variables or secure configuration management. Basic Usage Patterns -------------------- ### Available Subscriptions #### New Pools `subscribe_new_pools()` Real-time new liquidity pool notifications #### Token Updates `subscribe_token_updates()` Price and volume changes for specific tokens #### Wallet Trades `subscribe_wallet_trades()` Trading activity from specific wallets ### Multiple Subscriptions async def multi_subscription_example(): async with GmGnClient() as client: # Subscribe to multiple data streams await client.subscribe_new_pools(chain="sol") await client.subscribe_token_updates(tokens=["EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v"]) # Handle different event types @client.on_new_pool async def handle_pools(data): print(f"šŸ“Š New pool: {len(data.pools)} pools") @client.on_token_update async def handle_updates(data): print(f"šŸ’° Token update: ${data.price_usd}") @client.on_error async def handle_error(error): print(f"āŒ Error: {error}") await client.listen() ### Chain Selection # Solana (default) await client.subscribe_new_pools(chain="sol") # Ethereum await client.subscribe_new_pools(chain="eth") # Binance Smart Chain await client.subscribe_new_pools(chain="bsc") # Polygon await client.subscribe_new_pools(chain="polygon") Error Handling -------------- GmGnAPI includes comprehensive error handling and automatic reconnection: import logging from gmgnapi import GmGnClient, GmGnAPIError # Set up logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) async def robust_client_example(): try: async with GmGnClient() as client: # Error event handler @client.on_error async def handle_error(error): logger.error(f"WebSocket error: {error}") # Custom error handling logic here # Connection event handlers @client.on_connect async def handle_connect(): logger.info("āœ… Connected to GMGN WebSocket") @client.on_disconnect async def handle_disconnect(): logger.warning("āš ļø Disconnected from GMGN WebSocket") @client.on_reconnect async def handle_reconnect(): logger.info("šŸ”„ Reconnected to GMGN WebSocket") await client.subscribe_new_pools() await client.listen() except GmGnAPIError as e: logger.error(f"GMGN API Error: {e}") except Exception as e: logger.error(f"Unexpected error: {e}") raise ### Common Error Types #### ConnectionError Network connectivity issues - automatically retried with exponential backoff #### AuthenticationError Invalid or missing access token for authenticated channels #### ValidationError Invalid data received from the API - check your subscription parameters Next Steps ---------- [### View Examples\ \ Explore practical examples for common use cases](https://chipadevteam.github.io/GmGnAPI/examples.html) [### API Reference\ \ Complete documentation of all classes and methods](https://chipadevteam.github.io/GmGnAPI/api-reference.html) [### Advanced Features\ \ Learn about filtering, monitoring, and data export](https://chipadevteam.github.io/GmGnAPI/advanced.html) #### Pro Tips * Use the enhanced client (`GmGnEnhancedClient`) for production applications * Enable logging to debug connection issues * Consider implementing data persistence for important events * Monitor your application's memory usage with long-running connections * Join our GitHub Discussions for community support --- # Examples - GmGnAPI Documentation Basic Pool Monitoring --------------------- Simple example to monitor new liquidity pools in real-time. ### Basic Pool Monitor import asyncio import logging from datetime import datetime from gmgnapi import GmGnClient # Set up logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) async def basic_pool_monitor(): """ Monitor new Solana pools and log basic information. Perfect for getting started with real-time blockchain data. """ async with GmGnClient() as client: # Subscribe to new pools on Solana await client.subscribe_new_pools(chain="sol") @client.on_new_pool async def handle_new_pool(pool_data): timestamp = datetime.now().strftime("%H:%M:%S") logger.info(f"[{timestamp}] New pools detected: {len(pool_data.pools)}") for pool in pool_data.pools: if pool.bti: # Base token info available token_info = pool.bti logger.info(f" šŸŖ™ Token: {token_info.s} ({token_info.n})") logger.info(f" šŸ’° Market Cap: ${token_info.mc:,.2f}" if token_info.mc else " šŸ’° Market Cap: Unknown") logger.info(f" šŸŖ Exchange: {pool.ex}") logger.info(f" šŸ“ Pool Address: {pool.a}") logger.info(" " + "="*50) @client.on_connect async def handle_connect(): logger.info("āœ… Connected to GMGN WebSocket") @client.on_disconnect async def handle_disconnect(): logger.warning("āš ļø Disconnected from GMGN WebSocket") @client.on_error async def handle_error(error): logger.error(f"āŒ Error: {error}") logger.info("šŸš€ Starting basic pool monitor...") await client.listen() if __name__ == "__main__": try: asyncio.run(basic_pool_monitor()) except KeyboardInterrupt: logger.info("šŸ‘‹ Monitor stopped by user") except Exception as e: logger.error(f"šŸ’„ Unexpected error: {e}") raise Advanced Token Filtering ------------------------ Filter tokens based on market cap, volume, and risk criteria. ### Smart Token Filter import asyncio from decimal import Decimal from gmgnapi import GmGnEnhancedClient, TokenFilter class SmartTokenMonitor: """ Advanced token monitoring with intelligent filtering. Only shows tokens that meet specific quality criteria. """ def __init__(self): # Define filter for high-quality tokens self.quality_filter = TokenFilter( min_market_cap=Decimal("50000"), # Minimum $50K market cap max_market_cap=Decimal("50000000"), # Maximum $50M market cap min_liquidity=Decimal("25000"), # Minimum $25K liquidity min_volume_24h=Decimal("5000"), # Minimum $5K daily volume exchanges=["raydium", "orca", "meteora"], # Trusted exchanges only max_risk_score=0.4, # Medium-low risk only exclude_symbols=["SCAM", "TEST", "FAKE"] # Filter out obvious scams ) # Track statistics self.total_pools_seen = 0 self.quality_pools_found = 0 async def start_monitoring(self): """Start the smart monitoring system.""" async with GmGnEnhancedClient(token_filter=self.quality_filter) as client: await client.subscribe_new_pools(chain="sol") @client.on_new_pool async def handle_all_pools(pool_data): """Track all pools for statistics.""" self.total_pools_seen += len(pool_data.pools) @client.on_filtered_pool async def handle_quality_pool(pool_data): """Handle pools that pass quality filters.""" self.quality_pools_found += 1 for pool in pool_data.pools: if pool.bti: await self.analyze_quality_token(pool) # Log statistics filter_rate = (self.quality_pools_found / max(self.total_pools_seen, 1)) * 100 print(f"šŸ“Š Filter Rate: {filter_rate:.1f}% ({self.quality_pools_found}/{self.total_pools_seen})") @client.on_statistics_update async def handle_stats(stats): """Log monitoring statistics.""" print(f"šŸ“ˆ Monitoring Stats:") print(f" Messages: {stats.total_messages}") print(f" Uptime: {stats.connection_uptime:.1f}s") print(f" Unique Tokens: {stats.unique_tokens_seen}") print("šŸ§  Starting smart token monitor...") await client.listen() async def analyze_quality_token(self, pool): """Analyze a quality token that passed filters.""" token = pool.bti print(f"\nšŸŽÆ QUALITY TOKEN FOUND:") print(f" Symbol: {token.s}") print(f" Name: {token.n}") print(f" Market Cap: ${token.mc:,.2f}") print(f" 24h Volume: ${token.v24h:,.2f}" if token.v24h else " 24h Volume: Unknown") print(f" Exchange: {pool.ex}") print(f" Pool: {pool.a}") # Additional analysis if token.mc and token.v24h: volume_to_mcap_ratio = token.v24h / token.mc if volume_to_mcap_ratio > 0.1: # High activity print(f" šŸ”„ HIGH ACTIVITY: {volume_to_mcap_ratio:.2%} volume/mcap ratio") if token.hc and token.hc > 1000: # High holder count print(f" šŸ‘„ STRONG COMMUNITY: {token.hc:,} holders") print(" " + "="*60) async def main(): monitor = SmartTokenMonitor() try: await monitor.start_monitoring() except KeyboardInterrupt: print("\nšŸ‘‹ Smart monitor stopped") if __name__ == "__main__": asyncio.run(main()) Data Export & Storage --------------------- Export real-time data to various formats for analysis and archiving. ### Comprehensive Data Exporter import asyncio import json import csv import sqlite3 from datetime import datetime, timedelta from pathlib import Path from gmgnapi import GmGnEnhancedClient, DataExportConfig class DataExporter: """ Export blockchain data to multiple formats: - JSON for programmatic access - CSV for spreadsheet analysis - SQLite for structured queries """ def __init__(self, data_dir="./blockchain_data"): self.data_dir = Path(data_dir) self.data_dir.mkdir(exist_ok=True) # Configure different export formats self.json_config = DataExportConfig( enabled=True, format="json", file_path=str(self.data_dir / "pools.json"), max_file_size_mb=50, rotation_interval_hours=12, compress=True, include_metadata=True ) self.csv_config = DataExportConfig( enabled=True, format="csv", file_path=str(self.data_dir / "pools.csv"), max_file_size_mb=25, rotation_interval_hours=24 ) # Set up SQLite database self.setup_database() def setup_database(self): """Initialize SQLite database for structured storage.""" db_path = self.data_dir / "blockchain_data.db" with sqlite3.connect(db_path) as conn: conn.execute(""" CREATE TABLE IF NOT EXISTS pools ( id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp DATETIME DEFAULT CURRENT_TIMESTAMP, pool_address TEXT, token_symbol TEXT, token_name TEXT, market_cap REAL, volume_24h REAL, exchange TEXT, chain TEXT, holder_count INTEGER ) """) conn.execute(""" CREATE INDEX IF NOT EXISTS idx_timestamp ON pools(timestamp); """) conn.execute(""" CREATE INDEX IF NOT EXISTS idx_symbol ON pools(token_symbol); """) async def start_export_monitoring(self): """Start monitoring and exporting data.""" async with GmGnEnhancedClient(export_config=self.json_config) as client: await client.subscribe_new_pools(chain="sol") @client.on_new_pool async def export_pool_data(pool_data): """Export pool data to all configured formats.""" for pool in pool_data.pools: if pool.bti: # Export to CSV await self.export_to_csv(pool) # Export to SQLite await self.export_to_sqlite(pool) # JSON export is handled automatically by the client print(f"šŸ“ Exported: {pool.bti.s} to all formats") print("šŸ’¾ Starting data export monitoring...") await client.listen() async def export_to_csv(self, pool): """Export pool data to CSV format.""" csv_file = self.data_dir / f"pools_{datetime.now().strftime('%Y%m%d')}.csv" # Check if file exists to write headers write_headers = not csv_file.exists() with open(csv_file, 'a', newline='', encoding='utf-8') as f: writer = csv.writer(f) if write_headers: writer.writerow([\ 'timestamp', 'pool_address', 'token_symbol', 'token_name',\ 'market_cap', 'volume_24h', 'exchange', 'chain', 'holder_count'\ ]) token = pool.bti writer.writerow([\ datetime.now().isoformat(),\ pool.a,\ token.s or '',\ token.n or '',\ token.mc or 0,\ token.v24h or 0,\ pool.ex,\ 'sol',\ token.hc or 0\ ]) async def export_to_sqlite(self, pool): """Export pool data to SQLite database.""" db_path = self.data_dir / "blockchain_data.db" with sqlite3.connect(db_path) as conn: token = pool.bti conn.execute(""" INSERT INTO pools ( pool_address, token_symbol, token_name, market_cap, volume_24h, exchange, chain, holder_count ) VALUES (?, ?, ?, ?, ?, ?, ?, ?) """, ( pool.a, token.s, token.n, token.mc, token.v24h, pool.ex, 'sol', token.hc )) async def generate_daily_report(self): """Generate a daily summary report.""" db_path = self.data_dir / "blockchain_data.db" report_date = datetime.now().strftime('%Y-%m-%d') with sqlite3.connect(db_path) as conn: # Get daily statistics cursor = conn.execute(""" SELECT COUNT(*) as total_pools, COUNT(DISTINCT token_symbol) as unique_tokens, AVG(market_cap) as avg_market_cap, SUM(volume_24h) as total_volume, exchange, COUNT(*) as exchange_count FROM pools WHERE date(timestamp) = date('now') GROUP BY exchange ORDER BY exchange_count DESC """) results = cursor.fetchall() report_file = self.data_dir / f"daily_report_{report_date}.txt" with open(report_file, 'w') as f: f.write(f"Daily Blockchain Data Report - {report_date}\n") f.write("=" * 50 + "\n\n") for row in results: f.write(f"Exchange: {row[4]}\n") f.write(f" Total Pools: {row[5]}\n") f.write(f" Avg Market Cap: ${row[2]:,.2f}\n") f.write(f" Total Volume: ${row[3]:,.2f}\n\n") print(f"šŸ“Š Daily report generated: {report_file}") async def main(): exporter = DataExporter() # Start monitoring in background monitor_task = asyncio.create_task(exporter.start_export_monitoring()) # Generate daily reports every 24 hours async def daily_report_scheduler(): while True: await asyncio.sleep(24 * 60 * 60) # 24 hours await exporter.generate_daily_report() report_task = asyncio.create_task(daily_report_scheduler()) try: await asyncio.gather(monitor_task, report_task) except KeyboardInterrupt: print("\nšŸ“ Data export stopped") monitor_task.cancel() report_task.cancel() if __name__ == "__main__": asyncio.run(main()) Intelligent Alert System ------------------------ Set up smart alerts for significant market events and opportunities. ### Multi-Channel Alert System import asyncio import aiohttp import smtplib from email.mime.text import MimeText from datetime import datetime from decimal import Decimal from gmgnapi import GmGnEnhancedClient, AlertConfig, TokenFilter class IntelligentAlertSystem: """ Advanced alert system with multiple notification channels: - Slack/Discord webhooks - Email notifications - Console alerts with severity levels """ def __init__(self, slack_webhook=None, email_config=None): self.slack_webhook = slack_webhook self.email_config = email_config # Configure alerts for different scenarios self.alert_config = AlertConfig( enabled=True, webhook_url=slack_webhook, conditions=[\ # Large market cap opportunities\ {\ "name": "large_cap_opportunity",\ "field": "market_cap",\ "operator": ">",\ "value": 1000000,\ "severity": "high"\ },\ # High volume activity\ {\ "name": "volume_spike",\ "field": "volume_24h",\ "operator": ">",\ "value": 500000,\ "severity": "medium"\ },\ # New exchange listings\ {\ "name": "major_exchange",\ "field": "exchange",\ "operator": "in",\ "value": ["raydium", "orca"],\ "severity": "medium"\ }\ ], rate_limit_seconds=30 # Prevent spam ) # Filter for alert-worthy tokens self.alert_filter = TokenFilter( min_market_cap=Decimal("10000"), min_liquidity=Decimal("5000"), max_risk_score=0.5 ) async def start_alert_monitoring(self): """Start the intelligent alert monitoring system.""" async with GmGnEnhancedClient( token_filter=self.alert_filter, alert_config=self.alert_config ) as client: await client.subscribe_new_pools(chain="sol") @client.on_filtered_pool async def analyze_for_alerts(pool_data): """Analyze filtered pools for alert conditions.""" for pool in pool_data.pools: if pool.bti: await self.check_alert_conditions(pool) @client.on_alert_triggered async def handle_triggered_alert(alert_data): """Handle alerts triggered by the system.""" await self.send_alert_notifications(alert_data) print("šŸšØ Starting intelligent alert system...") await client.listen() async def check_alert_conditions(self, pool): """Check pool against custom alert conditions.""" token = pool.bti # Mega cap alert (>$10M) if token.mc and token.mc > 10000000: await self.trigger_custom_alert( "mega_cap_alert", f"šŸš€ MEGA CAP TOKEN: {token.s}", f"Market Cap: ${token.mc:,.2f}", "critical" ) # Rapid growth alert if token.v24h and token.mc: volume_ratio = token.v24h / token.mc if volume_ratio > 0.5: # Volume > 50% of market cap await self.trigger_custom_alert( "rapid_growth", f"šŸ“ˆ RAPID GROWTH: {token.s}", f"Volume/MarketCap: {volume_ratio:.1%}", "high" ) # Community strength alert if token.hc and token.hc > 5000: await self.trigger_custom_alert( "strong_community", f"šŸ‘„ STRONG COMMUNITY: {token.s}", f"Holders: {token.hc:,}", "medium" ) async def trigger_custom_alert(self, alert_type, title, message, severity): """Trigger a custom alert with notifications.""" alert_data = { "type": alert_type, "title": title, "message": message, "severity": severity, "timestamp": datetime.now().isoformat() } await self.send_alert_notifications(alert_data) async def send_alert_notifications(self, alert_data): """Send notifications through all configured channels.""" # Console alert self.log_console_alert(alert_data) # Slack notification if self.slack_webhook: await self.send_slack_alert(alert_data) # Email notification for high/critical severity if self.email_config and alert_data.get("severity") in ["high", "critical"]: await self.send_email_alert(alert_data) def log_console_alert(self, alert_data): """Log alert to console with color coding.""" severity = alert_data.get("severity", "low") # Color coding based on severity colors = { "critical": "šŸ”“", "high": "šŸŸ ", "medium": "šŸŸ”", "low": "šŸŸ¢" } icon = colors.get(severity, "ā„¹ļø") timestamp = datetime.now().strftime("%H:%M:%S") print(f"\n{icon} [{timestamp}] {alert_data['title']}") print(f" {alert_data['message']}") print(f" Severity: {severity.upper()}") print(" " + "="*50) async def send_slack_alert(self, alert_data): """Send alert to Slack webhook.""" if not self.slack_webhook: return severity_colors = { "critical": "#ff0000", "high": "#ff8800", "medium": "#ffaa00", "low": "#00ff00" } color = severity_colors.get(alert_data.get("severity"), "#808080") payload = { "attachments": [\ {\ "color": color,\ "title": alert_data["title"],\ "text": alert_data["message"],\ "fields": [\ {\ "title": "Severity",\ "value": alert_data.get("severity", "unknown").upper(),\ "short": True\ },\ {\ "title": "Time",\ "value": alert_data["timestamp"],\ "short": True\ }\ ]\ }\ ] } try: async with aiohttp.ClientSession() as session: async with session.post(self.slack_webhook, json=payload) as response: if response.status == 200: print("āœ… Slack alert sent successfully") else: print(f"āŒ Failed to send Slack alert: {response.status}") except Exception as e: print(f"āŒ Slack alert error: {e}") async def send_email_alert(self, alert_data): """Send email alert for high-priority events.""" if not self.email_config: return try: subject = f"šŸšØ Blockchain Alert: {alert_data['title']}" body = f""" Alert Details: Title: {alert_data['title']} Message: {alert_data['message']} Severity: {alert_data.get('severity', 'unknown').upper()} Time: {alert_data['timestamp']} This is an automated alert from GmGnAPI monitoring system. """ msg = MimeText(body) msg['Subject'] = subject msg['From'] = self.email_config['from'] msg['To'] = self.email_config['to'] # Send email (implementation depends on your email provider) print(f"šŸ“§ Email alert prepared: {subject}") except Exception as e: print(f"āŒ Email alert error: {e}") async def main(): # Configure your notification channels SLACK_WEBHOOK = "https://hooks.slack.com/services/YOUR/SLACK/WEBHOOK" EMAIL_CONFIG = { "from": "alerts@yourapp.com", "to": "you@email.com", "smtp_server": "smtp.gmail.com", "smtp_port": 587, "username": "your_email@gmail.com", "password": "your_app_password" } alert_system = IntelligentAlertSystem( slack_webhook=SLACK_WEBHOOK, email_config=EMAIL_CONFIG ) try: await alert_system.start_alert_monitoring() except KeyboardInterrupt: print("\nšŸšØ Alert system stopped") if __name__ == "__main__": asyncio.run(main()) Whale Activity Monitor ---------------------- Track large wallet movements and whale trading patterns. ### Whale Detection System import asyncio from decimal import Decimal from collections import defaultdict, deque from datetime import datetime, timedelta from gmgnapi import GmGnClient class WhaleActivityMonitor: """ Monitor and analyze whale trading activity. Tracks large transactions and wallet patterns. """ def __init__(self, whale_threshold_usd=50000): self.whale_threshold = Decimal(str(whale_threshold_usd)) # Track whale wallets and their activity self.whale_wallets = set() self.wallet_activity = defaultdict(lambda: { 'total_volume': Decimal('0'), 'trade_count': 0, 'first_seen': None, 'last_activity': None, 'tokens_traded': set() }) # Track recent large transactions self.recent_large_trades = deque(maxlen=100) # Known whale wallets (you can populate this from external sources) self.known_whales = { # Add known whale addresses here # "wallet_address": "Whale Name/Description" } async def start_whale_monitoring(self): """Start monitoring whale activity.""" # Note: Wallet monitoring requires authentication # Make sure you have GMGN_ACCESS_TOKEN set async with GmGnClient() as client: # Monitor new pools for whale involvement await client.subscribe_new_pools(chain="sol") # If you have whale wallet addresses, monitor them directly if self.known_whales: whale_addresses = list(self.known_whales.keys()) await client.subscribe_wallet_trades(whale_addresses, chain="sol") @client.on_new_pool async def analyze_pool_for_whales(pool_data): """Analyze new pools for whale involvement.""" for pool in pool_data.pools: if pool.bti and pool.bti.mc: await self.check_pool_whale_activity(pool) @client.on_wallet_trade async def track_whale_trades(trade_data): """Track direct whale trading activity.""" await self.analyze_whale_trade(trade_data) print("šŸ‹ Starting whale activity monitor...") print(f"šŸ’° Whale threshold: ${self.whale_threshold:,}") await client.listen() async def check_pool_whale_activity(self, pool): """Check if a new pool shows signs of whale involvement.""" token = pool.bti # Large initial liquidity suggests whale involvement if pool.il and pool.il > float(self.whale_threshold): await self.log_whale_activity( "large_liquidity", f"šŸŠ Large Initial Liquidity: {token.s}", f"Liquidity: ${pool.il:,.2f} | Pool: {pool.a[:8]}..." ) # High market cap at launch if token.mc and token.mc > float(self.whale_threshold * 2): await self.log_whale_activity( "high_mcap_launch", f"šŸ’Ž High Market Cap Launch: {token.s}", f"Market Cap: ${token.mc:,.2f}" ) # Check volume to market cap ratio (whales often create high volume) if token.v24h and token.mc: volume_ratio = token.v24h / token.mc if volume_ratio > 0.3: # High volume relative to market cap await self.log_whale_activity( "high_volume_ratio", f"šŸ“Š High Volume Activity: {token.s}", f"Volume/MCap: {volume_ratio:.1%} | Volume: ${token.v24h:,.2f}" ) async def analyze_whale_trade(self, trade_data): """Analyze individual whale trades.""" wallet = trade_data.wallet_address # Update wallet statistics wallet_stats = self.wallet_activity[wallet] for trade in trade_data.trades: # Check if this is a whale-sized trade if trade.amount_usd >= self.whale_threshold: await self.record_whale_trade(wallet, trade) # Update wallet activity wallet_stats['total_volume'] += trade_data.total_volume_24h_usd or Decimal('0') wallet_stats['trade_count'] += len(trade_data.trades) wallet_stats['last_activity'] = datetime.now() if not wallet_stats['first_seen']: wallet_stats['first_seen'] = datetime.now() # Add wallet to whale list if volume threshold exceeded if wallet_stats['total_volume'] >= self.whale_threshold: if wallet not in self.whale_wallets: self.whale_wallets.add(wallet) await self.log_whale_activity( "new_whale_detected", f"šŸ†• New Whale Detected: {wallet[:8]}...", f"Total Volume: ${wallet_stats['total_volume']:,.2f}" ) async def record_whale_trade(self, wallet, trade): """Record a significant whale trade.""" trade_info = { 'wallet': wallet, 'trade': trade, 'timestamp': datetime.now() } self.recent_large_trades.append(trade_info) # Determine whale name if known whale_name = self.known_whales.get(wallet, f"Whale {wallet[:8]}...") await self.log_whale_activity( "whale_trade", f"šŸ‹ {whale_name} {trade.trade_type.upper()}", f"Amount: ${trade.amount_usd:,.2f} | Token: {trade.token_address[:8]}..." ) # Check for unusual patterns await self.check_whale_patterns(wallet) async def check_whale_patterns(self, wallet): """Check for unusual whale trading patterns.""" # Check recent trades from this wallet recent_trades = [\ t for t in self.recent_large_trades \ if t['wallet'] == wallet and \ t['timestamp'] > datetime.now() - timedelta(hours=1)\ ] if len(recent_trades) >= 3: total_amount = sum(t['trade'].amount_usd for t in recent_trades) await self.log_whale_activity( "whale_activity_burst", f"āš” Whale Activity Burst: {wallet[:8]}...", f"3+ trades in 1h | Total: ${total_amount:,.2f}", severity="high" ) async def log_whale_activity(self, activity_type, title, details, severity="medium"): """Log whale activity with appropriate formatting.""" timestamp = datetime.now().strftime("%H:%M:%S") # Severity indicators indicators = { "low": "šŸŸ¢", "medium": "šŸŸ”", "high": "šŸ”“" } indicator = indicators.get(severity, "ā„¹ļø") print(f"\n{indicator} [{timestamp}] {title}") print(f" {details}") print(f" Activity Type: {activity_type}") print(" " + "="*60) def get_whale_statistics(self): """Get current whale monitoring statistics.""" print(f"\nšŸ‹ WHALE MONITORING STATISTICS") print("=" * 50) print(f"Known Whales: {len(self.whale_wallets)}") print(f"Recent Large Trades: {len(self.recent_large_trades)}") print(f"Monitoring Threshold: ${self.whale_threshold:,}") # Top whales by volume top_whales = sorted( self.wallet_activity.items(), key=lambda x: x[1]['total_volume'], reverse=True )[:5] print(f"\nTop 5 Whales by Volume:") for i, (wallet, stats) in enumerate(top_whales, 1): whale_name = self.known_whales.get(wallet, f"{wallet[:8]}...") print(f" {i}. {whale_name}") print(f" Volume: ${stats['total_volume']:,.2f}") print(f" Trades: {stats['trade_count']}") async def main(): # Create whale monitor with $50K threshold whale_monitor = WhaleActivityMonitor(whale_threshold_usd=50000) try: await whale_monitor.start_whale_monitoring() except KeyboardInterrupt: print("\nšŸ‹ Whale monitor stopped") # Show final statistics whale_monitor.get_whale_statistics() if __name__ == "__main__": asyncio.run(main()) Quick Examples -------------- Short, focused examples for specific use cases. #### Portfolio Tracker async def track_portfolio(): """Track specific tokens in your portfolio.""" portfolio_tokens = [\ "EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v", # USDC\ "So11111111111111111111111111111111111111112", # SOL\ ] async with GmGnClient() as client: await client.subscribe_token_updates(portfolio_tokens) @client.on_token_update async def handle_portfolio_update(update): print(f"šŸ’¼ {update.token_address[:8]}...") print(f" Price: ${update.price_usd}") print(f" 24h Change: {update.price_change_24h:.2%}") await client.listen() #### Price Alert Bot async def price_alert_bot(): """Simple price alert system.""" price_targets = { "SOL": {"above": 200, "below": 150}, "BTC": {"above": 100000, "below": 90000} } async with GmGnClient() as client: await client.subscribe_new_pools() @client.on_new_pool async def check_prices(pool_data): for pool in pool_data.pools: if pool.bti and pool.bti.s in price_targets: price = pool.bti.p symbol = pool.bti.s targets = price_targets[symbol] if price > targets["above"]: print(f"šŸš€ {symbol} above ${targets['above']}: ${price}") elif price < targets["below"]: print(f"šŸ“‰ {symbol} below ${targets['below']}: ${price}") await client.listen() #### Volume Scanner async def volume_scanner(): """Scan for high volume trading activity.""" volume_threshold = 1000000 # $1M+ volume async with GmGnClient() as client: await client.subscribe_new_pools() @client.on_new_pool async def scan_volume(pool_data): for pool in pool_data.pools: if pool.bti and pool.bti.v24h: if pool.bti.v24h > volume_threshold: print(f"šŸ“Š HIGH VOLUME: {pool.bti.s}") print(f" 24h Volume: ${pool.bti.v24h:,.2f}") print(f" Exchange: {pool.ex}") await client.listen() #### New Token Announcer async def new_token_announcer(): """Announce new tokens with basic info.""" async with GmGnClient() as client: await client.subscribe_new_pools() @client.on_new_pool async def announce_token(pool_data): for pool in pool_data.pools: if pool.bti: token = pool.bti print(f"šŸ†• NEW TOKEN: {token.s}") print(f" Name: {token.n}") print(f" Market Cap: ${token.mc:,.2f}") print(f" Pool: {pool.a}") print(f" Exchange: {pool.ex}") print(" " + "-"*30) await client.listen() --- # Advanced Usage - GmGnAPI Documentation Production Deployment --------------------- Best practices for deploying GmGnAPI in production environments. ### Docker Deployment # Dockerfile FROM python:3.11-slim WORKDIR /app # Install system dependencies RUN apt-get update && apt-get install -y \ gcc \ && rm -rf /var/lib/apt/lists/* # Copy requirements and install Python dependencies COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt # Copy application code COPY . . # Create non-root user RUN useradd -m -u 1000 gmgnapi && chown -R gmgnapi:gmgnapi /app USER gmgnapi # Health check HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \ CMD python -c "import gmgnapi; print('OK')" || exit 1 # Run the application CMD ["python", "main.py"] # docker-compose.yml version: '3.8' services: gmgnapi: build: . restart: unless-stopped environment: - GMGN_ACCESS_TOKEN=${GMGN_ACCESS_TOKEN} - LOG_LEVEL=INFO - DATA_DIR=/app/data volumes: - ./data:/app/data - ./logs:/app/logs networks: - gmgn-network redis: image: redis:7-alpine restart: unless-stopped volumes: - redis_data:/data networks: - gmgn-network prometheus: image: prom/prometheus:latest restart: unless-stopped ports: - "9090:9090" volumes: - ./prometheus.yml:/etc/prometheus/prometheus.yml - prometheus_data:/prometheus networks: - gmgn-network volumes: redis_data: prometheus_data: networks: gmgn-network: driver: bridge ### Kubernetes Deployment # k8s-deployment.yaml apiVersion: apps/v1 kind: Deployment metadata: name: gmgnapi labels: app: gmgnapi spec: replicas: 3 selector: matchLabels: app: gmgnapi template: metadata: labels: app: gmgnapi spec: containers: - name: gmgnapi image: gmgnapi:latest ports: - containerPort: 8080 env: - name: GMGN_ACCESS_TOKEN valueFrom: secretKeyRef: name: gmgn-secrets key: access-token resources: limits: memory: "512Mi" cpu: "500m" requests: memory: "256Mi" cpu: "250m" livenessProbe: httpGet: path: /health port: 8080 initialDelaySeconds: 30 periodSeconds: 10 readinessProbe: httpGet: path: /ready port: 8080 initialDelaySeconds: 5 periodSeconds: 5 --- apiVersion: v1 kind: Service metadata: name: gmgnapi-service spec: selector: app: gmgnapi ports: - protocol: TCP port: 80 targetPort: 8080 type: ClusterIP Advanced Custom Filtering ------------------------- Create sophisticated filtering logic for complex trading strategies. from decimal import Decimal from typing import Dict, Any, Callable, List from gmgnapi import TokenFilter, GmGnEnhancedClient import asyncio class AdvancedTokenFilter: """ Advanced filtering system with custom rules and machine learning integration. """ def __init__(self): self.custom_rules: List[Callable] = [] self.scoring_weights = { 'market_cap_score': 0.3, 'volume_score': 0.25, 'liquidity_score': 0.2, 'community_score': 0.15, 'risk_score': 0.1 } # Historical data for pattern analysis self.token_history = {} def add_custom_rule(self, rule_func: Callable[[Dict[str, Any]], bool]): """Add a custom filtering rule.""" self.custom_rules.append(rule_func) def calculate_token_score(self, token_data: Dict[str, Any]) -> float: """Calculate a comprehensive score for token quality.""" scores = { 'market_cap_score': self._score_market_cap(token_data), 'volume_score': self._score_volume(token_data), 'liquidity_score': self._score_liquidity(token_data), 'community_score': self._score_community(token_data), 'risk_score': self._score_risk(token_data) } # Weighted average total_score = sum( scores[key] * self.scoring_weights[key] for key in scores ) return min(max(total_score, 0), 100) # Clamp to 0-100 def _score_market_cap(self, token_data: Dict[str, Any]) -> float: """Score based on market cap (0-100).""" mc = token_data.get('market_cap', 0) if mc < 10000: return 0 elif mc < 100000: return 20 elif mc < 1000000: return 60 elif mc < 10000000: return 90 else: return 100 def _score_volume(self, token_data: Dict[str, Any]) -> float: """Score based on 24h volume activity.""" volume = token_data.get('volume_24h', 0) mc = token_data.get('market_cap', 1) volume_ratio = volume / mc if mc > 0 else 0 if volume_ratio < 0.01: return 0 elif volume_ratio < 0.05: return 30 elif volume_ratio < 0.1: return 60 elif volume_ratio < 0.3: return 90 else: return 100 def _score_liquidity(self, token_data: Dict[str, Any]) -> float: """Score based on liquidity depth.""" liquidity = token_data.get('liquidity', 0) if liquidity < 1000: return 0 elif liquidity < 10000: return 30 elif liquidity < 50000: return 60 elif liquidity < 200000: return 90 else: return 100 def _score_community(self, token_data: Dict[str, Any]) -> float: """Score based on community metrics.""" holders = token_data.get('holder_count', 0) if holders < 10: return 0 elif holders < 100: return 25 elif holders < 1000: return 50 elif holders < 5000: return 75 else: return 100 def _score_risk(self, token_data: Dict[str, Any]) -> float: """Score based on risk factors (higher score = lower risk).""" risk_factors = 0 # Check for risk indicators if token_data.get('creator_balance_ratio', 0) > 0.5: risk_factors += 30 # High creator holding if token_data.get('top_10_ratio', 0) > 0.8: risk_factors += 25 # High concentration if token_data.get('burn_rate', 0) > 0.1: risk_factors += 20 # High burn rate # Suspicious name patterns symbol = token_data.get('symbol', '').upper() suspicious_keywords = ['MOON', 'SAFE', 'BABY', 'DOGE', 'PEPE', 'TEST'] if any(keyword in symbol for keyword in suspicious_keywords): risk_factors += 15 return max(0, 100 - risk_factors) async def apply_filters(self, token_data: Dict[str, Any]) -> Dict[str, Any]: """Apply all filters and return results.""" # Calculate base score score = self.calculate_token_score(token_data) # Apply custom rules custom_results = [] for rule in self.custom_rules: try: result = rule(token_data) custom_results.append(result) except Exception as e: print(f"Custom rule error: {e}") custom_results.append(False) # Determine if token passes filters passes_score = score >= 60 # Minimum score threshold passes_custom = all(custom_results) if custom_results else True return { 'score': score, 'passes_filter': passes_score and passes_custom, 'custom_rule_results': custom_results, 'breakdown': { 'market_cap_score': self._score_market_cap(token_data), 'volume_score': self._score_volume(token_data), 'liquidity_score': self._score_liquidity(token_data), 'community_score': self._score_community(token_data), 'risk_score': self._score_risk(token_data) } } # Example usage with custom rules async def advanced_filtering_example(): """Example of advanced filtering with custom rules.""" filter_system = AdvancedTokenFilter() # Add custom rules def no_meme_coins(token_data): """Filter out obvious meme coins.""" symbol = token_data.get('symbol', '').upper() meme_keywords = ['DOGE', 'SHIB', 'PEPE', 'FLOKI', 'BABY'] return not any(keyword in symbol for keyword in meme_keywords) def minimum_age_rule(token_data): """Require tokens to be at least 1 hour old.""" created_time = token_data.get('created_timestamp', 0) current_time = asyncio.get_event_loop().time() age_hours = (current_time - created_time) / 3600 return age_hours >= 1 def whale_activity_rule(token_data): """Check for healthy whale activity.""" top_10_ratio = token_data.get('top_10_ratio', 1) return 0.3 <= top_10_ratio <= 0.7 # Balanced distribution # Register custom rules filter_system.add_custom_rule(no_meme_coins) filter_system.add_custom_rule(minimum_age_rule) filter_system.add_custom_rule(whale_activity_rule) # Use with enhanced client async with GmGnEnhancedClient() as client: await client.subscribe_new_pools() @client.on_new_pool async def analyze_with_advanced_filter(pool_data): for pool in pool_data.pools: if pool.bti: token_info = { 'symbol': pool.bti.s, 'market_cap': pool.bti.mc, 'volume_24h': pool.bti.v24h, 'holder_count': pool.bti.hc, 'liquidity': pool.il, 'created_timestamp': asyncio.get_event_loop().time(), 'top_10_ratio': pool.bti.t10hr or 0.5, 'creator_balance_ratio': pool.bti.cbr or 0.1 } result = await filter_system.apply_filters(token_info) if result['passes_filter']: print(f"āœ… HIGH QUALITY TOKEN: {token_info['symbol']}") print(f" Score: {result['score']:.1f}/100") print(f" Breakdown: {result['breakdown']}") else: print(f"āŒ Filtered out: {token_info['symbol']} (Score: {result['score']:.1f})") await client.listen() if __name__ == "__main__": asyncio.run(advanced_filtering_example()) Enterprise Data Persistence --------------------------- Advanced data storage and retrieval patterns for large-scale operations. ### Time-Series Database Integration import asyncio import asyncpg from datetime import datetime, timedelta from typing import Dict, List, Optional import json class TimeSeriesDataManager: """ Advanced time-series data management with PostgreSQL and TimescaleDB. """ def __init__(self, database_url: str): self.database_url = database_url self.pool: Optional[asyncpg.Pool] = None async def initialize(self): """Initialize database connection pool and create tables.""" self.pool = await asyncpg.create_pool( self.database_url, min_size=5, max_size=20, command_timeout=60 ) await self.create_tables() async def create_tables(self): """Create optimized tables for blockchain data.""" async with self.pool.acquire() as conn: # Main pools table await conn.execute(""" CREATE TABLE IF NOT EXISTS pools_timeseries ( time TIMESTAMPTZ NOT NULL, pool_address TEXT NOT NULL, token_symbol TEXT, token_name TEXT, market_cap NUMERIC, volume_24h NUMERIC, liquidity NUMERIC, price NUMERIC, holder_count INTEGER, exchange TEXT, chain TEXT, metadata JSONB, PRIMARY KEY (time, pool_address) ); """) # Create hypertable (TimescaleDB extension) try: await conn.execute(""" SELECT create_hypertable('pools_timeseries', 'time', if_not_exists => TRUE); """) except Exception as e: print(f"Note: TimescaleDB not available: {e}") # Indexes for fast queries await conn.execute(""" CREATE INDEX IF NOT EXISTS idx_pools_symbol_time ON pools_timeseries (token_symbol, time DESC); """) await conn.execute(""" CREATE INDEX IF NOT EXISTS idx_pools_mcap ON pools_timeseries (market_cap) WHERE market_cap IS NOT NULL; """) # Materialized view for aggregated data await conn.execute(""" CREATE MATERIALIZED VIEW IF NOT EXISTS daily_token_stats AS SELECT date_trunc('day', time) AS day, token_symbol, avg(market_cap) AS avg_market_cap, max(market_cap) AS max_market_cap, avg(volume_24h) AS avg_volume, avg(price) AS avg_price, count(*) AS data_points FROM pools_timeseries WHERE token_symbol IS NOT NULL GROUP BY date_trunc('day', time), token_symbol; """) async def store_pool_data(self, pool_data: Dict): """Store pool data with optimized batch insertion.""" async with self.pool.acquire() as conn: await conn.execute(""" INSERT INTO pools_timeseries ( time, pool_address, token_symbol, token_name, market_cap, volume_24h, liquidity, price, holder_count, exchange, chain, metadata ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12) ON CONFLICT (time, pool_address) DO UPDATE SET market_cap = EXCLUDED.market_cap, volume_24h = EXCLUDED.volume_24h, liquidity = EXCLUDED.liquidity, price = EXCLUDED.price, holder_count = EXCLUDED.holder_count, metadata = EXCLUDED.metadata; """, datetime.now(), pool_data['pool_address'], pool_data.get('token_symbol'), pool_data.get('token_name'), pool_data.get('market_cap'), pool_data.get('volume_24h'), pool_data.get('liquidity'), pool_data.get('price'), pool_data.get('holder_count'), pool_data.get('exchange'), pool_data.get('chain'), json.dumps(pool_data.get('metadata', {})) ) async def get_token_history(self, symbol: str, hours: int = 24) -> List[Dict]: """Get historical data for a token.""" async with self.pool.acquire() as conn: rows = await conn.fetch(""" SELECT time, market_cap, volume_24h, price, liquidity FROM pools_timeseries WHERE token_symbol = $1 AND time >= NOW() - INTERVAL '%s hours' ORDER BY time DESC; """, symbol, hours) return [dict(row) for row in rows] async def get_market_trends(self, timeframe: str = '1h') -> List[Dict]: """Get market trend analysis.""" async with self.pool.acquire() as conn: rows = await conn.fetch(f""" SELECT date_trunc('{timeframe}', time) AS period, count(DISTINCT token_symbol) AS new_tokens, avg(market_cap) AS avg_market_cap, sum(volume_24h) AS total_volume FROM pools_timeseries WHERE time >= NOW() - INTERVAL '7 days' GROUP BY date_trunc('{timeframe}', time) ORDER BY period DESC; """) return [dict(row) for row in rows] async def cleanup_old_data(self, days_to_keep: int = 30): """Clean up old data to manage storage.""" async with self.pool.acquire() as conn: result = await conn.execute(""" DELETE FROM pools_timeseries WHERE time < NOW() - INTERVAL '%s days'; """, days_to_keep) print(f"Cleaned up old data: {result}") # Refresh materialized view await conn.execute("REFRESH MATERIALIZED VIEW daily_token_stats;") async def close(self): """Close database connections.""" if self.pool: await self.pool.close() Performance Optimization ------------------------ Optimize GmGnAPI for high-throughput, low-latency operations. #### Connection Pooling import asyncio from gmgnapi import GmGnEnhancedClient from typing import List class OptimizedClientPool: """Manage multiple client connections for high throughput.""" def __init__(self, pool_size: int = 5): self.pool_size = pool_size self.clients: List[GmGnEnhancedClient] = [] self.current_client = 0 async def initialize(self): """Initialize client pool.""" for i in range(self.pool_size): client = GmGnEnhancedClient() await client.connect() self.clients.append(client) def get_client(self) -> GmGnEnhancedClient: """Get next client in round-robin fashion.""" client = self.clients[self.current_client] self.current_client = (self.current_client + 1) % self.pool_size return client #### Message Batching import asyncio from collections import deque from typing import Deque, Dict, Any class MessageBatcher: """Batch messages for efficient processing.""" def __init__(self, batch_size: int = 100, flush_interval: float = 1.0): self.batch_size = batch_size self.flush_interval = flush_interval self.buffer: Deque[Dict[str, Any]] = deque() self.last_flush = asyncio.get_event_loop().time() async def add_message(self, message: Dict[str, Any]): """Add message to batch.""" self.buffer.append(message) current_time = asyncio.get_event_loop().time() # Flush if batch is full or interval exceeded if (len(self.buffer) >= self.batch_size or current_time - self.last_flush >= self.flush_interval): await self.flush() async def flush(self): """Process batched messages.""" if not self.buffer: return batch = list(self.buffer) self.buffer.clear() self.last_flush = asyncio.get_event_loop().time() # Process batch await self.process_batch(batch) async def process_batch(self, batch: List[Dict[str, Any]]): """Override this method for custom batch processing.""" print(f"Processing batch of {len(batch)} messages") # Custom processing logic here #### Memory Management import gc import psutil import asyncio from typing import Optional class MemoryMonitor: """Monitor and manage memory usage.""" def __init__(self, max_memory_mb: int = 1024): self.max_memory_mb = max_memory_mb self.process = psutil.Process() async def start_monitoring(self): """Start memory monitoring loop.""" while True: await self.check_memory() await asyncio.sleep(30) # Check every 30 seconds async def check_memory(self): """Check current memory usage.""" memory_mb = self.process.memory_info().rss / 1024 / 1024 if memory_mb > self.max_memory_mb: print(f"āš ļø High memory usage: {memory_mb:.1f}MB") # Force garbage collection gc.collect() # Check again after GC new_memory_mb = self.process.memory_info().rss / 1024 / 1024 print(f"After GC: {new_memory_mb:.1f}MB") if new_memory_mb > self.max_memory_mb * 0.9: print("šŸšØ Memory usage still high after GC") # Implement additional cleanup strategies Security Best Practices ----------------------- Implement security measures for production deployments. #### Credential Management * Use environment variables for sensitive data * Implement credential rotation * Use encrypted storage for API tokens * Enable audit logging for credential access #### Network Security * Use TLS 1.3 for all connections * Implement connection rate limiting * Set up firewall rules for outbound connections * Monitor for suspicious network activity #### Data Protection * Encrypt sensitive data at rest * Implement data retention policies * Use secure data deletion methods * Regular security audits of stored data import os import ssl import hashlib from cryptography.fernet import Fernet from typing import Optional class SecureConfigManager: """Secure configuration and credential management.""" def __init__(self, encryption_key: Optional[bytes] = None): if encryption_key: self.cipher = Fernet(encryption_key) else: # Generate new key (store securely!) self.cipher = Fernet(Fernet.generate_key()) def encrypt_credential(self, credential: str) -> str: """Encrypt a credential for secure storage.""" return self.cipher.encrypt(credential.encode()).decode() def decrypt_credential(self, encrypted_credential: str) -> str: """Decrypt a stored credential.""" return self.cipher.decrypt(encrypted_credential.encode()).decode() def get_secure_token(self) -> str: """Get API token with fallback hierarchy.""" # Try environment variable first token = os.getenv('GMGN_ACCESS_TOKEN') if token: return token # Try encrypted config file config_file = os.getenv('GMGN_CONFIG_FILE') if config_file and os.path.exists(config_file): with open(config_file, 'r') as f: encrypted_token = f.read().strip() return self.decrypt_credential(encrypted_token) raise ValueError("No valid API token found") @staticmethod def create_ssl_context() -> ssl.SSLContext: """Create secure SSL context for connections.""" context = ssl.create_default_context() context.minimum_version = ssl.TLSVersion.TLSv1_3 context.check_hostname = True context.verify_mode = ssl.CERT_REQUIRED return context Troubleshooting Guide --------------------- Common issues and their solutions for production deployments. #### Connection Issues **Problem:** WebSocket connections dropping frequently **Solutions:** * Increase reconnection delay and max attempts * Check network stability and firewall settings * Implement connection health checks * Use connection pooling for redundancy #### Memory Leaks **Problem:** Memory usage continuously increasing **Solutions:** * Implement data rotation and cleanup * Use weak references for caches * Monitor with memory profiling tools * Set maximum buffer sizes #### Performance Degradation **Problem:** Slow message processing under high load **Solutions:** * Implement message batching * Use async processing queues * Optimize database queries * Scale horizontally with multiple instances #### Authentication Failures **Problem:** Random authentication errors **Solutions:** * Implement token refresh mechanism * Add retry logic with exponential backoff * Monitor token expiration * Use secure credential storage #### Diagnostic Commands # Memory usage analysis import tracemalloc import gc def analyze_memory(): """Analyze current memory usage.""" tracemalloc.start() # Your application code here current, peak = tracemalloc.get_traced_memory() print(f"Current memory usage: {current / 1024 / 1024:.1f} MB") print(f"Peak memory usage: {peak / 1024 / 1024:.1f} MB") # Get top memory consumers snapshot = tracemalloc.take_snapshot() top_stats = snapshot.statistics('lineno') print("\nTop 10 memory consumers:") for stat in top_stats[:10]: print(stat) # Connection diagnostics async def diagnose_connection(): """Diagnose WebSocket connection issues.""" import websockets import time start_time = time.time() try: async with websockets.connect("wss://gmgn.ai/ws") as ws: # Test ping/pong pong_waiter = await ws.ping() await pong_waiter latency = (time.time() - start_time) * 1000 print(f"āœ… Connection successful, latency: {latency:.1f}ms") except Exception as e: print(f"āŒ Connection failed: {e}") # Performance monitoring import asyncio import time from collections import deque class PerformanceMonitor: def __init__(self, window_size: int = 100): self.message_times = deque(maxlen=window_size) self.processing_times = deque(maxlen=window_size) def record_message(self, processing_time: float): """Record message processing time.""" current_time = time.time() self.message_times.append(current_time) self.processing_times.append(processing_time) def get_stats(self) -> dict: """Get performance statistics.""" if len(self.message_times) < 2: return {"error": "Not enough data"} # Calculate message rate (messages per second) time_span = self.message_times[-1] - self.message_times[0] message_rate = len(self.message_times) / time_span if time_span > 0 else 0 # Average processing time avg_processing_time = sum(self.processing_times) / len(self.processing_times) return { "message_rate": message_rate, "avg_processing_time_ms": avg_processing_time * 1000, "total_messages": len(self.message_times) } ---