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
import pandas as pd import numpy as np from datetime import datetime from typing import Dict, List, Tuple, Optional import logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger() expert4x grid trend multiplier
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 def execute_strategy(self, price_data: pd
# Initialize and run strategy strategy = GridTrendMultiplier( initial_balance=10000, grid_distance_pct=0.5, max_grid_levels=8, trend_multiplier=1.5, max_multiplier=4.0, risk_per_trade=0.02 ) volume_data: Optional[pd.Series] = None) ->
def update_multiplier(self, trend_strength: float): """ Update position multiplier based on trend strength """ if trend_strength > 50: # Strong trend - increase multiplier self.total_multiplier = min( self.max_multiplier, self.total_multiplier * self.trend_multiplier ) elif trend_strength < 25: # Weak trend - decrease multiplier self.total_multiplier = max( 1.0, self.total_multiplier / self.trend_multiplier ) def check_grid_execution(self, current_price: float, grid_levels: List[float], atr: float) -> Optional[Dict]: """ Check if price hit a grid level and execute order Returns: Order details if executed, None otherwise """ for level in grid_levels: # Check if price crossed a grid level if abs(current_price - level) / level < 0.0001: # Within 0.01% # Determine direction based on trend if self.current_trend == "BULLISH": direction = "BUY" stop_loss = level * (1 - 0.02) # 2% stop loss take_profit = level * (1 + self.grid_distance_pct / 100) elif self.current_trend == "BEARISH": direction = "SELL" stop_loss = level * (1 + 0.02) take_profit = level * (1 - self.grid_distance_pct / 100) else: # Neutral - alternate direction = "BUY" if len(self.open_positions) % 2 == 0 else "SELL" stop_loss = level * (1 - 0.02) if direction == "BUY" else level * (1 + 0.02) take_profit = level * (1 + self.grid_distance_pct / 100) if direction == "BUY" else level * (1 - self.grid_distance_pct / 100) position_size = self.calculate_position_size(level) order = { 'type': direction, 'entry_price': level, 'position_size': position_size, 'stop_loss': stop_loss, 'take_profit': take_profit, 'timestamp': datetime.now(), 'grid_level': level, 'multiplier': self.total_multiplier } return order return None