Expert4x Grid Trend Multiplier [ RELIABLE 2024 ]
def detect_trend(self, prices: pd.Series, volume: Optional[pd.Series] = None) -> Tuple[str, float]: """ Detect market trend using multiple indicators Returns: (trend_direction, trend_strength) """ # Calculate EMAs ema_fast = prices.ewm(span=20, adjust=False).mean() ema_slow = prices.ewm(span=50, adjust=False).mean() # Calculate ADX for trend strength high = prices.rolling(window=14).max() low = prices.rolling(window=14).min() plus_dm = high.diff() minus_dm = -low.diff() plus_dm[plus_dm < 0] = 0 minus_dm[minus_dm < 0] = 0 tr = self.calculate_atr( high, low, prices ) if hasattr(self, 'calculate_atr') else pd.Series(index=prices.index) plus_di = 100 * (plus_dm.rolling(14).mean() / tr) minus_di = 100 * (minus_dm.rolling(14).mean() / tr) dx = 100 * abs(plus_di - minus_di) / (plus_di + minus_di) adx = dx.rolling(14).mean() # Determine trend current_ema_fast = ema_fast.iloc[-1] current_ema_slow = ema_slow.iloc[-1] current_adx = adx.iloc[-1] if not pd.isna(adx.iloc[-1]) else 25 if current_ema_fast > current_ema_slow and current_adx > 25: trend = "BULLISH" trend_strength = min(100, current_adx) elif current_ema_fast < current_ema_slow and current_adx > 25: trend = "BEARISH" trend_strength = min(100, current_adx) else: trend = "NEUTRAL" trend_strength = 0 return trend, trend_strength
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 expert4x grid trend multiplier
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 detect_trend(self, prices: pd
def __init__(self, initial_balance: float = 10000, grid_distance_pct: float = 0.5, max_grid_levels: int = 10, trend_multiplier: float = 1.5, max_multiplier: float = 5.0, atr_period: int = 14, risk_per_trade: float = 0.02): """ Initialize Grid Trend Multiplier Args: initial_balance: Starting account balance grid_distance_pct: Distance between grid levels (% of price) max_grid_levels: Maximum grid levels trend_multiplier: Position size multiplier for trend direction max_multiplier: Maximum allowed multiplier atr_period: ATR calculation period risk_per_trade: Risk per trade (2% = 0.02) """ self.initial_balance = initial_balance self.balance = initial_balance self.grid_distance_pct = grid_distance_pct self.max_grid_levels = max_grid_levels self.trend_multiplier = trend_multiplier self.max_multiplier = max_multiplier self.atr_period = atr_period self.risk_per_trade = risk_per_trade # Strategy state self.grid_levels = [] self.open_positions = [] self.closed_trades = [] self.current_trend = "NEUTRAL" # BULLISH, BEARISH, NEUTRAL self.trend_strength = 0 # 0-100 self.total_multiplier = 1.0 # Performance metrics self.total_trades = 0 self.winning_trades = 0 self.losing_trades = 0 self.max_drawdown = 0 self.peak_balance = initial_balance def calculate_atr(self, high: pd.Series, low: pd.Series, close: pd.Series) -> pd.Series: """Calculate Average True Range""" tr1 = high - low tr2 = abs(high - close.shift()) tr3 = abs(low - close.shift()) tr = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1) atr = tr.rolling(window=self.atr_period).mean() return atr volume: Optional[pd.Series] = None) ->
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()