I used to write singleton in Python, which actually Alpha only need to initialize once and the trading bot will run forever. For example in main.py:

from time import sleep
import asyncio
from collections import deque
from typing import Dict, Union, Any, List
import aiohttp
from lib.utils import Config
from pprint import pprint
from logs.logger import logger
from binance.websocket.spot.websocket_client import SpotWebsocketClient as Client
from binance.um_futures import UMFutures
from redisUtil import redisConsumer
from bybit.usdt_perpetual import USDT_Pepertual

class Alpha(object):
    def __init__(self) -> None:
        self._config = Config()
        self.um_futures_client = UMFutures()
        self.my_client = Client()
        self.strategy_name = self._config.get_config_value("strategy_name")
        self.reference_window = self._config.get_config_value("reference_window")
        self.stop_loss = self._config.get_config_value("stop_loss")
        self.bet_size = self._config.get_config_value("bet_size")
        self.symbol = self._config.get_config_value("symbol")
        self.period = self._config.get_config_value("period")

        self.redis_client_consumer = redisConsumer(host=self._config.get_config_value("host"),
        self.subscription_list = self.update_subscription_list()
        self.current_close_price = 0
        # for storing the close price in a fixed length list
        self.close_price_list = deque([], maxlen=self.reference_window + 1)
        # for storing the funding rate in a fixed length list
        self.funding_rate_list = deque([], maxlen=self.reference_window + 1)

        self.bybit_client = USDT_Pepertual()
        self.__main_logger = logger("logging")

    def update_subscription_list(self) -> List[dict]:
        subscription_list = self.redis_client_consumer.get_subscript(strategy_name=self.strategy_name)
        return subscription_list

    def subscribe_binance_websocket_kline(self):
        self.my_client.kline(symbol=self.symbol, id=1, interval=self.period, callback=self.message_handler)

    def message_handler(self, message: dict) -> None:

        if message:
            if message.get('k', {}).get('x'):
                close_price = float(message.get('k', {}).get('c'))

    def generate_trading_signal(self) -> List[str]:
        signal_list = []
        # delete because contain private trading logic, return list := ["OPEN_LONG", "CLOSE_SHORT]/ ["OPEN_SHORT"]
        return signal_list

    def get_funding_rate(self) -> str:
        funding_rate = self.um_futures_client.mark_price(symbol=self.symbol).get("lastFundingRate")
        return funding_rate

    def store_data(self, close_price: float) -> None:
        funding_rate = self.get_funding_rate()

    def executor(self, close_price: float) -> None:
        if len(self.close_price_list) == self.close_price_list.maxlen:
            signal_list = self.generate_trading_signal()
            if signal_list:

    def _dispatch_request(self, signal: str) -> Dict[str, Union[str, Dict[str, Any]]]:
        match signal:
            case "OPEN_LONG":
                request_detail = self.bybit_client.open_market_long()
            case "OPEN_SHORT":
                request_detail = self.bybit_client.open_market_short()
            case "CLOSE_LONG":
                request_detail = self.bybit_client.close_market_long()
            case "CLOSE_SHORT":
                request_detail = self.bybit_client.close_market_short()
        return request_detail

    def add_size_to_client_list(self, request_dict: dict, position_dict_list: list) -> list:
        current_price = self.close_price_list[-1]
        reduce_only = request_dict.get("body", {}).get("reduce_only")
        if reduce_only == False:
            for subscription in position_dict_list:
                lot_size = int(
                    (float(subscription["invest_amount"]) / current_price * self.bet_size / 100) * 1000) / 1000
                subscription.update({"size": lot_size})
        return position_dict_list

    def _dispatch_order_pre_condition(self, request_dict: dict, request_detail: dict) -> bool:
        reduce_only = request_dict.get("body", {}).get("reduce_only")
        if isinstance(reduce_only, bool):
            side = request_dict["body"]["side"]
            if reduce_only:
                if side == "Buy":
                    return float(request_detail.get("size", 0)) < 0
                elif side == "Sell":
                    return float(request_detail.get("size", 0)) > 0
                return float(request_detail.get("size", -1)) == 0

    def _order_list_filtration(self, request_dict: dict, position_dict_list: list) -> List[dict]:
        filtered_order_list = [request_detail for request_detail in position_dict_list
                               if self._dispatch_order_pre_condition(request_dict=request_dict,
        final_order_list = self.add_size_to_client_list(request_dict=request_dict,
        return final_order_list

    def run(self) -> None:
            while True:
        except BaseException as e:

    async def execute_task(self, request_dict: dict, subscription_list: list = None):
        if subscription_list is None:
            subscription_list = self.subscription_list
        async with aiohttp.ClientSession() as session:
            all_response = await asyncio.gather(*[self.bybit_client.fetch(
            ) for request_detail in subscription_list])
            return all_response

    def logging_log(self, ret_msgs: list) -> None:
        for msg_dict in ret_msgs:

    def trade(self, signal_list: list) -> None:
        for signal in signal_list:
            request_dict = self._dispatch_request(signal=signal)
            # gather all client pre-order position status, with or without position
            position_dict_list = asyncio.run(self.execute_task(request_dict=self.bybit_client.my_position()))
            # if client have long position & signal is LONG, ignore it etc
            final_order_list = self._order_list_filtration(request_dict=request_dict,
            # execute the LONG/SHORT order, return a list of response message
            ret_msgs = asyncio.run(self.execute_task(request_dict=request_dict,
            # pprint(request_dict)
            # logging

if __name__ == "__main__":
    alpha = Alpha()

However, after I read something about singleton pattern in Python:

Python programmers almost never implement the Singleton Pattern. (Gang of Four book)

How should I improve the code in order to aviod the singleton pattern and handle a lot of original instance variables and object instances? Or any things else I could improve in this script? The purpose of this script is to execute a trading bot while receving the data, analysis signal, and execute the trade.

  • \$\begingroup\$ But I only need construct the object once, is it redundant to create an class to wrap all these code? If yes, is there any alternative for doing that? @PierreSchrodinger \$\endgroup\$
    – tung
    Oct 15, 2022 at 1:55

1 Answer 1


You're not using the singleton pattern since there is no logic that prevents you from constructing 2 objects of that type. Creating and using only one object is completely fine. Notable examples are:

app = Flask(__name__)


window = tkinter.Tk()

Classes are not only a way to create multiple entities of the same type, but also an encapsulation mechanism.

Your class does too many things. You should split it into smaller ones following single-responsibility principle.

Things to make separate classes for:

  • Logger
  • Signal Manager
  • ...
funding_rate = self.um_futures_client.mark_price(symbol=self.symbol).get("lastFundingRate")
    return funding_rate

This can be shortened to one string by not declaring a name and straight up returning:

return self.um_futures_client.mark_price(symbol=self.symbol).get("lastFundingRate")

This problem is present at least 4 times in the code, it just loads the brain of the reader with extra names to keep track of for no reason.

Also um_futures_client, funding_rate_list and get_funding_rate() are never actually used by your logic and can be deleted.

if subscription_list is None:
    subscription_list = self.subscription_list

This can be avoided by writing an overload for your method that doesn't take subscription_list parameter:

async def execute_task(self, request_dict: dict):
    execute_task(request_dict=dict, subscription_list=self.subscription_list)

All your comments explain what the code does instead of explaining why. If the code is well written we can figure out what it does just by looking at it, this type of comments is redundant.

You use __ and _ prefixes for some of your fields and methods. These are used as access modifiers, __a means that it can only be used inside this class and _b means that it can be used in all of its subclasses, but not from outside. You use them seemingly randomly which is confusing.

Thumbs up for typing annotations and if __name__ == "__main__":
Good luck!

P.S. The book you are referring to came out almost 30 years ago when python was in version 1.0. I'd recommend this site if you want to learn the topic.

  • \$\begingroup\$ Should I take out some of the content and split these to create different classes, for example: class SignalManager class AlphaLogger ? Also, I am a bit confused with //This can be shortened to one string by not declaring a name and straight up returning.// how can I improve it? Is there an example of writing an overload for updating the subscription_list? I am also confused with __ and _ , is there any examples or suggestion on how should I using these access modifiers? Received with a lot of useful comment and suggestion, thanks very much! \$\endgroup\$
    – tung
    Oct 16, 2022 at 17:28
  • 1
    \$\begingroup\$ 1. Yes. 2. Updated the answer with an example. 3. Updated the answer with an example. 4. Prefix your fields and methods with __ if you don't want other classes to access them. This way you hide implementation details and make the API cleaner. In Python this practice is often ommited. You can learn more here. \$\endgroup\$ Oct 16, 2022 at 19:50
  • \$\begingroup\$ For question 2, you mean I can delete if subscription is None... and add self.subscription_list into self.execute_task()? Also, may I ask is it good practise for initialize other classes inside def __init__() like self.config = Config self.my_client = Client()` or it is better to put these class instance under if __name__ == "__main__"? \$\endgroup\$
    – tung
    Oct 17, 2022 at 12:03
  • 1
    \$\begingroup\$ @tung you can write 2 methods that are called the same but accept different sets of parameters (this is called function overloading). You can remove if subscription_list is None... from your method and create a new one (see the answer) that doesn't take the subscription_list parameter and just calls the main method with self.subscription_list. As for the other question, don't initialize in the __init__ block, since you don't want to create a new instance of e.g. Logger every time your object is created. \$\endgroup\$ Oct 17, 2022 at 13:48
  • 1
    \$\begingroup\$ @tung under if __name__ == "__main__", otherwise it becomes a global variable. \$\endgroup\$ Oct 17, 2022 at 16:08

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.