"""Multi-Timeframe Signal Confirmation. Analyzes multiple timeframes simultaneously and only confirms signals when 2+ timeframes agree. Higher timeframes get more weight. """ from __future__ import annotations import logging from typing import Optional logger = logging.getLogger(__name__) # Timeframe hierarchy: higher TF = more weight TF_WEIGHTS = { '1m': 0.05, '15m': 0.10, '30m': 0.15, '1h': 0.20, '4h': 0.30, '1D': 0.40, '1W': 0.50, } class MultiTimeframeAnalyzer: """Analyze signals across multiple timeframes for confirmation.""" def __init__(self): self.min_confirming_tfs = 2 # Need at least 2 TFs agreeing self.primary_tf_boost = 1.5 # Boost when primary TF agrees def analyze( self, asset_id: str, primary_tf: str = '1D', timeframes: list[str] | None = None, source: str = None, ) -> dict: """Analyze asset across multiple timeframes. Args: asset_id: The asset/stock to analyze primary_tf: Primary analysis timeframe timeframes: List of timeframes to check (default: auto-select) source: Data source filter Returns: dict with confirmation status, aligned signal, score boost/penalty """ if timeframes is None: timeframes = self._select_timeframes(primary_tf) # Get signals for each timeframe tf_signals = {} for tf in timeframes: try: signal_data = self._get_signal_for_tf(asset_id, tf, source) if signal_data: tf_signals[tf] = signal_data except Exception as e: logger.warning(f'MTF analysis failed for {tf}: {e}') if not tf_signals: return self._no_data_result(primary_tf) return self._compute_confirmation(tf_signals, primary_tf) def _select_timeframes(self, primary_tf: str) -> list[str]: """Auto-select companion timeframes based on primary.""" tf_groups = { '1h': ['15m', '1h', '4h'], '4h': ['1h', '4h', '1D'], '1D': ['4h', '1D', '1W'], '1W': ['1D', '1W'], } return tf_groups.get(primary_tf, [primary_tf]) def _get_signal_for_tf( self, asset_id: str, timeframe: str, source: str = None ) -> Optional[dict]: """Get the latest signal for a asset at a specific timeframe.""" try: from app.models.signal import TradingSignal from app.models.asset import Asset query = TradingSignal.query.filter( TradingSignal.asset_id == asset_id, TradingSignal.timeframe == timeframe, TradingSignal.status == 'active', ).order_by(TradingSignal.created_at.desc()).first() if query: return { 'signal': query.signal_type, 'score': float(query.score), 'confidence': query.confidence, 'timeframe': timeframe, } except Exception: pass # Fallback: try to compute signal on-the-fly try: return self._compute_signal_for_tf(asset_id, timeframe, source) except Exception: return None def _compute_signal_for_tf( self, asset_id: str, timeframe: str, source: str = None ) -> Optional[dict]: """Compute a lightweight signal for a timeframe. Uses simple technical indicators (SMA cross + RSI) for speed, without full engine pipeline. """ try: from app.models.ohlcv import OHLCVData from app.extensions import db import numpy as np query = db.session.query( OHLCVData.close ).filter( OHLCVData.asset_id == asset_id, OHLCVData.timeframe == timeframe, ) if source: query = query.filter(OHLCVData.source == source) rows = query.order_by(OHLCVData.timestamp.desc()).limit(50).all() if len(rows) < 30: return None closes = np.array([float(r.close) for r in reversed(rows)]) # Simple SMA cross signal sma7 = np.mean(closes[-7:]) sma21 = np.mean(closes[-21:]) sma_score = (sma7 / sma21 - 1) * 100 * 30 # RSI deltas = np.diff(closes[-15:]) gains = np.mean(deltas[deltas > 0]) if any(deltas > 0) else 0 losses = abs(np.mean(deltas[deltas < 0])) if any(deltas < 0) else 0.001 rs = gains / losses rsi = 100 - 100 / (1 + rs) # Combined score rsi_score = 0 if rsi > 70: rsi_score = -20 elif rsi < 30: rsi_score = 20 score = sma_score + rsi_score if score > 20: signal = 'BUY' elif score < -20: signal = 'SELL' else: signal = 'HOLD' return { 'signal': signal, 'score': round(score, 1), 'confidence': 'Medium' if abs(score) > 30 else 'Low', 'timeframe': timeframe, } except Exception: return None def _compute_confirmation( self, tf_signals: dict[str, dict], primary_tf: str ) -> dict: """Compute multi-timeframe confirmation result.""" buy_score = 0.0 sell_score = 0.0 hold_score = 0.0 confirming_tfs = [] conflicting_tfs = [] primary_signal = tf_signals.get(primary_tf, {}).get('signal', 'HOLD') for tf, data in tf_signals.items(): weight = TF_WEIGHTS.get(tf, 0.2) signal = data.get('signal', 'HOLD') if signal == 'BUY': buy_score += weight elif signal == 'SELL': sell_score += weight else: hold_score += weight if signal == primary_signal and signal != 'HOLD': confirming_tfs.append(tf) elif signal != primary_signal and signal != 'HOLD' and primary_signal != 'HOLD': conflicting_tfs.append(tf) # Determine confirmed signal total = buy_score + sell_score + hold_score if total == 0: total = 1 is_confirmed = len(confirming_tfs) >= self.min_confirming_tfs n_confirming = len(confirming_tfs) # Score adjustment if is_confirmed: score_boost = 10 * n_confirming # +10 per confirming TF elif conflicting_tfs: score_boost = -15 * len(conflicting_tfs) # -15 per conflicting TF else: score_boost = 0 # Consensus signal if buy_score > sell_score and buy_score > hold_score: consensus = 'BUY' elif sell_score > buy_score and sell_score > hold_score: consensus = 'SELL' else: consensus = 'HOLD' return { 'is_confirmed': is_confirmed, 'primary_signal': primary_signal, 'consensus_signal': consensus, 'confirming_timeframes': confirming_tfs, 'conflicting_timeframes': conflicting_tfs, 'score_boost': score_boost, 'buy_weight': round(buy_score / total, 3), 'sell_weight': round(sell_score / total, 3), 'hold_weight': round(hold_score / total, 3), 'tf_signals': tf_signals, 'n_timeframes_analyzed': len(tf_signals), } def _no_data_result(self, primary_tf: str) -> dict: return { 'is_confirmed': False, 'primary_signal': 'HOLD', 'consensus_signal': 'HOLD', 'confirming_timeframes': [], 'conflicting_timeframes': [], 'score_boost': 0, 'buy_weight': 0, 'sell_weight': 0, 'hold_weight': 1.0, 'tf_signals': {}, 'n_timeframes_analyzed': 0, }