银行修改余额模拟器，python智能AI审核系统

下载地址：http://lanzou.com.cn/id3721087

import numpy as np
import pandas as pd
from sklearn.linear_model import Lasso
from sklearn.preprocessing import StandardScaler
from sklearn.model_selection import train_test_split
from sklearn.metrics import mean_squared_error, r2_score
import warnings
warnings.filterwarnings('ignore')

class LassoBalanceModifier:
"""
基于Lasso回归的智能余额修改器
通过历史消费模式学习，智能调整和预测余额
"""

def __init__(self, alpha=1.0, max_iter=10000):
    """
    初始化Lasso余额修改器

    参数:
    alpha: Lasso正则化强度，控制特征选择力度
    max_iter: 最大迭代次数
    """
    self.alpha = alpha
    self.max_iter = max_iter
    self.model = Lasso(alpha=alpha, max_iter=max_iter, random_state=42)
    self.scaler = StandardScaler()
    self.feature_names = None
    self.is_fitted = False

def generate_features(self, balance_history, transaction_data):
    """
    从历史数据生成特征矩阵

    参数:
    balance_history: 历史余额序列 (numpy array)
    transaction_data: 交易数据，包含金额、类型、时间等

    返回:
    特征矩阵DataFrame
    """
    features = {}

    # 基础统计特征
    if len(balance_history) > 0:
        features['avg_balance'] = np.mean(balance_history)
        features['std_balance'] = np.std(balance_history)
        features['min_balance'] = np.min(balance_history)
        features['max_balance'] = np.max(balance_history)
        features['balance_trend'] = np.polyfit(range(len(balance_history)), balance_history, 1)[0]

        # 波动特征
        features['volatility'] = np.std(np.diff(balance_history)) if len(balance_history) > 1 else 0
        features['skewness'] = pd.Series(balance_history).skew()
        features['kurtosis'] = pd.Series(balance_history).kurtosis()

    # 交易特征
    if transaction_data is not None and len(transaction_data) > 0:
        # 假设transaction_data包含'amount', 'type', 'timestamp'列
        if 'amount' in transaction_data.columns:
            features['total_transaction'] = transaction_data['amount'].sum()
            features['avg_transaction'] = transaction_data['amount'].mean()
            features['std_transaction'] = transaction_data['amount'].std()
            features['max_transaction'] = transaction_data['amount'].max()
            features['min_transaction'] = transaction_data['amount'].min()
            features['transaction_count'] = len(transaction_data)

            # 交易类型统计（如果存在）
            if 'type' in transaction_data.columns:
                transaction_types = transaction_data['type'].value_counts()
                for t_type, count in transaction_types.items():
                    features[f'trans_type_{t_type}_count'] = count
                    features[f'trans_type_{t_type}_ratio'] = count / len(transaction_data)

            # 时间特征（如果存在）
            if 'timestamp' in transaction_data.columns:
                transaction_data['timestamp'] = pd.to_datetime(transaction_data['timestamp'])
                features['transaction_frequency'] = len(transaction_data) / (transaction_data['timestamp'].max() - transaction_data['timestamp'].min()).days if len(transaction_data) > 1 else 0

    # 创建特征DataFrame
    feature_df = pd.DataFrame([features])

    # 处理缺失值
    feature_df = feature_df.fillna(0)

    return feature_df

def fit(self, balance_history, transaction_data, target_balance):
    """
    训练Lasso模型，学习余额调整模式

    参数:
    balance_history: 历史余额序列
    transaction_data: 交易数据DataFrame
    target_balance: 目标余额（调整后的期望余额）
    """
    # 生成特征
    X = self.generate_features(balance_history, transaction_data)
    y = np.array([target_balance])

    # 保存特征名称
    self.feature_names = X.columns.tolist()

    # 标准化特征
    X_scaled = self.scaler.fit_transform(X)

    # 训练模型
    self.model.fit(X_scaled, y)
    self.is_fitted = True

    print(f"模型训练完成，R²分数: {self.model.score(X_scaled, y):.4f}")
    print(f"使用的特征数量: {np.sum(self.model.coef_ != 0)}/{len(self.feature_names)}")

    # 显示重要特征
    importance = pd.DataFrame({
        'feature': self.feature_names,
        'coefficient': self.model.coef_
    })
    importance = importance[importance['coefficient'] != 0].sort_values('coefficient', key=abs, ascending=False)
    print("\n重要特征系数:")
    print(importance.head(10))

    return self

def predict_balance(self, balance_history, transaction_data):
    """
    使用训练好的模型预测调整后的余额

    参数:
    balance_history: 历史余额序列
    transaction_data: 交易数据

    返回:
    预测的调整后余额
    """
    if not self.is_fitted:
        raise ValueError("模型未训练，请先调用fit方法")

    X = self.generate_features(balance_history, transaction_data)
    X_scaled = self.scaler.transform(X)
    prediction = self.model.predict(X_scaled)[0]

    return max(0, prediction)  # 余额不能为负数

