AI 在金融科技中的应用

# 金融科技 AI 技术栈全景
from enum import Enum

class AITechnology(Enum):
    # 传统 ML - 仍然在结构化数据上表现优异
    LOGISTIC_REGRESSION = "信用评分卡"
    GRADIENT_BOOSTING = "XGBoost / LightGBM - 欺诈检测"
    RANDOM_FOREST = "客户分群"
    # 深度学习 - 处理复杂模式
    TRANSFORMER = "金融文本分析、舆情挖掘"
    GRAPH_NN = "关联交易网络分析"
    LSTM = "时间序列预测"
    # 前沿技术
    LLM = "智能客服、合规文档审查"
    REINFORCEMENT_LEARNING = "算法交易、投资组合优化"
    GAN = "合成金融数据生成、压力测试"

def select_model(data_type: str, task: str, data_size: int) -> str:
    if task == "fraud_detection":
        return "LightGBM" if data_size < 1_000_000 else "GraphSAGE"
    elif task == "credit_scoring":
        return "LogisticRegression" if data_size < 100_000 else "XGBoost"
    elif task == "algorithmic_trading":
        return "PPO" if data_type == "time_series" else "LSTM"
    else:
        return "BERT-Finance"

# 金融科技 AI 系统架构
class FinTechAISystem:
    """典型的金融科技 AI 系统架构"""
    
    def __init__(self):
        self.data_layer = {
            "structured": ["交易记录", "账户信息", "征信数据"],
            "unstructured": ["新闻", "财报", "社交媒体", "合同文本"],
            "alternative": ["卫星图像", "供应链数据", "支付流水"],
        }
        self.feature_store = {}
        self.model_registry = {}
        self.monitoring = {}
        
    def pipeline(self, raw_data):
        """端到端 AI 处理流水线"""
        features = self.feature_engineering(raw_data)
        predictions = self.model_inference(features)
        decisions = self.business_rules(predictions)
        self.log_decision(decisions)
        return decisions
        
    def model_inference(self, features):
        """多模型集成推理"""
        model_a = self.model_registry["fraud_model_v3"]
        model_b = self.model_registry["credit_model_v7"]
        return {
            "fraud_score": model_a.predict(features),
            "credit_score": model_b.predict(features),
        }

# 使用 XGBoost 构建信用风险评估模型
import xgboost as xgb
from sklearn.metrics import roc_auc_score
import numpy as np

def build_credit_model(X_train, y_train, X_val, y_val):
    """训练信用风险评估模型"""
    # 处理类别不平衡 - 金融数据中违约样本通常不到 5%
    n_positive = y_train.sum()
    n_negative = len(y_train) - n_positive
    scale_pos_weight = n_negative / max(n_positive, 1)
    
    params = {
        "objective": "binary:logistic",
        "eval_metric": "auc",
        "max_depth": 6,
        "learning_rate": 0.05,
        "subsample": 0.8,
        "colsample_bytree": 0.7,
        "scale_pos_weight": scale_pos_weight,
        "min_child_weight": 50,
    }
    
    dtrain = xgb.DMatrix(X_train, label=y_train)
    dval = xgb.DMatrix(X_val, label=y_val)
    
    model = xgb.train(
        params, dtrain, num_boost_round=1000,
        evals=[(dtrain, "train"), (dval, "val")],
        early_stopping_rounds=50,
        verbose_eval=100,
    )
    return model

# 模型解释性 - SHAP 值分析
import shap
explainer = shap.TreeExplainer(model)
shap_values = explainer.shap_values(X_val)

# 使用图神经网络进行关联信用评估
import torch
import torch.nn as nn
import torch.nn.functional as F
from torch_geometric.nn import SAGEConv

