fisk/core/trading/risk.py

def check_max_loss(risk_model, initial_balance, account_balance):
    """
    Check that the account balance is within the max loss limit.
    """
    max_loss_percent = risk_model.max_loss_percent
    max_loss = initial_balance * (max_loss_percent / 100)
    return account_balance > max_loss


def check_max_risk(risk_model, account_trades):
    """
    Check that all of the trades in the account are within the max risk limit.
    """
    max_risk_percent = risk_model.max_risk_percent
    total_risk = 0
    for trade in account_trades:
        max_tmp = []
        if "sl_percent" in trade:
            max_tmp.append(trade["sl_percent"])
        if "tsl_percent" in trade:
            max_tmp.append(trade["tsl_percent"])
        total_risk += max(max_tmp)
    return total_risk < max_risk_percent


def check_max_open_trades(risk_model, account_trades):
    """
    Check that the number of trades in the account is within the max open trades limit.
    """
    return len(account_trades) < risk_model.max_open_trades


def check_max_open_trades_per_symbol(risk_model, account_trades):
    """
    Check we cannot open more trades per symbol than permissible.
    """
    symbol_map = {}
    for trade in account_trades:
        symbol = trade["symbol"]
        if symbol not in symbol_map:
            symbol_map[symbol] = 0
        symbol_map[symbol] += 1
    for symbol, count in symbol_map.items():
        if count >= risk_model.max_open_trades_per_symbol:
            return False
    return True


def check_risk(risk_model, account, proposed_trade):
    """
    Run the risk checks on the account and the proposed trade.
    """
    # Check that the max loss is not exceeded
    max_loss_check = check_max_loss(
        risk_model, account.initial_balance, account.client.get_balance()
    )
    if not max_loss_check:
        return {"allowed": False, "reason": "Maximum loss exceeded."}

    # Check that the account max trades is not exceeded
    account_trades = account.client.get_all_open_trades()  # TODO
    account_trades.append(proposed_trade)  # TODO

    max_open_trades_check = check_max_open_trades(risk_model, account_trades)
    if not max_open_trades_check:
        return {"allowed": False, "reason": "Maximum open trades exceeded."}

    # Check that the max trades per symbol is not exceeded
    max_open_trades_per_symbol_check = check_max_open_trades_per_symbol(
        risk_model, account_trades
    )
    if not max_open_trades_per_symbol_check:
        return {"allowed": False, "reason": "Maximum open trades per symbol exceeded."}

    # Check that the max risk is not exceeded
    max_risk_check = check_max_risk(risk_model, account_trades)
    if not max_risk_check:
        return {"allowed": False, "reason": "Maximum risk exceeded."}