diff --git a/back_tester/enhanced_risk_management.py b/back_tester/enhanced_risk_management.py deleted file mode 100644 index 0bcc571..0000000 --- a/back_tester/enhanced_risk_management.py +++ /dev/null @@ -1,151 +0,0 @@ -""" -Enhanced risk management module for the backtester. - -This module provides advanced risk management functionality: -- Dynamic ATR-based stop losses -- Adaptive position sizing -- Improved risk-reward calculations -""" - -import pandas as pd -import numpy as np -from typing import Dict, Union - - -def calculate_tighter_stop_loss( - df: pd.DataFrame, - index: int, - entry_price: float, - signal_type: str, - atr_period: int = 14, - baseline_multiplier: float = 1.5, - min_distance_percent: float = 0.005, - max_distance_percent: float = 0.02, -) -> float: - """ - Calculate a tighter, smarter stop loss based on ATR and market structure. - - Args: - df: Price data DataFrame with OHLC - index: Current candle index - entry_price: Trade entry price - signal_type: 'Bullish' or 'Bearish' - atr_period: Period for ATR calculation - baseline_multiplier: Base ATR multiplier - min_distance_percent: Minimum distance for stop loss as percent of price - max_distance_percent: Maximum distance for stop loss as percent of price - - Returns: - Calculated stop loss price - """ - if index < atr_period: - # fallback if not enough data - stop_distance = entry_price * max_distance_percent - return ( - entry_price - stop_distance - if signal_type == "Bullish" - else entry_price + stop_distance - ) - - # ATR - high_low = ( - df["High"].iloc[index - atr_period : index] - - df["Low"].iloc[index - atr_period : index] - ) - high_close = np.abs( - df["High"].iloc[index - atr_period : index] - - df["Close"].iloc[index - atr_period - 1 : index - 1] - ) - low_close = np.abs( - df["Low"].iloc[index - atr_period : index] - - df["Close"].iloc[index - atr_period - 1 : index - 1] - ) - - true_range = pd.concat([high_low, high_close, low_close], axis=1).max(axis=1) - atr = true_range.mean() - - # Adjust ATR multiplier based on recent volatility - recent_volatility = ( - df["Close"].iloc[index - 10 : index].std() - / df["Close"].iloc[index - 10 : index].mean() - ) - volatility_factor = 1.0 - - if recent_volatility > 0.015: # High volatility - volatility_factor = 1.2 - elif recent_volatility < 0.005: # Low volatility - volatility_factor = 0.8 - - # Calculate stop loss distance with adjusted multiplier - stop_distance = atr * baseline_multiplier * volatility_factor - - # Apply min/max constraints - min_stop_distance = entry_price * min_distance_percent - max_stop_distance = entry_price * max_distance_percent - - stop_distance = max(min_stop_distance, min(stop_distance, max_stop_distance)) - - # Calculate stop loss price based on signal type - if signal_type == "Bullish": - stop_loss = entry_price - stop_distance - else: - stop_loss = entry_price + stop_distance - - return stop_loss - - -def calculate_dynamic_position_size( - balance: float, - risk_percentage: float, - entry_price: float, - stop_loss: float, - market_volatility: float = 1.0, -) -> Dict[str, Union[float, str]]: - """ - Calculate position size with dynamic risk adjustment based on market conditions. - - Args: - balance: Account balance - risk_percentage: Base percentage of account to risk - entry_price: Entry price for the trade - stop_loss: Stop loss price - market_volatility: Volatility factor (>1 means higher volatility) - - Returns: - Dictionary with position size and risk details - """ - # Adjust risk percentage based on market volatility - adjusted_risk = risk_percentage - - if market_volatility > 1.5: # High volatility - adjusted_risk = risk_percentage * 0.7 # Reduce risk - elif market_volatility < 0.7: # Low volatility - adjusted_risk = risk_percentage * 1.2 # Increase risk slightly - - # Ensure risk stays within reasonable bounds - adjusted_risk = min(adjusted_risk, risk_percentage * 1.5) - adjusted_risk = max(adjusted_risk, risk_percentage * 0.5) - - # Calculate dollar risk amount - risk_amount = balance * (adjusted_risk / 100.0) - - # Calculate risk per unit - risk_per_unit = abs(entry_price - stop_loss) - - if risk_per_unit <= 0: - return { - "position_size": 0.0, - "risk_amount": 0.0, - "risk_percentage": 0.0, - "error": "Invalid risk: entry price and stop loss are too close", - } - - # Calculate position size - position_size = risk_amount / risk_per_unit - - return { - "position_size": position_size, - "risk_amount": risk_amount, - "risk_percentage": adjusted_risk, - "error": None, - } diff --git a/back_tester/enhanced_trade_management.py b/back_tester/enhanced_trade_management.py deleted file mode 100644 index c3acaf6..0000000 --- a/back_tester/enhanced_trade_management.py +++ /dev/null @@ -1,192 +0,0 @@ -""" -Enhanced trade management module for the backtester. - -This module provides advanced trade management functionality: -- Trailing stops -- Time-based exits -- Volatility-based take profits -""" - -import numpy as np -import pandas as pd -from typing import Tuple - - -def calculate_trailing_stop( - entry_price: float, - current_price: float, - initial_stop: float, - trail_percent: float = 0.5, - activation_threshold: float = 1.0, -) -> float: - """ - Calculate a trailing stop price. - - Args: - entry_price: The entry price of the trade - current_price: The current price of the asset - initial_stop: The initial stop loss price - trail_percent: What percentage of profits to protect (0-1) - activation_threshold: How far price needs to move in profit (R multiple) to activate trailing - - Returns: - The updated stop loss price - """ - initial_risk = abs(entry_price - initial_stop) - profit_distance = current_price - entry_price - risk_multiple = profit_distance / initial_risk if initial_risk > 0 else 0 - - # Only activate trailing stop if price has moved **up** sufficiently in profit - if risk_multiple >= activation_threshold: - risk_to_protect = profit_distance * trail_percent - new_stop = entry_price + risk_to_protect - - return max(new_stop, initial_stop) - else: - return initial_stop - - -def calculate_atr_based_stops( - df: pd.DataFrame, multiplier: float = 1.5, period: int = 14 -) -> pd.Series: - """ - Calculate ATR-based stop loss values. - - Args: - df: DataFrame with OHLC data - multiplier: ATR multiplier for stop distance - period: Period for ATR calculation - - Returns: - Series of stop loss values - """ - high_low = df["High"] - df["Low"] - high_close = np.abs(df["High"] - df["Close"].shift()) - low_close = np.abs(df["Low"] - df["Close"].shift()) - - true_range = pd.concat([high_low, high_close, low_close], axis=1).max(axis=1) - atr = true_range.rolling(window=period).mean() - - # Fill NaN values with a reasonable default (1% of price) - if atr.isna().all(): - atr = df["Close"] * 0.01 - else: - atr = atr.bfill().ffill() - if atr.isna().any(): - atr = atr.fillna(df["Close"] * 0.01) - - # Calculate stop loss distances based on ATR - return atr * multiplier - - -def calculate_volatility_based_take_profits( - *args, **kwargs -) -> Tuple[float, float, float]: - """ - Calculate volatility-adjusted take profit levels. - - Function supports two calling methods for backward compatibility: - - Method 1 (recommended): - entry_price: The entry price of the trade - stop_loss: The stop loss price - atr: The current ATR value - signal_type: "Bullish" or "Bearish" - risk_reward_ratios: Base risk-reward ratios for the three targets (tuple of 3 floats) - - Method 2 (legacy): - df: DataFrame with OHLC data - i: Current index in dataframe - current_price: Current price - signal: "Bullish" or "Bearish" - atr_period: Period for ATR calculation - tp_multipliers: List of TP multipliers - min/max_distance_percent: Min/max TP distance - - Returns: - Tuple of three take profit levels - """ - # Check which calling method is being used - if len(args) > 0 and isinstance(args[0], pd.DataFrame): - # Legacy method with dataframe - df = args[0] - i = args[1] - current_price = args[2] - signal_type = args[3] - - # Extract parameters from kwargs - atr_period = kwargs.get("atr_period", 14) - tp_multipliers = kwargs.get("tp_multipliers", [2.0, 3.5, 5.0]) - - # Calculate ATR for volatility - atr_values = calculate_atr_based_stops( - df.iloc[: i + 1], multiplier=1.0, period=atr_period - ) - atr = atr_values.iloc[-1] if not atr_values.empty else current_price * 0.01 - - # Calculate stop loss if not provided - if "stop_loss" in kwargs: - stop_loss = kwargs["stop_loss"] - else: - stop_distance = atr * 1.5 # Default multiplier - if signal_type == "Bullish": - stop_loss = current_price * 0.95 # Default 5% below entry - else: - stop_loss = current_price * 1.05 # Default 5% above entry - - entry_price = current_price - risk_reward_ratios = tuple(tp_multipliers) - - else: - # New method with direct parameters - entry_price = kwargs.get("entry_price", args[0]) - stop_loss = kwargs.get("stop_loss", args[1]) - atr = kwargs.get("atr", args[2]) - signal_type = kwargs.get("signal_type", args[3]) - risk_reward_ratios = kwargs.get("risk_reward_ratios", (1.5, 2.5, 3.5)) - if len(args) > 4: - risk_reward_ratios = args[4] - - # Calculate initial risk - risk = abs(entry_price - stop_loss) - - # Adjust risk-reward ratio based on volatility - volatility_factor = atr / (entry_price * 0.01) # Normalize to 1% of price - adjusted_rr = [ - ( - max(rr - (0.2 * (volatility_factor - 1)), rr * 0.7) - if volatility_factor > 1 - else rr - ) - for rr in risk_reward_ratios - ] - - # Calculate take profit levels - if signal_type == "Bullish": - tp1 = entry_price + (risk * adjusted_rr[0]) - tp2 = entry_price + (risk * adjusted_rr[1]) - tp3 = entry_price + (risk * adjusted_rr[2]) - else: - tp1 = entry_price - (risk * adjusted_rr[0]) - tp2 = entry_price - (risk * adjusted_rr[1]) - tp3 = entry_price - (risk * adjusted_rr[2]) - - return tp1, tp2, tp3 - - -def should_exit_based_on_time( - entry_index: int, current_index: int, max_trade_duration: int = 24 -) -> bool: - """ - Determine if a trade should be exited based on time duration. - - Args: - entry_index: The candle index when the trade was entered - current_index: The current candle index - max_trade_duration: Maximum duration to hold a trade in candles - - Returns: - Boolean indicating whether to exit the trade - """ - trade_duration = current_index - entry_index - return trade_duration >= max_trade_duration diff --git a/back_tester/market_filters.py b/back_tester/market_filters.py deleted file mode 100644 index e75227d..0000000 --- a/back_tester/market_filters.py +++ /dev/null @@ -1,301 +0,0 @@ -""" -Market filters module for the backtester. - -This module provides functionality to filter trading opportunities based on: -- Market regime analysis -- Time-of-day filtering -- Enhanced entry confirmation criteria -""" - -import pandas as pd -import numpy as np -from typing import Dict, List, Tuple, Union - - -def detect_market_regime( - df: pd.DataFrame, - lookback_period: int = 20, - atr_period: int = 14, - rsi_period: int = 14, - atr_threshold: float = 1.5, - rsi_overbought: float = 70, - rsi_oversold: float = 30, -) -> Dict[str, Union[str, float]]: - """ - Detect the current market regime based on price action and indicators. - - Args: - df: Price data DataFrame with OHLC - lookback_period: Period to analyze for regime detection - atr_period: Period for ATR calculation - rsi_period: Period for RSI calculation - atr_threshold: Threshold for ATR ratio to detect volatility - rsi_overbought: RSI threshold for overbought - rsi_oversold: RSI threshold for oversold - - Returns: - Dictionary with market regime details - """ - if len(df) < lookback_period + atr_period: - return {"regime": "unknown", "strength": 0.0, "details": "Not enough data"} - - # Use the last lookback_period candles for analysis - analysis_window = df.iloc[-lookback_period:] - - # Calculate price direction - start_price = analysis_window.iloc[0]["Close"] - end_price = analysis_window.iloc[-1]["Close"] - price_change = (end_price - start_price) / start_price - - # Calculate ATR for volatility - high_low = analysis_window["High"] - analysis_window["Low"] - high_close = np.abs(analysis_window["High"] - analysis_window["Close"].shift()) - low_close = np.abs(analysis_window["Low"] - analysis_window["Close"].shift()) - - true_range = pd.concat([high_low, high_close, low_close], axis=1).max(axis=1) - atr = true_range.rolling(window=atr_period).mean().fillna(true_range.mean()) - - # Calculate average ATR as percentage of price - avg_atr_pct = atr.mean() / analysis_window["Close"].mean() - - # Calculate historical ATR for comparison - if len(df) > lookback_period + atr_period: - historical_window = df.iloc[-(lookback_period + atr_period) : -lookback_period] - historical_high_low = historical_window["High"] - historical_window["Low"] - historical_high_close = np.abs( - historical_window["High"] - historical_window["Close"].shift() - ) - historical_low_close = np.abs( - historical_window["Low"] - historical_window["Close"].shift() - ) - - historical_tr = pd.concat( - [historical_high_low, historical_high_close, historical_low_close], axis=1 - ).max(axis=1) - historical_atr = historical_tr.mean() - historical_atr_pct = historical_atr / historical_window["Close"].mean() - - atr_ratio = avg_atr_pct / historical_atr_pct if historical_atr_pct > 0 else 1.0 - else: - atr_ratio = 1.0 - - # Calculate RSI - delta = analysis_window["Close"].diff() - gain = delta.where(delta > 0, 0).fillna(0) - loss = -delta.where(delta < 0, 0).fillna(0) - - avg_gain = gain.rolling(window=rsi_period).mean().fillna(0) - avg_loss = loss.rolling(window=rsi_period).mean().fillna(0) - - rs = avg_gain / avg_loss.replace(0, 1e-9) # Avoid division by zero - rsi = 100 - (100 / (1 + rs)) - current_rsi = rsi.iloc[-1] - - # Calculate price range as percentage - price_range = ( - analysis_window["High"].max() - analysis_window["Low"].min() - ) / analysis_window["Low"].min() - - # Determine market regime - if atr_ratio > atr_threshold: - # Volatile market - regime = "volatile" - strength = min(atr_ratio / atr_threshold, 3.0) / 3.0 # Normalize to 0-1 - elif abs(price_change) > 0.05: # 5% price change - # Trending market - if price_change > 0: - regime = "trending_up" - else: - regime = "trending_down" - strength = min(abs(price_change) / 0.05, 3.0) / 3.0 # Normalize to 0-1 - elif price_range < 0.02: # Very tight range - regime = "quiet" - strength = 1.0 - (price_range / 0.02) # Higher strength for tighter ranges - else: - # Ranging market - regime = "ranging" - strength = min(price_range / 0.02, 2.0) / 2.0 # Normalize to 0-1 - - return { - "regime": regime, - "strength": strength, - "price_change": price_change, - "atr_ratio": atr_ratio, - "rsi": current_rsi, - "price_range": price_range, - } - - -def should_trade_in_regime( - market_regime: Dict[str, Union[str, float]], - signal_type: str, - min_regime_strength: float = 0.6, - allow_volatile: bool = False, -) -> Tuple[bool, str]: - """ - Determine if we should trade in the current market regime. - - Args: - market_regime: Dictionary with market regime details - signal_type: 'Bullish' or 'Bearish' - min_regime_strength: Minimum regime strength to consider valid - allow_volatile: Whether to allow trading in volatile markets - - Returns: - Tuple of (should_trade, reason) - """ - regime = market_regime.get("regime", "unknown") - strength = market_regime.get("strength", 0.0) - rsi = market_regime.get("rsi", 50.0) - - # Check regime strength - if strength < min_regime_strength and regime != "quiet": - return False, f"Weak {regime} regime (strength: {strength:.2f})" - - # Check for regime and signal alignment - if regime == "trending_up" and signal_type == "Bullish": - return True, f"Strong uptrend (strength: {strength:.2f})" - - elif regime == "trending_down" and signal_type == "Bearish": - return True, f"Strong downtrend (strength: {strength:.2f})" - - elif regime == "volatile": - if not allow_volatile: - return False, f"Avoiding volatile market (strength: {strength:.2f})" - # If we do allow volatile markets, check for extremes - if signal_type == "Bullish" and rsi < 30: - return True, f"Volatile market, oversold condition (RSI: {rsi:.2f})" - elif signal_type == "Bearish" and rsi > 70: - return True, f"Volatile market, overbought condition (RSI: {rsi:.2f})" - else: - return False, f"Volatile market without extreme RSI condition" - - elif regime == "ranging": - # In ranging markets, trade counter-trend - if signal_type == "Bullish" and rsi < 40: - return True, f"Ranging market, lower range (RSI: {rsi:.2f})" - elif signal_type == "Bearish" and rsi > 60: - return True, f"Ranging market, upper range (RSI: {rsi:.2f})" - else: - return False, f"Ranging market without favorable RSI position" - - elif regime == "quiet": - return False, "Market too quiet, avoiding low volatility" - - return False, f"Regime {regime} not aligned with {signal_type} signal" - - -def is_favorable_trading_time( - timestamp: pd.Timestamp, timezone: str = "UTC", favorable_hours: List[int] = None -) -> Tuple[bool, str]: - """ - Determine if the current time is favorable for trading. - - Args: - timestamp: The timestamp to check - timezone: Timezone to convert timestamp to - favorable_hours: List of hours (0-23) considered favorable for trading - - Returns: - Tuple of (is_favorable, reason) - """ - if favorable_hours is None: - # Default to major market hours (rough approximation) - favorable_hours = [1, 2, 3, 8, 9, 10, 11, 12, 13, 14, 15, 20, 21, 22] - - # Convert timestamp to specified timezone - local_time = timestamp.tz_localize("UTC").tz_convert(timezone) - hour = local_time.hour - day_of_week = local_time.dayofweek # 0=Monday, 6=Sunday - - # Check if weekend - if day_of_week >= 5: # Saturday or Sunday - return False, f"Weekend trading (day {day_of_week+1})" - - # Check hour - if hour in favorable_hours: - return True, f"Favorable trading hour ({hour}:00)" - - return False, f"Unfavorable trading hour ({hour}:00)" - - -def confirm_entry_criteria( - df: pd.DataFrame, - index: int, - signal_type: str, - confidence: float, - minimum_confidence: float = 0.6, -) -> Tuple[bool, str]: - """ - Apply enhanced entry confirmation criteria. - - Args: - df: Price data DataFrame - index: Current candle index - signal_type: 'Bullish' or 'Bearish' - confidence: Signal confidence value - minimum_confidence: Minimum confidence required - - Returns: - Tuple of (criteria_met, reason) - """ - # Check basic confidence threshold - if confidence < minimum_confidence: - return False, f"Low confidence ({confidence:.2f} < {minimum_confidence:.2f})" - - # Make sure we have enough historical data - if index < 20: - return False, "Not enough historical data" - - # Recent price action - recent_candles = df.iloc[index - 5 : index] - - # Check for trend confirmation in recent candles - if signal_type == "Bullish": - # Check if most recent close is above most recent open - last_candle_bullish = df["Close"].iloc[index - 1] > df["Open"].iloc[index - 1] - - # Check if at least 3 of last 5 candles are bullish - bullish_count = sum(recent_candles["Close"] > recent_candles["Open"]) - trend_confirmed = bullish_count >= 3 and last_candle_bullish - - if not trend_confirmed: - return ( - False, - f"Bullish signal not confirmed by recent price action ({bullish_count}/5 bullish candles)", - ) - - elif signal_type == "Bearish": - # Check if most recent close is below most recent open - last_candle_bearish = df["Close"].iloc[index - 1] < df["Open"].iloc[index - 1] - - # Check if at least 3 of last 5 candles are bearish - bearish_count = sum(recent_candles["Close"] < recent_candles["Open"]) - trend_confirmed = bearish_count >= 3 and last_candle_bearish - - if not trend_confirmed: - return ( - False, - f"Bearish signal not confirmed by recent price action ({bearish_count}/5 bearish candles)", - ) - - # Volume confirmation - # Check if volume is increasing - recent_volume = ( - df["Volume"].iloc[index - 5 : index] if "Volume" in df.columns else None - ) - - if recent_volume is not None: - avg_recent_volume = recent_volume.mean() - avg_prior_volume = ( - df["Volume"].iloc[index - 10 : index - 5].mean() - if index >= 10 - else avg_recent_volume - ) - - volume_increasing = avg_recent_volume > avg_prior_volume - - if not volume_increasing: - return False, "Volume not confirming the signal" - - return True, "All entry criteria confirmed" diff --git a/back_tester/strategy.py b/back_tester/strategy.py index 83ac44c..3850a48 100644 --- a/back_tester/strategy.py +++ b/back_tester/strategy.py @@ -8,33 +8,23 @@ project_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) sys.path.append(project_dir) +import pandas as pd # type: ignore # These imports use the system path we added above from data_fetching_instruments import fetch_candles, analyze_data from signal_detection import ( generate_price_prediction_signal_proba, + TradingSignal, calculate_position_size, ) +import sys +import os project_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) if project_dir not in sys.path: sys.path.append(project_dir) - from utils import create_true_preferences from .db_operations import ClickHouseDB -from .enhanced_trade_management import ( - calculate_trailing_stop, - should_exit_based_on_time, - calculate_volatility_based_take_profits, - calculate_atr_based_stops, -) -from .enhanced_risk_management import calculate_tighter_stop_loss -from .market_filters import ( - detect_market_regime, - should_trade_in_regime, - is_favorable_trading_time, - confirm_entry_criteria, -) def backtest_strategy( @@ -46,36 +36,6 @@ def backtest_strategy( liq_lev_tolerance: float = 0.05, weights: list = [], risk_percentage: float = 1.0, - # Trade entry/exit parameters - use_trailing_stop: bool = True, - trail_activation_threshold: float = 1.0, - trail_percent: float = 0.5, - use_tighter_stops: bool = True, - atr_stop_multiplier: float = 1.5, - min_stop_distance_percent: float = 0.005, - max_stop_distance_percent: float = 0.02, - max_trade_duration: int = 20, - # Market regime parameters - use_market_regime_filter: bool = True, - min_regime_strength: float = 0.6, - allow_volatile_regime: bool = False, - # Time of day parameters - use_time_filter: bool = True, - trading_timezone: str = "UTC", - favorable_hours: List[int] = None, - # Entry confirmation parameters - strengthen_entry_criteria: bool = True, - minimum_confidence: float = 0.7, - # Volatility-based take profit parameters - use_volatility_based_tps: bool = True, - atr_tp_multipliers: List[float] = None, # Multipliers for ATR to set TPs - min_tp_distance_percent: float = 0.01, # Minimum TP distance as percent - max_tp_distance_percent: float = 0.05, # Maximum TP distance as percent - # Multi-stage exit parameters - use_partial_exits: bool = True, - tp1_exit_percentage: float = 0.33, # Percentage of position to exit at TP1 - tp2_exit_percentage: float = 0.50, # Percentage of remaining position to exit at TP2 - tp3_exit_percentage: float = 1.0, # Percentage of remaining position to exit at TP3 iteration_id: Optional[str] = None, db: Optional[ClickHouseDB] = None, ) -> Tuple[float, list, Optional[str]]: @@ -153,381 +113,198 @@ def backtest_strategy( if position > 0 and current_trade: # Check take profit levels if ( - current_trade["take_profit_1"] is not None - and current_price >= current_trade["take_profit_1"] + current_price >= current_trade["take_profit_1"] and not current_trade["tp1_hit"] ): + # Close 1/3 of position at TP1 + close_amount = current_trade["initial_position"] / 3 + profit = close_amount * (current_price - entry_price) + balance += close_amount * current_price + position -= close_amount current_trade["tp1_hit"] = True - # Calculate amount to close based on configured percentage - if use_partial_exits: - # Close the configured percentage of initial position at TP1 - close_amount = ( - current_trade["initial_position"] * tp1_exit_percentage - ) - close_amount = min( - close_amount, position - ) # Don't close more than we have - - profit = close_amount * (current_price - entry_price) - balance += close_amount * current_price - position -= close_amount - - trade_log.append( - { - "type": "take_profit_1", - "price": current_price, - "index": i, - "signal": f"Take Profit 1 - {tp1_exit_percentage*100:.0f}% position closed", - "timestamp": current_time, - "amount": close_amount, - "remaining": position, - "profit": profit, - } - ) - print( - f"[Index {i}] TP1 hit at {current_price:.5f} - Closed {close_amount:.5f} units ({tp1_exit_percentage*100:.0f}%) - Remaining: {position:.5f}" - ) - else: - # Close entire position if partial exits are disabled - close_amount = position - profit = close_amount * (current_price - entry_price) - balance += close_amount * current_price - position = 0 - - trade_log.append( - { - "type": "take_profit_1", - "price": current_price, - "index": i, - "signal": "Take Profit 1 - 100% position closed", - "timestamp": current_time, - "amount": close_amount, - "profit": profit, - } - ) - print( - f"[Index {i}] TP1 hit at {current_price:.5f} - Closed entire position ({close_amount:.5f} units)" - ) - - # Reset trade tracking since position is closed - entry_price = None - entry_time = None - entry_index = None - entry_signal = None - current_trade = None - parent_trade_id = None + trade_log.append( + { + "type": "take_profit_1", + "price": current_price, + "index": i, + "signal": "Take Profit 1 - 33% position closed", + "timestamp": current_time, + "amount": close_amount, + "profit": profit, + } + ) + print( + f"[Index {i}] TP1 hit at {current_price:.5f} - Closed {close_amount:.5f} units" + ) # Store TP1 trade in database if db: - trade_id = ( - str(uuid.uuid4()) if use_partial_exits else parent_trade_id - ) - trade_data = { - "trade_id": trade_id, - "iteration_id": str(iteration_id), - "sub_iteration_id": str(sub_iteration_id), - "symbol": str(symbol), - "interval": str(interval), + "iteration_id": iteration_id, + "sub_iteration_id": sub_iteration_id, + "symbol": symbol, + "interval": interval, "trade_type": "take_profit_1", "entry_timestamp": entry_time, "exit_timestamp": current_time, - "entry_index": int(entry_index), - "exit_index": int(i), - "entry_price": float(entry_price), - "exit_price": float(current_price), - "profit_loss": float(profit), - "trade_duration": int(i - entry_index), - "entry_signal": str(entry_signal), + "entry_index": entry_index, + "exit_index": i, + "entry_price": entry_price, + "exit_price": current_price, + "profit_loss": profit, + "trade_duration": i - entry_index, + "entry_signal": entry_signal, "exit_signal": "Take Profit 1", - "risk_reward_ratio": float(current_trade["risk_reward_ratio"]), - "position_size": float(close_amount), - "stop_loss": float(current_trade["stop_loss"]), - "take_profit_1": float(current_trade["take_profit_1"]), - "take_profit_2": float(current_trade["take_profit_2"]), - "take_profit_3": float(current_trade["take_profit_3"]), - "risk_percentage": float(risk_percentage), - "amount_traded": float(close_amount * current_price), - "parent_trade_id": ( - parent_trade_id if use_partial_exits else None - ), + "risk_reward_ratio": current_trade["risk_reward_ratio"], + "position_size": close_amount, + "stop_loss": current_trade["stop_loss"], + "take_profit_1": current_trade["take_profit_1"], + "take_profit_2": current_trade["take_profit_2"], + "take_profit_3": current_trade["take_profit_3"], + "risk_percentage": risk_percentage, + "amount_traded": close_amount * current_price, + "parent_trade_id": parent_trade_id, } - - if use_partial_exits: - # For partial exits, insert a new trade record - db.insert_trade(trade_data) - else: - # For full exit, update the original trade record - db.update_trade( - parent_trade_id, - { - "exit_timestamp": current_time, - "exit_index": int(i), - "exit_price": float(current_price), - "profit_loss": float(profit), - "trade_duration": int(i - entry_index), - "exit_signal": "Take Profit 1", - }, - ) + db.insert_trade(trade_data) elif ( - current_trade["take_profit_2"] is not None - and current_price >= current_trade["take_profit_2"] - and current_trade["tp1_hit"] + current_price >= current_trade["take_profit_2"] and not current_trade["tp2_hit"] - and position > 0 ): + # Close 1/2 of remaining position at TP2 + close_amount = position / 2 + profit = close_amount * (current_price - entry_price) + balance += close_amount * current_price + position -= close_amount current_trade["tp2_hit"] = True - # Calculate amount to close at TP2 - if use_partial_exits: - # Calculate percentage of remaining position to close - remaining_percentage = tp2_exit_percentage - close_amount = position * remaining_percentage - close_amount = min( - close_amount, position - ) # Don't close more than we have - - profit = close_amount * (current_price - entry_price) - balance += close_amount * current_price - position -= close_amount - - trade_log.append( - { - "type": "take_profit_2", - "price": current_price, - "index": i, - "signal": f"Take Profit 2 - {remaining_percentage*100:.0f}% of remaining position closed", - "timestamp": current_time, - "amount": close_amount, - "remaining": position, - "profit": profit, - } - ) - print( - f"[Index {i}] TP2 hit at {current_price:.5f} - Closed {close_amount:.5f} units ({remaining_percentage*100:.0f}% of remaining) - Remaining: {position:.5f}" - ) - else: - # Close entire position if partial exits are disabled - close_amount = position - profit = close_amount * (current_price - entry_price) - balance += close_amount * current_price - position = 0 - - trade_log.append( - { - "type": "take_profit_2", - "price": current_price, - "index": i, - "signal": "Take Profit 2 - 100% position closed", - "timestamp": current_time, - "amount": close_amount, - "profit": profit, - } - ) - print( - f"[Index {i}] TP2 hit at {current_price:.5f} - Closed entire position ({close_amount:.5f} units)" - ) - - # Reset trade tracking since position is closed - entry_price = None - entry_time = None - entry_index = None - entry_signal = None - current_trade = None - parent_trade_id = None + trade_log.append( + { + "type": "take_profit_2", + "price": current_price, + "index": i, + "signal": "Take Profit 2 - 50% of remaining position closed", + "timestamp": current_time, + "amount": close_amount, + "profit": profit, + } + ) + print( + f"[Index {i}] TP2 hit at {current_price:.5f} - Closed {close_amount:.5f} units" + ) # Store TP2 trade in database if db: - # Generate a new trade ID for this partial exit if using partial exits - trade_id = ( - str(uuid.uuid4()) if use_partial_exits else parent_trade_id - ) - trade_data = { - "trade_id": trade_id, - "iteration_id": str(iteration_id), - "sub_iteration_id": str(sub_iteration_id), - "symbol": str(symbol), - "interval": str(interval), + "iteration_id": iteration_id, + "sub_iteration_id": sub_iteration_id, + "symbol": symbol, + "interval": interval, "trade_type": "take_profit_2", "entry_timestamp": entry_time, "exit_timestamp": current_time, - "entry_index": int(entry_index), - "exit_index": int(i), - "entry_price": float(entry_price), - "exit_price": float(current_price), - "profit_loss": float(profit), - "trade_duration": int(i - entry_index), - "entry_signal": str(entry_signal), + "entry_index": entry_index, + "exit_index": i, + "entry_price": entry_price, + "exit_price": current_price, + "profit_loss": profit, + "trade_duration": i - entry_index, + "entry_signal": entry_signal, "exit_signal": "Take Profit 2", - "risk_reward_ratio": float(current_trade["risk_reward_ratio"]), - "position_size": float(close_amount), - "stop_loss": float(current_trade["stop_loss"]), - "take_profit_1": float(current_trade["take_profit_1"]), - "take_profit_2": float(current_trade["take_profit_2"]), - "take_profit_3": float(current_trade["take_profit_3"]), - "risk_percentage": float(risk_percentage), - "amount_traded": float(close_amount * current_price), - "parent_trade_id": ( - parent_trade_id if use_partial_exits else None - ), + "risk_reward_ratio": current_trade["risk_reward_ratio"], + "position_size": close_amount, + "stop_loss": current_trade["stop_loss"], + "take_profit_1": current_trade["take_profit_1"], + "take_profit_2": current_trade["take_profit_2"], + "take_profit_3": current_trade["take_profit_3"], + "risk_percentage": risk_percentage, + "amount_traded": close_amount * current_price, + "parent_trade_id": parent_trade_id, } + db.insert_trade(trade_data) - if use_partial_exits: - # For partial exits, insert a new trade record - db.insert_trade(trade_data) - else: - # For full exit, update the original trade record - db.update_trade( - parent_trade_id, - { - "exit_timestamp": current_time, - "exit_index": int(i), - "exit_price": float(current_price), - "profit_loss": float(profit), - "trade_duration": int(i - entry_index), - "exit_signal": "Take Profit 2", - }, - ) - - # Check TP3 - always closes the full remaining position elif ( - current_trade["take_profit_3"] is not None - and current_price >= current_trade["take_profit_3"] - and current_trade["tp1_hit"] - and current_trade["tp2_hit"] + current_price >= current_trade["take_profit_3"] and not current_trade["tp3_hit"] - and position > 0 ): - current_trade["tp3_hit"] = True - - # Always close the full remaining position at TP3 + # Close remaining position at TP3 close_amount = position profit = close_amount * (current_price - entry_price) - balance += close_amount * current_price - position = 0 + balance += position * current_price trade_log.append( { "type": "take_profit_3", "price": current_price, "index": i, - "signal": "Take Profit 3 - 100% remaining position closed", + "signal": "Take Profit 3 - Remaining position closed", "timestamp": current_time, - "amount": close_amount, + "amount": position, "profit": profit, } ) print( - f"[Index {i}] TP3 hit at {current_price:.5f} - Closed remaining position ({close_amount:.5f} units)" + f"[Index {i}] TP3 hit at {current_price:.5f} - Closed {position:.5f} units" ) - # Reset trade tracking since position is fully closed - entry_price = None - entry_time = None - entry_index = None - entry_signal = None - parent_trade_id = None - # Store TP3 trade in database - if db and parent_trade_id: - # Always close the parent trade when TP3 is hit + if db: trade_data = { + "iteration_id": iteration_id, + "sub_iteration_id": sub_iteration_id, + "symbol": symbol, + "interval": interval, + "trade_type": "take_profit_3", + "entry_timestamp": entry_time, "exit_timestamp": current_time, - "exit_index": int(i), - "exit_price": float(current_price), - "profit_loss": float(profit), - "trade_duration": int(i - entry_index), + "entry_index": entry_index, + "exit_index": i, + "entry_price": entry_price, + "exit_price": current_price, + "profit_loss": profit, + "trade_duration": i - entry_index, + "entry_signal": entry_signal, "exit_signal": "Take Profit 3", + "risk_reward_ratio": current_trade["risk_reward_ratio"], + "position_size": close_amount, + "stop_loss": current_trade["stop_loss"], + "take_profit_1": current_trade["take_profit_1"], + "take_profit_2": current_trade["take_profit_2"], + "take_profit_3": current_trade["take_profit_3"], + "risk_percentage": risk_percentage, + "amount_traded": close_amount * current_price, + "parent_trade_id": parent_trade_id, } - db.update_trade(parent_trade_id, trade_data) - - # Also insert a TP3 trade record if using partial exits - if use_partial_exits: - tp3_trade_id = str(uuid.uuid4()) - tp3_trade_data = { - "trade_id": tp3_trade_id, - "iteration_id": str(iteration_id), - "sub_iteration_id": str(sub_iteration_id), - "symbol": str(symbol), - "interval": str(interval), - "trade_type": "take_profit_3", - "entry_timestamp": entry_time, - "exit_timestamp": current_time, - "entry_index": int(entry_index), - "exit_index": int(i), - "entry_price": float(entry_price), - "exit_price": float(current_price), - "profit_loss": float(profit), - "trade_duration": int(i - entry_index), - "entry_signal": str(entry_signal), - "exit_signal": "Take Profit 3", - "risk_reward_ratio": float( - current_trade["risk_reward_ratio"] - ), - "position_size": float(close_amount), - "stop_loss": float(current_trade["stop_loss"]), - "take_profit_1": float(current_trade["take_profit_1"]), - "take_profit_2": float(current_trade["take_profit_2"]), - "take_profit_3": float(current_trade["take_profit_3"]), - "risk_percentage": float(risk_percentage), - "amount_traded": float(close_amount * current_price), - "parent_trade_id": parent_trade_id, - } - db.insert_trade(tp3_trade_data) + db.insert_trade(trade_data) - # Reset all trade tracking since position is fully closed at TP3 - current_trade = None position = 0 entry_price = None entry_time = None entry_index = None entry_signal = None + current_trade = None parent_trade_id = None - # Update trailing stop if enabled - if use_trailing_stop and current_trade and current_price > entry_price: - updated_stop = calculate_trailing_stop( - entry_price, - current_price, - current_trade["stop_loss"], - trail_percent, - trail_activation_threshold, - ) - current_trade["stop_loss"] = updated_stop - - # Check time-based exit - time_exit_triggered = should_exit_based_on_time( - entry_index, i, max_trade_duration - ) - # Check stop loss - if current_price <= current_trade["stop_loss"] or time_exit_triggered: - # Close entire position at stop loss or time-based exit + elif current_price <= current_trade["stop_loss"]: + # Close entire position at stop loss loss = position * (current_price - entry_price) balance += position * current_price - exit_reason = ( - "Stop Loss" - if current_price <= current_trade["stop_loss"] - else "Time-Based Exit" - ) - trade_log.append( { - "type": exit_reason.lower().replace(" ", "_"), + "type": "stop_loss", "price": current_price, "index": i, - "signal": f"{exit_reason} - Position closed", + "signal": "Stop Loss - Position closed", "timestamp": current_time, "amount": position, "profit": loss, } ) print( - f"[Index {i}] {exit_reason} at {current_price:.5f} - Profit/Loss: {loss:.2f}" + f"[Index {i}] Stop Loss hit at {current_price:.5f} - Loss: {loss:.2f}" ) # Store stop loss trade in database @@ -537,7 +314,7 @@ def backtest_strategy( "sub_iteration_id": sub_iteration_id, "symbol": symbol, "interval": interval, - "trade_type": exit_reason.lower().replace(" ", "_"), + "trade_type": "stop_loss", "entry_timestamp": entry_time, "exit_timestamp": current_time, "entry_index": entry_index, @@ -547,7 +324,7 @@ def backtest_strategy( "profit_loss": loss, "trade_duration": i - entry_index, "entry_signal": entry_signal, - "exit_signal": exit_reason, + "exit_signal": "Stop Loss", "risk_reward_ratio": current_trade["risk_reward_ratio"], "position_size": position, "stop_loss": current_trade["stop_loss"], @@ -569,379 +346,49 @@ def backtest_strategy( parent_trade_id = None # Handle new signal - if signal in ["Bullish", "Bearish"] and position == 0 and trading_signal: - # Market regime filter disabled as it's too restrictive - # We'll keep the code but bypass the filter logic - if False and use_market_regime_filter: # Added False to bypass - # Detect current market regime - market_regime = detect_market_regime( - df.iloc[: i + 1], lookback_period=20 - ) - should_trade, regime_reason = should_trade_in_regime( - market_regime, - signal, - min_regime_strength=min_regime_strength, - allow_volatile=allow_volatile_regime, - ) - - if not should_trade: - print( - f"[Index {i}] Skipping trade due to market regime: {regime_reason}" - ) - continue - - # Time filter disabled as it's too restrictive - # We'll keep the code but bypass the filter logic - if False and use_time_filter: # Added False to bypass - # Set default favorable hours if not provided - if favorable_hours is None: - # Default to common active market hours (UTC) - favorable_hours = [ - 1, - 2, - 3, - 8, - 9, - 10, - 11, - 12, - 13, - 14, - 15, - 20, - 21, - 22, - ] - - favorable_time, time_reason = is_favorable_trading_time( - current_time, - timezone=trading_timezone, - favorable_hours=favorable_hours, - ) - - if not favorable_time: - print( - f"[Index {i}] Skipping trade due to time filter: {time_reason}" - ) - continue - - # Entry criteria filter disabled as it's too restrictive - # We'll keep the code but bypass the filter logic - if False and strengthen_entry_criteria: - criteria_met, criteria_reason = confirm_entry_criteria( - df.iloc[: i + 1], - i, - signal, - min_confidence=min_entry_confidence, - min_confirmation_candles=min_price_action_confirmation, - require_volume_confirmation=require_volume_confirmation, - ) - - if not criteria_met: - print( - f"[Index {i}] Skipping trade due to entry criteria: {criteria_reason}" - ) - continue - - # Proceed with the trade if it passes all filters - if signal == "Bullish": - # Validate price is reasonable - if current_price < 0.00000001: # Skip if price is too small - continue - - # Calculate position size based on risk management - position_sizing = calculate_position_size( - balance, risk_percentage, current_price, trading_signal.stop_loss - ) - - # Apply tighter stops if enabled - if use_tighter_stops: - tighter_stop = calculate_tighter_stop_loss( - df, - i, - current_price, - signal, - atr_period=14, - baseline_multiplier=atr_stop_multiplier, - min_distance_percent=min_stop_distance_percent, - max_distance_percent=max_stop_distance_percent, - ) - # Only update if tighter stop is actually better (closer but still safe) - if signal == "Bullish" and tighter_stop > trading_signal.stop_loss: - trading_signal.stop_loss = tighter_stop - elif ( - signal == "Bearish" and tighter_stop < trading_signal.stop_loss - ): - trading_signal.stop_loss = tighter_stop - - # Apply volatility-based take profits if enabled - if use_volatility_based_tps: - # Default ATR multipliers if not provided - if atr_tp_multipliers is None: - atr_tp_multipliers = [ - 2.0, - 3.5, - 5.0, - ] # Default multipliers for 3 TPs - - # Calculate volatility-based take profit levels - vol_based_tps = calculate_volatility_based_take_profits( - df, - i, - current_price, - signal, - atr_period=14, - tp_multipliers=atr_tp_multipliers, - min_distance_percent=min_tp_distance_percent, - max_distance_percent=max_tp_distance_percent, - ) - - # Update take profit levels based on volatility - # For bullish signals, only update if the volatility-based TP is lower (closer) - if signal == "Bullish": - if vol_based_tps[0] < trading_signal.take_profit_1: - trading_signal.take_profit_1 = vol_based_tps[0] - if vol_based_tps[1] < trading_signal.take_profit_2: - trading_signal.take_profit_2 = vol_based_tps[1] - if vol_based_tps[2] < trading_signal.take_profit_3: - trading_signal.take_profit_3 = vol_based_tps[2] - - # Recalculate position size with potentially updated stop loss - position_sizing = calculate_position_size( - balance, risk_percentage, current_price, trading_signal.stop_loss - ) - - # Enter position with calculated size - position = position_sizing["position_size"] - amount_to_invest = position * current_price - - # Ensure we don't invest more than available balance - if amount_to_invest > balance: - position = balance / current_price - amount_to_invest = balance - - # Skip if position size is unreasonably large - if position > 1e12: # 1 trillion units max - continue - - balance -= amount_to_invest - entry_price = current_price - entry_time = current_time - entry_index = i - entry_signal = signal - parent_trade_id = str(uuid.uuid4()) # Generate parent trade ID - - # Store trade details - current_trade = { - "stop_loss": float(trading_signal.stop_loss), - "take_profit_1": float(trading_signal.take_profit_1), - "take_profit_2": float(trading_signal.take_profit_2), - "take_profit_3": float(trading_signal.take_profit_3), - "risk_reward_ratio": float(trading_signal.risk_reward_ratio), - "position_size": float(position), - "initial_position": float(position), # Store initial position size - "tp1_hit": False, - "tp2_hit": False, - "tp3_hit": False, - } - - trade_log.append( - { - "type": "entry", - "price": float(entry_price), - "index": int(i), - "signal": f"{signal} - {reason.splitlines()[0]}", - "timestamp": current_time, - "amount": float(position), - "stop_loss": float(trading_signal.stop_loss), - "take_profit_1": float(trading_signal.take_profit_1), - "take_profit_2": float(trading_signal.take_profit_2), - "take_profit_3": float(trading_signal.take_profit_3), - } - ) - - # Store entry trade in database - if db: - trade_data = { - "trade_id": parent_trade_id, - "iteration_id": str(iteration_id), - "sub_iteration_id": str(sub_iteration_id), - "symbol": str(symbol), - "interval": str(interval), - "trade_type": "entry", - "entry_timestamp": current_time, - "exit_timestamp": current_time, # Will be updated on exit - "entry_index": int(i), - "exit_index": int(i), # Will be updated on exit - "entry_price": float(entry_price), - "exit_price": float(entry_price), # Will be updated on exit - "profit_loss": 0.0, # Will be calculated on exit - "trade_duration": 0, # Will be updated on exit - "entry_signal": str(signal), - "exit_signal": "", # Will be updated on exit - "risk_reward_ratio": float(trading_signal.risk_reward_ratio), - "position_size": float(position), - "stop_loss": float(trading_signal.stop_loss), - "take_profit_1": float(trading_signal.take_profit_1), - "take_profit_2": float(trading_signal.take_profit_2), - "take_profit_3": float(trading_signal.take_profit_3), - "risk_percentage": float(risk_percentage), - "amount_traded": float(amount_to_invest), - "parent_trade_id": None, # This is the parent trade - } - db.insert_trade(trade_data) - - print( - f"[Index {i}] {symbol}: ENTRY at {entry_price:.5f} | Size: {position:.5f} | Risk/Reward: {trading_signal.risk_reward_ratio:.2f} | Reason: {reason.splitlines()[0]}" - ) - - # Handle bearish signals - mirror of bullish signal handling - elif signal == "Bearish": - # Validate price is reasonable - if current_price < 0.00000001: # Skip if price is too small - continue - - # Calculate position size based on risk management - position_sizing = calculate_position_size( - balance, risk_percentage, current_price, trading_signal.stop_loss - ) - - # Apply tighter stops if enabled - if use_tighter_stops: - tighter_stop = calculate_tighter_stop_loss( - df, - i, - current_price, - signal, - atr_period=14, - baseline_multiplier=atr_stop_multiplier, - min_distance_percent=min_stop_distance_percent, - max_distance_percent=max_stop_distance_percent, - ) - # Only update if tighter stop is actually better (closer but still safe) - if signal == "Bearish" and tighter_stop < trading_signal.stop_loss: - trading_signal.stop_loss = tighter_stop - - # Apply volatility-based take profits if enabled - if use_volatility_based_tps: - # Default ATR multipliers if not provided - if atr_tp_multipliers is None: - atr_tp_multipliers = [ - 2.0, - 3.5, - 5.0, - ] # Default multipliers for 3 TPs - - # Calculate volatility-based take profit levels - vol_based_tps = calculate_volatility_based_take_profits( - df, - i, - current_price, - signal, - atr_period=14, - tp_multipliers=atr_tp_multipliers, - min_distance_percent=min_tp_distance_percent, - max_distance_percent=max_tp_distance_percent, - ) - - # Update take profit levels based on volatility - # For bearish signals, only update if the volatility-based TP is higher (closer) - if signal == "Bearish": - if vol_based_tps[0] > trading_signal.take_profit_1: - trading_signal.take_profit_1 = vol_based_tps[0] - if vol_based_tps[1] > trading_signal.take_profit_2: - trading_signal.take_profit_2 = vol_based_tps[1] - if vol_based_tps[2] > trading_signal.take_profit_3: - trading_signal.take_profit_3 = vol_based_tps[2] - - # Recalculate position size with potentially updated stop loss - position_sizing = calculate_position_size( - balance, risk_percentage, current_price, trading_signal.stop_loss - ) - - # Enter position with calculated size - position = position_sizing["position_size"] - amount_to_invest = position * current_price - - # Ensure we don't invest more than available balance - if amount_to_invest > balance: - position = balance / current_price - amount_to_invest = balance - - # Skip if position size is unreasonably large - if position > 1e12: # 1 trillion units max - continue - - balance -= amount_to_invest - entry_price = current_price - entry_time = current_time - entry_index = i - entry_signal = signal - parent_trade_id = str(uuid.uuid4()) # Generate parent trade ID - - # Store trade details - current_trade = { - "stop_loss": float(trading_signal.stop_loss), - "take_profit_1": float(trading_signal.take_profit_1), - "take_profit_2": float(trading_signal.take_profit_2), - "take_profit_3": float(trading_signal.take_profit_3), - "risk_reward_ratio": float(trading_signal.risk_reward_ratio), - "position_size": float(position), - "initial_position": float(position), # Store initial position size - "tp1_hit": False, - "tp2_hit": False, - "tp3_hit": False, - } - - trade_log.append( - { - "type": "entry", - "price": float(entry_price), - "index": int(i), - "signal": f"{signal} - {reason.splitlines()[0]}", - "timestamp": current_time, - "amount": float(position), - "stop_loss": float(trading_signal.stop_loss), - "take_profit_1": float(trading_signal.take_profit_1), - "take_profit_2": float(trading_signal.take_profit_2), - "take_profit_3": float(trading_signal.take_profit_3), - } - ) - - # Store entry trade in database - if db: - trade_data = { - "trade_id": parent_trade_id, - "iteration_id": str(iteration_id), - "sub_iteration_id": str(sub_iteration_id), - "symbol": str(symbol), - "interval": str(interval), - "trade_type": "entry", - "entry_timestamp": current_time, - "exit_timestamp": current_time, # Will be updated on exit - "entry_index": int(i), - "exit_index": int(i), # Will be updated on exit - "entry_price": float(entry_price), - "exit_price": float(entry_price), # Will be updated on exit - "profit_loss": 0.0, # Will be calculated on exit - "trade_duration": 0, # Will be updated on exit - "entry_signal": str(signal), - "exit_signal": "", # Will be updated on exit - "risk_reward_ratio": float(trading_signal.risk_reward_ratio), - "position_size": float(position), - "stop_loss": float(trading_signal.stop_loss), - "take_profit_1": float(trading_signal.take_profit_1), - "take_profit_2": float(trading_signal.take_profit_2), - "take_profit_3": float(trading_signal.take_profit_3), - "risk_percentage": float(risk_percentage), - "amount_traded": float(amount_to_invest), - "parent_trade_id": None, # This is the parent trade - } - db.insert_trade(trade_data) + if signal == "Bullish" and position == 0 and trading_signal: + # Validate price is reasonable + if current_price < 0.00000001: # Skip if price is too small + continue + + # Calculate position size based on risk management + position_sizing = calculate_position_size( + balance, risk_percentage, current_price, trading_signal.stop_loss + ) - print( - f"[Index {i}] {symbol}: ENTRY at {entry_price:.5f} | Size: {position:.5f} | Risk/Reward: {trading_signal.risk_reward_ratio:.2f} | Reason: {reason.splitlines()[0]}" - ) + # Enter position with calculated size + position = position_sizing["position_size"] + amount_to_invest = position * current_price + + # Ensure we don't invest more than available balance + if amount_to_invest > balance: + position = balance / current_price + amount_to_invest = balance + + # Skip if position size is unreasonably large + if position > 1e12: # 1 trillion units max + continue + + balance -= amount_to_invest + entry_price = current_price + entry_time = current_time + entry_index = i + entry_signal = signal + parent_trade_id = str(uuid.uuid4()) # Generate parent trade ID + + # Store trade details + current_trade = { + "stop_loss": float(trading_signal.stop_loss), + "take_profit_1": float(trading_signal.take_profit_1), + "take_profit_2": float(trading_signal.take_profit_2), + "take_profit_3": float(trading_signal.take_profit_3), + "risk_reward_ratio": float(trading_signal.risk_reward_ratio), + "position_size": float(position), + "initial_position": float(position), # Store initial position size + "tp1_hit": False, + "tp2_hit": False, + "tp3_hit": False, + } trade_log.append( {