Trading Algorithm Code

Our script

# main.py
import requests
from sell import sell_asset
from buy import buy_asset
from google_sheets import log_transaction

from strategy import strategy, Refresh_Time
import time  # Module to control intervals

Run = True  # Variable to start and pause the script

Refresh_Time = 1800  # The script waits this long before it goes through again, 3600 seconds = 1 hour

# Define the percentage of USDT to use for buying
buy_percentage = 0.2

api_key = "Add here your KEY"
secret_key = "Add here your KEY"

# List of API URLs to check
api_urls = ['https://api1.binance.com', 'https://api2.binance.com', 'https://api3.binance.com']

# Initialize the url variable
url = None

# Iterate through the API URLs
for api_url in api_urls:
    try:
        # Attempt to send a GET request to the API
        resp = requests.get(api_url)
        # Check if the status code is 200 (success)
        if resp.status_code == 200:
            # Set the url variable to the URL of the successful API
            url = api_url
            # Break the loop if a working API is found
            break
    except:
        # Continue if an exception occurs
        pass

# Check if a working API URL was found
if not url:
    print("All APIs are unreachable.")
    exit()

def order_signal(order_signal, api_key, secret_key, api_url):
    # Split the trading pair and the action
    symbol, action = order_signal.split(":")
    
    # Execute the corresponding action based on the signal
    if action == "BUY":
        print("Kaufsignal")
        response = buy_asset(symbol, api_url, api_key, secret_key, buy_percentage)
    elif action == "SELL":
        print("Verkaufssignal")
        response = sell_asset(symbol, api_url, api_key, secret_key)
    else:
        print("Ungültige Aktion")
        return

    if response:
        log_transaction(response, action, api_key, secret_key, api_url)

# structure main-file

symbols = ["HBARUSDT", "BTCUSDT", "ETHUSDT", "ALGOUSDT", "LINKUSDT"]

while Run == True:
    for symbol in symbols:
        order = strategy(symbol)
        if order:
            print(order)
            order_signal(order, api_key, secret_key, url)
    time.sleep(Refresh_Time)  # Then the loop is paused for the duration of Refresh_Time

import urllib.parse
import hashlib
import hmac
import requests
import time
import math

recv_window = 5000

# Function to generate the Binance US signature
def get_binanceus_signature(data, secret):
    postdata = urllib.parse.urlencode(data)
    message = postdata.encode()
    byte_key = bytes(secret, 'UTF-8')
    mac = hmac.new(byte_key, message, hashlib.sha256).hexdigest()
    return mac

# Function to send a POST request to Binance US
def buy_binanceus_request(uri_path, data, api_key, api_sec, api_url):
    headers = {
        'X-MBX-APIKEY': api_key
    }
    signature = get_binanceus_signature(data, api_sec)
    payload = {**data, "signature": signature}
    req = requests.post(f"{api_url}{uri_path}", headers=headers, data=payload)
    return req.json()

# Function to get the current price of a trading pair
def get_current_price(symbol, api_url):
    url = f"{api_url}/api/v3/ticker/price"
    params = {
        "symbol": symbol
    }
    response = requests.get(url, params=params)
    return response.json()

# Function to get the balance of USDT
def get_usdt_balance(api_key, secret_key, api_url):
    uri_path = "/api/v3/account"
    data = {
        "timestamp": int(round(time.time() * 1000)),
        "recvWindow": recv_window
    }
    headers = {
        'X-MBX-APIKEY': api_key
    }
    signature = get_binanceus_signature(data, secret_key)
    payload = f"{urllib.parse.urlencode(data)}&signature={signature}"
    req = requests.get(f"{api_url}{uri_path}?{payload}", headers=headers)
    response = req.json()

    if "balances" in response:
        for item in response["balances"]:
            if item["asset"] == "USDT":
                return float(item["free"])
    return 0.0

def round_down(value, decimals):
    factor = 10 ** decimals
    return math.floor(value * factor) / factor

# Function to buy an asset
def buy_asset(symbol, api_url, api_key, secret_key, buy_percentage):
    usdt_balance = get_usdt_balance(api_key, secret_key, api_url)
    amount_to_use = usdt_balance * buy_percentage

    if amount_to_use < 6.0:
        print(f"Not enough USDT to make a purchase. Minimum is 10 USDT, you have {usdt_balance} USDT.")
        return

    pair = symbol
    price_response = get_current_price(pair, api_url)
    if "price" not in price_response:
        print(f"Could not retrieve price for {pair}")
        return

    price = float(price_response["price"])
    quantity_to_buy = amount_to_use / price

    filters = get_trading_rules(pair, api_url)
    min_qty = None
    step_size = None

    for f in filters:
        if f['filterType'] == 'LOT_SIZE':
            min_qty = float(f['minQty'])
            step_size = float(f['stepSize'])
            quantity_to_buy = round_down(quantity_to_buy, int(-math.log10(step_size)))

    if quantity_to_buy < min_qty:
        print(f"Quantity to buy {quantity_to_buy} is less than the minimum quantity {min_qty}.")
        return

    uri_path = "/api/v3/order"
    data = {
        'symbol': pair,
        'side': 'BUY',
        'type': 'MARKET',
        'quantity': quantity_to_buy,
        "timestamp": int(round(time.time() * 1000)),
        "recvWindow": recv_window
    }

    resp = buy_binanceus_request(uri_path, data, api_key, secret_key, api_url)
    print(resp)
    return resp

