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 firstock import firstock 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 = 1 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 = 60 # Base profit level to trigger stop loss adjustment self.stop_loss = 30 # Initial stop loss offset from buy price self.stop_loss_move_int_level = self.baseProfitLevel - 20 # 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, 30) 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 = "04NOV25" self.testing_enabled = False if self.testing_enabled: self.buy_start_range = 110 self.buy_end_range = 190 self.retry_time = time(20, 30) self.exit_time = time(22, 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 = 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["tradingSymbol"]: float(item["lastTradedPrice"]) for item in mQ["data"] if "tradingSymbol" in item and "lastTradedPrice" in item } def calculate_metrics(self): print("") self.logger.info("Stage 3: All trades completed. Summary:") 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): print("") self.logger.info("Create Entries configuration : "+ str(self.createEntries)) self.logger.info(f"Test Mode Enabled : "+ str(self.testing_enabled)) self.logger.info(f"Buy Range: {self.buy_start_range} - {self.buy_end_range}") self.logger.info(f"Program Start Time: {self.pgm_start_time}, Retry Until: {self.retry_time}, Exit Time: {self.exit_time}") print("") 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) print("") self.logger.info(f"Stage 0: Starting Buy Loop") try: gQ = 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: print("") self.logger.info(f"Stage 1: Not bought any instruments yet ...") current_time = datetime.now(self.ist) try: mQ = firstock.getMultiQuotes( 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: time_module.sleep(1) # To avoid hitting API rate limits now_ist = datetime.now(self.ist) start_time = (now_ist - timedelta(minutes=4)).replace(second=0, microsecond=0) end_time = (now_ist - timedelta(minutes=1)).replace(second=0, microsecond=0) if self.testing_enabled: start_time = (now_ist - timedelta(minutes=5)).replace(hour=9, minute=30, second=0, microsecond=0) end_time = (now_ist - timedelta(minutes=1)).replace(hour=9, minute=33, second=0, microsecond=0) self.logger.info(f"Fetching timePriceSeries for {instrument} from {start_time.strftime("%H:%M:%S %d-%m-%Y")} to {end_time.strftime("%H:%M:%S %d-%m-%Y")}") response = firstock.timePriceSeries( userId=self.user_id, exchange='NFO', tradingSymbol=instrument, startTime=start_time.strftime("%H:%M:%S %d-%m-%Y"), endTime=end_time.strftime("%H:%M:%S %d-%m-%Y"), interval=str(self.tick_interval)+"mi" ) #pprint.pprint(response) intc_values = [round(float(item['close']), 2) for item in response['data']] self.logger.info(f"Checking {instrument} at price {price}. Previous Prices : {intc_values}") if all(value < self.buy_end_range for value in intc_values) and intc_values[0] < intc_values[-1]: 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 else: self.logger.info(f"Rejected {instrument} as some price above {self.buy_end_range}. Previous Prices : {intc_values}") if not bought_instruments: self.logger.info(f"No options found in {self.buy_start_range}-{self.buy_end_range}, waiting until {self.retry_time}...") time_module.sleep(1) if not bought_instruments: self.logger.info(f"No options found in {self.buy_start_range}-{self.buy_end_range} range by {self.retry_time}, exiting program.") return if self.testing_enabled: i =0 # Monitoring loop while bought_instruments: print("") self.logger.info(f"Stage 2: Bought instruments, monitoring for exit ...") 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 = firstock.getMultiQuotes(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) data_token_list = [{"exchange": "NFO", "tradingSymbol": symbol} for symbol in bought_instruments] mQ = firstock.getMultiQuotes(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['NIFTY07OCT25C24750']= 400 elif i == 1: current_price_dict['NIFTY07OCT25C24800']= 450 #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) self.calculate_metrics() if __name__ == "__main__": client_details = get_client_details() if client_details: createEntries = len(sys.argv) > 1 trader = Class180(client_details, createEntries) trader.login() trader.fetch_and_process_data() print("done") else: print("Failed to get client details")