金融数据分析师面试题库：风控建模、合规监管与量化策略实战题

## 信用评分模型目标
### 业务目标
- 提升风险识别准确性
- 降低坏账率和损失
- 提高审批效率
- 支持差异化定价

### 模型要求
- 准确性：区分度和预测能力
- 稳定性：时间稳定性和人群稳定性
- 可解释性：监管要求和业务理解
- 合规性：公平性和无歧视

# 信用评分模型数据准备
import pandas as pd
import numpy as np
from sklearn.preprocessing import WOE
import warnings
warnings.filterwarnings('ignore')

class CreditScoringPreprocessor:
    def __init__(self):
        self.woe_encoders = {}
        self.feature_bins = {}

    def preprocess_data(self, data):
        """数据预处理主流程"""
        # 1. 数据清洗
        cleaned_data = self.data_cleaning(data)

        # 2. 特征工程
        featured_data = self.feature_engineering(cleaned_data)

        # 3. 变量筛选
        selected_data = self.variable_selection(featured_data)

        return selected_data

    def data_cleaning(self, data):
        """数据清洗"""
        cleaned = data.copy()

        # 处理缺失值
        for col in cleaned.columns:
            if cleaned[col].dtype == 'object':
                cleaned[col].fillna('Unknown', inplace=True)
            else:
                cleaned[col].fillna(cleaned[col].median(), inplace=True)

        # 异常值处理
        numeric_cols = cleaned.select_dtypes(include=[np.number]).columns
        for col in numeric_cols:
            if col != 'target':
                Q1 = cleaned[col].quantile(0.25)
                Q3 = cleaned[col].quantile(0.75)
                IQR = Q3 - Q1
                lower_bound = Q1 - 1.5 * IQR
                upper_bound = Q3 + 1.5 * IQR

                cleaned[col] = np.where(
                    cleaned[col] < lower_bound, lower_bound,
                    np.where(cleaned[col] > upper_bound, upper_bound, cleaned[col])
                )

        return cleaned

    def feature_engineering(self, data):
        """特征工程"""
        featured = data.copy()

        # 1. 基础统计特征
        if 'monthly_income' in featured.columns and 'loan_amount' in featured.columns:
            featured['debt_to_income_ratio'] = featured['loan_amount'] / featured['monthly_income']

        if 'credit_limit' in featured.columns and 'credit_used' in featured.columns:
            featured['credit_utilization'] = featured['credit_used'] / featured['credit_limit']

        # 2. 年龄分组
        if 'age' in featured.columns:
            featured['age_group'] = pd.cut(
                featured['age'],
                bins=[0, 25, 35, 45, 55, 100],
                labels=['young', 'adult', 'middle', 'senior', 'elder']
            )

        # 3. 收入稳定性
        if 'employment_years' in featured.columns:
            featured['employment_stability'] = np.where(
                featured['employment_years'] >= 2, 'stable', 'unstable'
            )

        return featured

    def woe_binning(self, data, feature, target, max_bins=5):
        """WOE分箱"""
        # 等频分箱
        data_copy = datafeature, target.copy()
        data_copy['bin'] = pd.qcut(data_copy[feature], q=max_bins, duplicates='drop')

        # 计算WOE
        bin_stats = data_copy.groupby('bin').agg({
            target: ['count', 'sum']
        }).reset_index()

        bin_stats.columns = ['bin', 'total', 'bad']
        bin_stats['good'] = bin_stats['total'] - bin_stats['bad']

        total_good = bin_stats['good'].sum()
        total_bad = bin_stats['bad'].sum()

        bin_stats['good_rate'] = bin_stats['good'] / total_good
        bin_stats['bad_rate'] = bin_stats['bad'] / total_bad

        bin_stats['woe'] = np.log(
            (bin_stats['bad_rate'] + 0.0001) / (bin_stats['good_rate'] + 0.0001)
        )

        # 计算IV值
        bin_stats['iv'] = (bin_stats['bad_rate'] - bin_stats['good_rate']) * bin_stats['woe']
        iv_value = bin_stats['iv'].sum()

        return bin_stats, iv_value

    def calculate_scorecard(self, model, features, base_score=600, base_odds=50, pdo=20):
        """评分卡刻度"""
        # 计算分数转换参数
        factor = pdo / np.log(2)
        offset = base_score - factor * np.log(base_odds)

        # 获取模型系数
        coefficients = model.coef_[0]
        intercept = model.intercept_[0]

        # 计算各特征分数
        feature_scores = {}
        for i, feature in enumerate(features):
            feature_scores[feature] = {
                'coefficient': coefficients[i],
                'score_factor': -factor * coefficients[i]
            }

        # 基础分数
        base_points = offset - factor * intercept

        return {
            'base_points': base_points,
            'feature_scores': feature_scores,
            'factor': factor
        }

# 模型训练示例
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import train_test_split
from sklearn.metrics import roc_auc_score, roc_curve

def train_credit_model(X, y):
    """训练信用评分模型"""
    # 数据分割
    X_train, X_test, y_train, y_test = train_test_split(
        X, y, test_size=0.3, random_state=42, stratify=y
    )

    # 模型训练
    model = LogisticRegression(penalty='l1', solver='liblinear', random_state=42)
    model.fit(X_train, y_train)

    # 模型评估
    train_pred = model.predict_proba(X_train)[:, 1]
    test_pred = model.predict_proba(X_test)[:, 1]

    train_auc = roc_auc_score(y_train, train_pred)
    test_auc = roc_auc_score(y_test, test_pred)

    print(f"训练集AUC: {train_auc:.4f}")
    print(f"测试集AUC: {test_auc:.4f}")

    return model, X_test, y_test, test_pred

