飞算 JavaAI 金融科技场景实践：从风险防控到智能投顾的全链路技术革新

@Service
@Slf4j
public class CreditRiskDataService {
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    @Autowired
    private RiskDataMapper dataMapper;
    @Autowired
    private EncryptionService encryptionService;

    // 风控数据Topic
    private static final String RISK_DATA_TOPIC = "risk:data:realtime";
    // 用户数据缓存Key
    private static final String USER_RISK_DATA_KEY = "risk:user:data:";
    // 数据有效期（30天）
    private static final long DATA_EXPIRE_DAYS = 30;

    /**
     * 采集并预处理风控数据
     */
    public void collectRiskData(RiskDataDTO data) {
        // 1. 数据校验
        if (data.getUserId() == null) {
            log.warn("风控数据缺少用户ID，丢弃数据");
            return;
        }

        // 2. 敏感数据加密
        RiskDataDTO encryptedData = encryptSensitiveFields(data);

        // 3. 发送到Kafka进行实时处理
        kafkaTemplate.send(RISK_DATA_TOPIC, 
            data.getUserId().toString(), JSON.toJSONString(encryptedData));

        // 4. 本地缓存最近数据
        String cacheKey = USER_RISK_DATA_KEY + data.getUserId();
        redisTemplate.opsForList().leftPush(cacheKey, encryptedData);
        redisTemplate.opsForList().trim(cacheKey, 0, 999); // 保留最近1000条
        redisTemplate.expire(cacheKey, DATA_EXPIRE_DAYS, TimeUnit.DAYS);
    }

    /**
     * 实时处理风控数据
     */
    @KafkaListener(topics = RISK_DATA_TOPIC, groupId = "risk-data-processor")
    public void processRiskData(ConsumerRecord<String, String> record) {
        try {
            String userId = record.key();
            RiskDataDTO data = JSON.parseObject(record.value(), RiskDataDTO.class);

            // 1. 数据清洗与转换
            RiskDataCleaned cleanedData = dataCleaner.clean(data);
            if (cleanedData == null) {
                log.warn("用户{}的风控数据清洗失败", userId);
                return;
            }

            // 2. 数据 enrichment（补充外部数据）
            RiskDataEnriched enrichedData = dataEnricher.enrich(cleanedData);

            // 3. 提取特征
            Map<String, Object> features = featureExtractor.extract(enrichedData);

            // 4. 缓存特征数据
            String featureKey = "risk:user:features:" + userId;
            redisTemplate.opsForHash().putAll(featureKey, features);
            redisTemplate.expire(featureKey, DATA_EXPIRE_DAYS, TimeUnit.DAYS);

            // 5. 异步存储历史数据
            asyncService.saveRiskDataHistory(enrichedData);

            // 6. 如果用户有未完成的信贷申请，触发风控重评估
            if (hasPendingApplication(userId)) {
                kafkaTemplate.send("risk:reassess:trigger", userId, 
                    JSON.toJSONString(enrichedData.getTimestamp()));
            }

        } catch (Exception e) {
            log.error("处理风控数据失败", e);
        }
    }

    /**
     * 敏感数据加密
     */
    private RiskDataDTO encryptSensitiveFields(RiskDataDTO data) {
        RiskDataDTO encrypted = JSON.parseObject(JSON.toJSONString(data), RiskDataDTO.class);
        // 身份证加密
        if (encrypted.getIdCard() != null) {
            encrypted.setIdCard(encryptionService.encrypt(encrypted.getIdCard()));
        }
        // 银行卡号加密
        if (encrypted.getBankCard() != null) {
            encrypted.setBankCard(encryptionService.encrypt(encrypted.getBankCard()));
        }
        // 手机号加密
        if (encrypted.getPhone() != null) {
            encrypted.setPhone(encryptionService.encrypt(encrypted.getPhone()));
        }
        return encrypted;
    }
}

@Service
public class CreditRiskDecisionService {
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    @Autowired
    private RiskModelManager modelManager;
    @Autowired
    private DecisionRecordMapper recordMapper;
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    // 风控规则缓存Key
    private static final String RISK_RULES_KEY = "risk:rules:latest";
    // 决策结果缓存Key
    private static final String DECISION_RESULT_KEY = "risk:decision:";

    /**
     * 信贷申请实时风控决策
     */
    public RiskDecisionResult evaluateCreditApplication(CreditApplicationDTO application) {
        Long userId = application.getUserId();
        String applicationId = application.getApplicationId();
        log.info("开始评估用户{}的信贷申请{}", userId, applicationId);

        // 1. 初始化决策结果
        RiskDecisionResult result = new RiskDecisionResult();
        result.setApplicationId(applicationId);
        result.setUserId(userId);
        result.setDecisionTime(LocalDateTime.now());
        result.setDecisionSteps(new ArrayList<>());

        try {
            // 2. 基础校验（规则引擎）
            RuleCheckResult ruleResult = checkBasicRules(application);
            result.getDecisionSteps().add(buildDecisionStep("基础规则校验", ruleResult));
            
            if (!ruleResult.isPass()) {
                result.setFinalDecision("REJECT");
                result.setReason(ruleResult.getReason());
                saveDecisionResult(result);
                return result;
            }

            // 3. 获取用户特征数据
            Map<String, Object> features = getUserRiskFeatures(userId);
            if (features.isEmpty()) {
                result.setFinalDecision("REJECT");
                result.setReason("用户风险特征不足");
                saveDecisionResult(result);
                return result;
            }

            // 4. 机器学习模型评分
            ModelScoreResult modelResult = calculateModelScores(application, features);
            result.getDecisionSteps().add(buildDecisionStep("模型评分", modelResult));
            
            // 5. 综合决策
            String finalDecision = makeFinalDecision(modelResult, application);
            result.setFinalDecision(finalDecision);
            result.setScore(modelResult.getFinalScore());
            
            // 6. 高风险或大额申请需人工审核
            if ("MANUAL_REVIEW".equals(finalDecision)) {
                triggerManualReview(application, modelResult, features);
            }

            // 7. 保存决策结果
            saveDecisionResult(result);
            
            log.info("用户{}的信贷申请{}决策完成：{}", userId, applicationId, finalDecision);
            return result;
        } catch (Exception e) {
            log.error("信贷申请{}评估失败", applicationId, e);
            result.setFinalDecision("SYSTEM_ERROR");
            result.setReason("系统异常：" + e.getMessage());
            saveDecisionResult(result);
            return result;
        }
    }

