125 lines
4.9 KiB
Python
125 lines
4.9 KiB
Python
"""
|
|
Signal Tier Classifier
|
|
Classifies signals into Tier-1 (tradeable alone), Tier-2 (needs confirmation), or Ignore
|
|
"""
|
|
import pandas as pd
|
|
from utils.logger import logger
|
|
|
|
|
|
class SignalTierClassifier:
|
|
"""Service for classifying signals into tiers based on badge combinations"""
|
|
|
|
def __init__(self):
|
|
self.tier1_min_premium = 500000 # $500K minimum for Tier-1
|
|
|
|
def classify_tier(self, row) -> str:
|
|
"""
|
|
Classify signal tier based on badge combinations and premium
|
|
|
|
Tier-1: 🟢/🔴 + 💎 + ⭐ + premium > 500K + direction aligned
|
|
Tier-2: 🟢 + 💎 (no ⭐) OR ⭐ without 💎
|
|
Ignore: OTM-only, mixed signals, low volume/OI ratio
|
|
"""
|
|
# Use proper pandas Series access with fallback
|
|
def safe_get(row, key, default=None):
|
|
try:
|
|
val = row.get(key, default) if hasattr(row, 'get') else (row[key] if key in row.index else default)
|
|
return default if pd.isna(val) else val
|
|
except (KeyError, IndexError):
|
|
return default
|
|
|
|
badge_round = safe_get(row, 'badge_round', '') or ''
|
|
badge_more = safe_get(row, 'badge_more', '') or ''
|
|
premium = float(safe_get(row, 'premium_num', 0) or 0)
|
|
direction = safe_get(row, 'direction', '') or ''
|
|
bull_total = float(safe_get(row, 'bull_total', 0) or 0)
|
|
bear_total = float(safe_get(row, 'bear_total', 0) or 0)
|
|
|
|
has_diamond = '💎' in str(badge_more)
|
|
has_star = '⭐' in str(badge_more)
|
|
has_money = '💰' in str(badge_more)
|
|
|
|
# Check if OTM-only (no ITM premium)
|
|
prem_cb_itm = float(safe_get(row, 'prem_cb_itm', 0) or 0)
|
|
prem_ps_itm = float(safe_get(row, 'prem_ps_itm', 0) or 0)
|
|
prem_cs_itm = float(safe_get(row, 'prem_cs_itm', 0) or 0)
|
|
prem_pb_itm = float(safe_get(row, 'prem_pb_itm', 0) or 0)
|
|
|
|
bull_prem_itm = prem_cb_itm + prem_ps_itm
|
|
bear_prem_itm = prem_cs_itm + prem_pb_itm
|
|
|
|
# Ignore: OTM-only prints (no ITM premium)
|
|
if bull_prem_itm == 0 and bear_prem_itm == 0:
|
|
return 'IGNORE'
|
|
|
|
# Ignore: Mixed 🟢/🔴 with no net premium edge
|
|
if not badge_round or badge_round == '':
|
|
return 'IGNORE'
|
|
|
|
# Ignore: Big premium but volume << OI (likely rolls/hedges)
|
|
vol_num = float(safe_get(row, 'vol_num', 0) or 0)
|
|
oi_num = float(safe_get(row, 'oi_num', 0) or 0)
|
|
if premium > 500000 and vol_num > 0 and oi_num > 0:
|
|
vol_oi_ratio = vol_num / oi_num if oi_num > 0 else 0
|
|
if vol_oi_ratio < 0.3: # Volume is less than 30% of OI
|
|
return 'IGNORE'
|
|
|
|
# Tier-1 conditions: 🟢/🔴 + 💎 + ⭐ + premium > 500K + direction aligned
|
|
if (badge_round in ['🟢', '🔴'] and
|
|
has_diamond and
|
|
has_star and
|
|
premium >= self.tier1_min_premium):
|
|
|
|
# Check direction alignment
|
|
net_premium = bull_total - bear_total
|
|
if badge_round == '🟢' and direction == 'BULL' and net_premium > 0:
|
|
return 'TIER_1'
|
|
elif badge_round == '🔴' and direction == 'BEAR' and net_premium < 0:
|
|
return 'TIER_1'
|
|
|
|
# Tier-2 conditions: 🟢 + 💎 (no ⭐ yet) OR ⭐ without 💎
|
|
if badge_round == '🟢' and has_diamond and not has_star:
|
|
return 'TIER_2'
|
|
|
|
if has_star and not has_diamond:
|
|
return 'TIER_2'
|
|
|
|
# If we have direction and diamond but missing star, it's Tier-2
|
|
if badge_round in ['🟢', '🔴'] and has_diamond and not has_star:
|
|
return 'TIER_2'
|
|
|
|
# Default to ignore if doesn't meet criteria
|
|
return 'IGNORE'
|
|
|
|
def classify_tiers(self, df: pd.DataFrame) -> pd.DataFrame:
|
|
"""Classify tiers for all rows in DataFrame"""
|
|
df = df.copy()
|
|
|
|
if df.empty:
|
|
df['signal_tier'] = pd.Series(dtype=str)
|
|
df['is_tradeable'] = pd.Series(dtype=bool)
|
|
return df
|
|
|
|
logger.info(f"Classifying signal tiers for {len(df)} rows...")
|
|
|
|
# Apply classification and ensure it returns a Series
|
|
signal_tiers = df.apply(self.classify_tier, axis=1)
|
|
|
|
# Ensure it's a Series, not a DataFrame
|
|
if isinstance(signal_tiers, pd.DataFrame):
|
|
# If somehow it's a DataFrame, take the first column
|
|
signal_tiers = signal_tiers.iloc[:, 0]
|
|
|
|
df['signal_tier'] = signal_tiers
|
|
|
|
# Add is_tradeable flag (Tier-1 only)
|
|
df['is_tradeable'] = df['signal_tier'] == 'TIER_1'
|
|
|
|
# Log tier distribution
|
|
tier_counts = df['signal_tier'].value_counts()
|
|
logger.info(f"Signal tier distribution: {tier_counts.to_dict()}")
|
|
logger.info(f"Tradeable signals (Tier-1): {df['is_tradeable'].sum()}")
|
|
|
|
return df
|
|
|