# 模型验证框架
import matplotlib.pyplot as plt
from sklearn.metrics import classification_report, confusion_matrix

class ModelValidation:
    def __init__(self, model, X_test, y_test, y_pred_proba):
        self.model = model
        self.X_test = X_test
        self.y_test = y_test
        self.y_pred_proba = y_pred_proba

    def discrimination_analysis(self):
        """区分度分析"""
        # AUC值
        auc_score = roc_auc_score(self.y_test, self.y_pred_proba)

        # KS值计算
        fpr, tpr, thresholds = roc_curve(self.y_test, self.y_pred_proba)
        ks_score = max(tpr - fpr)

        # Gini系数
        gini_score = 2 * auc_score - 1

        return {
            'AUC': auc_score,
            'KS': ks_score,
            'Gini': gini_score
        }

    def stability_analysis(self, time_column=None):
        """稳定性分析"""
        if time_column is None:
            # 简化版本：按数据顺序分组
            data_size = len(self.X_test)
            periods = 4
            period_size = data_size // periods

            psi_results = []
            base_distribution = None

            for i in range(periods):
                start_idx = i * period_size
                end_idx = (i + 1) * period_size if i < periods - 1 else data_size

                period_scores = self.y_pred_proba[start_idx:end_idx]

                # 分箱
                score_bins = pd.cut(period_scores, bins=10, labels=False)
                current_dist = pd.Series(score_bins).value_counts(normalize=True).sort_index()

                if base_distribution is None:
                    base_distribution = current_dist
                else:
                    # 计算PSI
                    psi = self.calculate_psi(base_distribution, current_dist)
                    psi_results.append(psi)

            return {'PSI_values': psi_results, 'average_PSI': np.mean(psi_results)}

    def calculate_psi(self, expected, actual):
        """计算PSI值"""
        expected = expected + 0.0001  # 避免除零
        actual = actual + 0.0001

        psi_value = ((actual - expected) * np.log(actual / expected)).sum()
        return psi_value

    def generate_scorecard_report(self, scorecard_mapping):
        """生成评分卡报告"""
        # 计算各分数段的表现
        scores = self.convert_to_scores(scorecard_mapping)

        # 分数分组分析
        score_bins = pd.cut(scores, bins=10, labels=False)

        report_data = []
        for bin_id in range(10):
            mask = score_bins == bin_id
            if mask.sum() > 0:
                bin_bad_rate = self.y_test[mask].mean()
                bin_count = mask.sum()
                bin_min_score = scores[mask].min()
                bin_max_score = scores[mask].max()

                report_data.append({
                    'score_range': f"{bin_min_score:.0f}-{bin_max_score:.0f}",
                    'count': bin_count,
                    'bad_rate': bin_bad_rate,
                    'cum_bad_rate': None  # 需要计算累积坏账率
                })

        return pd.DataFrame(report_data)

    def convert_to_scores(self, scorecard_mapping):
        """转换为评分"""
        # 简化版本：使用概率转分数
        base_score = 600
        pdo = 20

        odds = (1 - self.y_pred_proba) / (self.y_pred_proba + 1e-10)
        scores = base_score + pdo * np.log(odds) / np.log(2)

        return scores

## 模型部署策略
### 部署架构
- 实时评分：API服务，毫秒级响应
- 批量评分：离线计算，定时更新
- 模型版本管理：A/B测试验证

### 监控体系
#### 模型性能监控
- AUC值：月度跟踪，设定预警阈值
- KS值：季度评估，稳定性检查
- PSI值：模型稳定性监控

#### 业务指标监控
- 通过率：申请通过率变化
- 坏账率：实际vs预期坏账率
- 盈利性：风险调整收益

### 模型治理
- 模型文档：详细记录建模过程
- 验证报告：独立验证团队评估
- 监管报告：定期向监管部门报告

## 金融欺诈类型分类
### 申请欺诈
- 身份欺诈：虚假身份信息
- 收入欺诈：夸大收入和资产
- 团伙欺诈：有组织批量申请

### 交易欺诈
- 账户盗用：他人账户非授权使用
- 洗钱行为：可疑资金流动
- 套现行为：信用额度违规套现

### 第三方欺诈
- 中介欺诈：非法中介包装
- 合作商欺诈：合作伙伴作假
- 内部欺诈：内部人员参与

