Trend Multiplier — Expert4x Grid
# 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 )
def calculate_grid_levels(self, current_price: float, atr_value: float) -> List[float]: """ Calculate dynamic grid levels based on ATR Args: current_price: Current market price atr_value: Current ATR value Returns: List of grid price levels """ grid_spacing = max( current_price * (self.grid_distance_pct / 100), atr_value * 0.5 # Minimum half of ATR ) levels = [] for i in range(1, self.max_grid_levels + 1): # Calculate multiplier with trend bias multiplier = self.total_multiplier if self.current_trend == "BULLISH": up_level = current_price + (grid_spacing * i * multiplier) down_level = current_price - (grid_spacing * i * (1 / multiplier)) elif self.current_trend == "BEARISH": up_level = current_price + (grid_spacing * i * (1 / multiplier)) down_level = current_price - (grid_spacing * i * multiplier) else: up_level = current_price + (grid_spacing * i) down_level = current_price - (grid_spacing * i) levels.extend([up_level, down_level]) return sorted(levels) expert4x grid trend multiplier
print("\n" + "="*50) print("GRID TREND MULTIPLIER STRATEGY RESULTS") print("="*50) for key, value in metrics.items(): if isinstance(value, float): print(f"{key.replace('_', ' ').title()}: {value:.2f}") else: print(f"{key.replace('_', ' ').title()}: {value}") return strategy, metrics if == " main ": strategy, metrics = run_backtest() risk_per_trade=0.02 ) def calculate_grid_levels(self
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) atr_value: float) ->
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()
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
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