136 lines
5.2 KiB
Python
136 lines
5.2 KiB
Python
"""
|
|
Trade Checklist
|
|
Evaluates signals against 5-point checklist - requires 4/5 to pass
|
|
"""
|
|
import pandas as pd
|
|
from utils.logger import logger
|
|
|
|
|
|
class TradeChecklist:
|
|
"""Service for evaluating trade checklist"""
|
|
|
|
def __init__(self):
|
|
self.min_score = 4 # Require 4/5 checks to pass
|
|
|
|
def _check_vwap_respect(self, row) -> bool:
|
|
"""Check if price respects VWAP (within reasonable distance)"""
|
|
# 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
|
|
|
|
price_vs_vwap = safe_get(row, 'price_vs_vwap_pct')
|
|
vwap_at_signal = safe_get(row, 'vwap_at_signal')
|
|
|
|
# If no VWAP data, can't check - return False
|
|
if vwap_at_signal is None or pd.isna(vwap_at_signal):
|
|
return False
|
|
|
|
# If price_vs_vwap is None, can't check
|
|
if price_vs_vwap is None or pd.isna(price_vs_vwap):
|
|
return False
|
|
|
|
# Convert to float
|
|
try:
|
|
price_vs_vwap = float(price_vs_vwap)
|
|
except (TypeError, ValueError):
|
|
return False
|
|
|
|
# Price respects VWAP if within ±2% (not too extended)
|
|
# For bullish: price should be at or above VWAP (or within 1% below)
|
|
# For bearish: price should be at or below VWAP (or within 1% above)
|
|
direction = safe_get(row, 'direction', '') or ''
|
|
badge_round = safe_get(row, 'badge_round', '') or ''
|
|
|
|
if badge_round == '🟢' or direction == 'BULL':
|
|
# Bullish: price should be near or above VWAP
|
|
return price_vs_vwap >= -1.0 # Within 1% below VWAP is acceptable
|
|
elif badge_round == '🔴' or direction == 'BEAR':
|
|
# Bearish: price should be near or below VWAP
|
|
return price_vs_vwap <= 1.0 # Within 1% above VWAP is acceptable
|
|
|
|
# Default: check if within ±2%
|
|
return abs(price_vs_vwap) <= 2.0
|
|
|
|
def evaluate(self, row) -> dict:
|
|
"""
|
|
Evaluate trade checklist
|
|
|
|
Checklist items:
|
|
1. 🟢 or 🔴 (has direction)
|
|
2. 💎 (has diamond)
|
|
3. ⭐ (has star)
|
|
4. Price respects VWAP
|
|
5. Index confirms (if available)
|
|
"""
|
|
# 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 ''
|
|
index_aligned = safe_get(row, 'index_aligned', False)
|
|
|
|
checks = {
|
|
'has_direction': badge_round in ['🟢', '🔴'],
|
|
'has_diamond': '💎' in str(badge_more),
|
|
'has_star': '⭐' in str(badge_more),
|
|
'price_respects_vwap': self._check_vwap_respect(row),
|
|
'index_confirms': bool(index_aligned) # Will be added by index correlation service
|
|
}
|
|
|
|
score = sum(checks.values())
|
|
passed = score >= self.min_score
|
|
|
|
return {
|
|
'checklist_score': int(score),
|
|
'checklist_passed': bool(passed),
|
|
'checks': checks
|
|
}
|
|
|
|
def evaluate_all(self, df: pd.DataFrame) -> pd.DataFrame:
|
|
"""Evaluate checklist for all rows in DataFrame"""
|
|
df = df.copy()
|
|
|
|
if df.empty:
|
|
df['checklist_score'] = pd.Series(dtype=int)
|
|
df['checklist_passed'] = pd.Series(dtype=bool)
|
|
df['checklist_details'] = pd.Series(dtype=object)
|
|
return df
|
|
|
|
logger.info(f"Evaluating trade checklist for {len(df)} signals...")
|
|
|
|
# Apply checklist evaluation - returns Series of dicts
|
|
checklist_results = df.apply(self.evaluate, axis=1)
|
|
|
|
# Ensure it's a Series
|
|
if isinstance(checklist_results, pd.DataFrame):
|
|
checklist_results = checklist_results.iloc[:, 0]
|
|
|
|
# Extract checklist data - ensure we get scalar values
|
|
df['checklist_score'] = checklist_results.apply(lambda x: x['checklist_score'] if isinstance(x, dict) else 0)
|
|
df['checklist_passed'] = checklist_results.apply(lambda x: x['checklist_passed'] if isinstance(x, dict) else False)
|
|
|
|
# Store detailed checks as JSON-serializable dict
|
|
def extract_checks(result):
|
|
if isinstance(result, dict):
|
|
return result.get('checks', {})
|
|
return {}
|
|
|
|
df['checklist_details'] = checklist_results.apply(extract_checks)
|
|
|
|
# Log checklist results
|
|
passed_count = df['checklist_passed'].sum()
|
|
avg_score = df['checklist_score'].mean() if len(df) > 0 else 0
|
|
logger.info(f"Checklist results: {passed_count}/{len(df)} passed (avg score: {avg_score:.2f}/5)")
|
|
|
|
return df
|
|
|