AI in Finance

Fraud Detection System

import numpy as np
import pandas as pd
from sklearn.ensemble import IsolationForest, RandomForestClassifier
from typing import Dict, List

class FraudDetector:
    def __init__(self):
        self.isolation_forest = IsolationForest(contamination=0.01)
        self.classifier = RandomForestClassifier(n_estimators=100)
    
    def extract_features(self, transactions: pd.DataFrame) -> pd.DataFrame:
        features = pd.DataFrame()
        
        features["amount"] = transactions["amount"]
        features["hour"] = pd.to_datetime(transactions["timestamp"]).dt.hour
        features["day_of_week"] = pd.to_datetime(transactions["timestamp"]).dt.dayofweek
        
        features["amount_log"] = np.log1p(transactions["amount"])
        
        features["is_international"] = transactions["country"] != "US"
        
        return features
    
    def train_anomaly_detector(self, normal_transactions: pd.DataFrame):
        features = self.extract_features(normal_transactions)
        self.isolation_forest.fit(features)
    
    def train_classifier(self, labeled_data: pd.DataFrame, labels: np.ndarray):
        features = self.extract_features(labeled_data)
        self.classifier.fit(features, labels)
    
    def predict_fraud(self, transaction: Dict) -> Dict:
        features = pd.DataFrame([transaction])
        
        anomaly_score = self.isolation_forest.decision_function(features)[0]
        is_anomaly = self.isolation_forest.predict(features)[0] == -1
        
        fraud_prob = self.classifier.predict_proba(features)[0][1]
        
        return {
            "is_fraud": is_anomaly or fraud_prob > 0.5,
            "fraud_probability": fraud_prob,
            "anomaly_score": anomaly_score,
            "confidence": max(fraud_prob, 1 - fraud_prob)
        }

detector = FraudDetector()
detector.train_anomaly_detector(normal_transactions)
result = detector.predict_fraud(new_transaction)

Trading Signal Generator

import torch
import torch.nn as nn

class TradingSignalGenerator:
    def __init__(self, sequence_length=60, n_features=10):
        self.model = nn.LSTM(
            input_size=n_features,
            hidden_size=64,
            num_layers=2,
            batch_first=True,
            dropout=0.2
        )
        
        self.signal_head = nn.Sequential(
            nn.Linear(64, 32),
            nn.ReLU(),
            nn.Linear(32, 3),
            nn.Softmax(dim=1)
        )
    
    def generate_signal(self, market_data: torch.Tensor) -> Dict:
        self.model.eval()
        
        with torch.no_grad():
            lstm_out, _ = self.model(market_data.unsqueeze(0))
            signals = self.signal_head(lstm_out[:, -1, :])
        
        action = torch.argmax(signals, dim=1).item()
        confidence = signals.max().item()
        
        action_map = {0: "buy", 1: "hold", 2: "sell"}
        
        return {
            "action": action_map[action],
            "confidence": confidence,
            "probabilities": {
                "buy": signals[0, 0].item(),
                "hold": signals[0, 1].item(),
                "sell": signals[0, 2].item()
            }
        }

generator = TradingSignalGenerator()
signal = generator.generate_signal(market_data_tensor)

Risk Assessment

class CreditRiskModel:
    def __init__(self):
        self.model = RandomForestClassifier(n_estimators=100)
        self.feature_importance = None
    
    def prepare_features(self, applicant_data: pd.DataFrame) -> pd.DataFrame:
        features = pd.DataFrame()
        
        features["income"] = applicant_data["annual_income"]
        features["debt_ratio"] = applicant_data["total_debt"] / applicant_data["annual_income"]
        features["credit_score"] = applicant_data["credit_score"]
        features["employment_years"] = applicant_data["employment_length"]
        features["num_accounts"] = applicant_data["number_of_accounts"]
        features["recent_inquiries"] = applicant_data["recent_credit_inquiries"]
        
        return features
    
    def predict_risk(self, applicant_data: pd.DataFrame) -> Dict:
        features = self.prepare_features(applicant_data)
        
        risk_prob = self.model.predict_proba(features)[0][1]
        risk_score = int(risk_prob * 1000)
        
        if risk_score < 300:
            risk_level = "low"
            recommendation = "approve"
        elif risk_score < 600:
            risk_level = "medium"
            recommendation = "conditional_approve"
        else:
            risk_level = "high"
            recommendation = "deny"
        
        return {
            "risk_score": risk_score,
            "risk_level": risk_level,
            "recommendation": recommendation,
            "factors": self.get_risk_factors(features)
        }
    
    def get_risk_factors(self, features: pd.DataFrame) -> List[Dict]:
        factors = []
        
        if features["debt_ratio"].values[0] > 0.4:
            factors.append({"factor": "High Debt Ratio", "impact": "negative"})
        
        if features["credit_score"].values[0] < 650:
            factors.append({"factor": "Low Credit Score", "impact": "negative"})
        
        return factors

risk_model = CreditRiskModel()
risk_assessment = risk_model.predict_risk(applicant_data)

Sentiment Analysis for Trading

import openai

class FinancialSentimentAnalyzer:
    def __init__(self, api_key: str):
        self.client = openai.OpenAI(api_key=api_key)
    
    def analyze_news(self, headlines: List[str]) -> Dict:
        combined = "\n".join([f"- {h}" for h in headlines[:20]])
        
        response = self.client.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": """Analyze financial news sentiment.
Return JSON with overall_sentiment (bullish/bearish/neutral), 
confidence, and sentiment_scores for each headline."""},
                {"role": "user", "content": f"Headlines:\n{combined}"}
            ],
            temperature=0,
            response_format={"type": "json_object"}
        )
        
        import json
        return json.loads(response.choices[0].message.content)

analyzer = FinancialSentimentAnalyzer(api_key="your-api-key")
sentiment = analyzer.analyze_news([
    "Tech stocks surge on strong earnings",
    "Fed signals potential rate cut",
    "Oil prices drop amid demand concerns"
])

Best Practices

• Implement real-time monitoring for fraud
• Use ensemble methods for robust predictions
• Maintain regulatory compliance (SEC, FINRA)
• Ensure explainability for regulatory requirements
• Monitor model drift in production
• Use proper backtesting for trading strategies