my-project-public

repository

loading code, commits, and activity

repositories

loading repo index

#1	"""Test Birdeye WebSocket Integration"""
#2
#3	import asyncio
#4	import sys
#5	from pathlib import Path
#6
#7	# Add parent to path
#8	sys.path.insert(0, str(Path(__file__).parent))
#9
#10	from config import load_config
#11	from clients.birdeye_client import BirdeyeWebSocketClient
#12
#13
#14	async def test_websocket():
#15	"""Test Birdeye WebSocket connection and subscriptions"""
#16
#17	print("🧪 Testing Birdeye WebSocket Integration...\n")
#18
#19	# Load config
#20	config = load_config()
#21	print(f"✓ Config loaded")
#22	print(f" API Key: {config.birdeye_api_key[:20]}...")
#23	print(f" WSS URL: {config.birdeye_wss_url[:60]}...\n")
#24
#25	# Create WebSocket client
#26	ws_client = BirdeyeWebSocketClient(api_key=config.birdeye_api_key)
#27	print(f"✓ WebSocket client created\n")
#28
#29	# Event counters
#30	event_counts = {
#31	"PRICE_UPDATE": 0,
#32	"NEW_LISTING": 0,
#33	"LARGE_TRADE": 0,
#34	"WALLET_TX": 0,
#35	}
#36
#37	# Register event handlers
#38	async def on_price_update(data):
#39	event_counts["PRICE_UPDATE"] += 1
#40	print(f"📊 Price Update #{event_counts['PRICE_UPDATE']}: {data.get('address', 'Unknown')[:8]}... = ${data.get('price', 0)}")
#41
#42	async def on_new_listing(data):
#43	event_counts["NEW_LISTING"] += 1
#44	print(f"🆕 New Listing #{event_counts['NEW_LISTING']}: {data.get('name', 'Unknown')} ({data.get('symbol', 'N/A')})")
#45
#46	async def on_large_trade(data):
#47	event_counts["LARGE_TRADE"] += 1
#48	print(f"🐋 Large Trade #{event_counts['LARGE_TRADE']}: ${data.get('amountUSD', 0):,.2f} - {data.get('side', 'N/A')}")
#49
#50	async def on_wallet_tx(data):
#51	event_counts["WALLET_TX"] += 1
#52	print(f"💼 Wallet TX #{event_counts['WALLET_TX']}: {data.get('type', 'Unknown')}")
#53
#54	ws_client.on("PRICE_UPDATE", on_price_update)
#55	ws_client.on("NEW_LISTING", on_new_listing)
#56	ws_client.on("LARGE_TRADE", on_large_trade)
#57	ws_client.on("WALLET_TX", on_wallet_tx)
#58	print(f"✓ Event handlers registered\n")
#59
#60	try:
#61	# Connect
#62	print("Connecting to Birdeye WebSocket...")
#63	await ws_client.connect()
#64	print(f"✓ Connected: {ws_client.is_connected}\n")
#65
#66	# Subscribe to new listings
#67	print("Subscribing to new token listings...")
#68	await ws_client.subscribe_new_listings()
#69	print("✓ Subscribed to new listings\n")
#70
#71	# Subscribe to BONK price updates
#72	bonk_address = "DezXAZ8z7PnrnRJjz3wXBoRgixCa6xjnB7YaB1pPB263"
#73	print(f"Subscribing to BONK price updates ({bonk_address[:8]}...)...")
#74	await ws_client.subscribe_price(bonk_address)
#75	print("✓ Subscribed to BONK prices\n")
#76
#77	# Subscribe to large trades
#78	print("Subscribing to large trades (>$10k)...")
#79	await ws_client.subscribe_large_trades(threshold_usd=10000)
#80	print("✓ Subscribed to large trades\n")
#81
#82	# Subscribe to wallet transactions (MAWD wallet)
#83	if config.wallet_address:
#84	print(f"Subscribing to wallet transactions ({config.wallet_address[:8]}...)...")
#85	await ws_client.subscribe_wallet_transactions(config.wallet_address)
#86	print("✓ Subscribed to wallet transactions\n")
#87
#88	print("=" * 60)
#89	print("📡 WebSocket Test Running - Listening for events...")
#90	print(" Press Ctrl+C to stop")
#91	print("=" * 60)
#92	print()
#93
#94	# Run for 60 seconds to test
#95	for i in range(60):
#96	await asyncio.sleep(1)
#97	if i % 10 == 0 and i > 0:
#98	print(f"\n⏱️ Running for {i} seconds...")
#99	print(f" Events received: {sum(event_counts.values())}")
#100	print(f" - Price updates: {event_counts['PRICE_UPDATE']}")
#101	print(f" - New listings: {event_counts['NEW_LISTING']}")
#102	print(f" - Large trades: {event_counts['LARGE_TRADE']}")
#103	print(f" - Wallet TXs: {event_counts['WALLET_TX']}\n")
#104
#105	print("\n" + "=" * 60)
#106	print("✅ Test Complete!")
#107	print(f"Total events received: {sum(event_counts.values())}")
#108	print(f"- Price updates: {event_counts['PRICE_UPDATE']}")
#109	print(f"- New listings: {event_counts['NEW_LISTING']}")
#110	print(f"- Large trades: {event_counts['LARGE_TRADE']}")
#111	print(f"- Wallet TXs: {event_counts['WALLET_TX']}")
#112	print("=" * 60)
#113
#114	except KeyboardInterrupt:
#115	print("\n\n⚠️ Test interrupted by user")
#116	except Exception as e:
#117	print(f"\n❌ Error during test: {e}")
#118	import traceback
#119	traceback.print_exc()
#120	finally:
#121	# Disconnect
#122	print("\nDisconnecting...")
#123	await ws_client.disconnect()
#124	print("✓ Disconnected\n")
#125	print("Test finished!")
#126
#127
#128	if __name__ == "__main__":
#129	asyncio.run(test_websocket())
#130

z6Mkq5mY3JWtxoxUobWcfNHm7AkRubgSWEZTkBVqZXJviFZ5/my-project-public