import os import time import pytz import pandas as pd import numpy as np np.NaN = np.nan # Add this line before importing pandas_ta import pandas_ta as ta import logging from datetime import datetime, timedelta from typing import List from thefirstock import thefirstock #import matplotlib.pyplot as plt import requests 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() print("\n\n\n\n") self.logger.info("**************************") self.total = {} 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"----------------------------------") 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 plot_bollinger_bands(self, df: pd.DataFrame, symbol: str, interval: int): """ Visualize Bollinger Bands with closing price, DMA status, and MACD in subplots All time displays converted to IST (Indian Standard Time) """ if df.empty: self.logger.warning("Empty DataFrame received for plotting") return # Create a copy and ensure proper timezone handling df = df.copy() if df['time'].dt.tz is None: df['time'] = df['time'].dt.tz_localize(self.ist) else: df['time'] = df['time'].dt.tz_convert(self.ist) # Convert timestamps to matplotlib-compatible format in IST plot_times = df['time'].dt.tz_convert(self.ist).values.astype('datetime64[ns]') # Create figure with 4 subplots fig, (ax0, ax1, ax2, ax3) = plt.subplots(4, 1, figsize=(15, 15), gridspec_kw={'height_ratios': [1, 2, 1, 1]}) # Top plot - Price with action status highlights ax0.plot(plot_times, df['intc'], label='Closing Price', color='blue', alpha=0.7, linewidth=2) # Highlight strong buy/sell signals where all three indicators agree buy_periods = df[df['action_status'] == 'Buy'] sell_periods = df[df['action_status'] == 'Sell'] # Add exit points as stars buy_exit_points = df[df['exitPoint'] == 'BuyExit'] if not buy_exit_points.empty: exit_times = buy_exit_points['time'].dt.tz_convert(self.ist).values.astype('datetime64[ns]') ax0.scatter(exit_times, buy_exit_points['intc'], marker='*', color='green', s=100, label='Buy Exit Point', zorder=5) sell_exit_points = df[df['exitPoint'] == 'SellExit'] if not sell_exit_points.empty: exit_times = sell_exit_points['time'].dt.tz_convert(self.ist).values.astype('datetime64[ns]') ax0.scatter(exit_times, sell_exit_points['intc'], marker='*', color='red', s=100, label='Sell Exit Point', zorder=5) # Add background color for action periods prev_status = None start_idx = 0 for i, status in enumerate(df['action_status']): if status != prev_status: if prev_status is not None: if prev_status == 'Buy': ax0.axvspan(plot_times[start_idx], plot_times[i-1], facecolor='lightgreen', alpha=0.3) elif prev_status == 'Sell': ax0.axvspan(plot_times[start_idx], plot_times[i-1], facecolor='lightcoral', alpha=0.3) start_idx = i prev_status = status # Fill the last region if needed if prev_status == 'Buy': ax0.axvspan(plot_times[start_idx], plot_times[-1], facecolor='lightgreen', alpha=0.3) elif prev_status == 'Sell': ax0.axvspan(plot_times[start_idx], plot_times[-1], facecolor='lightcoral', alpha=0.3) ax0.set_title(f'{symbol} - {interval} min - {df["date"].iloc[0].strftime("%d-%m-%Y")}') ax0.set_ylabel('Price') ax0.legend(loc='upper left') ax0.grid(True, which='both', linestyle='--', alpha=0.5) # Second plot - Bollinger Bands ax1.plot(plot_times, df['intc'], label='Closing Price', color='blue', alpha=0.7, linewidth=2) ax1.plot(plot_times, df['upper_band'], label='Upper Band', color='red', linestyle='--', alpha=0.7) ax1.plot(plot_times, df['middle_band'], label='Middle Band', color='green', alpha=0.7) ax1.plot(plot_times, df['lower_band'], label='Lower Band', color='red', linestyle='--', alpha=0.7) # Fill between the bands ax1.fill_between(plot_times, df['lower_band'], df['upper_band'], color='orange', alpha=0.1) # Highlight band touches upper_touches = df[df['intc'] >= df['upper_band']] lower_touches = df[df['intc'] <= df['lower_band']] if not upper_touches.empty: ax1.scatter(upper_touches['time'].dt.tz_convert(self.ist).values.astype('datetime64[ns]'), upper_touches['intc'], color='red', marker='^', s=100, label='Upper Band Touch') if not lower_touches.empty: ax1.scatter(lower_touches['time'].dt.tz_convert(self.ist).values.astype('datetime64[ns]'), lower_touches['intc'], color='green', marker='v', s=100, label='Lower Band Touch') # Add background color based on bollingerStatus prev_status = None start_idx = 0 for i, status in enumerate(df['bollingerStatus']): if status != prev_status: if prev_status is not None: if prev_status == 'Buy': ax1.axvspan(plot_times[start_idx], plot_times[i-1], facecolor='lightgreen', alpha=0.2) elif prev_status == 'Sell': ax1.axvspan(plot_times[start_idx], plot_times[i-1], facecolor='lightcoral', alpha=0.2) start_idx = i prev_status = status # Fill the last region if prev_status == 'Buy': ax1.axvspan(plot_times[start_idx], plot_times[-1], facecolor='lightgreen', alpha=0.2) elif prev_status == 'Sell': ax1.axvspan(plot_times[start_idx], plot_times[-1], facecolor='lightcoral', alpha=0.2) ax1.set_ylabel('Price with Bands') ax1.legend(loc='upper left') ax1.grid(True, which='both', linestyle='--', alpha=0.5) # Third plot - DMA status ax2.plot(plot_times, df['intc'], label='Price', color='blue', alpha=0.7, linewidth=1) ax2.plot(plot_times, df['dma'], label='200 DMA', color='black', alpha=0.9, linewidth=1.5) # Add background color based on DMA status prev_status = None start_idx = 0 for i, status in enumerate(df['dmaStatus']): if status != prev_status: if prev_status is not None: if prev_status == 'Buy': ax2.axvspan(plot_times[start_idx], plot_times[i-1], facecolor='lightgreen', alpha=0.3) elif prev_status == 'Sell': ax2.axvspan(plot_times[start_idx], plot_times[i-1], facecolor='lightcoral', alpha=0.3) start_idx = i prev_status = status # Fill the last region if it's Buy or Sell if prev_status == 'Buy': ax2.axvspan(plot_times[start_idx], plot_times[-1], facecolor='lightgreen', alpha=0.3) elif prev_status == 'Sell': ax2.axvspan(plot_times[start_idx], plot_times[-1], facecolor='lightcoral', alpha=0.3) ax2.set_ylabel('Price vs DMA') ax2.legend(loc='upper left') ax2.grid(True, which='both', linestyle='--', alpha=0.5) # Bottom plot - MACD ax3.plot(plot_times, df['macd_line'], label='MACD Line', color='blue', linewidth=1.5) ax3.plot(plot_times, df['signal_line'], label='Signal Line', color='orange', linewidth=1.5) # Add zero line for MACD ax3.axhline(0, color='gray', linestyle='--', linewidth=0.7) # Add background color based on MACD status prev_status = None start_idx = 0 for i, status in enumerate(df['macd_status']): if status != prev_status: if prev_status is not None: if prev_status == 'Buy': ax3.axvspan(plot_times[start_idx], plot_times[i-1], facecolor='lightgreen', alpha=0.3) elif prev_status == 'Sell': ax3.axvspan(plot_times[start_idx], plot_times[i-1], facecolor='lightcoral', alpha=0.3) start_idx = i prev_status = status # Fill the last region if it's Buy or Sell if prev_status == 'Buy': ax3.axvspan(plot_times[start_idx], plot_times[-1], facecolor='lightgreen', alpha=0.3) elif prev_status == 'Sell': ax3.axvspan(plot_times[start_idx], plot_times[-1], facecolor='lightcoral', alpha=0.3) ax3.set_xlabel('Time (IST)') ax3.set_ylabel('MACD') ax3.legend(loc='upper left') ax3.grid(True, which='both', linestyle='--', alpha=0.5) # Format x-axis dates with IST timezone ist_formatter = plt.matplotlib.dates.DateFormatter('%H:%M', tz=self.ist) for ax in [ax0, ax1, ax2, ax3]: ax.xaxis.set_major_formatter(ist_formatter) plt.xticks(rotation=45) plt.tight_layout() plt.show() def process_symbol_data(self, symbol: str, interval: int, start_time: datetime, end_time: datetime): print(f'\n\nprocess_symbol_data:: startTime: {start_time.strftime("%d/%m/%Y")} endTime: {end_time.strftime("%d/%m/%Y")} interval:{interval}') print('===============') 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), ) # Preprocess timestamps and ensure data integrity df['time'] = pd.to_datetime(df['time'], errors='coerce', dayfirst=True) df['time'] = df['time'].dt.tz_localize(self.ist) df['date'] = df['time'].dt.tz_convert(self.ist).dt.date numeric_cols = ['intc', 'intv', 'into', 'inth', 'intl'] for col in numeric_cols: df[col] = pd.to_numeric(df[col], errors='coerce') df = df.sort_values(by='time', ascending=True) # Calculate Bollinger Bands using pandas_ta # We'll use the closing price (intc) for calculations bbands = ta.bbands( close=df['intc'], length=20, # Standard 20-period lookback std=2, # 2 standard deviations mamode='sma' # Simple moving average for the basis ) # Add Bollinger Bands columns to the DataFrame df = pd.concat([df, bbands], axis=1) # Rename columns for clarity df.rename(columns={ 'BBU_20_2.0': 'upper_band', 'BBM_20_2.0': 'middle_band', 'BBL_20_2.0': 'lower_band' }, inplace=True) # Calculate bandwidth and %b (optional but useful indicators) df['bandwidth'] = (df['upper_band'] - df['lower_band']) / df['middle_band'] df['percent_b'] = (df['intc'] - df['lower_band']) / (df['upper_band'] - df['lower_band']) # Initialize bollingerStatus column df['bollingerStatus'] = 'No Action' buy_active = False sell_active = False for i in range(1, len(df)): # Check for upper band touch to initiate buy signal if df['intc'].iloc[i] >= df['upper_band'].iloc[i]: buy_active = False sell_active = True # Check for lower band touch to initiate sell signal elif df['intc'].iloc[i] <= df['lower_band'].iloc[i]: sell_active = False buy_active = True # Check for exit conditions if buy_active and df['intc'].iloc[i] > df['middle_band'].iloc[i]: buy_active = False elif sell_active and df['intc'].iloc[i] < df['middle_band'].iloc[i]: sell_active = False # Set the status if buy_active: df.at[df.index[i], 'bollingerStatus'] = 'Buy' elif sell_active: df.at[df.index[i], 'bollingerStatus'] = 'Sell' # Initialize exitPoint column df['exitPoint'] = 'None' # Check for status changes from Buy or Sell to something else for i in range(1, len(df)): prev_status = df['bollingerStatus'].iloc[i-1] current_status = df['bollingerStatus'].iloc[i] if (prev_status == 'Buy' and current_status != 'Buy'): df.at[df.index[i], 'exitPoint'] = 'SellExit' elif (prev_status == 'Sell' and current_status != 'Sell'): df.at[df.index[i], 'exitPoint'] = 'BuyExit' # Calculate 200-day moving average df['dma'] = df['intc'].rolling(window=200, min_periods=1).mean() # Forward fill the 200dma to handle initial periods with insufficient data df['dma'] = df['dma'].ffill() df['dmaStatus'] = np.where(df['intc'] > df['dma'], 'Buy', np.where(df['intc'] < df['dma'], 'Sell', 'No Action')) macd = ta.macd( close=df['intc'], fast=12, slow=26, signal=9 ) # Add MACD columns to the DataFrame df = pd.concat([df, macd], axis=1) # Rename MACD columns for clarity df.rename(columns={ 'MACD_12_26_9': 'macd_line', 'MACDs_12_26_9': 'signal_line', 'MACDh_12_26_9': 'macd_histogram' }, inplace=True) # Calculate MACD status df['macd_status'] = 'No Action' # Initialize with default value # Create conditions for Buy and Sell signals buy_condition = ( #(df['macd_line'] > df['signal_line']) & #(df['macd_line'].shift() <= df['signal_line'].shift()) & # Cross above (df['macd_line'] < -1) ) sell_condition = ( #(df['macd_line'] < df['signal_line']) & #(df['macd_line'].shift() >= df['signal_line'].shift()) & # Cross below (df['macd_line'] > 1) ) # Apply conditions df.loc[buy_condition, 'macd_status'] = 'Buy' df.loc[sell_condition, 'macd_status'] = 'Sell' # Forward fill the status until the next signal appears #df['macd_status'] = df['macd_status'].replace('No Action', method='ffill') #df['macd_status'].fillna('No Action', inplace=True) # Handle initial NaN values df['action_status'] = 'No Action' df.loc[(df['macd_status'] == 'Buy') & (df['dmaStatus'] == 'Buy') & (df['bollingerStatus'] == 'Buy'), 'action_status'] = 'Buy' df.loc[(df['macd_status'] == 'Sell') & (df['dmaStatus'] == 'Sell') & (df['bollingerStatus'] == 'Sell'), 'action_status'] = 'Sell' last_date = df['date'].max() df_last_day = df[df['date'] == last_date].copy() df_last_day = df_last_day[~((df_last_day['time'].dt.hour == 9) & (df_last_day['time'].dt.minute < 15))] df_last_day = df_last_day[~(((df_last_day['time'].dt.hour == 14) & (df_last_day['time'].dt.minute >= 53)) | (df_last_day['time'].dt.hour > 14) )] # Initialize trade tracking variables current_trade = None # 'Buy', 'Sell', or None trade_signals = [] buy_price = 0.0 sell_price = 0.0 nifty_step_value = 50 orderType = 0 CE_instrument = 'None' PE_instrument = 'None' CE_OrderType = 0 PE_OrderType = 0 target_price = 47 dmaTolenece = 5 buy_exit_met = False sell_exit_met = False overall_profit = 0 # Process each row in the last day's data for i in range(len(df_last_day)): row = df_last_day.iloc[i] prev_row = df_last_day.iloc[i-1] if i > 0 else None if prev_row is None: continue timeStr = row['time'] # Determine if this is a new signal new_buy = (row['action_status'] == 'Buy' and (prev_row['action_status'] != 'Buy')) new_sell = (row['action_status'] == 'Sell' and (prev_row['action_status'] != 'Sell')) buy_exit = (row['exitPoint'] == 'BuyExit' and current_trade == 'Buy') sell_exit = (row['exitPoint'] == 'SellExit' and current_trade == 'Sell') # Generate signals if current_trade is None: if new_buy: buy_price = row['intc'] trade_signals.append({ 'time': row['time'], 'signal': 'BUY', 'price': row['intc'], 'message': f"New BUY signal at {row['time']} - Price: {row['intc']:.2f}" }) current_trade = 'Buy' orderType = orderType + 1 closePrice = row['intc'] signal = 'BUY' instrument = 'NIFTY' + str(round(row['intc'] / nifty_step_value) * nifty_step_value ) + 'CE' CE_instrument = instrument CE_OrderType = orderType self.createEntry(timeStr, instrument, closePrice, signal, str(orderType)) buy_exit_met = False elif new_sell: sell_price = row['intc'] trade_signals.append({ 'time': row['time'], 'signal': 'SELL', 'price': row['intc'], 'message': f"New SELL signal at {row['time']} - Price: {row['intc']:.2f}" }) current_trade = 'Sell' orderType = orderType + 1 closePrice = row['intc'] signal = 'BUY' instrument = 'NIFTY' + str(round(row['intc'] / nifty_step_value) * nifty_step_value) + 'PE' self.createEntry(timeStr, instrument, closePrice, signal, str(orderType)) PE_instrument = instrument PE_OrderType = orderType sell_exit_met = False else: if current_trade == 'Buy': profit = row['intc'] - buy_price dmaChange = (row['intc'] < (row['dma'] - dmaTolenece)) if buy_exit: buy_exit_met = True if (buy_exit_met and row['intc'] <= row['lower_band']) or profit >= target_price or dmaChange: print (f'buy_exit_met:{buy_exit_met} intc:{row['intc']:.2f} lower_band:{row['lower_band']:.2f} profit:{profit:.2f} target_price:{target_price} dmaChange:{dmaChange}') trade_signals.append({ 'time': row['time'], 'signal': 'EXIT_BUY', 'price': row['intc'], 'message': f"Exit BUY position at {row['time']} - Price: {row['intc']:.2f} - profit: {profit:.2f}\n" }) current_trade = None closePrice = row['intc'] signal = 'SELL' instrument = CE_instrument self.createEntry(timeStr, instrument, closePrice, signal, str(CE_OrderType)) CE_instrument = 'None' CE_OrderType = 0 buy_exit_met = False overall_profit = overall_profit + profit elif current_trade == 'Sell': profit = sell_price - row['intc'] dmaChange = (row['intc'] > (row['dma'] + dmaTolenece)) if sell_exit: sell_exit_met = True if (sell_exit_met and row['intc'] >= row['upper_band']) or profit >= target_price or dmaChange: print (f'sell_exit_met:{sell_exit_met} intc:{row['intc']:.2f} upper_band:{row['upper_band']:.2f} profit:{profit:.2f} target_price:{target_price} dmaChange:{dmaChange}') trade_signals.append({ 'time': row['time'], 'signal': 'EXIT_SELL', 'price': row['intc'], 'message': f"Exit SELL position at {row['time']} - Price: {row['intc']:.2f} - profit: {profit:.2f}\n" }) current_trade = None closePrice = row['intc'] signal = 'SELL' instrument = PE_instrument self.createEntry(timeStr, instrument, closePrice, signal, str(PE_OrderType)) PE_instrument = 'None' PE_OrderType = 0 sell_exit_met = False overall_profit = overall_profit + profit # Print all trade signals print("\nTrade Signals:") print("--------------") if not trade_signals: print("No trade signals generated") else: for signal in trade_signals: print(signal['message']) print(f'\n overall_profit = {overall_profit:.2f} \n') self.total[row['time'].strftime('%Y-%m-%d')] = round(float(overall_profit),2) # Add signals to DataFrame df_last_day['trade_signal'] = 'None' for signal in trade_signals: mask = (df_last_day['time'] == signal['time']) df_last_day.loc[mask, 'trade_signal'] = signal['signal'] ''' file_name = "NiftyData.csv" df_last_day.to_csv(file_name, index=False) self.logger.info(f"Data saved to {file_name}") #self.plot_bollinger_bands(df_last_day, symbol, interval) print("\nFinal Data:") print("-----------") print(df_last_day[['time', 'intc', 'action_status', 'exitPoint', 'trade_signal']]) ''' def print_total(self): print(f"\n\n{'Date':<15} {'Profit':>10}") print("-" * 26) # Print each row for date, profit in self.total.items(): print(f"{date:<15} {profit:>10.2f}") # Print total if you want total_profit = sum(self.total.values()) print("-" * 26) print(f"{'Total':<15} {total_profit:>10.2f}") def createEntry(self, time, instrument, closePrice, signal, orderType): #return url = "http://143.244.141.41/php/createDeepSeekEntry.php" # Replace with your actual URL tickTime_str = time.strftime("%Y-%m-%d_%H:%M:%S") # 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": 5, } backTest = False no_of_backTestDays = 5 if not backTest: for interval in [1]: for symbol, days in symbols.items(): time.sleep(1) start_time = datetime.now(self.ist).replace(hour=9, minute=15, second=0) start_time = start_time - timedelta(days=days) end_time = datetime.now(self.ist).replace(hour=15, minute=30, second=0) self.process_symbol_data(symbol, interval, start_time, end_time) else: for interval in [1]: for symbol, days in symbols.items(): # Get current time in IST now = datetime.now(self.ist) # Loop through last 30 days for day_offset in range(no_of_backTestDays): time.sleep(1) # Add delay between requests print(day_offset) # Calculate end time (market close) for each day in the past end_time = now.replace(hour=15, minute=30, second=0) - timedelta(days=day_offset) # Calculate start time (market open) 5 days before the end time start_time = end_time.replace(hour=9, minute=15, second=0) - timedelta(days=5) # Process the data for this time period self.process_symbol_data(symbol, interval, start_time, end_time) if __name__ == "__main__": client_details = ['GA0810', 'O9i8u7y6$$', '08101994', 'GA0810_API', 'ada14c5f73f182ca724b90c5dde1e72d'] client_details = ['RA1383', 'O9i8u7y6$$', '13061983', 'RA1383_API', '85c819b0c188c44bc714b87645aef3d6'] stock_fetcher = StockDataFetcher(client_details) stock_fetcher.login() stock_fetcher.fetch_all_data() stock_fetcher.print_total()