stcs/app/helper_functions.py

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, leverage: 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 {}
        maintenance_margin_rate = safe_float(risk_level.get("maintenanceMargin"))

        liq_price_long = entry_price * (1 - 1 / leverage + maintenance_margin_rate)
        liq_price_short = entry_price * (1 + 1 / leverage - maintenance_margin_rate)

        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
) -> 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.

    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)

        r_quantity = set_quantity

        total += r_quantity
    return total