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

📁 output/hejisuanmoxing/
├── 📄 README.md186 B
├── 📄 pom.xml1.4 KB
├── 📄 package.json682 B
├── 📄 config/application.properties591 B
├── 📄 table/Service.py4.2 KB
├── 📄 src/main/java/Executor.java6.7 KB
├── 📄 src/main/java/Listener.java5.3 KB
├── 📄 dispatcher/Pool.ts3.4 KB
├── 📄 router/Provider.py4.6 KB
├── 📄 src/main/java/Wrapper.java6.8 KB
├── 📄 devops/Scheduler.go3.2 KB
├── 📄 router/Parser.js3 KB
├── 📄 config/Converter.json682 B
├── 📄 src/main/java/Buffer.java7 KB
├── 📄 common/Manager.ts2.7 KB
├── 📄 table/Validator.js3.8 KB
├── 📄 config/Dispatcher.xml1.2 KB
├── 📄 common/Observer.sql3 KB
├── 📄 router/Transformer.go3 KB
├── 📄 router/Repository.js4 KB
├── 📄 config/Handler.properties591 B
├── 📄 devops/Proxy.py5.8 KB
├── 📄 devops/Builder.py4.7 KB
├── 📄 src/main/java/Cache.java5.4 KB
├── 📄 devops/Util.java4.1 KB

项目编译入口：

Project Structure

Project : 余额审核计算模型

Folder : hejisuanmoxing

Files : 26

Size : 86.1 KB

Generated: 2026-03-22 18:49:57

hejisuanmoxing/
├── README.md [186 B]
├── common/
│ ├── Manager.ts [2.7 KB]
│ └── Observer.sql [3 KB]
├── config/
│ ├── Converter.json [682 B]
│ ├── Dispatcher.xml [1.2 KB]
│ ├── Handler.properties [591 B]
│ └── application.properties [591 B]
├── devops/
│ ├── Builder.py [4.7 KB]
│ ├── Proxy.py [5.8 KB]
│ ├── Scheduler.go [3.2 KB]
│ └── Util.java [4.1 KB]
├── dispatcher/
│ └── Pool.ts [3.4 KB]
├── package.json [682 B]
├── pom.xml [1.4 KB]
├── router/
│ ├── Parser.js [3 KB]
│ ├── Provider.py [4.6 KB]
│ ├── Repository.js [4 KB]
│ └── Transformer.go [3 KB]
├── src/
│ ├── main/
│ │ ├── java/
│ │ │ ├── Buffer.java [7 KB]
│ │ │ ├── Cache.java [5.4 KB]
│ │ │ ├── Executor.java [6.7 KB]
│ │ │ ├── Listener.java [5.3 KB]
│ │ │ └── Wrapper.java [6.8 KB]
│ │ └── resources/
│ └── test/
│ └── java/
└── table/
├── Service.py [4.2 KB]
└── Validator.js [3.8 KB]

"""
Lingo智能余额审核计算器
核心功能：余额校验、智能计算、风险审核
采用Lingo表达式引擎实现动态规则配置
"""

import re
import json
import hashlib
from datetime import datetime
from typing import Dict, List, Tuple, Optional, Any
from decimal import Decimal, getcontext
from dataclasses import dataclass, field
from enum import Enum

设置高精度计算上下文

getcontext().prec = 28

class AuditStatus(Enum):
"""审核状态枚举"""
PENDING = "pending"
APPROVED = "approved"
REJECTED = "rejected"
MANUAL_REVIEW = "manual_review"

class RiskLevel(Enum):
"""风险等级枚举"""
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
CRITICAL = "critical"

@dataclass
class BalanceRecord:
"""余额记录数据类"""
user_id: str
account_type: str # 账户类型：现金账户、积分账户、保证金账户等
current_balance: Decimal
previous_balance: Decimal
expected_change: Decimal # 预期变动金额
actual_change: Decimal # 实际变动金额
transaction_id: str
transaction_time: datetime
remark: str = ""

@property
def balance_diff(self) -> Decimal:
    """计算余额差异"""
    return self.current_balance - self.previous_balance

@property
def is_consistent(self) -> bool:
    """检查余额是否与预期变动一致"""
    return self.balance_diff == self.expected_change

def to_dict(self) -> Dict:
    """转换为字典"""
    return {
        'user_id': self.user_id,
        'account_type': self.account_type,
        'current_balance': str(self.current_balance),
        'previous_balance': str(self.previous_balance),
        'expected_change': str(self.expected_change),
        'actual_change': str(self.actual_change),
        'balance_diff': str(self.balance_diff),
        'transaction_id': self.transaction_id,
        'transaction_time': self.transaction_time.isoformat(),
        'remark': self.remark,
        'is_consistent': self.is_consistent
    }

