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 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 self.solana_ws.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 self.solana_ws.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 self.solana_ws.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"Initial Wallet States (All balances in {DISPLAY_CURRENCY}):\n\n"
f"Followed Wallet ({FOLLOWED_WALLET}):\n"
f"{chr(10).join(followed_wallet_state)}\n"
f"Total Value: {FOLLOWED_WALLET_VALUE:.2f} {DISPLAY_CURRENCY}\n\n"
f"Your Wallet ({YOUR_WALLET}):\n"
f"{chr(10).join(your_wallet_state)}\n"
f"Total Value: {YOUR_WALLET_VALUE:.2f} {DISPLAY_CURRENCY}\n\n"
f"Monitored Tokens:\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()