""" AI Plan Generation Service Generates daily trading plans using AI based on user's indicator preferences """ from typing import List, Optional, Dict from datetime import date from sqlalchemy.orm import Session import json from app.models.models import UserIndicatorPreferences, AIPlanGeneration from app.schemas.schemas import ( AIPlanGenerationRequest, AIPlanGenerationResponse, MarketBias, PriceData ) from app.services.openrouter import openrouter_service class AIPlanService: """Service for AI-powered trading plan generation""" def _get_user_indicator_preferences(self, db: Session, user_id: Optional[str] = None) -> List[UserIndicatorPreferences]: """Fetch user's enabled indicator preferences""" query = db.query(UserIndicatorPreferences).filter( UserIndicatorPreferences.enabled == True ) if user_id: query = query.filter(UserIndicatorPreferences.user_id == user_id) return query.order_by(UserIndicatorPreferences.priority.desc()).all() def _build_ai_prompt( self, request: AIPlanGenerationRequest, indicator_preferences: List[UserIndicatorPreferences] ) -> str: """Build comprehensive prompt for AI plan generation""" indicator_names = [pref.indicator_name for pref in indicator_preferences] if indicator_preferences else [] prompt = f"""You are an expert gold (XAU/USD) trading analyst. Generate a detailed daily trading plan based on the following information: CURRENT MARKET DATA: - Current Price: ${request.current_price:.2f} - User's Risk Tolerance: {request.risk_tolerance} - Available Capital: ${request.user_capital if request.user_capital else 'Not specified'} USER'S PREFERRED TECHNICAL INDICATORS: {', '.join(indicator_names) if indicator_names else 'No specific preferences - use standard analysis'} INDICATOR DETAILS: """ for pref in indicator_preferences: prompt += f"- {pref.indicator_name} (Priority: {pref.priority})" if pref.parameters: prompt += f" - Parameters: {json.dumps(pref.parameters)}" if pref.notes: prompt += f" - Notes: {pref.notes}" prompt += "\n" if request.price_data and len(request.price_data) > 0: recent_prices = request.price_data[-10:] # Last 10 data points prompt += f"\nRECENT PRICE ACTION (last {len(recent_prices)} periods):\n" for i, pd in enumerate(recent_prices, 1): prompt += f" {i}. Open: ${pd.open:.2f}, High: ${pd.high:.2f}, Low: ${pd.low:.2f}, Close: ${pd.close:.2f}\n" if request.indicators_data: prompt += f"\nCURRENT INDICATOR VALUES:\n" for indicator, value in request.indicators_data.items(): prompt += f"- {indicator}: {value}\n" prompt += """ Please generate a comprehensive daily trading plan with the following structure: 1. MARKET BIAS: Determine if the market is BULLISH, BEARISH, or NEUTRAL 2. CONFIDENCE: Your confidence level in this analysis (0-100) 3. DAILY TARGET: Suggested profit target in dollars (be realistic based on user's capital and risk tolerance) 4. MAX LOSS: Maximum acceptable loss for the day (align with risk tolerance) 5. ENTRY ZONE: Recommended price range for entering positions (min and max) 6. TARGET PRICE: Primary profit-taking level 7. STOP LOSS: Stop-loss level to protect capital 8. SUPPORT LEVELS: 3-5 key support levels below current price 9. RESISTANCE LEVELS: 3-5 key resistance levels above current price 10. MAX TRADES: Recommended maximum number of trades for the day 11. TRADING NOTES: Detailed strategy notes including: - Why this bias? - What indicators support this view? - What to watch for during the day? - Risk management considerations - Market conditions and factors 12. REASONING: Detailed explanation of your analysis and why you recommend this plan Format your response as a valid JSON object with these exact keys: { "market_bias": "BULLISH" | "BEARISH" | "NEUTRAL", "confidence": 75.0, "daily_target": 500.0, "max_loss": 250.0, "entry_zone_min": 2010.0, "entry_zone_max": 2015.0, "target_price": 2040.0, "stop_loss": 2005.0, "support_levels": [2000.0, 1990.0, 1980.0], "resistance_levels": [2020.0, 2030.0, 2040.0], "max_trades": 3, "trading_notes": "Detailed strategy notes here...", "reasoning": "Full analysis and reasoning here..." } Be specific, actionable, and realistic. Consider the user's risk tolerance and preferred indicators heavily in your analysis. """ return prompt async def generate_plan( self, db: Session, request: AIPlanGenerationRequest, user_id: Optional[str] = None ) -> AIPlanGenerationResponse: """Generate an AI-powered trading plan""" # Get user's indicator preferences if requested indicator_preferences = [] if request.use_indicator_preferences: indicator_preferences = self._get_user_indicator_preferences(db, user_id) # Build AI prompt prompt = self._build_ai_prompt(request, indicator_preferences) # Call AI service try: # Use OpenRouter service to get AI response ai_response = await openrouter_service.generate_trading_plan(prompt) # Parse AI response (assuming it returns JSON) if isinstance(ai_response, str): plan_data = json.loads(ai_response) else: plan_data = ai_response # Create database record db_plan = AIPlanGeneration( user_id=user_id, plan_date=date.today(), market_bias=plan_data.get("market_bias", "NEUTRAL"), confidence=plan_data.get("confidence", 50.0), daily_target=plan_data.get("daily_target"), max_loss=plan_data.get("max_loss"), entry_zone_min=plan_data.get("entry_zone_min"), entry_zone_max=plan_data.get("entry_zone_max"), target_price=plan_data.get("target_price"), stop_loss=plan_data.get("stop_loss"), support_levels=plan_data.get("support_levels", []), resistance_levels=plan_data.get("resistance_levels", []), max_trades=plan_data.get("max_trades", 3), trading_notes=plan_data.get("trading_notes"), reasoning=plan_data.get("reasoning"), indicators_used=[pref.indicator_name for pref in indicator_preferences], market_conditions={ "current_price": request.current_price, "risk_tolerance": request.risk_tolerance, }, ai_model=openrouter_service.model, accepted=False, modified=False ) db.add(db_plan) db.commit() db.refresh(db_plan) # Return response return AIPlanGenerationResponse( id=db_plan.id, plan_date=str(db_plan.plan_date), market_bias=MarketBias(db_plan.market_bias), confidence=db_plan.confidence, daily_target=db_plan.daily_target, max_loss=db_plan.max_loss, entry_zone_min=db_plan.entry_zone_min, entry_zone_max=db_plan.entry_zone_max, target_price=db_plan.target_price, stop_loss=db_plan.stop_loss, support_levels=db_plan.support_levels, resistance_levels=db_plan.resistance_levels, max_trades=db_plan.max_trades, trading_notes=db_plan.trading_notes, indicators_used=db_plan.indicators_used, reasoning=db_plan.reasoning, market_conditions=db_plan.market_conditions, ai_model=db_plan.ai_model, created_at=db_plan.created_at ) except json.JSONDecodeError as e: raise Exception(f"Failed to parse AI response: {str(e)}") except Exception as e: raise Exception(f"AI plan generation failed: {str(e)}") async def get_plan_history( self, db: Session, user_id: Optional[str] = None, limit: int = 10 ) -> List[AIPlanGenerationResponse]: """Get historical AI-generated plans""" query = db.query(AIPlanGeneration) if user_id: query = query.filter(AIPlanGeneration.user_id == user_id) plans = query.order_by(AIPlanGeneration.created_at.desc()).limit(limit).all() return [ AIPlanGenerationResponse( id=plan.id, plan_date=str(plan.plan_date), market_bias=MarketBias(plan.market_bias), confidence=plan.confidence, daily_target=plan.daily_target, max_loss=plan.max_loss, entry_zone_min=plan.entry_zone_min, entry_zone_max=plan.entry_zone_max, target_price=plan.target_price, stop_loss=plan.stop_loss, support_levels=plan.support_levels, resistance_levels=plan.resistance_levels, max_trades=plan.max_trades, trading_notes=plan.trading_notes, indicators_used=plan.indicators_used, reasoning=plan.reasoning, market_conditions=plan.market_conditions, ai_model=plan.ai_model, created_at=plan.created_at ) for plan in plans ] async def submit_feedback( self, db: Session, plan_id: int, accepted: bool, modified: bool = False, feedback: Optional[str] = None ): """Submit user feedback on an AI-generated plan""" plan = db.query(AIPlanGeneration).filter(AIPlanGeneration.id == plan_id).first() if not plan: raise Exception("Plan not found") plan.accepted = accepted plan.modified = modified plan.feedback = feedback db.commit() db.refresh(plan) return plan # Global instance ai_plan_service = AIPlanService()