# 反欺诈特征工程
class AntiFraudFeatureEngine:
    def __init__(self):
        self.device_profiles = {}
        self.network_features = {}

    def extract_device_features(self, data):
        """设备指纹特征"""
        features = {}

        # 设备基础信息
        features['device_type'] = data.get('device_type')
        features['os_version'] = data.get('os_version')
        features['screen_resolution'] = data.get('screen_resolution')

        # 设备行为特征
        features['typing_pattern'] = self.analyze_typing_pattern(data.get('keystroke_data', []))
        features['mouse_pattern'] = self.analyze_mouse_pattern(data.get('mouse_data', []))

        # 设备风险评分
        features['device_risk_score'] = self.calculate_device_risk(data)

        return features

    def extract_network_features(self, data):
        """网络行为特征"""
        features = {}

        # IP地址分析
        ip_info = self.analyze_ip_address(data.get('ip_address'))
        features.update(ip_info)

        # 位置一致性
        features['location_consistency'] = self.check_location_consistency(
            data.get('ip_location'),
            data.get('declared_address')
        )

        # 网络环境
        features['is_proxy'] = self.detect_proxy(data.get('ip_address'))
        features['is_vpn'] = self.detect_vpn(data.get('ip_address'))

        return features

    def extract_behavioral_features(self, user_history):
        """行为模式特征"""
        features = {}

        if not user_history:
            return features

        # 时间模式分析
        login_times = [record['timestamp'] for record in user_history]
        features['login_frequency'] = len(login_times)
        features['active_hours'] = self.analyze_active_hours(login_times)
        features['weekend_activity'] = self.analyze_weekend_activity(login_times)

        # 交易模式分析
        transactions = [r for r in user_history if r.get('type') == 'transaction']
        if transactions:
            amounts = [t['amount'] for t in transactions]
            features['avg_transaction_amount'] = np.mean(amounts)
            features['transaction_variance'] = np.var(amounts)
            features['max_transaction_amount'] = max(amounts)

        # 申请模式分析
        applications = [r for r in user_history if r.get('type') == 'application']
        features['application_frequency'] = len(applications)
        features['application_success_rate'] = self.calculate_success_rate(applications)

        return features

    def extract_social_network_features(self, user_id, connection_data):
        """社交网络特征"""
        features = {}

        # 关联账户分析
        connected_accounts = self.find_connected_accounts(user_id, connection_data)
        features['connected_account_count'] = len(connected_accounts)

        # 风险传播分析
        risk_scores = [self.get_user_risk_score(acc) for acc in connected_accounts]
        if risk_scores:
            features['network_avg_risk'] = np.mean(risk_scores)
            features['network_max_risk'] = max(risk_scores)
            features['high_risk_connections'] = sum(1 for score in risk_scores if score > 0.7)

        # 团伙识别
        features['potential_gang_member'] = self.detect_gang_behavior(
            user_id, connected_accounts, connection_data
        )

        return features

    def analyze_typing_pattern(self, keystroke_data):
        """分析打字模式"""
        if not keystroke_data:
            return {}

        intervals = []
        for i in range(1, len(keystroke_data)):
            interval = keystroke_data[i]['timestamp'] - keystroke_data[i-1]['timestamp']
            intervals.append(interval)

        if intervals:
            return {
                'avg_typing_speed': np.mean(intervals),
                'typing_rhythm_variance': np.var(intervals)
            }
        return {}

    def calculate_device_risk(self, data):
        """计算设备风险评分"""
        risk_score = 0.0

        # 检查设备是否在黑名单
        if self.is_device_in_blacklist(data.get('device_id')):
            risk_score += 0.5

        # 检查设备使用频率
        device_usage = self.get_device_usage_count(data.get('device_id'))
        if device_usage > 10:  # 一个设备多个账户
            risk_score += 0.3

        # 检查设备信息一致性
        if not self.check_device_consistency(data):
            risk_score += 0.2

        return min(risk_score, 1.0)

    def detect_gang_behavior(self, user_id, connected_accounts, connection_data):
        """检测团伙行为"""
        # 简化版团伙检测逻辑

        # 检查是否有大量关联账户
        if len(connected_accounts) > 20:
            return True

        # 检查关联账户的注册时间集中度
        reg_times = [self.get_user_registration_time(acc) for acc in connected_accounts]
        if reg_times:
            time_variance = np.var(reg_times)
            if time_variance < 86400:  # 24小时内注册
                return True

        # 检查相似的个人信息
        similar_info_count = self.count_similar_personal_info(connected_accounts)
        if similar_info_count > 5:
            return True

        return False

# 实时评分引擎
class RealTimeFraudScoring:
    def __init__(self, model, feature_engine):
        self.model = model
        self.feature_engine = feature_engine
        self.rule_engine = FraudRuleEngine()

    def score_transaction(self, transaction_data):
        """实时交易评分"""
        # 1. 特征提取
        features = self.feature_engine.extract_all_features(transaction_data)

        # 2. 规则引擎预筛选
        rule_result = self.rule_engine.evaluate_rules(transaction_data)
        if rule_result['block']:
            return {
                'score': 1.0,
                'decision': 'BLOCK',
                'reason': rule_result['reason']
            }

        # 3. 模型评分
        model_score = self.model.predict_proba([features])[0][1]

        # 4. 决策逻辑
        if model_score > 0.8:
            decision = 'BLOCK'
        elif model_score > 0.5:
            decision = 'REVIEW'
        else:
            decision = 'PASS'

        return {
            'score': model_score,
            'decision': decision,
            'features': features,
            'rule_triggers': rule_result.get('triggers', [])
        }

# 规则引擎
class FraudRuleEngine:
    def __init__(self):
        self.rules = self.load_rules()

    def load_rules(self):
        """加载反欺诈规则"""
        return [
            {
                'name': 'blacklist_check',
                'condition': lambda data: self.check_blacklist(data),
                'action': 'BLOCK',
                'priority': 1
            },
            {
                'name': 'velocity_check',
                'condition': lambda data: self.check_velocity(data),
                'action': 'REVIEW',
                'priority': 2
            },
            {
                'name': 'amount_threshold',
                'condition': lambda data: data.get('amount', 0) > 50000,
                'action': 'REVIEW',
                'priority': 3
            }
        ]

    def evaluate_rules(self, data):
        """评估规则"""
        triggered_rules = []

        for rule in self.rules:
            if rule['condition'](data):
                triggered_rules.append(rule)

        # 按优先级排序
        triggered_rules.sort(key=lambda x: x['priority'])

        if triggered_rules:
            highest_priority = triggered_rules[0]
            return {
                'block': highest_priority['action'] == 'BLOCK',
                'reason': highest_priority['name'],
                'triggers': [r['name'] for r in triggered_rules]
            }

        return {'block': False, 'triggers': []}