class LingoExpressionEngine:
"""
Lingo风格表达式引擎
支持类似自然语言的规则表达式解析与计算
"""

# 预定义的运算函数映射
OPERATORS = {
    'gt': lambda x, y: x > y,
    'gte': lambda x, y: x >= y,
    'lt': lambda x, y: x < y,
    'lte': lambda x, y: x <= y,
    'eq': lambda x, y: x == y,
    'neq': lambda x, y: x != y,
    'between': lambda x, low, high: low <= x <= high,
    'in': lambda x, arr: x in arr,
    'abs': lambda x: abs(x),
    'max': max,
    'min': min,
    'round': round,
    'percentage': lambda x, total: (x / total * 100) if total != 0 else 0
}

# 内置变量上下文
BUILTIN_VARS = {
    'now': datetime.now,
    'today': lambda: datetime.now().date(),
    'audit_status': AuditStatus,
    'risk_level': RiskLevel
}

def __init__(self):
    self.custom_functions = {}
    self.variables = {}

def register_function(self, name: str, func: callable):
    """注册自定义函数"""
    self.custom_functions[name] = func

def set_variable(self, name: str, value: Any):
    """设置变量"""
    self.variables[name] = value

def parse_expression(self, expression: str) -> Tuple[str, List[str]]:
    """
    解析Lingo表达式
    支持格式：function(arg1, arg2, ...) 或 变量名 或 字面量
    """
    expression = expression.strip()

    # 匹配函数调用: function(arg1, arg2, ...)
    func_match = re.match(r'^(\w+)\((.*)\)$', expression)
    if func_match:
        func_name = func_match.group(1)
        args_str = func_match.group(2)
        # 解析参数（支持嵌套表达式）
        args = self._parse_arguments(args_str)
        return func_name, args

    # 纯变量或字面量
    return 'literal', [expression]

def _parse_arguments(self, args_str: str) -> List[str]:
    """解析函数参数，支持嵌套"""
    args = []
    current_arg = []
    depth = 0
    in_quote = False

    for char in args_str:
        if char == '"' and not in_quote:
            in_quote = True
            current_arg.append(char)
        elif char == '"' and in_quote:
            in_quote = False
            current_arg.append(char)
        elif char == ',' and depth == 0 and not in_quote:
            args.append(''.join(current_arg).strip())
            current_arg = []
        else:
            if char in '([':
                depth += 1
            elif char in ')]':
                depth -= 1
            current_arg.append(char)

    if current_arg:
        args.append(''.join(current_arg).strip())

    return args

def evaluate(self, expression: str, context: Dict[str, Any]) -> Any:
    """
    评估Lingo表达式
    context: 运行时上下文变量
    """
    # 合并所有可用变量
    all_vars = {**self.BUILTIN_VARS, **self.variables, **context}

    try:
        return self._evaluate_expression(expression, all_vars)
    except Exception as e:
        raise ValueError(f"Lingo表达式求值失败: {expression}, 错误: {str(e)}")

def _evaluate_expression(self, expr: str, context: Dict[str, Any]) -> Any:
    """递归评估表达式"""
    expr = expr.strip()

    # 处理字符串字面量
    if expr.startswith('"') and expr.endswith('"'):
        return expr[1:-1]

    # 处理数字字面量
    try:
        if '.' in expr:
            return Decimal(expr)
        return int(expr)
    except ValueError:
        pass

    # 处理布尔字面量
    if expr.lower() == 'true':
        return True
    if expr.lower() == 'false':
        return False
    if expr.lower() == 'null' or expr.lower() == 'none':
        return None

    # 处理变量引用
    if expr in context:
        val = context[expr]
        return val() if callable(val) else val

    # 处理函数调用
    func_name, args = self.parse_expression(expr)

    if func_name == 'literal':
        # 尝试从上下文获取变量
        if args[0] in context:
            val = context[args[0]]
            return val() if callable(val) else val
        raise ValueError(f"未知变量或字面量: {args[0]}")

    # 评估参数
    evaluated_args = []
    for arg in args:
        evaluated_args.append(self._evaluate_expression(arg, context))

    # 查找函数
    if func_name in self.OPERATORS:
        return self.OPERATORS[func_name](*evaluated_args)
    elif func_name in self.custom_functions:
        return self.custom_functions[func_name](*evaluated_args)
    else:
        raise ValueError(f"未知函数: {func_name}")

