import os import time import pandas as pd import pandas_ta as ta import logging from datetime import datetime, timedelta import pytz from typing import List, Optional from thefirstock import thefirstock import sys import requests import pickle 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() 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: self.logger.info(f"Fetching data for {trading_symbol}") 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 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) return pd.concat([df, supertrend], axis=1) def save_state(self, file_path, data): with open(file_path, 'wb') as f: pickle.dump(data, f) # Function to load data from file def load_state(self, file_path): if os.path.exists(file_path): with open(file_path, 'rb') as f: return pickle.load(f) return None # Return None if file doesn't exist 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) ) if df.empty: self.logger.warning(f"No data for {trading_symbol}") return 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) df = df[~((df['time'].dt.hour == 9) & (df['time'].dt.minute < 15))] df = df.iloc[:-1] # Remove the last row # Calculate elapsed time in minutes current_time = datetime.now(self.ist) df['elapsedTime'] = (current_time - df['time']).dt.total_seconds() // 60 df = df.sort_values(by='time', ascending=True) # Calculate DEMA, MA10, and Supertrend df['intc'] = pd.to_numeric(df['intc'], errors='coerce') df['DEMA'] = df.ta.dema(close='intc', length=100) df['MA10'] = df.ta.sma(close='intc', length=10) df = self.calculate_supertrend(df, 10, 6.3) df = df.dropna(subset=['DEMA', 'MA10']) df['DEMA'] = pd.to_numeric(df['DEMA'], errors='coerce') df['MA10'] = pd.to_numeric(df['MA10'], errors='coerce') df['SUPERTd_10_6.3'] = pd.to_numeric(df['SUPERTd_10_6.3'], errors='coerce') # Define status based on DEMA and MA10 df['DEMAStatus'] = df.apply(lambda row: 'uptrend' if row['DEMA'] < row['intc'] else 'downtrend', axis=1) df['MA10Status'] = df.apply(lambda row: 'uptrend' if row['MA10'] < row['intc'] else 'downtrend', axis=1) df['superTrendStatus'] = df.apply(lambda row: 'uptrend' if row['SUPERTd_10_6.3'] == 1 else 'downtrend', axis=1) df['MA10_DEMA'] = df.apply(lambda row: 'high' if row['MA10'] > row['DEMA'] else 'low', axis=1) if symbol == 'NSE:Nifty 50': step_value = 50 elif symbol == 'NSE:Nifty Bank': step_value = 100 elif symbol == 'NSE:NIFTY MID SELECT': step_value = 25 elif symbol == 'NSE:Nifty Fin Service': step_value = 50 elif symbol == 'BSE:SENSEX': step_value = 100 else: step_value = 50 df['instrument'] = df['close'].apply(lambda x: str(round(x / step_value) * step_value)) ''' file_path = "output.csv" df.to_csv(file_path, index=False) ''' tmp_symbol=symbol if symbol == 'NSE:Nifty 50': tmp_symbol='NIFTY' elif symbol == 'NSE:Nifty Bank': tmp_symbol='BANKNIFTY' elif symbol == 'NSE:NIFTY MID SELECT': tmp_symbol='MIDCPNIFTY' elif symbol == 'NSE:Nifty Fin Service': tmp_symbol='FINNIFTY' elif symbol == 'BSE:SENSEX': tmp_symbol='SENSEX' file_path=f"/var/www/html/ACTIONS/{tmp_symbol}.txt" state = self.load_state(file_path) if state: entry_flag = state.get('entry_flag', False) prev_entry = state.get('prev_entry', None) prev_instrument = state.get('prev_instrument', None) else: entry_flag = False prev_entry = None prev_instrument = None df_new = pd.DataFrame(columns=df.columns) for i in range(1, len(df)): # Access rows using time or condition to ensure proper alignment current_row = df.iloc[i].copy() # Check for 'entry' conditions (only if all indicators align) if not entry_flag: # Check if all indicators are uptrend if (current_row['DEMAStatus'] == 'uptrend' and current_row['MA10Status'] == 'uptrend' and current_row['superTrendStatus'] == 'uptrend') and current_row['MA10_DEMA'] == 'high' : current_row['entryStatus'] = 'entry' new_instrument = current_row['instrument'] new_instrument += 'CE' current_row['instrument'] = new_instrument prev_entry = 'uptrend' prev_instrument = new_instrument entry_flag = True df_new = pd.concat([df_new, current_row.to_frame().T], ignore_index=True) # Check if all indicators are downtrend elif (current_row['DEMAStatus'] == 'downtrend' and current_row['MA10Status'] == 'downtrend' and current_row['superTrendStatus'] == 'downtrend') and current_row['MA10_DEMA'] == 'low' : current_row['entryStatus'] = 'entry' new_instrument = current_row['instrument'] new_instrument += 'PE' current_row['instrument'] = new_instrument prev_entry = 'downtrend' prev_instrument = new_instrument entry_flag = True df_new = pd.concat([df_new, current_row.to_frame().T], ignore_index=True) elif entry_flag: if prev_entry == 'uptrend' and current_row['MA10_DEMA'] == 'low': current_row['entryStatus'] = 'exit' current_row['instrument'] = prev_instrument entry_flag = False df_new = pd.concat([df_new, current_row.to_frame().T], ignore_index=True) elif prev_entry == 'downtrend' and current_row['MA10_DEMA'] == 'high': current_row['entryStatus'] = 'exit' current_row['instrument'] = prev_instrument entry_flag = False df_new = pd.concat([df_new, current_row.to_frame().T], ignore_index=True) columns_to_keep = ['time', 'intc', 'MA10', 'DEMA', 'MA10Status', 'DEMAStatus', 'superTrendStatus', 'MA10_DEMA', 'entryStatus' , 'instrument', 'elapsedTime'] # Create final_df with only the desired columns from df_new final_df = df_new[columns_to_keep] # Display final_df #print(final_df) self.save_state(file_path, { 'entry_flag': entry_flag, 'prev_entry': prev_entry, 'prev_instrument': prev_instrument }) for index, row in final_df.iterrows(): elapsed_time = int(row['elapsedTime']) #if elapsed_time > 3 and elapsed_time < 100: if elapsed_time < 600: url = "http://143.244.141.41/php/addAction.php" # Replace with your actual URL tickTime=row['time'] tickTime_str = tickTime.strftime("%Y-%m-%d_%H:%M:%S") # Define the parameters params = { 'tickTime': tickTime_str, 'symbol': tmp_symbol, 'instrument': row['instrument'], 'closePrice': str(row['intc']), 'entryStatus': row['entryStatus'] } # Perform the GET request if row['entryStatus'] == 'entry': self.entryDone = True if self.entryDone: self.logger.info(f"Get Request with params: {params}") response = requests.get(url, params=params) # Check the response status and print the result if response.status_code != 200: self.logger.error(f"Request failed with status code: {response.status_code}") self.logger.error(response.text) # Print error message if available else: self.logger.info(f"Request success: {response.text}") else: self.logger.info(f"Stopped Request with params: {params}") def setEntryDone(self): self.entryDone = False def fetch_all_data(self): start_time = datetime.now(self.ist) - timedelta(days=7) end_time = datetime.now(self.ist).replace(hour=15, minute=30) symbols = ["NSE:Nifty 50", "NSE:Nifty Bank", "NSE:NIFTY MID SELECT", "NSE:Nifty Fin Service", "BSE:SENSEX"] #symbols = ["NSE:Nifty 50"] for interval in [3]: for symbol in symbols: self.setEntryDone() time.sleep(1) self.process_symbol_data(symbol, interval, start_time, end_time) print ("") if __name__ == "__main__": client_details = ['TT1582', 'O9i8u7y6##', '15111982', 'TT1582_API', 'ed87fa73888bc882c5d79bf832c0f9eb'] stock_fetcher = StockDataFetcher(client_details) stock_fetcher.login() stock_fetcher.fetch_all_data()