# VaR模型实现
import numpy as np
import pandas as pd
from scipy import stats
import warnings
warnings.filterwarnings('ignore')

class VaRCalculator:
    def __init__(self, confidence_level=0.99, holding_period=1):
        self.confidence_level = confidence_level
        self.holding_period = holding_period

    def historical_simulation_var(self, returns, portfolio_weights=None):
        """历史模拟法计算VaR"""
        if portfolio_weights is not None:
            # 组合收益率
            portfolio_returns = np.dot(returns, portfolio_weights)
        else:
            portfolio_returns = returns

        # 排序收益率
        sorted_returns = np.sort(portfolio_returns)

        # 计算VaR
        alpha = 1 - self.confidence_level
        var_index = int(alpha * len(sorted_returns))
        var_value = -sorted_returns[var_index]

        # 调整持有期
        var_adjusted = var_value * np.sqrt(self.holding_period)

        return var_adjusted

    def parametric_var(self, returns, portfolio_weights=None):
        """参数法计算VaR"""
        if portfolio_weights is not None:
            portfolio_returns = np.dot(returns, portfolio_weights)
        else:
            portfolio_returns = returns

        # 计算统计量
        mean_return = np.mean(portfolio_returns)
        std_return = np.std(portfolio_returns)

        # 正态分布假设下的VaR
        z_score = stats.norm.ppf(1 - self.confidence_level)
        var_value = -(mean_return + z_score * std_return)

        # 调整持有期
        var_adjusted = var_value * np.sqrt(self.holding_period)

        return var_adjusted

    def monte_carlo_var(self, mu, sigma, correlation_matrix, portfolio_weights,
                       n_simulations=10000):
        """蒙特卡洛模拟法计算VaR"""
        n_assets = len(mu)

        # 生成随机收益率
        random_returns = np.random.multivariate_normal(
            mu, correlation_matrix, n_simulations
        )

        # 计算组合收益率
        portfolio_returns = np.dot(random_returns, portfolio_weights)

        # 计算VaR
        alpha = 1 - self.confidence_level
        var_value = -np.percentile(portfolio_returns, alpha * 100)

        # 调整持有期
        var_adjusted = var_value * np.sqrt(self.holding_period)

        return var_adjusted

    def expected_shortfall(self, returns, portfolio_weights=None):
        """预期损失（ES）计算"""
        if portfolio_weights is not None:
            portfolio_returns = np.dot(returns, portfolio_weights)
        else:
            portfolio_returns = returns

        # 排序收益率
        sorted_returns = np.sort(portfolio_returns)

        # 计算ES
        alpha = 1 - self.confidence_level
        var_index = int(alpha * len(sorted_returns))
        es_value = -np.mean(sorted_returns[:var_index])

        # 调整持有期
        es_adjusted = es_value * np.sqrt(self.holding_period)

        return es_adjusted

    def backtesting(self, var_forecasts, actual_returns):
        """VaR模型回测"""
        violations = actual_returns < -var_forecasts
        violation_rate = np.mean(violations)
        expected_violation_rate = 1 - self.confidence_level

        # Kupiec检验
        n = len(actual_returns)
        n_violations = np.sum(violations)

        if n_violations == 0:
            kupiec_stat = 0
        else:
            kupiec_stat = -2 * np.log(
                (expected_violation_rate ** n_violations) *
                ((1 - expected_violation_rate) ** (n - n_violations))
            ) + 2 * np.log(
                (violation_rate ** n_violations) *
                ((1 - violation_rate) ** (n - n_violations))
            )

        kupiec_p_value = 1 - stats.chi2.cdf(kupiec_stat, 1)

        return {
            'violation_rate': violation_rate,
            'expected_violation_rate': expected_violation_rate,
            'n_violations': n_violations,
            'kupiec_statistic': kupiec_stat,
            'kupiec_p_value': kupiec_p_value,
            'model_adequate': kupiec_p_value > 0.05
        }

# 使用示例
def demonstrate_var_calculation():
    """演示VaR计算"""
    # 模拟股票收益率数据
    np.random.seed(42)
    n_days = 1000
    n_assets = 3

    # 生成相关性矩阵
    correlation_matrix = np.array([
        [1.0, 0.3, 0.2],
        [0.3, 1.0, 0.4],
        [0.2, 0.4, 1.0]
    ])

    # 生成收益率
    mu = np.array([0.001, 0.0008, 0.0012])  # 日均收益率
    sigma = np.array([0.02, 0.025, 0.03])   # 日波动率

    cov_matrix = np.outer(sigma, sigma) * correlation_matrix
    returns = np.random.multivariate_normal(mu, cov_matrix, n_days)

    # 组合权重
    portfolio_weights = np.array([0.4, 0.35, 0.25])

    # 计算VaR
    var_calc = VaRCalculator(confidence_level=0.99)

    hist_var = var_calc.historical_simulation_var(returns, portfolio_weights)
    param_var = var_calc.parametric_var(returns, portfolio_weights)
    mc_var = var_calc.monte_carlo_var(mu, cov_matrix, portfolio_weights)
    es = var_calc.expected_shortfall(returns, portfolio_weights)

    print(f"历史模拟法VaR: {hist_var:.4f}")
    print(f"参数法VaR: {param_var:.4f}")
    print(f"蒙特卡洛VaR: {mc_var:.4f}")
    print(f"预期损失ES: {es:.4f}")

