154 lines
6.2 KiB
Python
154 lines
6.2 KiB
Python
"""
|
|
Price Reaction Tracker
|
|
Tracks how price moves after a signal appears - critical for filtering hedges/rolls
|
|
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
|
|
from utils.logger import logger
|
|
from services.yahoo_finance_service import YahooFinanceService
|
|
|
|
|
|
class PriceReactionTracker:
|
|
"""Service for tracking price reactions after flow signals"""
|
|
|
|
def __init__(self):
|
|
self.reaction_threshold_pct = 0.5 # 0.5% threshold for "flow led to move"
|
|
self.use_yahoo_finance = False # Temporarily disabled - focus on Phase 1 button
|
|
self.yahoo_service = YahooFinanceService() if self.use_yahoo_finance else None
|
|
|
|
async def get_price_at_time(
|
|
self,
|
|
symbol: str,
|
|
timestamp: datetime,
|
|
pool: asyncpg.Pool = None # Not used anymore, kept for compatibility
|
|
) -> Optional[float]:
|
|
"""Get price 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:
|
|
return self.yahoo_service.get_price_at_time(symbol, timestamp)
|
|
except Exception as e:
|
|
logger.debug(f"Error getting price at time from Yahoo Finance: {e}")
|
|
return None
|
|
|
|
async def enrich_with_reactions(
|
|
self,
|
|
flow_df: pd.DataFrame,
|
|
pool: asyncpg.Pool
|
|
) -> pd.DataFrame:
|
|
"""Enrich flow data with price reaction tracking"""
|
|
df = flow_df.copy()
|
|
|
|
if df.empty:
|
|
return df
|
|
|
|
logger.info(f"Tracking price reactions for {len(df)} signals...")
|
|
|
|
# Get unique symbols and timestamps for batch processing
|
|
unique_symbols = df['symbol_norm'].unique().tolist()
|
|
|
|
# Batch fetch price reactions
|
|
reaction_data = []
|
|
|
|
async with pool.acquire() as conn:
|
|
for idx, row in df.iterrows():
|
|
symbol = row['symbol_norm']
|
|
signal_time = row['flow_ts_utc']
|
|
price_at_signal = row.get('u_close')
|
|
|
|
if pd.isna(signal_time) or not price_at_signal or price_at_signal == 0:
|
|
reaction_data.append({
|
|
'price_reaction_5m_pct': None,
|
|
'price_reaction_15m_pct': None,
|
|
'price_reaction_30m_pct': None,
|
|
'high_break_5m': False,
|
|
'low_break_5m': False,
|
|
'flow_led_to_move': False
|
|
})
|
|
continue
|
|
|
|
# Get prices at 5m, 15m, 30m after signal
|
|
price_5m = None
|
|
price_15m = None
|
|
price_30m = None
|
|
|
|
try:
|
|
ts_5m = signal_time + timedelta(minutes=5)
|
|
price_5m = await self.get_price_at_time(symbol, ts_5m, pool)
|
|
except Exception as e:
|
|
logger.debug(f"Error fetching 5m price for {symbol}: {e}")
|
|
|
|
try:
|
|
ts_15m = signal_time + timedelta(minutes=15)
|
|
price_15m = await self.get_price_at_time(symbol, ts_15m, pool)
|
|
except Exception as e:
|
|
logger.debug(f"Error fetching 15m price for {symbol}: {e}")
|
|
|
|
try:
|
|
ts_30m = signal_time + timedelta(minutes=30)
|
|
price_30m = await self.get_price_at_time(symbol, ts_30m, pool)
|
|
except Exception as e:
|
|
logger.debug(f"Error fetching 30m price for {symbol}: {e}")
|
|
|
|
# Calculate reaction percentages
|
|
reaction_5m = None
|
|
reaction_15m = None
|
|
reaction_30m = None
|
|
|
|
if price_5m and price_at_signal:
|
|
try:
|
|
reaction_5m = round(((price_5m - price_at_signal) / price_at_signal) * 100, 2)
|
|
except (TypeError, ZeroDivisionError):
|
|
pass
|
|
|
|
if price_15m and price_at_signal:
|
|
try:
|
|
reaction_15m = round(((price_15m - price_at_signal) / price_at_signal) * 100, 2)
|
|
except (TypeError, ZeroDivisionError):
|
|
pass
|
|
|
|
if price_30m and price_at_signal:
|
|
try:
|
|
reaction_30m = round(((price_30m - price_at_signal) / price_at_signal) * 100, 2)
|
|
except (TypeError, ZeroDivisionError):
|
|
pass
|
|
|
|
# High/Low break confirmation
|
|
high_at_signal = row.get('u_high', 0) or 0
|
|
low_at_signal = row.get('u_low', 0) or 0
|
|
|
|
high_break_5m = False
|
|
low_break_5m = False
|
|
|
|
if price_5m:
|
|
if high_at_signal > 0 and price_5m > high_at_signal:
|
|
high_break_5m = True
|
|
if low_at_signal > 0 and price_5m < low_at_signal:
|
|
low_break_5m = True
|
|
|
|
# Determine if flow led to move
|
|
flow_led_to_move = False
|
|
if reaction_5m is not None:
|
|
flow_led_to_move = abs(reaction_5m) >= self.reaction_threshold_pct
|
|
|
|
reaction_data.append({
|
|
'price_reaction_5m_pct': reaction_5m,
|
|
'price_reaction_15m_pct': reaction_15m,
|
|
'price_reaction_30m_pct': reaction_30m,
|
|
'high_break_5m': high_break_5m,
|
|
'low_break_5m': low_break_5m,
|
|
'flow_led_to_move': flow_led_to_move
|
|
})
|
|
|
|
# Merge reaction data
|
|
reaction_df = pd.DataFrame(reaction_data, index=df.index)
|
|
df = pd.concat([df, reaction_df], axis=1)
|
|
|
|
logger.info(f"✅ Price reaction tracking complete. {df['flow_led_to_move'].sum()} signals led to price moves")
|
|
|
|
return df
|
|
|