Spaces:
Running
Running
| from datetime import datetime, timedelta | |
| import numpy as np | |
| from sqlalchemy.orm import Session | |
| from app.database.models import Transaction, FraudLog, Account, User | |
| def evaluate_transaction_for_fraud(db: Session, transaction_id: str): | |
| """ | |
| Evaluates a transaction for anomalies, generates a score, and logs alerts. | |
| """ | |
| txn = db.query(Transaction).filter(Transaction.id == transaction_id).first() | |
| if not txn: | |
| return {"error": "Transaction not found"} | |
| account = db.query(Account).filter(Account.id == txn.account_id).first() | |
| if not account: | |
| return {"error": "Account not found for transaction"} | |
| user_id = account.user_id | |
| # Fetch historical transactions to compare | |
| history = db.query(Transaction).join(Account).filter( | |
| Account.user_id == user_id, | |
| Transaction.type == "debit", | |
| Transaction.id != transaction_id | |
| ).order_by(Transaction.timestamp.desc()).limit(30).all() | |
| score = 0 | |
| reasons = [] | |
| # 1. Spikes in amount | |
| if history: | |
| amounts = [h.amount for h in history] | |
| avg_amount = np.mean(amounts) | |
| std_amount = np.std(amounts) if len(amounts) > 1 else 0.0 | |
| if txn.amount > avg_amount * 3.5: | |
| score += 40 | |
| reasons.append(f"Transaction amount (${txn.amount:,.2f}) is abnormally high compared to your historical average of ${avg_amount:,.2f}.") | |
| elif txn.amount > avg_amount * 2.0: | |
| score += 20 | |
| reasons.append(f"Transaction amount is significantly higher than usual (2x historical average).") | |
| else: | |
| avg_amount = 0.0 | |
| # 2. Timing anomaly (Late night 11 PM - 4 AM) | |
| hour = txn.timestamp.hour | |
| if hour >= 23 or hour < 4: | |
| score += 25 | |
| reasons.append("Unusual timing (transaction placed between 11 PM and 4 AM).") | |
| # 3. Frequency anomaly (rapid consecutive transactions) | |
| if history: | |
| latest_txn = history[0] | |
| time_diff = abs((txn.timestamp - latest_txn.timestamp).total_seconds()) | |
| if time_diff < 180: # Less than 3 minutes | |
| score += 20 | |
| reasons.append("High-frequency activity: multiple transactions placed within 3 minutes.") | |
| # 4. Duplicate transaction check (same merchant and amount within 10 minutes) | |
| if history: | |
| for prev in history[:5]: | |
| time_diff = abs((txn.timestamp - prev.timestamp).total_seconds()) | |
| if prev.merchant == txn.merchant and prev.amount == txn.amount and time_diff < 600: | |
| score += 30 | |
| reasons.append(f"Potential duplicate payment: identical debit of ${txn.amount:.2f} at {txn.merchant} detected within 10 minutes.") | |
| break | |
| # Normalize score to 100 max | |
| score = min(100, score) | |
| # Log to DB if score exceeds threshold | |
| if score >= 30: | |
| # Check if fraud log already exists | |
| existing_log = db.query(FraudLog).filter(FraudLog.transaction_id == txn.id).first() | |
| if not existing_log: | |
| fraud_log = FraudLog( | |
| transaction_id=txn.id, | |
| risk_score=score / 100.0, | |
| suspicious_activity_details="; ".join(reasons), | |
| status="pending" | |
| ) | |
| db.add(fraud_log) | |
| db.commit() | |
| return { | |
| "transaction_id": txn.id, | |
| "amount": txn.amount, | |
| "merchant": txn.merchant, | |
| "timestamp": txn.timestamp.isoformat(), | |
| "fraud_risk_score": score, | |
| "is_anomalous": score >= 30, | |
| "explanations": reasons, | |
| "status": "flagged" if score >= 50 else "suspicious" if score >= 30 else "verified" | |
| } | |
| def get_user_fraud_alerts(db: Session, user_id: str): | |
| """ | |
| Retrieves all flagged/suspicious transaction records and logs. | |
| """ | |
| logs = db.query(FraudLog).join(Transaction).join(Account).filter( | |
| Account.user_id == user_id | |
| ).order_by(Transaction.timestamp.desc()).all() | |
| alerts = [] | |
| for log in logs: | |
| txn = log.transaction | |
| alerts.append({ | |
| "fraud_log_id": log.id, | |
| "transaction_id": txn.id, | |
| "amount": txn.amount, | |
| "merchant": txn.merchant, | |
| "category": txn.category, | |
| "timestamp": txn.timestamp.isoformat(), | |
| "risk_score": round(log.risk_score * 100, 0), | |
| "details": log.suspicious_activity_details, | |
| "status": log.status | |
| }) | |
| return { | |
| "total_alerts": len(alerts), | |
| "pending_reviews": sum(1 for a in alerts if a["status"] == "pending"), | |
| "alerts": alerts | |
| } | |