    /**
     * 综合决策逻辑
     */
    private String makeFinalDecision(ModelScoreResult modelResult, CreditApplicationDTO application) {
        // 1. 基于模型分数的基础决策
        double score = modelResult.getFinalScore();
        if (score < 400) {
            return "REJECT"; // 低评分直接拒绝
        } else if (score < 600) {
            // 中等评分结合额度判断
            if (application.getAmount().compareTo(new BigDecimal("50000")) > 0) {
                return "MANUAL_REVIEW"; // 大额需人工审核
            } else {
                return "APPROVE"; // 小额自动通过
            }
        } else {
            // 高评分结合产品类型判断
            if ("CREDIT_CARD".equals(application.getProductType())) {
                return "APPROVE"; // 信用卡自动通过
            } else {
                // 贷款产品超过一定额度仍需审核
                if (application.getAmount().compareTo(new BigDecimal("200000")) > 0) {
                    return "MANUAL_REVIEW";
                } else {
                    return "APPROVE";
                }
            }
        }
    }

    /**
     * 基础规则校验
     */
    private RuleCheckResult checkBasicRules(CreditApplicationDTO application) {
        RuleCheckResult result = new RuleCheckResult();
        result.setPass(true);

        // 获取最新风控规则
        List<RiskRule> rules = (List<RiskRule>) redisTemplate.opsForValue().get(RISK_RULES_KEY);
        if (rules == null) {
            // 缓存未命中，从数据库加载
            rules = riskRuleMapper.selectAllEnabledRules();
            redisTemplate.opsForValue().set(RISK_RULES_KEY, rules, 1, TimeUnit.HOURS);
        }

        // 执行规则校验
        for (RiskRule rule : rules) {
            try {
                boolean rulePass = ruleExecutor.executeRule(rule, application);
                if (!rulePass) {
                    result.setPass(false);
                    result.setReason("触发规则：" + rule.getRuleName() + "(" + rule.getRuleCode() + ")");
                    result.setRuleCode(rule.getRuleCode());
                    return result; // 触发任何规则即返回
                }
            } catch (Exception e) {
                log.error("执行规则{}失败", rule.getRuleCode(), e);
                // 关键规则执行失败视为校验不通过
                if (rule.getSeverity() == RuleSeverity.CRITICAL) {
                    result.setPass(false);
                    result.setReason("关键规则执行失败：" + rule.getRuleCode());
                    return result;
                }
            }
        }

        return result;
    }
}

@Service
@Slf4j
public class FinancialTransactionService {
    @Autowired
    private AccountTccService accountTccService;
    @Autowired
    private TransactionMapper transactionMapper;
    @Autowired
    private TccTransactionManager tccManager;
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    /**
     * 执行转账交易（TCC模式）
     */
    public Result<TransactionVO> transfer(TransferDTO dto) {
        // 1. 参数校验
        validateTransferParams(dto);

        // 2. 生成交易号
        String transactionNo = generateTransactionNo();

        // 3. 创建本地交易记录（状态为处理中）
        Transaction transaction = createPendingTransaction(dto, transactionNo);
        transactionMapper.insert(transaction);

        // 4. 开启TCC事务
        TccTransaction tcc = tccManager.begin();
        try {
            // 5. 执行TCC Try阶段
            TransferResult result = accountTccService.transferTry(
                transactionNo, dto.getFromAccountId(), dto.getToAccountId(), 
                dto.getAmount(), dto.getRemark());

            if (!result.isSuccess()) {
                throw new BusinessException("转账预处理失败：" + result.getMsg());
            }

            // 6. 提交TCC事务
            tcc.commit();

            // 7. 更新交易状态为成功
            updateTransactionStatus(transactionNo, TransactionStatus.SUCCESS, null);

            // 8. 发送交易完成事件
            sendTransactionCompletedEvent(transactionNo, dto);

            // 9. 构建返回结果
            TransactionVO vo = convertToVO(transaction);
            vo.setStatus(TransactionStatus.SUCCESS);
            return Result.success(vo);

        } catch (Exception e) {
            // 10. 回滚TCC事务
            tcc.rollback();
            log.error("转账交易{}失败", transactionNo, e);

            // 11. 更新交易状态为失败
            updateTransactionStatus(transactionNo, TransactionStatus.FAILED, e.getMessage());

            return Result.fail("转账失败：" + e.getMessage());
        }
    }

