Spaces:
Paused
Paused
| """ | |
| analytics.py | |
| Analytics helper functions for Aqua Thistle prototype | |
| Handles exam/test score analysis, progress tracking, insights generation | |
| Added: | |
| - Scenario comparison | |
| - Query parsing for what-if analysis | |
| - Proactive insights generation | |
| - Interactive chart support | |
| """ | |
| import re | |
| def analyze_exam_performance(profile): | |
| """Analyze user's last exam performance and return insights""" | |
| scores = profile.get("last_scores", {}) | |
| recent_exam = profile.get("recent_exam", 0) | |
| goal = profile.get("goal", 1600) | |
| total_points = sum(scores.values()) | |
| score_percentages = {k: round((v/total_points)*100, 1) if total_points > 0 else 0 for k, v in scores.items()} | |
| weakest_subject = min(scores, key=scores.get) if scores else None | |
| strongest_subject = max(scores, key=scores.get) if scores else None | |
| progress_pct = round((recent_exam/goal)*100, 1) if goal > 0 else 0 | |
| points_needed = max(0, goal - recent_exam) | |
| insights = [] | |
| if profile.get("rushing"): | |
| insights.append("You're rushing questions near the end. Practice fundamentals to save time.") | |
| if profile.get("math_weakness"): | |
| insights.append("Math scores have most room for growth. Focus practice here.") | |
| if recent_exam >= 1100: | |
| insights.append(f"Reading score improved 15% over last 3 months! 🎉") | |
| return { | |
| "total_score": recent_exam, | |
| "goal_score": goal, | |
| "progress_pct": progress_pct, | |
| "points_needed": points_needed, | |
| "weakest_subject": weakest_subject, | |
| "strongest_subject": strongest_subject, | |
| "score_breakdown": scores, | |
| "score_percentages": score_percentages, | |
| "insights": insights, | |
| "expected_range": (max(1150, recent_exam-50), min(1320, recent_exam+120)) | |
| } | |
| def calculate_budget_metrics(profile): | |
| """Calculate financial/budget metrics for dashboard""" | |
| budget = profile.get("budget", 2000) | |
| spend = profile.get("spend", 0) | |
| remaining = budget - spend | |
| spend_pct = round((spend/budget)*100, 1) if budget > 0 else 0 | |
| categories = { | |
| "Savings": 1000, | |
| "Food": 400, | |
| "Leisure": 176 | |
| } | |
| potential_savings = [] | |
| if categories["Food"] > 300: | |
| potential_savings.append("Cut 15% on food costs (meal prep)") | |
| if categories["Leisure"] > 150: | |
| potential_savings.append("Reduce leisure spending by $50/month") | |
| return { | |
| "budget": budget, | |
| "spent": spend, | |
| "remaining": remaining, | |
| "spend_percentage": spend_pct, | |
| "categories": categories, | |
| "potential_savings": potential_savings, | |
| "on_track": spend_pct <= 75 | |
| } | |
| def parse_budget_query(query): | |
| """ | |
| Parse user queries for budget changes. | |
| Returns dict with changes or None if no budget query detected. | |
| Examples: | |
| - "What if I save $200 more?" → {"savings": 200} | |
| - "Show me if I spend $500 less on food" → {"food": -500} | |
| - "How about $300 extra income?" → {"budget": 300} | |
| """ | |
| query_lower = query.lower() | |
| # Check if it's a what-if/scenario query | |
| scenario_keywords = ["what if", "show me if", "how about", "suppose", "imagine", "if i"] | |
| if not any(keyword in query_lower for keyword in scenario_keywords): | |
| return None | |
| changes = {} | |
| # Parse amounts (e.g., $200, 200 dollars, $1,000) | |
| amount_patterns = [ | |
| r'\$(\d{1,3}(?:,\d{3})*(?:\.\d{2})?)', # $200 or $1,000.00 | |
| r'(\d{1,3}(?:,\d{3})*)\s*dollars?', # 200 dollars | |
| ] | |
| amounts = [] | |
| for pattern in amount_patterns: | |
| matches = re.findall(pattern, query_lower) | |
| amounts.extend([float(m.replace(',', '')) for m in matches]) | |
| if not amounts: | |
| return None | |
| # Determine direction (increase/decrease) | |
| is_increase = any(word in query_lower for word in ["more", "extra", "increase", "add", "save"]) | |
| is_decrease = any(word in query_lower for word in ["less", "reduce", "cut", "decrease", "lower"]) | |
| multiplier = 1 if is_increase else -1 if is_decrease else 0 | |
| # Determine category | |
| if any(word in query_lower for word in ["food", "eating", "meal", "grocery"]): | |
| changes["food"] = amounts[0] * multiplier | |
| elif any(word in query_lower for word in ["leisure", "entertainment", "fun", "hobby"]): | |
| changes["leisure"] = amounts[0] * multiplier | |
| elif any(word in query_lower for word in ["save", "saving"]): | |
| changes["savings"] = amounts[0] * multiplier | |
| elif any(word in query_lower for word in ["budget", "income", "earn"]): | |
| changes["budget"] = amounts[0] * multiplier | |
| elif any(word in query_lower for word in ["spend", "spending"]): | |
| changes["total_spend"] = amounts[0] * multiplier | |
| else: | |
| # Generic change | |
| changes["general"] = amounts[0] * multiplier | |
| return changes if changes else None | |
| def apply_scenario_changes(profile, changes): | |
| """ | |
| Apply scenario changes to profile and return modified copy. | |
| Original profile is NOT modified. | |
| """ | |
| scenario_profile = profile.copy() | |
| # Deep copy nested dicts | |
| if "last_scores" in scenario_profile: | |
| scenario_profile["last_scores"] = profile["last_scores"].copy() | |
| if "goals_today" in scenario_profile: | |
| scenario_profile["goals_today"] = profile["goals_today"].copy() | |
| if "budget" in changes: | |
| scenario_profile["budget"] = profile["budget"] + changes["budget"] | |
| if "total_spend" in changes: | |
| scenario_profile["spend"] = profile["spend"] + changes["total_spend"] | |
| if "savings" in changes: | |
| # More savings = less spending | |
| scenario_profile["spend"] = profile["spend"] - changes["savings"] | |
| if "food" in changes: | |
| scenario_profile["spend"] = profile["spend"] + changes["food"] | |
| if "leisure" in changes: | |
| scenario_profile["spend"] = profile["spend"] + changes["leisure"] | |
| # Ensure spend doesn't go negative | |
| scenario_profile["spend"] = max(0, scenario_profile["spend"]) | |
| return scenario_profile | |
| def compare_scenarios(current_profile, scenario_profile): | |
| """ | |
| Compare current vs scenario profiles. | |
| Returns dict with differences and insights. | |
| """ | |
| current_metrics = calculate_budget_metrics(current_profile) | |
| scenario_metrics = calculate_budget_metrics(scenario_profile) | |
| diff_spend = scenario_profile["spend"] - current_profile["spend"] | |
| diff_budget = scenario_profile["budget"] - current_profile["budget"] | |
| diff_remaining = scenario_metrics["remaining"] - current_metrics["remaining"] | |
| insights = [] | |
| if diff_spend < 0: | |
| insights.append(f"✅ You'd save {abs(diff_spend):.0f} dollars monthly") | |
| elif diff_spend > 0: | |
| insights.append(f"⚠️ Spending would increase by {diff_spend:.0f} dollars") | |
| if diff_remaining > 0: | |
| insights.append(f"💰 {diff_remaining:.0f} dollars more remaining at month end") | |
| elif diff_remaining < 0: | |
| insights.append(f"📉 {abs(diff_remaining):.0f} dollars less remaining") | |
| if scenario_metrics["on_track"] and not current_metrics["on_track"]: | |
| insights.append("🎯 This would put you back on track!") | |
| elif not scenario_metrics["on_track"] and current_metrics["on_track"]: | |
| insights.append("⚠️ This would put you over budget") | |
| return { | |
| "current": current_metrics, | |
| "scenario": scenario_metrics, | |
| "differences": { | |
| "spend": diff_spend, | |
| "budget": diff_budget, | |
| "remaining": diff_remaining | |
| }, | |
| "insights": insights | |
| } | |
| def generate_proactive_insights(profile, category): | |
| """ | |
| Generate proactive insights based on profile analysis. | |
| These appear automatically, not just in response to queries. | |
| """ | |
| insights = [] | |
| if category == "Finance": | |
| budget_metrics = calculate_budget_metrics(profile) | |
| spend_pct = budget_metrics["spend_percentage"] | |
| # Critical threshold | |
| if spend_pct > 90: | |
| insights.append({ | |
| "type": "error", | |
| "icon": "🚨", | |
| "text": f"You've spent {spend_pct}% of budget - urgent action needed", | |
| "action": "What can I cut immediately?" | |
| }) | |
| # Warning threshold | |
| elif spend_pct > 75: | |
| insights.append({ | |
| "type": "warning", | |
| "icon": "⚠️", | |
| "text": f"At {spend_pct}% of budget - pace yourself", | |
| "action": "Show me where I can save" | |
| }) | |
| # Good standing | |
| elif spend_pct < 60: | |
| insights.append({ | |
| "type": "success", | |
| "icon": "✅", | |
| "text": f"Great job! Only {spend_pct}% spent - you're ahead", | |
| "action": "How much could I save this month?" | |
| }) | |
| # Savings opportunity | |
| remaining = budget_metrics["remaining"] | |
| if remaining > 500: | |
| insights.append({ | |
| "type": "opportunity", | |
| "icon": "💡", | |
| "text": f"You have {remaining:.0f} dollars left - invest in savings?", | |
| "action": "Best high-yield accounts?" | |
| }) | |
| elif category == "Education": | |
| exam_data = analyze_exam_performance(profile) | |
| progress = exam_data["progress_pct"] | |
| if progress < 70: | |
| insights.append({ | |
| "type": "warning", | |
| "icon": "📚", | |
| "text": f"At {progress}% of goal - need {exam_data['points_needed']} more points", | |
| "action": "How can I improve fastest?" | |
| }) | |
| elif progress >= 90: | |
| insights.append({ | |
| "type": "success", | |
| "icon": "🎯", | |
| "text": f"Almost there! {progress}% to goal", | |
| "action": "Final push strategies?" | |
| }) | |
| if exam_data["weakest_subject"]: | |
| weak_score = exam_data["score_breakdown"][exam_data["weakest_subject"]] | |
| if weak_score < 120: | |
| insights.append({ | |
| "type": "opportunity", | |
| "icon": "💪", | |
| "text": f"{exam_data['weakest_subject']} is weakest - most room for growth", | |
| "action": f"Practice {exam_data['weakest_subject']} questions" | |
| }) | |
| return insights | |
| def get_daily_goals_summary(profile): | |
| """Summarize daily goal progress""" | |
| goals = profile.get("goals_today", {}) | |
| completed = sum(1 for v in goals.values() if v >= 100) | |
| in_progress = sum(1 for v in goals.values() if 50 <= v < 100) | |
| needs_attention = sum(1 for v in goals.values() if v < 50) | |
| avg_completion = round(sum(goals.values()) / len(goals), 1) if goals else 0 | |
| return { | |
| "total_goals": len(goals), | |
| "completed": completed, | |
| "in_progress": in_progress, | |
| "needs_attention": needs_attention, | |
| "avg_completion": avg_completion, | |
| "goals": goals | |
| } | |
| def generate_practice_questions(subject, difficulty="medium"): | |
| """Generate practice questions based on weak subject areas""" | |
| question_bank = { | |
| "Reading": [ | |
| "Read passage about climate change and identify main argument", | |
| "Analyze author's tone in historical narrative", | |
| "Compare two texts on same topic, find common themes" | |
| ], | |
| "Writing": [ | |
| "Fix grammar errors in paragraph about economics", | |
| "Improve sentence structure for clarity", | |
| "Choose best transition word for essay flow" | |
| ], | |
| "Reasoning": [ | |
| "Solve logic puzzle with 3 variables", | |
| "Identify pattern in number sequence", | |
| "Draw conclusion from given premises" | |
| ], | |
| "Algebra": [ | |
| "Solve quadratic equation: x² + 5x - 14 = 0", | |
| "Find slope of line through points (2,3) and (5,9)", | |
| "Simplify expression: (3x² - 2x + 1) - (x² + 4x - 3)" | |
| ], | |
| "Geometry": [ | |
| "Calculate area of triangle given base=8, height=6", | |
| "Find angle measure in complementary angle pair", | |
| "Determine volume of cylinder with r=3, h=10" | |
| ] | |
| } | |
| return question_bank.get(subject, ["No questions available for this subject"]) | |
| def get_time_improvement_suggestions(profile): | |
| """Generate time management suggestions based on performance""" | |
| suggestions = [] | |
| if profile.get("rushing"): | |
| suggestions.append("Practice timed sections: 15 min sprints") | |
| suggestions.append("Skip hard questions first, return later") | |
| suggestions.append("Use process of elimination to save time") | |
| if profile.get("math_weakness"): | |
| suggestions.append("Memorize common formulas before test") | |
| suggestions.append("Do 10 quick math drills daily") | |
| suggestions.append("Take full-length practice test this weekend") | |
| suggestions.append("Review mistakes immediately after practice") | |
| return suggestions | |
| def calculate_projected_improvement(current_score, weeks_until_test, study_hours_per_week): | |
| """Project score improvement based on study plan""" | |
| total_study_hours = weeks_until_test * study_hours_per_week | |
| projected_gain = (total_study_hours / 10) * 10 | |
| if current_score > 1400: | |
| projected_gain *= 0.7 | |
| elif current_score > 1200: | |
| projected_gain *= 0.85 | |
| projected_score = min(1600, current_score + int(projected_gain)) | |
| return { | |
| "current": current_score, | |
| "projected": projected_score, | |
| "gain": projected_score - current_score, | |
| "study_hours_needed": total_study_hours, | |
| "confidence": "high" if study_hours_per_week >= 10 else "medium" | |
| } | |
| def get_category_insights(category, profile): | |
| """Generate category-specific insights""" | |
| insights = [] | |
| if category == "Finance": | |
| budget_data = calculate_budget_metrics(profile) | |
| if not budget_data["on_track"]: | |
| insights.append("⚠️ Spending at {}%, consider budget review".format(budget_data["spend_percentage"])) | |
| insights.extend(budget_data["potential_savings"]) | |
| elif category == "Education": | |
| exam_data = analyze_exam_performance(profile) | |
| insights.extend(exam_data["insights"]) | |
| if exam_data["points_needed"] > 0: | |
| insights.append(f"Need {exam_data['points_needed']} more points to reach goal") | |
| elif category == "Family": | |
| insights.append("Schedule weekly check-in calls") | |
| insights.append("Send updates about school progress") | |
| elif category == "Friends": | |
| insights.append("Plan low-cost hangouts to stay connected") | |
| insights.append("Join study groups for accountability") | |
| elif category == "Weekend/Vacation": | |
| insights.append("Research free local events") | |
| insights.append("Budget $150-200 for weekend activities") | |
| return insights | |