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

actively listening to followed wallet (wip)

Browse Source
This commit is contained in:
Dobromir Popov 2024-10-02 16:21:53 +03:00
parent 990903db30
commit a7fdd033e9
3 changed files with 116 additions and 90 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

196
crypto/sol/app.py
View File

@ -8,6 +8,13 @@ from telegram import Bot
from telegram.constants import ParseMode
import datetime
import logging
from solana.rpc.websocket_api import connect
from solana.rpc.async_api import AsyncClient
from solana.rpc.commitment import Confirmed
import websockets
import json
app = Flask(__name__)
@ -17,7 +24,7 @@ dexscreener_client = DexscreenerClient()
# Configuration
DEVELOPER_CHAT_ID = "777826553"
FOLLOWED_WALLET = "9U7D916zuQ8qcL9kQZqkcroWhHGho5vD8VNekvztrutN" # traderrobot
FOLLOWED_WALLET = "9U7D916zuQ8qcL9kQZqkcroWhHGho5vD8VNekvztrutN" # traderrobot
YOUR_WALLET = "65nzyZXTLC81MthTo52a2gRJjqryTizWVqpK2fDKLye5"
TELEGRAM_BOT_TOKEN = "6805059978:AAHNJKuOeazMSJHc3-BXRCsFfEVyFHeFnjw"
@ -32,82 +39,33 @@ TOKEN_ADDRESSES = {
"SOL": "So11111111111111111111111111111111111111112",
"USDC": "EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v",
"TARD": "4nfn86ssbv7wiqcsw7bpvn46k24jhe334fudtyxhp1og"
# Add more known tokens here
}
async def send_telegram_message(message):
try:
await bot.send_message(chat_id=DEVELOPER_CHAT_ID, text=message, parse_mode=ParseMode.HTML)
logging.info(f"Telegram message sent: {message}")
except Exception as e:
logging.error(f"Error sending Telegram message: {str(e)}")
@app.route('/')
def index():
return render_template('index.html')
@app.route('/tokens', methods=['GET'])
async def get_tokens():
balances = await get_wallet_balances(YOUR_WALLET)
return jsonify(list(balances.keys()))
@app.route('/balances', methods=['GET'])
async def get_balances():
balances = await get_wallet_balances(YOUR_WALLET)
return jsonify(balances)
@app.route('/follow_move', methods=['POST'])
async def follow_move():
move = request.json
followed_balances = await get_wallet_balances(FOLLOWED_WALLET)
your_balances = await get_wallet_balances(YOUR_WALLET)
if move['token'] not in followed_balances or move['token'] not in your_balances:
return jsonify({'status': 'error', 'message': 'Invalid token'})
followed_balance = followed_balances[move['token']]
your_balance = your_balances[move['token']]
proportion = your_balance / followed_balance if followed_balance > 0 else 0
amount_to_swap = move['amount'] * proportion
if your_balance >= amount_to_swap:
# Implement actual swap logic here
pair = dexscreener_client.get_token_pair("solana", move['token'])
price = float(pair['priceUsd'])
received_amount = amount_to_swap * price
await send_telegram_message(
f"<b>Move Followed:</b>\n"
f"Swapped {amount_to_swap:.6f} {move['token']} "
f"for {received_amount:.6f} {move['to_token']}"
)
return jsonify({
'status': 'success',
'message': f"Swapped {amount_to_swap:.6f} {move['token']} for {received_amount:.6f} {move['to_token']}"
})
else:
await send_telegram_message(
f"<b>Move Failed:</b>\n"
f"Insufficient balance to swap {amount_to_swap:.6f} {move['token']}"
)
return jsonify({
'status': 'error',
'message': f"Insufficient balance to swap {amount_to_swap:.6f} {move['token']}"
})
async def get_token_balance(wallet_address, token_address):
try:
balance = await solana_client.get_token_account_balance(
Pubkey.from_string(token_address),
commitment=Confirmed
response = await solana_client.get_token_accounts_by_owner(
Pubkey.from_string(wallet_address),
{"mint": Pubkey.from_string(token_address)}
)
logging.debug(f"Balance for {token_address}: {balance}")
return float(balance['result']['value']['uiAmount'])
if response['result']['value']:
balance = await solana_client.get_token_account_balance(
response['result']['value'][0]['pubkey']
)
amount = float(balance['result']['value']['uiAmount'])
logging.debug(f"Balance for {token_address} in {wallet_address}: {amount}")
return amount
else:
logging.debug(f"No account found for {token_address} in {wallet_address}")
return 0
except Exception as e:
logging.error(f"Error getting balance for {token_address}: {str(e)}")
logging.error(f"Error getting balance for {token_address} in {wallet_address}: {str(e)}")
return 0
async def get_wallet_balances(wallet_address):
@ -139,9 +97,7 @@ async def list_initial_wallet_states():
your_non_zero = await get_non_zero_token_balances(YOUR_WALLET)
TOKEN_ADDRESSES = {**followed_non_zero, **your_non_zero}
# TOKEN_ADDRESSES.update(followed_non_zero)
# TOKEN_ADDRESSES.update(your_non_zero)
logging.info(f"Getting balances for tokens : {TOKEN_ADDRESSES.keys()}")
logging.info(f"Monitoring balances for tokens: {TOKEN_ADDRESSES.keys()}")
followed_wallet_state = "\n".join([f"{token}: {amount:.6f}" for token, amount in followed_wallet_balances.items() if amount > 0])
your_wallet_state = "\n".join([f"{token}: {amount:.6f}" for token, amount in your_wallet_balances.items() if amount > 0])
@ -156,26 +112,94 @@ async def list_initial_wallet_states():
f"{', '.join(TOKEN_ADDRESSES.keys())}"
)
logging.info(message)
await send_telegram_message(message)
# @app.route('/initialize', methods=['GET'])
# async def initialize():
# await list_initial_wallet_states()
# return jsonify({"status": "success", "message": "Initialization complete"})
async def follow_move(move):
followed_balances = await get_wallet_balances(FOLLOWED_WALLET)
your_balances = await get_wallet_balances(YOUR_WALLET)
if move['token'] not in followed_balances or move['token'] not in your_balances:
logging.error(f"Invalid token: {move['token']}")
return
async def send_startup_message():
current_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
message = (
f"<b>Solana Agent Application Started</b>\n\n"
f"Startup Time: {current_time}\n"
f"Followed Wallet: {FOLLOWED_WALLET}\n"
f"Your Wallet: {YOUR_WALLET}\n\n"
"The application is now running and ready to monitor wallet activities."
)
await send_telegram_message(message)
followed_balance = followed_balances[move['token']]
your_balance = your_balances[move['token']]
proportion = your_balance / followed_balance if followed_balance > 0 else 0
amount_to_swap = move['amount'] * proportion
if your_balance >= amount_to_swap:
# Implement actual swap logic here
pair = dexscreener_client.get_token_pair("solana", move['token'])
price = float(pair['priceUsd'])
received_amount = amount_to_swap * price
message = (
f"<b>Move Followed:</b>\n"
f"Swapped {amount_to_swap:.6f} {move['token']} "
f"for {received_amount:.6f} {move['to_token']}"
)
logging.info(message)
await send_telegram_message(message)
else:
message = (
f"<b>Move Failed:</b>\n"
f"Insufficient balance to swap {amount_to_swap:.6f} {move['token']}"
)
logging.warning(message)
await send_telegram_message(message)
async def on_logs(logs):
print(f"Received logs: {logs}")
if logs.value.err:
return
tx = logs.value.signature
tx_result = await solana_client.get_transaction(tx)
if tx_result and tx_result.value:
for instruction in tx_result.value.transaction.message.instructions:
if instruction.program_id == Pubkey.from_string(TOKEN_ADDRESSES['SOL']):
# This is a token transfer
from_pubkey = instruction.accounts[0]
to_pubkey = instruction.accounts[1]
amount = instruction.data[8:] # The amount is stored in the last 8 bytes
amount = int.from_bytes(amount, 'little') / 1e9 # Convert lamports to SOL
if from_pubkey == Pubkey.from_string(FOLLOWED_WALLET):
move = {
'token': 'SOL',
'amount': amount,
'to_token': 'Unknown' # You might want to determine this based on the receiving address
}
await follow_move(move)
async def subscribe_to_wallet():
uri = "wss://api.mainnet-beta.solana.com"
async with websockets.connect(uri) as websocket:
# Correct the `params` format to be an array
request = {
"jsonrpc": "2.0",
"id": 1,
"method": "logsSubscribe",
"params": [
{"mentions": ["YOUR_WALLET_ADDRESS"], "commitment": "confirmed"}
]
}
await websocket.send(json.dumps(request))
# Listen for messages
while True:
response = await websocket.recv()
print(response)
async def main():
logging.basicConfig(level=logging.INFO)
await send_telegram_message("Solana Agent Application Started")
await list_initial_wallet_states()
await subscribe_to_wallet()
if __name__ == '__main__':
# asyncio.run(send_startup_message())
asyncio.run(list_initial_wallet_states())
print(f"Initial Monitored Tokens: {', '.join(TOKEN_ADDRESSES.keys())}")
app.run(host='0.0.0.0', port=3009)
asyncio.run(main())

8
crypto/sol/r.txt
View File

@ -1,4 +1,6 @@
flask
Flask
solana
idna
httpx
dexscreener
python-telegram-bot
asyncio
base58

2
crypto/sol/readme.md
View File

@ -3,7 +3,7 @@ To run this Python Solana agent:
Install the required libraries:
`pip install flask solana dexscreener python-telegram-bot`
`pip install flask solana dexscreener python-telegram-bot asyncio base58`
Replace REPLACE_WITH_WALLET_ADDRESS with the wallet address you want to follow.
Replace REPLACE_WITH_YOUR_WALLET_ADDRESS with your own wallet address.