    /**
     * 创建待处理的交易记录
     */
    private Transaction createPendingTransaction(TransferDTO dto, String transactionNo) {
        Transaction transaction = new Transaction();
        transaction.setTransactionNo(transactionNo);
        transaction.setFromAccountId(dto.getFromAccountId());
        transaction.setToAccountId(dto.getToAccountId());
        transaction.setAmount(dto.getAmount());
        transaction.setTransactionType(TransactionType.TRANSFER);
        transaction.setStatus(TransactionStatus.PROCESSING);
        transaction.setRemark(dto.getRemark());
        transaction.setCreateTime(LocalDateTime.now());
        transaction.setCurrency("CNY");
        return transaction;
    }

    /**
     * 更新交易状态
     */
    @Transactional(rollbackFor = Exception.class)
    public void updateTransactionStatus(String transactionNo, String status, String failReason) {
        Transaction update = new Transaction();
        update.setTransactionNo(transactionNo);
        update.setStatus(status);
        update.setUpdateTime(LocalDateTime.now());
        if (TransactionStatus.FAILED.equals(status) && failReason != null) {
            update.setFailReason(failReason.length() > 500 ? 
                failReason.substring(0, 500) : failReason);
        }
        transactionMapper.updateByTransactionNo(update);
    }
}

@Service
public class AccountConsistencyService {
    @Autowired
    private AccountMapper accountMapper;
    @Autowired
    private TransactionMapper transactionMapper;
    @Autowired
    private AccountLogMapper logMapper;
    @Autowired
    private RedissonClient redissonClient;
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    // 账户余额缓存Key
    private static final String ACCOUNT_BALANCE_KEY = "account:balance:";
    // 账户锁定Key
    private static final String ACCOUNT_LOCK_KEY = "lock:account:";

    /**
     * 账户余额核对与修复（定时任务）
     */
    @Scheduled(cron = "0 0 */2 * * ?") // 每2小时执行一次
    public void checkAndRepairAccountBalances() {
        log.info("开始账户余额一致性检查");

        // 1. 获取需要检查的账户列表（按批次处理）
        List<Long> accountIds = accountMapper.selectAccountIdsForConsistencyCheck();
        if (accountIds.isEmpty()) {
            log.info("没有需要检查的账户");
            return;
        }

        // 2. 逐个账户检查
        int total = accountIds.size();
        int errorCount = 0;
        int repairedCount = 0;

        for (Long accountId : accountIds) {
            try {
                ConsistencyCheckResult result = checkAccountConsistency(accountId);
                if (!result.isConsistent()) {
                    errorCount++;
                    log.warn("账户{}余额不一致，系统记录:{}，计算值:{}", 
                        accountId, result.getRecordedBalance(), result.getCalculatedBalance());

                    // 尝试自动修复
                    if (canAutoRepair(result)) {
                        boolean repaired = repairAccountBalance(accountId, result);
                        if (repaired) {
                            repairedCount++;
                            log.info("账户{}余额已自动修复", accountId);
                        } else {
                            log.error("账户{}余额自动修复失败", accountId);
                        }
                    } else {
                        // 无法自动修复，触发人工干预
                        triggerManualIntervention(accountId, result);
                    }
                }
            } catch (Exception e) {
                log.error("检查账户{}余额失败", accountId, e);
            }
        }

        log.info("账户余额一致性检查完成，共检查{}个账户，发现{}个不一致，已修复{}个",
            total, errorCount, repairedCount);

        // 3. 发送检查报告
        if (errorCount > 0) {
            sendConsistencyReport(total, errorCount, repairedCount);
        }
    }

    /**
     * 检查单个账户余额一致性
     */
    private ConsistencyCheckResult checkAccountConsistency(Long accountId) {
        ConsistencyCheckResult result = new ConsistencyCheckResult();
        result.setAccountId(accountId);

        // 1. 获取系统记录的余额
        Account account = accountMapper.selectById(accountId);
        if (account == null) {
            result.setConsistent(false);
            result.setReason("账户不存在");
            return result;
        }
        result.setRecordedBalance(account.getBalance());

        // 2. 通过交易记录计算余额
        BigDecimal calculated = calculateBalanceFromTransactions(accountId);
        result.setCalculatedBalance(calculated);

        // 3. 比较是否一致（允许微小误差，如浮点数精度问题）
        BigDecimal diff = result.getRecordedBalance().subtract(calculated).abs();
        if (diff.compareTo(new BigDecimal("0.01")) <= 0) {
            result.setConsistent(true);
        } else {
            result.setConsistent(false);
            result.setDifference(diff);
            result.setReason("余额差异超过0.01元");
        }

        return result;
    }

    /**
     * 从交易记录计算账户余额
     */
    private BigDecimal calculateBalanceFromTransactions(Long accountId) {
        // 初始余额
        BigDecimal initialBalance = accountMapper.selectInitialBalance(accountId);
        
        // 收入总额
        BigDecimal totalIn = transactionMapper.sumAmountByAccount(
            accountId, TransactionType.INCOME, null, null);
        
        // 支出总额
        BigDecimal totalOut = transactionMapper.sumAmountByAccount(
            accountId, TransactionType.OUTCOME, null, null);
        
        // 计算当前应有余额
        return initialBalance.add(totalIn).subtract(totalOut);
    }

