import logging import sys import time as time_module # Use alias to avoid conflict with datetime.time from datetime import datetime, timedelta, time from typing import List import pytz import pandas as pd import requests import json from thefirstock import thefirstock import re import pprint 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 # Trading parameters self.timeframe_period = 21 self.smoothing_period = 3 self.tick_interval = 5 self.lotCount = 1 self.no_of_past_data_days = 4 self.max_loss = 40 # Dynamic stop loss and take profit offsets self.profit = 100 # Take profit offset from buy price self.baseProfitLevel = 40 # Base profit level to trigger stop loss adjustment self.stop_loss = 20 # Initial stop loss offset from buy price self.stop_loss_move_int_level = self.baseProfitLevel - 30 # Should be 20 point diff from self.baseProfitLevel # Track positions and P/L self.positions = {'CE': None, 'PE': None} self.total_pl = 0.0 self.trades = [] self.wins = 0 self.total_trades = 0 self.highest_prices = {} # Track highest price for each instrument self.stop_loss_states = {} # Track stop loss offset for each instrument self.pgm_start_time = time(9, 31) self.retry_time = time(10, 0) self.exit_time = time(13, 0) self.buy_start_range = 180 self.buy_end_range = 190 self.nifty_range_value = 6 self.nifty_step_value = 50 self.nifty_expiry_date = "02SEP25" self.testing_enabled = False if self.testing_enabled: self.retry_time = time(20, 45) self.buy_start_range = 160 self.buy_end_range = 180 self.exit_time = time(20, 0) ''' self.profit = 0 self.stop_loss = 0 ''' 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): instrument = instrument.replace(self.nifty_expiry_date, '') if instrument.startswith("NIFTYC"): instrument = instrument.replace("NIFTYC", "NIFTY") instrument = instrument + "CE" elif instrument.startswith("NIFTYP"): instrument = instrument.replace("NIFTYP", "NIFTY") instrument = instrument + "PE" if self.createEntries: url = "http://143.244.141.41/php/createEntries.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): try: price = float(current_price) remainder = price % 50 if remainder > 25: return int(price + (50 - remainder)) else: return int(price - remainder) except (ValueError, TypeError): self.logger.error(f"Invalid price for strike calculation: {current_price}") return None def get_trading_symbol_list(self, strikePriceList, range_value, expiry): trading_symbol_list = [] for i, strike in enumerate(strikePriceList): if i < range_value: trading_symbol_list.append(f"NIFTY{expiry}C{strike}") else: trading_symbol_list.append(f"NIFTY{expiry}P{strike}") return trading_symbol_list 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"]: float(item["result"]["lastTradedPrice"]) for item in mQ["data"] if "result" in item } def calculate_metrics(self): win_rate = (self.wins / self.total_trades * 100) if self.total_trades > 0 else 0 self.logger.info(f"Total P/L: {self.total_pl:.2f}") self.logger.info(f"Total Trades: {self.total_trades}, Wins: {self.wins}, Win Rate: {win_rate:.2f}%") 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) self.logger.info(f"Starting Buy Loop") try: gQ = thefirstock.firstock_getQuoteLTP( userId=self.user_id, exchange="NSE", tradingSymbol="Nifty 50" ) data = gQ.get('data', {}) last_price = data.get('lastTradedPrice', 0.0) base_strike_price = self.get_strike_price(last_price) self.logger.info(f"Base Strike Price List: {base_strike_price}") except Exception as e: self.logger.error(f"Error fetching Nifty 50 strike price exception: {e}") return range_value = self.nifty_range_value step_value = self.nifty_step_value expiry_date = self.nifty_expiry_date strikePriceList = [ base_strike_price + (i * step_value) for i in range(-range_value, range_value + 1) ] self.logger.info(f"Strike Price List: {strikePriceList}") trading_symbol_list = self.get_trading_symbol_list(strikePriceList, range_value, expiry_date) self.logger.info(f"Trading Symbol List: {trading_symbol_list}") data_token_list = [ {"exchange": "NFO", "tradingSymbol": symbol} for symbol in trading_symbol_list ] # Initial attempt to find matching instruments bought_instruments = [] buy_prices = {} self.highest_prices = {} self.stop_loss_states = {} while not bought_instruments and datetime.now(self.ist).time() < self.retry_time: current_time = datetime.now(self.ist) print(current_time) try: mQ = thefirstock.firstock_getMultiQuote( userId=self.user_id, dataToken=data_token_list ) except Exception as e: self.logger.error(f"Error fetching MultiQuote: {e}") time_module.sleep(5) continue price_dict = self.get_tradingsymbol_price_dict(mQ) self.logger.info(f"Price Dictionary: {price_dict}") current_time = datetime.now(self.ist) tickTimeStr = current_time.strftime('%Y-%m-%d %H:%M:%S') bought_instruments = [] buy_prices = {} self.highest_prices = {} self.stop_loss_states = {} for instrument, price in price_dict.items(): if self.buy_start_range <= price <= self.buy_end_range: self.create_order(tickTimeStr, instrument, price, 'BUY', self.lotCount) self.logger.info(f"Bought {instrument} at price {price}") bought_instruments.append(instrument) buy_prices[instrument] = price self.highest_prices[instrument] = price # Initialize highest price self.stop_loss_states[instrument] = self.stop_loss # Initialize stop loss state if not bought_instruments: self.logger.info("No options in 180-190 range, waiting until 9:45...") time_module.sleep(1) if not bought_instruments: self.logger.info("No options found in 180-190 range by 10:00 AM") return if self.testing_enabled: i =0 # Monitoring loop while bought_instruments: current_time = datetime.now(self.ist) tickTimeStr = current_time.strftime('%Y-%m-%d %H:%M:%S') if current_time.time() >= self.exit_time: for instrument in list(bought_instruments): data_token = [{"exchange": "NFO", "tradingSymbol": instrument}] mQ = thefirstock.firstock_getMultiQuote(userId=self.user_id, dataToken=data_token) current_price_dict = self.get_tradingsymbol_price_dict(mQ) current_price = current_price_dict.get(instrument, 0.0) pl = (current_price - buy_prices[instrument]) * self.lotCount self.total_pl += pl self.total_trades += 1 if pl > 0: self.wins += 1 self.trades.append((instrument, buy_prices[instrument], current_price, pl)) self.create_order(tickTimeStr, instrument, current_price, 'SELL', self.lotCount) self.logger.info(f"Time-based exit sell order for {instrument} at price {current_price}, P/L: {pl:.2f}") bought_instruments.remove(instrument) break time_module.sleep(2) print('sleep done') data_token_list = [{"exchange": "NFO", "tradingSymbol": symbol} for symbol in bought_instruments] mQ = thefirstock.firstock_getMultiQuote(userId=self.user_id, dataToken=data_token_list) current_price_dict = self.get_tradingsymbol_price_dict(mQ) if self.testing_enabled: if i == 0: current_price_dict['NIFTY02SEP25C24450']= 240 elif i == 1: current_price_dict['NIFTY02SEP25C24450']= 362 self.exit_time = time(15, 0) i = i+1 for instrument in list(bought_instruments): current_price = current_price_dict.get(instrument) if current_price is None: continue # Update highest price for trailing stop self.highest_prices[instrument] = max(self.highest_prices.get(instrument, buy_prices[instrument]), current_price) # Calculate dynamic stop loss and take profit for this instrument profit_level = current_price - buy_prices[instrument] take_profit_level = buy_prices[instrument] + self.profit # Adjust stop loss based on profit level, but only if it improves the stop loss if profit_level >= self.baseProfitLevel: # Calculate additional profit beyond baseProfitLevel additional_profit = profit_level - self.baseProfitLevel # Move stop loss up by additional_profit (1:1 with profit increase beyond baseProfitLevel) new_stop_loss_offset = -self.stop_loss_move_int_level - additional_profit # Only update if the new offset is more favorable (i.e., lower, since negative means higher stop loss price) if instrument not in self.stop_loss_states or new_stop_loss_offset < self.stop_loss_states[instrument]: self.stop_loss_states[instrument] = new_stop_loss_offset stop_loss_level = buy_prices[instrument] - self.stop_loss_states.get(instrument, self.stop_loss) self.logger.info(f"inst: {instrument}, price {current_price:.2f}, stopLoss: {stop_loss_level:.2f}, currentProfit: {profit_level:.2f}, take_profit_level: {take_profit_level:.2f}") if current_price <= stop_loss_level: pl = (current_price - buy_prices[instrument]) * self.lotCount self.total_pl += pl self.total_trades += 1 if pl > 0: self.wins += 1 self.trades.append((instrument, buy_prices[instrument], current_price, pl)) self.create_order(tickTimeStr, instrument, current_price, 'SELL', self.lotCount) self.logger.info(f"Stop loss hit for {instrument} at price {current_price}, P/L: {pl:.2f}") bought_instruments.remove(instrument) elif current_price >= take_profit_level: pl = (current_price - buy_prices[instrument]) * self.lotCount self.total_pl += pl self.total_trades += 1 if pl > 0: self.wins += 1 self.trades.append((instrument, buy_prices[instrument], current_price, pl)) self.create_order(tickTimeStr, instrument, current_price, 'SELL', self.lotCount) self.logger.info(f"Profit target hit for {instrument} at price {current_price}, P/L: {pl:.2f}") bought_instruments.remove(instrument) print('Sell monitoring ...') self.calculate_metrics() 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")