class CreditGNN(nn.Module):
    """基于图神经网络的信用评估模型
    
    核心思想：借款人的信用风险不仅取决于自身特征，
    还与其关联人的信用状况相关（社交传染效应）
    """
    
    def __init__(self, node_features: int, hidden: int = 128):
        super().__init__()
        self.conv1 = SAGEConv(node_features, hidden)
        self.conv2 = SAGEConv(hidden, hidden)
        self.classifier = nn.Sequential(
            nn.Linear(hidden, 64),
            nn.ReLU(),
            nn.Dropout(0.3),
            nn.Linear(64, 1),
        )
        
    def forward(self, x, edge_index):
        """
        x: 节点特征矩阵 [N, node_features]
        edge_index: 边索引 [2, E] - 关联关系
        """
        h = F.relu(self.conv1(x, edge_index))
        h = F.dropout(h, p=0.3, training=self.training)
        h = F.relu(self.conv2(h, edge_index))
        return torch.sigmoid(self.classifier(h)).squeeze(-1)

# 实时欺诈检测管道
import numpy as np
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler

class RealTimeFraudDetector:
    """实时欺诈检测器"""
    
    def __init__(self):
        self.scaler = StandardScaler()
        self.iso_forest = IsolationForest(
            contamination=0.001,  # 预计欺诈率 0.1%
            n_estimators=200,
            max_samples="auto",
            random_state=42,
        )
        self.threshold = -0.5
        
    def fit(self, historical_transactions):
        """用历史正常交易训练异常检测模型"""
        features = self._extract_features(historical_transactions)
        scaled = self.scaler.fit_transform(features)
        self.iso_forest.fit(scaled)
        scores = self.iso_forest.decision_function(scaled)
        self.threshold = np.percentile(scores, 1)
        
    def detect(self, transaction):
        """实时检测单笔交易"""
        features = self._extract_features([transaction])
        scaled = self.scaler.transform(features)
        score = self.iso_forest.decision_function(scaled)[0]
        is_fraud = score < self.threshold
        return {"is_fraud": bool(is_fraud), "anomaly_score": float(score)}
        
    def _extract_features(self, transactions):
        """从交易记录中提取特征"""
        pass

# 使用图注意力网络检测欺诈团伙
import torch
from torch_geometric.data import Data
from torch_geometric.nn import GATConv

class FraudGAT(torch.nn.Module):
    """基于图注意力网络的欺诈团伙检测
    
    GAT 的优势：为不同邻居分配不同注意力权重
    例如：可疑账户之间的关联比正常账户更值得关注
    """
    
    def __init__(self, in_channels, hidden_channels, num_heads=4):
        super().__init__()
        self.conv1 = GATConv(in_channels, hidden_channels, 
                            heads=num_heads, dropout=0.3)
        self.conv2 = GATConv(hidden_channels * num_heads, 
                            hidden_channels, heads=1, 
                            concat=False, dropout=0.3)
        self.classifier = torch.nn.Linear(hidden_channels, 1)
        
    def forward(self, data):
        x, edge_index = data.x, data.edge_index
        x = torch.relu(self.conv1(x, edge_index))
        x = torch.dropout(x, p=0.3, train=self.training)
        x = self.conv2(x, edge_index)
        return torch.sigmoid(self.classifier(x))

# 欺诈图谱构建
# 节点 = 账户，边 = 转账/共同设备/共同IP/共同地址
def build_fraud_graph(transactions, accounts):
    """构建欺诈检测图"""
    nodes = torch.tensor(accounts.features)
    edges = build_edges(transactions)
    return Data(x=nodes, edge_index=edges)

# 使用强化学习进行算法交易
import gym
import numpy as np
from stable_baselines3 import PPO
from stable_baselines3.common.vec_env import DummyVecEnv