    /**
     * 修复账户余额
     */
    private boolean repairAccountBalance(Long accountId, ConsistencyCheckResult result) {
        RLock lock = redissonClient.getLock(ACCOUNT_LOCK_KEY + accountId);
        try {
            // 获取锁
            boolean locked = lock.tryLock(5, 30, TimeUnit.SECONDS);
            if (!locked) {
                log.error("获取账户{}锁失败，无法修复", accountId);
                return false;
            }

            // 再次检查，防止并发问题
            ConsistencyCheckResult recheck = checkAccountConsistency(accountId);
            if (recheck.isConsistent()) {
                log.info("账户{}余额已自动恢复一致，无需修复", accountId);
                return true;
            }

            // 执行修复
            BigDecimal correctBalance = recheck.getCalculatedBalance();
            
            // 记录修复日志
            AccountRepairLog log = new AccountRepairLog();
            log.setAccountId(accountId);
            log.setBeforeBalance(recheck.getRecordedBalance());
            log.setAfterBalance(correctBalance);
            log.setDifference(recheck.getDifference());
            log.setRepairTime(LocalDateTime.now());
            log.setOperator("SYSTEM");
            log.setReason("自动修复余额不一致问题");
            logMapper.insert(log);
            
            // 更新账户余额
            Account update = new Account();
            update.setId(accountId);
            update.setBalance(correctBalance);
            update.setUpdateTime(LocalDateTime.now());
            accountMapper.updateById(update);
            
            // 更新缓存
            redisTemplate.opsForValue().set(
                ACCOUNT_BALANCE_KEY + accountId, correctBalance);
            
            // 发送修复事件
            kafkaTemplate.send("account:repair:completed", 
                accountId.toString(), JSON.toJSONString(log));
            
            return true;
        } catch (Exception e) {
            log.error("修复账户{}余额失败", accountId, e);
            return false;
        } finally {
            if (lock.isHeldByCurrentThread()) {
                lock.unlock();
            }
        }
    }
}

@Service
public class UserRiskProfileService {
    @Autowired
    private RiskAssessmentMapper assessmentMapper;
    @Autowired
    private UserBehaviorMapper behaviorMapper;
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    // 用户风险画像缓存Key
    private static final String RISK_PROFILE_KEY = "user:risk:profile:";
    // 风险评估有效期（3个月）
    private static final long PROFILE_EXPIRE_DAYS = 90;

    /**
     * 评估用户风险承受能力
     */
    public RiskProfile assessUserRiskProfile(Long userId, RiskAssessmentDTO assessment) {
        // 1. 记录用户风险评估答案
        saveRiskAssessmentAnswers(userId, assessment);

        // 2. 计算基础风险得分（基于问卷答案）
        int baseScore = calculateBaseRiskScore(assessment);

        // 3. 结合用户特征调整风险得分
        int adjustedScore = adjustScoreByUserFeatures(userId, baseScore);

        // 4. 确定风险等级
        String riskLevel = determineRiskLevel(adjustedScore);

        // 5. 构建风险画像
        RiskProfile profile = new RiskProfile();
        profile.setUserId(userId);
        profile.setRiskScore(adjustedScore);
        profile.setRiskLevel(riskLevel);
        profile.setAssessmentTime(LocalDateTime.now());
        profile.setValidUntil(LocalDateTime.now().plusDays(PROFILE_EXPIRE_DAYS));
        profile.setInvestmentHorizon(assessment.getInvestmentHorizon());
        profile.setInvestmentPurpose(assessment.getInvestmentPurpose());
        profile.setIncomeLevel(assessment.getIncomeLevel());
        profile.setNetWorth(assessment.getNetWorth());

        // 6. 保存风险画像
        assessmentMapper.insertRiskProfile(profile);

        // 7. 缓存风险画像
        String cacheKey = RISK_PROFILE_KEY + userId;
        redisTemplate.opsForValue().set(cacheKey, profile, PROFILE_EXPIRE_DAYS, TimeUnit.DAYS);

        // 8. 触发资产配置推荐
        kafkaTemplate.send("investment:recommendation:trigger", 
            userId.toString(), JSON.toJSONString(profile));

        return profile;
    }

    /**
     * 计算基础风险得分
     */
    private int calculateBaseRiskScore(RiskAssessmentDTO assessment) {
        int score = 0;

        // 1. 投资经验（0-30分）
        switch (assessment.getInvestmentExperience()) {
            case "NO_EXPERIENCE": score += 5; break;
            case "BEGINNER": score += 10; break;
            case "INTERMEDIATE": score += 20; break;
            case "ADVANCED": score += 30; break;
            default: score += 10;
        }

        // 2. 风险承受意愿（0-25分）
        for (RiskQuestionAnswer answer : assessment.getRiskToleranceAnswers()) {
            score += mapAnswerToScore(answer.getAnswer());
        }

        // 3. 投资期限（0-15分）
        switch (assessment.getInvestmentHorizon()) {
            case "LESS_THAN_1_YEAR": score += 5; break;
            case "1_TO_3_YEARS": score += 8; break;
            case "3_TO_5_YEARS": score += 12; break;
            case "MORE_THAN_5_YEARS": score += 15; break;
            default: score += 8;
        }

        // 4. 财务状况（0-30分）
        score += calculateFinancialSituationScore(assessment);

        return score;
    }

