The entire database has been changed to PostgresSQL. The entire code has been updated.

app/helper_functions.py

@@ -0,0 +1,196 @@
+import logging.config
+
+from app.bybit import get_bybit_client
+from logger_helper.logger_helper import LOGGING_CONFIG
+
+logging.config.dictConfig(LOGGING_CONFIG)
+logger = logging.getLogger("helper_functions")
+
+
+def safe_float(val) -> float:
+    """
+    Function to safely convert string to float
+    """
+    try:
+        if val is None or val == "":
+            return 0.0
+        return float(val)
+    except (ValueError, TypeError):
+        logger.error("Error converting value to float: %s", val)
+        return 0.0
+
+
+def is_number(value: str) -> bool:
+    """
+    Checks if a given string represents a number.
+
+    Args:
+        value (str): The string to check.
+
+    Returns:
+        bool: True if the string represents a number, False otherwise.
+    """
+    try:
+        # Convert the string to a float
+        num = float(value)
+        # Check if the number is positive
+        if num <= 0:
+            return False
+        # Check if the string contains "+" or "-"
+        if "+" in value or "-" in value:
+            return False
+        # Check if the string contains only digits
+        allowed_chars = set("0123456789.")
+        if not all(ch in allowed_chars for ch in value):
+            return False
+        return True
+    except ValueError:
+        return False
+
+
+def is_int(value: str) -> bool:
+    """
+    Checks if a given string represents an integer.
+
+    Args:
+        value (str): The string to check.
+
+    Returns:
+        bool: True if the string represents an integer, False otherwise.
+    """
+    # Check if the string contains only digits
+    if not value.isdigit():
+        return False
+    # Convert the string to an integer
+    num = int(value)
+    return num > 0
+
+
+def is_int_for_timer(value: str) -> bool | int:
+    """
+    Checks if a given string represents an integer for timer.
+
+    Args:
+        value (str): The string to check.
+
+    Returns:
+        bool: True if the string represents an integer, False otherwise.
+    """
+    # Check if the string contains only digits
+    try:
+        num = int(value)
+
+        if num >= 0:
+            return num
+        else:
+            return False
+    except ValueError:
+        return False
+
+
+def get_base_currency(symbol: str) -> str:
+    """
+    Extracts the base currency from a symbol string.
+
+    Args:
+        symbol (str): The symbol string to extract the base currency from.
+
+    Returns:
+        str: The base currency extracted from the symbol string.
+    """
+    if symbol.endswith("USDT"):
+        return symbol[:-4]
+    return symbol
+
+
+def safe_int(value, default=0) -> int:
+    """
+    Integer conversion with default value.
+    """
+    try:
+        return int(value)
+    except (ValueError, TypeError):
+        return default
+
+
+def format_value(value) -> str:
+    """
+    Function to format value
+    """
+    if not value or value.strip() == "":
+        return "Нет данных"
+    return value
+
+
+def check_limit_price(limit_price, min_price, max_price) -> str | None:
+    """
+    Function to check limit price
+    """
+    if limit_price < min_price:
+        return "Limit price is out min price"
+    if limit_price > max_price:
+        return "Limit price is out max price"
+    return None
+
+
+async def get_liquidation_price(
+    tg_id: int, symbol: str, entry_price: float, order_quantity: float
+) -> tuple[float, float]:
+    """
+    Function to get liquidation price
+    """
+    try:
+        client = await get_bybit_client(tg_id=tg_id)
+        get_risk_info = client.get_risk_limit(category="linear", symbol=symbol)
+        risk_list = get_risk_info.get("result", {}).get("list", [])
+        risk_level = risk_list[0] if risk_list else {}
+        maint_margin_rate = safe_float(risk_level.get("maintenanceMargin"))
+        maint_margin_deduction = safe_float(risk_level.get("initialMargin"))
+        max_leverage = safe_float(risk_level.get("maxLeverage"))
+        position_value = order_quantity / entry_price
+        initial_margin = position_value / max_leverage
+        maint_margin = (position_value * maint_margin_rate) - maint_margin_deduction
+
+        liq_price_long = order_quantity / (
+            position_value + (initial_margin - maint_margin)
+        )
+
+        liq_price_short = order_quantity / (
+            position_value - (initial_margin - maint_margin)
+        )
+
+        liq_price = liq_price_long, liq_price_short
+
+        return liq_price
+    except Exception as e:
+        logger.error("Error getting liquidation price: %s", e)
+        return 0, 0
+
+
+async def calculate_total_budget(
+    quantity, martingale_factor, max_steps, commission_fee_percent
+) -> float:
+    """
+    Calculate the total budget for a series of trading steps.
+
+    Args:
+        quantity (float): The initial quantity of the asset.
+        martingale_factor (float): The factor by which the quantity is multiplied for each step.
+        max_steps (int): The maximum number of trading steps.
+        commission_fee_percent (float): The commission fee percentage.
+
+    Returns:
+        float: The total budget for the series of trading steps.
+    """
+    total = 0
+    for step in range(max_steps):
+        set_quantity = quantity * (martingale_factor**step)
+        if commission_fee_percent == 0:
+            # Commission fee is not added to the position size
+            r_quantity = set_quantity
+        else:
+            # Commission fee is added to the position size
+            r_quantity = set_quantity * (1 + 2 * commission_fee_percent)
+
+        total += r_quantity
+    return total