# Function to get the trading rules for a symbol
def get_trading_rules(symbol, api_url):
    url = f"{api_url}/api/v3/exchangeInfo"
    response = requests.get(url)
    data = response.json()

    for s in data['symbols']:
        if s['symbol'] == symbol:
            return s['filters']
    return []

import urllib.parse
import hashlib
import hmac
import requests
import time
import math

recv_window = 5000

# Function to generate the Binance US signature
def get_binanceus_signature(data, secret):
    postdata = urllib.parse.urlencode(data)
    message = postdata.encode()
    byte_key = bytes(secret, 'UTF-8')
    mac = hmac.new(byte_key, message, hashlib.sha256).hexdigest()
    return mac

# Function to send a POST request to Binance US
def sell_binanceus_request(uri_path, data, api_key, api_sec, api_url):
    headers = {
        'X-MBX-APIKEY': api_key
    }
    signature = get_binanceus_signature(data, api_sec)
    payload = {**data, "signature": signature}
    req = requests.post(f"{api_url}{uri_path}", headers=headers, data=payload)
    return req.json()

# Function to get the current price of a trading pair
def get_current_price(symbol, api_url):
    url = f"{api_url}/api/v3/ticker/price"
    params = {
        "symbol": symbol
    }
    response = requests.get(url, params=params)
    return response.json()

# Function to get the trading rules for a symbol
def get_trading_rules(symbol, api_url):
    url = f"{api_url}/api/v3/exchangeInfo"
    response = requests.get(url)
    data = response.json()

    for s in data['symbols']:
        if s['symbol'] == symbol:
            return s['filters']
    return []

def round_down(value, decimals):
    factor = 10 ** decimals
    return math.floor(value * factor) / factor

# Function to sell an asset
def sell_asset(symbol, api_url, api_key, secret_key):
    asset = symbol.replace("USDT", "")
    quantity = get_asset_balance(asset, api_key, secret_key, api_url)
    pair = symbol
    
    price_response = get_current_price(pair, api_url)
    if "price" not in price_response:
        print(f"Could not retrieve price for {pair}")
        return

    price = float(price_response["price"])
    value_in_usdt = quantity * price

    min_usdt_threshold = 5.0
    if value_in_usdt < min_usdt_threshold:
        print(f"Value of {symbol} is below the minimum threshold of {min_usdt_threshold} USDT")
        return
    
    filters = get_trading_rules(pair, api_url)
    min_qty = None
    step_size = None

    for f in filters:
        if f['filterType'] == 'LOT_SIZE':
            min_qty = float(f['minQty'])
            step_size = float(f['stepSize'])
            quantity = round_down(quantity, int(-math.log10(step_size)))

    print(f"Prepared to sell {quantity} {symbol}")

    if quantity < min_qty:
        print(f"Quantity {quantity} is less than the minimum quantity {min_qty}.")
        return

    uri_path = "/api/v3/order"
    data = {
        'symbol': pair,
        'side': 'SELL',
        'type': 'MARKET',
        'quantity': quantity,
        "timestamp": int(round(time.time() * 1000)),
        "recvWindow": recv_window
    }

    resp = sell_binanceus_request(uri_path, data, api_key, secret_key, api_url)
    print(resp)
    return resp

# Function to get the balance of a specific asset
def get_asset_balance(asset, api_key, secret_key, api_url):
    uri_path = "/api/v3/account"
    data = {
        "timestamp": int(round(time.time() * 1000)),
        "recvWindow": recv_window
    }
    headers = {
        'X-MBX-APIKEY': api_key
    }
    signature = get_binanceus_signature(data, secret_key)
    payload = f"{urllib.parse.urlencode(data)}&signature={signature}"
    req = requests.get(f"{api_url}{uri_path}?{payload}", headers=headers)
    response = req.json()

    if "balances" in response:
        for item in response["balances"]:
            if item["asset"] == asset:
                free_quantity = float(item["free"])
                print(f"Available {asset} balance: {free_quantity}")
                return free_quantity
    return 0.0

import requests     # Module for API requests