# 多因子选股模型
import pandas as pd
import numpy as np
from sklearn.preprocessing import StandardScaler
from sklearn.linear_model import LinearRegression
import warnings
warnings.filterwarnings('ignore')

class MultiFactorModel:
    def __init__(self):
        self.factor_categories = {
            'value': ['PE', 'PB', 'PS', 'EV_EBITDA'],
            'quality': ['ROE', 'ROA', 'debt_ratio', 'current_ratio'],
            'growth': ['revenue_growth', 'profit_growth', 'eps_growth'],
            'momentum': ['price_momentum_1m', 'price_momentum_3m', 'price_momentum_12m'],
            'volatility': ['volatility_20d', 'volatility_60d', 'beta'],
            'size': ['market_cap', 'float_cap']
        }
        self.factor_weights = {}
        self.scaler = StandardScaler()

    def calculate_factors(self, stock_data, price_data, financial_data):
        """计算各类因子"""
        factors = pd.DataFrame(index=stock_data.index)

        # 价值因子
        factors['PE'] = stock_data['market_cap'] / financial_data['net_profit']
        factors['PB'] = stock_data['market_cap'] / financial_data['net_assets']
        factors['PS'] = stock_data['market_cap'] / financial_data['revenue']

        # 质量因子
        factors['ROE'] = financial_data['net_profit'] / financial_data['net_assets']
        factors['ROA'] = financial_data['net_profit'] / financial_data['total_assets']
        factors['debt_ratio'] = financial_data['total_debt'] / financial_data['total_assets']

        # 成长因子
        factors['revenue_growth'] = financial_data['revenue'].pct_change(4)  # 年化增长
        factors['profit_growth'] = financial_data['net_profit'].pct_change(4)

        # 动量因子
        factors['price_momentum_1m'] = price_data['close'].pct_change(20)
        factors['price_momentum_3m'] = price_data['close'].pct_change(60)
        factors['price_momentum_12m'] = price_data['close'].pct_change(240)

        # 波动率因子
        factors['volatility_20d'] = price_data['close'].rolling(20).std()
        factors['volatility_60d'] = price_data['close'].rolling(60).std()

        # 市值因子
        factors['market_cap'] = stock_data['market_cap']

        return factors.dropna()

    def factor_preprocessing(self, factors):
        """因子预处理"""
        processed_factors = factors.copy()

        # 1. 极值处理（MAD方法）
        for col in processed_factors.columns:
            median_val = processed_factors[col].median()
            mad = np.median(np.abs(processed_factors[col] - median_val))

            # 3倍MAD原则
            upper_bound = median_val + 3 * mad
            lower_bound = median_val - 3 * mad

            processed_factors[col] = np.clip(
                processed_factors[col], lower_bound, upper_bound
            )

        # 2. 标准化处理
        processed_factors = pd.DataFrame(
            self.scaler.fit_transform(processed_factors),
            index=processed_factors.index,
            columns=processed_factors.columns
        )

        # 3. 因子方向调整（确保方向一致）
        factor_directions = {
            'PE': -1, 'PB': -1, 'PS': -1,  # 价值因子：越小越好
            'ROE': 1, 'ROA': 1,            # 质量因子：越大越好
            'debt_ratio': -1,              # 债务比率：越小越好
            'revenue_growth': 1, 'profit_growth': 1,  # 成长因子：越大越好
            'price_momentum_1m': 1,        # 动量因子：越大越好
            'volatility_20d': -1,          # 波动率：越小越好
            'market_cap': -1               # 市值：小市值效应
        }

        for factor, direction in factor_directions.items():
            if factor in processed_factors.columns:
                processed_factors[factor] *= direction

        return processed_factors

    def ic_analysis(self, factors, returns, periods=[1, 5, 20]):
        """信息系数（IC）分析"""
        ic_results = {}

        for period in periods:
            forward_returns = returns.shift(-period)
            ic_values = {}

            for factor in factors.columns:
                # 计算横截面相关系数
                ic_series = []
                for date in factors.index:
                    if date in forward_returns.index:
                        factor_values = factors.loc[date]
                        return_values = forward_returns.loc[date]

                        # 去除NaN值
                        valid_mask = ~(factor_values.isna() | return_values.isna())
                        if valid_mask.sum() > 10:  # 至少10个有效样本
                            ic = factor_values[valid_mask].corr(return_values[valid_mask])
                            ic_series.append(ic)

                if ic_series:
                    ic_values[factor] = {
                        'mean_ic': np.mean(ic_series),
                        'std_ic': np.std(ic_series),
                        'ir': np.mean(ic_series) / np.std(ic_series) if np.std(ic_series) > 0 else 0,
                        'ic_win_rate': np.mean(np.array(ic_series) > 0)
                    }

            ic_results[f'{period}d'] = ic_values

        return ic_results

    def factor_combination(self, factors, method='equal_weight'):
        """因子合成"""
        if method == 'equal_weight':
            # 等权重合成
            combined_score = factors.mean(axis=1)

        elif method == 'ic_weight':
            # IC加权合成
            ic_weights = {}
            for factor in factors.columns:
                # 这里简化处理，实际应该基于历史IC计算权重
                ic_weights[factor] = np.random.uniform(0.5, 1.5)

            # 归一化权重
            total_weight = sum(ic_weights.values())
            normalized_weights = {k: v/total_weight for k, v in ic_weights.values()}

            combined_score = sum(factors[factor] * weight
                               for factor, weight in normalized_weights.items())

        elif method == 'pca':
            # 主成分分析合成
            from sklearn.decomposition import PCA

            pca = PCA(n_components=1)
            pca_scores = pca.fit_transform(factors)
            combined_score = pd.Series(pca_scores.flatten(), index=factors.index)

        return combined_score

    def portfolio_construction(self, scores, n_stocks=50, method='top_n'):
        """组合构建"""
        if method == 'top_n':
            # 选择得分最高的N只股票
            selected_stocks = scores.nlargest(n_stocks)
            weights = pd.Series(1/n_stocks, index=selected_stocks.index)

        elif method == 'score_weight':
            # 按得分加权
            positive_scores = scores[scores > 0]
            selected_stocks = positive_scores.nlargest(n_stocks)

            # 归一化权重
            weights = selected_stocks / selected_stocks.sum()

        elif method == 'risk_parity':
            # 风险平价组合（简化版本）
            selected_stocks = scores.nlargest(n_stocks)

            # 假设所有股票风险相等
            weights = pd.Series(1/n_stocks, index=selected_stocks.index)

        return weights

    def backtest_performance(self, weights, returns, benchmark_returns=None):
        """回测绩效分析"""
        # 计算组合收益率
        portfolio_returns = (weights * returns).sum(axis=1)

        # 绩效指标计算
        annual_return = portfolio_returns.mean() * 252
        annual_volatility = portfolio_returns.std() * np.sqrt(252)
        sharpe_ratio = annual_return / annual_volatility if annual_volatility > 0 else 0

        # 最大回撤
        cumulative_returns = (1 + portfolio_returns).cumprod()
        running_max = cumulative_returns.expanding().max()
        drawdown = (cumulative_returns - running_max) / running_max
        max_drawdown = drawdown.min()

        performance_metrics = {
            'annual_return': annual_return,
            'annual_volatility': annual_volatility,
            'sharpe_ratio': sharpe_ratio,
            'max_drawdown': max_drawdown,
            'calmar_ratio': annual_return / abs(max_drawdown) if max_drawdown != 0 else 0
        }

        # 如果有基准，计算相对指标
        if benchmark_returns is not None:
            excess_returns = portfolio_returns - benchmark_returns
            tracking_error = excess_returns.std() * np.sqrt(252)
            information_ratio = excess_returns.mean() * 252 / tracking_error if tracking_error > 0 else 0

            performance_metrics.update({
                'excess_return': excess_returns.mean() * 252,
                'tracking_error': tracking_error,
                'information_ratio': information_ratio
            })

        return performance_metrics

