RapidBeats by Grav & Torrin 🩺

Instructions

  1. Save this code to a file called RapidBeats.py inside your affection-bots folder.
  2. Type pip install colorama from a terminal inside the folder
  3. Run the script with python RapidBeats.py

If you are confused by these instructions, go back to the tutorial and revist this later

Source Code:

⎗
✓
import time
import requests
import json
import unicodedata
from colorama import init, Fore, Style
from datetime import datetime
import colorsys
from web3 import Web3
import asyncio

# Initialize colorama for colored output
init(autoreset=True)

# Configuration
CONFIG = {
    'UPDATE_INTERVAL': 5,  # seconds
    'GAS_PRICE_MIN': 150000,  # minimum gas price in Beats for color scaling
    'GAS_PRICE_MAX': 1500000,  # maximum gas price in Beats for color scaling
    'COLOR_STOPS': [
        # Beats, (H,S,V) Hue, Saturation, and Value
        (100000, (240, 1, 1)),  # Blue
        (150000, (210, 0.9, 1)),  # Light Blue
        (250000, (180, 0.8, 1)),  # Cyan
        (350000, (120, 0.7, 1)),  # Green
        (450000, (90, 0.8, 1)),  # Light Green
        (550000, (60, 0.9, 1)),  # Yellow
        (700000, (30, 1, 1)),  # Orange
        (1000000, (10, 1, 1)),  # Red
        (2000000, (25, 1, 1)),  # Orange
    ],
    'MINTING_THRESHOLDS': [
        (250000, "[ Excellent mint Affection! ]"),
        (400000, "[   Good mint Affection.    ]"),
        (550000, "[  Acceptable for minting.  ]"),
        (800000, "[   Wait for lower Beats.   ]"),
        (1000000, "[   Rapid Beats too HIGH.   ]"),
    ]
}

ARB_REPORT_INTERVAL = 20  # seconds

# Token addresses
pdai_address = "0x6B175474E89094C44Da98b954EedeAC495271d0F"
pusdc_address = "0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48"
affection_address = "0x24F0154C1dCe548AdF15da2098Fdd8B8A3B8151D"
wpls_address = "0xA1077a294dDE1B09bB078844df40758a5D0f9a27"
pi_address = '0xA2262D7728C689526693aE893D0fD8a352C7073C'
math11_address = '0xB680F0cc810317933F234f67EB6A9E923407f05D'
g5_address = '0x2fc636E7fDF9f3E8d61033103052079781a6e7D2'

# PulseX_v2 router address and ABI
PULSEX_V2_ROUTER = "0x165C3410fC91EF562C50559f7d2289fEbed552d9"
ROUTER_ABI = [
    {
        "inputs": [
            {"internalType": "uint256", "name": "amountIn", "type": "uint256"},
            {"internalType": "address[]", "name": "path", "type": "address[]"}
        ],
        "name": "getAmountsOut",
        "outputs": [{"internalType": "uint256[]", "name": "amounts", "type": "uint256[]"}],
        "stateMutability": "view",
        "type": "function"
    }
]

# Web3 setup
WEB3_PROVIDER_URL = "https://rpc.pulsechain.com"
w3 = Web3(Web3.HTTPProvider(WEB3_PROVIDER_URL))
router_contract = w3.eth.contract(address=PULSEX_V2_ROUTER, abi=ROUTER_ABI)