# General variables
ShortRange_EMA_Length = 20  # Number of values for the ShortRange EMA
LongRange_EMA_Length = 100  # Number of values for the LongRange EMA
Smoothing_Factor = 2        # Smoothing factor for the calculation of the EMAs
Chart_Interval = '1h'       # Interval of the values, takes one value per hour
Chart_Length = 200          # The last 200 values are stored in a list
Refresh_Time = 3600           # The script waits this long before it goes through again, 3600 seconds = 1 hour
Stop_Signal = 0.99          # Percentage at which it will be sold, 0.99 = 1% loss from the highest value in the LongRange, currently not in use
Run = True                  # Variable to start and pause the script
Order_History = [["HBARUSDT", ""], 
              ["BTCUSDT", ""], 
              ["ETHUSDT", ""],
              ["ALGOUSDT", ""],
              ["LINKUSDT", ""]
              ]

# Price lists of individual currencies
Prices_HBARUSDT = []    # Hedera
Prices_BTCUSDT = []     # Bitcoin
Prices_ETHUSDT = []     # Ethereum
Prices_ALGOUSDT = []    # Algorand
Prices_LINKUSDT = []    # Chainlink

# Maximum prices of the currencies since the beginning of the respective measurement
Max_Prices = [["HBARUSDT", 0], 
              ["BTCUSDT", 0], 
              ["ETHUSDT", 0],
              ["ALGOUSDT", 0],
              ["LINKUSDT", 0]
              ]


# Function to calculate the SMA (Simple Moving Average), needed to calculate the EMA
def SMA_Calculation(Prices, Length):        # Requires a list of prices and the length of the period (20 for ShortRange, 100 for LongRange)
    Sum = sum(Prices[:Length])              # The first Length prices are summed up
    SMA = Sum / Length                      # Then they are divided by the number of measurements
    return SMA                              # The function returns the SMA value


# Function to calculate the EMA (Exponential Moving Average), an indicator for trading strategies
def EMA_Calculation(Prices, Length, Smoothing_Factor):                                          # Requires a list of prices and the length of the period (20 for ShortRange, 100 for LongRange), as well as the smoothing factor
    Current_EMA = SMA_Calculation(Prices, Length)                                               # The initial SMA is calculated using the SMA_Calculation function, which corresponds to the EMA of the first period
    Smoothing_Factor = Smoothing_Factor / (Length + 1)                                          # The smoothing factor is adjusted relative to the length
    for Price in Prices[Length:]:                                                               # For each additional price in the list (excluding those already used for the initial SMA)
        Current_EMA = (Price * Smoothing_Factor) + (Current_EMA * (1 - Smoothing_Factor))       # The current EMA is calculated using the current price, the previous EMA, and the smoothing factor
    return Current_EMA                                                                          # The current EMA is returned


# Function to update the price lists
def List_Update(Currency):                                                                                                                          # Requires the currency
    global Max_Prices                                                                                                                               # Edits the global variable Max_Prices (highest price since the beginning of the measurement)
    Data = requests.get('https://api.binance.com/api/v3/klines', params={'symbol': Currency, 'interval': Chart_Interval, 'limit': Chart_Length})    # Requests the values from Binance with the specified parameters and stores them in the variable Data
    Data = Data.json()                                                                                                                              # Converts the data from JSON to a Python list 
    Prices = [float(entry[4]) for entry in Data]                                                                                                    # Takes the important data (close prices) from the list and changes the data type to float
    for Max_Price in Max_Prices:                                                                                                                    # For each entry in the Max_Prices list
        if Max_Price[0] == Currency and Max_Price[1] < max(Prices):                                                                                 # It is checked whether the current highest value is smaller than the highest after the update
            Max_Price[1] = max(Prices)                                                                                                              # If yes, the highest value is adjusted
    return Prices                                                                                                                                   # The list of new prices is returned


# Function to execute the orders (buy/sell)
def Order(LongRange_EMA, Current_Price, Currency):                      # Requires the LongRange EMA, the current price, and the associated currency
    order = ""                                                         # Initialize order as empty
    print("Currency:", Currency, "EMA:", LongRange_EMA, "Price:", Current_Price)
    if LongRange_EMA > Current_Price:                                   # Checks whether the LongRange EMA is greater than the current price
        for Value in Order_History:                                     # Each currency goes through
            if Value[0] == Currency and Value[1] != "Sell":             # Checks whether this is the correct currency and whether it does not have the status Sell
                if Value[1] == "Buy":                                   # If yes, it is checked whether Buy is inside
                    order = f"{Currency}:SELL"                          # If yes, it should be sold
                    print("\033[31mSell", Currency, "because LongRange EMA:", LongRange_EMA, "is higher than the current Price", Current_Price, "and it has changed since the last hour.\033[0m")
                else:                                                   # If no, it should not progress any action
                    print("No action. This is the initial price of", Currency)
                Value[1] = "Sell"                                       # Changes the History to Sell
            else:
                if Value[0] == Currency:                                # If no, it should not progress any action
                    print("No action. The current Price of", Currency, "is still lower than the EMA, it hasn't changed.")
    else:                                                               # If the current price is greater than the LongRange EMA
        for Value in Order_History:                                     # Each currency goes through
            if Value[0] == Currency and Value[1] != "Buy":              # Checks whether this is the correct currency and whether it does not have the status Buy
                if Value[1] == "Sell":                                  # If yes, it is checked whether Sell is inside
                    order = f"{Currency}:BUY"                           # If no, it should be bought
                    print("\033[31mBuy", Currency, "because LongRange EMA:", LongRange_EMA, "is lower than current Price", Current_Price, "and it has changed since the last hour.\033[0m")
                else:                                                   # If no, it should not progress any action
                    print("No action. This is the initial price of", Currency)
                Value[1] = "Buy"                                        # Changes the History to Buy
            else:
                if Value[0] == Currency:                                # If no, it should not progress any action
                    print("No action. The current Price of", Currency, "is still higher than the EMA, it hasn't changed.")
    return order


