popov/gogo2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

get transactionInfo RPC call works

Browse Source
This commit is contained in:
Dobromir Popov
2024-10-04 18:12:52 +03:00
parent c952edb363
commit 2d92dc6340
3 changed files with 279 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

117
crypto/sol/app.py
View File

@ -5,7 +5,7 @@ from flask import Flask, render_template, request, jsonify
from solana.rpc.async_api import AsyncClient
from solana.transaction import Signature
from solana.rpc.websocket_api import connect
from solana.rpc.types import TokenAccountOpts
from solana.rpc.types import TokenAccountOpts, TxOpts
from solana.rpc.commitment import Confirmed
from solders.pubkey import Pubkey
from dexscreener import DexscreenerClient
@ -56,19 +56,22 @@ def retry_last_log():
# Use the production Solana RPC endpoint
solana_client = AsyncClient("https://api.mainnet-beta.solana.com")
dexscreener_client = DexscreenerClient()
# Configuration
DEVELOPER_CHAT_ID = os.getenv("DEVELOPER_CHAT_ID")
FOLLOWED_WALLET = os.getenv("FOLLOWED_WALLET")
YOUR_WALLET = os.getenv("YOUR_WALLET")
TELEGRAM_BOT_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN")
SOLANA_URL = os.getenv("SOLANA_NET_URL")
SOLANA_WS_URL = os.getenv("SOLANA_WS_URL")
SOLANA_HTTP_URL = os.getenv("SOLANA_HTTP_URL")
DISPLAY_CURRENCY = os.getenv('DISPLAY_CURRENCY', 'USD')
# Use the production Solana RPC endpoint
solana_client = AsyncClient(SOLANA_HTTP_URL)
dexscreener_client = DexscreenerClient()
# Initialize Telegram Bot
bot = Bot(token=TELEGRAM_BOT_TOKEN)
@ -473,13 +476,51 @@ def perform_swap(input_token, output_token, amount):
"success": False,
"error": str(e)
}
from base58 import b58decode
from solders.pubkey import Pubkey
from solders.transaction import Transaction
from solders.signature import Signature
async def get_transaction_details_rpc(tx_signature):
url = SOLANA_HTTP_URL
# url = 'https://solana.drpc.org'
headers = {"Content-Type": "application/json"}
data = {
"jsonrpc": "2.0",
"id": 1,
"method": "getTransaction",
"params": [
tx_signature,
{
"encoding": "jsonParsed",
"maxSupportedTransactionVersion": 0
}
]
}
# request = {
# "jsonrpc": "2.0",
# "id": 1,
# "method": "getConfirmedTransaction",
# "params": [
# tx_signature,
# "json"
# ]
# }
try:
response = requests.post(url, headers=headers, data=json.dumps(data))
response.raise_for_status() # Raises an error for bad responses
transaction_details = response.json()
if 'result' in transaction_details:
print(transaction_details['result'])
return transaction_details['result']
else:
print("Unexpected response:", transaction_details)
except requests.exceptions.RequestException as e:
print("Error fetching transaction details:", e)
async def save_log(log):
try:
os.makedirs('./logs', exist_ok=True)
@ -491,6 +532,7 @@ async def save_log(log):
except Exception as e:
logging.error(f"Error saving RPC log: {e}")
async def process_log(log_result):
if log_result['value']['err']:
return
@ -499,32 +541,42 @@ async def process_log(log_result):
logs = log_result['value']['logs']
try:
try:
transaction = await get_transaction_details_rpc(tx_signature_str)
except Exception as e:
logging.error(f"Error fetching transaction details: {e}")
return
# Convert the base58 signature string to bytes
tx_signature = Signature(b58decode(tx_signature_str))
# Fetch transaction details
tx_result = await solana_client.get_transaction(tx_signature)
if tx_result and tx_result.value:
transaction = Transaction.from_json(tx_result.value)
message = transaction.message
tx_result = await solana_client.get_transaction(tx_signature, max_supported_transaction_version=0)
#tx_result = await get_transaction_details(tx_signature_str)
if tx_result.value is None:
logging.error(f"Transaction not found: {tx_signature_str}")
return
transaction = tx_result.value.transaction
for log_entry in logs:
if 'Program log: Instruction: Swap' in log_entry:
for instruction in message.instructions:
if instruction.program_id == TOKEN_ADDRESSES['SOL']:
from_pubkey = instruction.accounts[0]
to_pubkey = instruction.accounts[1]
amount = int(instruction.data, 16) / 1e9
for log_entry in logs:
if 'Program log: Instruction: Swap' in log_entry:
for instruction in message.instructions:
if instruction.program_id == TOKEN_ADDRESSES['SOL']:
from_pubkey = instruction.accounts[0]
to_pubkey = instruction.accounts[1]
amount = int(instruction.data, 16) / 1e9
if from_pubkey == FOLLOWED_WALLET:
move = {
'token': 'SOL',
'amount': amount,
'to_token': 'Unknown'
}
message_text = f"Swap detected:\nFrom: {from_pubkey}\nTo: {to_pubkey}\nAmount: {amount} SOL"
await send_telegram_message(message_text)
await follow_move(move)
if from_pubkey == FOLLOWED_WALLET:
move = {
'token': 'SOL',
'amount': amount,
'to_token': 'Unknown'
}
message_text = f"Swap detected:\nFrom: {from_pubkey}\nTo: {to_pubkey}\nAmount: {amount} SOL"
await send_telegram_message(message_text)
await follow_move(move)
except Exception as e:
logging.error(f"Error processing log: {e}")
@ -541,7 +593,7 @@ async def subscribe_to_wallet():
"wss://rpc.ankr.com/solana",
"wss://mainnet.rpcpool.com",
]
uri = SOLANA_URL # wss://api.mainnet-beta.solana.com
uri = SOLANA_WS_URL # wss://api.mainnet-beta.solana.com
reconnect_delay = 5 # Start with a 5-second delay
max_reconnect_delay = 60 # Maximum delay of 60 seconds
@ -606,14 +658,17 @@ async def subscribe_to_wallet():
# Implement exponential backoff
reconnect_delay = min(reconnect_delay * 2, max_reconnect_delay)
logger = logging.getLogger(__name__)
logger = logging.getLogger(__name__)
async def main():
# Initialize logging
logging.basicConfig(level=logging.DEBUG)
# logging.basicConfig(level=logging.INFO)
await send_telegram_message("Solana Agent Started. Connecting to mainnet...")
await subscribe_to_wallet()
#await subscribe_to_wallet()
if __name__ == '__main__':