""" Price Context Service Handles price data enrichment for options flow Uses Yahoo Finance for real-time data instead of database """ import pandas as pd import asyncpg from datetime import datetime, timedelta from typing import Dict, Optional import pytz from services.yahoo_finance_service import YahooFinanceService from utils.logger import logger class PriceContextService: """Service for enriching flow data with price context""" def __init__(self, pool: asyncpg.Pool): self.pool = pool self.ct_tz = pytz.timezone('America/Chicago') self.use_yahoo_finance = False # Temporarily disabled - focus on Phase 1 button self.yahoo_service = YahooFinanceService() if self.use_yahoo_finance else None def get_session_bucket(self, flow_ts_local: datetime) -> str: """Determine session bucket from flow timestamp""" if pd.isna(flow_ts_local): return 'OFF' hour = flow_ts_local.hour minute = flow_ts_local.minute if 4 <= hour < 9 or (hour == 9 and minute < 30): return 'PRE' elif (hour == 9 and minute >= 30) or (9 < hour < 16): return 'RTH' elif 16 <= hour < 20: return 'POST' else: return 'OFF' async def get_price_at_time( self, symbol: str, timestamp: datetime, pool: asyncpg.Pool = None # Not used anymore, kept for compatibility ) -> Optional[Dict]: """Get price data at or before a specific timestamp using Yahoo Finance""" if not self.use_yahoo_finance or not self.yahoo_service: return None # Yahoo Finance disabled try: price = self.yahoo_service.get_price_at_time(symbol, timestamp) if price: return { 'close': price, 'high': price, # Approximate 'low': price, # Approximate 'volume': None, 'ts': timestamp } return None except Exception as e: logger.debug(f"Error getting price at time from Yahoo Finance: {e}") return None async def get_rth_open( self, symbol: str, flow_date_cst: datetime.date, pool: asyncpg.Pool = None # Not used anymore, kept for compatibility ) -> Optional[Dict]: """Get first RTH bar for a given date using Yahoo Finance""" if not self.use_yahoo_finance or not self.yahoo_service: return None # Yahoo Finance disabled try: rth_open_price = self.yahoo_service.get_rth_open(symbol, flow_date_cst) if rth_open_price: rth_time = self.ct_tz.localize( datetime.combine(flow_date_cst, datetime.min.time().replace(hour=9, minute=30)) ) return { 'open': rth_open_price, 'ts': rth_time } return None except Exception as e: logger.debug(f"Error getting RTH open from Yahoo Finance: {e}") return None async def get_prior_close( self, symbol: str, flow_date_cst: datetime.date, pool: asyncpg.Pool = None # Not used anymore, kept for compatibility ) -> Optional[float]: """Get prior day's close using Yahoo Finance""" if not self.use_yahoo_finance or not self.yahoo_service: return None # Yahoo Finance disabled try: return self.yahoo_service.get_prior_close(symbol, flow_date_cst) except Exception as e: logger.debug(f"Error getting prior close from Yahoo Finance: {e}") return None async def calculate_vwap_at_time( self, symbol: str, timestamp: datetime, pool: asyncpg.Pool = None # Not used anymore, kept for compatibility ) -> Optional[Dict]: """ Calculate VWAP (Volume Weighted Average Price) up to the given timestamp for the trading day using Yahoo Finance. VWAP = SUM(price * volume) / SUM(volume) """ try: # Convert timestamp to CST if needed if timestamp.tzinfo is None: timestamp = self.ct_tz.localize(timestamp) else: timestamp = timestamp.astimezone(self.ct_tz) # Convert to Eastern Time for market hours check (US market opens at 9:30 AM ET) et_tz = pytz.timezone('America/New_York') timestamp_et = timestamp.astimezone(et_tz) if timestamp.tzinfo else et_tz.localize(timestamp) trade_time = timestamp_et.time() # Only calculate VWAP if we're at or after RTH open (9:30 AM ET) if trade_time < pd.Timestamp('09:30:00').time(): # Before RTH, return None (no VWAP yet) logger.debug(f"Before RTH open (9:30 AM ET): {timestamp_et.strftime('%H:%M:%S %Z')}") return None # Calculate VWAP from RTH open (9:30 AM) to the given timestamp if not self.use_yahoo_finance or not self.yahoo_service: return None # Yahoo Finance disabled vwap = await self.yahoo_service.calculate_vwap(symbol, timestamp) if vwap: return { 'vwap': vwap, 'total_volume': None, # Yahoo Finance doesn't provide this easily 'bar_count': None } return None except Exception as e: logger.debug(f"Error calculating VWAP from Yahoo Finance: {e}") return None async def enrich_flow_with_prices( self, flow_rows: pd.DataFrame, pool: asyncpg.Pool ) -> pd.DataFrame: """Enrich flow data with price context (optimized with batch queries)""" df = flow_rows.copy() # Add session bucket df['session_bucket'] = df['flow_ts_local'].apply(self.get_session_bucket) if df.empty: return df # Batch fetch all price data using Yahoo Finance # Get unique symbols and dates unique_symbols = df['symbol_norm'].unique().tolist() unique_dates = df['flow_date_cst'].unique().tolist() # Batch fetch prices at flow times using Yahoo Finance # Group by symbol to reduce API calls price_data_dict = {} for symbol in unique_symbols: symbol_flows = df[df['symbol_norm'] == symbol] timestamps = symbol_flows['flow_ts_utc'].dropna().unique().tolist() if not timestamps: continue # For each timestamp, get the latest price <= that timestamp # Only fetch unique timestamps per symbol for ts in timestamps: try: price_data = await self.get_price_at_time(symbol, ts) if price_data: price_data_dict[(symbol.upper(), ts)] = price_data # Also get 5m and 15m ago prices (only if needed) # Skip if we already have this timestamp ts_5m = ts - timedelta(minutes=5) if (symbol.upper(), ts_5m) not in price_data_dict and self.use_yahoo_finance and self.yahoo_service: try: price_5m = self.yahoo_service.get_price_at_time(symbol, ts_5m) if price_5m: price_data_dict[(symbol.upper(), ts_5m)] = {'close': price_5m} except Exception as e: pass ts_15m = ts - timedelta(minutes=15) if (symbol.upper(), ts_15m) not in price_data_dict and self.use_yahoo_finance and self.yahoo_service: try: price_15m = self.yahoo_service.get_price_at_time(symbol, ts_15m) if price_15m: price_data_dict[(symbol.upper(), ts_15m)] = {'close': price_15m} except Exception as e: pass except Exception as e: logger.debug(f"Error fetching price for {symbol} at {ts}: {e}") # Add small delay on error to avoid rate limiting import asyncio await asyncio.sleep(0.2) # Batch fetch RTH opens using Yahoo Finance rth_open_dict = {} for symbol in unique_symbols: for date in unique_dates: try: rth_data = await self.get_rth_open(symbol, date) if rth_data: rth_open_dict[(symbol.upper(), date)] = rth_data except Exception as e: logger.debug(f"Error fetching RTH open for {symbol} on {date}: {e}") # Batch fetch prior closes using Yahoo Finance prior_close_dict = {} for symbol in unique_symbols: for date in unique_dates: try: prior_close = await self.get_prior_close(symbol, date) if prior_close: prior_close_dict[(symbol.upper(), date)] = prior_close except Exception as e: logger.debug(f"Error fetching prior close for {symbol} before {date}: {e}") # Batch fetch VWAP at signal times using Yahoo Finance vwap_dict = {} for symbol in unique_symbols: symbol_flows = df[df['symbol_norm'] == symbol] timestamps = symbol_flows['flow_ts_utc'].dropna().unique().tolist() for ts in timestamps: try: vwap_data = await self.calculate_vwap_at_time(symbol, ts) if vwap_data: vwap_dict[(symbol.upper(), ts)] = vwap_data except Exception as e: logger.debug(f"Error calculating VWAP for {symbol} at {ts}: {e}") # Build price data for each flow row price_data = [] for idx, row in df.iterrows(): symbol = row['symbol_norm'] flow_ts_utc = row['flow_ts_utc'] flow_date_cst = row['flow_date_cst'] # Get price at flow time price_at_time = price_data_dict.get((symbol.upper(), flow_ts_utc)) if not pd.isna(flow_ts_utc) else None # Get RTH open rth_open_data = rth_open_dict.get((symbol.upper(), flow_date_cst)) # Get prior close prior_close = prior_close_dict.get((symbol.upper(), flow_date_cst)) # Get 5m and 15m ago prices price_5m_ago = None price_15m_ago = None if not pd.isna(flow_ts_utc): ts_5m = flow_ts_utc - timedelta(minutes=5) price_5m_key = (symbol.upper(), ts_5m) price_5m_ago_data = price_data_dict.get(price_5m_key) price_5m_ago = price_5m_ago_data.get('close') if price_5m_ago_data else None ts_15m = flow_ts_utc - timedelta(minutes=15) price_15m_key = (symbol.upper(), ts_15m) price_15m_ago_data = price_data_dict.get(price_15m_key) price_15m_ago = price_15m_ago_data.get('close') if price_15m_ago_data else None # Get VWAP at signal time vwap_data = vwap_dict.get((symbol.upper(), flow_ts_utc)) if not pd.isna(flow_ts_utc) else None vwap_at_signal = vwap_data['vwap'] if vwap_data else None # Calculate price vs VWAP percentage price_vs_vwap_pct = None if vwap_at_signal and price_at_time and price_at_time.get('close'): try: price_vs_vwap_pct = round( ((price_at_time['close'] - vwap_at_signal) / vwap_at_signal) * 100, 2 ) except (TypeError, ZeroDivisionError): pass price_data.append({ 'u_close': price_at_time['close'] if price_at_time else None, 'u_high': price_at_time['high'] if price_at_time else None, 'u_low': price_at_time['low'] if price_at_time else None, 'u_vol_1m': price_at_time['volume'] if price_at_time else None, 'rth_open': rth_open_data['open'] if rth_open_data else None, 'prior_close': prior_close, 'close_5m_ago': price_5m_ago, 'close_15m_ago': price_15m_ago, 'vwap_at_signal': vwap_at_signal, 'price_vs_vwap_pct': price_vs_vwap_pct, }) # Merge price data price_df = pd.DataFrame(price_data, index=df.index) df = pd.concat([df, price_df], axis=1) # Calculate price features def calc_pct_vs_prior_close(row): if row['prior_close'] and row['u_close'] and row['prior_close'] != 0: try: return round(((row['u_close'] - row['prior_close']) * 100.0 / row['prior_close']), 2) except (TypeError, ValueError): return None return None df['pct_vs_prior_close'] = df.apply(calc_pct_vs_prior_close, axis=1) def calc_pct_vs_rth_open(row): if (row['rth_open'] and row['u_close'] and row['session_bucket'] == 'RTH' and row['rth_open'] != 0): try: return round(((row['u_close'] - row['rth_open']) * 100.0 / row['rth_open']), 2) except (TypeError, ValueError): return None return None df['pct_vs_rth_open'] = df.apply(calc_pct_vs_rth_open, axis=1) def calc_pct_5m_momo(row): if row['close_5m_ago'] and row['u_close'] and row['close_5m_ago'] != 0: try: return round(((row['u_close'] - row['close_5m_ago']) * 100.0 / row['close_5m_ago']), 2) except (TypeError, ValueError): return None return None df['pct_5m_momo'] = df.apply(calc_pct_5m_momo, axis=1) def calc_pct_15m_momo(row): if row['close_15m_ago'] and row['u_close'] and row['close_15m_ago'] != 0: try: return round(((row['u_close'] - row['close_15m_ago']) * 100.0 / row['close_15m_ago']), 2) except (TypeError, ValueError): return None return None df['pct_15m_momo'] = df.apply(calc_pct_15m_momo, axis=1) # Calculate tape alignment df['tape_alignment'] = df.apply( lambda row: self.calculate_tape_alignment(row), axis=1 ) return df def calculate_tape_alignment(self, row) -> int: """Calculate tape alignment based on direction and price movement""" session = row.get('session_bucket', '') direction = row.get('direction', '') tol_pct = 0.20 # tolerance percentage if session == 'PRE' and row.get('pct_vs_prior_close') is not None: pct = row['pct_vs_prior_close'] if (direction == 'BULL' and pct >= tol_pct) or \ (direction == 'BEAR' and pct <= -tol_pct): return 1 elif session == 'RTH' and row.get('pct_vs_rth_open') is not None: pct = row['pct_vs_rth_open'] if (direction == 'BULL' and pct >= tol_pct) or \ (direction == 'BEAR' and pct <= -tol_pct): return 1 elif session == 'POST' and row.get('pct_vs_prior_close') is not None: pct = row['pct_vs_prior_close'] if (direction == 'BULL' and pct >= tol_pct) or \ (direction == 'BEAR' and pct <= -tol_pct): return 1 return 0