127 lines
4.0 KiB
Python
127 lines
4.0 KiB
Python
from flask import Flask, render_template, request, jsonify
|
|
from solana.rpc.api import Client
|
|
from dexscreener import DexscreenerClient
|
|
import asyncio
|
|
from telegram import Bot
|
|
from telegram.constants import ParseMode
|
|
import os
|
|
|
|
app = Flask(__name__)
|
|
solana_client = Client("https://api.mainnet-beta.solana.com")
|
|
dexscreener_client = DexscreenerClient()
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# This can be your own ID, or one for a developer group/channel.
|
|
# You can use the /start command of this bot to see your chat id.
|
|
DEVELOPER_CHAT_ID = "777826553"
|
|
|
|
|
|
# Replace with the wallet address you want to follow
|
|
FOLLOWED_WALLET = "9U7D916zuQ8qcL9kQZqkcroWhHGho5vD8VNekvztrutN"
|
|
|
|
# Replace with your wallet address
|
|
YOUR_WALLET = "65nzyZXTLC81MthTo52a2gRJjqryTizWVqpK2fDKLye5"
|
|
|
|
|
|
|
|
TELEGRAM_CHAT_ID = "777826553"
|
|
# Replace with your Telegram Bot Token
|
|
TELEGRAM_BOT_TOKEN = "6749075936:AAHUHiPTDEIu6JH7S2fQdibwsu6JVG3FNG0"
|
|
|
|
# Telegram Bot Token
|
|
# t.me/kevin_ai_robot
|
|
# TOKEN = '6805059978:AAHNJKuOeazMSJHc3-BXRCsFfEVyFHeFnjw'
|
|
# t.me/artitherobot 6749075936:AAHUHiPTDEIu6JH7S2fQdibwsu6JVG3FNG0
|
|
|
|
# Initialize Telegram Bot
|
|
bot = Bot(token=TELEGRAM_BOT_TOKEN)
|
|
|
|
# Simulated wallet balances (replace with actual balance fetching logic)
|
|
wallet_balances = {
|
|
FOLLOWED_WALLET: {"SOL": 100, "USDC": 1000},
|
|
YOUR_WALLET: {"SOL": 10, "USDC": 100}
|
|
}
|
|
|
|
async def send_telegram_message(message):
|
|
await bot.send_message(chat_id=TELEGRAM_CHAT_ID, text=message, parse_mode=ParseMode.HTML)
|
|
|
|
@app.route('/')
|
|
def index():
|
|
return render_template('index.html')
|
|
|
|
@app.route('/tokens', methods=['GET'])
|
|
def get_tokens():
|
|
return jsonify(list(wallet_balances[YOUR_WALLET].keys()))
|
|
|
|
@app.route('/balances', methods=['GET'])
|
|
def get_balances():
|
|
return jsonify(wallet_balances[YOUR_WALLET])
|
|
|
|
@app.route('/followed_wallet_moves', methods=['GET'])
|
|
def get_followed_wallet_moves():
|
|
# In a real-world scenario, you'd use a blockchain explorer API to get recent transactions
|
|
# For this example, we'll simulate a move
|
|
simulated_move = {
|
|
"token": "SOL",
|
|
"action": "swap",
|
|
"amount": 5,
|
|
"to_token": "USDC"
|
|
}
|
|
|
|
# Send Telegram notification about the detected move
|
|
asyncio.run(send_telegram_message(
|
|
f"<b>Move Detected:</b>\n"
|
|
f"Followed wallet swapped {simulated_move['amount']} {simulated_move['token']} "
|
|
f"to {simulated_move['to_token']}"
|
|
))
|
|
|
|
return jsonify(simulated_move)
|
|
|
|
@app.route('/follow_move', methods=['POST'])
|
|
def follow_move():
|
|
move = request.json
|
|
followed_balance = wallet_balances[FOLLOWED_WALLET][move['token']]
|
|
your_balance = wallet_balances[YOUR_WALLET][move['token']]
|
|
|
|
proportion = your_balance / followed_balance
|
|
amount_to_swap = move['amount'] * proportion
|
|
|
|
if wallet_balances[YOUR_WALLET][move['token']] >= amount_to_swap:
|
|
wallet_balances[YOUR_WALLET][move['token']] -= amount_to_swap
|
|
|
|
pair = dexscreener_client.get_token_pair("solana", move['token'])
|
|
price = float(pair['priceUsd'])
|
|
|
|
received_amount = amount_to_swap * price
|
|
wallet_balances[YOUR_WALLET][move['to_token']] += received_amount
|
|
|
|
# Send Telegram notification about the followed move
|
|
asyncio.run(send_telegram_message(
|
|
f"<b>Move Followed:</b>\n"
|
|
f"Swapped {amount_to_swap:.4f} {move['token']} "
|
|
f"for {received_amount:.4f} {move['to_token']}"
|
|
))
|
|
|
|
return jsonify({
|
|
'status': 'success',
|
|
'message': f"Swapped {amount_to_swap:.4f} {move['token']} for {received_amount:.4f} {move['to_token']}"
|
|
})
|
|
else:
|
|
# Send Telegram notification about the failed move
|
|
asyncio.run(send_telegram_message(
|
|
f"<b>Move Failed:</b>\n"
|
|
f"Insufficient balance to swap {amount_to_swap:.4f} {move['token']}"
|
|
))
|
|
|
|
return jsonify({
|
|
'status': 'error',
|
|
'message': f"Insufficient balance to swap {amount_to_swap:.4f} {move['token']}"
|
|
})
|
|
|
|
if __name__ == '__main__':
|
|
app.run(debug=True) |