from datetime import datetime from decimal import Decimal as D from core.lib.notify import sendmsg from core.util import logs log = logs.get_logger("checks") def run_checks(strategy, x): pass def within_trading_times(strategy, ts=None): if not ts: ts = datetime.utcnow() # Check if we can trade now! trading_times = strategy.trading_times.all() if not trading_times: log.error("No trading times set for strategy") return False matches = [x.within_range(ts) for x in trading_times] if not any(matches): log.debug("Not within trading time range") return False return True def within_callback_price_deviation(strategy, price, current_price): # Convert the callback price deviation to a ratio if strategy.risk_model is not None: callback_price_deviation_as_ratio = D( strategy.risk_model.callback_price_deviation_percent ) / D(100) else: callback_price_deviation_as_ratio = D(0.5) / D(100) log.debug(f"Callback price deviation as ratio: {callback_price_deviation_as_ratio}") maximum_price_deviation = D(current_price) * D(callback_price_deviation_as_ratio) # Ensure the current price is within price_slippage_as_ratio of the callback price if abs(current_price - price) <= maximum_price_deviation: log.debug("Current price is within price deviation of callback price") return True else: log.error("Current price is not within price deviation of callback price") log.debug(f"Difference: {abs(current_price - price)}") return False def within_max_loss(strategy): pass def within_trends(strategy, symbol, direction): if strategy.trend_signals.exists(): if strategy.trends is None: log.debug("Refusing to trade with no trend signals received") sendmsg( strategy.user, f"Refusing to trade {symbol} with no trend signals received", title="Trend not ready", ) return None if symbol not in strategy.trends: log.debug("Refusing to trade asset without established trend") sendmsg( strategy.user, f"Refusing to trade {symbol} without established trend", title="Trend not ready", ) return None else: if strategy.trends[symbol] != direction: log.debug("Refusing to trade against the trend") sendmsg( strategy.user, f"Refusing to trade {symbol} against the trend", title="Trend rejection", ) return False else: log.debug(f"Trend check passed for {symbol} - {direction}") return True else: log.debug("No trend signals configured") return True def within_position_size(strategy): pass def within_protection(strategy): pass def within_max_open_trades(strategy): pass def within_max_open_trades_per_asset(strategy): pass def within_max_risk(strategy): pass def within_crossfilter(strategy): pass