import logging import sys from datetime import datetime, timedelta from typing import List, Dict, Any import pytz import pandas as pd import requests import json import numpy as np from firstock import firstock import os import glob import time from pathlib import Path import json LOCK_FILE = "/tmp/puthayal.lock" def acquire_lock(timeout=30, interval=3): """Try to acquire lock within timeout seconds.""" start_time = time.time() print(f"[INFO] Attempting to acquire lock: {LOCK_FILE}") while True: try: # Try to create the lock file exclusively fd = os.open(LOCK_FILE, os.O_CREAT | os.O_EXCL | os.O_WRONLY) os.close(fd) print(f"[INFO] Lock acquired successfully: {LOCK_FILE}") return True # Lock acquired except FileExistsError: # Lock file already exists print(f"[WARN] Lock file already exists: {LOCK_FILE}") if time.time() - start_time >= timeout: print("[ERROR] Timeout reached. Could not acquire lock.") return False print(f"[INFO] Retrying in {interval} seconds...") time.sleep(interval) def release_lock(): """Remove lock file if it exists.""" try: os.remove(LOCK_FILE) print(f"[INFO] Lock file removed: {LOCK_FILE}") except FileNotFoundError: print(f"[WARN] Lock file not found, nothing to remove: {LOCK_FILE}") pass def _get_config_path() -> Path: """Get the path to the config file in the same directory as this Python file""" return Path(__file__).parent / 'symbol_config.json' def _load_symbol_config() -> dict: """ Load symbol configuration from JSON file. Returns empty dict if config file doesn't exist or has issues. """ config_path = _get_config_path() print(f"[TRACE] Loading symbol config from: {config_path}") try: if config_path.exists(): print(f"[TRACE] Config file exists at: {config_path}") with open(config_path, 'r') as f: config = json.load(f) print(f"[TRACE] Successfully loaded config file") print(f"[TRACE] Config contents: {config}") # Get symbol_mapping, return empty dict if not present symbol_mapping = config.get('symbol_mapping', {}) print(f"[TRACE] Extracted symbol_mapping: {symbol_mapping}") if not symbol_mapping: print(f"[WARNING] symbol_mapping is empty or not found in config") else: print(f"[INFO] Loaded {len(symbol_mapping)} symbol mappings") return symbol_mapping else: print(f"[WARNING] Config file does not exist at: {config_path}") print(f"[INFO] Returning empty dict (no default mappings)") return {} except json.JSONDecodeError as e: print(f"[ERROR] JSON decode error in config file: {e}") print(f"[ERROR] File may be malformed. Returning empty dict.") return {} except IOError as e: print(f"[ERROR] IO error reading config file: {e}") print(f"[ERROR] Returning empty dict.") return {} except Exception as e: print(f"[ERROR] Unexpected error loading config: {e}") print(f"[ERROR] Returning empty dict.") return {} class StockDataFetcher: def __init__(self, client_details: List[str]): self.client_details = client_details self.user_id = client_details[0] self.ist = pytz.timezone('Asia/Kolkata') self.logger = self.setup_logger() # Data fetch parameters self.interval = 1 # Create entries flag self.create_entries = False # Fixed profit target (in points) self.fixed_profit = 200 # Adjust this value as needed # Store symbol for entry creation self.current_symbol = 'NIFTY' # Load config once at module level self.SYMBOL_MAPPING = _load_symbol_config() 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}') def addLogDataInfo(self, text): self.logger.info(f'{text}') 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 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 translate_symbol(self, symbol: str) -> str: self.logger.info(f"[translate_symbol] Called with symbol: '{symbol}'") # Convert to lowercase for case-insensitive matching symbol_lower = symbol.lower() self.logger.debug(f"[translate_symbol] Lowercase symbol: '{symbol_lower}'") # Check if symbol exists in mapping if symbol_lower in self.SYMBOL_MAPPING: result = self.SYMBOL_MAPPING[symbol_lower] self.logger.info(f"[translate_symbol] Found mapping: '{symbol_lower}' -> '{result}'") return result else: self.logger.warning(f"[translate_symbol] Symbol '{symbol_lower}' not found in mapping") self.logger.warning(f"[translate_symbol] Available mappings: {list(self.SYMBOL_MAPPING.keys())}") # Default to NSE:NIFTY self.logger.info(f"[translate_symbol] Using default: 'NSE:NIFTY' for symbol: '{symbol}'") return 'NSE:NIFTY' 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) + "mi", ) if df.empty: return df df = df.sort_values(by='epochTime', ascending=True) 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 fetch_time_price_series(self, exchange: str, trading_symbol: str, start_time: str, end_time: str, interval: str) -> pd.DataFrame: try: self.addLogDataInfo(f"API Request: exchange={exchange}, symbol={trading_symbol}") self.addLogDataInfo(f"Start: {start_time}, End: {end_time}, Interval: {interval}") response = firstock.timePriceSeries( userId=self.user_id, exchange=exchange, tradingSymbol=trading_symbol, startTime=start_time, endTime=end_time, interval=interval ) self.addLogDataInfo(f"API Response Status: {response.get('status')}") if response.get("status") == "success": data = response.get("data", []) if data: self.addLogDataInfo(f"API returned {len(data)} candles") if len(data) > 0: first_date = data[0].get('time', 'N/A') last_date = data[-1].get('time', 'N/A') self.addLogDataInfo(f"First candle: {first_date}") self.addLogDataInfo(f"Last candle: {last_date}") df_temp = pd.DataFrame(data) if 'time' in df_temp.columns: df_temp['datetime'] = pd.to_datetime(df_temp['time'], format='%H:%M:%S %d-%m-%Y') unique_dates = df_temp['datetime'].dt.date.nunique() self.addLogDataInfo(f"Unique trading days in data: {unique_dates}") return df_temp else: self.addLogDataInfo("API returned empty data list") return pd.DataFrame() else: self.addLogDataDebug(f"API error response: {response}") return pd.DataFrame() except Exception as e: self.addLogDataDebug(f"Error fetching series: {e}") import traceback self.addLogDataDebug(traceback.format_exc()) return pd.DataFrame() def calculate_emas(self, df: pd.DataFrame) -> pd.DataFrame: """ Calculate EMA 9, 21, and 50 for close price """ df = df.copy() # Calculate EMAs df['ema_9'] = df['close'].ewm(span=9, adjust=False).mean() df['ema_21'] = df['close'].ewm(span=21, adjust=False).mean() df['ema_50'] = df['close'].ewm(span=50, adjust=False).mean() self.addLogDataInfo(f"Calculated EMAs: 9, 21, 50") return df def _send_entry_request(self, url: str, params: dict, endpoint_name: str): """Send entry request to a single endpoint using GET""" try: self.addLogDataInfo(f"📤 Sending entry to {endpoint_name}: {url}") self.addLogDataInfo(f"📋 Params: {params}") # Use GET instead of POST response = requests.get(url, params=params, timeout=10) self.addLogDataInfo(f"📊 Response Status Code: {response.status_code}") if response.status_code == 200: self.addLogDataInfo(f"✅ {endpoint_name} - Entry created successfully") self.addLogDataInfo(f"📄 Response: {response.text[:200]}") # Log first 200 chars else: self.addLogDataInfo(f"❌ {endpoint_name} - Failed with status: {response.status_code}") self.addLogDataInfo(f"📄 Response: {response.text[:200]}") except requests.exceptions.Timeout: self.addLogDataInfo(f"⏱️ {endpoint_name} - Request timeout") except requests.exceptions.ConnectionError: self.addLogDataInfo(f"🔌 {endpoint_name} - Connection error") except Exception as e: self.addLogDataInfo(f"❌ {endpoint_name} - Error: {e}") import traceback self.addLogDataInfo(traceback.format_exc()) def create_trend_entry(self, tick_time_str, instrument, close_price, signal, lot_count=1, trigger_type=None, is_exit=False, exit_instrument=None): """Create entry/exit order using GET parameters""" self.addLogDataInfo(f"🔍 create_trend_entry called:") self.addLogDataInfo(f" tick_time_str: {tick_time_str}") self.addLogDataInfo(f" instrument: {instrument}") self.addLogDataInfo(f" close_price: {close_price}") self.addLogDataInfo(f" signal: {signal}") self.addLogDataInfo(f" create_entries flag: {self.create_entries}") if not self.create_entries: self.addLogDataInfo("⚠️ create_entries is False, skipping entry creation") return # Convert tick_time_str from YYYYMMDDHHMMSS to human readable format (with hyphen) try: dt = datetime.strptime(tick_time_str, '%Y%m%d%H%M%S') # Format as: DD-MM-YYYY-HH:MM:SS (with hyphen between date and time) human_readable_time = dt.strftime('%d-%m-%Y-%H:%M:%S') self.addLogDataInfo(f" Converted time: {human_readable_time}") except (ValueError, TypeError) as e: self.logger.error(f"Invalid tick_time_str: {tick_time_str}, Error: {e}") return # Get the base instrument name (without CE/PE) base_instrument = instrument self.addLogDataInfo(f" Base instrument: {base_instrument}") # Determine the strike price multiplier based on instrument if 'SENSEX' in instrument or 'BANKEX' in instrument: multiplier = 100 else: multiplier = 50 # Default for NIFTY and others self.addLogDataInfo(f" Multiplier: {multiplier}") # Calculate the nearest multiple of multiplier try: close_price_int = int(float(close_price)) # Round to nearest multiple of multiplier nearest_multiple = round(close_price_int / multiplier) * multiplier # Ensure we don't go below minimum if nearest_multiple < 0: nearest_multiple = multiplier self.addLogDataInfo(f" Close price: {close_price_int}, Nearest multiple: {nearest_multiple}") except (ValueError, TypeError) as e: self.logger.error(f"Invalid close_price: {close_price}, Error: {e}") return # Determine the instrument with strike and signal if is_exit: # For SELL, use the instrument from the previous BUY instrument_with_strike = exit_instrument clean_signal = 'SELL' self.addLogDataInfo(f" EXIT: Using instrument: {instrument_with_strike}") else: # For BUY, create new instrument with CE/PE if signal == 'BUY_CE': instrument_with_strike = f"{base_instrument}{nearest_multiple}CE" clean_signal = 'BUY' self.addLogDataInfo(f" ENTRY CE: {instrument_with_strike}") elif signal == 'BUY_PE': instrument_with_strike = f"{base_instrument}{nearest_multiple}PE" clean_signal = 'BUY' self.addLogDataInfo(f" ENTRY PE: {instrument_with_strike}") else: instrument_with_strike = f"{base_instrument}{nearest_multiple}" clean_signal = 'BUY' self.addLogDataInfo(f" ENTRY: {instrument_with_strike}") # Use int value for closePrice try: close_price_int = int(float(close_price)) except (ValueError, TypeError) as e: self.logger.error(f"Invalid close_price: {close_price}, Error: {e}") return params = { 'tickTime': str(human_readable_time), 'instrument': str(instrument_with_strike), 'closePrice': str(close_price_int), 'signal': str(clean_signal), 'orderType': str(lot_count), 'remarks': str(trigger_type) if trigger_type else 'EMA' } self.addLogDataInfo(f"📝 Final params: {params}") if is_exit: self.addLogDataInfo(f"🔄 Creating SELL entry for instrument: {instrument_with_strike} (previous BUY)") else: self.addLogDataInfo(f"🟢 Creating BUY entry with instrument: {instrument_with_strike} (nearest multiple: {nearest_multiple})") endpoints = [ ("Multibagger", "http://143.244.141.41/php/createEntriesMft.php"), ("LastSupper", "http://139.59.6.25/php/createEntriesMft.php"), ("GoodFriday", "http://68.183.85.105/php/createEntriesMft.php"), ] for endpoint_name, url in endpoints: self._send_entry_request(url, params, endpoint_name) def detect_ema_signals(self, df: pd.DataFrame, symbol: str) -> tuple: """ Detect trading signals based on EMA crossover: - BUY CE: (EMA9 > EMA21 and prevEMA9 <= prevEMA21) AND close > EMA50 - BUY PE: (EMA9 < EMA21 and prevEMA9 >= prevEMA21) AND close < EMA50 - EXIT CE: - Profit: entry price + fixed_profit >= current close - Crossover: price < EMA50 - EXIT PE: - Profit: entry price - fixed_profit <= current close - Crossover: price > EMA50 Returns: - signals: List of signal dictionaries - signal_summary: List of signal summaries with P&L """ self.addLogDataInfo(f"🔍 detect_ema_signals called with symbol: {symbol}") self.addLogDataInfo(f"🔍 create_entries flag: {self.create_entries}") self.addLogDataInfo(f"🔍 Fixed Profit Target: {self.fixed_profit} points") signals = [] signal_summary = [] df = df.copy() df = df.dropna(subset=['ema_9', 'ema_21', 'ema_50']) if len(df) < 2: self.addLogDataInfo("⚠️ Not enough data for signal detection (need at least 2 rows)") return signals, signal_summary # Track position state in_position = False position_type = None # 'CE' or 'PE' entry_price = None entry_datetime = None entry_instrument = None entry_signal = None # 'BUY_CE' or 'BUY_PE' self.addLogDataInfo("=" * 80) self.addLogDataInfo("SCANNING FOR EMA CROSSOVER SIGNALS...") self.addLogDataInfo("=" * 80) self.addLogDataInfo(f"Initial State: No position") self.addLogDataInfo(f"Total rows to scan: {len(df)}") self.addLogDataInfo(f"Fixed Profit Target: {self.fixed_profit} points") self.addLogDataInfo("-" * 40) for i in range(1, len(df)): prev_row = df.iloc[i-1] curr_row = df.iloc[i] # Get EMA values ema9_prev = prev_row['ema_9'] ema21_prev = prev_row['ema_21'] ema9_curr = curr_row['ema_9'] ema21_curr = curr_row['ema_21'] ema50_curr = curr_row['ema_50'] close_curr = curr_row['close'] # Check crossover conditions # BUY CE: (EMA9 > EMA21 and prevEMA9 <= prevEMA21) AND close > EMA50 buy_ce_condition = (ema9_curr > ema21_curr) and (ema9_prev <= ema21_prev) and (close_curr > ema50_curr) # BUY PE: (EMA9 < EMA21 and prevEMA9 >= prevEMA21) AND close < EMA50 buy_pe_condition = (ema9_curr < ema21_curr) and (ema9_prev >= ema21_prev) and (close_curr < ema50_curr) # Debug logging for each row if i % 10 == 0 or buy_ce_condition or buy_pe_condition: self.addLogDataInfo(f"Row {i}: Time={curr_row['datetime'].strftime('%H:%M:%S')}, Close={close_curr:.2f}, EMA9={ema9_curr:.2f}, EMA21={ema21_curr:.2f}, EMA50={ema50_curr:.2f}") self.addLogDataInfo(f" buy_ce_condition={buy_ce_condition}, buy_pe_condition={buy_pe_condition}") if in_position: self.addLogDataInfo(f" In position: {position_type} at ₹{entry_price:.2f}") if position_type == 'CE': profit_target = entry_price + self.fixed_profit self.addLogDataInfo(f" CE Profit Target: ₹{profit_target:.2f}") self.addLogDataInfo(f" Current Close: ₹{close_curr:.2f}") self.addLogDataInfo(f" Target Hit: {close_curr >= profit_target}") self.addLogDataInfo(f" Price < EMA50: {close_curr < ema50_curr} (Exit condition)") elif position_type == 'PE': profit_target = entry_price - self.fixed_profit self.addLogDataInfo(f" PE Profit Target: ₹{profit_target:.2f}") self.addLogDataInfo(f" Current Close: ₹{close_curr:.2f}") self.addLogDataInfo(f" Target Hit: {close_curr <= profit_target}") self.addLogDataInfo(f" Price > EMA50: {close_curr > ema50_curr} (Exit condition)") # Check if we're in a position and exit conditions are met if in_position: exit_signal = False exit_reason = "" if position_type == 'CE': # CE Exit Condition 1: Profit target hit profit_target = entry_price + self.fixed_profit if close_curr >= profit_target: exit_signal = True exit_reason = f"Profit target hit (₹{profit_target:.2f})" # CE Exit Condition 2: Price crosses below EMA50 elif close_curr < ema50_curr: exit_signal = True exit_reason = f"Price crossed below EMA50 (₹{ema50_curr:.2f})" elif position_type == 'PE': # PE Exit Condition 1: Profit target hit profit_target = entry_price - self.fixed_profit if close_curr <= profit_target: exit_signal = True exit_reason = f"Profit target hit (₹{profit_target:.2f})" # PE Exit Condition 2: Price crosses above EMA50 elif close_curr > ema50_curr: exit_signal = True exit_reason = f"Price crossed above EMA50 (₹{ema50_curr:.2f})" if exit_signal: self.addLogDataInfo(f"🎯 EXIT CONDITION MET at {curr_row['datetime'].strftime('%Y-%m-%d %H:%M:%S')}") self.addLogDataInfo(f" Reason: {exit_reason}") signal = { 'type': f'{position_type} EXIT', 'datetime': curr_row['datetime'].isoformat(), 'price': float(round(close_curr, 2)), 'ema_9': float(round(ema9_curr, 2)), 'ema_21': float(round(ema21_curr, 2)), 'ema_50': float(round(ema50_curr, 2)), 'exit_reason': exit_reason } signals.append(signal) # Calculate P&L if position_type == 'CE': pl = round(close_curr - entry_price, 2) else: # PE pl = round(entry_price - close_curr, 2) # Add to signal summary signal_summary.append({ 'type': position_type, 'entry_datetime': entry_datetime.isoformat() if entry_datetime else None, 'entry_price': float(round(entry_price, 2)), 'exit_price': float(round(close_curr, 2)), 'pl': pl, 'exit_datetime': curr_row['datetime'].isoformat(), 'exit_reason': exit_reason }) # Create SELL entry if enabled if self.create_entries and entry_instrument: self.addLogDataInfo(f"🚀 Creating EXIT order for {position_type}...") sell_time = curr_row['datetime'] - timedelta(minutes=5) sell_tick_time = sell_time.strftime('%Y%m%d%H%M%S') self.create_trend_entry( tick_time_str=sell_tick_time, instrument=symbol, close_price=close_curr, signal='SELL', lot_count=1, trigger_type='EXIT', is_exit=True, exit_instrument=entry_instrument ) else: self.addLogDataInfo(f"⚠️ create_entries is False or no entry_instrument, skipping exit order") in_position = False position_type = None entry_price = None entry_datetime = None entry_instrument = None entry_signal = None self.addLogDataInfo(f"🎯 {signal['type']} SIGNAL DETECTED!") self.addLogDataInfo(f" Time: {curr_row['datetime'].strftime('%Y-%m-%d %H:%M:%S')}") self.addLogDataInfo(f" Price: ₹{signal['price']}") self.addLogDataInfo(f" P&L: ₹{pl:.2f}") self.addLogDataInfo(f" Reason: {exit_reason}") self.addLogDataInfo("-" * 40) # After exit, continue to next row to allow new entry continue # BUY CE: (EMA9 > EMA21 and prevEMA9 <= prevEMA21) AND close > EMA50 if buy_ce_condition and not in_position: self.addLogDataInfo(f"🟢 BUY CE CONDITION MET at {curr_row['datetime'].strftime('%Y-%m-%d %H:%M:%S')}") signal = { 'type': 'BUY CE', 'datetime': curr_row['datetime'].isoformat(), 'price': float(round(close_curr, 2)), 'ema_9': float(round(ema9_curr, 2)), 'ema_21': float(round(ema21_curr, 2)), 'ema_50': float(round(ema50_curr, 2)) } signals.append(signal) in_position = True position_type = 'CE' entry_price = close_curr entry_datetime = curr_row['datetime'] entry_signal = 'BUY_CE' # Create instrument for this BUY if 'SENSEX' in symbol or 'BANKEX' in symbol: multiplier = 100 else: multiplier = 50 nearest_multiple = round(int(float(close_curr)) / multiplier) * multiplier entry_instrument = f"{symbol}{nearest_multiple}CE" self.addLogDataInfo(f" Entry Instrument: {entry_instrument}") self.addLogDataInfo(f" Profit Target: ₹{entry_price + self.fixed_profit:.2f}") self.addLogDataInfo(f" Exit Condition: Price < EMA50 (₹{ema50_curr:.2f})") # Create BUY entry if enabled if self.create_entries: self.addLogDataInfo(f"🚀 Creating BUY CE order...") tick_time = curr_row['datetime'].strftime('%Y%m%d%H%M%S') self.create_trend_entry( tick_time_str=tick_time, instrument=symbol, close_price=close_curr, signal='BUY_CE', lot_count=1, trigger_type='EMA_CROSS_ABOVE', is_exit=False, exit_instrument=None ) else: self.addLogDataInfo(f"⚠️ create_entries is False, skipping order creation") self.addLogDataInfo(f"🟢 BUY CE SIGNAL DETECTED!") self.addLogDataInfo(f" Time: {curr_row['datetime'].strftime('%Y-%m-%d %H:%M:%S')}") self.addLogDataInfo(f" Price: ₹{signal['price']}") self.addLogDataInfo(f" EMA 9: {signal['ema_9']:.2f}") self.addLogDataInfo(f" EMA 21: {signal['ema_21']:.2f}") self.addLogDataInfo(f" EMA 50: {signal['ema_50']:.2f}") self.addLogDataInfo(f" Condition: EMA9 > EMA21 and prevEMA9 <= prevEMA21 AND Close > EMA50") self.addLogDataInfo(f" Profit Target: ₹{entry_price + self.fixed_profit:.2f}") self.addLogDataInfo(f" Exit Condition: Price < EMA50") self.addLogDataInfo("-" * 40) # BUY PE: (EMA9 < EMA21 and prevEMA9 >= prevEMA21) AND close < EMA50 elif buy_pe_condition and not in_position: self.addLogDataInfo(f"🔴 BUY PE CONDITION MET at {curr_row['datetime'].strftime('%Y-%m-%d %H:%M:%S')}") signal = { 'type': 'BUY PE', 'datetime': curr_row['datetime'].isoformat(), 'price': float(round(close_curr, 2)), 'ema_9': float(round(ema9_curr, 2)), 'ema_21': float(round(ema21_curr, 2)), 'ema_50': float(round(ema50_curr, 2)) } signals.append(signal) in_position = True position_type = 'PE' entry_price = close_curr entry_datetime = curr_row['datetime'] entry_signal = 'BUY_PE' # Create instrument for this BUY if 'SENSEX' in symbol or 'BANKEX' in symbol: multiplier = 100 else: multiplier = 50 nearest_multiple = round(int(float(close_curr)) / multiplier) * multiplier entry_instrument = f"{symbol}{nearest_multiple}PE" self.addLogDataInfo(f" Entry Instrument: {entry_instrument}") self.addLogDataInfo(f" Profit Target: ₹{entry_price - self.fixed_profit:.2f}") self.addLogDataInfo(f" Exit Condition: Price > EMA50 (₹{ema50_curr:.2f})") # Create BUY entry if enabled (BUY PE for short) if self.create_entries: self.addLogDataInfo(f"🚀 Creating BUY PE order...") tick_time = curr_row['datetime'].strftime('%Y%m%d%H%M%S') self.create_trend_entry( tick_time_str=tick_time, instrument=symbol, close_price=close_curr, signal='BUY_PE', lot_count=1, trigger_type='EMA_CROSS_BELOW', is_exit=False, exit_instrument=None ) else: self.addLogDataInfo(f"⚠️ create_entries is False, skipping order creation") self.addLogDataInfo(f"🔴 BUY PE SIGNAL DETECTED!") self.addLogDataInfo(f" Time: {curr_row['datetime'].strftime('%Y-%m-%d %H:%M:%S')}") self.addLogDataInfo(f" Price: ₹{signal['price']}") self.addLogDataInfo(f" EMA 9: {signal['ema_9']:.2f}") self.addLogDataInfo(f" EMA 21: {signal['ema_21']:.2f}") self.addLogDataInfo(f" EMA 50: {signal['ema_50']:.2f}") self.addLogDataInfo(f" Condition: EMA9 < EMA21 and prevEMA9 >= prevEMA21 AND Close < EMA50") self.addLogDataInfo(f" Profit Target: ₹{entry_price - self.fixed_profit:.2f}") self.addLogDataInfo(f" Exit Condition: Price > EMA50") self.addLogDataInfo("-" * 40) # Add final open position to signal summary (if any) if in_position and entry_price is not None: last_row = df.iloc[-1] signal_summary.append({ 'type': f'{position_type} (Open)', 'entry_datetime': entry_datetime.isoformat() if entry_datetime else None, 'entry_price': float(round(entry_price, 2)), 'exit_price': float(round(last_row['close'], 2)), 'pl': None, # No P&L for open position 'exit_datetime': last_row['datetime'].isoformat(), 'exit_reason': 'Position still open' }) # Print final state self.addLogDataInfo("=" * 80) self.addLogDataInfo("FINAL STATE:") if in_position: self.addLogDataInfo(f" Position: {position_type} at ₹{entry_price:.2f}") self.addLogDataInfo(f" Current Price: ₹{df.iloc[-1]['close']:.2f}") if position_type == 'CE': unrealized_pl = round(df.iloc[-1]['close'] - entry_price, 2) profit_target = entry_price + self.fixed_profit self.addLogDataInfo(f" Unrealized P&L: ₹{unrealized_pl:.2f}") self.addLogDataInfo(f" Profit Target: ₹{profit_target:.2f}") self.addLogDataInfo(f" Target Distance: ₹{profit_target - df.iloc[-1]['close']:.2f}") self.addLogDataInfo(f" EMA50: ₹{df.iloc[-1]['ema_50']:.2f}") self.addLogDataInfo(f" Price < EMA50: {df.iloc[-1]['close'] < df.iloc[-1]['ema_50']}") elif position_type == 'PE': unrealized_pl = round(entry_price - df.iloc[-1]['close'], 2) profit_target = entry_price - self.fixed_profit self.addLogDataInfo(f" Unrealized P&L: ₹{unrealized_pl:.2f}") self.addLogDataInfo(f" Profit Target: ₹{profit_target:.2f}") self.addLogDataInfo(f" Target Distance: ₹{df.iloc[-1]['close'] - profit_target:.2f}") self.addLogDataInfo(f" EMA50: ₹{df.iloc[-1]['ema_50']:.2f}") self.addLogDataInfo(f" Price > EMA50: {df.iloc[-1]['close'] > df.iloc[-1]['ema_50']}") else: self.addLogDataInfo(" Position: None") self.addLogDataInfo("=" * 80) # Print summary self.addLogDataInfo("=" * 80) self.addLogDataInfo(f"SCAN COMPLETE: Found {len(signals)} signals") if len(signals) > 0: buy_ce_count = sum(1 for s in signals if s['type'] == 'BUY CE') buy_pe_count = sum(1 for s in signals if s['type'] == 'BUY PE') exit_count = sum(1 for s in signals if 'EXIT' in s['type']) self.addLogDataInfo(f" BUY CE: {buy_ce_count}") self.addLogDataInfo(f" BUY PE: {buy_pe_count}") self.addLogDataInfo(f" EXITS: {exit_count}") # Print signal summary with P&L if signal_summary: self.addLogDataInfo("\n" + "=" * 80) self.addLogDataInfo("SIGNAL SUMMARY WITH P&L") self.addLogDataInfo("=" * 80) for idx, summary in enumerate(signal_summary, 1): if summary['pl'] is not None: pl_str = f"₹{summary['pl']:.2f}" pl_color = "🟢" if summary['pl'] > 0 else "🔴" if summary['pl'] < 0 else "⚪" self.addLogDataInfo(f"{idx}. {summary['type']}") self.addLogDataInfo(f" Entry: {summary['entry_datetime']} at ₹{summary['entry_price']:.2f}") self.addLogDataInfo(f" Exit: {summary['exit_datetime']} at ₹{summary['exit_price']:.2f}") self.addLogDataInfo(f" P&L: {pl_color} {pl_str}") if 'exit_reason' in summary: self.addLogDataInfo(f" Reason: {summary['exit_reason']}") else: self.addLogDataInfo(f"{idx}. {summary['type']}") self.addLogDataInfo(f" Entry: {summary['entry_datetime']} at ₹{summary['entry_price']:.2f}") self.addLogDataInfo(f" Current: {summary['exit_datetime']} at ₹{summary['exit_price']:.2f}") self.addLogDataInfo(f" Status: OPEN POSITION") if 'exit_reason' in summary: self.addLogDataInfo(f" Reason: {summary['exit_reason']}") self.addLogDataInfo("-" * 40) self.addLogDataInfo("=" * 80) return signals, signal_summary def fetch_all_data(self, symbol: str = 'nifty', elapsed: int = 0, create_entries: bool = False): """ Fetch data for a given symbol Args: symbol: Symbol name (e.g., 'nifty', 'banknifty', 'sensex') elapsed: Number of days to subtract from both start_time and end_time start_time = 9:15 AM of (current date - elapsed - 3) end_time = 15:30 PM of (current date - elapsed) create_entries: Whether to create entries """ # Set the create_entries flag and current symbol self.create_entries = create_entries self.current_symbol = symbol.upper() self.addLogDataInfo(f"🚀 fetch_all_data called with:") self.addLogDataInfo(f" symbol: {symbol}") self.addLogDataInfo(f" elapsed: {elapsed}") self.addLogDataInfo(f" create_entries: {create_entries}") self.addLogDataInfo(f" fixed_profit: {self.fixed_profit}") now_ist = datetime.now(self.ist) current_date = now_ist.date() # Calculate start_time: 9:15 AM of (current date - elapsed - 3 days) start_date = current_date - timedelta(days=elapsed + 3) start_time = datetime.combine(start_date, datetime.min.time()).replace(hour=9, minute=15, second=0) start_time = self.ist.localize(start_time) # Calculate end_time: 15:30 PM of (current date - elapsed) end_date = current_date - timedelta(days=elapsed) end_time = datetime.combine(end_date, datetime.min.time()).replace(hour=15, minute=30, second=0) end_time = self.ist.localize(end_time) # If end_time is in the future (today), use current time if end_time > now_ist: end_time = now_ist # Translate symbol to exchange:format translated_symbol = self.translate_symbol(symbol) self.addLogDataInfo(f"📊 Fetching data for {translated_symbol}") self.addLogDataInfo(f" Start: {start_time.strftime('%Y-%m-%d %H:%M:%S')}") self.addLogDataInfo(f" End: {end_time.strftime('%Y-%m-%d %H:%M:%S')}") self.addLogDataInfo(f" Elapsed parameter: {elapsed} days") self.addLogDataInfo(f" Start date = today - {elapsed} - 3 = {start_date}") self.addLogDataInfo(f" End date = today - {elapsed} = {end_date}") self.addLogDataInfo(f" Create Entries: {create_entries}") df = self.process_symbol_data(translated_symbol, self.interval, start_time, end_time) if df.empty: self.addLogDataInfo("❌ No data fetched") return df['datetime'] = pd.to_datetime(df['time'], format='%H:%M:%S %d-%m-%Y') self.addLogDataInfo(f"📊 Raw data shape: {df.shape}") # Calculate EMAs before filtering df = self.calculate_emas(df) self.addLogDataInfo(f"📊 Data shape after EMA calculation: {df.shape}") # Filter to keep only the last date's data if not df.empty: last_date = df['datetime'].dt.date.max() df_before_filter = len(df) df = df[df['datetime'].dt.date == last_date] self.addLogDataInfo(f"📅 Filtered data: kept only last date ({last_date})") self.addLogDataInfo(f" Removed {df_before_filter - len(df)} rows from other dates") self.addLogDataInfo(f"📊 Data shape after filtering: {df.shape}") total_points = len(df) self.addLogDataInfo(f"📊 Total data points: {total_points}") if total_points > 0: first_date = df['datetime'].min() last_date = df['datetime'].max() self.addLogDataInfo(f"📅 Data date range: {first_date} to {last_date}") unique_dates = df['datetime'].dt.date.nunique() self.addLogDataInfo(f"📅 Unique trading days in filtered data: {unique_dates}") dates_list = sorted(df['datetime'].dt.date.unique()) self.addLogDataInfo(f"📅 Dates in data: {[d.strftime('%Y-%m-%d') for d in dates_list]}") # Log EMA values for the last row last_row = df.iloc[-1] self.addLogDataInfo(f"📊 Latest EMA values:") self.addLogDataInfo(f" EMA 9: {last_row['ema_9']:.2f}") self.addLogDataInfo(f" EMA 21: {last_row['ema_21']:.2f}") self.addLogDataInfo(f" EMA 50: {last_row['ema_50']:.2f}") self.addLogDataInfo(f" Close: {last_row['close']:.2f}") # Detect EMA crossover signals self.addLogDataInfo(f"🔍 Starting signal detection with symbol: {self.current_symbol}") signals, signal_summary = self.detect_ema_signals(df, self.current_symbol) self.addLogDataInfo(f"✅ Data fetch completed for {symbol.upper()}") self.addLogDataInfo(f" Total signals found: {len(signals)}") self.addLogDataInfo(f" Total signal summaries: {len(signal_summary)}") if __name__ == "__main__": print("[INFO] Script started.") if not acquire_lock(): print("Another instance is running. Exiting.") sys.exit(1) try: print("[INFO] Lock acquired, proceeding with execution.") # Hardcoded client details client_details = ['DB1485', 'ABcd#1234', '14121985', 'DB1485_API', 'd9ce8d1261a834458847929c11ea1047'] # Parse command line arguments symbol = sys.argv[1] if len(sys.argv) > 1 else 'nifty' elapsed = int(sys.argv[2]) if len(sys.argv) > 2 else 0 create_entries = (sys.argv[3].lower() in ('true', '1') if len(sys.argv) > 3 else False) print(f"[INFO] Parameters: symbol={symbol}, elapsed={elapsed}, create_entries={create_entries}") stock_fetcher = StockDataFetcher(client_details) stock_fetcher.login() stock_fetcher.fetch_all_data(symbol, elapsed, create_entries) print("[INFO] Execution completed.") except Exception as e: print("[ERROR] Exception occurred:", e) raise finally: release_lock() print("[INFO] Lock released. Script finished.")