107 lines
3.7 KiB
Python
107 lines
3.7 KiB
Python
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}
|