# This is the main function where the functions are called
# The function is executed in another file
# First, the price list of the currency is updated
# Then the EMAs are calculated, compared, and it is decided whether to buy or sell 
def strategy(Symbol):
        if Symbol == "HBARUSDT":    # HBARUSDT - Hedera
            Prices_HBARUSDT = List_Update("HBARUSDT")
            return Order(EMA_Calculation(Prices_HBARUSDT, LongRange_EMA_Length, Smoothing_Factor), Prices_HBARUSDT[-1], "HBARUSDT")
        elif Symbol == "BTCUSDT":   # BTCUSDT - Bitcoin
            Prices_BTCUSDT = List_Update("BTCUSDT")
            return Order(EMA_Calculation(Prices_BTCUSDT, LongRange_EMA_Length, Smoothing_Factor), Prices_BTCUSDT[-1], "BTCUSDT")
        elif Symbol == "ETHUSDT":   # ETHUSDT - Ethereum
            Prices_ETHUSDT = List_Update("ETHUSDT")
            return Order(EMA_Calculation(Prices_ETHUSDT, LongRange_EMA_Length, Smoothing_Factor), Prices_ETHUSDT[-1], "ETHUSDT")
        elif Symbol == "ALGOUSDT":  # AGLOUSDT - Algorand
            Prices_ALGOUSDT = List_Update("ALGOUSDT")
            return Order(EMA_Calculation(Prices_ALGOUSDT, LongRange_EMA_Length, Smoothing_Factor), Prices_ALGOUSDT[-1], "ALGOUSDT")
        elif Symbol == "LINKUSDT":  # LINKUSDT - Chainlink
            Prices_LINKUSDT = List_Update("LINKUSDT")
            return Order(EMA_Calculation(Prices_LINKUSDT, LongRange_EMA_Length, Smoothing_Factor), Prices_LINKUSDT[-1], "LINKUSDT")

google-sheets-credentials.json has to be in the same folder

from google.oauth2 import service_account
from googleapiclient.discovery import build
from datetime import datetime
import urllib.parse
import hashlib
import hmac
import requests
import time

# Path to the JSON file with the service account information
credentials_file_path = "google-sheets-credentials.json"

# Spreadsheet ID (from the document URL)
spreadsheet_id = "15mWxtbm7b44fQDpbRrc7aASJ9Zcn9NQfTGU-UqYi33o"

# Create the service account credentials
credentials = service_account.Credentials.from_service_account_file(credentials_file_path, scopes=["https://www.googleapis.com/auth/spreadsheets"])

# Build the Sheets service
service = build("sheets", "v4", credentials=credentials)

recv_window = 5000

def get_binanceus_signature(data, secret):
    postdata = urllib.parse.urlencode(data)
    message = postdata.encode()
    byte_key = bytes(secret, 'UTF-8')
    mac = hmac.new(byte_key, message, hashlib.sha256).hexdigest()
    return mac

def get_all_balances(api_key, secret_key, api_url):
    uri_path = "/api/v3/account"
    data = {
        "timestamp": int(round(time.time() * 1000)),
        "recvWindow": recv_window
    }
    headers = {
        'X-MBX-APIKEY': api_key
    }
    signature = get_binanceus_signature(data, secret_key)
    payload = f"{urllib.parse.urlencode(data)}&signature={signature}"
    try:
        req = requests.get(f"{api_url}{uri_path}?{payload}", headers=headers)
        req.raise_for_status()
        response = req.json()
    except requests.exceptions.RequestException as e:
        print(f"Failed to get account balances: {e}")
        return []

    balances = []
    if "balances" in response:
        for item in response["balances"]:
            asset = item["asset"]
            free_quantity = float(item["free"])
            if free_quantity > 0:
                balances.append((asset, free_quantity))
    return balances

