import os import time import pandas as pd import pandas_ta as ta import logging from datetime import datetime, timedelta import pytz from thefirstock import thefirstock from typing import List import numpy as np import requests import sys import glob class StockDataFetcher: def __init__(self, client_details: List[str]): self.client_details = client_details self.user_id = client_details[0] self.order_cutoff_time = "14:29" # No new orders after this time self.forced_sell_time = "14:52" # All outsanding orders will be closed after this self.forced_sell_enabled = True # True - force close enabled; False - force close disabled self.ist = pytz.timezone('Asia/Kolkata') self.logger = self.setup_logger() print("\n\n\n\n") self.logger.info("**************************") def setup_logger(self) -> logging.Logger: logger = logging.getLogger(__name__) logger.setLevel(logging.INFO) 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 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: response = thefirstock.firstock_login(*self.client_details) if response.get("status") == "success": self.logger.info("Login successful") else: self.logger.error(f"Login failed: {response}") except Exception as e: self.logger.error(f"Login error: {e}") def fetch_time_price_series( self, exchange: str, trading_symbol: str, start_time: str, end_time: str, interval: str ) -> pd.DataFrame: print('\n') self.logger.info(f"----------------------------------") self.logger.info(f"Fetching data for {trading_symbol}") self.logger.info(f"----------------------------------") 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", [])) else: self.logger.error(f"Fetch failed: {response}") return pd.DataFrame() except Exception as e: self.logger.error(f"Error fetching series: {e}") return pd.DataFrame() def getStepValueAndShortSymbol(self, symbol): tmp_symbol=symbol if symbol == 'NSE:Nifty 50': step_value = 50 tmp_symbol='NIFTY' elif symbol == 'NSE:Nifty Bank': step_value = 100 tmp_symbol='BANKNIFTY' elif symbol == 'NSE:NIFTY MID SELECT': step_value = 25 tmp_symbol='MIDCPNIFTY' elif symbol == 'NSE:Nifty Fin Service': step_value = 50 tmp_symbol='FINNIFTY' elif symbol == 'BSE:SENSEX': step_value = 100 tmp_symbol='SENSEX' else: step_value = 50 return step_value, tmp_symbol def calculate_supertrend(self, df: pd.DataFrame, atr_period: int, multiplier: float) -> pd.DataFrame: df[['high', 'low', 'close']] = df[['inth', 'intl', 'intc']].apply(pd.to_numeric, errors='coerce') supertrend = df.ta.supertrend(high='high', low='low', close='close', length=atr_period, multiplier=multiplier) supertrend['ST_Signal'] = supertrend['SUPERTd_21_6.3'].apply(lambda x: 'Buy' if x == 1 else 'Sell') #supertrend['ST_Signal'] = supertrend['SUPERTd_6_13.0'].apply(lambda x: 'Buy' if x == 1 else 'Sell') return pd.concat([df, supertrend], axis=1) def get_undelying_strike_price(self, df, time, intc, step_value): strikePrice = int(intc) indexPrice = 0 for index, row in df.iterrows(): if time == row['time']: strikePrice = int(float(row['intc'])) indexPrice = strikePrice strikePrice = round(strikePrice / step_value) * step_value return strikePrice, indexPrice def process_symbol_data(self, symbol: str, interval: int, start_time: datetime, end_time: datetime): step_value, tmp_symbol = self.getStepValueAndShortSymbol(symbol) exchange, trading_symbol = symbol.split(":") # Fetch Data if exchange == 'NSE' and trading_symbol == 'Nifty 50': original_exchange = exchange original_trading_symbol = trading_symbol exchange = 'NFO' trading_symbol = 'NIFTY27MAR25F' 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), ) original_df = self.fetch_time_price_series( original_exchange, original_trading_symbol, start_time.strftime("%d/%m/%Y %H:%M:%S"), end_time.strftime("%d/%m/%Y %H:%M:%S"), str(interval), ) else: 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), ) if df.empty: self.logger.warning(f"No data for {trading_symbol}") return # Preprocess timestamps and ensure data integrity #df['time'] = pd.to_datetime(df['time']).dt.tz_localize(self.ist) df['time'] = pd.to_datetime(df['time'], errors='coerce', dayfirst=True) if not df['time'].dt.tz: df['time'] = df['time'].dt.tz_localize(self.ist) original_df['time'] = pd.to_datetime(original_df['time'], errors='coerce', dayfirst=True) if not original_df['time'].dt.tz: original_df['time'] = original_df['time'].dt.tz_localize(self.ist) df = df[~((df['time'].dt.hour == 9) & (df['time'].dt.minute < 15))] df = df[~(((df['time'].dt.hour == 15) & (df['time'].dt.minute >= 31)) | ((df['time'].dt.hour > 15) & (df['time'].dt.hour < 23)))] current_time = datetime.now(self.ist) df = df.sort_values(by='time', ascending=True) df['intc'] = pd.to_numeric(df['intc'], errors='coerce') df['intv'] = pd.to_numeric(df['intv'], errors='coerce') df['into'] = pd.to_numeric(df['into'], errors='coerce') df.loc[df['into'] >= df['intc'], 'intv'] *= -1 df['date'] = df['time'].dt.tz_convert(self.ist).dt.date df['dayVolume'] = df.groupby('date')['intv'].cumsum() # Define threshold levels for buy/sell signals volume_levels = [ 100000, 200000, 300000, 400000, 600000, 800000, 1000000, 1200000] volume_reset_levels = [-100000, 0, 100000, 200000, 400000, 600000, 800000, 1000000] negative_volume_levels = [ -100000, -200000, -300000, -400000, -600000, -800000, -1000000, -1200000] negative_volume_reset_levels = [100000, 0, -100000, -200000, -400000, -600000, -800000, -1000000] tmp_symbol='NIFTY' for date, group in df.groupby('date'): sell_actions = {level: None for level in volume_levels} # Track last sell action buy_actions = {level: None for level in negative_volume_levels} # Track last buy action print(f"\n\nDate: {date}") last_time, last_price, last_volume = None, None, None # Track last values for idx, row in group.iterrows(): day_volume = row['dayVolume'] price = row['intc'] time = row['time'] new_order_time = time.strftime("%H:%M") # Update last recorded values last_time, last_price, last_volume = time, price, day_volume # Check for Sell Action for i, level in enumerate(volume_levels): reset_level = volume_reset_levels[i] if sell_actions[level] is None and day_volume >= level: if new_order_time <= self.order_cutoff_time: st_instrument, indexPrice = self.get_undelying_strike_price(original_df, time, int(price), step_value) instrument = tmp_symbol + str(st_instrument) + 'PE' tickTime_str = time.strftime("%Y-%m-%d_%H:%M:%S") closePrice = indexPrice signal = 'BUY' orderType = level self.createEntry(tickTime_str, instrument, closePrice, signal, orderType) sell_actions[level] = (time, price, instrument, indexPrice) # Store action print(f"{time} : Buy {instrument} at index price {indexPrice}, future price {price} for Level {level} volume : {day_volume} ") else: #print('NEW PE Order stopped as it exceeded cut off time') pass if sell_actions[level] is not None and day_volume <= reset_level: sell_time, sell_price, sell_instrument, sell_index_price = sell_actions[level] st_instrument, indexPrice = self.get_undelying_strike_price(original_df, time, int(price), step_value) price_diff = int(sell_price - price) index_price_diff = int(sell_index_price - indexPrice) sell_actions[level] = None # Reset after reaching reset level tickTime_str = time.strftime("%Y-%m-%d_%H:%M:%S") closePrice = indexPrice signal = 'SELL' orderType = level self.createEntry(tickTime_str, sell_instrument, closePrice, signal, orderType) print(f"{time} : Sell {sell_instrument} at index price {indexPrice}, future price {price} for Level {level}, volume : {day_volume} Index Price Diff: {index_price_diff}, Price Diff: {price_diff} ") # Check for Buy Action for i, level in enumerate(negative_volume_levels): reset_level = negative_volume_reset_levels[i] if buy_actions[level] is None and day_volume <= level: if new_order_time <= self.order_cutoff_time: st_instrument, indexPrice = self.get_undelying_strike_price(original_df, time, int(price), step_value) instrument = tmp_symbol + str(st_instrument) + 'CE' tickTime_str = time.strftime("%Y-%m-%d_%H:%M:%S") closePrice = indexPrice signal = 'BUY' orderType = level self.createEntry(tickTime_str, instrument, closePrice, signal, orderType) buy_actions[level] = (time, price, instrument, indexPrice) # Store action print(f"{time} : Buy {instrument} at index price {indexPrice}, future price {price} for Level {level} volume : {day_volume} ") else: #print('NEW CE Order stopped as it exceeded cut off time') pass if buy_actions[level] is not None and day_volume >= reset_level: buy_time, buy_price, buy_instrument, buy_index_price = buy_actions[level] st_instrument, indexPrice = self.get_undelying_strike_price(original_df, time, int(price), step_value) price_diff = int(price - buy_price) index_price_diff = int(indexPrice - buy_index_price) buy_actions[level] = None # Reset after reaching reset level tickTime_str = time.strftime("%Y-%m-%d_%H:%M:%S") closePrice = indexPrice signal = 'SELL' orderType = level self.createEntry(tickTime_str, buy_instrument, closePrice, signal, orderType) print(f"{time} : Sell {buy_instrument} at index price {indexPrice}, future price {price} for Level {level}, volume : {day_volume} Index Price Diff: {index_price_diff}, Price Diff: {price_diff} ") if self.forced_sell_enabled and new_order_time >= self.forced_sell_time: st_instrument, indexPrice = self.get_undelying_strike_price(original_df, last_time, int(last_price), step_value) self.close_outstanding_actions(sell_actions, buy_actions, indexPrice, last_price, time) sell_actions = {level: None for level in volume_levels} buy_actions = {level: None for level in negative_volume_levels} # After exiting the loop for a date, print the last recorded values st_instrument, indexPrice = self.get_undelying_strike_price(original_df, last_time, int(last_price), step_value) self.log_volume(last_volume) print(f"End of {date} - Last Time: {last_time}, index Price: {indexPrice} future Price: {last_price}, Last Volume: {last_volume}") # Save DataFrame to CSV file_path = f"/var/www/html/clientData/{trading_symbol}_data.csv" df.to_csv(file_path, index=False) #print(df) def close_outstanding_actions(self, sell_actions, buy_actions, indexPrice, futurePrice, time): tickTime_str = time.strftime("%Y-%m-%d_%H:%M:%S") closePrice = indexPrice signal = 'SELL' for level, action in sell_actions.items(): if action is not None: bTime, price, instrument, index_price = action orderType = level self.createEntry(tickTime_str, instrument, closePrice, signal, orderType) print(f"{time} : Forced Sell {instrument} at index price {indexPrice}, future price {futurePrice} for Level {level}") for level, action in buy_actions.items(): if action is not None: bTtime, price, instrument, index_price = action orderType = level self.createEntry(tickTime_str, instrument, closePrice, signal, orderType) print(f"{time} : Forced Sell {instrument} at index price {indexPrice}, future price {futurePrice} for Level {level}") def log_volume(self, last_volume): self.base_dir = "/var/www/html/clientData" # Construct the file path file_path = f"{self.base_dir}/FinalVolume_{last_volume}.txt" # Delete old files matching the pattern old_files = glob.glob(f"{self.base_dir}/FinalVolume_*.txt") for old_file in old_files: try: os.remove(old_file) print(f"Deleted old file: {old_file}") except OSError as e: print(f"Error deleting file {old_file}: {e}") # Create an empty file with the new name try: with open(file_path, "w") as file: pass # No content is written to the file print(f"Created new file: {file_path}") except IOError as e: print(f"Error creating file {file_path}: {e}") def createEntry(self, tickTime_str, instrument, closePrice, signal, orderType): url = "http://143.244.141.41/php/createDeepSeekEntry.php" # Replace with your actual URL # Define the parameters params = { 'tickTime': str(tickTime_str), 'instrument': str(instrument), 'closePrice': str(closePrice), 'signal': str(signal), 'orderType': str(orderType) } # Perform the GET request #self.logger.info(f"Get Request with params: {params}") response = requests.get(url, params=params) if response.status_code != 200: self.logger.error(f"Request failed with status code: {response.status_code}") if response.text != '': self.logger.error(response.text + "\n") # Print error message if available else: #self.logger.info(f"Request success: {response.text}\n") pass def fetch_all_data(self): symbols = { "NSE:Nifty 50": 1, } for interval in [1]: for symbol, days in symbols.items(): time.sleep(1) start_time = datetime.now(self.ist).replace(hour=9, minute=15) end_time = datetime.now(self.ist).replace(hour=15, minute=30) #start_time = datetime.now(self.ist) - timedelta(days=days) #start_time = start_time.replace(hour=9, minute=15) self.process_symbol_data(symbol, interval, start_time, end_time) if __name__ == "__main__": client_details = ['GA0810', 'O9i8u7y6$$', '08101994', 'GA0810_API', 'ada14c5f73f182ca724b90c5dde1e72d'] stock_fetcher = StockDataFetcher(client_details) stock_fetcher.login() stock_fetcher.fetch_all_data()