my-project-public

repository

loading code, commits, and activity

repositories

loading repo index

#1	"""Unified Solana Blockchain Analyzer - Detects and analyzes contracts, wallets, and transactions"""
#2
#3	import re
#4	import httpx
#5	from typing import Optional, Dict, Any, Literal
#6	from dataclasses import dataclass
#7	from enum import Enum
#8
#9
#10	class SolanaAddressType(Enum):
#11	"""Type of Solana address"""
#12	TOKEN_CONTRACT = "token_contract"
#13	WALLET = "wallet"
#14	TRANSACTION = "transaction"
#15	UNKNOWN = "unknown"
#16
#17
#18	@dataclass
#19	class SolanaAnalysisResult:
#20	"""Result of Solana address analysis"""
#21	address: str
#22	address_type: SolanaAddressType
#23	data: Dict[str, Any]
#24	source: str # Which API provided the data
#25
#26	def to_dict(self) -> Dict[str, Any]:
#27	return {
#28	"address": self.address,
#29	"type": self.address_type.value,
#30	"data": self.data,
#31	"source": self.source,
#32	}
#33
#34
#35	class SolanaAnalyzer:
#36	"""
#37	Unified analyzer for Solana blockchain addresses.
#38	Automatically detects whether an address is a token contract, wallet, or transaction.
#39	"""
#40
#41	def __init__(
#42	self,
#43	birdeye_api_key: str,
#44	helius_api_key: str,
#45	helius_rpc_url: str,
#46	):
#47	"""
#48	Initialize Solana analyzer.
#49
#50	Args:
#51	birdeye_api_key: Birdeye API key (for token data)
#52	helius_api_key: Helius API key (for wallet/transaction data)
#53	helius_rpc_url: Helius RPC URL
#54	"""
#55	self.birdeye_api_key = birdeye_api_key
#56	self.helius_api_key = helius_api_key
#57	self.helius_rpc_url = helius_rpc_url
#58
#59	self._client = httpx.AsyncClient(timeout=30.0)
#60
#61	def detect_address_type(self, address: str) -> SolanaAddressType:
#62	"""
#63	Detect the type of Solana address.
#64
#65	Solana addresses characteristics:
#66	- Token contracts: 32-44 chars, base58 encoded, typically start with specific patterns
#67	- Wallets: Same format as token contracts (need to check if it's a program/token vs wallet)
#68	- Transactions: 88 chars (base58 encoded signature)
#69
#70	Args:
#71	address: The Solana address to analyze
#72
#73	Returns:
#74	SolanaAddressType enum value
#75	"""
#76	# Remove whitespace
#77	address = address.strip()
#78
#79	# Transaction signatures are 88 characters (base58 encoded)
#80	if len(address) == 88:
#81	return SolanaAddressType.TRANSACTION
#82
#83	# Solana addresses are 32-44 characters
#84	if 32 <= len(address) <= 44:
#85	# Check if valid base58
#86	base58_pattern = r'^[1-9A-HJ-NP-Za-km-z]+$'
#87	if re.match(base58_pattern, address):
#88	# We'll need to check via API whether it's a token or wallet
#89	# For now, return WALLET and let the analyze method determine
#90	return SolanaAddressType.WALLET
#91
#92	return SolanaAddressType.UNKNOWN
#93
#94	async def analyze(self, address: str) -> SolanaAnalysisResult:
#95	"""
#96	Analyze a Solana address and return comprehensive data.
#97
#98	This method:
#99	1. Detects the type of address
#100	2. Routes to the appropriate API
#101	3. Returns formatted analysis
#102
#103	Args:
#104	address: Solana address (contract, wallet, or transaction)
#105
#106	Returns:
#107	SolanaAnalysisResult with complete analysis
#108	"""
#109	address_type = self.detect_address_type(address)
#110
#111	if address_type == SolanaAddressType.TRANSACTION:
#112	return await self._analyze_transaction(address)
#113	elif address_type in [SolanaAddressType.WALLET, SolanaAddressType.TOKEN_CONTRACT]:
#114	# Try token first, then wallet
#115	try:
#116	return await self._analyze_token(address)
#117	except Exception:
#118	# If not a token, analyze as wallet
#119	return await self._analyze_wallet(address)
#120	else:
#121	return SolanaAnalysisResult(
#122	address=address,
#123	address_type=SolanaAddressType.UNKNOWN,
#124	data={"error": "Invalid Solana address format"},
#125	source="none",
#126	)
#127
#128	async def _analyze_token(self, address: str) -> SolanaAnalysisResult:
#129	"""Analyze a token contract using Birdeye API"""
#130
#131	# Get token overview
#132	headers = {"X-API-KEY": self.birdeye_api_key}
#133
#134	response = await self._client.get(
#135	"https://public-api.birdeye.so/defi/token_overview",
#136	params={"address": address},
#137	headers=headers,
#138	)
#139	response.raise_for_status()
#140	token_data = response.json()
#141
#142	if not token_data.get("success"):
#143	raise Exception("Not a valid token address")
#144
#145	data = token_data.get("data", {})
#146
#147	# Get security analysis
#148	try:
#149	security_response = await self._client.get(
#150	"https://public-api.birdeye.so/defi/token_security",
#151	params={"address": address},
#152	headers=headers,
#153	)
#154	security_data = security_response.json().get("data", {})
#155	except Exception:
#156	security_data = {}
#157
#158	# Get OHLCV data for price chart (15m timeframe, 24h)
#159	try:
#160	ohlcv_response = await self._client.get(
#161	"https://public-api.birdeye.so/defi/ohlcv",
#162	params={
#163	"address": address,
#164	"type": "15m",
#165	"time_from": int((httpx.get("https://timeapi.io/api/Time/current/zone?timeZone=UTC").json()["timestamp"] - 86400)),
#166	"time_to": int(httpx.get("https://timeapi.io/api/Time/current/zone?timeZone=UTC").json()["timestamp"]),
#167	},
#168	headers=headers,
#169	)
#170	ohlcv_data = ohlcv_response.json().get("data", {}).get("items", [])
#171	except Exception:
#172	ohlcv_data = []
#173
#174	return SolanaAnalysisResult(
#175	address=address,
#176	address_type=SolanaAddressType.TOKEN_CONTRACT,
#177	data={
#178	"token_info": {
#179	"name": data.get("name"),
#180	"symbol": data.get("symbol"),
#181	"decimals": data.get("decimals"),
#182	"supply": data.get("supply"),
#183	},
#184	"market_data": {
#185	"price": data.get("price"),
#186	"price_change_24h": data.get("priceChange24hPercent"),
#187	"volume_24h": data.get("v24hUSD"),
#188	"volume_change_24h": data.get("v24hChangePercent"),
#189	"market_cap": data.get("mc"),
#190	"liquidity": data.get("liquidity"),
#191	},
#192	"security": security_data,
#193	"chart_data": {
#194	"timeframe": "15m",
#195	"period": "24h",
#196	"candles": ohlcv_data,
#197	},
#198	"holder_stats": {
#199	"holders": data.get("holder"),
#200	},
#201	},
#202	source="Birdeye API",
#203	)
#204
#205	async def _analyze_wallet(self, address: str) -> SolanaAnalysisResult:
#206	"""Analyze a wallet address using Helius DAS API"""
#207
#208	# Use Helius DAS API to get assets
#209	payload = {
#210	"jsonrpc": "2.0",
#211	"id": "wallet-analysis",
#212	"method": "getAssetsByOwner",
#213	"params": {
#214	"ownerAddress": address,
#215	"page": 1,
#216	"limit": 1000,
#217	},
#218	}
#219
#220	response = await self._client.post(
#221	self.helius_rpc_url,
#222	json=payload,
#223	headers={"Content-Type": "application/json"},
#224	)
#225	response.raise_for_status()
#226	assets_data = response.json()
#227
#228	# Get SOL balance
#229	balance_payload = {
#230	"jsonrpc": "2.0",
#231	"id": "balance-check",
#232	"method": "getBalance",
#233	"params": [address],
#234	}
#235
#236	balance_response = await self._client.post(
#237	self.helius_rpc_url,
#238	json=balance_payload,
#239	headers={"Content-Type": "application/json"},
#240	)
#241	balance_data = balance_response.json()
#242	sol_balance = balance_data.get("result", {}).get("value", 0) / 1e9
#243
#244	# Parse assets
#245	assets = assets_data.get("result", {}).get("items", [])
#246
#247	# Categorize assets
#248	tokens = []
#249	nfts = []
#250
#251	for asset in assets:
#252	if asset.get("interface") == "FungibleToken":
#253	tokens.append({
#254	"mint": asset.get("id"),
#255	"name": asset.get("content", {}).get("metadata", {}).get("name"),
#256	"symbol": asset.get("content", {}).get("metadata", {}).get("symbol"),
#257	"balance": asset.get("token_info", {}).get("balance", 0),
#258	"decimals": asset.get("token_info", {}).get("decimals", 0),
#259	})
#260	elif asset.get("interface") in ["NFT", "ProgrammableNFT"]:
#261	nfts.append({
#262	"mint": asset.get("id"),
#263	"name": asset.get("content", {}).get("metadata", {}).get("name"),
#264	"collection": asset.get("grouping", [{}])[0].get("group_value") if asset.get("grouping") else None,
#265	})
#266
#267	return SolanaAnalysisResult(
#268	address=address,
#269	address_type=SolanaAddressType.WALLET,
#270	data={
#271	"balance": {
#272	"sol": sol_balance,
#273	},
#274	"tokens": {
#275	"count": len(tokens),
#276	"holdings": tokens[:20], # Top 20
#277	},
#278	"nfts": {
#279	"count": len(nfts),
#280	"holdings": nfts[:20], # Top 20
#281	},
#282	"total_assets": len(assets),
#283	},
#284	source="Helius DAS API",
#285	)
#286
#287	async def _analyze_transaction(self, signature: str) -> SolanaAnalysisResult:
#288	"""Analyze a transaction using Helius API"""
#289
#290	# Use enhanced transaction API
#291	response = await self._client.get(
#292	f"{self.helius_rpc_url.replace('?', '/transactions/')}?api-key={self.helius_api_key}",
#293	params={"transactions": [signature]},
#294	)
#295
#296	if response.status_code != 200:
#297	# Fall back to standard RPC
#298	payload = {
#299	"jsonrpc": "2.0",
#300	"id": "tx-analysis",
#301	"method": "getTransaction",
#302	"params": [
#303	signature,
#304	{"encoding": "jsonParsed", "maxSupportedTransactionVersion": 0},
#305	],
#306	}
#307
#308	rpc_response = await self._client.post(
#309	self.helius_rpc_url,
#310	json=payload,
#311	headers={"Content-Type": "application/json"},
#312	)
#313	tx_data = rpc_response.json().get("result", {})
#314	else:
#315	tx_data = response.json()[0] if response.json() else {}
#316
#317	return SolanaAnalysisResult(
#318	address=signature,
#319	address_type=SolanaAddressType.TRANSACTION,
#320	data={
#321	"signature": signature,
#322	"slot": tx_data.get("slot"),
#323	"block_time": tx_data.get("blockTime"),
#324	"fee": tx_data.get("meta", {}).get("fee", 0) / 1e9, # Convert to SOL
#325	"status": "success" if not tx_data.get("meta", {}).get("err") else "failed",
#326	"error": tx_data.get("meta", {}).get("err"),
#327	"type": tx_data.get("type", "Unknown"),
#328	"description": tx_data.get("description", ""),
#329	"accounts_involved": len(tx_data.get("transaction", {}).get("message", {}).get("accountKeys", [])),
#330	},
#331	source="Helius API",
#332	)
#333
#334	async def close(self):
#335	"""Close the HTTP client"""
#336	await self._client.aclose()
#337
#338	async def __aenter__(self):
#339	return self
#340
#341	async def __aexit__(self, exc_type, exc_val, exc_tb):
#342	await self.close()
#343

z6Mkq5mY3JWtxoxUobWcfNHm7AkRubgSWEZTkBVqZXJviFZ5/my-project-public