Core Features: - Dynamic grid levels based on ATR - Trend detection using multiple timeframes - Position size multiplier based on trend strength - Martingale-style recovery with risk management - Auto grid adjustment during strong trends """

def reset_strategy(self): """ Reset strategy to initial state """ self.balance = self.initial_balance self.grid_levels = [] self.open_positions = [] self.closed_trades = [] self.current_trend = "NEUTRAL" self.trend_strength = 0 self.total_multiplier = 1.0 self.total_trades = 0 self.winning_trades = 0 self.losing_trades = 0 self.max_drawdown = 0 self.peak_balance = self.initial_balance logger.info("Strategy reset to initial state") def run_backtest(): """ Run backtest with sample data """ # Generate sample price data np.random.seed(42) dates = pd.date_range('2023-01-01', periods=1000, freq='1H') price = 100 prices = []

for i in range(1000): price += np.random.randn() * 0.5 if i > 200 and i < 600: # Uptrend price += 0.1 elif i > 600: # Downtrend price -= 0.05 prices.append(max(price, 10)) df = pd.DataFrame({ 'high': [p * (1 + abs(np.random.randn() * 0.002)) for p in prices], 'low': [p * (1 - abs(np.random.randn() * 0.002)) for p in prices], 'close': prices }, index=dates)

def calculate_position_size(self, price: float, stop_loss_pct: float = 0.02) -> float: """ Calculate position size based on trend multiplier and risk management Args: price: Entry price stop_loss_pct: Stop loss percentage Returns: Position size in units """ # Base risk amount risk_amount = self.balance * self.risk_per_trade # Apply trend multiplier if self.current_trend == "BULLISH": position_multiplier = self.total_multiplier elif self.current_trend == "BEARISH": position_multiplier = self.total_multiplier else: position_multiplier = 1.0 # Calculate position size stop_loss_distance = price * stop_loss_pct position_size = (risk_amount * position_multiplier) / stop_loss_distance # Cap position size based on available balance max_position = self.balance * 0.1 / price # Max 10% of balance per trade position_size = min(position_size, max_position) return position_size

metrics = strategy.execute_strategy(df)

def execute_strategy(self, price_data: pd.DataFrame, volume_data: Optional[pd.Series] = None) -> Dict: """ Main strategy execution loop Args: price_data: DataFrame with 'high', 'low', 'close' columns volume_data: Optional volume series Returns: Strategy performance metrics """ logger.info("Starting Grid Trend Multiplier Strategy") for i in range(len(price_data)): current_close = price_data['close'].iloc[i] current_high = price_data['high'].iloc[i] current_low = price_data['low'].iloc[i] # Use enough data for indicators if i < 50: continue # Get price series up to current point price_series = price_data['close'].iloc[:i+1] # Detect trend self.current_trend, self.trend_strength = self.detect_trend(price_series) # Update multiplier based on trend strength self.update_multiplier(self.trend_strength) # Calculate ATR atr_series = self.calculate_atr( price_data['high'].iloc[:i+1], price_data['low'].iloc[:i+1], price_data['close'].iloc[:i+1] ) current_atr = atr_series.iloc[-1] if not pd.isna(atr_series.iloc[-1]) else current_close * 0.01 # Calculate grid levels self.grid_levels = self.calculate_grid_levels(current_close, current_atr) # Check for grid execution order = self.check_grid_execution(current_close, self.grid_levels, current_atr) if order: self.open_positions.append(order) logger.info(f"Order executed: {order['type']} at {order['entry_price']:.4f} " f"with multiplier {order['multiplier']:.2f}") # Update existing positions closed_trades = self.update_positions(current_close) if closed_trades: for trade in closed_trades: logger.info(f"Trade closed: {trade['result']} with profit ${trade['profit']:.2f}") # Calculate final metrics metrics = self.get_performance_metrics() return metrics