DCAv2Earn/main.py

from libraries.wrappers import earn_binance
from libraries.wrappers import earn_kucoin
from libraries.wrappers import earn_okx
from libraries.wrappers import earn_gateio
from libraries.earner import earner
from libraries.colors import colors
from concurrent.futures import ThreadPoolExecutor, as_completed
from threading import Thread
from flask import Flask, jsonify, request
from waitress import serve
from functools import wraps
import time
import datetime
import json
import sqlite3
import signal
import os


def load_keys_from_db(file_name: str) -> list:
    '''
    Load valid API keys to a set
    
    Parameters
    ----------
    file_name : str
        Name of the database file

    Returns
    -------
    valid_keys : list
        List of valid API keys
    '''
    
    database_connection = sqlite3.connect(file_name)
    database_cursor = database_connection.cursor()
    database_cursor.execute("SELECT * FROM credentials_table")
    data = database_cursor.fetchall()
    database_connection.close()    
    
    valid_keys = {line[1] for line in data}
    
    return valid_keys


def seconds_to_time(total_seconds: float) -> str:
    '''
    Takes an int or float as an input and it returns a D:HH:MM:SS formatted string.
    
    Parameters:
        total_seconds (float): The number of seconds to convert
    
    Returns:
        str: The formatted string
    '''

    time_delta = datetime.timedelta(seconds=total_seconds)

    hours = time_delta.seconds//3600
    remainder = time_delta.seconds%3600
    minutes = remainder//60
    seconds = remainder%60

    return f"{time_delta.days}:{hours:02d}:{minutes:02d}:{seconds:02d}"


def main():
    executor = ThreadPoolExecutor(max_workers=len(earners))
    while True:
        #Run earners
        futures = [executor.submit(e.run) for e in earners]
        for future in as_completed(futures):
            try:
                future.result()
            except Exception as e:
                print(f"Error in thread - {e}")
        
        #Print status
        subscriptions = []
        redemptions = []
        for item in earners:
            print(item.get_status_string())
            subscriptions.append(item.get_last_subscription())
            redemptions.append(item.get_last_redemption())
        last_subscription = max(subscriptions, key=lambda d: next(iter(d.values())))
        last_subscription_key = next(iter(last_subscription.keys()))
        last_subscription_value = next(iter(last_subscription.values()))
        if last_subscription_value == 0:
            last_subscription_key = "Never"
            last_subscription_value = ""
        else:
            last_subscription_value = datetime.datetime.fromtimestamp(last_subscription_value).strftime('@%Y/%m/%d %H:%M:%S')
        last_redemption = max(redemptions, key=lambda d: next(iter(d.values())))
        last_redemption_key = next(iter(last_redemption.keys()))
        last_redemption_value = next(iter(last_redemption.values()))
        if last_redemption_value == 0:
            last_redemption_key = "Never"
            last_redemption_value = ""
        else:
            last_redemption_value = datetime.datetime.fromtimestamp(last_redemption_value).strftime('@%Y/%m/%d %H:%M:%S')
        print("-"*90)
        total_on_trading = sum([item.get_trading_balance() for item in earners])
        total_on_earning = sum([item.get_earning_balance() for item in earners])
        total_funds = total_on_earning+total_on_trading
        time_of_day = datetime.datetime.now().strftime('[%Y/%m/%d %H:%M:%S]')
        print(f"Last subscription: {last_subscription_key}{last_subscription_value} | Last redemption: {last_redemption_key}{last_redemption_value}")
        print(f"Version {version} | Total funds: {total_funds:.2f} USDT | On earn: {total_on_earning:.2f} USDT ({total_on_earning/total_funds*100:.2f}%)")
        print(f"{time_of_day} | Uptime: {seconds_to_time(time.time()-start_time)}")
        print(colors.blue+"="*90+colors.white)
        
        #Wait for next lap
        time.sleep(config["lap_time"])
        


#########################
######### API ###########
#########################

earn_api = Flask(__name__)

#Helper functions
def require_api_key(func):
    '''
    Validates API key
    '''
    @wraps(func)
    def wrapper(*args, **kwargs):
        key = request.headers.get("X-API-KEY")
        if not key or key not in valid_keys:
            return jsonify({'Error': 'API key not valid'}), 401
        return func(*args, **kwargs)
    return wrapper