class StockTradingEnv(gym.Env):
    """股票交易环境 - 强化学习的交互接口"""
    
    def __init__(self, prices, features, initial_balance=100000):
        super().__init__()
        self.prices = prices
        self.features = features
        self.initial_balance = initial_balance
        self.action_space = gym.spaces.Discrete(3)  # 0=卖出, 1=持有, 2=买入
        self.observation_space = gym.spaces.Box(
            low=-np.inf, high=np.inf,
            shape=(features.shape[1] + 3,), dtype=np.float32
        )
        self.reset()
        
    def reset(self):
        self.step_num = 0
        self.balance = self.initial_balance
        self.shares = 0
        return self._get_observation()
        
    def step(self, action):
        self._execute_action(action)
        self.step_num += 1
        reward = self._calculate_reward()
        done = self.step_num >= len(self.prices) - 1
        return self._get_observation(), reward, done, {}
        
    def _execute_action(self, action):
        current_price = self.prices[self.step_num]
        if action == 2 and self.balance > current_price:
            shares_to_buy = int(self.balance * 0.95 / current_price)
            self.shares += shares_to_buy
            self.balance -= shares_to_buy * current_price
        elif action == 0 and self.shares > 0:
            self.balance += self.shares * current_price * 0.999
            
    def _calculate_reward(self):
        current_value = self.balance + self.shares * self.prices[self.step_num]
        return (current_value - self.portfolio_value) / self.portfolio_value

# 训练 PPO 交易策略
env = DummyVecEnv([lambda: StockTradingEnv(prices, features)])
model = PPO("MlpPolicy", env, verbose=1, n_steps=2048)
model.learn(total_timesteps=500_000)

# 多因子量化策略 + 机器学习
import pandas as pd
import numpy as np
from sklearn.ensemble import GradientBoostingRegressor

class MLQuantStrategy:
    """机器学习驱动的量化交易策略"""
    
    FACTORS = [
        "momentum_20d", "momentum_60d", "volatility_20d",
        "volume_ratio", "rsi_14", "macd_signal",
        "pe_ratio", "market_cap_rank",
    ]
    
    def __init__(self):
        self.model = GradientBoostingRegressor(
            n_estimators=200, max_depth=4, learning_rate=0.05,
            subsample=0.8, min_samples_leaf=50,
        )
        
    def train(self, historical_data):
        """用历史数据训练收益预测模型"""
        X = historical_data[self.FACTORS].values
        y = historical_data["close"].shift(-5) / historical_data["close"] - 1
        y = y.dropna()
        X = X[:len(y)]
        self.model.fit(X, y)
        
    def predict_signals(self, current_data):
        """生成交易信号"""
        X = current_data[self.FACTORS].values
        expected_returns = self.model.predict(X)
        top_quintile = np.percentile(expected_returns, 80)
        bottom_quintile = np.percentile(expected_returns, 20)
        
        signals = pd.DataFrame({
            "ticker": current_data["ticker"],
            "expected_return": expected_returns,
            "signal": np.where(expected_returns >= top_quintile, 1,
                      np.where(expected_returns <= bottom_quintile, -1, 0)),
        })
        return signals

# 智能投顾核心：客户风险画像
import numpy as np
from sklearn.cluster import KMeans
from dataclasses import dataclass

@dataclass
class InvestorProfile:
    """投资者画像"""
    risk_tolerance: float      # 风险承受力 0-10
    investment_horizon: int    # 投资期限（年）
    expected_return: float     # 期望年化收益率
    max_drawdown: float        # 最大可接受回撤
    income_level: float
    age: int
    dependents: int

def profile_investor(questionnaire_responses, transaction_history):
    """基于问卷和交易行为自动构建投资者画像"""
    # 问卷得分
    survey_risk = np.mean(questionnaire_responses["risk_questions"])
    
    # 行为分析 - 从历史交易中推断真实风险偏好
    # 人们说的和做的往往不一致
    behavior_signals = analyze_trading_behavior(transaction_history)
    actual_risk = behavior_signals["avg_portfolio_volatility"]
    panic_selling = behavior_signals["panic_sell_frequency"]
    
    # 综合评分 - 行为数据权重更高
    composite_risk = 0.3 * survey_risk + 0.7 * actual_risk
    if panic_selling > 0.3:
        composite_risk *= 0.8
        
    return InvestorProfile(
        risk_tolerance=composite_risk,
        investment_horizon=questionnaire_responses["horizon"],
        expected_return=questionnaire_responses["target_return"],
        max_drawdown=estimate_max_drawdown(composite_risk),
        income_level=transaction_history["avg_monthly_income"],
        age=questionnaire_responses["age"],
        dependents=questionnaire_responses["dependents"],
    )

