Add new python project

Program.cs

@@ -21,4 +21,4 @@ static void Main()
         }
     }
 
-
+}

defi-dex-swap-sniper-bot/.python-version

@@ -0,0 +1 @@
+3.12

defi-dex-swap-sniper-bot/README.md

@@ -0,0 +1,3 @@
+# Defi Dex Swap Sniper Bot
+
+This is the python implementation of the sniper bot

defi-dex-swap-sniper-bot/pyproject.toml

@@ -0,0 +1,29 @@
+[project]
+name = "defi-dex-swap-sniper-bot"
+version = "0.1.0"
+description = "Add your description here"
+readme = "README.md"
+requires-python = ">=3.12"
+dependencies = [
+    "anchorpy>=0.20.1",
+    "beautifulsoup4>=4.13.3",
+    "pydantic-settings>=2.7.1",
+    "requests>=2.32.3",
+    "selenium>=4.28.1",
+    "solana>=0.35.1",
+    "web3>=7.8.0",
+    "webdriver-manager>=4.0.2",
+]
+
+[project.scripts]
+start-bot = "defi_dex_swap_sniper_bot.trading_bot:run"
+
+[dependency-groups]
+dev = [
+    "ruff>=0.9.6",
+]
+
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"

defi-dex-swap-sniper-bot/src/defi_dex_swap_sniper_bot/abi.py

@@ -0,0 +1,43 @@
+# -*- coding: utf-8 -*-
+"""
+Created on Sun Apr 29 10:11:13 2024
+
+"""
+
+from bs4 import BeautifulSoup as bsp
+from selenium import webdriver
+from selenium.webdriver.firefox.webdriver import WebDriver
+from selenium.webdriver.firefox.options import Options
+from webdriver_manager.firefox import GeckoDriverManager
+
+
+def token_abi(address: str, driver: WebDriver | None=None):
+    try:
+        filename = f'ABI_{address}.txt'
+        with open(f"data/{filename}") as f:
+            abi = f.readlines()
+            return abi[0]
+    except IOError:
+        return find_abi(address, driver)
+
+
+def find_abi(address: str, driver: WebDriver | None = None) -> str:
+    url = f'https://bscscan.com/address/{address}#code'
+
+    if not driver:
+        options = Options()
+        options.headless = True
+        # TODO: Remove unexpected keyword argument
+        driver = webdriver.Firefox(executable_path=GeckoDriverManager().install(), options=options)
+
+    driver.get(url)
+    page_soup = bsp(driver.page_source, features="lxml")
+    abi = page_soup.find_all("pre", {"class": "wordwrap js-copytextarea2"})
+
+    with open(f'data/ABI_{address}.txt', 'w') as f:
+        f.write(abi[0].text)
+
+    driver.delete_all_cookies()
+    driver.get("chrome://settings/clearBrowserData")
+    # driver.close()
+    return abi[0].text

defi-dex-swap-sniper-bot/src/defi_dex_swap_sniper_bot/buy_tokens.py

@@ -0,0 +1,54 @@
+from typing import TypedDict, Unpack,Type
+from web3 import Web3
+from web3.contract import Contract
+from config import settings
+import time
+from eth_typing import Address,ChecksumAddress
+from web3.types import ENS
+
+
+class BuyTokenOptions(TypedDict):
+    symbol: str
+    web3: Web3
+    wallet_address: Address|ChecksumAddress|ENS
+    contract_pancake: Type[Contract] | Contract
+    token_to_buy_address: Address|ChecksumAddress|ENS
+    wbnb_address: Address|ChecksumAddress|ENS
+
+def buy_tokens(**options: Unpack[BuyTokenOptions]):
+    symbol = options.get('symbol')
+    web3 = options.get('web3')
+    wallet_address = options.get('wallet_address')
+    contract_pancake = options.get('contract_pancake')
+    token_to_buy_address = options.get('token_to_buy_address')
+    wbnb_address = options.get('wbnb_address')
+
+    if not all([symbol,web3,wallet_address,contract_pancake,token_to_buy_address,wbnb_address]):
+        raise ValueError('incomplete options provided to buy token. required parameters are `symbol, web3, '
+                         'wallet_address, contract_pancake, token_to_buy_address, wbnb_address`')
+
+    to_buy_bnb_amount = input(f"Enter amount of BNB you want to buy {symbol}: ")
+    to_buy_bnb_amount = web3.to_wei(to_buy_bnb_amount, 'ether')
+
+    pancake_swap_txn = contract_pancake.functions.swapExactETHForTokens(0,
+                                                                      [wbnb_address, token_to_buy_address],
+                                                                      wallet_address,
+                                                                      (int(time.time() + 10000))).buildTransaction({
+        'from': wallet_address,
+        'value': to_buy_bnb_amount,  # Amount of BNB
+        'gas': 160000,
+        'gasPrice': web3.to_wei('5', 'gwei'),
+        'nonce': web3.eth.get_transaction_count(wallet_address)
+    })
+
+    signed_txn = web3.eth.account.sign_transaction(pancake_swap_txn, private_key=settings.YOUR_PRIVATE_KEY)
+    try:
+        tx_token = web3.eth.send_raw_transaction(signed_txn.rawTransaction)
+        result = [web3.to_hex(tx_token), f"Bought {web3.from_wei(to_buy_bnb_amount, 'ether')} BNB of {symbol}"]
+        return result
+    except ValueError as e:
+        if e.args[0].get('message') in 'intrinsic gas too low':
+            result = ["Failed", f"ERROR: {e.args[0].get('message')}"]
+        else:
+            result = ["Failed", f"ERROR: {e.args[0].get('message')} : {e.args[0].get('code')}"]
+        return result

