"""Swing Failure Pattern (SFP) Detector — identifies false breakouts. A Swing Failure Pattern occurs when price sweeps beyond a key level (high or low) but fails to sustain above/below it, signalling a reversal. Bullish SFP: price sweeps *below* a key low then recovers above it. -> The stop-hunt traps shorts, then reverses up. Bearish SFP: price sweeps *above* a key high then falls below it. -> The stop-hunt traps longs, then reverses down. Also identifies liquidity pools (clusters of equal highs/lows) where stop losses likely accumulate, making them prime SFP targets. Data source: Asset.sparkline_in_7d (168 hourly data points). """ from __future__ import annotations from app.helpers.sparkline import load_sparkline class SwingFailurePatternService: """Detect Swing Failure Patterns and liquidity pools.""" MIN_POINTS = 30 # Minimum penetration beyond level to count as a sweep (pct) MIN_SWEEP_PCT = 0.2 # Maximum bars for price to recover after sweep MAX_RECOVERY_BARS = 8 # Tolerance for "equal" highs/lows (pct) EQUAL_LEVEL_TOLERANCE = 0.5 def __init__(self): pass # ------------------------------------------------------------------ # Public API # ------------------------------------------------------------------ def analyze(self, symbol: str, asset_type: str = 'crypto') -> dict: """Detect SFPs for a single asset.""" import json import numpy as np from app.models.asset import Asset try: asset = Asset.query.filter_by( symbol=symbol.upper(), asset_type=asset_type ).first() if not asset: return {'status': 'error', 'message': f'Asset {symbol} not found'} prices = load_sparkline(asset) if len(prices) < self.MIN_POINTS: return {'status': 'error', 'message': 'Insufficient data'} prices = [float(p) for p in prices if p and float(p) > 0] if len(prices) < self.MIN_POINTS: return {'status': 'error', 'message': 'Insufficient valid data points'} prices_arr = np.array(prices, dtype=float) current_price = float(asset.current_price) if asset.current_price else float(prices_arr[-1]) # Identify key levels (significant highs and lows) key_highs = self._find_key_levels(prices_arr, level_type='high') key_lows = self._find_key_levels(prices_arr, level_type='low') # Detect bullish SFPs (sweep below key low, recover above) bullish_sfps = self._detect_bullish_sfp(prices_arr, key_lows) # Detect bearish SFPs (sweep above key high, fall below) bearish_sfps = self._detect_bearish_sfp(prices_arr, key_highs) all_sfps = bullish_sfps + bearish_sfps all_sfps.sort(key=lambda x: x.get('reliability', 0), reverse=True) # Find liquidity pools liquidity_pools = self._find_liquidity_pools(prices_arr) # Calculate entry/target for each SFP for sfp in all_sfps: entry, sl, t1, t2, rr = self._calculate_target_price( sfp['type'], sfp.get('recovery_price', current_price), sfp.get('sweep_extreme', current_price), current_price ) sfp['entry'] = round(entry, 8) sfp['stop_loss'] = round(sl, 8) sfp['target_1'] = round(t1, 8) sfp['target_2'] = round(t2, 8) sfp['risk_reward'] = round(rr, 1) # Active SFP (most recent actionable one) active_sfp = None for sfp in all_sfps: if sfp.get('time_index', 0) >= len(prices_arr) - 24: active_sfp = sfp break # Score score = self._calculate_score(all_sfps, active_sfp, liquidity_pools) # Overall bias bullish_count = sum(1 for s in all_sfps if s['type'] == 'bullish_sfp') bearish_count = sum(1 for s in all_sfps if s['type'] == 'bearish_sfp') if bullish_count > bearish_count: bias = 'bullish' elif bearish_count > bullish_count: bias = 'bearish' else: bias = 'neutral' return { 'status': 'success', 'data': { 'symbol': asset.symbol, 'name': asset.name, 'current_price': current_price, 'sfp_patterns': all_sfps, 'liquidity_pools': liquidity_pools, 'active_sfp': active_sfp, 'score': score, 'bias': bias, }, } except Exception as e: return {'status': 'error', 'message': str(e)} def scan_all(self, asset_type: str = 'crypto', limit: int = 50) -> dict: """Scan top assets for Swing Failure Patterns.""" from app.models.asset import Asset try: assets = ( Asset.query .filter_by(asset_type=asset_type, is_active=True) .order_by(Asset.market_cap_rank.asc()) .limit(limit) .all() ) results = [] for asset in assets: result = self.analyze(symbol=asset.symbol, asset_type=asset_type) if result.get('status') == 'success': results.append(result['data']) results.sort(key=lambda x: x.get('score', 0), reverse=True) return { 'status': 'success', 'data': {'items': results, 'total': len(results)}, } except Exception as e: return {'status': 'error', 'message': str(e)} # ------------------------------------------------------------------ # Key Level Detection # ------------------------------------------------------------------ def _find_key_levels(self, prices, level_type: str = 'high', lookback: int = 50, order: int = 5) -> list: """Identify significant highs or lows using swing detection. Returns list of dicts with 'index' and 'price'. """ import numpy as np levels = [] end = len(prices) start = max(0, end - lookback) for i in range(start + order, end - order): if level_type == 'high': is_swing = all(prices[i] >= prices[i - j] for j in range(1, order + 1)) and \ all(prices[i] >= prices[i + j] for j in range(1, order + 1)) else: is_swing = all(prices[i] <= prices[i - j] for j in range(1, order + 1)) and \ all(prices[i] <= prices[i + j] for j in range(1, order + 1)) if is_swing: levels.append({'index': int(i), 'price': float(prices[i])}) # Also add the absolute high/low from the lookback subset = prices[start:end] if level_type == 'high': abs_idx = int(np.argmax(subset)) + start levels.append({'index': abs_idx, 'price': float(prices[abs_idx])}) else: abs_idx = int(np.argmin(subset)) + start levels.append({'index': abs_idx, 'price': float(prices[abs_idx])}) # Deduplicate: merge levels within 0.5% of each other levels.sort(key=lambda x: x['price']) merged = [] for lvl in levels: if merged and abs(lvl['price'] - merged[-1]['price']) / merged[-1]['price'] * 100 < self.EQUAL_LEVEL_TOLERANCE: # Keep the more recent one if lvl['index'] > merged[-1]['index']: merged[-1] = lvl else: merged.append(lvl) return merged # ------------------------------------------------------------------ # SFP Detection # ------------------------------------------------------------------ def _detect_bullish_sfp(self, prices, key_lows: list) -> list: """Bullish SFP: price sweeps below key low then recovers above it. For each key low, look for a subsequent bar that goes below the level and then a recovery bar that closes back above. """ import numpy as np sfps = [] for level in key_lows: level_price = level['price'] level_idx = level['index'] # Scan bars after this level was established for i in range(level_idx + 1, len(prices)): sweep_pct = (level_price - prices[i]) / level_price * 100 if level_price > 0 else 0 if sweep_pct >= self.MIN_SWEEP_PCT: # Found a sweep below the level — now check for recovery sweep_low = float(prices[i]) recovery_price, recovery_idx, recovery_speed = self._check_recovery( prices, i, level_price, direction='up' ) if recovery_price is not None: sweep_depth = self._calculate_sweep_depth(sweep_low, level_price) vol_confirm = self._estimate_volume_confirm(prices, i) reliability = self._calculate_sfp_reliability( sweep_depth, recovery_speed, vol_confirm ) sfps.append({ 'type': 'bullish_sfp', 'key_level': round(level_price, 8), 'sweep_low': round(sweep_low, 8), 'sweep_extreme': round(sweep_low, 8), 'recovery_price': round(recovery_price, 8), 'sweep_depth_pct': round(sweep_depth, 2), 'recovery_speed': recovery_speed, 'reliability': reliability, 'time_index': int(recovery_idx), 'description': ( f'Price swept below key support at ' f'{level_price:.8g}, recovered ' f'{recovery_speed}ly - bullish SFP' ), }) break # one SFP per level return sfps def _detect_bearish_sfp(self, prices, key_highs: list) -> list: """Bearish SFP: price sweeps above key high then falls below it.""" import numpy as np sfps = [] for level in key_highs: level_price = level['price'] level_idx = level['index'] for i in range(level_idx + 1, len(prices)): sweep_pct = (prices[i] - level_price) / level_price * 100 if level_price > 0 else 0 if sweep_pct >= self.MIN_SWEEP_PCT: sweep_high = float(prices[i]) recovery_price, recovery_idx, recovery_speed = self._check_recovery( prices, i, level_price, direction='down' ) if recovery_price is not None: sweep_depth = self._calculate_sweep_depth(sweep_high, level_price) vol_confirm = self._estimate_volume_confirm(prices, i) reliability = self._calculate_sfp_reliability( sweep_depth, recovery_speed, vol_confirm ) sfps.append({ 'type': 'bearish_sfp', 'key_level': round(level_price, 8), 'sweep_high': round(sweep_high, 8), 'sweep_extreme': round(sweep_high, 8), 'recovery_price': round(recovery_price, 8), 'sweep_depth_pct': round(sweep_depth, 2), 'recovery_speed': recovery_speed, 'reliability': reliability, 'time_index': int(recovery_idx), 'description': ( f'Price swept above key resistance at ' f'{level_price:.8g}, reversed ' f'{recovery_speed}ly - bearish SFP' ), }) break return sfps def _check_recovery(self, prices, sweep_idx: int, level_price: float, direction: str): """Check if price recovers back through the level after a sweep. Returns (recovery_price, recovery_index, speed) or (None, None, None). """ max_bars = min(self.MAX_RECOVERY_BARS, len(prices) - sweep_idx - 1) for j in range(1, max_bars + 1): idx = sweep_idx + j if idx >= len(prices): break if direction == 'up' and prices[idx] > level_price: speed = self._classify_recovery_speed(j) return float(prices[idx]), idx, speed elif direction == 'down' and prices[idx] < level_price: speed = self._classify_recovery_speed(j) return float(prices[idx]), idx, speed return None, None, None def _classify_recovery_speed(self, bars: int) -> str: """Classify recovery speed by number of bars.""" if bars <= 2: return 'fast' elif bars <= 5: return 'moderate' else: return 'slow' # ------------------------------------------------------------------ # Calculations # ------------------------------------------------------------------ def _calculate_sweep_depth(self, sweep_price: float, level_price: float) -> float: """How far price penetrated the level (percentage).""" if level_price == 0: return 0.0 return abs(sweep_price - level_price) / level_price * 100 def _calculate_recovery_speed(self, prices, sweep_index: int) -> str: """How quickly price recovered (based on bars to reclaim level).""" # Simplified — actual speed is calculated in _check_recovery return 'moderate' def _estimate_volume_confirm(self, prices, sweep_idx: int) -> float: """Estimate volume confirmation at sweep point. Higher volume at sweep -> more confirmation (0-1 scale). """ import numpy as np if sweep_idx < 5 or sweep_idx >= len(prices): return 0.5 # Estimated volume at sweep vol_at_sweep = abs(prices[sweep_idx] - prices[sweep_idx - 1]) * prices[sweep_idx] # Average volume over prior 20 bars start = max(0, sweep_idx - 20) vols = [abs(prices[i] - prices[i - 1]) * prices[i] for i in range(start + 1, sweep_idx)] avg_vol = np.mean(vols) if vols else vol_at_sweep if avg_vol == 0: return 0.5 ratio = vol_at_sweep / avg_vol # Normalize to 0-1 return min(1.0, ratio / 3.0) def _calculate_sfp_reliability(self, sweep_depth: float, recovery_speed: str, volume_confirm: float) -> int: """SFP reliability score 0-100. Factors: - Sweep depth: moderate sweep (0.5-3%) is ideal - Recovery speed: faster = more reliable - Volume confirmation: higher = more reliable """ score = 0.0 # Sweep depth (ideal 0.5-3%) if 0.3 <= sweep_depth <= 4.0: score += 35 elif 0.1 <= sweep_depth <= 6.0: score += 20 else: score += 5 # Recovery speed speed_bonus = {'fast': 35, 'moderate': 22, 'slow': 10} score += speed_bonus.get(recovery_speed, 10) # Volume confirmation (0-1 scale -> 0-30 pts) score += volume_confirm * 30 return max(0, min(100, int(score))) def _calculate_target_price(self, sfp_type: str, entry: float, stop_extreme: float, current_price: float): """Calculate entry, stop-loss, and targets based on R:R. Bullish SFP: entry above recovery, SL below sweep low, targets up. Bearish SFP: entry below recovery, SL above sweep high, targets down. """ if sfp_type == 'bullish_sfp': entry_price = entry stop_loss = stop_extreme * 0.99 # 1% below sweep low risk = abs(entry_price - stop_loss) target_1 = entry_price + 2 * risk # 2:1 R:R target_2 = entry_price + 3 * risk # 3:1 R:R rr = 2.0 else: # bearish_sfp entry_price = entry stop_loss = stop_extreme * 1.01 # 1% above sweep high risk = abs(stop_loss - entry_price) target_1 = entry_price - 2 * risk target_2 = entry_price - 3 * risk rr = 2.0 target_1 = max(target_1, 0) target_2 = max(target_2, 0) return entry_price, stop_loss, target_1, target_2, rr # ------------------------------------------------------------------ # Liquidity Pools # ------------------------------------------------------------------ def _find_liquidity_pools(self, prices) -> list: """Find clusters of equal highs/lows where stop losses cluster. Equal lows = multiple swing lows at approximately the same price -> stops from long positions likely rest just below. Equal highs = same for short stops above. """ import numpy as np pools = [] # Find swing lows swing_lows = [] swing_highs = [] order = 3 for i in range(order, len(prices) - order): if all(prices[i] <= prices[i - j] for j in range(1, order + 1)) and \ all(prices[i] <= prices[i + j] for j in range(1, order + 1)): swing_lows.append(float(prices[i])) if all(prices[i] >= prices[i - j] for j in range(1, order + 1)) and \ all(prices[i] >= prices[i + j] for j in range(1, order + 1)): swing_highs.append(float(prices[i])) # Cluster equal lows low_clusters = self._cluster_levels(swing_lows) for price_level, count in low_clusters: if count >= 2: risk = 'high' if count >= 3 else 'medium' pools.append({ 'type': 'equal_lows', 'price': round(price_level, 8), 'count': count, 'risk': risk, }) # Cluster equal highs high_clusters = self._cluster_levels(swing_highs) for price_level, count in high_clusters: if count >= 2: risk = 'high' if count >= 3 else 'medium' pools.append({ 'type': 'equal_highs', 'price': round(price_level, 8), 'count': count, 'risk': risk, }) pools.sort(key=lambda x: x['count'], reverse=True) return pools[:10] def _cluster_levels(self, levels: list) -> list: """Group nearby price levels into clusters. Returns list of (cluster_center, count). """ if not levels: return [] sorted_levels = sorted(levels) clusters = [] current_cluster = [sorted_levels[0]] for i in range(1, len(sorted_levels)): if sorted_levels[i] == 0: continue pct_diff = abs(sorted_levels[i] - current_cluster[-1]) / current_cluster[-1] * 100 if pct_diff <= self.EQUAL_LEVEL_TOLERANCE: current_cluster.append(sorted_levels[i]) else: if len(current_cluster) >= 2: center = sum(current_cluster) / len(current_cluster) clusters.append((center, len(current_cluster))) current_cluster = [sorted_levels[i]] # Last cluster if len(current_cluster) >= 2: center = sum(current_cluster) / len(current_cluster) clusters.append((center, len(current_cluster))) return clusters # ------------------------------------------------------------------ # Scoring # ------------------------------------------------------------------ def _calculate_score(self, all_sfps: list, active_sfp, liquidity_pools: list) -> int: """Composite SFP score 0-100.""" score = 0 # SFP count and reliability if all_sfps: avg_reliability = sum(s.get('reliability', 0) for s in all_sfps) / len(all_sfps) score += min(40, int(avg_reliability * 0.4)) # Active SFP bonus if active_sfp: score += min(30, active_sfp.get('reliability', 0) * 30 // 100) # Liquidity pool presence (more pools = more actionable) pool_count = len(liquidity_pools) score += min(15, pool_count * 5) # SFP count bonus sfp_count = len(all_sfps) score += min(15, sfp_count * 5) return max(0, min(100, score))