from decimal import Decimal as D from core.exchanges import convert from core.models import Trade from core.trading import market 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 # Calculate the inverse of the max loss percent as a ratio inverse_loss_multiplier = 1 - max_loss_percent / 100 minimum_balance = initial_balance * inverse_loss_multiplier return account_balance > minimum_balance def check_max_risk(risk_model, account_balance_usd, account_trades): """ Check that all of the trades in the account are within the max risk limit. """ max_risk_percent = D(risk_model.max_risk_percent) # Calculate the max risk of the account in USD max_risk_usd = account_balance_usd * (max_risk_percent / D(100)) total_risk = 0 for trade in account_trades: max_tmp = [] # Need to calculate the max risk in base account currency # Percentages relate to the price movement, without accounting the # size of the trade if "stop_loss_usd" in trade: max_tmp.append(trade["stop_loss_usd"]) if "trailing_stop_loss_usd" in trade: max_tmp.append(trade["trailing_stop_loss_usd"]) if max_tmp: total_risk += max(max_tmp) allowed = total_risk < max_risk_usd return allowed 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() if isinstance(proposed_trade, Trade): proposed_trade = proposed_trade.__dict__ account_trades.append(proposed_trade) account_trades = convert.convert_trades(account_trades) account_trades = market.convert_trades_to_usd(account, account_trades) 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 account_balance_usd = account.client.get_balance(return_usd=True) max_risk_check = check_max_risk(risk_model, account_balance_usd, account_trades) if not max_risk_check: return {"allowed": False, "reason": "Maximum risk exceeded."} return {"allowed": True}