""" Tier-0 Noise Rejection Service Filters out low-quality signals before enrichment to reduce processing overhead """ import pandas as pd import numpy as np from typing import Optional from datetime import timedelta from utils.logger import logger from utils.error_handler import safe_divide class NoiseRejector: """ Rejects early-stage noise before enrichment to optimize processing. Rejects if: - Single isolated trade (no repeat activity) - Far OTM weekly lottos - Delta-adjusted premium below threshold """ def __init__(self): # Configuration self.repeat_activity_window_minutes = 30 # Minutes to look for repeat activity self.min_delta_adjusted_premium = 50000 # Minimum delta-adjusted premium self.max_otm_percentage = 15.0 # Reject strikes >15% OTM self.min_expiry_days = 1 # Minimum days to expiry (reject same-day/weekly lottos) def is_isolated_trade( self, df: pd.DataFrame, row_idx: int ) -> bool: """ Check if trade is isolated (no repeat activity within window). Returns True if isolated (should reject). """ if row_idx >= len(df): return True current_row = df.iloc[row_idx] symbol = current_row.get('symbol_norm') direction = current_row.get('direction') flow_ts_utc = current_row.get('flow_ts_utc') if pd.isna(symbol) or pd.isna(flow_ts_utc): return True # Look for trades in same direction within window window_start = flow_ts_utc - timedelta(minutes=self.repeat_activity_window_minutes) same_symbol = df['symbol_norm'] == symbol same_direction = df['direction'] == direction in_window = (df['flow_ts_utc'] >= window_start) & (df['flow_ts_utc'] < flow_ts_utc) # Exclude current row other_trades = df[same_symbol & same_direction & in_window] # If no other trades in window, it's isolated return len(other_trades) == 0 def is_far_otm_lotto( self, row: pd.Series ) -> bool: """ Check if trade is a far OTM weekly lotto (low probability, low signal value). Returns True if should reject. """ spot_num = row.get('spot_num', 0) or 0 strike_num = row.get('strike_num', 0) or 0 cp_norm = row.get('cp_norm', '') exp_date = row.get('exp_date') flow_date = row.get('flow_date_cst') if pd.isna(spot_num) or spot_num == 0 or pd.isna(strike_num) or pd.isna(cp_norm): return False # Calculate moneyness percentage if cp_norm == 'CALL': otm_pct = ((strike_num - spot_num) / spot_num * 100.0) if spot_num > 0 else 0 else: # PUT otm_pct = ((spot_num - strike_num) / spot_num * 100.0) if spot_num > 0 else 0 # Reject if >15% OTM if otm_pct > self.max_otm_percentage: return True # Check if weekly lotto (expiry within 7 days) if pd.notna(exp_date) and pd.notna(flow_date): try: if isinstance(exp_date, str): from datetime import datetime exp_date = datetime.strptime(exp_date, '%Y-%m-%d').date() if isinstance(flow_date, str): from datetime import datetime flow_date = datetime.strptime(flow_date, '%Y-%m-%d').date() days_to_expiry = (exp_date - flow_date).days # Reject weekly lottos that are also far OTM if days_to_expiry <= 7 and otm_pct > 10.0: return True except (ValueError, TypeError, AttributeError): pass return False def calculate_delta_adjusted_premium( self, row: pd.Series ) -> float: """ Calculate delta-adjusted premium (premium * |delta|). Approximates intrinsic value component of premium. For simplicity, we estimate delta from moneyness: - ATM: delta ≈ 0.5 - ITM: delta increases toward 1.0 - OTM: delta decreases toward 0.0 """ premium_num = row.get('premium_num', 0) or 0 spot_num = row.get('spot_num', 0) or 0 strike_num = row.get('strike_num', 0) or 0 cp_norm = row.get('cp_norm', '') if pd.isna(premium_num) or premium_num == 0 or pd.isna(spot_num) or spot_num == 0: return 0.0 # Estimate delta from moneyness if cp_norm == 'CALL': if strike_num <= spot_num: # ITM call: delta from 0.5 to 1.0 moneyness_ratio = strike_num / spot_num if spot_num > 0 else 1.0 # Strike at spot = 0.5, strike at 0 = 1.0 estimated_delta = 0.5 + (1.0 - moneyness_ratio) * 0.5 else: # OTM call: delta from 0.5 to 0.0 moneyness_ratio = strike_num / spot_num if spot_num > 0 else 1.0 # Strike at spot = 0.5, strike at 1.15x spot = 0.0 estimated_delta = max(0.0, 0.5 * (1.5 - moneyness_ratio) / 0.5) else: # PUT if strike_num >= spot_num: # ITM put: delta from -0.5 to -1.0 (use absolute value) moneyness_ratio = strike_num / spot_num if spot_num > 0 else 1.0 estimated_delta = 0.5 + (moneyness_ratio - 1.0) * 0.5 else: # OTM put: delta from 0.5 to 0.0 (use absolute value) moneyness_ratio = strike_num / spot_num if spot_num > 0 else 1.0 estimated_delta = max(0.0, 0.5 * (1.0 - moneyness_ratio) / 0.5) # Delta-adjusted premium delta_adj_premium = premium_num * abs(estimated_delta) return float(delta_adj_premium) def should_reject( self, df: pd.DataFrame, row_idx: int ) -> bool: """ Determine if row should be rejected as noise. Returns True if should reject (mark early_noise_reject = True). """ if row_idx >= len(df): return True row = df.iloc[row_idx] # Reject isolated trades if self.is_isolated_trade(df, row_idx): return True # Reject far OTM lottos if self.is_far_otm_lotto(row): return True # Reject if delta-adjusted premium below threshold delta_adj_premium = self.calculate_delta_adjusted_premium(row) if delta_adj_premium < self.min_delta_adjusted_premium: return True return False def mark_noise_rejections(self, df: pd.DataFrame) -> pd.DataFrame: """ Mark rows for early noise rejection. Adds: early_noise_reject (boolean) """ if df.empty: return df df = df.copy() # Initialize column df['early_noise_reject'] = False # Sort by timestamp for proper isolation detection df_sorted = df.sort_values('flow_ts_utc').reset_index(drop=True) # Mark rejections for idx in df_sorted.index: if self.should_reject(df_sorted, idx): original_idx = df_sorted.index[idx] df.at[original_idx, 'early_noise_reject'] = True rejection_count = df['early_noise_reject'].sum() logger.info(f"Noise rejection: {rejection_count}/{len(df)} rows marked for rejection ({rejection_count/len(df)*100:.1f}%)") return df