async def estimate_mempool_gas_prices():
    try:
        pending = w3.eth.get_block('pending', full_transactions=True)
        gas_prices = []
        for tx in pending['transactions']:
            if 'maxFeePerGas' in tx:
                gas_prices.append(w3.from_wei(tx['maxFeePerGas'], 'gwei'))
            elif 'gasPrice' in tx:
                gas_prices.append(w3.from_wei(tx['gasPrice'], 'gwei'))

        if not gas_prices:
            return None

        gas_prices.sort()
        result = {
            'slow': float(round(gas_prices[int(len(gas_prices) * 0.25)], 2)),  # 25th percentile
            'standard': float(round(gas_prices[int(len(gas_prices) * 0.5)], 2)),  # 50th percentile
            'fast': float(round(gas_prices[int(len(gas_prices) * 0.70)], 2)),  # 70th percentile
            'rapid': float(round(gas_prices[int(len(gas_prices) * 0.90)], 2)),  # 90th percentile
        }
        return {k: int(v * 1e9) for k, v in result.items()}  # Convert to wei
    except Exception as e:
        print(f"Error estimating gas prices: {e}")
        return None

def get_token_decimals(token_address):
    token_decimals = {
        pdai_address: 18,
        pusdc_address: 6,
        affection_address: 18,
        wpls_address: 18,
        pi_address: 18,
        math11_address: 18,
        g5_address: 18,
    }
    return token_decimals.get(token_address, 18)

def sample_exchange_rate(token1, token2):
    try:
        decimals1 = get_token_decimals(token1)
        decimals2 = get_token_decimals(token2)
        amount_in = 10 ** decimals1
        amounts_out = router_contract.functions.getAmountsOut(
            amount_in,
            [token1, token2]
        ).call()
        amount_out = amounts_out[-1]
        rate = (amount_in / 10**decimals1) / (amount_out / 10**decimals2)
        return rate
    except Exception as e:
        print(f"Error fetching exchange rate for {token1} to {token2}: {e}")
        return None

def report_arb_opportunities():
    tokens = {
        "pDAI": pdai_address,
        "pUSDC": pusdc_address,
        "PI": pi_address,
        "MATH": math11_address,
        "G5": g5_address,
    }

    affection_rate = sample_exchange_rate(wpls_address, affection_address)
    if not affection_rate:
        print(f"{Fore.RED}Failed to fetch rate for AFFECTIONâ„¢{Style.RESET_ALL}")
        return

    arb_opportunities = []

    for name, address in tokens.items():
        rate = sample_exchange_rate(wpls_address, address)
        if rate is not None:
            price_diff = ((rate - affection_rate) / affection_rate) * 100
            price_diff = round(price_diff)  # Round to nearest integer
            arb_opportunities.append((name, price_diff, rate))
        else:
            print(f"{Fore.YELLOW}Failed to fetch rate for {name}{Style.RESET_ALL}")

    arb_opportunities.sort(key=lambda x: x[1], reverse=True)

    # Print header
    print(f"\n{'Token':<6} {'WPLS':>8} {'% from â’¶â„¢':>12}")
    print("-" * 28)

    # Print AFFECTIONâ„¢ price
    affection_price_str = format_rate(affection_rate)
    print(f"{Fore.MAGENTA}{'â’¶ ':<6} {affection_price_str:>8} {'0 %':>12}{Style.RESET_ALL}")

    for name, price_diff, rate in arb_opportunities:
        rate_str = format_rate(rate)
        price_diff_str = f"{price_diff:+d} %"

        if price_diff == 0:
            color = Fore.YELLOW
        elif price_diff > 0:
            color = Fore.GREEN
        else:
            color = Fore.RED

        print(f"{color}{name:<6} {rate_str:>8} {price_diff_str:>12}{Style.RESET_ALL}")

def format_rate(rate):
    if rate >= 1000000:
        adjusted_rate = rate / 1000000000000
        return f"{adjusted_rate:.2f}T"
    elif rate < 0.01:
        return f"{rate:.6f}"
    else:
        return f"{rate:.2f}"

def format_number_with_comma(number):
    return f"{number:,.0f}"