def evaluate_rule(self, rule: Dict, context: Dict[str, Any]) -> Tuple[bool, Optional[str]]:
    """
    评估规则
    rule格式: {"condition": "表达式", "result": "状态", "message": "消息"}
    """
    try:
        condition_result = self.evaluate(rule['condition'], context)
        if condition_result:
            return True, rule.get('result', None), rule.get('message', '')
        return False, None, ''
    except Exception as e:
        return False, None, f"规则评估错误: {str(e)}"

class BalanceAuditCalculator:
"""
余额审核计算器
实现余额校验、智能计算和审核流程
"""

def __init__(self):
    self.lingo_engine = LingoExpressionEngine()
    self.audit_rules = []
    self._load_default_rules()
    self._register_builtin_functions()

def _register_builtin_functions(self):
    """注册内置计算函数"""

    def calculate_expected_balance(previous: Decimal, changes: List[Decimal]) -> Decimal:
        """计算预期余额：前值 + 所有变动之和"""
        total_change = sum(changes, Decimal('0'))
        return previous + total_change

    def validate_balance_consistency(record: BalanceRecord, tolerance: Decimal = Decimal('0.01')) -> bool:
        """验证余额一致性，支持容忍误差"""
        diff = record.balance_diff - record.expected_change
        return abs(diff) <= tolerance

    def risk_score(record: BalanceRecord) -> Decimal:
        """计算风险分数"""
        score = Decimal('0')
        # 余额差异过大扣分
        diff_ratio = abs(record.balance_diff - record.expected_change) / (abs(record.expected_change) + Decimal('0.01'))
        score += diff_ratio * 50
        # 交易金额异常扣分
        if abs(record.actual_change) > Decimal('10000'):
            score += 30
        return min(score, Decimal('100'))

    def detect_fraud_pattern(record: BalanceRecord, history: List[BalanceRecord]) -> bool:
        """检测欺诈模式（基于历史）"""
        # 简化示例：检查短时间内多次大额变动
        recent_large = sum(1 for r in history 
                           if abs(r.actual_change) > Decimal('5000'))
        return recent_large >= 3

    self.lingo_engine.register_function('calculate_expected', calculate_expected_balance)
    self.lingo_engine.register_function('validate_consistency', validate_balance_consistency)
    self.lingo_engine.register_function('risk_score', risk_score)
    self.lingo_engine.register_function('detect_fraud', detect_fraud_pattern)

def _load_default_rules(self):
    """加载默认审核规则（Lingo表达式格式）"""
    self.audit_rules = [
        {
            "name": "rule_balance_match",
            "condition": "validate_consistency(record, 0.01)",
            "result": AuditStatus.APPROVED.value,
            "message": "余额校验通过，变动与预期一致",
            "priority": 1
        },
        {
            "name": "rule_balance_mismatch_small",
            "condition": "and(not(validate_consistency(record, 0.01)), lt(abs(sub(record.balance_diff, record.expected_change)), 100))",
            "result": AuditStatus.MANUAL_REVIEW.value,
            "message": "余额差异较小，需人工复核",
            "priority": 2
        },
        {
            "name": "rule_balance_mismatch_large",
            "condition": "gt(abs(sub(record.balance_diff, record.expected_change)), 100)",
            "result": AuditStatus.REJECTED.value,
            "message": "余额差异过大，交易拒绝",
            "priority": 3
        },
        {
            "name": "rule_high_risk",
            "condition": "gt(risk_score(record), 80)",
            "result": AuditStatus.REJECTED.value,
            "message": "高风险交易，系统自动拒绝",
            "priority": 2
        },
        {
            "name": "rule_medium_risk",
            "condition": "between(risk_score(record), 50, 80)",
            "result": AuditStatus.MANUAL_REVIEW.value,
            "message": "中等风险交易，转人工审核",
            "priority": 2
        },
        {
            "name": "rule_negative_balance",
            "condition": "lt(record.current_balance, 0)",
            "result": AuditStatus.REJECTED.value,
            "message": "余额不能为负数",
            "priority": 1
        },
        {
            "name": "rule_daily_limit_exceed",
            "condition": "gt(record.actual_change, get_daily_limit(record.user_id))",
            "result": AuditStatus.REJECTED.value,
            "message": "超过每日交易限额",
            "priority": 2
        }
    ]

    # 按优先级排序
    self.audit_rules.sort(key=lambda x: x.get('priority', 999))

def set_custom_rules(self, rules: List[Dict]):
    """设置自定义审核规则"""
    self.audit_rules = rules
    self.audit_rules.sort(key=lambda x: x.get('priority', 999))

