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, 'rma_direction_exits': 0, 'exit_reasons': { 'PROFIT_TARGET': 0, 'SIGNAL_EXIT': 0, 'TIME_EXIT': 0, 'FINAL_EXIT': 0, 'STOP_LOSS': 0, 'RMA_DIRECTION_CHANGE': 0 }, 'rma_direction_trades': { 'CE_WITH_RMA_UP': 0, 'CE_WITH_RMA_DOWN': 0, 'PE_WITH_RMA_UP': 0, 'PE_WITH_RMA_DOWN': 0 } } self.lotCount = 1 # number of lots at each strike price # RMA Parameters self.rma_length = 20 # RMA period self.rma_direction_smoothing = 9 # Smoothing period for RMA direction # Fixed Profit Targets and Stop Loss self.fixed_profit_target = 16 # Fixed profit target in points self.fixed_stop_loss = 24 # Fixed stop loss in points # Time-based entry and exit parameters self.trading_start_hour = 2 self.trading_start_minute = 18 self.time_exit_hour = 14 self.time_exit_minute = 54 # Tick Interval self.interval = 1 # 1-minute interval self.debug_data = '' 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_rma_direction(self, rma_series: pd.Series, smoothing: int) -> pd.Series: """Calculate RMA direction for trend identification""" return rma_series >= rma_series.shift(smoothing) 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.Series) -> pd.Series: """Calculate Time Weighted Average Price (TWAP)""" try: # For TWAP, we use the average of high, low, and close prices if 'high' not in df.columns or 'low' not in df.columns or 'close' not in df.columns: self.addLogDataDebug("Missing required columns for TWAP calculation") return pd.Series([0] * len(df), index=df.index) # Calculate typical price (same as for VWAP) typical_price = (df['high'] + df['low'] + df['close']) / 3 # TWAP is the cumulative average of typical price twap = typical_price.expanding().mean() return twap except Exception as e: self.addLogDataDebug(f"Error calculating TWAP: {e}") return pd.Series([0] * len(df), index=df.index) def update_trade_statistics(self, position_type: str, pl: float, exit_reason: str, rma_direction: bool = None): """Update comprehensive trade statistics""" self.trade_statistics['total_trades'] += 1 if position_type == 'CE': self.trade_statistics['ce_trades'] += 1 # Track RMA direction for CE trades if rma_direction is not None: if rma_direction: self.trade_statistics['rma_direction_trades']['CE_WITH_RMA_UP'] += 1 else: self.trade_statistics['rma_direction_trades']['CE_WITH_RMA_DOWN'] += 1 else: self.trade_statistics['pe_trades'] += 1 # Track RMA direction for PE trades if rma_direction is not None: if rma_direction: self.trade_statistics['rma_direction_trades']['PE_WITH_RMA_UP'] += 1 else: self.trade_statistics['rma_direction_trades']['PE_WITH_RMA_DOWN'] += 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 elif 'RMA DIRECTION CHANGE' in exit_reason: self.trade_statistics['rma_direction_exits'] += 1 self.trade_statistics['exit_reasons']['RMA_DIRECTION_CHANGE'] += 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')) # Calculate RMA direction success rates ce_with_rma_up = self.trade_statistics['rma_direction_trades']['CE_WITH_RMA_UP'] ce_with_rma_down = self.trade_statistics['rma_direction_trades']['CE_WITH_RMA_DOWN'] pe_with_rma_up = self.trade_statistics['rma_direction_trades']['PE_WITH_RMA_UP'] pe_with_rma_down = self.trade_statistics['rma_direction_trades']['PE_WITH_RMA_DOWN'] 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("RMA DIRECTION ANALYSIS") self.addLogDataInfo("-" * 30) self.addLogDataInfo(f"CE Trades with RMA UP: {ce_with_rma_up}") self.addLogDataInfo(f"CE Trades with RMA DOWN: {ce_with_rma_down}") self.addLogDataInfo(f"PE Trades with RMA UP: {pe_with_rma_up}") self.addLogDataInfo(f"PE Trades with RMA DOWN: {pe_with_rma_down}") 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"RMA Direction Change Exits: {self.trade_statistics['exit_reasons']['RMA_DIRECTION_CHANGE']}") 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): 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 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): strike_prices = [] try: price = float(closePrice) remainder = price % 50 base_strike = int(price + (50 - remainder)) if remainder > 25 else int(price - remainder) except (ValueError, TypeError): raise ValueError("Invalid closePrice input. Must be a number or numeric string.") if base_strike < price and type == 'PE': base_strike += 50 elif base_strike > price and type == 'CE': base_strike -= 50 # Return single strike price return [base_strike] 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 # Track last position type to avoid reentry last_position_type = None wait_for_crossover = False for i in range(1, len(df)): _time = df.loc[i, 'time'] price = df.loc[i, 'close'] current_rma = df.loc[i, 'rma'] current_rma_direction = df.loc[i, 'rma_direction'] current_vwap = df.loc[i, 'vwap'] current_twap = df.loc[i, 'twap'] prev_price = df.loc[i-1, 'close'] prev_rma = df.loc[i-1, 'rma'] prev_rma_direction = df.loc[i-1, 'rma_direction'] prev_vwap = df.loc[i-1, 'vwap'] prev_twap = df.loc[i-1, 'twap'] if time_exit_executed: continue # Check if trading should start (after 9:45) 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 09:45") else: # Skip processing until trading start time continue # Check for time-based exit (14:54) 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', current_rma_direction) self.addLogDataInfo(f"[TIME EXIT CE] at {_time} | Price: {price} | P/L: ₹{pl} | RMA Direction: {'UP' if current_rma_direction else 'DOWN'} | Time: 14:54") 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', current_rma_direction) self.addLogDataInfo(f"[TIME EXIT PE] at {_time} | Price: {price} | P/L: ₹{pl} | RMA Direction: {'UP' if current_rma_direction else 'DOWN'} | Time: 14:54") self.open_positions_price['PE'].clear() last_position_type = position position = None entry_price = None current_trade_entry_time = None wait_for_crossover = True # Wait for crossover after exit self.addLogDataInfo(f"[TIME EXIT EXECUTED] at {_time} | No new positions will be created after this time") # Skip further processing after time exit time_exit_executed = True continue # Check for RMA direction change exits if position == 'CE' and current_rma_direction != prev_rma_direction and not current_rma_direction: # RMA direction changed to false - close CE position 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, 'RMA DIRECTION CHANGE', current_rma_direction) self.addLogDataInfo(f"[RMA DIRECTION EXIT CE] at {_time} | Price: {price} | P/L: ₹{pl} | RMA Direction Changed to: {'UP' if current_rma_direction else 'DOWN'}") self.open_positions_price['CE'].clear() last_position_type = 'CE' position = None entry_price = None current_trade_entry_time = None wait_for_crossover = True # Wait for crossover after RMA direction exit continue # Skip further processing after exit if position == 'PE' and current_rma_direction != prev_rma_direction and current_rma_direction: # RMA direction changed to true - close PE position 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, 'RMA DIRECTION CHANGE', current_rma_direction) self.addLogDataInfo(f"[RMA DIRECTION EXIT PE] at {_time} | Price: {price} | P/L: ₹{pl} | RMA Direction Changed to: {'UP' if current_rma_direction else 'DOWN'}") self.open_positions_price['PE'].clear() last_position_type = 'PE' position = None entry_price = None current_trade_entry_time = None wait_for_crossover = True # Wait for crossover after RMA direction exit continue # Skip further processing after exit # 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: 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', current_rma_direction) self.addLogDataInfo(f"[PROFIT TARGET CE] at {_time} | Price: {price} | Profit: ₹{pl} | RMA Direction: {'UP' if current_rma_direction else 'DOWN'} | Target: {self.fixed_profit_target}") self.open_positions_price['CE'].clear() last_position_type = 'CE' position = None entry_price = None current_trade_entry_time = None wait_for_crossover = True # Wait for crossover after profit exit continue # Skip signal check after profit exit # 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', current_rma_direction) self.addLogDataInfo(f"[STOP LOSS CE] at {_time} | Price: {price} | Loss: ₹{abs(pl)} | RMA Direction: {'UP' if current_rma_direction else 'DOWN'} | Stop Loss: {self.fixed_stop_loss}") self.open_positions_price['CE'].clear() last_position_type = 'CE' position = None entry_price = None current_trade_entry_time = None wait_for_crossover = True # Wait for crossover after stop loss continue # Skip signal check after stop loss if position == 'PE' and entry_price is not None: current_profit = entry_price - price if current_profit >= self.fixed_profit_target: 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', current_rma_direction) self.addLogDataInfo(f"[PROFIT TARGET PE] at {_time} | Price: {price} | Profit: ₹{pl} | RMA Direction: {'UP' if current_rma_direction else 'DOWN'} | Target: {self.fixed_profit_target}") self.open_positions_price['PE'].clear() last_position_type = 'PE' position = None entry_price = None current_trade_entry_time = None wait_for_crossover = True # Wait for crossover after profit exit continue # Skip signal check after profit exit # 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', current_rma_direction) self.addLogDataInfo(f"[STOP LOSS PE] at {_time} | Price: {price} | Loss: ₹{abs(pl)} | RMA Direction: {'UP' if current_rma_direction else 'DOWN'} | Stop Loss: {self.fixed_stop_loss}") self.open_positions_price['PE'].clear() last_position_type = 'PE' position = None entry_price = None current_trade_entry_time = None wait_for_crossover = True # Wait for crossover after stop loss continue # Skip signal check after stop loss # Check for crossover conditions price_cross_above_rma = (price > current_rma) and (prev_price <= prev_rma) price_cross_below_rma = (price < current_rma) and (prev_price >= prev_rma) rma_direction_change = current_rma_direction != prev_rma_direction # Reset wait_for_crossover if we get a proper crossover if wait_for_crossover and (price_cross_above_rma or price_cross_below_rma or rma_direction_change): wait_for_crossover = False self.addLogDataInfo(f"[CROSSOVER DETECTED] at {_time} | Price Cross: Above={price_cross_above_rma}, Below={price_cross_below_rma} | RMA Direction Change: {rma_direction_change}") # CE BUY LOGIC: Close price > RMA AND RMA direction is UP AND no existing CE position AND crossover condition ce_buy_condition = (price > current_rma and current_rma_direction and price > current_twap and position != 'CE') pe_buy_condition = (price < current_rma and not current_rma_direction and price < current_twap and position != 'PE') #self.addLogDataInfo(f"[ce_buy]: {ce_buy_condition} | [pe_buy]: {pe_buy_condition}, price={price}, current_rma_direction={current_rma_direction}, current_rma ={current_rma}, current_twap={current_twap}, position={position} | ") if ce_buy_condition: # If we're waiting for crossover, check if we have a valid crossover if wait_for_crossover and last_position_type == 'CE': if not (price_cross_above_rma or rma_direction_change): self.addLogDataInfo(f"[SKIP CE REENTRY] at {_time} | Waiting for crossover | Last Position: {last_position_type}") continue # Close any open PE position first if exists 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', current_rma_direction) self.addLogDataInfo(f"[CLOSE PE] Before CE Buy at {_time} | Price: {price} | P/L: ₹{pl} | RMA Direction: {'UP' if current_rma_direction else 'DOWN'}") self.open_positions_price['PE'].clear() last_position_type = 'PE' position = None entry_price = None # Enter CE position only if no existing position if position is None: strikePrices = self.getNiftyStrikePriceGivenCurrentValue(price, 'CE') 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)) crossover_info = "" if price_cross_above_rma: crossover_info = " | Price Crossed Above RMA" elif rma_direction_change: crossover_info = " | RMA Direction Changed" self.addLogDataInfo(f"[BUY CE] at {_time} | Price: {price} > RMA: {current_rma:.2f} | RMA Direction: UP{crossover_info} | VWAP: {current_vwap:.2f} | TWAP: {current_twap:.2f}") position = 'CE' entry_price = price current_trade_entry_time = _time wait_for_crossover = False # Reset crossover wait else: self.addLogDataInfo(f"[SKIP CE BUY] at {_time} | Position already exists: {position}") # PE BUY LOGIC: Close price < RMA AND RMA direction is DOWN AND no existing PE position AND crossover condition elif pe_buy_condition: # If we're waiting for crossover, check if we have a valid crossover if wait_for_crossover and last_position_type == 'PE': if not (price_cross_below_rma or rma_direction_change): self.addLogDataInfo(f"[SKIP PE REENTRY] at {_time} | Waiting for crossover | Last Position: {last_position_type}") continue # Close any open CE position first if exists 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', current_rma_direction) self.addLogDataInfo(f"[CLOSE CE] Before PE Buy at {_time} | Price: {price} | P/L: ₹{pl} | RMA Direction: {'UP' if current_rma_direction else 'DOWN'}") self.open_positions_price['CE'].clear() last_position_type = 'CE' position = None entry_price = None # Enter PE position only if no existing position if position is None: strikePrices = self.getNiftyStrikePriceGivenCurrentValue(price, 'PE') 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)) crossover_info = "" if price_cross_below_rma: crossover_info = " | Price Crossed Below RMA" elif rma_direction_change: crossover_info = " | RMA Direction Changed" self.addLogDataInfo(f"[BUY PE] at {_time} | Price: {price} < RMA: {current_rma:.2f} | RMA Direction: DOWN{crossover_info} | VWAP: {current_vwap:.2f} | TWAP: {current_twap:.2f}") position = 'PE' entry_price = price current_trade_entry_time = _time wait_for_crossover = False # Reset crossover wait else: self.addLogDataInfo(f"[SKIP PE BUY] at {_time} | Position already exists: {position}") def fetch_all_data(self, elapsed: int): now_ist = datetime.now(self.ist) start_time = now_ist.replace(hour=9, minute=15, second=0) - timedelta(days=elapsed+1) 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 RMA Direction df['rma_direction'] = self.calculate_rma_direction(df['rma'], self.rma_direction_smoothing) # 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.reset_index(drop=True) # This resets index to 0, 1, 2, 3... # Handle trend following orders self.addLogDataInfo("") self.addLogDataInfo("Handling RMA-based trend following orders...") self.addLogDataInfo(f"RMA Length: {self.rma_length}") self.addLogDataInfo(f"RMA Direction Smoothing: {self.rma_direction_smoothing}") 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"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"RMA Direction Change Exit: Enabled") 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"Final RMA Direction: {'UP' if df.iloc[-1]['rma_direction'] else 'DOWN'}") 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() 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)