def get_color_for_gas_price(price_beats):
    stops = CONFIG['COLOR_STOPS']
    if price_beats <= stops[0][0]:
        return hsv_to_ansi(*stops[0][1])
    if price_beats >= stops[-1][0]:
        return hsv_to_ansi(*stops[-1][1])
    for i in range(len(stops) - 1):
        low, high = stops[i], stops[i + 1]
        if low[0] <= price_beats < high[0]:
            position = (price_beats - low[0]) / (high[0] - low[0])
            h = low[1][0] + (high[1][0] - low[1][0]) * position
            s = low[1][1] + (high[1][1] - low[1][1]) * position
            v = low[1][2] + (high[1][2] - low[1][2]) * position
            return hsv_to_ansi(h, s, v)

def hsv_to_ansi(h, s, v):
    rgb = colorsys.hsv_to_rgb(h / 360, s, v)
    r, g, b = [int(x * 255) for x in rgb]
    return f"\033[38;2;{r};{g};{b}m"

def get_minting_recommendation(price_beats):
    for threshold, recommendation in CONFIG['MINTING_THRESHOLDS']:
        if price_beats <= threshold:
            return recommendation
    if price_beats > 2000000:
        formatted_beats = format_number_with_comma(price_beats)
        return f"WHOA! Beats HIGH AF! >[{formatted_beats}]<"
    return CONFIG['MINTING_THRESHOLDS'][-1][1]

async def main():
    last_arb_report_time = 0
    while True:
        current_time = time.time()
        try:
            gas_data = await estimate_mempool_gas_prices()
            if gas_data:
                current_time_str = datetime.now().strftime("%H:%M:%S")
                print(f"\n[{current_time_str}] Current Gas Fees:")

                rapid_color = ""
                rapid_beats = 0
                for speed in ['rapid', 'fast', 'standard', 'slow']:
                    if speed in gas_data:
                        fee_wei = gas_data[speed]
                        fee_beats = fee_wei / 1e9

                        price_color = get_color_for_gas_price(fee_beats)
                        if speed == 'rapid':
                            rapid_color = price_color
                            rapid_beats = fee_beats
                        formatted_beats = format_number_with_comma(fee_beats)
                        print(f"  {price_color}{speed.capitalize():>8}: {formatted_beats:>10} Beats{Style.RESET_ALL}")

                recommendation = get_minting_recommendation(rapid_beats)
                rec_text = f"{recommendation}"
                print(f"\n{rapid_color}{rec_text}{Style.RESET_ALL}")

                # Display meter
                meter_width = len(rec_text)
                meter_position = min(int((rapid_beats - CONFIG['GAS_PRICE_MIN']) / (
                        CONFIG['GAS_PRICE_MAX'] - CONFIG['GAS_PRICE_MIN']) * meter_width), meter_width)
                acceptable_position = int((500000 - CONFIG['GAS_PRICE_MIN']) / (
                        CONFIG['GAS_PRICE_MAX'] - CONFIG['GAS_PRICE_MIN']) * meter_width)

                if rapid_beats > 2000000:
                    meter = '|' + '*' * meter_position + ' ' * (meter_width - meter_position - 2) + '|'
                else:
                    meter = '|' + '=' * meter_position + ' ' * (meter_width - meter_position - 2) + '|'

                meter = meter[:acceptable_position] + '|' + meter[acceptable_position + 1:]
                print(f"{rapid_color}{meter}{Style.RESET_ALL}")

                # Generate Arb Report every ARB_REPORT_INTERVAL seconds
                if current_time - last_arb_report_time >= ARB_REPORT_INTERVAL:
                    report_arb_opportunities()
                    last_arb_report_time = current_time

            else:
                print(f"{Fore.RED}Failed to fetch gas prices. Retrying in {CONFIG['UPDATE_INTERVAL']} seconds...{Style.RESET_ALL}")
        except Exception as e:
            print(f"{Fore.RED}An unexpected error occurred: {e}{Style.RESET_ALL}")

        await asyncio.sleep(CONFIG['UPDATE_INTERVAL'])

if __name__ == "__main__":
    asyncio.run(main())
Edit
Pub: 29 Aug 2024 13:12 UTC
Edit: 18 Oct 2024 15:14 UTC
Views: 327