def estimate_max_drawdown(risk_score):
    """根据风险评分估算最大可接受回撤"""
    return 0.05 + 0.035 * risk_score

# 动态资产配置引擎
import numpy as np
from scipy.optimize import minimize

class RoboAdvisorEngine:
    """智能投顾资产配置引擎"""
    
    ASSET_CLASSES = [
        "US_EQUITY", "INTL_EQUITY", "BONDS",
        "REAL_ESTATE", "COMMODITIES", "CASH",
    ]
    
    def optimize_portfolio(self, investor_profile, expected_returns, cov_matrix):
        """为投资者优化资产配置"""
        n_assets = len(self.ASSET_CLASSES)
        max_equity = investor_profile.risk_tolerance / 10 * 0.8
        
        def negative_sharpe(weights):
            """负夏普比率 - 最小化负值即最大化夏普"""
            port_return = np.sum(weights * expected_returns)
            port_vol = np.sqrt(weights.T @ cov_matrix @ weights)
            return -(port_return - 0.03) / port_vol
            
        constraints = [
            {"type": "eq", "fun": lambda w: np.sum(w) - 1.0},
            {"type": "ineq", "fun": lambda w: max_equity - w[0] - w[1]},
        ]
        bounds = tuple((0.0, 0.5) for _ in range(n_assets))
        
        result = minimize(negative_sharpe, np.ones(n_assets)/n_assets,
                         method="SLSQP", bounds=bounds, constraints=constraints)
        
        return dict(zip(self.ASSET_CLASSES, np.round(result.x, 4)))
        
    def rebalance_check(self, current_weights, target_weights, threshold=0.05):
        """检查是否需要再平衡"""
        deviations = np.abs(current_weights - target_weights)
        needs_rebalance = deviations > threshold
        return needs_rebalance, deviations

# AML 交易监控 - 智能警报过滤
import pandas as pd
from sklearn.ensemble import RandomForestClassifier
from imblearn.pipeline import Pipeline
from imblearn.over_sampling import SMOTE

class AMLAlertFilter:
    """AML 警报智能过滤器 - 降低误报率"""
    
    def __init__(self):
        self.pipeline = Pipeline([
            ("smote", SMOTE(sampling_strategy=0.1)),
            ("classifier", RandomForestClassifier(
                n_estimators=300, max_depth=10,
                class_weight="balanced", n_jobs=-1,
            ))
        ])
        
    def train(self, historical_alerts, investigation_outcomes):
        """用历史调查结果训练过滤器"""
        self.pipeline.fit(historical_alerts, investigation_outcomes)
        
    def filter_alerts(self, new_alerts):
        """过滤新警报，返回高风险警报列表"""
        probabilities = self.pipeline.predict_proba(new_alerts)[:, 1]
        new_alerts["fraud_probability"] = probabilities
        high_risk = new_alerts[new_alerts["fraud_probability"] > 0.3]
        high_risk = high_risk.sort_values("fraud_probability", ascending=False)
        return high_risk
        
    def generate_report(self, alert):
        """自动生成可疑活动报告草稿"""
        report = f"账户 {alert['account_id']} 触发 {alert['rule_name']} 规则。"
        report += "交易金额 $" + format(alert['amount'], ',.2f') + "，频率 " + str(alert['tx_count']) + " 笔/天。"
        report += f"风险评分 {alert['fraud_probability']:.2%}。"
        report += f"关联账户数 {alert['linked_accounts']}。"
        action = "立即冻结" if alert["fraud_probability"] > 0.7 else "进一步调查"
        report += f"建议行动：{action}。"
        return report