    /**
     * 确定风险等级
     */
    private String determineRiskLevel(int score) {
        if (score < 20) {
            return "CONSERVATIVE"; // 保守型
        } else if (score < 40) {
            return "MODERATE_CONSERVATIVE"; // 稳健偏保守型
        } else if (score < 60) {
            return "MODERATE"; // 稳健型
        } else if (score < 80) {
            return "MODERATE_AGGRESSIVE"; // 稳健偏进取型
        } else {
            return "AGGRESSIVE"; // 进取型
        }
    }

    /**
     * 根据市场变化和用户行为动态调整风险画像
     */
    @Scheduled(cron = "0 0 1 * * ?") // 每天凌晨1点执行
    public void dynamicallyAdjustRiskProfiles() {
        log.info("开始动态调整用户风险画像");

        // 获取需要调整的用户列表（最近30天有交易活动的用户）
        List<Long> userIds = behaviorMapper.selectActiveUserIds(30);
        if (userIds.isEmpty()) {
            log.info("没有需要调整风险画像的活跃用户");
            return;
        }

        // 获取最新市场状况
        MarketSituation market = marketService.getLatestMarketSituation();

        // 逐个用户调整
        for (Long userId : userIds) {
            try {
                adjustUserRiskProfile(userId, market);
            } catch (Exception e) {
                log.error("调整用户{}风险画像失败", userId, e);
            }
        }

        log.info("用户风险画像动态调整完成，共处理{}个用户", userIds.size());
    }
}

@Service
public class AssetAllocationService {
    @Autowired
    private RiskProfileService riskProfileService;
    @Autowired
    private MarketDataService marketDataService;
    @Autowired
    private PortfolioMapper portfolioMapper;
    @Autowired
    private AllocationStrategyManager strategyManager;
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    // 资产配置缓存Key
    private static final String ASSET_ALLOCATION_KEY = "portfolio:allocation:";
    // 调仓阈值配置
    private static final double REBALANCE_THRESHOLD = 0.05; // 5%偏差触发调仓

    /**
     * 为用户生成资产配置方案
     */
    public AssetAllocation generateAssetAllocation(Long userId, BigDecimal investmentAmount) {
        // 1. 获取用户风险画像
        RiskProfile profile = riskProfileService.getUserRiskProfile(userId);
        if (profile == null) {
            throw new BusinessException("请先完成风险评估");
        }

        // 2. 获取当前市场数据
        MarketData marketData = marketDataService.getLatestMarketData();

        // 3. 根据风险等级选择配置策略
        AllocationStrategy strategy = strategyManager.getStrategyByRiskLevel(profile.getRiskLevel());

        // 4. 生成基础资产配置比例
        Map<String, Double> baseAllocation = strategy.calculateBaseAllocation(marketData);

        // 5. 根据市场状况调整配置
        Map<String, Double> adjustedAllocation = strategy.adjustForMarketConditions(
            baseAllocation, marketData);

        // 6. 根据用户特征进一步调整
        Map<String, Double> finalAllocation = strategy.adjustForUserFeatures(
            adjustedAllocation, profile);

        // 7. 计算具体金额
        Map<String, BigDecimal> assetAmounts = calculateAssetAmounts(finalAllocation, investmentAmount);

        // 8. 选择具体投资标的
        List<InvestmentInstrument> instruments = selectInstruments(finalAllocation, marketData);

        // 9. 构建配置方案
        AssetAllocation allocation = new AssetAllocation();
        allocation.setUserId(userId);
        allocation.setAllocationId(generateAllocationId());
        allocation.setRiskLevel(profile.getRiskLevel());
        allocation.setTotalAmount(investmentAmount);
        allocation.setAssetAllocation(finalAllocation);
        allocation.setAssetAmounts(assetAmounts);
        allocation.setInstruments(instruments);
        allocation.setCreationTime(LocalDateTime.now());
        allocation.setStrategyName(strategy.getName());

        // 10. 保存配置方案
        portfolioMapper.insertAssetAllocation(allocation);

        // 11. 缓存配置方案
        String cacheKey = ASSET_ALLOCATION_KEY + allocation.getAllocationId();
        redisTemplate.opsForValue().set(cacheKey, allocation, 30, TimeUnit.DAYS);

        return allocation;
    }

    /**
     * 检查并执行调仓操作
     */
    public RebalanceResult checkAndRebalancePortfolio(Long portfolioId) {
        // 1. 获取投资组合当前状况
        Portfolio portfolio = portfolioMapper.selectById(portfolioId);
        if (portfolio == null) {
            throw new BusinessException("投资组合不存在");
        }

        // 2. 获取目标配置
        AssetAllocation targetAllocation = portfolioMapper.selectLatestAssetAllocation(portfolioId);
        if (targetAllocation == null) {
            throw new BusinessException("未找到资产配置方案");
        }

        // 3. 获取当前持仓
        List<PortfolioPosition> currentPositions = portfolioMapper.selectCurrentPositions(portfolioId);
        Map<String, Double> currentAllocation = calculateCurrentAllocation(currentPositions);

        // 4. 计算偏差
        Map<String, Double> deviations = calculateAllocationDeviations(
            targetAllocation.getAssetAllocation(), currentAllocation);

        // 5. 判断是否需要调仓
        boolean needRebalance = deviations.values().stream()
            .anyMatch(dev -> Math.abs(dev) > REBALANCE_THRESHOLD);

        if (!needRebalance) {
            RebalanceResult result = new RebalanceResult();
            result.setPortfolioId(portfolioId);
            result.setRebalanced(false);
            result.setReason("资产配置偏差在允许范围内，无需调仓");
            result.setDeviations(deviations);
            return result;
        }

        // 6. 执行调仓操作
        return executeRebalance(portfolioId, targetAllocation, currentPositions, deviations);
    }