# 策略回测框架
class QuantStrategy:
    def __init__(self, factor_model):
        self.factor_model = factor_model
        self.rebalance_frequency = 20  # 20个交易日调仓一次

    def run_backtest(self, start_date, end_date, stock_data, price_data, financial_data):
        """运行回测"""
        results = []
        current_date = start_date

        while current_date <= end_date:
            # 计算因子
            factors = self.factor_model.calculate_factors(
                stock_data.loc[current_date],
                price_data.loc[:current_date],
                financial_data.loc[current_date]
            )

            # 因子预处理
            processed_factors = self.factor_model.factor_preprocessing(factors)

            # 因子合成
            scores = self.factor_model.factor_combination(processed_factors)

            # 组合构建
            weights = self.factor_model.portfolio_construction(scores)

            # 计算下期收益
            next_date = current_date + pd.Timedelta(days=self.rebalance_frequency)
            if next_date <= end_date:
                period_returns = price_data.loc[next_date] / price_data.loc[current_date] - 1
                portfolio_return = (weights * period_returns).sum()

                results.append({
                    'date': current_date,
                    'portfolio_return': portfolio_return,
                    'weights': weights.to_dict()
                })

            current_date = next_date

        return pd.DataFrame(results)

# 风险模型
class RiskModel:
    def __init__(self):
        self.factor_exposure = None
        self.factor_covariance = None
        self.specific_risk = None

    def calculate_risk_exposure(self, factors):
        """计算风险暴露"""
        # 行业暴露
        industry_exposure = pd.get_dummies(factors['industry'])

        # 风格暴露
        style_factors = ['size', 'value', 'quality', 'growth', 'momentum', 'volatility']
        style_exposure = factors[style_factors]

        # 合并暴露矩阵
        exposure_matrix = pd.concat([industry_exposure, style_exposure], axis=1)
        return exposure_matrix

    def estimate_factor_returns(self, exposure_matrix, stock_returns):
        """估计因子收益率"""
        # 横截面回归
        factor_returns = {}

        for date in stock_returns.index:
            if date in exposure_matrix.index:
                X = exposure_matrix.loc[date].dropna()
                y = stock_returns.loc[date].dropna()

                # 确保X和y的索引对齐
                common_stocks = X.index.intersection(y.index)
                if len(common_stocks) > 50:  # 确保足够的样本量
                    X_aligned = X.loc[common_stocks]
                    y_aligned = y.loc[common_stocks]

                    # 加权最小二乘回归（市值加权）
                    model = LinearRegression()
                    model.fit(X_aligned, y_aligned)

                    factor_returns[date] = dict(zip(X_aligned.columns, model.coef_))

        return pd.DataFrame(factor_returns).T

    def calculate_portfolio_risk(self, weights, exposure_matrix,
                                factor_covariance, specific_risk):
        """计算组合风险"""
        # 组合因子暴露
        portfolio_exposure = weights @ exposure_matrix

        # 因子风险
        factor_risk = portfolio_exposure @ factor_covariance @ portfolio_exposure.T

        # 特异性风险
        specific_risk_contrib = weights @ np.diag(specific_risk) @ weights.T

        # 总风险
        total_risk = factor_risk + specific_risk_contrib

        return {
            'total_risk': np.sqrt(total_risk),
            'factor_risk': np.sqrt(factor_risk),
            'specific_risk': np.sqrt(specific_risk_contrib),
            'factor_contribution': factor_risk / total_risk,
            'specific_contribution': specific_risk_contrib / total_risk
        }