def get_current_price(symbol, api_url):
    url = f"{api_url}/api/v3/ticker/price"
    params = {"symbol": symbol}
    try:
        response = requests.get(url, params=params)
        response.raise_for_status()
        data = response.json()
        return float(data['price'])
    except requests.exceptions.RequestException as e:
        print(f"Failed to get current price for {symbol}: {e}")
        return None

def calculate_total_balance_in_usdt(api_key, secret_key, api_url):
    balances = get_all_balances(api_key, secret_key, api_url)
    total_usdt = 0.0
    for asset, quantity in balances:
        if asset == "USDT":
            total_usdt += quantity
        else:
            price = get_current_price(f"{asset}USDT", api_url)
            if price is not None:
                total_usdt += quantity * price
    return total_usdt

def get_last_row(sheet_name):
    result = service.spreadsheets().values().get(spreadsheetId=spreadsheet_id, range=f"{sheet_name}!A:A").execute()
    values = result.get('values', [])
    return len(values) + 1

def log_transaction(response, action, api_key, secret_key, api_url):
    sheet_name = "Sheet1"
    next_row = get_last_row(sheet_name)
    range_name = f"{sheet_name}!A{next_row}"

    pair = response['symbol']
    price = float(response['fills'][0]['price'])
    executed_qty = float(response['executedQty'])
    cummulative_quote_qty = round(float(response['cummulativeQuoteQty']), 2)
    commission = float(response['fills'][0]['commission'])
    transact_time = response['transactTime']
    
    total_balance_usdt = calculate_total_balance_in_usdt(api_key, secret_key, api_url)
    
    readable_time = datetime.fromtimestamp(transact_time / 1000).strftime('%d.%m.%Y %H:%M:%S')
    values = [[pair, action, price, executed_qty, commission, cummulative_quote_qty, total_balance_usdt, readable_time]]

    body = {'values': values}
    value_input_option = 'RAW'

    request = service.spreadsheets().values().update(spreadsheetId=spreadsheet_id, range=range_name, valueInputOption=value_input_option, body=body)
    response = request.execute()

    print(f"Transaction logged in row {next_row}.")

Last updated 1 year ago

Was this helpful?

import urllib.parse import hashlib import hmac import requests import time import math recv_window = 5000 # Function to generate the Binance US signature def get_binanceus_signature(data, secret): postdata = urllib.parse.urlencode(data) message = postdata.encode() byte_key = bytes(secret, 'UTF-8') mac = hmac.new(byte_key, message, hashlib.sha256).hexdigest() return mac # Function to send a POST request to Binance US def buy_binanceus_request(uri_path, data, api_key, api_sec, api_url): headers = { 'X-MBX-APIKEY': api_key } signature = get_binanceus_signature(data, api_sec) payload = {**data, "signature": signature} req = requests.post(f"{api_url}{uri_path}", headers=headers, data=payload) return req.json() # Function to get the current price of a trading pair def get_current_price(symbol, api_url): url = f"{api_url}/api/v3/ticker/price" params = { "symbol": symbol } response = requests.get(url, params=params) return response.json() # Function to get the balance of USDT def get_usdt_balance(api_key, secret_key, api_url): uri_path = "/api/v3/account" data = { "timestamp": int(round(time.time() * 1000)), "recvWindow": recv_window } headers = { 'X-MBX-APIKEY': api_key } signature = get_binanceus_signature(data, secret_key) payload = f"{urllib.parse.urlencode(data)}&signature={signature}" req = requests.get(f"{api_url}{uri_path}?{payload}", headers=headers) response = req.json() if "balances" in response: for item in response["balances"]: if item["asset"] == "USDT": return float(item["free"]) return 0.0 def round_down(value, decimals): factor = 10 ** decimals return math.floor(value * factor) / factor # Function to buy an asset def buy_asset(symbol, api_url, api_key, secret_key, buy_percentage): usdt_balance = get_usdt_balance(api_key, secret_key, api_url) amount_to_use = usdt_balance * buy_percentage if amount_to_use < 6.0: print(f"Not enough USDT to make a purchase. Minimum is 10 USDT, you have {usdt_balance} USDT.") return pair = symbol price_response = get_current_price(pair, api_url) if "price" not in price_response: print(f"Could not retrieve price for {pair}") return price = float(price_response["price"]) quantity_to_buy = amount_to_use / price filters = get_trading_rules(pair, api_url) min_qty = None step_size = None for f in filters: if f['filterType'] == 'LOT_SIZE': min_qty = float(f['minQty']) step_size = float(f['stepSize']) quantity_to_buy = round_down(quantity_to_buy, int(-math.log10(step_size))) if quantity_to_buy < min_qty: print(f"Quantity to buy {quantity_to_buy} is less than the minimum quantity {min_qty}.") return uri_path = "/api/v3/order" data = { 'symbol': pair, 'side': 'BUY', 'type': 'MARKET', 'quantity': quantity_to_buy, "timestamp": int(round(time.time() * 1000)), "recvWindow": recv_window } resp = buy_binanceus_request(uri_path, data, api_key, secret_key, api_url) print(resp) return resp # Function to get the trading rules for a symbol def get_trading_rules(symbol, api_url): url = f"{api_url}/api/v3/exchangeInfo" response = requests.get(url) data = response.json() for s in data['symbols']: if s['symbol'] == symbol: return s['filters'] return []