# 监管法规 NLP 解析引擎
import spacy
from typing import List, Dict

class RegulationAnalyzer:
    """监管法规自动解析引擎
    
    功能：
    1. 解析监管文件，提取合规要求
    2. 与现有合规规则对比，识别新增/变更要求
    3. 生成合规差距报告
    """
    
    def __init__(self):
        self.nlp = spacy.load("zh_core_web_trf")
        
    def parse_regulation(self, regulation_text: str) -> Dict:
        """解析监管文件，提取关键合规要求"""
        doc = self.nlp(regulation_text)
        
        # 提取义务性语句（包含"应当"、"必须"、"不得"）
        obligations = []
        for sent in doc.sents:
            text = sent.text
            if any(kw in text for kw in ["应当", "必须", "不得", "禁止"]):
                obligations.append({
                    "text": text,
                    "keywords": [k for k in ["应当", "必须", "不得", "禁止"] if k in text],
                })
                
        return {
            "total_sentences": len(list(doc.sents)),
            "obligations": obligations,
            "compliance_items": len(obligations),
        }
        
    def compare_regulations(self, old_reg: str, new_reg: str) -> List[Dict]:
        """对比新旧法规，识别变化"""
        old_items = self.parse_regulation(old_reg)["obligations"]
        new_items = self.parse_regulation(new_reg)["obligations"]
        changes = []
        old_texts = {item["text"] for item in old_items}
        for item in new_items:
            if item["text"] not in old_texts:
                changes.append({"type": "new", "content": item["text"]})
        return changes

# 端到端欺诈检测系统
import pandas as pd
import numpy as np
import lightgbm as lgb
from sklearn.model_selection import train_test_split
from sklearn.metrics import roc_auc_score
import json

class FraudDetectionSystem:
    """完整的欺诈检测系统"""
    
    def __init__(self, config_path="config.json"):
        with open(config_path) as f:
            self.config = json.load(f)
        self.feature_engineer = FeatureEngineer()
        self.models = {}
        self.threshold = 0.5
        
    def train(self, transactions_df, labels_df):
        """训练欺诈检测模型"""
        print("1. 特征工程...")
        features = self.feature_engineer.transform(transactions_df)
        
        print("2. 数据集划分...")
        X_train, X_val, y_train, y_val = train_test_split(
            features, labels_df, test_size=0.2, 
            stratify=labels_df, random_state=42
        )
        
        print("3. 训练 LightGBM 主模型...")
        fraud_ratio = y_train.sum() / len(y_train)
        train_data = lgb.Dataset(X_train, label=y_train)
        val_data = lgb.Dataset(X_val, label=y_val, reference=train_data)
        
        params = {
            "objective": "binary", "metric": "auc",
            "boosting_type": "gbdt", "num_leaves": 63,
            "learning_rate": 0.05, "feature_fraction": 0.8,
            "bagging_fraction": 0.8, "bagging_freq": 5,
            "scale_pos_weight": (1 - fraud_ratio) / fraud_ratio,
            "verbose": -1,
        }
        
        self.models["lightgbm"] = lgb.train(
            params, train_data, num_boost_round=1000,
            valid_sets=[val_data],
            callbacks=[lgb.early_stopping(50), lgb.log_evaluation(100)],
        )
        
        val_preds = self.models["lightgbm"].predict(X_val)
        auc = roc_auc_score(y_val, val_preds)
        print(f"验证集 AUC: {auc:.4f}")
        self.threshold = self._optimize_threshold(y_val, val_preds)
        
    def predict(self, transaction_df):
        """实时推理 - 交易到来时的处理"""
        features = self.feature_engineer.transform(transaction_df)
        score = self.models["lightgbm"].predict(features)[0]
        if score >= self.threshold:
            return "BLOCK"
        elif score >= self.threshold * 0.7:
            return "REVIEW"
        else:
            return "ALLOW"
            
    def _optimize_threshold(self, y_true, y_scores):
        """基于业务成本优化决策阈值"""
        FN_COST, FP_COST = 100, 5
        best_threshold, best_cost = 0.5, float("inf")
        for t in np.arange(0.1, 0.9, 0.01):
            preds = (y_scores >= t).astype(int)
            fn = np.sum((y_true == 1) & (preds == 0))
            fp = np.sum((y_true == 0) & (preds == 1))
            cost = fn * FN_COST + fp * FP_COST
            if cost < best_cost:
                best_cost = cost
                best_threshold = t
        return best_threshold