@earn_api.route("/get_global_status", methods=['GET'])
@require_api_key
def get_global_status():
    '''
    GET request
    
    Parameters:
        None
    ''' 
       
    response = {}
    for item in earners:
        response[str(item.connector)] = {"currency": item.get_currency(),
                                            "trading_balance": item.get_trading_balance(),
                                            "earning_balance": item.get_earning_balance(),
                                            "is_paused": item.get_is_paused(),
                                            "step_size": item.get_step_size(),
                                            "percentage": item.get_percentage(),
                                            "minimum_amount_in_trading_account": item.get_minimum_amount_in_trading_account(),
                                            "time_between_subscriptions": item.get_time_between_subscriptions(),
                                            "time_between_redemptions": item.get_time_between_redemptions(),
                                            "last_subscription": item.get_last_subscription(),
                                            "last_redemption": item.get_last_redemption()}
        response["uptime"] = time.time() - start_time
    return jsonify(response)


@earn_api.route("/toggle_pause", methods=['POST'])
@require_api_key
def toggle_pause():
    '''
    GET request
    
    Parameters:
        broker: str
    ''' 
         
    if request.json is None:
        return jsonify({'Error': 'request.json is None'})    
    broker = request.json["broker"]
    if broker is None:
        return jsonify({'Error': 'broker is None'})
    if broker not in valid_broker_list:
        return jsonify({'Error': 'broker not valid'})
    for item in earners:
        if str(item.connector)==broker:
            item.toggle_pause()
            return jsonify({'Status': item.get_is_paused()})
    return jsonify({'Error': 'broker not found'})


@earn_api.route("/get_step_size", methods=["GET"])
@require_api_key
def get_step_size():
    '''
    GET request
    
    Parameters:
        broker: str
    '''    
        
    broker = request.args.get("broker")
    if broker is None:
        return jsonify({'Error': 'broker is None'})
    if broker not in valid_broker_list:
        return jsonify({'Error': 'broker not valid'})        
    for item in earners:
        if str(item.connector)==broker:
            return jsonify({'step_size': item.get_step_size()})
    return jsonify({'Error': 'broker not found'})
        

@earn_api.route("/set_step_size", methods=["POST"])
@require_api_key
def set_step_size():  
    if request.json is None:
        return jsonify({'Error': 'request.json is None'})
    broker = request.json["broker"]
    new_step_size = request.json["new_step_size"]
    if broker is None:
        return jsonify({'Error': 'broker is None'})
    if broker not in valid_broker_list:
        return jsonify({'Error': 'broker not valid'})        
    if new_step_size is None:
        return jsonify({'Error': 'new_step_size is None'})
    for item in earners:
        if str(item.connector)==broker:
            item.set_step_size(new_step_size)
            return jsonify({'step_size': new_step_size})
    return jsonify({'Error': 'broker not found'})


@earn_api.route("/get_percentage", methods=["GET"])     
@require_api_key   
def get_percentage():
    broker = request.args.get("broker")
    if broker is None:
        return jsonify({'Error': 'broker is None'})
    if broker not in valid_broker_list:
        return jsonify({'Error': 'broker not valid'})        
    for item in earners:
        if str(item.connector)==broker:
            return jsonify({'percentage': item.get_percentage()})
    return jsonify({'Error': 'broker not found'})


@earn_api.route("/set_percentage", methods=["POST"])
@require_api_key
def set_percentage():  
    if request.json is None:
        return jsonify({'Error': 'request.json is None'})
    broker = request.json["broker"]
    new_percentage = request.json["new_percentage"]
    if broker is None:
        return jsonify({'Error': 'broker is None'})
    if broker not in valid_broker_list:
        return jsonify({'Error': 'broker not valid'})        
    if new_percentage is None:
        return jsonify({'Error': 'new_step_size is None'})
    for item in earners:
        if str(item.connector)==broker:
            item.set_percentage(new_percentage)
            return jsonify({'percentage': new_percentage})
    return jsonify({'Error': 'broker not found'})
            

@earn_api.route("/get_time_between_subscriptions", methods=["GET"])   
@require_api_key     
def get_time_between_subscriptions():   
    broker = request.args.get("broker")
    if broker is None:
        return jsonify({'Error': 'broker is None'})
    if broker not in valid_broker_list:
        return jsonify({'Error': 'broker not valid'})        
    for item in earners:
        if str(item.connector)==broker:
            return jsonify({'time_between_subscriptions': item.get_time_between_subscriptions()})
    return jsonify({'Error': 'broker not found'})


