import pandas as pd from datetime import datetime, timedelta import logging from typing import Dict, List from .data_fetcher import DataFetcher from .indicators import IndicatorCalculator from .divergence import DivergenceDetector from .strategies import StrategyEvaluator logger = logging.getLogger("TradingEngine") logger.setLevel(logging.INFO) class TradingEngine: TIMEFRAMES = ['5min', '15min', '30min', '1h', 'day'] STRATEGIES = { 'momentum': StrategyEvaluator.strategy_momentum, 'vwap_pullback': StrategyEvaluator.strategy_vwap_pullback, 'ema_crossover': StrategyEvaluator.strategy_ema_crossover, 'rsi_volume_reversal': StrategyEvaluator.strategy_rsi_volume_reversal } def __init__(self, zerodha_client, db_manager=None): self.data_fetcher = DataFetcher(zerodha_client) self.db_manager = db_manager def analyze_symbol(self, symbol: str, timeframes: List[str] = None) -> dict: logger.info(f"Analyzing {symbol} with multi-timeframe strategy") if timeframes is None: timeframes = self.TIMEFRAMES # Fetch multi-timeframe data timeframe_data = self.data_fetcher.fetch_multi_timeframe(symbol, timeframes) # Calculate indicators and pivots for each timeframe pivots = {} for tf, df in timeframe_data.items(): df = IndicatorCalculator.add_all_indicators(df) df['supertrend'], df['supertrend_direction'] = IndicatorCalculator.calculate_supertrend(df) timeframe_data[tf] = df # Calculate pivots for relevant timeframes if tf in ['30min', 'day', 'week', 'month']: pivots[tf] = IndicatorCalculator.calculate_pivot_points(df) # Run multi-timeframe strategy strategy_result = StrategyEvaluator.analyze_multi_timeframe(timeframe_data, pivots) # Detect divergences daily_df = timeframe_data.get('day') rsi_div = DivergenceDetector.detect_rsi_divergence(daily_df) if daily_df is not None else None macd_div = DivergenceDetector.detect_macd_divergence(daily_df) if daily_df is not None else None # Get 52-week high/low yearly_df = self.data_fetcher.fetch_ohlc(symbol, 'day', days_back=365) yearly_high = yearly_df['high'].max() if not yearly_df.empty else 0 yearly_low = yearly_df['low'].min() if not yearly_df.empty else 0 # Format final output signal_data = { "symbol": symbol, "decision": strategy_result['signal'], "strategy": strategy_result.get('strategy_name', 'Multi-Timeframe'), "entry_price": daily_df['close'].iloc[-1] if daily_df is not None else 0, "stop_loss": strategy_result.get('stop_loss', 0), "target_1": strategy_result.get('targets', [])[0] if strategy_result.get('targets') else 0, "target_2": strategy_result.get('targets', [])[1] if strategy_result.get('targets') and len(strategy_result['targets']) > 1 else 0, "confidence_score": strategy_result['confidence'], "divergence": { "RSI": rsi_div or "None", "MACD": macd_div or "None" }, "pivots": pivots, "yearly_high": yearly_high, "yearly_low": yearly_low, "status": "Ready", "timestamp": datetime.now().isoformat() } # Save to database if available if self.db_manager: self.db_manager.save_signal(signal_data) return signal_data def analyze_symbol_old(self, symbol: str, timeframes: List[str] = None) -> dict: """Full analysis pipeline for a single symbol""" logger.info(f"Analyzing {symbol}") if timeframes is None: timeframes = self.TIMEFRAMES # Fetch multi-timeframe data timeframe_data = self.data_fetcher.fetch_multi_timeframe(symbol, timeframes) # Calculate indicators for each timeframe for tf, df in timeframe_data.items(): timeframe_data[tf] = IndicatorCalculator.add_all_indicators(df) # Use daily data for primary analysis daily_df = timeframe_data.get('day') if daily_df is None or daily_df.empty: logger.error(f"No daily data for {symbol}") return self._create_hold_signal(symbol) current_close = daily_df['close'].iloc[-1] # Run all strategies strategy_results = [] for name, strategy in self.STRATEGIES.items(): result = strategy(daily_df) result['strategy_name'] = name strategy_results.append(result) # Find the strongest signal best_signal = {'signal': 'HOLD', 'confidence': 0} for result in strategy_results: if result['confidence'] > best_signal['confidence']: best_signal = result # Detect divergences rsi_div = DivergenceDetector.detect_rsi_divergence(daily_df) macd_div = DivergenceDetector.detect_macd_divergence(daily_df) # Format final output signal_data = { "symbol": symbol, "decision": best_signal['signal'], "strategy": best_signal.get('strategy_name', 'momentum'), "entry_price": current_close, "stop_loss": best_signal.get('stop_loss', current_close * 0.97), "target_1": best_signal.get('targets', [])[0] if best_signal.get('targets') else current_close * 1.03, "target_2": best_signal.get('targets', [])[1] if best_signal.get('targets') and len(best_signal['targets']) > 1 else current_close * 1.06, "confidence_score": best_signal['confidence'], "divergence": { "RSI": rsi_div or "None", "MACD": macd_div or "None" }, "status": "Ready", "timestamp": datetime.now().isoformat() } # Save to database if available if self.db_manager: self.db_manager.save_signal(signal_data) return signal_data def _create_hold_signal(self, symbol): return { "symbol": symbol, "decision": "HOLD", "strategy": "No signal", "entry_price": 0, "stop_loss": 0, "target_1": 0, "target_2": 0, "confidence_score": 0, "divergence": {"RSI": "None", "MACD": "None"}, "status": "No data", "timestamp": datetime.now().isoformat() } def execute_daily_pipeline(self, user_id: str, auto_trade: bool = False): """Full daily execution pipeline""" logger.info(f"Running daily pipeline for user {user_id}") # Fetch portfolio from DB (pseudo-code) portfolio = self._get_user_portfolio(user_id) # Implement this signals = [] for item in portfolio: try: symbol = item['symbol'] signal = self.analyze_symbol(symbol) signals.append(signal) # Execute orders if enabled if auto_trade and signal['decision'] in ('BUY', 'SELL') and signal['confidence_score'] > 80: self.execute_order(signal, user_id) except Exception as e: logger.error(f"Error processing {item['symbol']}: {str(e)}") return signals def _get_user_portfolio(self, user_id: str) -> List[Dict]: """Fetch user portfolio from database""" # Implement your DB logic here return [{"symbol": "RELIANCE"}, {"symbol": "TCS"}] # Mock data def execute_order(self, signal: dict, user_id: str): """Execute order through Zerodha API with risk management""" # Implement order execution with proper position sizing # Use your existing ZerodhaClient.place_order method pass