import urllib.parse import hashlib import hmac import requests import time import math recv_window = 5000 # Function to generate the Binance US signature def get_binanceus_signature(data, secret): postdata = urllib.parse.urlencode(data) message = postdata.encode() byte_key = bytes(secret, 'UTF-8') mac = hmac.new(byte_key, message, hashlib.sha256).hexdigest() return mac # Function to send a POST request to Binance US def sell_binanceus_request(uri_path, data, api_key, api_sec, api_url): headers = { 'X-MBX-APIKEY': api_key } signature = get_binanceus_signature(data, api_sec) payload = {**data, "signature": signature} req = requests.post(f"{api_url}{uri_path}", headers=headers, data=payload) return req.json() # Function to get the current price of a trading pair def get_current_price(symbol, api_url): url = f"{api_url}/api/v3/ticker/price" params = { "symbol": symbol } response = requests.get(url, params=params) return response.json() # Function to get the trading rules for a symbol def get_trading_rules(symbol, api_url): url = f"{api_url}/api/v3/exchangeInfo" response = requests.get(url) data = response.json() for s in data['symbols']: if s['symbol'] == symbol: return s['filters'] return [] def round_down(value, decimals): factor = 10 ** decimals return math.floor(value * factor) / factor # Function to sell an asset def sell_asset(symbol, api_url, api_key, secret_key): asset = symbol.replace("USDT", "") quantity = get_asset_balance(asset, api_key, secret_key, api_url) pair = symbol price_response = get_current_price(pair, api_url) if "price" not in price_response: print(f"Could not retrieve price for {pair}") return price = float(price_response["price"]) value_in_usdt = quantity * price min_usdt_threshold = 5.0 if value_in_usdt < min_usdt_threshold: print(f"Value of {symbol} is below the minimum threshold of {min_usdt_threshold} USDT") return filters = get_trading_rules(pair, api_url) min_qty = None step_size = None for f in filters: if f['filterType'] == 'LOT_SIZE': min_qty = float(f['minQty']) step_size = float(f['stepSize']) quantity = round_down(quantity, int(-math.log10(step_size))) print(f"Prepared to sell {quantity} {symbol}") if quantity < min_qty: print(f"Quantity {quantity} is less than the minimum quantity {min_qty}.") return uri_path = "/api/v3/order" data = { 'symbol': pair, 'side': 'SELL', 'type': 'MARKET', 'quantity': quantity, "timestamp": int(round(time.time() * 1000)), "recvWindow": recv_window } resp = sell_binanceus_request(uri_path, data, api_key, secret_key, api_url) print(resp) return resp # Function to get the balance of a specific asset def get_asset_balance(asset, api_key, secret_key, api_url): uri_path = "/api/v3/account" data = { "timestamp": int(round(time.time() * 1000)), "recvWindow": recv_window } headers = { 'X-MBX-APIKEY': api_key } signature = get_binanceus_signature(data, secret_key) payload = f"{urllib.parse.urlencode(data)}&signature={signature}" req = requests.get(f"{api_url}{uri_path}?{payload}", headers=headers) response = req.json() if "balances" in response: for item in response["balances"]: if item["asset"] == asset: free_quantity = float(item["free"]) print(f"Available {asset} balance: {free_quantity}") return free_quantity return 0.0