@earn_api.route("/set_time_between_subscriptions", methods=["POST"])
@require_api_key
def set_time_between_subscriptions():
    if request.json is None:
        return jsonify({'Error': 'request.json is None'})
    broker = request.json["broker"]
    new_time_between_subscriptions = request.json["new_time_between_subscriptions"]
    if broker is None:
        return jsonify({'Error': 'broker is None'})
    if broker not in valid_broker_list:
        return jsonify({'Error': 'broker not valid'})        
    if new_time_between_subscriptions is None:
        return jsonify({'Error': 'new_step_size is None'})
    for item in earners:
        if str(item.connector)==broker:
            item.set_time_between_subscriptions(new_time_between_subscriptions)
            return jsonify({'time_between_subscriptions': new_time_between_subscriptions})
    return jsonify({'Error': 'broker not found'})


@earn_api.route("/get_time_between_redemptions", methods=["GET"])
@require_api_key
def get_time_between_redemptions():
    broker = request.args.get("broker")
    if broker is None:
        return jsonify({'Error': 'broker is None'})
    if broker not in valid_broker_list:
        return jsonify({'Error': 'broker not valid'})        
    for item in earners:
        if str(item.connector)==broker:
            return jsonify({'time_between_redemptions': item.get_time_between_redemptions()})
    return jsonify({'Error': 'broker not found'})


@earn_api.route("/set_time_between_redemptions", methods=["POST"])
@require_api_key
def set_time_between_redemptions():
    if request.json is None:
        return jsonify({'Error': 'request.json is None'})
    broker = request.json["broker"]
    new_time_between_redemptions = request.json["new_time_between_redemptions"]
    if broker is None:
        return jsonify({'Error': 'broker is None'})
    if broker not in valid_broker_list:
        return jsonify({'Error': 'broker not valid'})        
    if new_time_between_redemptions is None:
        return jsonify({'Error': 'new_step_size is None'})
    for item in earners:
        if str(item.connector)==broker:
            item.set_time_between_redemptions(new_time_between_redemptions)
            return jsonify({'time_between_redemptions': new_time_between_redemptions})
    return jsonify({'Error': 'broker not found'})


@earn_api.route("/get_minimum_amount_in_trading_account", methods=["GET"])
@require_api_key
def get_minimum_amount_in_trading_account():
    broker = request.args.get("broker")
    if broker is None:
        return jsonify({'Error': 'broker is None'})
    if broker not in valid_broker_list:
        return jsonify({'Error': 'broker not valid'})        
    for item in earners:
        if str(item.connector)==broker:
            return jsonify({'minimum_amount_in_trading_account': item.get_minimum_amount_in_trading_account()})
    return jsonify({'Error': 'broker not found'})


@earn_api.route("/set_minimum_amount_in_trading_account", methods=["POST"])
@require_api_key
def set_minimum_amount_in_trading_account():
    broker = request.json["broker"]
    new_minimum_amount_in_trading_account = request.json["new_minimum_amount_in_trading_account"]
    if broker is None:
        return jsonify({'Error': 'broker is None'})
    if broker not in valid_broker_list:
        return jsonify({'Error': 'broker not valid'})        
    if new_minimum_amount_in_trading_account is None:
        return jsonify({'Error': 'new_step_size is None'})
    for item in earners:
        if str(item.connector)==broker:
            item.set_minimum_amount_in_trading_account(new_minimum_amount_in_trading_account)
            return jsonify({'minimum_amount_in_trading_account': new_minimum_amount_in_trading_account})
    return jsonify({'Error': 'broker not found'})


@earn_api.route("/get_last_subscription", methods=["GET"])
@require_api_key
def get_last_subscription():
    broker = request.args.get("broker")
    if broker is None:
        return jsonify({'Error': 'broker is None'})
    if broker not in valid_broker_list:
        return jsonify({'Error': 'broker not valid'})           
    for item in earners:
        if str(item.connector)==broker:
            return jsonify({'last_subscription': item.get_last_subscription()})
    return jsonify({'Error': 'broker not found'})


