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 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))] # 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) columns_to_remove = ['ssboe', 'into', 'inth', 'intl', 'intvwap', 'intv', 'intoi', 'v', 'oi'] df = df.drop(columns=columns_to_remove, errors='ignore') # Calculate DEMA, MA10, and Supertrend df['intc'] = pd.to_numeric(df['intc'], errors='coerce') 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 df = df.copy() # Add this line before modifying the DataFrame df.loc[:, 'instrument'] = df['intc'].apply(lambda x: str(round(x / step_value) * step_value)) df['DEMA10'] = df.ta.dema(close='intc', length=10) df['MA100'] = df.ta.sma(close='intc', length=100) df.loc[:, 'DEMA10'] = pd.to_numeric(df['DEMA10'], errors='coerce') df.loc[:, 'MA100'] = pd.to_numeric(df['MA100'], errors='coerce') df['MA100-5'] = df['MA100'] - df['MA100'].shift(5) df['DEMA10-MA100'] = df['DEMA10'] - df['MA100'] df['DEMA10-MA100-Trend'] = df['DEMA10-MA100'].apply(lambda x: 'UP' if x > 0 else 'DOWN') df['MA100-Trend'] = df['MA100-5'].apply(lambda x: 'UP' if x > 0.3 else ('DOWN' if x < -0.3 else 'FLAT')) df['MA100-INTC-Trend'] = df.apply(lambda row: 'DOWN' if row['intc'] < row['MA100'] else 'UP', axis=1) df['Action'] = df.apply( lambda row: 'CE' if row['DEMA10-MA100-Trend'] == 'UP' and row['MA100-Trend'] == 'UP' and row['MA100-INTC-Trend'] == 'UP' else ('PE' if row['DEMA10-MA100-Trend'] == 'DOWN' and row['MA100-Trend'] == 'DOWN' and row['MA100-INTC-Trend'] == 'DOWN' else ''), axis=1 ) df = df.dropna(subset=['DEMA10', 'MA100', 'MA100-5']) df = df[df['elapsedTime'] <= 4000] file_name = f"{tmp_symbol}.csv" output_path = os.path.join("/var/www/html/ACTIONS/", file_name) # Change the directory as needed df.to_csv(output_path, index=False) self.logger.info(f"Data saved to {output_path}") print(df) 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: 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()