defi-dex-swap-sniper-bot/src/defi_dex_swap_sniper_bot/config.py

@@ -0,0 +1,27 @@
+from pydantic_settings import BaseSettings, SettingsConfigDict
+from eth_typing import Address,ChecksumAddress
+from web3.types import ENS
+
+class Settings(BaseSettings):
+    model_config = SettingsConfigDict(
+        env_file="./.env",
+        env_ignore_empty=True,
+        extra="ignore",
+    )
+
+    # Add Your Wallet Address by setting it in the .env file or providing it in your environmental variables
+    YOUR_WALLET_ADDRESS: Address|ChecksumAddress|ENS
+    # Add Your Private key by setting it in the .env file or providing it in your environmental variables
+    YOUR_PRIVATE_KEY: str
+
+    # Add your token address by setting it in the .env file or providing it in your environmental variables
+    # Example : "0xc66c8b40e9712708d0b4f27c9775dc934b65f0d9"
+    TRADE_TOKEN_ADDRESS: Address|ChecksumAddress|ENS | None = None  # Add token address here example : "0xc66c8b40e9712708d0b4f27c9775dc934b65f0d9"
+    WBNB_ADDRESS: Address|ChecksumAddress|ENS = "0xbb4CdB9CBd36B01bD1cBaEBF2De08d9173bc095c"
+    PANCAKE_ROUTER_ADDRESS: Address|ChecksumAddress|ENS = "0x10ED43C718714eb63d5aA57B78B54704E256024E"
+    SHOW_TX_ON_BROWSER = True
+
+    SELL_TOKENS = None
+    BUY_TOKENS = None
+
+settings = Settings() # type: ignore

defi-dex-swap-sniper-bot/src/defi_dex_swap_sniper_bot/exchange_api.py

@@ -0,0 +1,54 @@
+from http import HTTPMethod
+from typing import Literal, Any
+
+import requests
+import hmac
+import hashlib
+import time
+
+HTTPMethods = Literal['GET','POST','DELETE']
+
+class ExchangeAPI:
+    def __init__(self, api_key: str, api_secret: str, base_url="https://api.exchange.com"):
+        self.api_key = api_key
+        self.api_secret = api_secret
+        self.base_url = base_url
+
+    def _sign_request(self, method: HTTPMethods, path: str, params: dict[str,Any] | None=None):
+        timestamp = str(int(time.time()))
+        message = timestamp + method + path
+        if params:
+            message += str(params)
+        signature = hmac.new(self.api_secret.encode(), message.encode(), hashlib.sha256).hexdigest()
+        return signature
+
+    def _send_request(self, method: HTTPMethods, path: str, params: dict[str,Any] | None=None):
+        headers = {
+            "X-API-KEY": self.api_key,
+            "X-API-SIGNATURE": self._sign_request(method, path, params),
+            "X-API-TIMESTAMP": str(int(time.time()))
+        }
+        response = requests.request(method, self.base_url + path, headers=headers, json=params)
+        response.raise_for_status()
+        return response.json()
+
+    def get_balance(self):
+        return self._send_request("GET", "/balance")
+
+    def place_order(self, pair, side, price, quantity):
+        params = {
+            "pair": pair,
+            "side": side,
+            "price": price,
+            "quantity": quantity
+        }
+        return self._send_request("POST", "/orders", params)
+
+    def cancel_order(self, order_id):
+        return self._send_request("DELETE", f"/orders/{order_id}")
+
+    def get_order(self, order_id):
+        return self._send_request("GET", f"/orders/{order_id}")
+
+    def get_price(self, pair):
+        raise NotImplementedError

defi-dex-swap-sniper-bot/src/defi_dex_swap_sniper_bot/sol.py

@@ -0,0 +1,43 @@
+import anchorpy
+from solana.transaction import Keypair, Pubkey
+# from solana.system_program import SYS_PROGRAM_ID
+from solana.transaction import Transaction
+
+# Set up the wallet and RPC endpoint
+wallet = Keypair.from_seed(bytes.fromhex("YOUR_PRIVATE_KEY"))
+
+rpc_endpoint = "https://ssc-dao.genesysgo.net"
+
+# Connect to the Solana network
+connection = anchorpy.Connection(rpc_endpoint)
+
+# Install the token sniper program
+token_sniper_program_id = Pubkey("YOUR_IDENTIFICATOR_PROGRAM_SNIPPING_TOKENS")
+token_sniper_program = anchorpy.Program(connection, token_sniper_program_id, wallet)
+
+# Set the address of the liquidity pool you want to snipe
+pool_address = Pubkey("LICIDITY POOL_ADDRESS")
+
+# Set the amount of tokens you want to buy
+amount_to_buy = 1000
+
+# Set the price you are willing to pay for the tokens
+price_to_pay = 0.01
+
+# Create a transaction for sniping tokens
+transaction = Transaction()
+transaction.add(
+    token_sniper_program.instruction.snipe_tokens(
+        pool_address,
+        amount_to_buy,
+        price_to_pay
+    )
+)
+
+# Sign the transaction and send it to the network
+try:
+    transaction.sign([wallet])
+    connection.send_transaction(transaction)
+    print("Transaction successfully sent!")
+except Exception as e:
+    print("Error during transaction execution:", e)

defi-dex-swap-sniper-bot/src/defi_dex_swap_sniper_bot/strategies.py

@@ -0,0 +1,9 @@
+import random
+
+class MeanReversionStrategy:
+    def __init__(self, settings):
+        self.settings = settings
+
+    def decide(self, pair: str, amount):
+        # This is a very basic strategy that randomly decides to buy or sell
+        return "buy" if random.random() < 0.5 else "sell"