import logging import sys import time from datetime import datetime, timedelta from typing import List import pytz import pandas as pd import requests import json import math from thefirstock import thefirstock def get_client_details(): try: url = 'http://143.244.141.41/php/getUserDetails.php' response = requests.get(url) response.raise_for_status() # Raise exception for HTTP errors data = response.json() if not data.get('success', False): raise ValueError("getUserDetails endpoint returned unsuccessful response") user_data = data['data'] client_details = [ user_data['field1'], # userId user_data['field2'], # password user_data['field3'], # TOTP user_data['field4'], # apiKey user_data['field5'] # vendorCode ] return client_details 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 def get_open_price_and_expiry_date(date): default_base_strike_price = 24000 default_expiry_date = '15MAY25' # Make request to PHP endpoint url = f"http://143.244.141.41/php/getNiftyOpenPrice.php?date={date}" msg = '' return_price = default_base_strike_price return_expiry = default_expiry_date try: response = requests.get(url) try: data = response.json() price = int(float(data['price'])) base = math.floor(price / 50) * 50 remainder = price - base rounded_price = base + (50 if remainder > 25 else 0) return_price = rounded_price return_expiry = data['expiry'] msg = data['message'] except Exception as e: msg = f"Error processing response: {e}" except Exception as e: msg = f"Error reaching server. {e}" return return_price, return_expiry, msg 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.dataset = {} self.createEntries = createEntries self.storeCsvFilesOnServer = True # set this to false except for debug self.storeStrikePriceCsvFilesOnServer = False self.storeCombinedCsvFilesOnServer = True self.storeDiffCsvFilesOnServer = True self.sleep_duration_bw_calls = 0.2 self.nifty_step_value = 50 self.debug_data = '' self.transactionsCounter = 0 self.lotCount = 1 self.pendingTransactions = {} self.pendingTransactionsBuyPrice = {} self.allowed_order_entries = [] self.exit_counter_profit = 0 self.exit_counter_cutover = 0 self.force_close_pending_orders = False self.max_profit_limit = 250 self.max_trailing_profit_limit = 50 self.stop_loss_limit = -150 self.combined_df = None # Option Data related Configurations self.tick_interval = 1 self.expiry_date = "" self.base_strike_price = 0 self.no_of_steps = 10 # defines the range self.order_allowed_steps = 2 # order range from open price self.profit_sumDiff_threshold = 450 self.profit_sumDiff_trailing_threshold = 50 self.close_all_pending_time = "15:20:00" self.g_Allow_CE = True self.g_Allow_PE = True def print_configuration(self): self.addLogDataInfo("-" * 40) self.addLogDataInfo("Current Configuration:") self.addLogDataInfo("-" * 40) self.addLogDataInfo(f"Tick Interval: {self.tick_interval}") self.addLogDataInfo(f"Expiry Date: {self.expiry_date}") self.addLogDataInfo(f"Base Strike Price: {self.base_strike_price}") self.addLogDataInfo(f"Number of Steps (total): {self.no_of_steps}") self.addLogDataInfo(f"Number of Steps (order allowed): {self.order_allowed_steps}") self.addLogDataInfo(f"Max profit Limit: {self.max_profit_limit:.1f}") self.addLogDataInfo(f"Max StopLoss Limit: {self.stop_loss_limit:.1f}") self.addLogDataInfo(f"Close all pending transactions Time: {self.close_all_pending_time}") self.addLogDataInfo("-" * 40) 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 addLogDataDebug(self, text): self.logger.debug(f'{text}') self.debug_data = self.debug_data + text + '\n' def addLogDataInfo(self, text): self.logger.info(f'{text}') self.debug_data = self.debug_data + text + '\n' 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 login(self): try: self.addLogDataInfo(f"Attempting login for {self.client_details[0]}") response = thefirstock.firstock_login(*self.client_details) if response.get("status") == "success": self.addLogDataInfo("Login successful") else: self.addLogDataInfo(f"Login failed: {response}") sys.exit() except Exception as e: self.addLogDataInfo(f"Login error: {e}") sys.exit() def fetch_time_price_series( self, exchange: str, trading_symbol: str, start_time: str, end_time: str, interval: str ) -> pd.DataFrame: self.addLogDataDebug(f"Fetching data for {trading_symbol} from {start_time} to {end_time}. interval:{interval}") try: response = thefirstock.firstock_TimePriceSeries( userId=self.user_id, exchange=exchange, tradingSymbol=trading_symbol, startTime=start_time, endTime=end_time, interval=interval, ) 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 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("%d/%m/%Y %H:%M:%S"), end_time.strftime("%d/%m/%Y %H:%M:%S"), str(interval), ) numeric_cols = ['intc', 'intvwap', 'oi', 'intoi'] for col in numeric_cols: df[col] = pd.to_numeric(df[col], errors='coerce') df = df.sort_values(by='time', ascending=True) df['cost'] = (df['intc'] * df['oi']) / 10000000 df['cost'] = df['cost'].round(2) return df def getPerformanceMetrics(self): self.addLogDataInfo("-" * 40) self.addLogDataInfo("Performance Metrics:") self.addLogDataInfo("-" * 40) self.addLogDataInfo(f"Cutover flip count: {self.exit_counter_cutover}") self.addLogDataInfo(f"Profit count: {self.exit_counter_profit}") self.addLogDataInfo("-" * 40) def detect_transitions(self, combined_df): self.combined_df = combined_df all_events = [] diff_columns = [col for col in combined_df.columns if col.startswith('diff_')] open_positions = {col: {'ce': False, 'pe': False} for col in diff_columns} reverse_open_positions = {col: {'ce': False, 'pe': False} for col in diff_columns} ce_buy_price = {col: 0.0 for col in diff_columns} pe_buy_price = {col: 0.0 for col in diff_columns} reverse_ce_buy_price = {col: 0.0 for col in diff_columns} reverse_pe_buy_price = {col: 0.0 for col in diff_columns} wait_for_zero_cross = {col: False for col in diff_columns} for idx in range(1, len(combined_df)): current_time = combined_df['time'].iloc[idx] sumDiff_current_value = combined_df['sumDiff_ps'].iloc[idx] sumDiff_prev_value = combined_df['sumDiff_ps'].iloc[idx - 1] current_events = [] for col in diff_columns: if col not in self.allowed_order_entries: continue ce_allowed = True pe_allowed = True ce_price_col = col.replace('diff', 'intc') + '_ce' pe_price_col = col.replace('diff', 'intc') + '_pe' current_strike = int(pe_price_col.split("_")[1]) if current_strike < self.base_strike_price: pe_allowed = False #print (f'pe not allowed {current_strike}') elif current_strike > self.base_strike_price: ce_allowed = False #print (f'ce not allowed {current_strike}') if not self.g_Allow_CE: ce_allowed = False if not self.g_Allow_PE: pe_allowed = False current_ce_price = combined_df[ce_price_col].iloc[idx] current_pe_price = combined_df[pe_price_col].iloc[idx] #====== REVERSE #close ce reverse entry if reverse_open_positions[col]['ce'] and sumDiff_current_value > -600: pnl = current_ce_price - reverse_ce_buy_price[col] current_events.append({ 'time': current_time, 'column': col, 'type': 'ce', 'event': 'SELL', 'closePrice': current_ce_price, 'pnl': pnl, 'reason': '-- REVERSE CE EXIT PROFIT' }) reverse_open_positions[col]['ce'] = False self.exit_counter_profit = self.exit_counter_profit + 1 #close pe reverse entry elif reverse_open_positions[col]['pe'] and sumDiff_current_value < 600: reverse_open_positions[col]['pe'] = False pnl = current_pe_price - reverse_pe_buy_price[col] current_events.append({ 'time': current_time, 'column': col, 'type': 'pe', 'event': 'SELL', 'closePrice': current_pe_price, 'pnl': pnl, 'reason': '-- REVERSE PE EXIT PROFIT' }) reverse_open_positions[col]['pe'] = False self.exit_counter_profit = self.exit_counter_profit + 1 #open ce reverse entry if ce_allowed and not reverse_open_positions[col]['ce'] and sumDiff_current_value < -950: reverse_ce_buy_price[col] = current_ce_price reverse_open_positions[col]['ce'] = True current_events.append({ 'time': current_time, 'column': col, 'type': 'ce', 'event': 'BUY', 'closePrice': current_ce_price, 'pnl': 0.0, 'reason': '++ REVERSE CE ENTRY' }) #open pe reverse entry elif pe_allowed and not reverse_open_positions[col]['pe'] and sumDiff_current_value > 950: reverse_pe_buy_price[col] = current_pe_price reverse_open_positions[col]['pe'] = True current_events.append({ 'time': current_time, 'column': col, 'type': 'pe', 'event': 'BUY', 'closePrice': current_pe_price, 'pnl': 0.0, 'reason': '++ REVERSE PE ENTRY' }) # ===== EXIT CONDITIONS ===== if open_positions[col]['ce'] and sumDiff_current_value > self.profit_sumDiff_threshold: pnl = current_ce_price - ce_buy_price[col] current_events.append({ 'time': current_time, 'column': col, 'type': 'ce', 'event': 'SELL', 'closePrice': current_ce_price, 'pnl': pnl, 'reason': '-- CE EXIT PROFIT' }) open_positions[col]['ce'] = False wait_for_zero_cross[col] = True # Set flag after exit self.exit_counter_profit = self.exit_counter_profit + 1 elif open_positions[col]['pe'] and sumDiff_current_value < -self.profit_sumDiff_threshold: pnl = current_pe_price - pe_buy_price[col] current_events.append({ 'time': current_time, 'column': col, 'type': 'pe', 'event': 'SELL', 'closePrice': current_pe_price, 'pnl': pnl, 'reason': '-- PE EXIT PROFIT' }) open_positions[col]['pe'] = False wait_for_zero_cross[col] = True # Set flag after exit self.exit_counter_profit = self.exit_counter_profit + 1 # ===== WAIT FOR ZERO CROSS ===== if wait_for_zero_cross[col]: # Check for zero crossing if (sumDiff_prev_value > 0 and sumDiff_current_value <= 0) or \ (sumDiff_prev_value < 0 and sumDiff_current_value >= 0): wait_for_zero_cross[col] = False # Reset flag after zero cross else: continue # Still waiting, skip entry # ===== ENTRY CONDITIONS ===== if sumDiff_current_value > 0 and not open_positions[col]['ce']: if open_positions[col]['pe']: pnl = current_pe_price - pe_buy_price[col] current_events.append({ 'time': current_time, 'column': col, 'type': 'pe', 'event': 'SELL', 'closePrice': current_pe_price, 'pnl': pnl, 'reason': '-- PE EXIT (CE ENTRY DUE)' }) open_positions[col]['pe'] = False wait_for_zero_cross[col] = True # Wait after forced PE exit self.exit_counter_cutover = self.exit_counter_cutover + 1 if ce_allowed: ce_buy_price[col] = current_ce_price open_positions[col]['ce'] = True current_events.append({ 'time': current_time, 'column': col, 'type': 'ce', 'event': 'BUY', 'closePrice': ce_buy_price[col], 'pnl': 0.0, 'reason': '++ CE ENTRY' }) elif sumDiff_current_value < 0 and not open_positions[col]['pe']: if open_positions[col]['ce']: pnl = current_ce_price - ce_buy_price[col] current_events.append({ 'time': current_time, 'column': col, 'type': 'ce', 'event': 'SELL', 'closePrice': current_ce_price, 'pnl': pnl, 'reason': '-- CE EXIT (PE ENTRY DUE)' }) open_positions[col]['ce'] = False wait_for_zero_cross[col] = True # Wait after forced CE exit self.exit_counter_cutover = self.exit_counter_cutover + 1 if pe_allowed: pe_buy_price[col] = current_pe_price open_positions[col]['pe'] = True current_events.append({ 'time': current_time, 'column': col, 'type': 'pe', 'event': 'BUY', 'closePrice': pe_buy_price[col], 'pnl': 0.0, 'reason': '++ PE ENTRY' }) #add the new code here sell_events = [e for e in current_events if e['event'] == 'SELL'] buy_events = [e for e in current_events if e['event'] == 'BUY'] all_events.extend(sell_events + buy_events) return sorted(all_events, key=lambda x: x['time']) def detect_transitions_trail(self, combined_df): self.combined_df = combined_df all_events = [] diff_columns = [col for col in combined_df.columns if col.startswith('diff_')] open_positions = {col: {'ce': False, 'pe': False} for col in diff_columns} ce_buy_price = {col: 0.0 for col in diff_columns} pe_buy_price = {col: 0.0 for col in diff_columns} wait_for_zero_cross = {col: False for col in diff_columns} max_favorable_ce = {col: float('-inf') for col in diff_columns} max_favorable_pe = {col: float('inf') for col in diff_columns} trailing_ce_armed = {col: False for col in diff_columns} trailing_pe_armed = {col: False for col in diff_columns} for idx in range(1, len(combined_df)): current_time = combined_df['time'].iloc[idx] sumDiff_current_value = combined_df['sumDiff_ps'].iloc[idx] sumDiff_prev_value = combined_df['sumDiff_ps'].iloc[idx - 1] for col in diff_columns: if col not in self.allowed_order_entries: continue ce_price_col = col.replace('diff', 'intc') + '_ce' pe_price_col = col.replace('diff', 'intc') + '_pe' current_ce_price = combined_df[ce_price_col].iloc[idx] current_pe_price = combined_df[pe_price_col].iloc[idx] # ===== CE TRAILING EXIT ===== if open_positions[col]['ce']: if sumDiff_current_value > self.profit_sumDiff_threshold: max_favorable_ce[col] = max(max_favorable_ce[col], sumDiff_current_value) trailing_ce_armed[col] = True self.addLogDataInfo(f"New profit threshold for {col} CE : {max_favorable_ce[col]}") if trailing_ce_armed[col] and (max_favorable_ce[col] - sumDiff_current_value) >= self.profit_sumDiff_trailing_threshold: pnl = current_ce_price - ce_buy_price[col] all_events.append({ 'time': current_time, 'column': col, 'type': 'ce', 'event': 'SELL', 'closePrice': current_ce_price, 'pnl': pnl, 'reason': '-- CE EXIT TRAIL' }) open_positions[col]['ce'] = False wait_for_zero_cross[col] = True self.exit_counter_profit += 1 max_favorable_ce[col] = float('-inf') trailing_ce_armed[col] = False continue # ===== PE TRAILING EXIT ===== elif open_positions[col]['pe']: if sumDiff_current_value < -self.profit_sumDiff_threshold: max_favorable_pe[col] = min(max_favorable_pe[col], sumDiff_current_value) trailing_pe_armed[col] = True self.addLogDataInfo(f"New profit threshold for {col} PE : {max_favorable_pe[col]}") if trailing_pe_armed[col] and (sumDiff_current_value - max_favorable_pe[col]) >= self.profit_sumDiff_trailing_threshold: pnl = current_pe_price - pe_buy_price[col] all_events.append({ 'time': current_time, 'column': col, 'type': 'pe', 'event': 'SELL', 'closePrice': current_pe_price, 'pnl': pnl, 'reason': '-- PE EXIT TRAIL' }) open_positions[col]['pe'] = False wait_for_zero_cross[col] = True self.exit_counter_profit += 1 max_favorable_pe[col] = float('inf') trailing_pe_armed[col] = False continue # ===== WAIT FOR ZERO CROSS ===== if wait_for_zero_cross[col]: if (sumDiff_prev_value > 0 and sumDiff_current_value <= 0) or \ (sumDiff_prev_value < 0 and sumDiff_current_value >= 0): wait_for_zero_cross[col] = False else: continue # still waiting, skip entry # ===== CE ENTRY ===== if sumDiff_current_value > 0 and not open_positions[col]['ce']: if open_positions[col]['pe']: pnl = current_pe_price - pe_buy_price[col] all_events.append({ 'time': current_time, 'column': col, 'type': 'pe', 'event': 'SELL', 'closePrice': current_pe_price, 'pnl': pnl, 'reason': '-- PE EXIT (CE ENTRY DUE)' }) open_positions[col]['pe'] = False wait_for_zero_cross[col] = True self.exit_counter_cutover += 1 ce_buy_price[col] = current_ce_price open_positions[col]['ce'] = True max_favorable_ce[col] = sumDiff_current_value trailing_ce_armed[col] = False all_events.append({ 'time': current_time, 'column': col, 'type': 'ce', 'event': 'BUY', 'closePrice': ce_buy_price[col], 'pnl': 0.0, 'reason': '++ CE ENTRY' }) # ===== PE ENTRY ===== elif sumDiff_current_value < 0 and not open_positions[col]['pe']: if open_positions[col]['ce']: pnl = current_ce_price - ce_buy_price[col] all_events.append({ 'time': current_time, 'column': col, 'type': 'ce', 'event': 'SELL', 'closePrice': current_ce_price, 'pnl': pnl, 'reason': '-- CE EXIT (PE ENTRY DUE)' }) open_positions[col]['ce'] = False wait_for_zero_cross[col] = True self.exit_counter_cutover += 1 pe_buy_price[col] = current_pe_price open_positions[col]['pe'] = True max_favorable_pe[col] = sumDiff_current_value trailing_pe_armed[col] = False all_events.append({ 'time': current_time, 'column': col, 'type': 'pe', 'event': 'BUY', 'closePrice': pe_buy_price[col], 'pnl': 0.0, 'reason': '++ PE ENTRY' }) return sorted(all_events, key=lambda x: x['time']) def createEntry(self, tickTimeStr, instrument, closePrice, signal, lotCount): instrument = 'NIFTY' + str(instrument) self.update_pending_transactions(instrument, signal, closePrice) self.transactionsCounter = self.transactionsCounter + 1 if self.createEntries: url = "http://143.244.141.41/php/createMoneyHeistEntry.php" tickTimeStr = tickTimeStr + ':' + str(self.transactionsCounter).zfill(3) # Define the parameters params = { 'tickTime': str(tickTimeStr), 'instrument': str(instrument), 'closePrice': str(closePrice), 'signal': str(signal), 'orderType': str(lotCount) } response = requests.get(url, params=params) if response.status_code != 200: self.addLogDataDebug(f"Request failed with status code: {response.status_code}") if response.text != '': self.addLogDataDebug(response.text + "\n") # Print error message if available def update_pending_transactions(self, instrument, action, closePrice): # Get current count or default to 0 current_count = self.pendingTransactions.get(instrument, 0) # Update count based on action if action == 'BUY': self.pendingTransactions[instrument] = current_count + 1 self.pendingTransactionsBuyPrice[instrument] = closePrice elif action == 'SELL': self.pendingTransactions[instrument] = current_count - 1 def print_pending_transactions(self, totalPnL, idx): positions_to_be_closed = 0 # Filter non-zero positions and sort by count (highest to lowest) active_positions = {k: v for k, v in sorted( self.pendingTransactions.items(), key=lambda item: abs(item[1]), reverse=True) if v != 0} # Create formatted output if not active_positions: self.addLogDataDebug("All positions are squared off (zero pending transactions)") return positions_to_be_closed, totalPnL self.addLogDataDebug(f" Pending Transactions ".center(40, "+")) self.addLogDataDebug("-" * 60) self.addLogDataDebug(f"{'Instrument':<12} | {'Qty':>3} contracts | {'Buy':>6} | {'LTP':>6} | {'PnL':>7}") self.addLogDataDebug("-" * 60) for instrument, count in active_positions.items(): price_col = instrument.replace('NIFTY','intc_') price_col = price_col.replace('CE', '_ce') price_col = price_col.replace('PE', '_pe') buy_price = self.pendingTransactionsBuyPrice[instrument] #current_price = self.combined_df[price_col].iloc[-1] current_price = self.combined_df.iloc[idx][price_col] PnL = current_price - buy_price totalPnL = totalPnL + PnL self.addLogDataDebug(f"{instrument:<12} | {abs(count):>3} contracts | {buy_price:>6.2f} | {current_price:>6.2f} | {PnL:>7.2f}") positions_to_be_closed = positions_to_be_closed + 1 self.addLogDataDebug("-" * 60) return positions_to_be_closed, totalPnL def close_all_pending_transactions(self, timeStr): pendingTransactions = self.pendingTransactions # Filter non-zero positions and sort by count (highest to lowest) active_positions = {k: v for k, v in sorted( pendingTransactions.items(), key=lambda item: abs(item[1]), reverse=True) if v != 0} self.addLogDataDebug(f" Closing Pending Transactions ".center(40, "-")) for instrument, count in active_positions.items(): signal = "SELL" if count > 0 else "BUY" self.addLogDataDebug(f"closing position for {instrument}") instrument = instrument.replace('NIFTY','') self.createEntry(timeStr, instrument, 'xxx.xx', signal, self.lotCount) self.addLogDataDebug("-" * 40) def fetch_all_data(self, elapsed: int): elapsed = int(elapsed) if elapsed != 0: self.storeCsvFilesOnServer = False self.storeCsvFilesOnServer = True start_time = datetime.now(self.ist).replace( hour=9, minute=15, second=0 ) - timedelta(days=elapsed) end_time = datetime.now(self.ist).replace( hour=15, minute=31, second=0 ) - timedelta(days=elapsed) start_time_date = str(start_time).split(' ')[0] base_price, expiry , msg = get_open_price_and_expiry_date(start_time_date) self.base_strike_price = base_price self.expiry_date = expiry self.addLogDataInfo(f"Get Open price returned {base_price}, {expiry} for {start_time_date}. msg = {msg}") stock_fetcher.print_configuration() interval = self.tick_interval expiry_date = self.expiry_date base_strike_price = self.base_strike_price step_value = self.nifty_step_value no_of_steps = self.no_of_steps range_delta = step_value * no_of_steps order_allowed_range_delta = step_value * self.order_allowed_steps for strike in range(base_strike_price - order_allowed_range_delta, base_strike_price + order_allowed_range_delta + 1, step_value): self.allowed_order_entries.append(f'diff_{strike}') final_dfs = [] for strike in range(base_strike_price - range_delta, base_strike_price + range_delta + 1, step_value): pe_symbol = f"NFO:NIFTY{expiry_date}P{strike}" ce_symbol = f"NFO:NIFTY{expiry_date}C{strike}" # Retry fetching PE data until successful no_of_retries_pending = 3 while True: try: time.sleep(self.sleep_duration_bw_calls) pe_df = self.process_symbol_data(pe_symbol, interval, start_time, end_time) break # Exit loop on success except Exception as e: no_of_retries_pending = no_of_retries_pending - 1 self.addLogDataInfo(f"Error fetching PE data for strikePrice: {strike}. Retrying... Exception: {e}") time.sleep(2*self.sleep_duration_bw_calls) if no_of_retries_pending == 0: sys.exit() # Retry fetching CE data until successful no_of_retries_pending = 3 while True: try: time.sleep(self.sleep_duration_bw_calls) ce_df = self.process_symbol_data(ce_symbol, interval, start_time, end_time) break # Exit loop on success except Exception as e: no_of_retries_pending = no_of_retries_pending - 1 self.addLogDataInfo(f"Error fetching CE data for strikePrice: {strike}. Retrying... Exception: {e}") time.sleep(2*self.sleep_duration_bw_calls) if no_of_retries_pending == 0: sys.exit() # Merge on time column merged_df = pd.merge(pe_df, ce_df, on='time', suffixes=(f'_{strike}_pe', f'_{strike}_ce')) # Calculate CE - PE difference merged_df[f'diff_{strike}'] = merged_df[f'cost_{strike}_ce'] - merged_df[f'cost_{strike}_pe'] merged_df[f'sum_{strike}'] = merged_df[f'cost_{strike}_ce'] + merged_df[f'cost_{strike}_pe'] if self.storeCsvFilesOnServer and self.storeStrikePriceCsvFilesOnServer: try: merged_df.to_csv(f'/var/www/html/clientData/NIFTY_{strike}_{start_time_date}.csv', index=False) except: pass # Keep only time and difference columns #result_df = merged_df[['time', f'diff_{strike}', f'cost_{strike}_pe', f'cost_{strike}_ce']] result_df = merged_df[['time', f'cost_{strike}_ce', f'cost_{strike}_pe', f'diff_{strike}', f'sum_{strike}', f'intc_{strike}_pe', f'intc_{strike}_ce']] final_dfs.append(result_df) if final_dfs: combined_df = pd.concat(final_dfs, axis=1) # Remove duplicate time columns combined_df = combined_df.loc[:,~combined_df.columns.duplicated()] # Get list of columns that start with 'diff_' diff_cols = [col for col in combined_df.columns if col.startswith('diff_')] sum_cols = [col for col in combined_df.columns if col.startswith('sum_')] # Calculate sum only for diff_ columns combined_df['sumDiff'] = combined_df[diff_cols].sum(axis=1) first_row_sum = combined_df['sumDiff'].iloc[0] combined_df['sumDiff_ps'] = combined_df['sumDiff'] - first_row_sum def count_signs(row): positive_count = sum(row[col] > 0 for col in diff_cols) negative_count = sum(row[col] < 0 for col in diff_cols) return positive_count - negative_count combined_df['signCount'] = combined_df.apply(count_signs, axis=1) if combined_df.iloc[-1].isna().any(): combined_df = combined_df.iloc[:-1] no_of_retries_pending = 3 while True: try: time.sleep(self.sleep_duration_bw_calls) niftydf = self.process_symbol_data('NSE:Nifty 50', interval, start_time, end_time) niftydf = niftydf.rename(columns={'intc': 'niftyPrice'}) break # Exit loop on success except Exception as e: no_of_retries_pending = no_of_retries_pending - 1 self.addLogDataInfo(f"Error fetching data for Nifty 50. Retrying... Exception: {e}") time.sleep(2*self.sleep_duration_bw_calls) if no_of_retries_pending == 0: sys.exit() diff_df = pd.concat([ combined_df[['time']], combined_df[[col for col in combined_df.columns if col.startswith('diff_')]], combined_df[[col for col in combined_df.columns if col.startswith('sum_')]], combined_df['sumDiff'], combined_df[['sumDiff_ps']], combined_df['signCount'] ], axis=1) diff_df = pd.merge(diff_df, niftydf[['time', 'niftyPrice']], on='time', how='left') if self.storeCsvFilesOnServer and self.storeDiffCsvFilesOnServer: try: diff_df.to_csv(f'/var/www/html/clientData/diff_{start_time_date}.csv', index=False) except: diff_df.to_csv(f'diff_{start_time_date}.csv', index=False) pass all_events = self.detect_transitions(combined_df) self.addLogDataDebug("=" * 40) self.addLogDataDebug('Start the Music :') self.addLogDataDebug(f"Initial amount in play: {int(first_row_sum)}") self.addLogDataDebug(f"Money profit threshold: {self.profit_sumDiff_threshold}") self.addLogDataDebug("=" * 40) totalPnL = 0.0 max_trailing_profit = None last_time = '' for event in all_events: if not self.force_close_pending_orders: strike_price = event['column'].replace('diff_','') pnl = round(event['pnl'], 2) totalPnL = round(totalPnL + pnl, 2) self.addLogDataInfo(f"{event['time']} : NIFTY{strike_price}{event['type'].upper()} {event['event']:<4} at {event['closePrice']:>7.2f} profit : {pnl:>7.2f} TotalPnL:{totalPnL:>7.2f} reason: {event['reason']}") timeStr = str(event['time']) instrument = strike_price + event['type'].upper() self.createEntry(timeStr, instrument, event['closePrice'], event['event'], self.lotCount) if max_trailing_profit is None: if totalPnL > self.max_profit_limit: max_trailing_profit = totalPnL self.addLogDataInfo(f"Max Profit Limit reached. Starting trailing logic. Peak profit: {max_trailing_profit:.1f}") else: if totalPnL > max_trailing_profit: max_trailing_profit = totalPnL self.addLogDataInfo(f"New peak profit updated to: {max_trailing_profit:.1f}") elif totalPnL < max_trailing_profit - self.max_trailing_profit_limit: self.addLogDataInfo(f"Profit dropped > {self.max_trailing_profit_limit} points from peak. Peak: {max_trailing_profit:.1f}") self.addLogDataInfo(f"Current totalPnL: {totalPnL:.1f}. Triggering force close.") self.force_close_pending_orders = True last_time = timeStr if totalPnL < self.stop_loss_limit: self.addLogDataInfo(f"Current totalPnL: {totalPnL:.1f}. stop loss limit reached . Triggering force close.") self.force_close_pending_orders = True last_time = timeStr if not self.force_close_pending_orders: last_time = str(combined_df['time'].iloc[-1]) last_time_without_date = last_time.split(' ')[1] self.getPerformanceMetrics() self.addLogDataDebug("checking for the pending transactions ...") index = combined_df[combined_df['time'].astype(str) == last_time].index iloc_index = combined_df.index.get_indexer(index)[0] positions_to_be_closed, totalPnL = self.print_pending_transactions(totalPnL, iloc_index) if not self.force_close_pending_orders: if totalPnL < self.stop_loss_limit: self.addLogDataInfo(f"Current totalPnL(after pending transactions): {totalPnL:.1f}. stop loss limit reached . Triggering force close.") self.force_close_pending_orders = True if max_trailing_profit is not None: if totalPnL < max_trailing_profit - self.max_trailing_profit_limit: self.addLogDataInfo(f"Profit dropped (after pending transactions) > {self.max_trailing_profit_limit} points from peak. Peak: {max_trailing_profit:.1f}") self.force_close_pending_orders = True self.addLogDataInfo("=" * 40) self.addLogDataInfo(f'Total PnL : {totalPnL:.1f}') self.addLogDataInfo(f'Number of Transactions : {self.transactionsCounter}') self.addLogDataInfo("=" * 40) self.addLogDataDebug(f"Last tick time - {last_time_without_date}") if last_time_without_date == self.close_all_pending_time and positions_to_be_closed > 0 or self.force_close_pending_orders: if not self.force_close_pending_orders: self.addLogDataDebug("*** End time reached. ") self.close_all_pending_transactions(last_time) if self.storeCsvFilesOnServer and self.storeCombinedCsvFilesOnServer: try: combined_df.to_csv(f'/var/www/html/clientData/combined_{start_time_date}.csv', index=False) except: pass self.addLogDataInfo("\n\n") if __name__ == "__main__": client_details = get_client_details() elapsed = float(sys.argv[1]) if len(sys.argv) > 1 else 0 # optional 1st argument - for backtesting createEntries = len(sys.argv) > 2 # optional 2nd argument stock_fetcher = StockDataFetcher(client_details, createEntries) stock_fetcher.login() stock_fetcher.fetch_all_data(elapsed)