import logging import sys import time from datetime import datetime, timedelta, time from typing import List import pytz import pandas as pd import requests import json import math from firstock import firstock import numpy as np import pprint import os def get_client_details(): try: url = 'http://143.244.141.41/php/getUserDetails.php' response = requests.get(url) response.raise_for_status() data = response.json() if not data.get('success', False): raise ValueError("getUserDetails endpoint returned unsuccessful response") user_data = data['data'] return [user_data['field1'], user_data['field2'], user_data['field3'], user_data['field4'], user_data['field5']] except requests.exceptions.RequestException as e: print(f"HTTP Request failed: {e}") return None except (json.JSONDecodeError, KeyError) as e: print(f"Failed to parse response: {e}") return None class StockDataFetcher: def __init__(self, client_details: List[str], createEntries: bool = False): self.client_details = client_details self.user_id = client_details[0] self.ist = pytz.timezone('Asia/Kolkata') self.logger = self.setup_logger() self.createEntries = createEntries self.trendTransactionsCounter = 0 self.total_pl = 0.0 # Trade Statistics self.trade_statistics = { 'total_trades': 0, 'successful_trades': 0, 'lost_trades': 0, 'ce_trades': 0, 'pe_trades': 0, 'total_profit': 0.0, 'total_loss': 0.0, 'largest_win': 0.0, 'largest_loss': 0.0, 'profit_target_hits': 0, 'signal_exits': 0, 'time_based_exits': 0, 'final_exits': 0, 'stop_loss_exits': 0, 'exit_reasons': { 'PROFIT_TARGET': 0, 'SIGNAL_EXIT': 0, 'TIME_EXIT': 0, 'FINAL_EXIT': 0, 'STOP_LOSS': 0 } } self.lotCount = 1 # number of lots at each strike price self.strikeDepth = 2 # how much depth in ITM # RMA Parameters self.rma_length = 200 # RMA period # EMA Parameters self.ema_length = 2 # EMA period # Fixed Profit Targets and Stop Loss self.fixed_profit_target = 99.2 # Fixed profit target in points self.fixed_stop_loss = 100 # Fixed stop loss in points # Time-based entry and exit parameters self.trading_start_hour = 9 self.trading_start_minute = 18 self.time_exit_hour = 15 self.time_exit_minute = 26 # Tick Interval self.interval = 1 # 1-minute interval self.debug_data = '' # Daily P&L tracking self.daily_pl_threshold = -150 # Stop trading when daily P&L crosses -40 self.trading_stopped = False # Flag to track if trading has been stopped for the day # New flag to track if profit target has been hit for the day self.profit_target_hit = False # Flag to track if profit target has been hit once class ISTFormatter(logging.Formatter): def formatTime(self, record, datefmt=None): ist = pytz.timezone('Asia/Kolkata') record_time = datetime.fromtimestamp(record.created, tz=ist) return record_time.strftime(datefmt or '%Y-%m-%d %H:%M:%S') def addLogDataDebug(self, text): self.logger.debug(f'{text}') self.debug_data += text + '\n' def addLogDataInfo(self, text): self.logger.info(f'{text}') self.debug_data += text + '\n' def setup_logger(self) -> logging.Logger: logger = logging.getLogger(__name__) logger.setLevel(logging.DEBUG) if not logger.handlers: formatter = self.ISTFormatter('%(asctime)s - %(levelname)s - %(message)s') console_handler = logging.StreamHandler() console_handler.setFormatter(formatter) logger.addHandler(console_handler) logging.getLogger().handlers.clear() return logger def is_time_for_trading_start(self, current_time_str: str) -> bool: """Check if current time is after or equal to trading start time (09:45)""" try: # Parse current time string to datetime object current_time = datetime.strptime(current_time_str, '%Y-%m-%d %H:%M:%S') # Check if time is 09:45 or later if current_time.hour > self.trading_start_hour or ( current_time.hour == self.trading_start_hour and current_time.minute >= self.trading_start_minute ): return True return False except ValueError: # If time format is different, try alternative parsing try: current_time = datetime.strptime(current_time_str, '%H:%M:%S %d-%m-%Y') if current_time.hour > self.trading_start_hour or ( current_time.hour == self.trading_start_hour and current_time.minute >= self.trading_start_minute ): return True return False except ValueError: return False def is_time_for_exit(self, current_time_str: str) -> bool: """Check if current time is after or equal to exit time (14:54)""" try: # Parse current time string to datetime object current_time = datetime.strptime(current_time_str, '%Y-%m-%d %H:%M:%S') # Check if time is 14:54 or later if current_time.hour > self.time_exit_hour or ( current_time.hour == self.time_exit_hour and current_time.minute >= self.time_exit_minute ): return True return False except ValueError: # If time format is different, try alternative parsing try: current_time = datetime.strptime(current_time_str, '%H:%M:%S %d-%m-%Y') if current_time.hour > self.time_exit_hour or ( current_time.hour == self.time_exit_hour and current_time.minute >= self.time_exit_minute ): return True return False except ValueError: return False def calculate_ema(self, series: pd.Series, length: int) -> pd.Series: """Calculate EMA (Exponential Moving Average)""" try: ema = series.ewm(span=length, adjust=False).mean() self.addLogDataDebug(f"EMA {length} calculation completed - First 5 values: {ema.head().tolist()}") return ema except Exception as e: self.addLogDataDebug(f"Error calculating EMA {length}: {e}") return pd.Series([0] * len(series), index=series.index) def calculate_vwap(self, df: pd.DataFrame) -> pd.Series: """Calculate Volume Weighted Average Price (VWAP)""" try: # Check if required columns exist if 'volume' not in df.columns or 'high' not in df.columns or 'low' not in df.columns or 'close' not in df.columns: self.addLogDataDebug("Missing required columns for VWAP calculation") return pd.Series([0] * len(df), index=df.index) # Calculate typical price typical_price = (df['high'] + df['low'] + df['close']) / 3 # Calculate cumulative typical price * volume cumulative_tpv = (typical_price * df['volume']).cumsum() # Calculate cumulative volume cumulative_volume = df['volume'].cumsum() # Calculate VWAP vwap = cumulative_tpv / cumulative_volume return vwap except Exception as e: self.addLogDataDebug(f"Error calculating VWAP: {e}") return pd.Series([0] * len(df), index=df.index) def calculate_twap(self, df: pd.DataFrame, anchor_period: str = 'D') -> pd.Series: """Calculate Time Weighted Average Price (TWAP) matching Pine Script logic""" try: # Check for required columns required_columns = ['open', 'high', 'low', 'close'] missing_columns = [col for col in required_columns if col not in df.columns] if missing_columns: self.addLogDataDebug(f"Missing required columns for TWAP calculation: {missing_columns}") return pd.Series([0] * len(df), index=df.index) # Create proper datetime index from time and epochTime columns working_df = df.copy() if 'epochTime' in working_df.columns: # Convert epochTime to datetime (assuming it's in seconds) working_df['datetime'] = pd.to_datetime(working_df['epochTime'], unit='s') working_df.set_index('datetime', inplace=True) self.addLogDataDebug("Using epochTime column as datetime index for TWAP calculation") elif 'time' in working_df.columns: # If you have date information, combine with time # This assumes you have a date column or can infer date try: # If time is in format 'HH:MM:SS' working_df['datetime'] = pd.to_datetime(working_df['time'], format='%H:%M:%S') working_df.set_index('datetime', inplace=True) self.addLogDataDebug("Using time column as datetime index for TWAP calculation") except: self.addLogDataDebug("Cannot parse time column, using simple cumulative TWAP") return self._calculate_simple_twap(working_df) else: self.addLogDataDebug("No datetime columns available, using simple cumulative TWAP") return self._calculate_simple_twap(working_df) # Use OHLC4 as source (same as Pine Script default) typical_price = (working_df['open'] + working_df['high'] + working_df['low'] + working_df['close']) / 4 # Create period change flags period_changes = working_df.index.to_series().dt.floor(anchor_period) != working_df.index.to_series().shift(1).dt.floor(anchor_period) period_changes.iloc[0] = True # First row always starts a new period # Calculate TWAP with period resets twap_values = [] current_sum = 0.0 current_count = 0 for i, (price, period_change) in enumerate(zip(typical_price, period_changes)): if period_change: # Reset for new period current_sum = 0.0 current_count = 0 current_sum += price current_count += 1 twap_values.append(current_sum / current_count) # Return with original index to maintain alignment with input dataframe return pd.Series(twap_values, index=df.index) except Exception as e: self.addLogDataDebug(f"Error calculating TWAP: {e}") return pd.Series([0] * len(df), index=df.index) def _calculate_simple_twap(self, df: pd.DataFrame) -> pd.Series: """Fallback simple TWAP calculation without datetime requirements""" try: typical_price = (df['open'] + df['high'] + df['low'] + df['close']) / 4 return typical_price.expanding().mean() except Exception as e: self.addLogDataDebug(f"Error in simple TWAP calculation: {e}") return pd.Series([0] * len(df), index=df.index) def update_trade_statistics(self, position_type: str, pl: float, exit_reason: str): """Update comprehensive trade statistics""" self.trade_statistics['total_trades'] += 1 if position_type == 'CE': self.trade_statistics['ce_trades'] += 1 else: self.trade_statistics['pe_trades'] += 1 if pl > 0: self.trade_statistics['successful_trades'] += 1 self.trade_statistics['total_profit'] += pl if pl > self.trade_statistics['largest_win']: self.trade_statistics['largest_win'] = pl else: self.trade_statistics['lost_trades'] += 1 self.trade_statistics['total_loss'] += abs(pl) if abs(pl) > self.trade_statistics['largest_loss']: self.trade_statistics['largest_loss'] = abs(pl) # Track exit reasons if 'PROFIT TARGET' in exit_reason: self.trade_statistics['profit_target_hits'] += 1 self.trade_statistics['exit_reasons']['PROFIT_TARGET'] += 1 elif 'TIME EXIT' in exit_reason: self.trade_statistics['time_based_exits'] += 1 self.trade_statistics['exit_reasons']['TIME_EXIT'] += 1 elif 'FINAL CLOSE' in exit_reason: self.trade_statistics['final_exits'] += 1 self.trade_statistics['exit_reasons']['FINAL_EXIT'] += 1 elif 'STOP LOSS' in exit_reason: self.trade_statistics['stop_loss_exits'] += 1 self.trade_statistics['exit_reasons']['STOP_LOSS'] += 1 else: self.trade_statistics['signal_exits'] += 1 self.trade_statistics['exit_reasons']['SIGNAL_EXIT'] += 1 def print_trade_statistics(self): """Print comprehensive trade statistics""" total_trades = self.trade_statistics['total_trades'] if total_trades == 0: self.addLogDataInfo("No trades executed") return win_rate = (self.trade_statistics['successful_trades'] / total_trades) * 100 avg_win = (self.trade_statistics['total_profit'] / self.trade_statistics['successful_trades'] if self.trade_statistics['successful_trades'] > 0 else 0) avg_loss = (self.trade_statistics['total_loss'] / self.trade_statistics['lost_trades'] if self.trade_statistics['lost_trades'] > 0 else 0) profit_factor = (self.trade_statistics['total_profit'] / self.trade_statistics['total_loss'] if self.trade_statistics['total_loss'] > 0 else float('inf')) self.addLogDataInfo("=" * 60) self.addLogDataInfo("TRADE STATISTICS SUMMARY") self.addLogDataInfo("=" * 60) self.addLogDataInfo(f"Total Trades: {total_trades}") self.addLogDataInfo(f"Successful Trades: {self.trade_statistics['successful_trades']}") self.addLogDataInfo(f"Lost Trades: {self.trade_statistics['lost_trades']}") self.addLogDataInfo(f"Win Rate: {win_rate:.2f}%") self.addLogDataInfo(f"CE Trades: {self.trade_statistics['ce_trades']}") self.addLogDataInfo(f"PE Trades: {self.trade_statistics['pe_trades']}") self.addLogDataInfo("") self.addLogDataInfo("P&L ANALYSIS") self.addLogDataInfo("-" * 30) self.addLogDataInfo(f"Total Profit: ₹{self.trade_statistics['total_profit']:.2f}") self.addLogDataInfo(f"Total Loss: ₹{self.trade_statistics['total_loss']:.2f}") self.addLogDataInfo(f"Net P&L: ₹{self.total_pl:.2f}") self.addLogDataInfo(f"Profit Factor: {profit_factor:.2f}") self.addLogDataInfo(f"Average Win: ₹{avg_win:.2f}") self.addLogDataInfo(f"Average Loss: ₹{avg_loss:.2f}") self.addLogDataInfo(f"Largest Win: ₹{self.trade_statistics['largest_win']:.2f}") self.addLogDataInfo(f"Largest Loss: ₹{self.trade_statistics['largest_loss']:.2f}") self.addLogDataInfo("") self.addLogDataInfo("EXIT REASONS BREAKDOWN") self.addLogDataInfo("-" * 30) self.addLogDataInfo(f"Profit Target Hits: {self.trade_statistics['exit_reasons']['PROFIT_TARGET']}") self.addLogDataInfo(f"Signal Exits: {self.trade_statistics['exit_reasons']['SIGNAL_EXIT']}") self.addLogDataInfo(f"Time-based Exits (14:54): {self.trade_statistics['exit_reasons']['TIME_EXIT']}") self.addLogDataInfo(f"Stop Loss Exits: {self.trade_statistics['exit_reasons']['STOP_LOSS']}") self.addLogDataInfo(f"Final Session Exits: {self.trade_statistics['exit_reasons']['FINAL_EXIT']}") self.addLogDataInfo("=" * 60) def login(self): try: self.logger.info(f"Attempting login for {self.client_details[0]}") response = firstock.login(*self.client_details) if response.get("status") == "success": self.logger.info("Login successful") else: self.logger.error(f"Login failed: {response}") sys.exit() except Exception as e: self.logger.error(f"Login error: {e}") sys.exit() def process_symbol_data(self, symbol: str, interval: int, start_time: datetime, end_time: datetime): exchange, trading_symbol = symbol.split(":") df = self.fetch_time_price_series( exchange, trading_symbol, start_time.strftime("%H:%M:%S %d-%m-%Y"), end_time.strftime("%H:%M:%S %d-%m-%Y"), str(interval), ) df = df.sort_values(by='epochTime', ascending=True) # Calculate VWAP and TWAP before dropping volume and oi df['vwap'] = self.calculate_vwap(df) df['twap'] = self.calculate_twap(df) # Drop columns only if they exist columns_to_drop = ['volume', 'oi'] df = df.drop([col for col in columns_to_drop if col in df.columns], axis=1) return df def calculate_rma(self, series: pd.Series, length: int) -> pd.Series: """Calculate RMA (Running Moving Average)""" return series.ewm(alpha=1/length, adjust=False).mean() def createTrendEntry(self, tickTimeStr, instrument, closePrice, signal, lotCount): self.createTrendEntryLastSupper(tickTimeStr, instrument, closePrice, signal, lotCount) instrument = 'NIFTY' + str(instrument) self.trendTransactionsCounter += 1 if self.createEntries: # File to store last signal last_action_file = "last_action.txt" # Check previous signal previous_signal = None if os.path.exists(last_action_file): with open(last_action_file, "r") as f: previous_signal = f.read().strip() # If signal is same as previous, skip if previous_signal == str(signal): self.addLogDataDebug(f"Skipping request as signal '{signal}' is same as previous.") return url = "http://143.244.141.41/php/createEntriesMft.php" tickTimeStr += ':' + str(self.trendTransactionsCounter).zfill(3) params = { 'tickTime': str(tickTimeStr), 'instrument': str(instrument), 'closePrice': str(closePrice), 'signal': str(signal), 'orderType': str(lotCount) } try: response = requests.get(url, params=params, timeout=5) if response.status_code == 200: # Save the new signal to file with open(last_action_file, "w") as f: f.write(str(signal)) self.addLogDataDebug(f"Request successful. Signal '{signal}' stored.") else: self.addLogDataDebug(f"Request failed with status code: {response.status_code}") if response.text != '': self.addLogDataDebug(response.text + "\n") except requests.RequestException as e: self.addLogDataDebug(f"Request failed: {e}") def createTrendEntryLastSupper(self, tickTimeStr, instrument, closePrice, signal, lotCount): instrument = 'NIFTY' + str(instrument) self.trendTransactionsCounter += 1 if self.createEntries: # File to store last signal last_action_file = "last_action_ls.txt" # Check previous signal previous_signal = None if os.path.exists(last_action_file): with open(last_action_file, "r") as f: previous_signal = f.read().strip() # If signal is same as previous, skip if previous_signal == str(signal): self.addLogDataDebug(f"Skipping request as signal '{signal}' is same as previous.") return url = "http://139.59.6.25/php/createEntriesMft.php" tickTimeStr += ':' + str(self.trendTransactionsCounter).zfill(3) params = { 'tickTime': str(tickTimeStr), 'instrument': str(instrument), 'closePrice': str(closePrice), 'signal': str(signal), 'orderType': str(lotCount) } try: response = requests.get(url, params=params, timeout=5) if response.status_code == 200: # Save the new signal to file with open(last_action_file, "w") as f: f.write(str(signal)) self.addLogDataDebug(f"Request successful. Signal '{signal}' stored.") else: self.addLogDataDebug(f"Request failed with status code: {response.status_code}") if response.text != '': self.addLogDataDebug(response.text + "\n") except requests.RequestException as e: self.addLogDataDebug(f"Request failed: {e}") def fetch_time_price_series(self, exchange: str, trading_symbol: str, start_time: str, end_time: str, interval: str) -> pd.DataFrame: print(f"userId: {self.user_id}, exchange: {exchange}, tradingSymbol: {trading_symbol}, startTime: {start_time}, endTime: {end_time}, interval: {interval}") try: response = firstock.timePriceSeries( userId=self.user_id, exchange=exchange, tradingSymbol=trading_symbol, startTime=start_time, endTime=end_time, interval=str(interval) + "mi" ) if response.get("status") == "success": return pd.DataFrame(response.get("data", [])) self.addLogDataDebug(f"Fetch failed: {response}") return pd.DataFrame() except Exception as e: self.addLogDataDebug(f"Error fetching series: {e}") return pd.DataFrame() def getNiftyStrikePriceGivenCurrentValue(self, closePrice, type, depth=1): try: price = float(closePrice) remainder = price % 50 # Round to nearest 50 if remainder > 25: base_strike = int(price + (50 - remainder)) else: base_strike = int(price - remainder) # For ITM options: # CE (Call) ITM: strike < current price (lower strikes) # PE (Put) ITM: strike > current price (higher strikes) if type == 'CE': # For CE ITM, we want strikes BELOW current price if base_strike >= price: base_strike -= 50 # Calculate ITM strike based on depth itm_strike = base_strike - (50 * (depth - 1)) elif type == 'PE': # For PE ITM, we want strikes ABOVE current price if base_strike <= price: base_strike += 50 # Calculate ITM strike based on depth itm_strike = base_strike + (50 * (depth - 1)) return [itm_strike] except (ValueError, TypeError): raise ValueError("Invalid closePrice input. Must be a number or numeric string.") def check_daily_pl_threshold(self, current_time: str) -> bool: """Check if daily P&L has crossed the threshold and stop trading if needed""" if self.total_pl <= self.daily_pl_threshold and not self.trading_stopped: self.trading_stopped = True self.addLogDataInfo(f"[DAILY P&L THRESHOLD BREACH] at {current_time}") self.addLogDataInfo(f"Daily P&L: ₹{self.total_pl:.2f} has crossed threshold: ₹{self.daily_pl_threshold:.2f}") self.addLogDataInfo("Trading stopped for the day. No new positions will be created.") return True return False def check_profit_target_hit(self, current_time: str) -> bool: """Check if profit target has been hit and stop trading if needed""" if self.profit_target_hit and not self.trading_stopped: self.trading_stopped = True self.addLogDataInfo(f"[PROFIT TARGET HIT] at {current_time}") self.addLogDataInfo(f"Profit target of {self.fixed_profit_target} points has been achieved once today.") self.addLogDataInfo("Trading stopped for the day. No new positions will be created.") return True return False def handle_trend_following_orders(self, df: pd.DataFrame): self.open_positions_price = {'CE': [], 'PE': []} position = None entry_price = None current_trade_entry_time = None time_exit_executed = False trading_started = False self.trading_stopped = False # Reset trading stopped flag for new day self.profit_target_hit = False # Reset profit target flag for new day for i in range(1, len(df)): _time = df.loc[i, 'time'] price = df.loc[i, 'close'] current_rma = df.loc[i, 'rma'] current_ema = df.loc[i, 'ema'] current_vwap = df.loc[i, 'vwap'] current_twap = df.loc[i, 'twap'] prev_rma = df.loc[i-1, 'rma'] prev_ema = df.loc[i-1, 'ema'] # self.addLogDataInfo(f"T:{_time} | Price:{price:.4f} | RMA:{current_rma:.4f} | EMA:{current_ema:.4f} | VWAP:{current_vwap:.4f} | TWAP:{current_twap:.4f}") if time_exit_executed: continue # Check if trading should start if not trading_started: if self.is_time_for_trading_start(str(_time)): trading_started = True self.addLogDataInfo(f"[TRADING STARTED] at {_time} | Trading begins after{self.trading_start_hour}:{self.trading_start_minute}") else: # Skip processing until trading start time continue # Check for daily P&L threshold breach if self.check_daily_pl_threshold(str(_time)): # Close any open positions due to daily P&L threshold breach if position is not None: if position == 'CE': for instr, buy_price in self.open_positions_price['CE']: pl = round((price - buy_price) * self.lotCount, 2) self.createTrendEntry(str(_time), instr, price, 'SELL', self.lotCount) self.total_pl += pl self.update_trade_statistics('CE', pl, 'DAILY P&L THRESHOLD') self.addLogDataInfo(f"[DAILY P&L THRESHOLD CE EXIT] at {_time} | Price: {price} | P/L: ₹{pl}") self.open_positions_price['CE'].clear() elif position == 'PE': for instr, buy_price in self.open_positions_price['PE']: pl = round((buy_price - price) * self.lotCount, 2) self.createTrendEntry(str(_time), instr, price, 'SELL', self.lotCount) self.total_pl += pl self.update_trade_statistics('PE', pl, 'DAILY P&L THRESHOLD') self.addLogDataInfo(f"[DAILY P&L THRESHOLD PE EXIT] at {_time} | Price: {price} | P/L: ₹{pl}") self.open_positions_price['PE'].clear() position = None entry_price = None current_trade_entry_time = None # Skip further processing for the day continue # Check for profit target hit if self.check_profit_target_hit(str(_time)): # Close any open positions due to profit target hit if position is not None: if position == 'CE': for instr, buy_price in self.open_positions_price['CE']: pl = round((price - buy_price) * self.lotCount, 2) self.createTrendEntry(str(_time), instr, price, 'SELL', self.lotCount) self.total_pl += pl self.update_trade_statistics('CE', pl, 'PROFIT TARGET HIT') self.addLogDataInfo(f"[PROFIT TARGET HIT CE EXIT] at {_time} | Price: {price} | P/L: ₹{pl}") self.open_positions_price['CE'].clear() elif position == 'PE': for instr, buy_price in self.open_positions_price['PE']: pl = round((buy_price - price) * self.lotCount, 2) self.createTrendEntry(str(_time), instr, price, 'SELL', self.lotCount) self.total_pl += pl self.update_trade_statistics('PE', pl, 'PROFIT TARGET HIT') self.addLogDataInfo(f"[PROFIT TARGET HIT PE EXIT] at {_time} | Price: {price} | P/L: ₹{pl}") self.open_positions_price['PE'].clear() position = None entry_price = None current_trade_entry_time = None # Skip further processing for the day continue # Check for time-based exit if not time_exit_executed and self.is_time_for_exit(str(_time)): if position is not None: # Close any open position due to time exit if position == 'CE': for instr, buy_price in self.open_positions_price['CE']: pl = round((price - buy_price) * self.lotCount, 2) self.createTrendEntry(str(_time), instr, price, 'SELL', self.lotCount) self.total_pl += pl self.update_trade_statistics('CE', pl, 'TIME EXIT') self.addLogDataInfo(f"[TIME EXIT CE] at {_time} | Price: {price} | P/L: ₹{pl}") self.open_positions_price['CE'].clear() elif position == 'PE': for instr, buy_price in self.open_positions_price['PE']: pl = round((buy_price - price) * self.lotCount, 2) self.createTrendEntry(str(_time), instr, price, 'SELL', self.lotCount) self.total_pl += pl self.update_trade_statistics('PE', pl, 'TIME EXIT') self.addLogDataInfo(f"[TIME EXIT PE] at {_time} | Price: {price} | P/L: ₹{pl}") self.open_positions_price['PE'].clear() position = None entry_price = None current_trade_entry_time = None self.addLogDataInfo(f"[TIME EXIT EXECUTED] at {_time} | No new positions will be created after this time") time_exit_executed = True continue # Check for fixed profit exit first if position == 'CE' and entry_price is not None: current_profit = price - entry_price if current_profit >= self.fixed_profit_target: # Mark profit target as hit self.profit_target_hit = True for instr, buy_price in self.open_positions_price['CE']: pl = round((price - buy_price) * self.lotCount, 2) self.createTrendEntry(str(_time), instr, price, 'SELL', self.lotCount) self.total_pl += pl self.update_trade_statistics('CE', pl, 'PROFIT TARGET') self.addLogDataInfo(f"[PROFIT TARGET CE] at {_time} | Price: {price} | Profit: ₹{pl} | Target: {self.fixed_profit_target}") self.open_positions_price['CE'].clear() position = None entry_price = None current_trade_entry_time = None # After hitting profit target, stop trading for the day self.addLogDataInfo(f"[PROFIT TARGET ACHIEVED] at {_time}") self.addLogDataInfo(f"Profit target of {self.fixed_profit_target} points has been hit. Trading stopped for the day.") continue # Check for stop loss exit for CE if position == 'CE' and entry_price is not None: current_loss = entry_price - price if current_loss >= self.fixed_stop_loss: for instr, buy_price in self.open_positions_price['CE']: pl = round((price - buy_price) * self.lotCount, 2) self.createTrendEntry(str(_time), instr, price, 'SELL', self.lotCount) self.total_pl += pl self.update_trade_statistics('CE', pl, 'STOP LOSS') self.addLogDataInfo(f"[STOP LOSS CE] at {_time} | Price: {price} | Loss: ₹{abs(pl)} | Stop Loss: {self.fixed_stop_loss}") self.open_positions_price['CE'].clear() position = None entry_price = None current_trade_entry_time = None continue if position == 'PE' and entry_price is not None: current_profit = entry_price - price if current_profit >= self.fixed_profit_target: # Mark profit target as hit self.profit_target_hit = True for instr, buy_price in self.open_positions_price['PE']: pl = round((buy_price - price) * self.lotCount, 2) self.createTrendEntry(str(_time), instr, price, 'SELL', self.lotCount) self.total_pl += pl self.update_trade_statistics('PE', pl, 'PROFIT TARGET') self.addLogDataInfo(f"[PROFIT TARGET PE] at {_time} | Price: {price} | Profit: ₹{pl} | Target: {self.fixed_profit_target}") self.open_positions_price['PE'].clear() position = None entry_price = None current_trade_entry_time = None # After hitting profit target, stop trading for the day self.addLogDataInfo(f"[PROFIT TARGET ACHIEVED] at {_time}") self.addLogDataInfo(f"Profit target of {self.fixed_profit_target} points has been hit. Trading stopped for the day.") continue # Check for stop loss exit for PE if position == 'PE' and entry_price is not None: current_loss = price - entry_price if current_loss >= self.fixed_stop_loss: for instr, buy_price in self.open_positions_price['PE']: pl = round((buy_price - price) * self.lotCount, 2) self.createTrendEntry(str(_time), instr, price, 'SELL', self.lotCount) self.total_pl += pl self.update_trade_statistics('PE', pl, 'STOP LOSS') self.addLogDataInfo(f"[STOP LOSS PE] at {_time} | Price: {price} | Loss: ₹{abs(pl)} | Stop Loss: {self.fixed_stop_loss}") self.open_positions_price['PE'].clear() position = None entry_price = None current_trade_entry_time = None continue # Check for EMA crossover conditions ema_cross_above_rma = (current_ema > current_rma) and (prev_ema <= prev_rma) ema_cross_below_rma = (current_ema < current_rma) and (prev_ema >= prev_rma) # BUY CE: EMA9 crosses above RMA if ema_cross_above_rma and position != 'CE' and not self.trading_stopped: self.addLogDataInfo(f"[EMA CROSS DETECTED] EMA {current_ema:.4f} crossed above RMA {current_rma:.4f}") # Close any existing PE position first if position == 'PE': for instr, buy_price in self.open_positions_price['PE']: pl = round((buy_price - price) * self.lotCount, 2) self.createTrendEntry(str(_time), instr, price, 'SELL', self.lotCount) self.total_pl += pl self.update_trade_statistics('PE', pl, 'SIGNAL EXIT') self.addLogDataInfo(f"[CLOSE PE] Before CE Buy at {_time} | Price: {price} | P/L: ₹{pl}") self.open_positions_price['PE'].clear() position = None entry_price = None # Enter CE position strikePrices = self.getNiftyStrikePriceGivenCurrentValue(price, 'CE', self.strikeDepth) for strike in strikePrices: instrument = f"{strike}CE" self.createTrendEntry(str(_time), instrument, price, 'BUY', self.lotCount) self.open_positions_price['CE'].append((instrument, price)) self.addLogDataInfo(f"[BUY CE] at {_time} | EMA {current_ema:.4f} crossed above RMA {current_rma:.4f} | Price: {price}") position = 'CE' entry_price = price current_trade_entry_time = _time # BUY PE: EMA9 crosses below RMA elif ema_cross_below_rma and position != 'PE' and not self.trading_stopped: self.addLogDataInfo(f"[EMA CROSS DETECTED] EMA {current_ema:.4f} crossed below RMA {current_rma:.4f}") # Close any existing CE position first if position == 'CE': for instr, buy_price in self.open_positions_price['CE']: pl = round((price - buy_price) * self.lotCount, 2) self.createTrendEntry(str(_time), instr, price, 'SELL', self.lotCount) self.total_pl += pl self.update_trade_statistics('CE', pl, 'SIGNAL EXIT') self.addLogDataInfo(f"[CLOSE CE] Before PE Buy at {_time} | Price: {price} | P/L: ₹{pl}") self.open_positions_price['CE'].clear() position = None entry_price = None # Enter PE position strikePrices = self.getNiftyStrikePriceGivenCurrentValue(price, 'PE', self.strikeDepth) for strike in strikePrices: instrument = f"{strike}PE" self.createTrendEntry(str(_time), instrument, price, 'BUY', self.lotCount) self.open_positions_price['PE'].append((instrument, price)) self.addLogDataInfo(f"[BUY PE] at {_time} | EMA {current_ema:.4f} crossed below RMA {current_rma:.4f} | Price: {price}") position = 'PE' entry_price = price current_trade_entry_time = _time def fetch_all_data(self, elapsed: int): now_ist = datetime.now(self.ist) hour = now_ist.hour # 24-hour format minute = now_ist.minute # Minutes if hour < 8 or (hour == 8 and minute <= 16): print('Waiting till 9:17 am... exiting for now') return start_time = now_ist.replace(hour=9, minute=15, second=0) - timedelta(days=elapsed+3) end_time = now_ist.replace(hour=15, minute=31, second=0) - timedelta(days=elapsed) df = self.process_symbol_data('NSE:NIFTY', self.interval, start_time, end_time) if df.empty: self.addLogDataInfo("No data fetched") return # Calculate RMA df['rma'] = self.calculate_rma(df['close'], self.rma_length) # Calculate EMA df['ema'] = self.calculate_ema(df['close'], self.ema_length) # Now filter to keep only the last date's data df['datetime'] = pd.to_datetime(df['time'], format='%H:%M:%S %d-%m-%Y') last_date = df['datetime'].dt.date.max() df = df[df['datetime'].dt.date == last_date].copy() df = df.drop('datetime', axis=1) df = df[:-1] df = df.reset_index(drop=True) # Handle trend following orders self.addLogDataInfo("") self.addLogDataInfo("EMA-RMA Crossover Strategy") self.addLogDataInfo(f"RMA Length: {self.rma_length}") self.addLogDataInfo(f"EMA Length: {self.ema_length}") self.addLogDataInfo(f"Fixed Profit Target: {self.fixed_profit_target} points") self.addLogDataInfo(f"Fixed Stop Loss: {self.fixed_stop_loss} points") self.addLogDataInfo(f"Daily P&L Stop Threshold: ₹{self.daily_pl_threshold}") self.addLogDataInfo(f"Profit Target Hit Once: Stops trading for the day") self.addLogDataInfo(f"Trading Start Time: {self.trading_start_hour:02d}:{self.trading_start_minute:02d}") self.addLogDataInfo(f"Time-based Exit: {self.time_exit_hour:02d}:{self.time_exit_minute:02d}") self.addLogDataInfo(f"No new positions after {self.time_exit_hour:02d}:{self.time_exit_minute:02d}") self.handle_trend_following_orders(df) self.addLogDataInfo("") self.addLogDataInfo(f"Final close price: {df.iloc[-1]['close']}") self.addLogDataInfo(f"RMA ({self.rma_length}): {df.iloc[-1]['rma']:.2f}") self.addLogDataInfo(f"EMA ({self.ema_length}): {df.iloc[-1]['ema']:.2f}") self.addLogDataInfo(f"Final VWAP: {df.iloc[-1]['vwap']:.2f}") self.addLogDataInfo(f"Final TWAP: {df.iloc[-1]['twap']:.2f}") self.addLogDataInfo(f"Total Realized P&L: ₹{round(self.total_pl, 2)}") # Print comprehensive trade statistics self.print_trade_statistics() # Add daily P&L threshold status if self.trading_stopped: if self.profit_target_hit: self.addLogDataInfo(f"[PROFIT TARGET HIT] Trading was stopped after hitting profit target of {self.fixed_profit_target} points") else: self.addLogDataInfo(f"[DAILY P&L THRESHOLD ACTIVATED] Trading was stopped due to P&L breaching ₹{self.daily_pl_threshold}") self.addLogDataInfo("\n") if __name__ == "__main__": client_details = get_client_details() elapsed = float(sys.argv[1]) if len(sys.argv) > 1 else 0 createEntries = len(sys.argv) > 2 stock_fetcher = StockDataFetcher(client_details, createEntries) stock_fetcher.login() stock_fetcher.fetch_all_data(elapsed)