from datetime import datetime, timedelta import numpy as np from sqlalchemy.orm import Session from app.database.models import Account, Transaction, Goal, Investment, Subscription def get_cashflow_metrics(db: Session, user_id: str, days: int = 90): """ Computes daily average income and spending based on historical transactions. """ # Fetch checking & savings accounts for user accounts = db.query(Account).filter(Account.user_id == user_id).all() account_ids = [acc.id for acc in accounts] if not account_ids: return 0.0, 0.0, 0.0 # Current balance, avg daily income, avg daily spending current_balance = sum(acc.balance for acc in accounts) # Fetch recent transactions cutoff = datetime.utcnow() - timedelta(days=days) txns = db.query(Transaction).filter( Transaction.account_id.in_(account_ids), Transaction.timestamp >= cutoff ).all() if not txns: return current_balance, 0.0, 0.0 total_income = sum(t.amount for t in txns if t.type.lower() == "credit") total_spending = sum(t.amount for t in txns if t.type.lower() == "debit") avg_daily_income = total_income / days avg_daily_spending = total_spending / days return current_balance, avg_daily_income, avg_daily_spending def predict_future_balance(db: Session, user_id: str, projection_days: int = 90): """ Predicts future balances and returns trend description. """ current_balance, daily_income, daily_spending = get_cashflow_metrics(db, user_id) net_daily = daily_income - daily_spending projected_balance = max(0.0, current_balance + (net_daily * projection_days)) # Calculate percentage change if current_balance > 0: percent_change = (projected_balance - current_balance) / current_balance * 100 else: percent_change = 0.0 # Generate human-friendly description if percent_change < 0: insight = f"If current spending continues, your total balance may decrease by {abs(percent_change):.1f}% (down to ${projected_balance:,.2f}) in {projection_days} days." elif percent_change > 0: insight = f"Based on current trends, your total balance is projected to grow by {percent_change:.1f}% (up to ${projected_balance:,.2f}) in {projection_days} days." else: insight = "Your financial trajectory is steady with minor balance fluctuations." # Generate daily data points for charts chart_data = [] for day in range(0, projection_days + 1, 5): val = max(0.0, current_balance + (net_daily * day)) date_str = (datetime.utcnow() + timedelta(days=day)).strftime("%Y-%m-%d") chart_data.append({"date": date_str, "balance": round(val, 2)}) return { "current_balance": round(current_balance, 2), "projected_balance": round(projected_balance, 2), "percent_change": round(percent_change, 2), "net_daily": round(net_daily, 2), "insight": insight, "chart_data": chart_data } def forecast_savings_and_investments(db: Session, user_id: str, projection_months: int = 12): """ Projects savings and investment growth. """ accounts = db.query(Account).filter(Account.user_id == user_id).all() savings_balance = sum(acc.balance for acc in accounts if acc.type.lower() == "savings") checking_balance = sum(acc.balance for acc in accounts if acc.type.lower() == "checking") investments = db.query(Investment).filter(Investment.user_id == user_id).all() total_invested = sum(inv.current_value for inv in investments) # Subscriptions and recurring bills subs = db.query(Subscription).filter(Subscription.user_id == user_id, Subscription.active == True).all() monthly_sub_cost = sum(sub.amount if sub.billing_cycle.lower() == "monthly" else (sub.amount / 12) for sub in subs) # Let's assume standard default rates if not specified savings_apr = 0.04 # 4% APY investment_apr = 0.08 # 8% APY # We assume the user saves 10% of their net income monthly (derived from transaction history) _, daily_income, daily_spending = get_cashflow_metrics(db, user_id) monthly_income = daily_income * 30.4 monthly_spending = daily_spending * 30.4 net_monthly = max(0.0, monthly_income - monthly_spending) monthly_savings_addition = net_monthly * 0.5 # Put 50% of net into savings monthly_investment_addition = net_monthly * 0.3 # Put 30% of net into investments savings_data = [] investment_data = [] debt_data = [] current_savings = savings_balance current_inv = total_invested # Let's model a baseline debt if the user has a Goal of type "debt" or a general dummy debt # We will look for Goals containing "debt" or "loan" goals = db.query(Goal).filter(Goal.user_id == user_id).all() debt_goal = next((g for g in goals if "debt" in g.title.lower() or "loan" in g.title.lower()), None) total_debt = 5000.0 # Default initial simulated debt if none found if debt_goal: total_debt = max(0.0, debt_goal.target_amount - debt_goal.current_amount) monthly_debt_payment = min(total_debt, max(150.0, net_monthly * 0.1)) # Assume 10% of net or at least $150 for month in range(0, projection_months + 1): # Compounding savings if month > 0: current_savings = (current_savings + monthly_savings_addition) * (1 + savings_apr / 12) current_inv = (current_inv + monthly_investment_addition) * (1 + investment_apr / 12) total_debt = max(0.0, total_debt - monthly_debt_payment) label = f"Month {month}" savings_data.append({"month": label, "amount": round(current_savings, 2)}) investment_data.append({"month": label, "amount": round(current_inv, 2)}) debt_data.append({"month": label, "amount": round(total_debt, 2)}) return { "projection_months": projection_months, "monthly_savings_addition": round(monthly_savings_addition, 2), "monthly_investment_addition": round(monthly_investment_addition, 2), "savings_growth": savings_data, "investment_growth": investment_data, "debt_decline": debt_data, "total_projected_savings": round(current_savings, 2), "total_projected_investments": round(current_inv, 2), "total_remaining_debt": round(total_debt, 2) } def simulate_future_scenarios(db: Session, user_id: str, projection_months: int = 6): """ Simulates three scenarios: Status Quo, Frugal (cut spending 20%), and Luxury (increase spending 15%). """ current_balance, daily_income, daily_spending = get_cashflow_metrics(db, user_id) monthly_income = daily_income * 30.4 monthly_spending = daily_spending * 30.4 scenarios = { "status_quo": {"spend_mult": 1.0, "name": "Status Quo (Current spending)"}, "frugal": {"spend_mult": 0.8, "name": "Frugal Mode (Cut non-essentials by 20%)"}, "lifestyle_inflation": {"spend_mult": 1.15, "name": "Lifestyle Inflation (+15% spending)"} } results = {} for key, config in scenarios.items(): mult = config["spend_mult"] projected_spend = monthly_spending * mult net_monthly = monthly_income - projected_spend balance_trend = [] balance = current_balance for m in range(0, projection_months + 1): if m > 0: balance = max(0.0, balance + net_monthly) balance_trend.append({"month": f"M{m}", "balance": round(balance, 2)}) results[key] = { "name": config["name"], "monthly_income": round(monthly_income, 2), "monthly_spending": round(projected_spend, 2), "net_monthly": round(net_monthly, 2), "balance_projection": balance_trend, "final_balance": round(balance, 2), "savings_change_pct": round(((balance - current_balance) / current_balance * 100) if current_balance > 0 else 0.0, 2) } return results