import logging import sys import time as time_module from datetime import datetime, time, timedelta from typing import List, Tuple import pytz import requests import json 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() data = response.json() if not data.get('success', False): raise ValueError("getUserDetails endpoint returned unsuccessful response") user_data = data['data'] return [user_data['field1'], user_data['field2'], user_data['field3'], user_data['field4'], user_data['field5']] 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 class Class180: 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.createEntries = createEntries self.lotCount = 1 self.nifty_50_derivatives = { "ADANIENT" : {"lotSize": 300, "strikeInterval": 50, "exchange": "NFO"}, "ADANIPORTS" : {"lotSize": 475, "strikeInterval": 20, "exchange": "NFO"}, "APOLLOHOSP" : {"lotSize": 125, "strikeInterval": 50, "exchange": "NFO"}, "ASIANPAINT" : {"lotSize": 250, "strikeInterval": 20, "exchange": "NFO"}, "AXISBANK" : {"lotSize": 625, "strikeInterval": 10, "exchange": "NFO"}, "BAJAJ-AUTO" : {"lotSize": 75, "strikeInterval": 100, "exchange": "NFO"}, "BAJFINANCE" : {"lotSize": 750, "strikeInterval": 100, "exchange": "NFO"}, "BAJAJFINSV" : {"lotSize": 500, "strikeInterval": 20, "exchange": "NFO"}, "BHARTIARTL" : {"lotSize": 475, "strikeInterval": 20, "exchange": "NFO"}, "BPCL" : {"lotSize": 1975, "strikeInterval": 10, "exchange": "NFO"}, "BRITANNIA" : {"lotSize": 125, "strikeInterval": 50, "exchange": "NFO"}, "CIPLA" : {"lotSize": 375, "strikeInterval": 20, "exchange": "NFO"}, "COALINDIA" : {"lotSize": 1350, "strikeInterval": 10, "exchange": "NFO"}, "DIVISLAB" : {"lotSize": 100, "strikeInterval": 50, "exchange": "BFO"}, "DMART" : {"lotSize": 150, "strikeInterval": 50, "exchange": "NFO"}, "DRREDDY" : {"lotSize": 625, "strikeInterval": 20, "exchange": "NFO"}, "EICHERMOT" : {"lotSize": 175, "strikeInterval": 50, "exchange": "NFO"}, "GRASIM" : {"lotSize": 250, "strikeInterval": 20, "exchange": "NFO"}, "HCLTECH" : {"lotSize": 350, "strikeInterval": 20, "exchange": "NFO"}, "HDFCBANK" : {"lotSize": 1100, "strikeInterval": 10, "exchange": "NFO"}, "HDFCLIFE" : {"lotSize": 1100, "strikeInterval": 10, "exchange": "NFO"}, "HEROMOTOCO" : {"lotSize": 150, "strikeInterval": 50, "exchange": "NFO"}, "HINDALCO" : {"lotSize": 1400, "strikeInterval": 10, "exchange": "NFO"}, "HINDUNILVR" : {"lotSize": 300, "strikeInterval": 20, "exchange": "NFO"}, "ICICIBANK" : {"lotSize": 700, "strikeInterval": 10, "exchange": "NFO"}, "INDUSINDBK" : {"lotSize": 700, "strikeInterval": 10, "exchange": "NFO"}, "INFY" : {"lotSize": 400, "strikeInterval": 20, "exchange": "NFO"}, "ITC" : {"lotSize": 1600, "strikeInterval": 10, "exchange": "NFO"}, "JSWSTEEL" : {"lotSize": 675, "strikeInterval": 10, "exchange": "NFO"}, "KOTAKBANK" : {"lotSize": 400, "strikeInterval": 20, "exchange": "NFO"}, "LT" : {"lotSize": 175, "strikeInterval": 50, "exchange": "NFO"}, "M&M" : {"lotSize": 200, "strikeInterval": 50, "exchange": "NFO"}, "MARUTI" : {"lotSize": 50, "strikeInterval": 100, "exchange": "NFO"}, "NESTLEIND" : {"lotSize": 500, "strikeInterval": 50, "exchange": "NFO"}, "NTPC" : {"lotSize": 1500, "strikeInterval": 5, "exchange": "NFO"}, "ONGC" : {"lotSize": 2250, "strikeInterval": 5, "exchange": "NFO"}, "POWERGRID" : {"lotSize": 1900, "strikeInterval": 5, "exchange": "NFO"}, "RELIANCE" : {"lotSize": 500, "strikeInterval": 50, "exchange": "NFO"}, "SBILIFE" : {"lotSize": 375, "strikeInterval": 20, "exchange": "NFO"}, "SBIN" : {"lotSize": 750, "strikeInterval": 10, "exchange": "NFO"}, "SHREECEM" : {"lotSize": 25, "strikeInterval": 250, "exchange": "NFO"}, "SUNPHARMA" : {"lotSize": 350, "strikeInterval": 20, "exchange": "NFO"}, "TATACONSUM" : {"lotSize": 550, "strikeInterval": 10, "exchange": "NFO"}, "TATAMOTORS" : {"lotSize": 800, "strikeInterval": 10, "exchange": "NFO"}, "TATASTEEL" : {"lotSize": 5500, "strikeInterval": 5, "exchange": "NFO"}, "TCS" : {"lotSize": 175, "strikeInterval": 20, "exchange": "NFO"}, "TECHM" : {"lotSize": 600, "strikeInterval": 20, "exchange": "NFO"}, "TITAN" : {"lotSize": 175, "strikeInterval": 50, "exchange": "NFO"}, "ULTRACEMCO" : {"lotSize": 50, "strikeInterval": 100, "exchange": "NFO"}, "WIPRO" : {"lotSize": 3000, "strikeInterval": 10, "exchange": "NFO"} } self.pgm_start_time = time(9, 27) self.stock_derivative_expiry_bfo = "25SEP" self.stock_derivative_expiry_nfo = "30SEP25" self.top_gainer_count = 2 self.top_loser_count = 2 self.testing_enabled = False if self.testing_enabled: self.pgm_start_time = time(7, 27) 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 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 login(self): try: self.logger.info(f"Attempting login for {self.client_details[0]}") 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}") sys.exit() except Exception as e: self.logger.error(f"Login error: {e}") sys.exit() def create_order(self, tickTimeStr, instrument, price, action, lotCount): if self.createEntries: url = "http://143.244.141.41/php/getUserDetails.php" params = { 'tickTime': str(tickTimeStr), 'instrument': str(instrument), 'closePrice': str(price), 'signal': str(action), 'orderType': str(lotCount) } try: response = requests.get(url, params=params) if response.status_code != 200: self.logger.error(f"Order failed with status {response.status_code}: {response.text}") except Exception as e: self.logger.error(f"Order request failed: {e}") def get_strike_price(self, current_price, strike_interval): try: price = float(current_price) remainder = price % strike_interval if remainder > strike_interval / 2: return int(price + (strike_interval - remainder)) else: return int(price - remainder) except (ValueError, TypeError): self.logger.error(f"Invalid price for strike calculation: {current_price}") return None def get_tradingsymbol_price_dict(self, mQ): if not isinstance(mQ, dict) or mQ.get("status") != "success" or "data" not in mQ: return {} return { item["result"]["tradingSymbol"]: { "lastTradedPrice": float(item["result"].get("lastTradedPrice", 0)), "dayClosePrice": float(item["result"].get("dayClosePrice", 0)), "openPrice": float(item["result"].get("openPrice", 0)), "lotSize": int(item["result"].get("lotSize", 0)), "changePercent": round( ((float(item["result"].get("lastTradedPrice", 0)) - float(item["result"].get("dayClosePrice", 0))) / float(item["result"].get("dayClosePrice", 1))) * 100, 2 ) } for item in mQ["data"] if "result" in item } def get_top_gainers_losers(self, price_dict: dict, top_gainer_count: int, top_loser_count: int) -> Tuple[List[str], List[str]]: sorted_items = sorted( price_dict.items(), key=lambda x: x[1]['changePercent'], reverse=False ) losers = [item[0].replace('-EQ', '') for item in sorted_items[:top_loser_count]] gainers = [item[0].replace('-EQ', '') for item in sorted_items[-top_gainer_count:]][::-1] return gainers, losers def fetch_and_process_data(self, elapsed: int): self.logger.info(f"Waiting Till : {self.pgm_start_time}") while True: now = datetime.now(self.ist).time() if now >= self.pgm_start_time: break time_module.sleep(1) nifty_50_stocks = list(self.nifty_50_derivatives.keys()) nse_equity_symbols = [f"{stock}-EQ" for stock in nifty_50_stocks] data_token_list = [{"exchange": "NSE", "tradingSymbol": symbol} for symbol in nse_equity_symbols] try: self.logger.info(f"Wait Complete. Fetching Nifty 50 stock prices...") mQ = thefirstock.firstock_getMultiQuote( userId=self.user_id, dataToken=data_token_list ) price_dict = self.get_tradingsymbol_price_dict(mQ) if self.testing_enabled: self.logger.info("Testing code to fetch previous day close price ..") for symbol, price in price_dict.items(): self.logger.info(f"Getting close price for {symbol} ..") now_ist = datetime.now(self.ist) elapsed = 3 start_time = now_ist.replace(hour=15, minute=30, second=0) - timedelta(days=elapsed) end_time = now_ist.replace(hour=15, minute=33, second=0) - timedelta(days=elapsed) try: df = thefirstock.firstock_TimePriceSeries( userId=self.user_id, exchange='NSE', tradingSymbol=symbol, startTime=start_time.strftime("%d/%m/%Y %H:%M:%S"), endTime=end_time.strftime("%d/%m/%Y %H:%M:%S"), interval="1" ) if df.get('data') and len(df['data']) > 0: cp = df['data'][0].get('intc') if cp is not None: ltp = price_dict[symbol]['lastTradedPrice'] price_dict[symbol]['dayClosePrice'] = float(cp) price_dict[symbol]['changePercent'] = round(((float(ltp) - float(cp)) / float(cp)) * 100, 2) else: self.logger.warning(f"No close price found for {symbol}") else: self.logger.warning(f"No data returned for {symbol} in TimePriceSeries") except Exception as e: self.logger.error(f"Error fetching historical data for {symbol}: {e}") self.logger.info("") self.logger.info("Nifty 50 Stock Prices:") self.logger.info("=" * 80) self.logger.info(f"{'Symbol':<20} {'Current Price':>15} {'PrevClose':>15} {'Change%':>10}") self.logger.info("=" * 80) for symbol, price in price_dict.items(): self.logger.info( f"{symbol:<20} {price['lastTradedPrice']:>15.2f} {price['dayClosePrice']:>15.2f} {price['changePercent']:>10.2f}%" ) self.logger.info("=" * 80) top_gainers, top_losers = self.get_top_gainers_losers(price_dict, self.top_gainer_count, self.top_loser_count) self.logger.info("") self.logger.info(f"Top {self.top_gainer_count} Gainers:") self.logger.info(", ".join(top_gainers)) self.logger.info(f"Top {self.top_loser_count} Losers:") self.logger.info(", ".join(top_losers)) buy_list = [] self.logger.info("") self.logger.info("Creating buy_list with next strike for gainers and previous strike for losers:") for symbol in top_gainers + top_losers: eq_symbol = f"{symbol}-EQ" if eq_symbol in price_dict and symbol in self.nifty_50_derivatives: ltp = price_dict[eq_symbol]['lastTradedPrice'] strike_interval = self.nifty_50_derivatives[symbol]['strikeInterval'] exchange = self.nifty_50_derivatives[symbol]['exchange'] nearest_strike = self.get_strike_price(ltp, strike_interval) if nearest_strike is not None: next_strike = nearest_strike + strike_interval prev_strike = nearest_strike - strike_interval if symbol in top_gainers: strike = next_strike option_type = "CE" else: strike = prev_strike option_type = "PE" if exchange == "BFO": expiry = self.stock_derivative_expiry_bfo trading_symbol = f"{symbol}{expiry}{strike}{option_type}" elif exchange == "NFO": expiry = self.stock_derivative_expiry_nfo trading_symbol = f"{symbol}{expiry}{option_type[0]}{strike}" buy_list.append({"exchange": exchange, "tradingSymbol": trading_symbol}) else: self.logger.info(f"{symbol:<15} {'Invalid Price':>10} {strike_interval:>15} {'N/A':>15} {'N/A':>15} {'N/A'}") else: self.logger.info(f"{symbol:<15} {'N/A':>10} {'N/A':>15} {'N/A':>15} {'N/A':>15} {'N/A'}") if buy_list: self.logger.info("") self.logger.info("Buy List Trading Symbols:") self.logger.info(", ".join(item["tradingSymbol"] for item in buy_list)) self.logger.info("") self.logger.info("Fetching quotes for buy_list:") try: mQ = thefirstock.firstock_getMultiQuote( userId=self.user_id, dataToken=buy_list ) buy_price_dict = self.get_tradingsymbol_price_dict(mQ) self.logger.info("=" * 90) self.logger.info(f"{'Symbol':<30} {'Current Price':>15} {'Open Price':>15} {'Change%':>10} {'Amt Required':>15}") self.logger.info("=" * 90) # Prepare Markdown content timestamp = datetime.now(self.ist).strftime('%Y-%m-%d %H:%M:%S') for symbol, price in buy_price_dict.items(): base_symbol = symbol.split(self.stock_derivative_expiry_nfo)[0] if self.stock_derivative_expiry_nfo in symbol else symbol.split(self.stock_derivative_expiry_bfo)[0] lot_size = self.nifty_50_derivatives.get(base_symbol, {}).get('lotSize', 0) amt_required = price['lastTradedPrice'] * lot_size if lot_size > 0 else 0.0 self.logger.info( f"{symbol:<30} {price['lastTradedPrice']:>15.2f} {price['openPrice']:>15.2f} {price['changePercent']:>10.2f}% {amt_required:>15.2f}" ) self.logger.info("=" * 90) except Exception as e: self.logger.error(f"Error fetching quotes for buy_list: {e}") else: self.logger.info("No symbols in buy_list to fetch quotes for.") self.logger.info("Completed processing buy_list for top gainers and losers.") except Exception as e: self.logger.error(f"Error fetching Nifty 50 stock prices: {e}") return if __name__ == "__main__": client_details = get_client_details() if client_details: elapsed = float(sys.argv[1]) if len(sys.argv) > 1 else 0 createEntries = len(sys.argv) > 2 trader = Class180(client_details, createEntries) trader.login() trader.fetch_and_process_data(elapsed) print("done") else: print("Failed to get client details")