import requests # Module for API requests # General variables ShortRange_EMA_Length = 20 # Number of values for the ShortRange EMA LongRange_EMA_Length = 100 # Number of values for the LongRange EMA Smoothing_Factor = 2 # Smoothing factor for the calculation of the EMAs Chart_Interval = '1h' # Interval of the values, takes one value per hour Chart_Length = 200 # The last 200 values are stored in a list Refresh_Time = 3600 # The script waits this long before it goes through again, 3600 seconds = 1 hour Stop_Signal = 0.99 # Percentage at which it will be sold, 0.99 = 1% loss from the highest value in the LongRange, currently not in use Run = True # Variable to start and pause the script Order_History = [["HBARUSDT", ""], ["BTCUSDT", ""], ["ETHUSDT", ""], ["ALGOUSDT", ""], ["LINKUSDT", ""] ] # Price lists of individual currencies Prices_HBARUSDT = [] # Hedera Prices_BTCUSDT = [] # Bitcoin Prices_ETHUSDT = [] # Ethereum Prices_ALGOUSDT = [] # Algorand Prices_LINKUSDT = [] # Chainlink # Maximum prices of the currencies since the beginning of the respective measurement Max_Prices = [["HBARUSDT", 0], ["BTCUSDT", 0], ["ETHUSDT", 0], ["ALGOUSDT", 0], ["LINKUSDT", 0] ] # Function to calculate the SMA (Simple Moving Average), needed to calculate the EMA def SMA_Calculation(Prices, Length): # Requires a list of prices and the length of the period (20 for ShortRange, 100 for LongRange) Sum = sum(Prices[:Length]) # The first Length prices are summed up SMA = Sum / Length # Then they are divided by the number of measurements return SMA # The function returns the SMA value # Function to calculate the EMA (Exponential Moving Average), an indicator for trading strategies def EMA_Calculation(Prices, Length, Smoothing_Factor): # Requires a list of prices and the length of the period (20 for ShortRange, 100 for LongRange), as well as the smoothing factor Current_EMA = SMA_Calculation(Prices, Length) # The initial SMA is calculated using the SMA_Calculation function, which corresponds to the EMA of the first period Smoothing_Factor = Smoothing_Factor / (Length + 1) # The smoothing factor is adjusted relative to the length for Price in Prices[Length:]: # For each additional price in the list (excluding those already used for the initial SMA) Current_EMA = (Price * Smoothing_Factor) + (Current_EMA * (1 - Smoothing_Factor)) # The current EMA is calculated using the current price, the previous EMA, and the smoothing factor return Current_EMA # The current EMA is returned # Function to update the price lists def List_Update(Currency): # Requires the currency global Max_Prices # Edits the global variable Max_Prices (highest price since the beginning of the measurement) Data = requests.get('https://api.binance.com/api/v3/klines', params={'symbol': Currency, 'interval': Chart_Interval, 'limit': Chart_Length}) # Requests the values from Binance with the specified parameters and stores them in the variable Data Data = Data.json() # Converts the data from JSON to a Python list Prices = [float(entry[4]) for entry in Data] # Takes the important data (close prices) from the list and changes the data type to float for Max_Price in Max_Prices: # For each entry in the Max_Prices list if Max_Price[0] == Currency and Max_Price[1] < max(Prices): # It is checked whether the current highest value is smaller than the highest after the update Max_Price[1] = max(Prices) # If yes, the highest value is adjusted return Prices # The list of new prices is returned # Function to execute the orders (buy/sell) def Order(LongRange_EMA, Current_Price, Currency): # Requires the LongRange EMA, the current price, and the associated currency order = "" # Initialize order as empty print("Currency:", Currency, "EMA:", LongRange_EMA, "Price:", Current_Price) if LongRange_EMA > Current_Price: # Checks whether the LongRange EMA is greater than the current price for Value in Order_History: # Each currency goes through if Value[0] == Currency and Value[1] != "Sell": # Checks whether this is the correct currency and whether it does not have the status Sell if Value[1] == "Buy": # If yes, it is checked whether Buy is inside order = f"{Currency}:SELL" # If yes, it should be sold print("\033[31mSell", Currency, "because LongRange EMA:", LongRange_EMA, "is higher than the current Price", Current_Price, "and it has changed since the last hour.\033[0m") else: # If no, it should not progress any action print("No action. This is the initial price of", Currency) Value[1] = "Sell" # Changes the History to Sell else: if Value[0] == Currency: # If no, it should not progress any action print("No action. The current Price of", Currency, "is still lower than the EMA, it hasn't changed.") else: # If the current price is greater than the LongRange EMA for Value in Order_History: # Each currency goes through if Value[0] == Currency and Value[1] != "Buy": # Checks whether this is the correct currency and whether it does not have the status Buy if Value[1] == "Sell": # If yes, it is checked whether Sell is inside order = f"{Currency}:BUY" # If no, it should be bought print("\033[31mBuy", Currency, "because LongRange EMA:", LongRange_EMA, "is lower than current Price", Current_Price, "and it has changed since the last hour.\033[0m") else: # If no, it should not progress any action print("No action. This is the initial price of", Currency) Value[1] = "Buy" # Changes the History to Buy else: if Value[0] == Currency: # If no, it should not progress any action print("No action. The current Price of", Currency, "is still higher than the EMA, it hasn't changed.") return order # This is the main function where the functions are called # The function is executed in another file # First, the price list of the currency is updated # Then the EMAs are calculated, compared, and it is decided whether to buy or sell def strategy(Symbol): if Symbol == "HBARUSDT": # HBARUSDT - Hedera Prices_HBARUSDT = List_Update("HBARUSDT") return Order(EMA_Calculation(Prices_HBARUSDT, LongRange_EMA_Length, Smoothing_Factor), Prices_HBARUSDT[-1], "HBARUSDT") elif Symbol == "BTCUSDT": # BTCUSDT - Bitcoin Prices_BTCUSDT = List_Update("BTCUSDT") return Order(EMA_Calculation(Prices_BTCUSDT, LongRange_EMA_Length, Smoothing_Factor), Prices_BTCUSDT[-1], "BTCUSDT") elif Symbol == "ETHUSDT": # ETHUSDT - Ethereum Prices_ETHUSDT = List_Update("ETHUSDT") return Order(EMA_Calculation(Prices_ETHUSDT, LongRange_EMA_Length, Smoothing_Factor), Prices_ETHUSDT[-1], "ETHUSDT") elif Symbol == "ALGOUSDT": # AGLOUSDT - Algorand Prices_ALGOUSDT = List_Update("ALGOUSDT") return Order(EMA_Calculation(Prices_ALGOUSDT, LongRange_EMA_Length, Smoothing_Factor), Prices_ALGOUSDT[-1], "ALGOUSDT") elif Symbol == "LINKUSDT": # LINKUSDT - Chainlink Prices_LINKUSDT = List_Update("LINKUSDT") return Order(EMA_Calculation(Prices_LINKUSDT, LongRange_EMA_Length, Smoothing_Factor), Prices_LINKUSDT[-1], "LINKUSDT")