    /**
     * 执行调仓操作
     */
    private RebalanceResult executeRebalance(
            Long portfolioId, AssetAllocation target, 
            List<PortfolioPosition> currentPositions, Map<String, Double> deviations) {
        // 1. 开启事务
        TransactionStatus status = transactionManager.getTransaction(
            new DefaultTransactionDefinition());

        try {
            // 2. 计算调仓方案
            RebalancePlan plan = calculateRebalancePlan(
                portfolioId, target, currentPositions, deviations);

            // 3. 执行交易指令
            List<TransactionResult> transactionResults = executeTransactions(plan);

            // 4. 更新持仓记录
            updatePortfolioPositions(portfolioId, transactionResults);

            // 5. 记录调仓历史
            RebalanceRecord record = createRebalanceRecord(portfolioId, plan, transactionResults);
            portfolioMapper.insertRebalanceRecord(record);

            // 6. 提交事务
            transactionManager.commit(status);

            // 7. 构建返回结果
            RebalanceResult result = new RebalanceResult();
            result.setPortfolioId(portfolioId);
            result.setRebalanced(true);
            result.setRebalanceId(record.getRebalanceId());
            result.setDeviations(deviations);
            result.setTransactions(transactionResults);
            result.setRebalanceTime(LocalDateTime.now());

            // 8. 发送调仓完成事件
            kafkaTemplate.send("portfolio:rebalanced", 
                portfolioId.toString(), JSON.toJSONString(result));

            return result;
        } catch (Exception e) {
            // 9. 回滚事务
            transactionManager.rollback(status);
            log.error("投资组合{}调仓失败", portfolioId, e);
            
            RebalanceResult result = new RebalanceResult();
            result.setPortfolioId(portfolioId);
            result.setRebalanced(false);
            result.setReason("调仓失败：" + e.getMessage());
            return result;
        }
    }

    /**
     * 计算当前资产配置比例
     */
    private Map<String, Double> calculateCurrentAllocation(List<PortfolioPosition> positions) {
        Map<String, Double> allocation = new HashMap<>();
        
        // 计算总资产价值
        BigDecimal totalValue = positions.stream()
            .map(PortfolioPosition::getCurrentValue)
            .reduce(BigDecimal.ZERO, BigDecimal::add);
        
        if (totalValue.compareTo(BigDecimal.ZERO) <= 0) {
            return allocation;
        }
        
        // 计算各类资产占比
        Map<String, BigDecimal> assetCategoryValues = new HashMap<>();
        for (PortfolioPosition position : positions) {
            String category = position.getAssetCategory();
            BigDecimal currentValue = position.getCurrentValue();
            
            assetCategoryValues.put(category, 
                assetCategoryValues.getOrDefault(category, BigDecimal.ZERO).add(currentValue));
        }
        
        // 转换为百分比
        assetCategoryValues.forEach((category, value) -> {
            double percentage = value.divide(totalValue, 4, RoundingMode.HALF_UP).doubleValue();
            allocation.put(category, percentage);
        });
        
        return allocation;
    }
}

// 金融交易合规引擎
public class FinancialComplianceEngine {
    private final List<ComplianceRule> rules = new ArrayList<>();
    
    public FinancialComplianceEngine() {
        // 初始化金融合规规则
        rules.add(new AntiMoneyLaunderingRule()); // 反洗钱规则
        rules.add(new TransactionLimitRule()); // 交易限额规则
        rules.add(new SanctionsListRule()); // 制裁名单规则
        rules.add(new InsiderTradingRule()); // 内幕交易规则
        rules.add(new MarketManipulationRule()); // 市场操纵规则
        rules.add(new KYCVerificationRule()); // 客户身份验证规则
    }
    
    /**
     * 交易前合规校验
     */
    public ComplianceCheckResult checkTransaction(FinancialTransaction transaction) {
        ComplianceCheckResult result = new ComplianceCheckResult();
        result.setTransactionId(transaction.getTransactionId());
        result.setCheckTime(LocalDateTime.now());
        result.setPass(true);
        
        for (ComplianceRule rule : rules) {
            RuleViolation violation = rule.check(transaction);
            if (violation != null) {
                result.setPass(false);
                result.addViolation(violation);
                // 严重违规直接终止检查
                if (violation.getSeverity() == Severity.CRITICAL) {
                    // 记录可疑交易
                    reportSuspiciousTransaction(transaction, violation);
                    return result;
                }
            }
        }
        
        return result;
    }
    
    /**
     * 报告可疑交易
     */
    private void reportSuspiciousTransaction(FinancialTransaction transaction, RuleViolation violation) {
        try {
            SuspiciousTransactionReport report = new SuspiciousTransactionReport();
            report.setTransactionId(transaction.getTransactionId());
            report.setReportId("STR_" + System.currentTimeMillis());
            report.setReportTime(LocalDateTime.now());
            report.setReason(violation.getDescription());
            report.setRuleCode(violation.getRuleCode());
            report.setTransactionType(transaction.getTransactionType());
            report.setAmount(transaction.getAmount());
            report.setFromAccount(transaction.getFromAccountId());
            report.setToAccount(transaction.getToAccountId());
            
            // 保存可疑交易报告
            suspiciousTransactionRepository.save(report);
            
            // 发送可疑交易通知
            kafkaTemplate.send("compliance:suspicious:transaction", 
                report.getReportId(), JSON.toJSONString(report));
        } catch (Exception e) {
            log.error("报告可疑交易失败", e);
        }
    }
}

