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): # Make request to PHP endpoint url = f"http://143.244.141.41/php/getSensexOpenPrice.php?date={date}" msg = '' try: response = requests.get(url) data = response.json() # Check for error flag from PHP if data.get('error', False): print(f"ERROR: {data.get('message', 'Unknown error')}") sys.exit(1) price = int(float(data['price'])) base = math.floor(price / 100) * 100 remainder = price - base rounded_price = base + (100 if remainder > 50 else 0) return_price = rounded_price return_expiry = data['expiry'] msg = data['message'] except Exception as e: print(f"Unexpected error: {e}") sys.exit(1) 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 = 1 self.sensex_step_value = 100 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 = 150 self.close_all_pending_time = "14:30:00" 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"Profit threshold: {self.profit_sumDiff_threshold}") 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'] * 20 ) / 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} 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} 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] # ===== EXIT CONDITIONS ===== if open_positions[col]['ce'] and sumDiff_current_value > self.profit_sumDiff_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 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] all_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] 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 # Wait after forced PE exit self.exit_counter_cutover = self.exit_counter_cutover + 1 ce_buy_price[col] = current_ce_price open_positions[col]['ce'] = True all_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] 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 # Wait after forced CE exit self.exit_counter_cutover = self.exit_counter_cutover + 1 pe_buy_price[col] = current_pe_price open_positions[col]['pe'] = True 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 = 'SENSEX' + str(instrument) self.update_pending_transactions(instrument, signal, closePrice) self.transactionsCounter = self.transactionsCounter + 1 if self.createEntries: url = "http://143.244.141.41/php/createMoneyHeistSensexEntry.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('SENSEX','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('Sensex','') 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 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.sensex_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"BFO:SENSEX{expiry_date}{strike}PE" ce_symbol = f"BFO:SENSEX{expiry_date}{strike}CE" # 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'] merged_df[f'sum_{strike}_zero'] = merged_df[f'sum_{strike}'] - merged_df[f'sum_{strike}'].iloc[0] if self.storeCsvFilesOnServer and self.storeStrikePriceCsvFilesOnServer: try: merged_df.to_csv(f'/var/www/html/clientDataSensex/SENSEX_{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'sum_{strike}_zero', 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_')] #hack to address inaccurate data given by firstock - initial few rows were zero #combined_df = combined_df[~(combined_df[diff_cols] == 0).all(axis=1)].reset_index(drop=True) # Calculate sum only for diff_ columns combined_df['sumDiff'] = combined_df[diff_cols].sum(axis=1) combined_df['sumDiff_ps'] = combined_df['sumDiff'] 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) sensexdf = self.process_symbol_data('BSE:SENSEX', interval, start_time, end_time) sensexdf = sensexdf.rename(columns={'intc': 'sensexPrice'}) 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 Sensex. 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['sumDiff'], combined_df[['sumDiff_ps']], combined_df['signCount'] ], axis=1) diff_df = pd.merge(diff_df, sensexdf[['time', 'sensexPrice']], on='time', how='left') if self.storeCsvFilesOnServer and self.storeDiffCsvFilesOnServer: try: diff_df.to_csv(f'/var/www/html/clientDataSensex/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("=" * 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']} : SENSEX{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/clientDataSensex/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)