import os import sys import uuid import random from datetime import datetime, timedelta # Add parent directory to path so we can import from app sys.path.append(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))) from app.database.database import SessionLocal, engine, SQLALCHEMY_DATABASE_URL from app.database.models import ( Base, User, Account, Transaction, Subscription, Goal, Investment, AIInsight, FraudLog, Notification, AnalyticsSnapshot ) # Create tables Base.metadata.create_all(bind=engine) def seed_data(): db = SessionLocal() print(f"Seeding into: {SQLALCHEMY_DATABASE_URL}") # Check if we already have users if db.query(User).count() > 0: print("Database already seeded.") db.close() return print("Seeding database...") personas = ["Saver", "Investor", "Impulsive Spender", "Minimalist", "Risk Taker"] merchants = ["Swiggy", "Amazon", "Netflix", "Uber", "Fuel", "Salary", "SIP", "Starbucks", "Apple", "Walmart"] categories = ["Food", "Shopping", "Entertainment", "Transport", "Income", "Investment", "Groceries", "Tech", "Utilities"] for persona in personas: try: user = User( email=f"{persona.lower().replace(' ', '_')}@example.com", password_hash="hashed_password", profile_data={"name": f"{persona} User", "phone": "+1234567890"}, financial_personality=persona, ai_personalization_settings={"theme": "dark", "notifications": "all"} ) db.add(user) db.flush() # get user.id without committing # ── Accounts ────────────────────────────────────────────────────── checking = Account( user_id=user.id, type="checking", balance=random.uniform(1000.0, 10000.0), currency="USD" ) savings = Account( user_id=user.id, type="savings", balance=random.uniform(5000.0, 50000.0), currency="USD" ) db.add_all([checking, savings]) db.flush() # ── Subscriptions ───────────────────────────────────────────────── # Active subscription (high usage) db.add(Subscription( user_id=user.id, merchant="Netflix", amount=15.99, billing_cycle="monthly", active=True, ai_usage_detection={"usage_frequency": "high", "recommendation": "keep"} )) # Unused subscription — triggers unused detection in subscriptions.py db.add(Subscription( user_id=user.id, merchant="Spotify", amount=9.99, billing_cycle="monthly", active=True, ai_usage_detection={"usage_frequency": "low", "recommendation": "cancel"} )) # Duplicate subscription — triggers duplicate detection in subscriptions.py # (second Netflix entry for the same user) db.add(Subscription( user_id=user.id, merchant="Netflix", amount=15.99, billing_cycle="monthly", active=True, ai_usage_detection={"usage_frequency": "medium", "recommendation": "review"} )) # ── Goals ───────────────────────────────────────────────────────── db.add(Goal( user_id=user.id, title="Emergency Fund", target_amount=10000.0, current_amount=random.uniform(1000.0, 5000.0), target_date=datetime.utcnow() + timedelta(days=365), ai_generated_plan={"monthly_saving_required": 500.0, "risk": "low"} )) # ── Investments ─────────────────────────────────────────────────── db.add(Investment( user_id=user.id, asset_name="S&P 500", type="stock", amount_invested=random.uniform(1000.0, 10000.0), current_value=random.uniform(1100.0, 12000.0), portfolio_allocation=50.0, ai_risk_analysis={"risk_level": "medium", "recommendation": "hold"} )) # ── Transactions ────────────────────────────────────────────────── start_date = datetime.utcnow() - timedelta(days=90) # Monthly salary (3 months) for i in range(3): tx_date = start_date + timedelta(days=i * 30) db.add(Transaction( account_id=checking.id, amount=5000.0, type="credit", category="Income", timestamp=tx_date, merchant="Salary", tags=["salary", "income"], ai_generated_metadata={"is_recurring": True, "confidence": 0.99}, spending_emotion_label="neutral" )) # Regular expense transactions for _ in range(30): tx_date = start_date + timedelta(days=random.randint(0, 89)) amount = random.uniform(10.0, 500.0) merchant = random.choice(merchants) if merchant == "Salary": continue # Persona-based spending adjustments if user.financial_personality == "Saver" and amount > 200: amount = random.uniform(10.0, 100.0) elif user.financial_personality == "Impulsive Spender": amount = random.uniform(50.0, 800.0) tx = Transaction( account_id=checking.id, amount=amount, type="debit", category=random.choice(categories), timestamp=tx_date, merchant=merchant, tags=["expense"], ai_generated_metadata={"category_confidence": 0.9}, spending_emotion_label=random.choice(["happy", "regret", "neutral", "essential"]) ) db.add(tx) db.flush() # Seed a fraud log for ~5% of transactions if random.random() < 0.05: db.add(FraudLog( transaction_id=tx.id, risk_score=random.uniform(0.7, 0.99), suspicious_activity_details="Unusual location and high amount for this merchant.", status="pending" )) # Late-night transaction — ensures behavior.py late-night detection fires late_night_date = start_date + timedelta(days=random.randint(1, 80), hours=23, minutes=random.randint(0, 59)) db.add(Transaction( account_id=checking.id, amount=random.uniform(50.0, 300.0), type="debit", category="Entertainment", timestamp=late_night_date, merchant="Online Store", tags=["late-night", "impulse"], ai_generated_metadata={"category_confidence": 0.85}, spending_emotion_label="regret" )) # ── Supporting records ──────────────────────────────────────────── db.add(AIInsight( user_id=user.id, type="cashflow", content=f"You are spending 20% more on {random.choice(categories)} this month." )) db.add(Notification( user_id=user.id, title="Weekly Summary", message="Your weekly financial summary is ready.", type="insight" )) db.add(AnalyticsSnapshot( user_id=user.id, date=datetime.utcnow(), total_balance=checking.balance + savings.balance, total_spending=2000.0, total_savings=3000.0, financial_score=random.uniform(60.0, 95.0), trends_json={"spending_trend": "down", "savings_trend": "up"} )) db.commit() print(f" ✓ Seeded user: {persona}") except Exception as e: db.rollback() print(f" ✗ Failed to seed user '{persona}': {e}") db.close() print("Database seeded successfully!") if __name__ == "__main__": seed_data()