747 lines
34 KiB
Python
747 lines
34 KiB
Python
import sys
|
|
import os
|
|
|
|
import aiohttp
|
|
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
import asyncio
|
|
import json
|
|
import logging
|
|
import random
|
|
import websockets
|
|
from typing import Dict, List, Optional
|
|
import requests
|
|
from datetime import datetime
|
|
from solana.rpc.types import TokenAccountOpts, TxOpts
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
SOLANA_ENDPOINTS = [
|
|
"wss://api.mainnet-beta.solana.com",
|
|
]
|
|
PING_INTERVAL = 30
|
|
SUBSCRIBE_INTERVAL = 1*60 # Resubscribe every 1 minute
|
|
|
|
from config import (
|
|
FOLLOWED_WALLET, SOLANA_HTTP_URL, DISPLAY_CURRENCY
|
|
)
|
|
|
|
from modules.utils import telegram_utils
|
|
|
|
class SolanaWS:
|
|
def __init__(self, on_message: Optional[callable] = None):
|
|
self.websocket = None
|
|
self.subscription_id = None
|
|
self.message_queue = asyncio.Queue()
|
|
self.on_message = on_message
|
|
self.websocket = None
|
|
|
|
async def connect(self):
|
|
while True:
|
|
try:
|
|
current_url = random.choice(SOLANA_ENDPOINTS)
|
|
self.websocket = await websockets.connect(current_url, ping_interval=30, ping_timeout=20)
|
|
logger.info(f"Connected to Solana websocket: {current_url}")
|
|
return
|
|
except Exception as e:
|
|
logger.error(f"Failed to connect to {current_url}: {e}")
|
|
await asyncio.sleep(5)
|
|
|
|
async def ws_jsonrpc(self, ws, method, params=None, doProcessResponse = True):
|
|
if not isinstance(params, list):
|
|
params = [params] if params is not None else []
|
|
|
|
request = {
|
|
"jsonrpc": "2.0",
|
|
"id": 1,
|
|
"method": method,
|
|
"params": params
|
|
}
|
|
|
|
await ws.send(json.dumps(request))
|
|
if not doProcessResponse:
|
|
return None
|
|
else:
|
|
response = await self.websocket.recv()
|
|
response_data = json.loads(response)
|
|
|
|
if 'result' in response_data:
|
|
return response_data['result']
|
|
elif 'error' in response_data:
|
|
logger.error(f"Error in WebSocket RPC call: {response_data['error']}")
|
|
return None
|
|
else:
|
|
logger.warning(f"Unexpected response: {response_data}")
|
|
return None
|
|
|
|
async def subscribe(self):
|
|
params = [
|
|
{"mentions": [FOLLOWED_WALLET]},
|
|
{"commitment": "confirmed"}
|
|
]
|
|
result = await self.ws_jsonrpc("logsSubscribe", params, doProcessResponse=False)
|
|
response = process_messages(self.websocket)
|
|
if result is not None:
|
|
self.subscription_id = result
|
|
logger.info(f"Subscription successful. Subscription id: {self.subscription_id}")
|
|
else:
|
|
logger.error("Failed to subscribe")
|
|
|
|
async def unsubscribe(self):
|
|
if self.subscription_id:
|
|
result = await self.ws_jsonrpc("logsUnsubscribe", [self.subscription_id])
|
|
if result:
|
|
logger.info(f"Unsubscribed from subscription id: {self.subscription_id}")
|
|
self.subscription_id = None
|
|
else:
|
|
logger.error(f"Failed to unsubscribe from subscription id: {self.subscription_id}")
|
|
|
|
async def receive_messages(self):
|
|
while True:
|
|
try:
|
|
message = await self.websocket.recv()
|
|
await self.message_queue.put(message)
|
|
except websockets.exceptions.ConnectionClosedError:
|
|
logger.error("WebSocket connection closed")
|
|
break
|
|
except Exception as e:
|
|
logger.error(f"Error receiving message: {e}")
|
|
break
|
|
|
|
async def process_messages(self):
|
|
while True:
|
|
message = await self.message_queue.get()
|
|
if self.on_message:
|
|
await self.on_message(message)
|
|
logger.info(f"Received message: {message}")
|
|
|
|
async def close(self):
|
|
if self.websocket:
|
|
await self.websocket.close()
|
|
logger.info("WebSocket connection closed")
|
|
|
|
async def solana_jsonrpc(method, params=None, jsonParsed=True):
|
|
if not isinstance(params, list):
|
|
params = [params] if params is not None else []
|
|
|
|
data = {
|
|
"jsonrpc": "2.0",
|
|
"id": 1,
|
|
"method": method,
|
|
"params": params
|
|
}
|
|
|
|
if jsonParsed:
|
|
data["params"].append({"encoding": "jsonParsed", "maxSupportedTransactionVersion": 0})
|
|
else:
|
|
data["params"].append({"maxSupportedTransactionVersion": 0})
|
|
|
|
try:
|
|
response = requests.post(SOLANA_HTTP_URL, headers={"Content-Type": "application/json"}, data=json.dumps(data))
|
|
response.raise_for_status()
|
|
result = response.json()
|
|
if 'result' not in result or 'error' in result:
|
|
logger.error("Error fetching data from Solana RPC:", result)
|
|
return None
|
|
return result['result']
|
|
except Exception as e:
|
|
logger.error(f"Error fetching data from Solana RPC: {e}")
|
|
return None
|
|
|
|
class SolanaAPI:
|
|
def __init__(self, process_transaction_callback, on_initial_subscription_callback = None, on_bot_message=None):
|
|
self.process_transaction = process_transaction_callback
|
|
self.on_initial_subscription = on_initial_subscription_callback
|
|
self.on_bot_message = on_bot_message,
|
|
|
|
self.dex = SolanaDEX(DISPLAY_CURRENCY)
|
|
self.solana_ws = SolanaWS(on_message=self.process_transaction)
|
|
|
|
async def process_messages(self, solana_ws):
|
|
while True:
|
|
message = await solana_ws.message_queue.get()
|
|
await self.process_transaction(message)
|
|
|
|
async def wallet_watch_loop(self):
|
|
|
|
solana_ws = SolanaWS(on_message=self.process_transaction)
|
|
first_subscription = True
|
|
|
|
while True:
|
|
try:
|
|
await solana_ws.connect()
|
|
await solana_ws.subscribe()
|
|
|
|
if first_subscription:
|
|
asyncio.create_task(self.on_initial_subscription())
|
|
first_subscription = False
|
|
|
|
await self.on_bot_message(f"Solana mainnet connected ({solana_ws.subscription_id})...")
|
|
|
|
receive_task = asyncio.create_task(solana_ws.receive_messages())
|
|
process_task = asyncio.create_task(solana_ws.process_messages())
|
|
|
|
try:
|
|
await asyncio.gather(receive_task, process_task)
|
|
except asyncio.CancelledError:
|
|
pass
|
|
finally:
|
|
receive_task.cancel()
|
|
process_task.cancel()
|
|
|
|
except Exception as e:
|
|
logger.error(f"An unexpected error occurred: {e}")
|
|
finally:
|
|
await solana_ws.unsubscribe()
|
|
if solana_ws.websocket:
|
|
await solana_ws.close()
|
|
await self.on_bot_message("Reconnecting...")
|
|
await asyncio.sleep(5)
|
|
|
|
async def process_transaction(self, signature):
|
|
print(f"Processing transaction: {signature['signature']}")
|
|
# Add your transaction processing logic here
|
|
|
|
async def get_last_transactions(self, account_address, check_interval=300, limit=1000):
|
|
last_check_time = None
|
|
last_signature = None
|
|
|
|
while True:
|
|
current_time = datetime.now()
|
|
|
|
if last_check_time is None or (current_time - last_check_time).total_seconds() >= check_interval:
|
|
params = [
|
|
account_address,
|
|
{
|
|
"limit": limit
|
|
}
|
|
]
|
|
|
|
if last_signature:
|
|
params[1]["before"] = last_signature
|
|
|
|
result = await solana_jsonrpc("getSignaturesForAddress", params)
|
|
|
|
if result:
|
|
for signature in result:
|
|
if last_signature and signature['signature'] == last_signature:
|
|
break
|
|
|
|
await self.process_transaction(signature)
|
|
|
|
if result:
|
|
last_signature = result[0]['signature']
|
|
|
|
last_check_time = current_time
|
|
|
|
await asyncio.sleep(1)
|
|
|
|
async def get_token_metadata_symbol(mint_address):
|
|
global TOKENS_INFO
|
|
|
|
if mint_address in TOKENS_INFO and 'symbol' in TOKENS_INFO[mint_address]:
|
|
return TOKENS_INFO[mint_address].get('symbol')
|
|
|
|
try:
|
|
account_data_result = await self.solana_ws.solana_jsonrpc("getAccountInfo", mint_address)
|
|
if 'value' in account_data_result and 'data' in account_data_result['value']:
|
|
account_data_data = account_data_result['value']['data']
|
|
if 'parsed' in account_data_data and 'info' in account_data_data['parsed']:
|
|
account_data_info = account_data_data['parsed']['info']
|
|
if 'decimals' in account_data_info:
|
|
if mint_address in TOKENS_INFO:
|
|
TOKENS_INFO[mint_address]['decimals'] = account_data_info['decimals']
|
|
else:
|
|
TOKENS_INFO[mint_address] = {'decimals': account_data_info['decimals']}
|
|
if 'tokenName' in account_data_info:
|
|
if mint_address in TOKENS_INFO:
|
|
TOKENS_INFO[mint_address]['name'] = account_data_info['tokenName']
|
|
else:
|
|
TOKENS_INFO[mint_address] = {'name': account_data_info['tokenName']}
|
|
|
|
metadata = await get_token_metadata(mint_address)
|
|
if metadata:
|
|
if mint_address in TOKENS_INFO:
|
|
TOKENS_INFO[mint_address].update(metadata)
|
|
else:
|
|
TOKENS_INFO[mint_address] = metadata
|
|
await save_token_info()
|
|
# TOKENS_INFO[mint_address] = metadata
|
|
# return metadata.get('symbol') or metadata.get('name')
|
|
return TOKENS_INFO[mint_address].get('symbol')
|
|
except Exception as e:
|
|
logging.error(f"Error fetching token name for {mint_address}: {str(e)}")
|
|
return None
|
|
|
|
async def get_transaction_details_rpc(tx_signature, readfromDump=False):
|
|
global FOLLOWED_WALLET_VALUE, YOUR_WALLET_VALUE, TOKEN_PRICES, TOKENS_INFO
|
|
try:
|
|
if readfromDump and os.path.exists('./logs/transation_details.json'):
|
|
with open('./logs/transation_details.json', 'r') as f: # trump_swap_tr_details
|
|
transaction_details = json.load(f)
|
|
return transaction_details
|
|
else:
|
|
transaction_details = await solana_jsonrpc("getTransaction", tx_signature)
|
|
with open('./logs/transation_details.json', 'w') as f:
|
|
json.dump(transaction_details, f, indent=2)
|
|
|
|
if transaction_details is None:
|
|
logging.error(f"Error fetching transaction details for {tx_signature}")
|
|
return None
|
|
|
|
# Initialize default result structure
|
|
parsed_result = {
|
|
"order_id": None,
|
|
"token_in": None,
|
|
"token_out": None,
|
|
"amount_in": 0,
|
|
"amount_out": 0,
|
|
"amount_in_USD": 0,
|
|
"amount_out_USD": 0,
|
|
"percentage_swapped": 0
|
|
}
|
|
|
|
# Extract order_id from logs
|
|
log_messages = transaction_details.get("meta", {}).get("logMessages", [])
|
|
for log in log_messages:
|
|
if "order_id" in log:
|
|
parsed_result["order_id"] = log.split(":")[2].strip()
|
|
break
|
|
|
|
# Extract token transfers from innerInstructions
|
|
inner_instructions = transaction_details.get('meta', {}).get('innerInstructions', [])
|
|
for instruction_set in inner_instructions:
|
|
for instruction in instruction_set.get('instructions', []):
|
|
if instruction.get('program') == 'spl-token' and instruction.get('parsed', {}).get('type') == 'transferChecked':
|
|
info = instruction['parsed']['info']
|
|
mint = info['mint']
|
|
amount = float(info['tokenAmount']['amount']) / 10 ** info['tokenAmount']['decimals'] # Adjust for decimals
|
|
|
|
# Determine which token is being swapped in and out based on zero balances
|
|
if parsed_result["token_in"] is None and amount > 0:
|
|
parsed_result["token_in"] = mint
|
|
parsed_result["amount_in"] = amount
|
|
|
|
|
|
if parsed_result["token_in"] is None or parsed_result["token_out"] is None:
|
|
# if we've failed to extract token_in and token_out from the transaction details, try a second method
|
|
inner_instructions = transaction_details.get('meta', {}).get('innerInstructions', [])
|
|
transfers = []
|
|
|
|
for instruction_set in inner_instructions:
|
|
for instruction in instruction_set.get('instructions', []):
|
|
if instruction.get('program') == 'spl-token' and instruction.get('parsed', {}).get('type') in ['transfer', 'transferChecked']:
|
|
info = instruction['parsed']['info']
|
|
amount = float(info['amount']) if 'amount' in info else float(info['tokenAmount']['amount'])
|
|
decimals = info['tokenAmount']['decimals'] if 'tokenAmount' in info else 0
|
|
adjusted_amount = amount / (10 ** decimals)
|
|
# adjusted_amount = float(info["amount"]) / (10 ** (info["tokenAmount"]["decimals"] if 'tokenAmount' in info else 0))
|
|
transfers.append({
|
|
'mint': info.get('mint'),
|
|
'amount': adjusted_amount,
|
|
'source': info['source'],
|
|
'destination': info['destination']
|
|
})
|
|
|
|
# Identify token_in and token_out
|
|
if len(transfers) >= 2:
|
|
parsed_result["token_in"] = transfers[0]['mint']
|
|
parsed_result["amount_in"] = transfers[0]['amount']
|
|
parsed_result["token_out"] = transfers[-1]['mint']
|
|
parsed_result["amount_out"] = transfers[-1]['amount']
|
|
|
|
# If mint is not provided, query the Solana network for the account data
|
|
if parsed_result["token_in"] is None or parsed_result["token_out"] is None:
|
|
#for transfer in transfers:
|
|
# do only first and last transfer
|
|
for transfer in [transfers[0], transfers[-1]]:
|
|
if transfer['mint'] is None:
|
|
# Query the Solana network for the account data
|
|
account_data_result = await solana_jsonrpc("getAccountInfo", transfer['source'])
|
|
|
|
if 'value' in account_data_result and 'data' in account_data_result['value']:
|
|
account_data_value = account_data_result['value']
|
|
account_data_data = account_data_value['data']
|
|
if 'parsed' in account_data_data and 'info' in account_data_data['parsed']:
|
|
account_data_info = account_data_data['parsed']['info']
|
|
if 'mint' in account_data_info:
|
|
transfer['mint'] = account_data_info['mint']
|
|
if transfer['mint'] in TOKENS_INFO or 'decimals' not in TOKENS_INFO[transfer['mint']]:
|
|
await get_token_metadata_symbol(transfer['mint'])
|
|
# get actual prices
|
|
current_price = await get_token_prices([transfer['mint']])
|
|
|
|
if parsed_result["token_in"] is None:
|
|
parsed_result["token_in"] = transfer['mint']
|
|
parsed_result["symbol_in"] = TOKENS_INFO[transfer['mint']]['symbol']
|
|
parsed_result["amount_in"] = transfer['amount']/10**TOKENS_INFO[transfer['mint']]['decimals']
|
|
parsed_result["amount_in_USD"] = parsed_result["amount_in"] * TOKENS_INFO[transfer['mint']].get('price', current_price[transfer['mint']])
|
|
elif parsed_result["token_out"] is None:
|
|
parsed_result["token_out"] = transfer['mint']
|
|
parsed_result["symbol_out"] = TOKENS_INFO[transfer['mint']]['symbol']
|
|
parsed_result["amount_out"] = transfer['amount']/10**TOKENS_INFO[transfer['mint']]['decimals']
|
|
parsed_result["amount_out_USD"] = parsed_result["amount_out"] * TOKENS_INFO[transfer['mint']]['price']
|
|
|
|
pre_balalnces = transaction_details.get('meta', {}).get('preTokenBalances', [])
|
|
for balance in pre_balalnces:
|
|
if balance['mint'] == parsed_result["token_in"] and balance['owner'] == FOLLOWED_WALLET:
|
|
parsed_result["before_source_balance"] = float(balance['uiTokenAmount']['amount']) / 10 ** balance['uiTokenAmount']['decimals']
|
|
break
|
|
|
|
|
|
# Calculate percentage swapped
|
|
try:
|
|
if parsed_result["amount_in"] > 0 and 'before_source_balance' in parsed_result and parsed_result["before_source_balance"] > 0:
|
|
parsed_result["percentage_swapped"] = (parsed_result["amount_in"] / parsed_result["before_source_balance"]) * 100
|
|
else:
|
|
# calculate based on total wallet value: FOLLOWED_WALLET_VALUE
|
|
parsed_result["percentage_swapped"] = (parsed_result["amount_in_USD"] / FOLLOWED_WALLET_VALUE) * 100
|
|
except Exception as e:
|
|
logging.error(f"Error calculating percentage swapped: {e}")
|
|
|
|
return parsed_result
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
print("Error fetching transaction details:", e)
|
|
|
|
|
|
|
|
|
|
|
|
class SolanaDEX:
|
|
def __init__(self, DISPLAY_CURRENCY):
|
|
self.DISPLAY_CURRENCY = DISPLAY_CURRENCY
|
|
pass
|
|
|
|
async def get_token_prices(token_addresses: List[str]) -> Dict[str, float]:
|
|
global TOKENS_INFO
|
|
|
|
# Skip for USD
|
|
prices = {addr: 1.0 for addr in token_addresses if addr == "EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v"}
|
|
remaining_tokens = [addr for addr in token_addresses if addr not in prices]
|
|
|
|
# Try CoinGecko
|
|
coingecko_prices = await self.get_prices_from_coingecko(remaining_tokens)
|
|
prices.update(coingecko_prices)
|
|
|
|
|
|
# For remaining missing tokens, try Jupiter
|
|
missing_tokens = set(remaining_tokens) - set(prices.keys())
|
|
if missing_tokens:
|
|
jupiter_prices = await get_prices_from_jupiter(list(missing_tokens))
|
|
prices.update(jupiter_prices)
|
|
|
|
|
|
# For tokens not found in CoinGecko, use DexScreener
|
|
missing_tokens = set(remaining_tokens) - set(coingecko_prices.keys())
|
|
if missing_tokens:
|
|
dexscreener_prices = await get_prices_from_dexscreener(list(missing_tokens))
|
|
prices.update(dexscreener_prices)
|
|
|
|
# For remaining missing tokens, try Raydium
|
|
missing_tokens = set(remaining_tokens) - set(prices.keys())
|
|
if missing_tokens:
|
|
raydium_prices = await get_prices_from_raydium(list(missing_tokens))
|
|
prices.update(raydium_prices)
|
|
|
|
# For remaining missing tokens, try Orca
|
|
missing_tokens = set(remaining_tokens) - set(prices.keys())
|
|
if missing_tokens:
|
|
orca_prices = await get_prices_from_orca(list(missing_tokens))
|
|
prices.update(orca_prices)
|
|
|
|
# If any tokens are still missing, set their prices to 0
|
|
for token in set(token_addresses) - set(prices.keys()):
|
|
prices[token] = 0.0
|
|
logging.warning(f"Price not found for token {token}. Setting to 0.")
|
|
|
|
for token, price in prices.items():
|
|
token_info = TOKENS_INFO.setdefault(token, {})
|
|
if 'symbol' not in token_info:
|
|
token_info['symbol'] = await get_token_metadata_symbol(token)
|
|
token_info['price'] = price
|
|
|
|
return prices
|
|
|
|
async def get_prices_from_coingecko(token_addresses: List[str]) -> Dict[str, float]:
|
|
base_url = "https://api.coingecko.com/api/v3/simple/token_price/solana"
|
|
prices = {}
|
|
|
|
async def fetch_single_price(session, address):
|
|
params = {
|
|
"contract_addresses": address,
|
|
"vs_currencies": DISPLAY_CURRENCY.lower()
|
|
}
|
|
try:
|
|
async with session.get(base_url, params=params) as response:
|
|
if response.status == 200:
|
|
data = await response.json()
|
|
if address in data and DISPLAY_CURRENCY.lower() in data[address]:
|
|
return address, data[address][DISPLAY_CURRENCY.lower()]
|
|
else:
|
|
logging.warning(f"Failed to get price for {address} from CoinGecko. Status: {response.status}")
|
|
except Exception as e:
|
|
logging.error(f"Error fetching price for {address} from CoinGecko: {str(e)}")
|
|
return address, None
|
|
|
|
async with aiohttp.ClientSession() as session:
|
|
tasks = [fetch_single_price(session, address) for address in token_addresses]
|
|
results = await asyncio.gather(*tasks)
|
|
|
|
for address, price in results:
|
|
if price is not None:
|
|
prices[address] = price
|
|
|
|
return prices
|
|
|
|
async def get_prices_from_dexscreener(token_addresses: List[str]) -> Dict[str, float]:
|
|
base_url = "https://api.dexscreener.com/latest/dex/tokens/"
|
|
prices = {}
|
|
|
|
try:
|
|
async with aiohttp.ClientSession() as session:
|
|
tasks = [fetch_token_data(session, f"{base_url}{address}") for address in token_addresses]
|
|
results = await asyncio.gather(*tasks)
|
|
|
|
for address, result in zip(token_addresses, results):
|
|
if result and 'pairs' in result and result['pairs']:
|
|
pair = result['pairs'][0] # Use the first pair (usually the most liquid)
|
|
prices[address] = float(pair['priceUsd'])
|
|
else:
|
|
logging.warning(f"No price data found on DexScreener for token {address}")
|
|
except Exception as e:
|
|
logging.error(f"Error fetching token prices from DexScreener: {str(e)}")
|
|
|
|
return prices
|
|
|
|
async def get_prices_from_jupiter(token_addresses: List[str]) -> Dict[str, float]:
|
|
url = "https://price.jup.ag/v4/price"
|
|
params = {
|
|
"ids": ",".join(token_addresses)
|
|
}
|
|
prices = {}
|
|
|
|
try:
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.get(url, params=params) as response:
|
|
if response.status == 200:
|
|
data = await response.json()
|
|
for address, price_info in data.get('data', {}).items():
|
|
if 'price' in price_info:
|
|
prices[address] = float(price_info['price'])
|
|
else:
|
|
logging.error(f"Failed to get token prices from Jupiter. Status: {response.status}")
|
|
except Exception as e:
|
|
logging.error(f"Error fetching token prices from Jupiter: {str(e)}")
|
|
return prices
|
|
|
|
# New function for Raydium
|
|
async def get_prices_from_raydium(token_addresses: List[str]) -> Dict[str, float]:
|
|
url = "https://api.raydium.io/v2/main/price"
|
|
prices = {}
|
|
|
|
try:
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.get(url) as response:
|
|
if response.status == 200:
|
|
data = await response.json()
|
|
for address in token_addresses:
|
|
if address in data:
|
|
prices[address] = float(data[address])
|
|
else:
|
|
logging.error(f"Failed to get token prices from Raydium. Status: {response.status}")
|
|
except Exception as e:
|
|
logging.error(f"Error fetching token prices from Raydium: {str(e)}")
|
|
return prices
|
|
|
|
# New function for Orca
|
|
async def get_prices_from_orca(token_addresses: List[str]) -> Dict[str, float]:
|
|
url = "https://api.orca.so/allTokens"
|
|
prices = {}
|
|
|
|
try:
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.get(url) as response:
|
|
if response.status == 200:
|
|
data = await response.json()
|
|
for token_info in data:
|
|
if token_info['mint'] in token_addresses:
|
|
prices[token_info['mint']] = float(token_info['price'])
|
|
else:
|
|
logging.error(f"Failed to get token prices from Orca. Status: {response.status}")
|
|
except Exception as e:
|
|
logging.error(f"Error fetching token prices from Orca: {str(e)}")
|
|
return prices
|
|
|
|
async def fetch_token_data(session, url):
|
|
try:
|
|
async with session.get(url) as response:
|
|
if response.status == 200:
|
|
return await response.json()
|
|
else:
|
|
logging.error(f"Failed to fetch data from {url}. Status: {response.status}")
|
|
return None
|
|
except Exception as e:
|
|
logging.error(f"Error fetching data from {url}: {str(e)}")
|
|
return None
|
|
|
|
async def get_sol_price() -> float:
|
|
sol_address = "So11111111111111111111111111111111111111112" # Solana's wrapped SOL address
|
|
return await get_token_prices([sol_address]).get(sol_address, 0.0)
|
|
|
|
async def get_wallet_balances(wallet_address, doGetTokenName=True):
|
|
balances = {}
|
|
logging.info(f"Getting balances for wallet: {wallet_address}")
|
|
global TOKENS_INFO
|
|
try:
|
|
response = await solana_client.get_token_accounts_by_owner_json_parsed(
|
|
Pubkey.from_string(wallet_address),
|
|
opts=TokenAccountOpts(
|
|
program_id=Pubkey.from_string("TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA")
|
|
),
|
|
commitment=Confirmed
|
|
)
|
|
|
|
if response.value:
|
|
for account in response.value:
|
|
try:
|
|
parsed_data = account.account.data.parsed
|
|
if isinstance(parsed_data, dict) and 'info' in parsed_data:
|
|
info = parsed_data['info']
|
|
if isinstance(info, dict) and 'mint' in info and 'tokenAmount' in info:
|
|
mint = info['mint']
|
|
decimals = info['tokenAmount']['decimals']
|
|
amount = float(info['tokenAmount']['amount'])/10**decimals
|
|
if amount > 0:
|
|
if mint in TOKENS_INFO:
|
|
token_name = TOKENS_INFO[mint].get('symbol')
|
|
elif doGetTokenName:
|
|
token_name = await get_token_metadata_symbol(mint) or 'N/A'
|
|
# sleep for 1 second to avoid rate limiting
|
|
await asyncio.sleep(2)
|
|
|
|
TOKENS_INFO[mint]['holdedAmount'] = round(amount,decimals)
|
|
TOKENS_INFO[mint]['decimals'] = decimals
|
|
balances[mint] = {
|
|
'name': token_name or 'N/A',
|
|
'address': mint,
|
|
'amount': amount,
|
|
'decimals': decimals
|
|
}
|
|
# sleep for 1 second to avoid rate limiting
|
|
logging.debug(f"Account balance for {token_name} ({mint}): {amount}")
|
|
else:
|
|
logging.warning(f"Unexpected data format for account: {account}")
|
|
except Exception as e:
|
|
logging.error(f"Error parsing account data: {str(e)}")
|
|
|
|
sol_balance = await solana_client.get_balance(Pubkey.from_string(wallet_address))
|
|
if sol_balance.value is not None:
|
|
balances['SOL'] = {
|
|
'name': 'SOL',
|
|
'address': 'SOL',
|
|
'amount': sol_balance.value / 1e9
|
|
}
|
|
else:
|
|
logging.warning(f"SOL balance response missing for wallet: {wallet_address}")
|
|
|
|
except Exception as e:
|
|
logging.error(f"Error getting wallet balances: {str(e)}")
|
|
logging.info(f"Found {len(response.value)} ({len(balances)} non zero) token accounts for wallet: {wallet_address}")
|
|
return balances
|
|
|
|
async def convert_balances_to_currency(balances , sol_price):
|
|
converted_balances = {}
|
|
for address, info in balances.items():
|
|
converted_balance = info.copy() # Create a copy of the original info
|
|
if info['name'] == 'SOL':
|
|
converted_balance['value'] = info['amount'] * sol_price
|
|
elif address in TOKEN_PRICES:
|
|
converted_balance['value'] = info['amount'] * TOKEN_PRICES[address]
|
|
else:
|
|
converted_balance['value'] = None # Price not available
|
|
logging.warning(f"Price not available for token {info['name']} ({address})")
|
|
converted_balances[address] = converted_balance
|
|
return converted_balances
|
|
|
|
|
|
async def list_initial_wallet_states():
|
|
global TOKEN_ADDRESSES, FOLLOWED_WALLET_VALUE, YOUR_WALLET_VALUE, TOKEN_PRICES
|
|
global TOKENS_INFO # new
|
|
|
|
followed_wallet_balances = await get_wallet_balances(FOLLOWED_WALLET)
|
|
your_wallet_balances = await get_wallet_balances(YOUR_WALLET)
|
|
|
|
all_token_addresses = list(set(followed_wallet_balances.keys()) |
|
|
set(your_wallet_balances.keys()) |
|
|
set(TOKEN_ADDRESSES.values()))
|
|
|
|
TOKEN_PRICES = await get_token_prices(all_token_addresses)
|
|
sol_price = await get_sol_price()
|
|
|
|
followed_converted_balances = await convert_balances_to_currency(followed_wallet_balances, sol_price)
|
|
your_converted_balances = await convert_balances_to_currency(your_wallet_balances, sol_price)
|
|
|
|
|
|
TOKEN_ADDRESSES = {
|
|
address: info for address,
|
|
info in {**followed_converted_balances, **your_converted_balances}.items() if info['value'] is not None and info['value'] > 0
|
|
}
|
|
logging.info(f"Monitoring balances for tokens: {[info['name'] for info in TOKEN_ADDRESSES.values()]}")
|
|
|
|
followed_wallet_state = []
|
|
FOLLOWED_WALLET_VALUE = 0
|
|
for address, info in followed_converted_balances.items():
|
|
if info['value'] is not None and info['value'] > 0:
|
|
followed_wallet_state.append(f"{info['name']}: {info['value']:.2f} {DISPLAY_CURRENCY} ({info['address']})")
|
|
FOLLOWED_WALLET_VALUE += info['value']
|
|
|
|
your_wallet_state = []
|
|
YOUR_WALLET_VALUE = 0
|
|
for address, info in your_converted_balances.items():
|
|
if info['value'] is not None and info['value'] > 0:
|
|
your_wallet_state.append(f"{info['name']}: {info['value']:.2f} {DISPLAY_CURRENCY}")
|
|
YOUR_WALLET_VALUE += info['value']
|
|
|
|
message = (
|
|
f"<b>Initial Wallet States (All balances in {DISPLAY_CURRENCY}):</b>\n\n"
|
|
f"<b>Followed Wallet ({FOLLOWED_WALLET}):</b>\n"
|
|
f"{chr(10).join(followed_wallet_state)}\n"
|
|
f"<b>Total Value:</b> {FOLLOWED_WALLET_VALUE:.2f} {DISPLAY_CURRENCY}\n\n"
|
|
f"<b>Your Wallet ({YOUR_WALLET}):</b>\n"
|
|
f"{chr(10).join(your_wallet_state)}\n"
|
|
f"<b>Total Value:</b> {YOUR_WALLET_VALUE:.2f} {DISPLAY_CURRENCY}\n\n"
|
|
f"<b>Monitored Tokens:</b>\n"
|
|
f"{', '.join([safe_get_property(info, 'name') for info in TOKEN_ADDRESSES.values()])}"
|
|
)
|
|
|
|
logging.info(message)
|
|
await telegram_utils.send_telegram_message(message)
|
|
|
|
# save token info to file
|
|
await save_token_info()
|
|
|
|
|
|
|
|
#example
|
|
# async def main():
|
|
# await telegram_utils.initialize()
|
|
|
|
# async def process_log(log):
|
|
# print(f"Processing log: {log}")
|
|
|
|
# async def list_initial_wallet_states():
|
|
# print("Listing initial wallet states")
|
|
|
|
|
|
# wallet_watch_task = asyncio.create_task(solana_api.wallet_watch_loop())
|
|
|
|
# try:
|
|
# await asyncio.gather(wallet_watch_task)
|
|
# except asyncio.CancelledError:
|
|
# pass
|
|
# finally:
|
|
# await telegram_utils.close()
|
|
|
|
# if __name__ == "__main__":
|
|
# asyncio.run(main()) |