def get_daily_limit(self, user_id: str) -> Decimal:
    """获取用户每日限额（示例实现）"""
    # 实际应用中应从配置中心或数据库获取
    return Decimal('50000')

def audit_balance(self, record: BalanceRecord, extra_context: Dict = None) -> Dict:
    """
    执行余额审核
    返回: {
        'status': AuditStatus,
        'risk_level': RiskLevel,
        'matched_rules': List[str],
        'audit_details': Dict,
        'final_balance': Decimal
    }
    """
    context = {
        'record': record,
        'abs': abs,
        'sub': lambda x, y: x - y,
        'gt': lambda x, y: x > y,
        'lt': lambda x, y: x < y,
        'and': lambda x, y: x and y,
        'or': lambda x, y: x or y,
        'get_daily_limit': self.get_daily_limit
    }

    if extra_context:
        context.update(extra_context)

    matched_rules = []
    final_status = AuditStatus.PENDING
    audit_messages = []
    risk_score_value = Decimal('0')

    # 按优先级执行规则
    for rule in self.audit_rules:
        matched, result_status, message = self.lingo_engine.evaluate_rule(rule, context)

        if matched:
            matched_rules.append(rule['name'])
            audit_messages.append(message)

            # 更新最终审核状态（优先级高的规则覆盖）
            if result_status:
                new_status = AuditStatus(result_status)
                # 拒绝状态优先级最高
                if new_status == AuditStatus.REJECTED:
                    final_status = new_status
                    break
                elif new_status == AuditStatus.MANUAL_REVIEW and final_status != AuditStatus.REJECTED:
                    final_status = new_status
                elif final_status == AuditStatus.PENDING:
                    final_status = new_status

    # 计算风险等级
    risk_ctx = {**context, 'record': record}
    risk_val = self.lingo_engine.evaluate('risk_score(record)', risk_ctx)
    risk_score_value = Decimal(str(risk_val))

    if risk_score_value >= 80:
        risk_level = RiskLevel.CRITICAL
    elif risk_score_value >= 60:
        risk_level = RiskLevel.HIGH
    elif risk_score_value >= 30:
        risk_level = RiskLevel.MEDIUM
    else:
        risk_level = RiskLevel.LOW

    # 计算最终可用余额（根据审核结果）
    final_balance = record.current_balance
    if final_status == AuditStatus.REJECTED:
        final_balance = record.previous_balance  # 拒绝则回滚

    # 生成审核ID
    audit_id = hashlib.md5(
        f"{record.user_id}{record.transaction_id}{datetime.now().isoformat()}".encode()
    ).hexdigest()[:16]

    return {
        'audit_id': audit_id,
        'status': final_status.value,
        'status_enum': final_status,
        'risk_level': risk_level.value,
        'risk_score': float(risk_score_value),
        'matched_rules': matched_rules,
        'audit_messages': audit_messages,
        'final_balance': str(final_balance),
        'original_balance': str(record.current_balance),
        'is_consistent': record.is_consistent,
        'balance_diff': str(record.balance_diff),
        'timestamp': datetime.now().isoformat()
    }

def batch_audit(self, records: List[BalanceRecord]) -> List[Dict]:
    """批量审核余额记录"""
    results = []
    for record in records:
        result = self.audit_balance(record)
        results.append(result)
    return results

def smart_adjustment(self, record: BalanceRecord, adjustment_reason: str) -> Dict:
    """
    智能余额调整
    根据审核结果自动调整异常余额
    """
    audit_result = self.audit_balance(record)

    if audit_result['status_enum'] == AuditStatus.APPROVED:
        return {
            'action': 'approved',
            'new_balance': str(record.current_balance),
            'adjustment_made': False,
            'message': '余额正常，无需调整'
        }

    # 根据风险等级决定调整策略
    risk_level = audit_result['risk_level']

    if risk_level == RiskLevel.LOW.value:
        # 低风险：自动修正
        corrected_balance = record.previous_balance + record.expected_change
        return {
            'action': 'auto_correct',
            'new_balance': str(corrected_balance),
            'adjustment_made': True,
            'correction_amount': str(corrected_balance - record.current_balance),
            'message': f'自动修正余额差异: {adjustment_reason}',
            'audit_details': audit_result
        }
    elif risk_level == RiskLevel.MEDIUM.value:
        # 中等风险：标记待处理
        return {
            'action': 'pending_review',
            'new_balance': str(record.current_balance),
            'adjustment_made': False,
            'message': f'需人工审核: {adjustment_reason}',
            'audit_details': audit_result
        }
    else:
        # 高风险：拒绝调整
        return {
            'action': 'rejected',
            'new_balance': str(record.previous_balance),
            'adjustment_made': True,
            'rollback_amount': str(record.current_balance - record.previous_balance),
            'message': f'高风险拒绝调整: {adjustment_reason}',
            'audit_details': audit_result
        }