## 客户流失分析框架
### 1. 流失定义
- 主动流失：客户主动销卡
- 被动流失：超过X个月无交易
- 潜在流失：交易频率大幅下降

### 2. 流失原因分析
#### 产品因素
- 信用额度：是否满足客户需求
- 费用结构：年费、手续费合理性
- 产品功能：是否满足客户偏好

#### 服务因素
- 客服体验：响应时间、解决效率
- 渠道体验：线上线下服务质量
- 个性化程度：推荐和服务的精准性

#### 竞争因素
- 竞品优势：其他银行的优惠政策
- 市场变化：客户偏好转移
- 新兴产品：支付宝、微信支付等

### 3. 客户细分分析
#### 高价值流失客户
- 特征：高净值、高活跃度、高贡献度
- 挽回策略：专属客户经理、定制化产品

#### 普通流失客户
- 特征：中等价值、中等活跃度
- 挽回策略：优惠活动、产品升级

#### 低价值流失客户
- 特征：低活跃度、低贡献度
- 策略：自然流失，降低挽回成本

# 贷款定价模型
class LoanPricingModel:
    def __init__(self):
        self.risk_free_rate = 0.03  # 无风险利率
        self.funding_cost = 0.035   # 资金成本
        self.operation_cost_rate = 0.005  # 运营成本率
        self.target_roe = 0.15      # 目标ROE

    def calculate_risk_premium(self, pd_score, lgd_rate=0.4):
        """计算风险溢价"""
        # 违约概率转换为风险溢价
        expected_loss = pd_score * lgd_rate
        risk_premium = expected_loss * 1.5  # 风险调整系数
        return risk_premium

    def calculate_loan_rate(self, customer_profile):
        """计算贷款利率"""
        # 基础成本
        base_cost = self.funding_cost + self.operation_cost_rate

        # 风险溢价
        risk_premium = self.calculate_risk_premium(
            customer_profile['pd_score'],
            customer_profile.get('lgd_rate', 0.4)
        )

        # 资本成本
        capital_ratio = 0.08  # 资本充足率要求
        capital_cost = capital_ratio * self.target_roe

        # 竞争调整
        market_rate = customer_profile.get('market_rate', 0.12)
        competitive_adjustment = min(0.01, max(-0.01, market_rate - base_cost - risk_premium))

        # 最终利率
        final_rate = base_cost + risk_premium + capital_cost + competitive_adjustment

        return {
            'final_rate': final_rate,
            'base_cost': base_cost,
            'risk_premium': risk_premium,
            'capital_cost': capital_cost,
            'competitive_adjustment': competitive_adjustment
        }

    def sensitivity_analysis(self, customer_profile):
        """敏感性分析"""
        base_rate = self.calculate_loan_rate(customer_profile)['final_rate']

        scenarios = {
            'pd_+10%': customer_profile.copy(),
            'pd_-10%': customer_profile.copy(),
            'funding_cost_+50bp': customer_profile.copy(),
            'funding_cost_-50bp': customer_profile.copy()
        }

        # PD变化情景
        scenarios['pd_+10%']['pd_score'] *= 1.1
        scenarios['pd_-10%']['pd_score'] *= 0.9

        # 资金成本变化情景
        original_funding_cost = self.funding_cost

        sensitivity_results = {}
        for scenario, profile in scenarios.items():
            if 'funding_cost' in scenario:
                if '+50bp' in scenario:
                    self.funding_cost = original_funding_cost + 0.005
                else:
                    self.funding_cost = original_funding_cost - 0.005

            scenario_rate = self.calculate_loan_rate(profile)['final_rate']
            sensitivity_results[scenario] = {
                'rate': scenario_rate,
                'change': scenario_rate - base_rate,
                'change_bps': (scenario_rate - base_rate) * 10000
            }

            # 恢复原始资金成本
            self.funding_cost = original_funding_cost

        return sensitivity_results

## 实时风控系统架构
### 数据采集层
- 交易数据：实时交易流水
- 行为数据：用户操作行为
- 外部数据：征信、黑名单等

### 数据传输层
- Kafka集群：高吞吐量消息队列
- 分区策略：按用户ID分区保证顺序
- 容错机制：多副本、自动故障转移

### 实时计算层
- Flink流处理：毫秒级延迟
- 状态管理：用户行为状态维护
- 窗口计算：滑动窗口风险指标

### 决策引擎层
- 规则引擎：预定义风控规则
- 模型评分：机器学习模型
- 决策融合：多维度决策逻辑

### 存储层
- Redis：热数据缓存
- HBase：历史数据查询
- ES：日志检索分析

### 监控告警层
- 系统监控：性能、可用性
- 业务监控：风险指标、拦截率
- 实时告警：异常情况通知

