"""
Options Flow Processor
Converts the complex SQL logic from optionflowrockerscorer.sql to Python/pandas
"""
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Tuple
import pytz
import re
from utils.logger import logger
from utils.error_handler import handle_processing_error, validate_dataframe, safe_divide


class OptionsFlowProcessor:
    """Process options flow data with complex analytics"""
    
    def __init__(self, tol_pct: float = 0.20):
        self.tol_pct = tol_pct
        self.ct_tz = pytz.timezone('America/Chicago')
        self.utc_tz = pytz.UTC
    
    def normalize_call_put(self, value: str) -> Optional[str]:
        """Normalize CallPut field"""
        if pd.isna(value):
            return None
        val = str(value).upper().strip()
        if val in ('C', 'CALL', 'CALLS', 'CE'):
            return 'CALL'
        elif val in ('P', 'PUT', 'PUTS', 'PE'):
            return 'PUT'
        return None
    
    def normalize_side(self, value: str) -> Optional[str]:
        """Normalize Side field"""
        if pd.isna(value):
            return None
        val = str(value).upper().strip()
        if val in ('A', 'AA', 'ASK', 'BUY', 'BOT', 'BTO', 'AT_ASK'):
            return 'BUY'
        elif val in ('B', 'BB', 'BID', 'SELL', 'SLD', 'STO', 'AT_BID'):
            return 'SELL'
        return None
    
    def clean_numeric(self, value) -> Optional[float]:
        """Clean and convert numeric field"""
        if pd.isna(value):
            return None
        # Convert to string, remove whitespace, $, commas
        text = str(value).strip()
        text = re.sub(r'[\s$,]', '', text)
        if not text:
            return None
        try:
            return float(text)
        except (ValueError, TypeError):
            return None
    
    def parse_date(self, date_str) -> Optional[datetime]:
        """Parse date string (supports YYYY-MM-DD or MM/DD/YYYY)"""
        if pd.isna(date_str):
            return None
        
        date_str = str(date_str).strip()
        
        # Try YYYY-MM-DD
        if re.match(r'^\d{4}-\d{2}-\d{2}$', date_str):
            try:
                return datetime.strptime(date_str, '%Y-%m-%d')
            except ValueError:
                pass
        
        # Try MM/DD/YYYY
        if re.match(r'^\d{1,2}/\d{1,2}/\d{2,4}$', date_str):
            try:
                return datetime.strptime(date_str, '%m/%d/%Y')
            except ValueError:
                try:
                    return datetime.strptime(date_str, '%m/%d/%y')
                except ValueError:
                    pass
        
        return None
    
    def parse_timestamp(self, date_str, time_str) -> Optional[datetime]:
        """Parse timestamp from date and time strings"""
        if pd.isna(date_str) or pd.isna(time_str):
            return None
        
        date_str = str(date_str).strip()
        time_str = str(time_str).strip()
        
        # Try various formats
        formats = [
            ('%Y-%m-%d', '%I:%M:%S %p'),
            ('%Y-%m-%d', '%I:%M %p'),
            ('%Y-%m-%d', '%H:%M:%S'),
            ('%m/%d/%Y', '%I:%M:%S %p'),
            ('%m/%d/%Y', '%I:%M %p'),
            ('%m/%d/%Y', '%H:%M:%S'),
        ]
        
        for date_fmt, time_fmt in formats:
            try:
                date_obj = datetime.strptime(date_str, date_fmt)
                time_obj = datetime.strptime(time_str, time_fmt).time()
                return datetime.combine(date_obj.date(), time_obj)
            except (ValueError, TypeError):
                continue
        
        return None
    
    def process_base(self, df: pd.DataFrame) -> pd.DataFrame:
        """Step 1: Normalize fields and clean numerics"""
        df = df.copy()
        
        # Filter initial data
        df = df[
            df['Premium'].notna() &
            (df['Premium'].astype(str).str.strip() != '') &
            (df['StockEtf'] == 'STOCK') &
            (~df['Symbol'].isin(['TSLA', 'NVDA']))
        ].copy()
        
        # Add row ID (using index as stable identifier)
        df['rid'] = df.index
        
        # Normalize symbol
        df['symbol_norm'] = df['Symbol'].astype(str).str.upper().str.strip()
        
        # Normalize CallPut
        df['cp_norm'] = df['CallPut'].apply(self.normalize_call_put)
        
        # Normalize Side
        df['side_norm'] = df['Side'].apply(self.normalize_side)
        
        # Clean numerics
        df['strike_num'] = df['Strike'].apply(self.clean_numeric)
        df['spot_num'] = df['Spot'].apply(self.clean_numeric)
        df['premium_num'] = df['Premium'].apply(self.clean_numeric)
        df['vol_num'] = df['Volume'].apply(self.clean_numeric)
        df['oi_num'] = df['OI'].apply(self.clean_numeric)
        
        # Parse expiration date
        df['exp_date'] = df['ExpirationDate'].apply(self.parse_date)
        
        # Parse flow timestamp
        df['flow_ts_local'] = df.apply(
            lambda row: self.parse_timestamp(row['CreatedDate'], row['CreatedTime']),
            axis=1
        )
        
        # Debug: Check how many timestamps were parsed successfully
        parsed_count = df['flow_ts_local'].notna().sum()
        logger.info(f"Parsed timestamps: {parsed_count} out of {len(df)} rows")
        
        if parsed_count > 0:
            # Show date range of parsed data
            min_date = df['flow_ts_local'].min()
            max_date = df['flow_ts_local'].max()
            logger.info(f"Parsed date range: {min_date} to {max_date}")
        
        if parsed_count == 0 and len(df) > 0:
            # Show sample of what we're trying to parse
            sample_row = df.iloc[0]
            logger.warning(f"Failed to parse timestamps. Sample row:")
            logger.warning(f"  CreatedDate: {sample_row.get('CreatedDate')} (type: {type(sample_row.get('CreatedDate'))})")
            logger.warning(f"  CreatedTime: {sample_row.get('CreatedTime')} (type: {type(sample_row.get('CreatedTime'))})")
            # Show unique date values to help debug
            unique_dates = df['CreatedDate'].unique()[:10]
            logger.warning(f"  Sample CreatedDate values: {unique_dates.tolist()}")
        
        return df
    
    def process_flow(self, df: pd.DataFrame, start_day: datetime, end_day: datetime) -> pd.DataFrame:
        """Step 2: Filter by date window and compute UTC"""
        df = df.copy()
        
        if df.empty or 'flow_ts_local' not in df.columns:
            logger.warning("Empty DataFrame or missing flow_ts_local column")
            return df
        
        # Debug: Check flow_ts_local values
        null_timestamps = df['flow_ts_local'].isna().sum()
        logger.info(f"flow_ts_local: {len(df)} total rows, {null_timestamps} null timestamps")
        if null_timestamps < len(df):
            sample_ts = df['flow_ts_local'].dropna().iloc[0] if len(df['flow_ts_local'].dropna()) > 0 else None
            logger.info(f"Sample flow_ts_local value: {sample_ts} (type: {type(sample_ts)})")
        
        # Filter by date window
        # Handle null timestamps gracefully
        df['flow_date_cst'] = pd.to_datetime(df['flow_ts_local'], errors='coerce').dt.date
        
        # Filter rows within date range
        start_date = start_day.date() if isinstance(start_day, datetime) else start_day
        end_date = end_day.date() if isinstance(end_day, datetime) else end_day
        
        logger.info(f"Filtering by date range: {start_date} to {end_date}")
        
        # Only filter if we have valid dates
        valid_dates = df['flow_date_cst'].notna()
        valid_count = valid_dates.sum()
        logger.info(f"Valid dates: {valid_count} out of {len(df)} rows")
        
        if valid_dates.any():
            # Show sample of dates we're comparing
            if valid_count > 0:
                sample_dates = df[valid_dates]['flow_date_cst'].head(5).tolist()
                logger.info(f"Sample flow_date_cst values: {sample_dates}")
            
            date_mask = valid_dates & (df['flow_date_cst'] >= start_date) & (df['flow_date_cst'] <= end_date)
            matching_count = date_mask.sum()
            logger.info(f"Rows matching date range: {matching_count} out of {valid_count} valid dates")
            df = df[date_mask].copy()
        else:
            # No valid dates, return empty
            logger.warning(f"No valid dates found in flow_ts_local. Date range: {start_date} to {end_date}")
            logger.warning(f"First few flow_ts_local values: {df['flow_ts_local'].head(10).tolist()}")
            df = df.iloc[0:0].copy()
            return df
        
        if df.empty:
            logger.warning(f"No data in date range {start_date} to {end_date}")
            logger.warning(f"Date range might not match data. Check if data exists for this date.")
            return df
        
        # Convert to UTC (assuming flow_ts_local is in CST)
        def convert_to_utc(ts_local):
            if pd.isna(ts_local) or ts_local is None:
                return None
            if isinstance(ts_local, datetime):
                if ts_local.tzinfo is None:
                    # Naive datetime, assume CST
                    return self.ct_tz.localize(ts_local).astimezone(self.utc_tz)
                else:
                    # Already timezone-aware, convert to UTC
                    return ts_local.astimezone(self.utc_tz)
            return None
        
        df['flow_ts_utc'] = df['flow_ts_local'].apply(convert_to_utc)
        
        return df
    
    def process_moneyness(self, df: pd.DataFrame) -> pd.DataFrame:
        """Step 3: Calculate direction and moneyness"""
        df = df.copy()
        
        # Moneyness
        def calc_moneyness(row):
            if pd.isna(row['cp_norm']) or pd.isna(row['strike_num']) or pd.isna(row['spot_num']):
                return None
            if row['cp_norm'] == 'CALL':
                return 'OTM' if row['strike_num'] > row['spot_num'] else 'ITM'
            else:  # PUT
                return 'OTM' if row['strike_num'] < row['spot_num'] else 'ITM'
        
        df['moneyness'] = df.apply(calc_moneyness, axis=1)
        
        # Direction
        def calc_direction(row):
            if pd.isna(row['cp_norm']) or pd.isna(row['side_norm']):
                return None
            if (row['cp_norm'] == 'CALL' and row['side_norm'] == 'BUY') or \
               (row['cp_norm'] == 'PUT' and row['side_norm'] == 'SELL'):
                return 'BULL'
            elif (row['cp_norm'] == 'PUT' and row['side_norm'] == 'BUY') or \
                 (row['cp_norm'] == 'CALL' and row['side_norm'] == 'SELL'):
                return 'BEAR'
            return None
        
        df['direction'] = df.apply(calc_direction, axis=1)
        
        # Moneyness percentage
        def calc_mny_pct(row):
            spot = row.get('spot_num')
            strike = row.get('strike_num')
            cp = row.get('cp_norm')
            
            if pd.isna(spot) or spot == 0 or pd.isna(strike) or pd.isna(cp):
                return None
            
            if cp == 'CALL':
                return safe_divide((spot - strike) * 100.0, spot)
            else:  # PUT
                return safe_divide((strike - spot) * 100.0, spot)
        
        df['mny_pct'] = df.apply(calc_mny_pct, axis=1)
        
        return df
    
    def process_aggregations(self, df: pd.DataFrame) -> pd.DataFrame:
        """Step 4: Calculate running sums and aggregations (complete window functions)"""
        df = df.copy()
        
        # Sort for window functions
        df = df.sort_values(['exp_date', 'symbol_norm', 'flow_ts_utc', 'rid']).reset_index(drop=True)
        
        # Fill NaN values for calculations
        df['premium_num'] = df['premium_num'].fillna(0)
        df['vol_num'] = df['vol_num'].fillna(0)
        df['oi_num'] = df['oi_num'].fillna(0)
        
        # Group by exp_date and symbol_norm for window functions
        groups = df.groupby(['exp_date', 'symbol_norm'], group_keys=False)
        
        # Premium aggregations - all 8 combinations
        def calc_prem_value(row, cp, side, moneyness):
            if (row['cp_norm'] == cp and 
                row['side_norm'] == side and 
                row['moneyness'] == moneyness):
                return row['premium_num']
            return 0.0
        
        # CALL BUY OTM
        df['prem_cb_otm'] = groups.apply(
            lambda g: g.apply(lambda row: calc_prem_value(row, 'CALL', 'BUY', 'OTM'), axis=1).cumsum()
        ).values
        
        # CALL BUY ITM
        df['prem_cb_itm'] = groups.apply(
            lambda g: g.apply(lambda row: calc_prem_value(row, 'CALL', 'BUY', 'ITM'), axis=1).cumsum()
        ).values
        
        # CALL SELL OTM
        df['prem_cs_otm'] = groups.apply(
            lambda g: g.apply(lambda row: calc_prem_value(row, 'CALL', 'SELL', 'OTM'), axis=1).cumsum()
        ).values
        
        # CALL SELL ITM
        df['prem_cs_itm'] = groups.apply(
            lambda g: g.apply(lambda row: calc_prem_value(row, 'CALL', 'SELL', 'ITM'), axis=1).cumsum()
        ).values
        
        # PUT BUY OTM
        df['prem_pb_otm'] = groups.apply(
            lambda g: g.apply(lambda row: calc_prem_value(row, 'PUT', 'BUY', 'OTM'), axis=1).cumsum()
        ).values
        
        # PUT BUY ITM
        df['prem_pb_itm'] = groups.apply(
            lambda g: g.apply(lambda row: calc_prem_value(row, 'PUT', 'BUY', 'ITM'), axis=1).cumsum()
        ).values
        
        # PUT SELL OTM
        df['prem_ps_otm'] = groups.apply(
            lambda g: g.apply(lambda row: calc_prem_value(row, 'PUT', 'SELL', 'OTM'), axis=1).cumsum()
        ).values
        
        # PUT SELL ITM
        df['prem_ps_itm'] = groups.apply(
            lambda g: g.apply(lambda row: calc_prem_value(row, 'PUT', 'SELL', 'ITM'), axis=1).cumsum()
        ).values
        
        # Volume and OI aggregations
        df['vol_all'] = groups['vol_num'].transform(lambda x: x.cumsum())
        df['oi_all'] = groups['oi_num'].transform(lambda x: x.cumsum())
        
        # Bull/Bear volume
        def calc_bull_vol(row):
            if ((row['cp_norm'] == 'CALL' and row['side_norm'] == 'BUY') or
                (row['cp_norm'] == 'PUT' and row['side_norm'] == 'SELL')):
                return row['vol_num']
            return 0.0
        
        df['bull_vol'] = groups.apply(
            lambda g: g.apply(calc_bull_vol, axis=1).cumsum()
        ).values
        
        def calc_bear_vol(row):
            if ((row['cp_norm'] == 'PUT' and row['side_norm'] == 'BUY') or
                (row['cp_norm'] == 'CALL' and row['side_norm'] == 'SELL')):
                return row['vol_num']
            return 0.0
        
        df['bear_vol'] = groups.apply(
            lambda g: g.apply(calc_bear_vol, axis=1).cumsum()
        ).values
        
        # Bull/Bear OI
        def calc_bull_oi(row):
            if ((row['cp_norm'] == 'CALL' and row['side_norm'] == 'BUY') or
                (row['cp_norm'] == 'PUT' and row['side_norm'] == 'SELL')):
                return row['oi_num']
            return 0.0
        
        df['bull_oi'] = groups.apply(
            lambda g: g.apply(calc_bull_oi, axis=1).cumsum()
        ).values
        
        def calc_bear_oi(row):
            if ((row['cp_norm'] == 'PUT' and row['side_norm'] == 'BUY') or
                (row['cp_norm'] == 'CALL' and row['side_norm'] == 'SELL')):
                return row['oi_num']
            return 0.0
        
        df['bear_oi'] = groups.apply(
            lambda g: g.apply(calc_bear_oi, axis=1).cumsum()
        ).values
        
        # OI for badges (CALL BUY OTM and PUT BUY OTM)
        def calc_oi_cb_otm(row):
            if row['cp_norm'] == 'CALL' and row['side_norm'] == 'BUY' and row['moneyness'] == 'OTM':
                return row['oi_num']
            return 0.0
        
        df['oi_cb_otm'] = groups.apply(
            lambda g: g.apply(calc_oi_cb_otm, axis=1).cumsum()
        ).values
        
        def calc_oi_pb_otm(row):
            if row['cp_norm'] == 'PUT' and row['side_norm'] == 'BUY' and row['moneyness'] == 'OTM':
                return row['oi_num']
            return 0.0
        
        df['oi_pb_otm'] = groups.apply(
            lambda g: g.apply(calc_oi_pb_otm, axis=1).cumsum()
        ).values
        
        # Direction count (within exp_date, symbol_norm, direction)
        # Need to group by exp_date, symbol_norm, and direction for this count
        # Filter out rows with None direction to avoid grouping issues
        valid_direction = df['direction'].notna()
        if valid_direction.any():
            dir_groups = df[valid_direction].groupby(['exp_date', 'symbol_norm', 'direction'], group_keys=False)
            df.loc[valid_direction, 'dir_count'] = dir_groups.cumcount() + 1
            df.loc[~valid_direction, 'dir_count'] = 0
        else:
            df['dir_count'] = 0
        
        return df
    
    def process_badges(self, df: pd.DataFrame) -> pd.DataFrame:
        """Step 5: Calculate badges and decorations (complete implementation)"""
        df = df.copy()
        
        # Fill NaN values
        df['prem_cb_itm'] = df['prem_cb_itm'].fillna(0)
        df['prem_ps_itm'] = df['prem_ps_itm'].fillna(0)
        df['prem_cs_itm'] = df['prem_cs_itm'].fillna(0)
        df['prem_pb_itm'] = df['prem_pb_itm'].fillna(0)
        df['prem_cb_otm'] = df['prem_cb_otm'].fillna(0)
        df['prem_ps_otm'] = df['prem_ps_otm'].fillna(0)
        df['prem_pb_otm'] = df['prem_pb_otm'].fillna(0)
        df['prem_cs_otm'] = df['prem_cs_otm'].fillna(0)
        df['bull_vol'] = df['bull_vol'].fillna(0)
        df['bear_vol'] = df['bear_vol'].fillna(0)
        df['bull_oi'] = df['bull_oi'].fillna(0)
        df['bear_oi'] = df['bear_oi'].fillna(0)
        df['oi_cb_otm'] = df['oi_cb_otm'].fillna(0)
        df['oi_pb_otm'] = df['oi_pb_otm'].fillna(0)
        
        # Calculate net values
        df['net_vol_raw'] = df['bull_vol'] - df['bear_vol']
        df['net_oi_raw'] = df['bull_oi'] - df['bear_oi']
        
        # Calculate premium totals
        df['bull_prem_itm'] = df['prem_cb_itm'] + df['prem_ps_itm']
        df['bear_prem_itm'] = df['prem_cs_itm'] + df['prem_pb_itm']
        df['otm_bull_raw'] = df['prem_cb_otm'] + df['prem_ps_otm']
        df['otm_bear_raw'] = df['prem_pb_otm'] + df['prem_cs_otm']
        df['bull_total'] = df['bull_prem_itm'] + df['otm_bull_raw']
        df['bear_total'] = df['bear_prem_itm'] + df['otm_bear_raw']
        
        # Badge round
        df['badge_round'] = df.apply(
            lambda row: '🟢' if row['bull_prem_itm'] > row['bear_prem_itm']
                       else '🔴' if row['bull_prem_itm'] < row['bear_prem_itm']
                       else '',
            axis=1
        )
        
        # Badge more (complete implementation)
        def calc_badge_more(row):
            badge_parts = []
            bull_prem_itm = row['bull_prem_itm']
            bear_prem_itm = row['bear_prem_itm']
            
            # Diamond badge
            if bull_prem_itm > bear_prem_itm and row['prem_cb_itm'] > row['prem_ps_itm']:
                badge_parts.append('💎')
            elif bull_prem_itm < bear_prem_itm and row['prem_pb_itm'] > row['prem_cs_itm']:
                badge_parts.append('💎')
            
            # Star badge
            if bull_prem_itm > bear_prem_itm and (row['otm_bull_raw'] - row['otm_bear_raw']) > 10000:
                badge_parts.append('⭐')
            elif bull_prem_itm < bear_prem_itm and (row['otm_bear_raw'] - row['otm_bull_raw']) > 10000:
                badge_parts.append('⭐')
            
            # Money badge
            if bull_prem_itm > bear_prem_itm and row['oi_cb_otm'] > 100000:
                badge_parts.append('💰')
            elif bull_prem_itm < bear_prem_itm and row['oi_pb_otm'] > 100000:
                badge_parts.append('💰')
            
            # Check badge (vol > oi)
            vol_num = row.get('vol_num', 0) or 0
            oi_num = row.get('oi_num', 0) or 0
            if vol_num > oi_num:
                badge_parts.append('✔')
            
            return ''.join(badge_parts)
        
        df['badge_more'] = df.apply(calc_badge_more, axis=1)
        
        # Flash badge (complete logic)
        def calc_flash(row):
            premium_num = row.get('premium_num', 0) or 0
            if premium_num <= 10000:
                return ''
            
            bull_prem_itm = row['bull_prem_itm']
            bear_prem_itm = row['bear_prem_itm']
            cp_norm = row.get('cp_norm', '')
            side = str(row.get('Side', '')).upper()
            
            # Check for AA/BB in Side field
            has_aa = 'AA' in side
            has_bb = 'BB' in side
            
            if bull_prem_itm > bear_prem_itm:
                if (cp_norm == 'CALL' and has_aa) or (cp_norm == 'PUT' and has_bb):
                    return '⚡'
            elif bull_prem_itm < bear_prem_itm:
                if (cp_norm == 'CALL' and has_bb) or (cp_norm == 'PUT' and has_aa):
                    return '⚡'
            
            return ''
        
        df['flash'] = df.apply(calc_flash, axis=1)
        
        # Rocket badge - matches SQL logic exactly
        def calc_rocket(row):
            vol_num = row.get('vol_num', 0) or 0
            oi_num = row.get('oi_num', 0) or 0
            premium_num = row.get('premium_num', 0) or 0
            badge_round = row.get('badge_round', '')
            badge_more = row.get('badge_more', '')
            flash = row.get('flash', '')
            
            has_diamond = '💎' in badge_more
            has_star = '⭐' in badge_more
            has_money = '💰' in badge_more
            # Check vol > oi directly (matches SQL: COALESCE(d.vol_num,0) > COALESCE(d.oi_num,0))
            vol_gt_oi = vol_num > oi_num
            
            # Full rocket conditions (matches SQL exactly)
            if (badge_round == '🟢' and has_diamond and has_star and has_money and 
                flash == '⚡' and premium_num > 1000000):
                return '🚀'
            
            # Triple rocket (matches SQL: vol > oi AND flash AND money AND premium > 500K)
            if (vol_gt_oi and flash == '⚡' and has_money and premium_num > 500000):
                return '🚀🚀🚀'
            
            # Double rocket (matches SQL: vol > oi AND flash AND money)
            if (vol_gt_oi and flash == '⚡' and has_money):
                return '🚀🚀'
            
            # Single rocket (matches SQL: vol > oi AND (flash OR premium > 500K))
            if (vol_gt_oi and (flash == '⚡' or premium_num > 500000)):
                return '🚀'
            
            return ''
        
        df['rocket'] = df.apply(calc_rocket, axis=1)
        
        return df
    
    def process_rocket_score(self, df: pd.DataFrame) -> pd.DataFrame:
        """Calculate rocket score and final rocket label"""
        df = df.copy()
        
        def calc_score(row):
            score = 0.0
            
            # Premium tier - ensure it's a number
            prem = row.get('premium_num')
            if prem is None or pd.isna(prem):
                prem = 0.0
            else:
                prem = float(prem) if prem else 0.0
            if prem >= 2000000:
                score += 3.0
            elif prem >= 800000:
                score += 2.0
            elif prem >= 200000:
                score += 1.0
            
            # Net premium imbalance - ensure values are numbers
            bull = row.get('bull_total')
            bear = row.get('bear_total')
            bull = float(bull) if bull is not None and not pd.isna(bull) else 0.0
            bear = float(bear) if bear is not None and not pd.isna(bear) else 0.0
            total = bull + bear
            if total > 0:
                net_diff = bull - bear
                score += 1.5 * safe_divide(net_diff, total)
            
            # Vol > OI - ensure values are numbers
            vol_num = row.get('vol_num')
            oi_num = row.get('oi_num')
            vol_num = float(vol_num) if vol_num is not None and not pd.isna(vol_num) else 0.0
            oi_num = float(oi_num) if oi_num is not None and not pd.isna(oi_num) else 0.0
            if vol_num > oi_num:
                score += 1.2
            
            # Session weight
            session = row.get('session_bucket', '')
            if session == 'RTH':
                score += 1.0
            elif session == 'POST':
                score += 0.5
            elif session == 'PRE':
                score += 0.3
            
            # Catalyst flag
            if row.get('near_alert_type'):
                score += 1.0
            
            # OTM bias - ensure values are numbers
            cp_norm = row.get('cp_norm', '')
            strike_num = row.get('strike_num')
            spot_num = row.get('spot_num')
            strike_num = float(strike_num) if strike_num is not None and not pd.isna(strike_num) else 0.0
            spot_num = float(spot_num) if spot_num is not None and not pd.isna(spot_num) else 0.0
            if cp_norm == 'CALL' and strike_num > spot_num:
                score += 0.8
            elif cp_norm == 'PUT' and strike_num < spot_num:
                score += 0.8
            
            # Tape alignment
            tape_align = row.get('tape_alignment')
            if tape_align == 1 or tape_align is True:
                score += 0.5
            
            # Ensure score is a number before rounding
            if score is None or pd.isna(score):
                score = 0.0
            
            return round(float(score), 2)
        
        df['rocket_score'] = df.apply(calc_score, axis=1)
        
        # Calculate final rocket label with moneyness
        def calc_rocket_with_mny(row):
            score = row.get('rocket_score', 0)
            mny_pct = row.get('mny_pct')
            rocket = row.get('rocket', '')
            
            # Determine rocket base from score
            if score >= 5.0:
                rocket_base = '🚀🚀🚀'
            elif score >= 3.5:
                rocket_base = '🚀🚀'
            elif score >= 2.0:
                rocket_base = '🚀'
            else:
                rocket_base = rocket if rocket else ''
            
            # Append moneyness if available
            if mny_pct is not None and not pd.isna(mny_pct):
                mny_label = 'ITM' if mny_pct >= 0 else 'OTM'
                mny_str = f"{abs(mny_pct):.1f}"
                return f"{rocket_base} [{mny_label} {mny_str}%]"
            
            return rocket_base
        
        df['Rocket_with_mny'] = df.apply(calc_rocket_with_mny, axis=1)
        
        return df
    
    def process(self, df: pd.DataFrame, start_day: datetime, end_day: datetime) -> pd.DataFrame:
        """Main processing pipeline"""
        try:
            logger.info(f"Starting options flow processing: {len(df)} rows, {start_day} to {end_day}")
            
            if df.empty:
                logger.warning("Input DataFrame is empty")
                return df
            
            # Step 1: Base normalization
            logger.debug("Step 1: Base normalization")
            df = self.process_base(df)
            logger.debug(f"After base normalization: {len(df)} rows")
            
            if df.empty:
                logger.warning("No data after base normalization")
                return df
            
            # Step 2: Flow filtering
            logger.debug("Step 2: Flow filtering")
            df = self.process_flow(df, start_day, end_day)
            logger.debug(f"After flow filtering: {len(df)} rows")
            
            if df.empty:
                logger.warning("No data after flow filtering - date range may not match data")
                # Return empty DataFrame with required columns to avoid KeyError
                return pd.DataFrame(columns=[
                    'rid', 'symbol_norm', 'cp_norm', 'side_norm', 'strike_num', 'spot_num',
                    'premium_num', 'vol_num', 'oi_num', 'exp_date', 'flow_ts_local', 'flow_ts_utc',
                    'flow_date_cst', 'moneyness', 'direction', 'mny_pct', 'badge_round', 'badge_more',
                    'flash', 'rocket', 'rocket_score', 'Rocket_with_mny'
                ])
            
            # Step 3: Moneyness
            logger.debug("Step 3: Moneyness calculation")
            df = self.process_moneyness(df)
            
            # Step 4: Aggregations
            logger.debug("Step 4: Aggregations")
            df = self.process_aggregations(df)
            
            # Step 5: Badges
            logger.debug("Step 5: Badge calculation")
            df = self.process_badges(df)
            
            # Step 6: Rocket score
            logger.debug("Step 6: Rocket score calculation")
            df = self.process_rocket_score(df)
            
            logger.info(f"Processing complete: {len(df)} rows processed")
            return df
            
        except Exception as e:
            handle_processing_error(
                e,
                context={'start_day': start_day, 'end_day': end_day, 'input_rows': len(df)},
                raise_error=True
            )