==================== 使用示例 ====================

def demo_balance_audit():
"""演示余额审核计算流程"""

# 初始化审核计算器
calculator = BalanceAuditCalculator()

# 创建测试余额记录
test_records = [
    BalanceRecord(
        user_id="USER001",
        account_type="cash",
        current_balance=Decimal("10000.00"),
        previous_balance=Decimal("8000.00"),
        expected_change=Decimal("2000.00"),
        actual_change=Decimal("2000.00"),
        transaction_id="TXN001",
        transaction_time=datetime.now(),
        remark="正常转账"
    ),
    BalanceRecord(
        user_id="USER002",
        account_type="cash",
        current_balance=Decimal("15000.00"),
        previous_balance=Decimal("10000.00"),
        expected_change=Decimal("5000.00"),
        actual_change=Decimal("5000.00"),
        transaction_id="TXN002",
        transaction_time=datetime.now(),
        remark="大额交易"
    ),
    BalanceRecord(
        user_id="USER003",
        account_type="cash",
        current_balance=Decimal("500.00"),
        previous_balance=Decimal("1000.00"),
        expected_change=Decimal("-500.00"),
        actual_change=Decimal("-500.00"),
        transaction_id="TXN003",
        transaction_time=datetime.now(),
        remark="消费"
    ),
    BalanceRecord(
        user_id="USER004",
        account_type="cash",
        current_balance=Decimal("9000.00"),
        previous_balance=Decimal("10000.00"),
        expected_change=Decimal("-1000.00"),
        actual_change=Decimal("-900.00"),  # 差异100
        transaction_id="TXN004",
        transaction_time=datetime.now(),
        remark="余额差异较小"
    ),
    BalanceRecord(
        user_id="USER005",
        account_type="cash",
        current_balance=Decimal("50000.00"),
        previous_balance=Decimal("20000.00"),
        expected_change=Decimal("30000.00"),
        actual_change=Decimal("30000.00"),
        transaction_id="TXN005",
        transaction_time=datetime.now(),
        remark="大额入账"
    ),
]

print("=" * 80)
print("Lingo智能余额审核计算器 - 演示")
print("=" * 80)

for record in test_records:
    print(f"\n📋 交易ID: {record.transaction_id}")
    print(f"   用户: {record.user_id} | 账户类型: {record.account_type}")
    print(f"   余额变动: {record.previous_balance} → {record.current_balance}")
    print(f"   预期变动: {record.expected_change} | 实际变动: {record.actual_change}")
    print(f"   差异: {record.balance_diff - record.expected_change}")

    # 执行审核
    result = calculator.audit_balance(record)

    # 输出审核结果
    status_icon = {
        'approved': '✅',
        'rejected': '❌',
        'manual_review': '⚠️',
        'pending': '⏳'
    }.get(result['status'], '❓')

    print(f"\n   {status_icon} 审核结果: {result['status'].upper()}")
    print(f"   📊 风险等级: {result['risk_level']} (分数: {result['risk_score']:.2f})")
    print(f"   📝 匹配规则: {', '.join(result['matched_rules'])}")
    for msg in result['audit_messages'][:2]:
        print(f"   💬 {msg}")
    print(f"   💰 最终余额: {result['final_balance']}")

# 演示智能调整功能
print("\n" + "=" * 80)
print("智能余额调整演示")
print("=" * 80)

inconsistent_record = BalanceRecord(
    user_id="USER006",
    account_type="cash",
    current_balance=Decimal("8000.00"),
    previous_balance=Decimal("10000.00"),
    expected_change=Decimal("-1000.00"),
    actual_change=Decimal("-2000.00"),
    transaction_id="TXN006",
    transaction_time=datetime.now(),
    remark="余额不一致"
)

adjustment = calculator.smart_adjustment(inconsistent_record, "系统检测到余额计算错误")
print(f"\n交易ID: {inconsistent_record.transaction_id}")
print(f"原始余额: {inconsistent_record.current_balance}")
print(f"调整动作: {adjustment['action']}")
print(f"调整后余额: {adjustment['new_balance']}")
print(f"说明: {adjustment['message']}")

if name == "main":
demo_balance_audit()