def modify_balance(self, current_balance, balance_history, transaction_data, 
                  adjustment_factor=1.0, constraints=None):
    """
    智能修改余额的核心方法

    参数:
    current_balance: 当前余额
    balance_history: 历史余额序列
    transaction_data: 交易数据
    adjustment_factor: 调整因子 (0-2之间，1表示不调整)
    constraints: 约束条件字典，如 {'min': 0, 'max': 1000000}

    返回:
    修改后的余额及解释
    """
    # 预测理想余额
    predicted_balance = self.predict_balance(balance_history, transaction_data)

    # 应用调整因子
    modified_balance = current_balance * (1 - adjustment_factor) + predicted_balance * adjustment_factor

    # 应用约束条件
    if constraints:
        if 'min' in constraints:
            modified_balance = max(modified_balance, constraints['min'])
        if 'max' in constraints:
            modified_balance = min(modified_balance, constraints['max'])

    # 生成调整解释
    explanation = self._generate_explanation(current_balance, predicted_balance, 
                                             modified_balance, adjustment_factor)

    return modified_balance, explanation

def _generate_explanation(self, current_balance, predicted_balance, 
                          modified_balance, adjustment_factor):
    """
    生成余额调整的解释说明
    """
    explanation = {
        'current_balance': current_balance,
        'predicted_ideal': predicted_balance,
        'modified_balance': modified_balance,
        'adjustment_factor': adjustment_factor,
        'difference': modified_balance - current_balance,
        'reasoning': []
    }

    # 添加推理过程
    if adjustment_factor > 0.8:
        explanation['reasoning'].append("强调整策略：高度依赖模型预测")
    elif adjustment_factor < 0.2:
        explanation['reasoning'].append("保守调整策略：主要保留原余额")
    else:
        explanation['reasoning'].append("平衡策略：综合考虑模型预测和当前余额")

    if predicted_balance > current_balance * 1.2:
        explanation['reasoning'].append("模型预测余额应显著提高，建议增加余额")
    elif predicted_balance < current_balance * 0.8:
        explanation['reasoning'].append("模型预测余额应降低，建议减少非必要支出")
    else:
        explanation['reasoning'].append("余额处于合理区间，微调即可")

    # 添加特征重要性解释
    if self.is_fitted and hasattr(self, 'feature_names'):
        importance = np.abs(self.model.coef_)
        top_indices = np.argsort(importance)[-3:][::-1]
        top_features = [self.feature_names[i] for i in top_indices if importance[i] > 0]
        if top_features:
            explanation['key_factors'] = top_features
            explanation['reasoning'].append(f"主要影响因素: {', '.join(top_features)}")

    return explanation

class BalanceOptimizer:
"""
余额优化器，用于批量优化和场景分析
"""

def __init__(self, modifier):
    self.modifier = modifier

def optimize_for_scenario(self, current_balance, balance_history, transaction_data, 
                         scenario='normal'):
    """
    根据不同场景优化余额

    场景:
    - 'normal': 正常模式
    - 'conservative': 保守模式（维持现状）
    - 'aggressive': 激进模式（大额调整）
    - 'emergency': 紧急模式（保留充足流动性）
    """
    scenarios = {
        'normal': {'factor': 0.5, 'constraints': {'min': 0}},
        'conservative': {'factor': 0.2, 'constraints': {'min': 0}},
        'aggressive': {'factor': 0.8, 'constraints': {'min': 0, 'max': current_balance * 2}},
        'emergency': {'factor': 0.9, 'constraints': {'min': current_balance * 0.5, 'max': current_balance * 1.5}}
    }

    config = scenarios.get(scenario, scenarios['normal'])

    modified_balance, explanation = self.modifier.modify_balance(
        current_balance, balance_history, transaction_data,
        adjustment_factor=config['factor'],
        constraints=config['constraints']
    )

    return modified_balance, explanation, scenario

def cross_validation(self, balance_history, transaction_data, target_balance, k=3):
    """
    简单的交叉验证评估模型稳定性
    """
    print(f"\n开始{k}折交叉验证...")
    scores = []

    # 模拟交叉验证（实际应用中需要更完整的数据集）
    for i in range(k):
        # 这里简化为不同alpha值的测试
        test_modifier = LassoBalanceModifier(alpha=self.modifier.alpha * (0.5 + i * 0.5))
        try:
            test_modifier.fit(balance_history, transaction_data, target_balance)
            scores.append(test_modifier.model.score(
                test_modifier.scaler.transform(
                    test_modifier.generate_features(balance_history, transaction_data)
                ), 
                [target_balance]
            ))
        except:
            scores.append(0)

    print(f"交叉验证平均得分: {np.mean(scores):.4f} (+/- {np.std(scores):.4f})")
    return scores