from google.oauth2 import service_account from googleapiclient.discovery import build from datetime import datetime import urllib.parse import hashlib import hmac import requests import time # Path to the JSON file with the service account information credentials_file_path = "google-sheets-credentials.json" # Spreadsheet ID (from the document URL) spreadsheet_id = "15mWxtbm7b44fQDpbRrc7aASJ9Zcn9NQfTGU-UqYi33o" # Create the service account credentials credentials = service_account.Credentials.from_service_account_file(credentials_file_path, scopes=["https://www.googleapis.com/auth/spreadsheets"]) # Build the Sheets service service = build("sheets", "v4", credentials=credentials) recv_window = 5000 def get_binanceus_signature(data, secret): postdata = urllib.parse.urlencode(data) message = postdata.encode() byte_key = bytes(secret, 'UTF-8') mac = hmac.new(byte_key, message, hashlib.sha256).hexdigest() return mac def get_all_balances(api_key, secret_key, api_url): uri_path = "/api/v3/account" data = { "timestamp": int(round(time.time() * 1000)), "recvWindow": recv_window } headers = { 'X-MBX-APIKEY': api_key } signature = get_binanceus_signature(data, secret_key) payload = f"{urllib.parse.urlencode(data)}&signature={signature}" try: req = requests.get(f"{api_url}{uri_path}?{payload}", headers=headers) req.raise_for_status() response = req.json() except requests.exceptions.RequestException as e: print(f"Failed to get account balances: {e}") return [] balances = [] if "balances" in response: for item in response["balances"]: asset = item["asset"] free_quantity = float(item["free"]) if free_quantity > 0: balances.append((asset, free_quantity)) return balances def get_current_price(symbol, api_url): url = f"{api_url}/api/v3/ticker/price" params = {"symbol": symbol} try: response = requests.get(url, params=params) response.raise_for_status() data = response.json() return float(data['price']) except requests.exceptions.RequestException as e: print(f"Failed to get current price for {symbol}: {e}") return None def calculate_total_balance_in_usdt(api_key, secret_key, api_url): balances = get_all_balances(api_key, secret_key, api_url) total_usdt = 0.0 for asset, quantity in balances: if asset == "USDT": total_usdt += quantity else: price = get_current_price(f"{asset}USDT", api_url) if price is not None: total_usdt += quantity * price return total_usdt def get_last_row(sheet_name): result = service.spreadsheets().values().get(spreadsheetId=spreadsheet_id, range=f"{sheet_name}!A:A").execute() values = result.get('values', []) return len(values) + 1 def log_transaction(response, action, api_key, secret_key, api_url): sheet_name = "Sheet1" next_row = get_last_row(sheet_name) range_name = f"{sheet_name}!A{next_row}" pair = response['symbol'] price = float(response['fills'][0]['price']) executed_qty = float(response['executedQty']) cummulative_quote_qty = round(float(response['cummulativeQuoteQty']), 2) commission = float(response['fills'][0]['commission']) transact_time = response['transactTime'] total_balance_usdt = calculate_total_balance_in_usdt(api_key, secret_key, api_url) readable_time = datetime.fromtimestamp(transact_time / 1000).strftime('%d.%m.%Y %H:%M:%S') values = [[pair, action, price, executed_qty, commission, cummulative_quote_qty, total_balance_usdt, readable_time]] body = {'values': values} value_input_option = 'RAW' request = service.spreadsheets().values().update(spreadsheetId=spreadsheet_id, range=range_name, valueInputOption=value_input_option, body=body) response = request.execute() print(f"Transaction logged in row {next_row}.")