@earn_api.route("/get_last_redemption", methods=["GET"])
@require_api_key
def get_last_redemption():
    broker = request.args.get("broker")
    if broker is None:
        return jsonify({'Error': 'broker is None'})
    if broker not in valid_broker_list:
        return jsonify({'Error': 'broker not valid'})           
    for item in earners:
        if str(item.connector)==broker:
            return jsonify({'last_redemption': item.get_last_redemption()})
    return jsonify({'Error': 'broker not found'})


@earn_api.route("/get_total_balance", methods=["GET"])
@require_api_key
def get_total_balance():
    broker = request.args.get("broker")
    if broker is None:
        return jsonify({'Error': 'broker is None'})
    if broker not in valid_broker_list:
        return jsonify({'Error': 'broker not valid'})           
    for item in earners:
        if str(item.connector)==broker:
            return jsonify({'trading_balance': item.get_trading_balance(), 'earning_balance': item.get_earning_balance()})
    return jsonify({'Error': 'broker not found'})


@earn_api.route("/subscribe", methods=["POST"])
@require_api_key
def subscribe():
    '''
    args:
        broker: broker name
        amount: amount to subscribe
        force_pause: True or False
    '''
    if request.json is None:
        return jsonify({'Error': 'request.json is None'})
    broker = request.json["broker"]
    amount = request.json["amount"]
    force_pause = False
    if "force_pause" in request.json:
        force_pause = request.json["force_pause"]
    if broker is None:
        return jsonify({'Error': 'broker is None'})
    if broker not in valid_broker_list:
        return jsonify({'Error': 'broker not valid'})
    if amount is None:
        return jsonify({'Error': 'amount is None'})
    if force_pause is None:
        return jsonify({'Error': 'force_pause is None'})
    for item in earners:
        if str(item.connector)==broker:
            subscription = item.subscribe(amount, force_pause=force_pause)
            return jsonify({'Success': str(subscription)})
    return jsonify({'Error': 'broker not found'})
     

@earn_api.route("/redeem", methods=["POST"])
@require_api_key
def redeem():
    '''
    args:
        broker: broker name
        amount: amount to redeem
        force_pause: True or False
    '''
    if request.json is None:
        return jsonify({'Error': 'request.json is None'})
    broker = request.json["broker"]
    amount = request.json["amount"]
    force_pause = False
    if "force_pause" in request.json:
        force_pause = request.json["force_pause"]
    if broker is None:
        return jsonify({'Error': 'broker is None'})
    if broker not in valid_broker_list:
        return jsonify({'Error': 'broker not valid'})
    if amount is None:
        return jsonify({'Error': 'amount is None'})
    if force_pause is None:
        return jsonify({'Error': 'force_pause is None'})
    for item in earners:
        if str(item.connector)==broker:
            redemption = item.redeem(amount, force_pause=force_pause)
            return jsonify({'Success': str(redemption)})
    return jsonify({'Error': 'broker not found'})

            
            
def run_API(port):
    serve(earn_api, host="0.0.0.0", port=port)
    #earn_api.run(host="0.0.0.0", port=port)


executor = None
#Shutdown handler
def shutdown_handler(signum, _):
    print(f"Received signal {signum}, shutting down as gracefully as possible...")
    if executor:
        executor.shutdown(wait=True, timeout=5)
    os._exit(0)

# Register signals for shutdown handler
signal.signal(signal.SIGINT, shutdown_handler)
signal.signal(signal.SIGTERM, shutdown_handler)


if __name__=="__main__":

    version = "2025.10.11"
    start_time = time.time()
    
    with open("config.json") as f:
        config = json.load(f)
    
    connectors = {"binance": earn_binance.binance_earn(),
                  "gateio": earn_gateio.gateio_earn(),
                  "kucoin": earn_kucoin.kucoin_earn(),
                  "okx": earn_okx.okx_earn()}
    earners = []
    
    for item in config["exchanges"]:
        earners.append(earner(connectors[item], config["exchanges"][item]))
    
    #Valid broker list
    valid_broker_list = [str(item.connector) for item in earners]
    
    #Load valid API keys
    valid_keys = load_keys_from_db("keys/api_credentials.db")

    #Threads to run: main loop and flask api
    api_thread = Thread(target=run_API, args=(config["api_port"],), daemon=True)

    #Iterate indefinitely:
    api_thread.start()
        
    try:
        main()
    except KeyboardInterrupt:
        api_thread.join()
        shutdown_handler(signal.SIGINT, None)