""" Strategic Scoring Engine based on GEMINI.md Framework Implements the automated strategic fit analysis from Part IV, Section 4.4, Step 3 """ import re import logging from typing import Dict, List, Tuple from app.models import User, GrantOpportunity, OpportunityScore from app import db class ScoringEngine: """ AI-powered scoring engine that evaluates grant opportunities based on: 1. Agency Profile Score (0-30 points) 2. Market Need Score (0-40 points) 3. Competitive Landscape Score (0-30 points) """ def __init__(self): self.logger = logging.getLogger(__name__) # Agency Profiles from GEMINI.md Table 1 self.agency_profiles = { 'Department of Education': { 'type': 'product', 'focus': 'commercialization', 'keywords': ['product', 'commercial', 'market', 'viable', 'scalable', 'deployment'] }, 'ED': { 'type': 'product', 'focus': 'commercialization', 'keywords': ['product', 'commercial', 'market', 'viable', 'scalable', 'deployment'] }, 'IES': { 'type': 'product', 'focus': 'commercialization', 'keywords': ['product', 'commercial', 'market', 'viable', 'scalable', 'deployment'] }, 'National Science Foundation': { 'type': 'research', 'focus': 'knowledge', 'keywords': ['research', 'knowledge', 'curriculum', 'education', 'workforce', 'inclusive'] }, 'NSF': { 'type': 'research', 'focus': 'knowledge', 'keywords': ['research', 'knowledge', 'curriculum', 'education', 'workforce', 'inclusive'] } } # Market Need Keywords from GEMINI.md Table 2 self.market_need_keywords = { 'adaptive assessment': 20, 'special education': 25, 'students with disabilities': 25, 'digital literacy': 15, 'misinformation': 15, 'immersive learning': 10, 'virtual reality': 10, 'augmented reality': 10, 'civics education': 20, 'dyslexia': 20, 'teacher workload': 15, 'personalized learning': 10, 'artificial intelligence': 15, 'machine learning': 12, 'natural language processing': 12, 'educational technology': 8, 'learning analytics': 10, 'intelligent tutoring': 15, 'adaptive tutoring': 15, 'assessment generation': 12, 'feedback systems': 8, 'educational games': 10, 'simulation': 12, 'ethics': 10, 'ai ethics': 15 } # Competitive Landscape from GEMINI.md Table 2 self.competitive_landscape = { 'red_ocean': [ 'lesson planner', 'teacher assistant', 'grading tool', 'general ai tutor', 'basic chatbot', 'simple quiz generator' ], 'blue_ocean': [ 'civics simulation', 'ethical ai', 'dyslexia tutor', 'adaptive assessment', 'cultural responsiveness', 'special education ai', 'immersive learning', 'performance-based assessment', 'misinformation detection' ] } def calculate_agency_profile_score(self, opportunity: GrantOpportunity, user: User) -> Tuple[int, str]: """ Calculate Agency Profile Score (0-30 points) Based on alignment between agency mission and opportunity content """ text = (opportunity.title + ' ' + (opportunity.description or '') + ' ' + (opportunity.full_text or '')).lower() agency = opportunity.agency score = 0 reason = "No agency profile match found." # Find matching agency profile agency_profile = None for agency_name, profile in self.agency_profiles.items(): if agency_name.lower() in agency.lower(): agency_profile = profile break if not agency_profile: return score, reason # Check for alignment keywords keyword_matches = [] for keyword in agency_profile['keywords']: if keyword in text: keyword_matches.append(keyword) if keyword_matches: if agency_profile['type'] == 'product': score = 30 if len(keyword_matches) >= 3 else 20 if len(keyword_matches) >= 2 else 15 reason = f"Strong alignment with {agency} product-focused mission. Keywords found: {', '.join(keyword_matches)}" else: # research type score = 30 if len(keyword_matches) >= 3 else 20 if len(keyword_matches) >= 2 else 15 reason = f"Strong alignment with {agency} research-focused mission. Keywords found: {', '.join(keyword_matches)}" else: score = 5 reason = f"Weak alignment with {agency} mission. No key alignment keywords found." return score, reason def calculate_market_need_score(self, opportunity: GrantOpportunity, user: User) -> Tuple[int, str]: """ Calculate Market Need Score (0-40 points) Based on high-value market trends and technology opportunities """ text = (opportunity.title + ' ' + (opportunity.description or '') + ' ' + (opportunity.full_text or '')).lower() total_score = 0 matched_keywords = [] # Check for market need keywords for keyword, weight in self.market_need_keywords.items(): if keyword.lower() in text: total_score += weight matched_keywords.append(keyword.title()) # Cap the score at 40 score = min(total_score, 40) if matched_keywords: reason = f"High market alignment. Keywords found: {', '.join(matched_keywords[:5])}{'...' if len(matched_keywords) > 5 else ''}" else: reason = "No specific high-value market keywords identified." return score, reason def calculate_competitive_landscape_score(self, opportunity: GrantOpportunity, user: User) -> Tuple[int, str]: """ Calculate Competitive Landscape Score (0-30 points) Based on whether opportunity targets crowded vs. niche markets """ text = (opportunity.title + ' ' + (opportunity.description or '') + ' ' + (opportunity.full_text or '')).lower() # Check for blue ocean opportunities (favorable) blue_ocean_matches = [] for keyword in self.competitive_landscape['blue_ocean']: if keyword.lower() in text: blue_ocean_matches.append(keyword) # Check for red ocean opportunities (challenging) red_ocean_matches = [] for keyword in self.competitive_landscape['red_ocean']: if keyword.lower() in text: red_ocean_matches.append(keyword) if blue_ocean_matches: score = 30 if len(blue_ocean_matches) >= 2 else 25 reason = f"Favorable competitive landscape. Targets niche 'blue ocean' areas: {', '.join(blue_ocean_matches)}" elif red_ocean_matches: score = 5 reason = f"Challenging competitive landscape. Enters crowded 'red ocean' market: {', '.join(red_ocean_matches)}" else: score = 15 reason = "Neutral competitive landscape. No clear competitive indicators found." return score, reason def calculate_total_score(self, opportunity: GrantOpportunity, user: User) -> OpportunityScore: """ Calculate comprehensive opportunity score for a user Returns OpportunityScore object ready for database storage """ # Calculate individual scores agency_score, agency_reason = self.calculate_agency_profile_score(opportunity, user) market_score, market_reason = self.calculate_market_need_score(opportunity, user) competitive_score, competitive_reason = self.calculate_competitive_landscape_score(opportunity, user) total_score = agency_score + market_score + competitive_score # Generate AI recommendation based on score if total_score >= 80: recommendation = "HIGHLY RECOMMENDED: This opportunity shows exceptional alignment with your profile and market positioning." confidence = 0.9 elif total_score >= 60: recommendation = "RECOMMENDED: This opportunity shows strong potential and good strategic fit." confidence = 0.75 elif total_score >= 40: recommendation = "CONSIDER: This opportunity has moderate potential but may require careful evaluation." confidence = 0.6 else: recommendation = "LOW PRIORITY: This opportunity shows limited alignment with optimal strategic positioning." confidence = 0.4 # Create OpportunityScore object score_obj = OpportunityScore( user_id=user.id, opportunity_id=opportunity.id, agency_profile_score=agency_score, market_need_score=market_score, competitive_landscape_score=competitive_score, total_score=total_score, agency_profile_reason=agency_reason, market_need_reason=market_reason, competitive_landscape_reason=competitive_reason, ai_recommendation=recommendation, confidence_level=confidence, model_version="ScoringEngine_v1.0" ) self.logger.info(f"Calculated score for opportunity {opportunity.id}: {total_score}/100") return score_obj def batch_score_opportunities(self, user: User, opportunities: List[GrantOpportunity]) -> List[OpportunityScore]: """ Score multiple opportunities for a user in batch """ scores = [] for opportunity in opportunities: try: # Check if score already exists existing_score = OpportunityScore.query.filter_by( user_id=user.id, opportunity_id=opportunity.id ).first() if existing_score: scores.append(existing_score) continue # Calculate new score score = self.calculate_total_score(opportunity, user) db.session.add(score) scores.append(score) except Exception as e: self.logger.error(f"Error scoring opportunity {opportunity.id}: {str(e)}") continue try: db.session.commit() self.logger.info(f"Successfully scored {len(scores)} opportunities for user {user.id}") except Exception as e: db.session.rollback() self.logger.error(f"Error saving scores: {str(e)}") raise return scores def get_top_opportunities(self, user: User, limit: int = 10) -> List[Tuple[GrantOpportunity, OpportunityScore]]: """ Get top-scored opportunities for a user """ results = db.session.query(GrantOpportunity, OpportunityScore)\ .join(OpportunityScore, OpportunityScore.opportunity_id == GrantOpportunity.id)\ .filter(OpportunityScore.user_id == user.id)\ .filter(GrantOpportunity.application_due_date > db.func.now())\ .order_by(OpportunityScore.total_score.desc())\ .limit(limit)\ .all() return results def update_scoring_criteria(self, new_keywords: Dict[str, int] = None, new_agencies: Dict = None): """ Update scoring criteria dynamically (for system learning/adaptation) """ if new_keywords: self.market_need_keywords.update(new_keywords) self.logger.info(f"Updated market need keywords: {new_keywords}") if new_agencies: self.agency_profiles.update(new_agencies) self.logger.info(f"Updated agency profiles: {new_agencies}")