// 反洗钱规则示例
public class AntiMoneyLaunderingRule implements ComplianceRule {
    @Autowired
    private CustomerProfileService customerProfileService;
    @Autowired
    private TransactionHistoryService historyService;
    
    @Override
    public RuleViolation check(FinancialTransaction transaction) {
        // 1. 检查交易金额是否超过大额交易标准
        if (isLargeAmountTransaction(transaction)) {
            // 大额交易需额外审核
            if (!isVerifiedForLargeTransactions(transaction.getFromAccountId())) {
                return new RuleViolation(
                    "AML-001",
                    "大额交易未通过审核",
                    Severity.CRITICAL,
                    "该交易金额超过大额交易标准，且账户未通过大额交易审核"
                );
            }
            
            // 记录大额交易报告
            reportLargeTransaction(transaction);
        }
        
        // 2. 检查是否存在拆分交易嫌疑（规避大额交易报告）
        if (hasStructuringSuspicion(transaction)) {
            return new RuleViolation(
                "AML-002",
                "存在拆分交易嫌疑",
                Severity.CRITICAL,
                "短时间内发生多笔接近大额标准的交易，可能存在拆分规避嫌疑"
            );
        }
        
        // 3. 检查交易模式是否与客户 profile 匹配
        if (!isTransactionPatternConsistent(transaction)) {
            return new RuleViolation(
                "AML-003",
                "交易模式与客户历史行为不符",
                Severity.HIGH,
                "该交易金额、频率或对手方与客户历史行为模式存在显著差异"
            );
        }
        
        // 4. 检查是否涉及高风险国家/地区
        if (involvesHighRiskCountry(transaction)) {
            return new RuleViolation(
                "AML-004",
                "交易涉及高风险国家/地区",
                Severity.HIGH,
                "交易对手方所在国家/地区被列为高风险洗钱地区"
            );
        }
        
        return null;
    }
    
    /**
     * 判断是否为大额交易
     */
    private boolean isLargeAmountTransaction(FinancialTransaction transaction) {
        // 人民币5万元以上，外币等值1万美元以上视为大额交易
        if ("CNY".equals(transaction.getCurrency())) {
            return transaction.getAmount().compareTo(new BigDecimal("50000")) > 0;
        } else {
            // 转换为美元比较
            BigDecimal usdAmount = currencyConverter.convert(
                transaction.getAmount(), transaction.getCurrency(), "USD");
            return usdAmount.compareTo(new BigDecimal("10000")) > 0;
        }
    }
}

@Aspect
@Component
public class FinancialDataSecurityAspect {
    @Autowired
    private EncryptionService encryptionService;
    @Autowired
    private AuditLogService auditLogService;
    @Autowired
    private DataMaskingService maskingService;

    /**
     * 金融敏感数据加密存储切面
     */
    @Around("@annotation(financialDataEncrypt)")
    public Object encryptFinancialData(
            ProceedingJoinPoint joinPoint, FinancialDataEncrypt financialDataEncrypt) throws Throwable {
        // 1. 获取方法参数
        Object[] args = joinPoint.getArgs();
        if (args == null || args.length == 0) {
            return joinPoint.proceed();
        }
        
        // 2. 对金融敏感字段进行加密
        for (Object arg : args) {
            if (arg instanceof FinancialEntity) {
                encryptFinancialEntity((FinancialEntity) arg);
            }
        }
        
        // 3. 执行目标方法
        return joinPoint.proceed(args);
    }

    /**
     * 金融敏感数据解密返回切面
     */
    @Around("@annotation(financialDataDecrypt)")
    public Object decryptFinancialData(
            ProceedingJoinPoint joinPoint, FinancialDataDecrypt financialDataDecrypt) throws Throwable {
        // 1. 执行目标方法
        Object result = joinPoint.proceed();
        
        // 2. 对返回结果解密（根据用户权限决定解密程度）
        UserContext context = SecurityUtils.getCurrentUserContext();
        if (context == null) {
            throw new AccessDeniedException("未授权访问");
        }
        
        decryptObject(result, context.getPermissionLevel());
        
        return result;
    }

    /**
     * 金融数据访问审计切面
     */
    @AfterReturning(
        pointcut = "@annotation(financialDataAudit)",
        returning = "result"
    )
    public void auditFinancialDataAccess(
            JoinPoint joinPoint, FinancialDataAudit financialDataAudit, Object result) {
        // 记录金融数据访问日志
        FinancialDataAccessLog log = new FinancialDataAccessLog();
        log.setAccessId(UUID.randomUUID().toString());
        log.setOperatorId(SecurityUtils.getCurrentUserId());
        log.setOperatorName(SecurityUtils.getCurrentUserName());
        log.setAccessTime(LocalDateTime.now());
        log.setModule(financialDataAudit.module());
        log.setOperation(financialDataAudit.operation());
        log.setIpAddress(IpUtils.getIpAddr());
        log.setUserAgent(SecurityUtils.getUserAgent());
        
        // 提取访问的关键ID（如账户ID、交易ID）
        String targetId = extractTargetId(joinPoint.getArgs(), financialDataAudit.idParamIndex());
        log.setTargetId(targetId);
        
        // 记录访问结果
        log.setSuccess(result != null && !(result instanceof Result) || 
            (result instanceof Result && ((Result<?>) result).isSuccess()));
        
        auditLogService.saveFinancialDataAccessLog(log);
    }