class FeatureEngineer:
    def transform(self, df):
        df = df.copy()
        df["hour"] = pd.to_datetime(df["TransactionDT"], unit="s").dt.hour
        df["is_night"] = (df["hour"] < 6).astype(int)
        df["log_amount"] = np.log1p(df["TransactionAmt"])
        df["amount_deviation"] = df["TransactionAmt"] / df["TransactionAmt"].median()
        df["device_type_encoded"] = df["DeviceType"].map({"desktop": 0, "mobile": 1}).fillna(2)
        return df.fillna(-999)

# 模型监控与持续学习
import numpy as np
from datetime import datetime

class ModelMonitor:
    """欺诈检测模型监控器"""
    
    def __init__(self, baseline_features=None, baseline_scores=None):
        self.baseline_features = baseline_features
        self.baseline_scores = baseline_scores
        self.prediction_log = []
        self.feedback_log = []
        
    def log_prediction(self, transaction_id, features, score, decision):
        """记录每次预测"""
        self.prediction_log.append({
            "timestamp": datetime.now(),
            "transaction_id": transaction_id,
            "score": score,
            "decision": decision,
        })
        
    def log_feedback(self, transaction_id, actual_outcome):
        """记录真实结果（来自客户投诉或人工审核）"""
        self.feedback_log.append({
            "timestamp": datetime.now(),
            "transaction_id": transaction_id,
            "actual_outcome": actual_outcome,
        })
        
    def check_drift(self, recent_features, threshold=0.05):
        """检测特征漂移 - PSI (Population Stability Index)"""
        if self.baseline_features is None:
            return {"drift_detected": False, "reason": "无基线数据"}
        psi_values = {}
        drift_detected = False
        for feature in recent_features.columns:
            psi = self._calculate_psi(
                self.baseline_features[feature], recent_features[feature]
            )
            psi_values[feature] = psi
            if psi > 0.25:
                drift_detected = True
        return {
            "drift_detected": drift_detected,
            "psi_values": psi_values,
            "high_drift_features": [f for f, psi in psi_values.items() if psi > 0.25],
        }
        
    def _calculate_psi(self, expected, actual, buckets=10):
        """计算群体稳定性指数"""
        breakpoints = np.linspace(0, 100, buckets + 1)
        expected_percents = np.percentile(expected, breakpoints)
        psi = 0
        for i in range(len(expected_percents) - 1):
            lo, hi = expected_percents[i], expected_percents[i + 1]
            ep = np.sum((expected >= lo) & (expected < hi)) / len(expected)
            ap = np.sum((actual >= lo) & (actual < hi)) / len(actual)
            ep = max(ep, 0.0001)
            ap = max(ap, 0.0001)
            psi += (ap - ep) * np.log(ap / ep)
        return psi

def check_retrain_needed(monitor, psi_threshold=0.25):
    """检查是否需要重新训练模型"""
    recent = get_recent_transactions(hours=24)
    drift_result = monitor.check_drift(recent)
    if drift_result["drift_detected"]:
        print(f"检测到特征漂移！高漂移特征: {drift_result['high_drift_features']}")
        trigger_retraining_pipeline()
        return True
    return False