import pandas as pd import os import numpy as np from datetime import datetime import requests import logging class OptionSignalAnalyzer: def __init__(self, input_filename="nifty_consolidated_data_31OCT.csv", createEntries=True): self.input_filename = input_filename self.current_positions = [] # Changed to list to track multiple positions self.trade_log = [] self.createEntries = createEntries self.order_sequence = 1 # Sequence counter for orders self.portfolio_value = 0 # Track P/L self.trade_pnl = [] # Track P/L for each trade self.strikeOffset = -50 # New: Strike price offset self.profitLevel = 9.65 # Profit target in points self.stopLossLevel = 18 # Stop loss in points self.tradeStopTime = "14:45:00" # Stop trading after this time self.last_buy_signals = {} # Track last buy signal for each strike+type self.numEntriesForEntryPattern = 2 # Number of consecutive signals needed for ENTRY pattern matching self.numEntriesForExitPattern = 1 # Number of consecutive signals needed for EXIT pattern matching # Setup logger self.logger = logging.getLogger(__name__) def create_order(self, tickTimeStr, instrument, price, action, lotCount): return """Your provided order creation function with sequence number""" # Add sequence number to timestamp sequence_str = f"{self.order_sequence:04d}" # Format as 0001, 0002, etc. tickTimeStr_with_seq = f"{tickTimeStr}:{sequence_str}" if self.createEntries: url = "http://143.244.141.41/php/createEntries.php" params = { 'tickTime': str(tickTimeStr_with_seq), # Use timestamp with sequence '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}") else: print(f" ✓ Order {sequence_str} created: {action} {instrument} at price {price}") except Exception as e: self.logger.error(f"Order request failed: {e}") # Increment sequence for next order self.order_sequence += 1 def read_consolidated_csv(self): """Read the consolidated CSV file with proper data type handling""" if not os.path.isfile(self.input_filename): print(f"Input file {self.input_filename} not found!") return None try: df = pd.read_csv(self.input_filename) print(f"Successfully read {len(df)} records from {self.input_filename}") # Convert timestamp to datetime for proper sorting df['timestamp'] = pd.to_datetime(df['timestamp']) df = df.sort_values('timestamp').reset_index(drop=True) return df except Exception as e: print(f"Error reading CSV file: {e}") return None def extract_strike_columns(self, df): """Extract strike price columns from the dataframe""" strike_columns = [col for col in df.columns if col.startswith('strike_')] print(f"Found {len(strike_columns)} strike columns") return strike_columns def get_adjusted_strike(self, strike_price, option_type): """Get the adjusted strike price based on option type""" if option_type == 'CE': return int(strike_price) + self.strikeOffset elif option_type == 'PE': return int(strike_price) - self.strikeOffset else: return strike_price def get_price_columns(self, strike_price, option_type): """Get the price column name for a given strike and option type""" adjusted_strike = self.get_adjusted_strike(strike_price, option_type) if option_type == 'CE': return f"ce_{adjusted_strike}_ltp" elif option_type == 'PE': return f"pe_{adjusted_strike}_ltp" return None def get_current_price(self, row, strike_price, option_type): """Get the current price for a given strike and option type from the row USING ADJUSTED STRIKE""" price_column = self.get_price_columns(strike_price, option_type) if price_column and price_column in row: price = row[price_column] if pd.isna(price) or price == 'NA': return None try: return float(price) except (ValueError, TypeError): return None return None def format_timestamp_for_order(self, timestamp): """Format timestamp to string with seconds accuracy""" if isinstance(timestamp, pd.Timestamp): return timestamp.strftime('%Y-%m-%d %H:%M:%S') return str(timestamp) def create_instrument_name(self, strike_price, option_type): """Create instrument name in format NIFTY+strikePrice+CE/PE with strike offset""" # Apply strike offset based on option type adjusted_strike = self.get_adjusted_strike(strike_price, option_type) return f"NIFTY{adjusted_strike}{option_type}" def check_profit_stop_loss(self, timestamp, current_row): """Check if profit target or stop loss is hit for all current positions""" positions_to_close = [] for position in self.current_positions: strike = position.get('strike') option_type = position.get('type') buy_price = position.get('buy_price', 0) # Get current price current_price = self.get_current_price(current_row, strike, option_type) if current_price is None: continue # Calculate unrealized P/L unrealized_pnl = current_price - buy_price # Check profit target if unrealized_pnl >= self.profitLevel: positions_to_close.append((position, f"Profit target reached (+{unrealized_pnl:.2f} points)")) # Check stop loss elif unrealized_pnl <= -self.stopLossLevel: positions_to_close.append((position, f"Stop loss hit ({unrealized_pnl:.2f} points)")) # Close positions that hit profit/stop loss for position, reason in positions_to_close: self.close_single_position(timestamp, current_row, position, reason) return len(positions_to_close) > 0 def is_already_holding_same_position(self, strike_price, option_type): """Check if we're already holding a position for the same strike and option type""" for position in self.current_positions: if (position.get('strike') == strike_price and position.get('type') == option_type): return True return False def check_open_pattern(self, signals, option_type): """Check if the signal pattern matches the opening pattern for the given option type""" if len(signals) != self.numEntriesForEntryPattern + 1: return False # For CE: 0 followed by numEntriesForEntryPattern consecutive 1s if option_type == 'CE': if signals[0] != 0: return False return all(signal == 1 for signal in signals[1:]) # For PE: 1 followed by numEntriesForEntryPattern consecutive 0s elif option_type == 'PE': if signals[0] != 1: return False return all(signal == 0 for signal in signals[1:]) return False def check_close_pattern(self, signals, option_type): """Check if the signal pattern matches the closing pattern for the given option type""" if len(signals) != self.numEntriesForExitPattern + 1: return False # For CE close: 1 followed by numEntriesForExitPattern consecutive 0s if option_type == 'CE': if signals[0] != 1: return False return all(signal == 0 for signal in signals[1:]) # For PE close: 0 followed by numEntriesForExitPattern consecutive 1s elif option_type == 'PE': if signals[0] != 0: return False return all(signal == 1 for signal in signals[1:]) return False def get_required_previous_rows(self): """Get the maximum number of previous rows needed for both entry and exit patterns""" return max(self.numEntriesForEntryPattern, self.numEntriesForExitPattern) def get_signals_for_pattern(self, current_row, previous_rows, strike_col, pattern_type): """Get signals for the specified pattern type (entry or exit)""" if pattern_type == 'entry': num_signals_needed = self.numEntriesForEntryPattern else: # exit num_signals_needed = self.numEntriesForExitPattern signals = [] # Add previous signals (oldest first to current) start_index = len(previous_rows) - num_signals_needed if start_index < 0: return [] # Not enough previous rows for j in range(start_index, len(previous_rows)): signal_val = previous_rows[j][strike_col] if signal_val == 'NA': return [] try: signals.append(int(signal_val)) except (ValueError, TypeError): return [] # Add current signal current_signal = current_row[strike_col] if current_signal == 'NA': return [] try: signals.append(int(current_signal)) except (ValueError, TypeError): return [] return signals def analyze_signal_changes(self, df): """Analyze signal changes and generate buy/sell actions using pattern-based logic for both open and close""" if df is None or df.empty: print("No data to analyze") return strike_columns = self.extract_strike_columns(df) if not strike_columns: print("No strike columns found in the data") return max_pattern_length = self.get_required_previous_rows() print("\n=== ANALYZING SIGNAL CHANGES ===") print(f"Using strike offset: {self.strikeOffset} (CE: +{self.strikeOffset}, PE: -{self.strikeOffset})") print(f"Profit Target: {self.profitLevel} points, Stop Loss: {self.stopLossLevel} points") print(f"Trade Stop Time: {self.tradeStopTime}") print(f"ENTRY Pattern Length: {self.numEntriesForEntryPattern} consecutive signals") print(f"EXIT Pattern Length: {self.numEntriesForExitPattern} consecutive signals") print(f"ENTRY Logic: 0→{','.join(['1']*self.numEntriesForEntryPattern)} for CE | 1→{','.join(['0']*self.numEntriesForEntryPattern)} for PE") print(f"EXIT Logic: 1→{','.join(['0']*self.numEntriesForExitPattern)} for CE | 0→{','.join(['1']*self.numEntriesForExitPattern)} for PE") print("Multiple positions allowed but NO duplicate buys at same strike") # Process each timestamp in chronological order # Start from max_pattern_length to have enough previous rows for both entry and exit patterns for i in range(max_pattern_length, len(df)): current_row = df.iloc[i] # Get previous rows based on maximum pattern length needed previous_rows = [] for j in range(1, max_pattern_length + 1): previous_rows.append(df.iloc[i - j]) timestamp = current_row['timestamp'] print(f"\n--- Timestamp: {timestamp} ---") print(f" Current open positions: {len(self.current_positions)}") # Check if current time is after trade stop time current_time_str = timestamp.strftime('%H:%M:%S') if current_time_str > self.tradeStopTime: print(f" ⏰ Trade stop time {self.tradeStopTime} reached - Closing all open positions and stopping trading") self.close_all_positions(timestamp, current_row, f"Trade stop time {self.tradeStopTime} reached") break # Exit the loop completely # First, check profit/stop loss for current positions if self.current_positions: if self.check_profit_stop_loss(timestamp, current_row): # Some positions were closed due to profit/stop loss pass # Check each strike for pattern-based OPEN and CLOSE signals for strike_col in strike_columns: strike_price = strike_col.replace('strike_', '') # Get signals for ENTRY pattern checking entry_signals = self.get_signals_for_pattern(current_row, previous_rows, strike_col, 'entry') # Get signals for EXIT pattern checking exit_signals = self.get_signals_for_pattern(current_row, previous_rows, strike_col, 'exit') # Skip if we don't have enough valid signals for either pattern if not entry_signals and not exit_signals: continue # CHECK FOR CLOSE PATTERNS FIRST (using exit pattern) if self.current_positions and exit_signals: for position in self.current_positions[:]: # Use slice copy for safe iteration position_strike = position.get('strike') position_type = position.get('type') # Only check close patterns for positions at this strike if strike_price == position_strike: if self.check_close_pattern(exit_signals, position_type): pattern_desc = f"1→{','.join(['0']*self.numEntriesForExitPattern)}" if position_type == 'CE' else f"0→{','.join(['1']*self.numEntriesForExitPattern)}" print(f"Strike {strike_price}: Close pattern {pattern_desc} detected for {position_type} - CLOSING") self.close_single_position(timestamp, current_row, position, f"Close pattern {pattern_desc} detected") # CHECK FOR OPEN PATTERNS (after closing, using entry pattern) if entry_signals: # OPEN CE when pattern: 0 → 1,1,1,... (one 0 followed by numEntriesForEntryPattern consecutive 1s) if self.check_open_pattern(entry_signals, 'CE'): # Check if we already have an open CE position at this strike if not self.is_already_holding_same_position(strike_price, 'CE'): pattern_desc = f"0→{','.join(['1']*self.numEntriesForEntryPattern)}" print(f"Strike {strike_price}: Open pattern {pattern_desc} detected - OPENING CE") self.buy_instrument(strike_price, 'CE', 1, timestamp, current_row) else: pattern_desc = f"0→{','.join(['1']*self.numEntriesForEntryPattern)}" print(f"Strike {strike_price}: Open pattern {pattern_desc} detected but SKIPPING - Already holding CE at this strike") # OPEN PE when pattern: 1 → 0,0,0,... (one 1 followed by numEntriesForEntryPattern consecutive 0s) elif self.check_open_pattern(entry_signals, 'PE'): # Check if we already have an open PE position at this strike if not self.is_already_holding_same_position(strike_price, 'PE'): pattern_desc = f"1→{','.join(['0']*self.numEntriesForEntryPattern)}" print(f"Strike {strike_price}: Open pattern {pattern_desc} detected - OPENING PE") self.buy_instrument(strike_price, 'PE', 0, timestamp, current_row) else: pattern_desc = f"1→{','.join(['0']*self.numEntriesForEntryPattern)}" print(f"Strike {strike_price}: Open pattern {pattern_desc} detected but SKIPPING - Already holding PE at this strike") def close_all_positions(self, timestamp, current_row, reason=""): """Close all current positions""" if not self.current_positions: return print(f" Closing all {len(self.current_positions)} positions: {reason}") # Close all positions (iterate over copy of list) for position in self.current_positions[:]: self.close_single_position(timestamp, current_row, position, reason) def close_single_position(self, timestamp, current_row, position, reason=""): """Close a single position and create SELL order""" if position is None: return strike = position.get('strike', 'UNKNOWN') option_type = position.get('type', 'UNKNOWN') signal = position.get('signal', 'UNKNOWN') buy_price = position.get('buy_price', 0) # Get current sell price from the data USING ADJUSTED STRIKE sell_price = self.get_current_price(current_row, strike, option_type) if sell_price is None: print(f" ⚠ WARNING: No valid price found for {option_type} at ADJUSTED strike, using buy_price {buy_price} as default") sell_price = buy_price # Calculate P/L pnl = sell_price - buy_price self.portfolio_value += pnl # Create SELL order FIRST with adjusted strike tickTimeStr = self.format_timestamp_for_order(timestamp) instrument = self.create_instrument_name(strike, option_type) self.create_order(tickTimeStr, instrument, sell_price, "SELL", 1) # Record P/L for this trade trade_pnl = { 'timestamp': timestamp, 'strike': strike, 'option_type': option_type, 'adjusted_strike': self.get_adjusted_strike(strike, option_type), 'buy_price': buy_price, 'sell_price': sell_price, 'pnl': pnl, 'cumulative_pnl': self.portfolio_value, 'close_reason': reason } self.trade_pnl.append(trade_pnl) action = { 'timestamp': timestamp, 'action': 'CLOSE', 'strike': strike, 'option_type': option_type, 'adjusted_strike': self.get_adjusted_strike(strike, option_type), 'signal': signal, 'buy_price': buy_price, 'sell_price': sell_price, 'pnl': pnl, 'reason': reason } self.trade_log.append(action) # Remove from current positions if position in self.current_positions: self.current_positions.remove(position) # Show adjusted strike in close message adjusted_strike = self.get_adjusted_strike(strike, option_type) print(f" ✓ CLOSE {option_type} at adjusted strike {adjusted_strike} (Original: {strike}) - P/L: {pnl:.2f} - {reason}") def buy_instrument(self, strike_price, option_type, signal, timestamp, current_row): """Buy a new instrument and create BUY order (without closing existing same-type positions)""" # Get current buy price from the data USING ADJUSTED STRIKE buy_price = self.get_current_price(current_row, strike_price, option_type) if buy_price is None: print(f" ⚠ WARNING: No valid price found for {option_type} at ADJUSTED strike, using 100 as default") buy_price = 100 # Create BUY order with adjusted strike tickTimeStr = self.format_timestamp_for_order(timestamp) instrument = self.create_instrument_name(strike_price, option_type) self.create_order(tickTimeStr, instrument, buy_price, "BUY", 1) action = { 'timestamp': timestamp, 'action': 'BUY', 'strike': strike_price, 'option_type': option_type, 'adjusted_strike': self.get_adjusted_strike(strike_price, option_type), 'signal': signal, 'buy_price': buy_price, 'reason': f"Pattern detected" } self.trade_log.append(action) # Create new position and add to current positions (allow multiple) new_position = { 'strike': strike_price, 'type': option_type, 'signal': signal, 'buy_price': buy_price } self.current_positions.append(new_position) # Show adjusted strike information adjusted_strike = self.get_adjusted_strike(strike_price, option_type) if option_type == 'CE': print(f" ✓ BUY {option_type} at adjusted strike {adjusted_strike} (Original: {strike_price}) at price {buy_price:.2f}") print(f" → Strike adjustment: {strike_price} → {adjusted_strike} (+{self.strikeOffset})") else: # PE print(f" ✓ BUY {option_type} at adjusted strike {adjusted_strike} (Original: {strike_price}) at price {buy_price:.2f}") print(f" → Strike adjustment: {strike_price} → {adjusted_strike} (-{self.strikeOffset})") print(f" Total open positions: {len(self.current_positions)}") def safe_get_position_info(self): """Safely get current position information""" if not self.current_positions: return "No positions" position_info = [] for pos in self.current_positions: strike = pos.get('strike', 'UNKNOWN') option_type = pos.get('type', 'UNKNOWN') signal = pos.get('signal', 'UNKNOWN') buy_price = pos.get('buy_price', 'UNKNOWN') adjusted_strike = self.get_adjusted_strike(strike, option_type) position_info.append(f"{option_type}@{adjusted_strike}({strike})") return ", ".join(position_info) def print_trade_summary(self): """Print all buy/sell actions with P/L""" print("\n" + "="*80) print("TRADE ACTION SUMMARY WITH P/L") print("="*80) if not self.trade_log: print("No trade actions generated.") return # Group trades by timestamp to show close-open sequences clearly current_timestamp = None for i, trade in enumerate(self.trade_log, 1): if trade['timestamp'] != current_timestamp: current_timestamp = trade['timestamp'] print(f"\n--- {current_timestamp} ---") if trade['action'] == 'BUY': strike_info = f"{trade['strike']}→{trade.get('adjusted_strike', 'N/A')}" print(f"{i:2d}. {trade['action']:5s} {trade['option_type']:2s} " f"at strike {strike_info:10s} (Signal: {trade['signal']}) - Price: {trade.get('buy_price', 'N/A'):.2f}") else: # CLOSE action strike_info = f"{trade['strike']}→{trade.get('adjusted_strike', 'N/A')}" print(f"{i:2d}. {trade['action']:5s} {trade['option_type']:2s} " f"at strike {strike_info:10s} (Signal: {trade['signal']}) - " f"Buy: {trade.get('buy_price', 'N/A'):.2f}, Sell: {trade.get('sell_price', 'N/A'):.2f}, " f"P/L: {trade.get('pnl', 'N/A'):.2f} - Reason: {trade.get('reason', 'N/A')}") print(f"\nTotal trade actions: {len(self.trade_log)}") print(f"Total orders created: {self.order_sequence - 1}") print(f"Strike offset applied: {self.strikeOffset} (CE: +{self.strikeOffset}, PE: -{self.strikeOffset})") print(f"Profit Target: {self.profitLevel} points, Stop Loss: {self.stopLossLevel} points") print(f"Trade Stop Time: {self.tradeStopTime}") print(f"ENTRY Pattern Length: {self.numEntriesForEntryPattern} consecutive signals") print(f"EXIT Pattern Length: {self.numEntriesForExitPattern} consecutive signals") # Calculate summary statistics buy_actions = [t for t in self.trade_log if t['action'] == 'BUY'] close_actions = [t for t in self.trade_log if t['action'] == 'CLOSE'] ce_buys = [t for t in buy_actions if t['option_type'] == 'CE'] pe_buys = [t for t in buy_actions if t['option_type'] == 'PE'] print(f"BUY actions: {len(buy_actions)} (CE: {len(ce_buys)}, PE: {len(pe_buys)})") print(f"CLOSE actions: {len(close_actions)}") # P/L Summary total_pnl = sum(trade.get('pnl', 0) for trade in self.trade_log if trade['action'] == 'CLOSE') winning_trades = len([t for t in self.trade_log if t.get('pnl', 0) > 0 and t['action'] == 'CLOSE']) losing_trades = len([t for t in self.trade_log if t.get('pnl', 0) < 0 and t['action'] == 'CLOSE']) # Analyze close reasons profit_closes = len([t for t in self.trade_log if 'Profit target' in t.get('reason', '')]) stop_loss_closes = len([t for t in self.trade_log if 'Stop loss' in t.get('reason', '')]) pattern_closes = len([t for t in self.trade_log if 'Close pattern' in t.get('reason', '')]) time_stop_closes = len([t for t in self.trade_log if 'Trade stop time' in t.get('reason', '')]) other_closes = len(close_actions) - profit_closes - stop_loss_closes - pattern_closes - time_stop_closes print(f"\n=== P/L SUMMARY ===") print(f"Total P/L: {total_pnl:.2f}") print(f"Winning Trades: {winning_trades}") print(f"Losing Trades: {losing_trades}") print(f"Win Rate: {winning_trades/len(close_actions)*100:.1f}%" if close_actions else "Win Rate: N/A") print(f"Final Portfolio Value: {self.portfolio_value:.2f}") print(f"\n=== CLOSE REASONS ===") print(f"Profit Target Hits: {profit_closes}") print(f"Stop Loss Hits: {stop_loss_closes}") print(f"Pattern Closes: {pattern_closes}") print(f"Time Stop ({self.tradeStopTime}): {time_stop_closes}") print(f"Other Reasons: {other_closes}") # Show final positions position_info = self.safe_get_position_info() print(f"\nFINAL POSITIONS: {position_info}") def export_trade_log(self, output_filename="trade_actions.csv"): """Export trade log to CSV""" if not self.trade_log: print("No trade actions to export") return df_trades = pd.DataFrame(self.trade_log) df_trades.to_csv(output_filename, index=False) print(f"Trade actions exported to {output_filename}") # Export P/L data separately if self.trade_pnl: df_pnl = pd.DataFrame(self.trade_pnl) df_pnl.to_csv("trade_pnl.csv", index=False) print(f"P/L data exported to trade_pnl.csv") def analyze(self): """Main analysis method""" print(f"Analyzing signal changes from {self.input_filename}...") # Reset sequence and portfolio for new analysis self.order_sequence = 1 self.portfolio_value = 0 self.trade_pnl = [] self.current_positions = [] # Reset positions self.last_buy_signals = {} # Reset buy signals # Read consolidated CSV df = self.read_consolidated_csv() if df is None: return # Print data overview print(f"Data period: {df['timestamp'].min()} to {df['timestamp'].max()}") print(f"Total timestamps: {len(df)}") # Analyze signal changes self.analyze_signal_changes(df) # Print final summary self.print_trade_summary() # Export to CSV self.export_trade_log() def main(): """Main function to run the analysis""" analyzer = OptionSignalAnalyzer(createEntries=True) # You can change the pattern lengths here: # analyzer.numEntriesForEntryPattern = 4 # For 0→1,1,1,1 or 1→0,0,0,0 entry patterns # analyzer.numEntriesForExitPattern = 3 # For 1→0,0,0 or 0→1,1,1 exit patterns analyzer.analyze() if __name__ == "__main__": main()