## 银行财务仪表板设计
### KPI概览区域
- 净利润：当期vs目标vs同期
- ROE/ROA：盈利能力指标
- 资本充足率：监管指标
- 不良贷款率：资产质量

### 业务分析区域
- 资产负债结构：资产负债配置
- 收入结构分析：利息收入vs非息收入
- 成本结构分析：资金成本vs运营成本
- 地区业务分布：各地区贡献度

### 风险监控区域
- 风险敞口分布：行业、地区集中度
- 拨备覆盖率：风险缓释能力
- 流动性指标：流动性风险监控
- 市场风险VaR：交易风险监控

### 趋势分析区域
- 历史趋势图：关键指标时间序列
- 同业对比：与竞争对手比较
- 预测分析：基于历史的趋势预测

## 智能风控产品设计
### 产品定位
- 目标客户：中小银行、消金公司
- 核心价值：降低风险成本、提升审批效率
- 差异化优势：轻量化部署、快速上线

### 功能架构
#### 风险评估模块
- 信用评分模型：个人、企业评分
- 反欺诈模型：设备指纹、行为分析
- 预警模型：贷前贷中贷后预警

#### 决策引擎模块
- 规则配置：可视化规则编辑
- 策略管理：多策略版本管理
- A/B测试：策略效果验证

#### 监控分析模块
- 实时监控：风险指标监控
- 绩效分析：模型效果分析
- 报告生成：监管报告、经营报告

### 商业模式
- SaaS订阅：按用户数收费
- 按量计费：按评分次数收费
- 定制服务：个性化开发服务

# 反洗钱可疑交易识别
class AMLDetection:
    def __init__(self):
        self.suspicious_patterns = {
            'structuring': self.detect_structuring,
            'unusual_volume': self.detect_unusual_volume,
            'rapid_movement': self.detect_rapid_movement,
            'geographic_anomaly': self.detect_geographic_anomaly
        }

    def detect_structuring(self, transactions):
        """检测拆分交易（化整为零）"""
        # 查找接近报告阈值的多笔交易
        threshold = 50000  # 大额交易报告阈值

        suspicious_cases = []
        grouped = transactions.groupby(['account_id', 'date'])

        for (account, date), group in grouped:
            daily_total = group['amount'].sum()
            transaction_count = len(group)

            # 检测规律：多笔接近阈值的交易
            if (transaction_count >= 3 and
                daily_total >= threshold * 0.8 and
                group['amount'].max() < threshold):

                suspicious_cases.append({
                    'account_id': account,
                    'date': date,
                    'pattern': 'structuring',
                    'transaction_count': transaction_count,
                    'total_amount': daily_total,
                    'risk_score': self.calculate_structuring_risk(group)
                })

        return suspicious_cases

    def detect_unusual_volume(self, transactions, lookback_days=30):
        """检测异常交易量"""
        suspicious_cases = []

        for account in transactions['account_id'].unique():
            account_txns = transactions[transactions['account_id'] == account]

            # 计算历史平均交易量
            historical_avg = account_txns['amount'].rolling(lookback_days).mean()
            current_amount = account_txns['amount'].iloc[-1]

            # 异常检测：当前交易超过历史平均的5倍
            if current_amount > historical_avg.iloc[-1] * 5:
                suspicious_cases.append({
                    'account_id': account,
                    'pattern': 'unusual_volume',
                    'current_amount': current_amount,
                    'historical_avg': historical_avg.iloc[-1],
                    'risk_score': min(current_amount / historical_avg.iloc[-1] / 5, 1.0)
                })

        return suspicious_cases

    def detect_rapid_movement(self, transactions, time_threshold_hours=24):
        """检测资金快速流转"""
        suspicious_cases = []

        # 按账户分组，按时间排序
        for account in transactions['account_id'].unique():
            account_txns = transactions[
                transactions['account_id'] == account
            ].sort_values('timestamp')

            for i in range(len(account_txns) - 1):
                current_txn = account_txns.iloc[i]
                next_txn = account_txns.iloc[i + 1]

                time_diff = (next_txn['timestamp'] - current_txn['timestamp']).total_seconds() / 3600

                # 检测：大额入账后短时间内大额出账
                if (time_diff <= time_threshold_hours and
                    current_txn['type'] == 'credit' and
                    next_txn['type'] == 'debit' and
                    abs(current_txn['amount'] - next_txn['amount']) / current_txn['amount'] < 0.1):

                    suspicious_cases.append({
                        'account_id': account,
                        'pattern': 'rapid_movement',
                        'time_diff_hours': time_diff,
                        'amount': current_txn['amount'],
                        'risk_score': max(0, 1 - time_diff / time_threshold_hours)
                    })

        return suspicious_cases

    def generate_sar_report(self, suspicious_cases):
        """生成可疑交易报告"""
        high_risk_cases = [
            case for case in suspicious_cases
            if case.get('risk_score', 0) > 0.7
        ]

        sar_report = {
            'report_date': pd.Timestamp.now(),
            'total_suspicious_cases': len(suspicious_cases),
            'high_risk_cases': len(high_risk_cases),
            'patterns_detected': {},
            'recommendations': []
        }

        # 统计各类模式
        for case in suspicious_cases:
            pattern = case['pattern']
            if pattern not in sar_report['patterns_detected']:
                sar_report['patterns_detected'][pattern] = 0
            sar_report['patterns_detected'][pattern] += 1

        # 生成建议
        if high_risk_cases:
            sar_report['recommendations'].append(
                f"建议对{len(high_risk_cases)}个高风险案例进行人工审核"
            )

        return sar_report