AI Trading Bot

# Complete source code for DeepDown AI Trading Bot
import os
import re
import requests
import tweepy
import sqlite3
import numpy as np
from solana.rpc.api import Client
from solana.publickey import PublicKey
from solana.transaction import Transaction
from solana.system_program import TransferParams, transfer
from solana.rpc.types import TxOpts
from solana.rpc.commitment import Confirmed
from solana.account import Account
from dotenv import load_dotenv

# Load environment variables
load_dotenv()

# Initialize Solana client
solana_client = Client("https://api.mainnet-beta.solana.com")

# Set up wallet (private key)
private_key = os.getenv('SOLANA_PRIVATE_KEY')
wallet = Account(bytes.fromhex(private_key))

# Regex to match Solana and Ethereum contract addresses
SOLANA_ADDRESS_REGEX = re.compile(r'\b[1-9A-HJ-NP-Za-km-z]{32,44}\b')
ETHEREUM_ADDRESS_REGEX = re.compile(r'0x[a-fA-F0-9]{40}')

# Create local database
def create_database():
    conn = sqlite3.connect('memecoins.db')
    c = conn.cursor()
    c.execute('''CREATE TABLE IF NOT EXISTS contracts
                 (address TEXT PRIMARY KEY, chain TEXT, pumpfun_data TEXT, gmgn_data TEXT)''')
    conn.commit()
    conn.close()

create_database()

# Fetch contract score
def get_contract_score(address, chain):
    url = f"https://api.gmgn.ai/v1/contract/{chain}/{address}/score"
    headers = {
        "Authorization": f"Bearer {os.getenv('GMGN_API_KEY')}"
    }
    try:
        response = requests.get(url, headers=headers)
        if response.status_code == 200:
            return response.json().get('score', 100)
    except Exception as e:
        print(f"Error fetching contract score: {e}")
    return None

# Send Telegram alert
def send_telegram_alert(message):
    bot_token = os.getenv('TELEGRAM_BOT_TOKEN')
    chat_id = os.getenv('TELEGRAM_CHAT_ID')
    url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
    payload = {
        "chat_id": chat_id,
        "text": message
    }
    try:
        response = requests.post(url, json=payload)
        if response.status_code != 200:
            print(f"Failed to send Telegram alert: {response.text}")
    except Exception as e:
        print(f"Error sending Telegram alert: {e}")

# Check contract score and send alert if below threshold
def check_contract_score(address, chain, threshold=80):
    score = get_contract_score(address, chain)
    if score is not None and score < threshold:
        message = f"ALERT: Contract {address} on {chain} has a low score of {score}!"
        print(message)
        send_telegram_alert(message)

# Fetch SOL price
def get_sol_price():
    url = "https://api.coingecko.com/api/v3/simple/price?ids=solana&vs_currencies=usd"
    try:
        response = requests.get(url)
        if response.status_code == 200:
            return response.json()['solana']['usd']
    except Exception as e:
        print(f"Error fetching SOL price: {e}")
    return None

# Fetch token price
def get_token_price(token_address):
    url = f"https://api.gmgn.ai/v1/token/{token_address}/price"
    headers = {
        "Authorization": f"Bearer {os.getenv('GMGN_API_KEY')}"
    }
    try:
        response = requests.get(url, headers=headers)
        if response.status_code == 200:
            return response.json().get('price')
    except Exception as e:
        print(f"Error fetching token price: {e}")
    return None

# Buy tool
def buy_token(token_address, amount_in_sol, priority_fee, max_slippage=0.3):
    if amount_in_sol > 1:
        print("Error: Amount exceeds 1 SOL limit.")
        return None

    sol_price = get_sol_price()
    if not sol_price:
        print("Error: Failed to fetch SOL price.")
        return None

    token_price = get_token_price(token_address)
    if not token_price:
        print("Error: Failed to fetch token price.")
        return None

    max_price = token_price * (1 + max_slippage)

    transaction = Transaction()
    transaction.add(
        transfer(
            TransferParams(
                from_pubkey=wallet.public_key(),
                to_pubkey=PublicKey(token_address),
                lamports=int(amount_in_sol * 1e9)
            )
        )
    )

    transaction.recent_blockhash = solana_client.get_recent_blockhash()['result']['value']['blockhash']
    transaction.fee_payer = wallet.public_key()
    transaction.sign(wallet)

    try:
        result = solana_client.send_transaction(
            transaction, 
            wallet, 
            opts=TxOpts(skip_confirmation=False, preflight_commitment=Confirmed)
        )
        if result['result']:
            print(f"Buy order successful: {result['result']}")
            return result['result']
        else:
            print(f"Buy order failed: {result}")
            return None
    except Exception as e:
        print(f"Error sending buy order: {e}")
        return None

# Sell tool
def sell_token(token_address, amount_in_tokens, priority_fee, profit_target=15):
    token_balance = get_token_balance(token_address)
    if token_balance < amount_in_tokens:
        print("Error: Insufficient token balance.")
        return None

    token_price = get_token_price(token_address)
    if not token_price:
        print("Error: Failed to fetch token price.")
        return None

    target_price = token_price * profit_target

    transaction = Transaction()
    transaction.add(
        transfer(
            TransferParams(
                from_pubkey=wallet.public_key(),
                to_pubkey=PublicKey(token_address),
                lamports=int(amount_in_tokens * 1e9)
            )
        )
    )

    transaction.recent_blockhash = solana_client.get_recent_blockhash()['result']['value']['blockhash']
    transaction.fee_payer = wallet.public_key()
    transaction.sign(wallet)

    try:
        result = solana_client.send_transaction(
            transaction, 
            wallet, 
            opts=TxOpts(skip_confirmation=False, preflight_commitment=Confirmed)
        )
        if result['result']:
            print(f"Sell order successful: {result['result']}")
            return result['result']
        else:
            print(f"Sell order failed: {result}")
            return None
    except Exception as e:
        print(f"Error sending sell order: {e}")
        return None

# Twitter listener
class MemecoinListener(tweepy.StreamingClient):
    def on_tweet(self, tweet):
        print(f"New tweet from @{tweet.user.screen_name}: {tweet.text}")
        
        solana_addresses = SOLANA_ADDRESS_REGEX.findall(tweet.text)
        ethereum_addresses = ETHEREUM_ADDRESS_REGEX.findall(tweet.text)
        
        for address in solana_addresses:
            print(f"Found Solana address: {address}")
            self.process_address(address, 'solana')
        
        for address in ethereum_addresses:
            print(f"Found Ethereum address: {address}")
            self.process_address(address, 'ethereum')

    def process_address(self, address, chain):
        check_contract_score(address, chain)
        if should_buy(address, chain):
            buy_token(address, amount_in_sol=1, priority_fee=1000)

def should_buy(address, chain):
    # Add your buying conditions here
    return True

if __name__ == "__main__":
    listener = MemecoinListener(bearer_token=os.getenv('TWITTER_BEARER_TOKEN'))
    listener.filter(follow=["123456789", "987654321"])  # Replace with KOL user IDs