    /**
     * 加密金融实体
     */
    private void encryptFinancialEntity(FinancialEntity entity) {
        // 反射获取所有字段，对标记金融敏感注解的字段加密
        Field[] fields = entity.getClass().getDeclaredFields();
        for (Field field : fields) {
            if (field.isAnnotationPresent(FinancialSensitiveField.class)) {
                try {
                    field.setAccessible(true);
                    Object value = field.get(entity);
                    if (value != null && value instanceof String) {
                        String encryptedValue = encryptionService.encryptFinancialData((String) value);
                        field.set(entity, encryptedValue);
                    }
                } catch (Exception e) {
                    log.error("加密金融字段失败: {}", field.getName(), e);
                }
            }
        }
    }
    
    /**
     * 根据权限解密对象
     */
    private void decryptObject(Object obj, int permissionLevel) {
        if (obj == null) {
            return;
        }
        
        // 反射获取所有字段，对标记金融敏感注解的字段根据权限解密
        Field[] fields = obj.getClass().getDeclaredFields();
        for (Field field : fields) {
            if (field.isAnnotationPresent(FinancialSensitiveField.class)) {
                try {
                    field.setAccessible(true);
                    Object value = field.get(obj);
                    if (value != null && value instanceof String) {
                        String processedValue;
                        if (permissionLevel >= 3) {
                            // 高权限：完全解密
                            processedValue = encryptionService.decryptFinancialData((String) value);
                        } else if (permissionLevel >= 2) {
                            // 中权限：部分显示
                            String decrypted = encryptionService.decryptFinancialData((String) value);
                            processedValue = maskingService.maskFinancialData(decrypted, field.getName());
                        } else {
                            // 低权限：完全掩码
                            processedValue = maskingService.completelyMask(field.getName());
                        }
                        field.set(obj, processedValue);
                    }
                } catch (Exception e) {
                    log.error("处理金融字段失败: {}", field.getName(), e);
                }
            }
        }
    }
}

@Service
public class IntelligentRiskEngine {
    @Autowired
    private RuleEngine ruleEngine;
    @Autowired
    private ModelInferenceService modelService;
    @Autowired
    private FeatureService featureService;
    @Autowired
    private RiskDecisionMapper decisionMapper;

    /**
     * 智能风控决策（规则+模型）
     */
    public RiskDecision intelligentRiskDecision(CreditApplication application) {
        Long userId = application.getUserId();
        String applicationId = application.getApplicationId();
        
        // 1. 构建决策上下文
        RiskDecisionContext context = new RiskDecisionContext();
        context.setApplicationId(applicationId);
        context.setUserId(userId);
        context.setDecisionTime(LocalDateTime.now());
        
        try {
            // 2. 实时特征计算
            Map<String, Object> features = featureService.calculateRealTimeFeatures(userId, application);
            context.setFeatures(features);
            
            // 3. 执行规则引擎（快速拒绝明显不符合条件的申请）
            RuleExecutionResult ruleResult = ruleEngine.executeRules(
                "CREDIT_APPLICATION", features, application);
            context.setRuleResult(ruleResult);
            
            if (!ruleResult.isPass()) {
                // 规则拒绝
                return buildRejectionDecision(context, "RULE", 
                    ruleResult.getRejectionReason(), ruleResult.getRuleCode());
            }
            
            // 4. 执行模型评分
            ModelScoreResult modelResult = modelService.predict(
                "CREDIT_RISK_MODEL_V3", features);
            context.setModelResult(modelResult);
            
            // 5. 综合决策
            String finalDecision = makeFinalDecision(ruleResult, modelResult, application);
            context.setFinalDecision(finalDecision);
            
            // 6. 生成决策解释
            context.setExplanation(generateExplanation(context));
            
            // 7. 保存决策结果
            saveDecisionResult(context);
            
            return buildDecisionResult(context);
        } catch (Exception e) {
            log.error("智能风控决策失败，applicationId:{}", applicationId, e);
            return buildSystemErrorDecision(context, e.getMessage());
        }
    }

    /**
     * 生成决策解释（满足可解释性要求）
     */
    private DecisionExplanation generateExplanation(RiskDecisionContext context) {
        DecisionExplanation explanation = new DecisionExplanation();
        explanation.setApplicationId(context.getApplicationId());
        
        // 规则通过原因
        List<String> ruleReasons = context.getRuleResult().getPassReasons().stream()
            .map(Reason -> "符合规则: " + reason)
            .collect(Collectors.toList());
        
        // 模型评分解释
        List<String> modelFactors = modelExplainer.explainTopFactors(
            context.getModelResult(), 5); // 取影响最大的5个因素
        
        // 综合解释
        if ("APPROVE".equals(context.getFinalDecision())) {
            explanation.setSummary("您的贷款申请已通过审批，主要基于您良好的信用记录和还款能力");
        } else if ("REJECT".equals(context.getFinalDecision())) {
            explanation.setSummary("很抱歉，您的贷款申请未通过审批，主要原因是" + 
                modelFactors.get(0));
        } else {
            explanation.setSummary("您的贷款申请需要进一步人工审核");
        }
        
        explanation.setRuleReasons(ruleReasons);
        explanation.setModelFactors(modelFactors);
        explanation.setCalculationTime(context.getDecisionTime());
        
        return explanation;
    }
}