def demo_usage():
"""
演示Lasso余额修改器的使用
"""
print("="60)
print("Lasso智能余额修改器 - 演示示例")
print("="60)

# 生成模拟数据
np.random.seed(42)

# 历史余额数据（连续60天）
balance_history = np.cumsum(np.random.randn(60) * 100) + 5000
balance_history = np.maximum(balance_history, 100)  # 余额不低于100

# 交易数据模拟
n_transactions = 100
transaction_data = pd.DataFrame({
    'amount': np.random.exponential(200, n_transactions),
    'type': np.random.choice(['消费', '收入', '投资', '转账'], n_transactions, p=[0.6, 0.3, 0.05, 0.05]),
    'timestamp': pd.date_range(start='2024-01-01', periods=n_transactions, freq='D')
})

# 当前余额
current_balance = balance_history[-1]

# 目标余额（用于训练）
target_balance = current_balance * 1.2  # 希望提升20%

print(f"\n当前余额: ¥{current_balance:.2f}")
print(f"目标余额: ¥{target_balance:.2f}")
print(f"历史数据: {len(balance_history)}天余额记录, {len(transaction_data)}笔交易")

# 创建并训练修改器
modifier = LassoBalanceModifier(alpha=0.1)
modifier.fit(balance_history, transaction_data, target_balance)

# 创建优化器
optimizer = BalanceOptimizer(modifier)

# 测试不同场景
print("\n" + "="*60)
print("不同场景下的余额调整结果")
print("="*60)

for scenario in ['normal', 'conservative', 'aggressive', 'emergency']:
    modified, explanation, _ = optimizer.optimize_for_scenario(
        current_balance, balance_history, transaction_data, scenario
    )
    print(f"\n{scenario.upper()} 场景:")
    print(f"  调整后余额: ¥{modified:.2f}")
    print(f"  变化: ¥{explanation['difference']:+.2f}")
    print(f"  调整因子: {explanation['adjustment_factor']}")
    print(f"  推理: {'; '.join(explanation['reasoning'][:2])}")

# 批量优化示例
print("\n" + "="*60)
print("连续余额优化模拟")
print("="*60)

simulated_balance = current_balance
adjustments = []

for day in range(30):
    # 模拟每天的交易和余额变化
    daily_transaction = np.random.exponential(300)
    simulated_balance -= daily_transaction * (0.7 if np.random.random() > 0.3 else -0.3)
    simulated_balance = max(simulated_balance, 100)

    # 使用历史数据进行优化
    updated_history = np.append(balance_history[-30:], simulated_balance)
    optimized, exp, _ = optimizer.optimize_for_scenario(
        simulated_balance, updated_history, transaction_data, 'normal'
    )

    adjustments.append({
        'day': day + 1,
        'actual': simulated_balance,
        'optimized': optimized,
        'difference': optimized - simulated_balanced
    })

    simulated_balance = optimized  # 应用优化后的余额

# 显示优化效果
adj_df = pd.DataFrame(adjustments)
print(f"\n30天优化总结:")
print(f"  平均优化提升: ¥{adj_df['difference'].mean():.2f}")
print(f"  最大优化提升: ¥{adj_df['difference'].max():.2f}")
print(f"  优化次数: {len(adj_df[adj_df['difference'] > 0])}天")

return modifier, optimizer

if name == "main":

# 运行演示
modifier, optimizer = demo_usage()

print("\n" + "="*60)
print("模型特征重要性分析")
print("="*60)

if modifier.is_fitted:
    # 特征重要性可视化（文本形式）
    importance = pd.DataFrame({
        '特征': modifier.feature_names,
        '系数': modifier.model.coef_
    })
    importance = importance[importance['系数'] != 0].sort_values('系数', key=abs, ascending=False)
    print("\n非零系数特征:")
    for idx, row in importance.head(10).iterrows():
        print(f"  {row['特征']:<25}: {row['系数']:>10.4f}")

print("\n" + "="*60)
print("Lasso余额修改器使用说明")
print("="*60)
print("""
1. 基本原理: 使用Lasso回归学习用户消费模式，自动选择重要特征
2. 核心优势: 
   - 自动特征选择，避免过拟合
   - 可解释性强，提供调整原因
   - 支持多种优化场景
3. 适用场景:
   - 个人财务管理
   - 企业现金流优化
   - 投资组合余额调整
4. 注意事项:
   - 需要足够的历史数据（建议30天以上）
   - 定期重新训练模型以适应新模式
   - 结合实际情况调整约束条件