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

Folder : shengchengqi

Files : 31

Size : 90.9 KB

Generated: 2026-03-20 17:12:36

shengchengqi/
├── README.md [175 B]
├── config/
│ ├── Buffer.properties [583 B]
│ ├── Observer.xml [1.6 KB]
│ ├── Parser.json [678 B]
│ └── application.properties [585 B]
├── datastore/
│ ├── Adapter.py [4.4 KB]
│ ├── Cache.py [4.3 KB]
│ ├── Listener.ts [3.9 KB]
│ ├── Processor.ts [3.1 KB]
│ ├── Scheduler.go [3.4 KB]
│ └── Util.php [3.3 KB]
├── generators/
│ ├── Helper.js [3 KB]
│ └── Manager.php [3.2 KB]
├── layouts/
│ ├── Executor.cpp [1.5 KB]
│ ├── Provider.go [2.5 KB]
│ ├── Proxy.js [3.1 KB]
│ └── Repository.java [4.3 KB]
├── lib/
│ └── Engine.jar [629 B]
├── package.json [678 B]
├── pom.xml [1.4 KB]
├── services/
│ ├── Queue.sql [2.3 KB]
│ ├── Service.py [4.9 KB]
│ └── Worker.py [4.7 KB]
├── src/
│ ├── main/
│ │ ├── java/
│ │ │ ├── Client.java [6.6 KB]
│ │ │ ├── Dispatcher.java [6.7 KB]
│ │ │ └── Validator.java [7 KB]
│ │ └── resources/
│ └── test/
│ └── java/
├── startup/
│ ├── Controller.cpp [1.6 KB]
│ └── Converter.py [4 KB]
└── view/
├── Loader.js [4.1 KB]
└── Server.ts [2.7 KB]

基础数据管理：员工信息、考勤数据、薪酬结构等

薪酬计算引擎：基本工资、加班费、奖金、扣款等计算逻辑

税务计算模块：个人所得税、社保公积金等

AI分析模块：薪酬预测、异常检测、优化建议

报表生成：工资条、统计报表、可视化图表

让我们一步步来实现这个系统。

二、基础数据模型设计
首先，我们需要设计合理的数据模型来存储员工信息和薪酬数据。

python
from datetime import datetime
from dataclasses import dataclass
from typing import List, Dict, Optional
import pandas as pd
import numpy as np

@dataclass
class Employee:
"""员工基础信息类"""
emp_id: str
name: str
department: str
position: str
base_salary: float # 基本工资
join_date: datetime
bank_account: str
tax_threshold: float = 5000.0 # 个税起征点

@dataclass
class Attendance:
"""考勤数据类"""
emp_id: str
year: int
month: int
work_days: int # 实际工作天数
leave_days: int # 请假天数
overtime_hours: float # 加班小时数
absent_days: int # 旷工天数

@dataclass
class SalaryDetail:
"""薪酬明细类"""
emp_id: str
year: int
month: int
base_salary: float
overtime_pay: float
bonus: float
allowance: float # 津贴
social_security: float # 社保扣款
housing_fund: float # 公积金扣款
personal_tax: float
other_deductions: float
net_salary: float # 实发工资
三、薪酬计算引擎实现
薪酬计算是系统的核心，我们需要实现各种复杂的计算逻辑。

python
class SalaryCalculator:
"""薪酬计算器"""

def __init__(self):
    # 社保公积金配置（示例数据）
    self.social_security_rate = {
        'pension': 0.08,  # 养老保险
        'medical': 0.02,  # 医疗保险
        'unemployment': 0.005,  # 失业保险
    }
    self.housing_fund_rate = 0.12  # 公积金比例

    # 加班费配置
    self.overtime_rate = {
        'weekday': 1.5,  # 工作日加班1.5倍
        'weekend': 2.0,  # 周末加班2倍
        'holiday': 3.0   # 法定节假日3倍
    }

def calculate_overtime_pay(self, base_salary: float, 
                            overtime_hours: float, 
                            work_days: int) -> float:
    """计算加班费"""
    daily_rate = base_salary / work_days
    hourly_rate = daily_rate / 8  # 按8小时工作制

    # 简化处理，实际应用中需要区分不同类型的加班
    overtime_pay = hourly_rate * overtime_hours * self.overtime_rate['weekday']
    return round(overtime_pay, 2)

def calculate_social_security(self, salary: float) -> float:
    """计算社保个人缴纳部分"""
    # 社保缴费基数上限和下限（示例）
    base_min, base_max = 4000, 30000
    base = max(min(salary, base_max), base_min)

    total = 0
    for _, rate in self.social_security_rate.items():
        total += base * rate
    return round(total, 2)

def calculate_housing_fund(self, salary: float) -> float:
    """计算公积金"""
    base = min(max(salary, 4000), 30000)
    return round(base * self.housing_fund_rate, 2)

def calculate_tax(self, taxable_income: float) -> float:
    """计算个人所得税（综合所得税率表）"""
    if taxable_income <= 0:
        return 0

    # 七级超额累进税率表
    tax_brackets = [
        (36000, 0.03, 0),
        (144000, 0.10, 2520),
        (300000, 0.20, 16920),
        (420000, 0.25, 31920),
        (660000, 0.30, 52920),
        (960000, 0.35, 85920),
        (float('inf'), 0.45, 181920)
    ]

    for limit, rate, quick_deduction in tax_brackets:
        if taxable_income <= limit:
            return round(taxable_income * rate - quick_deduction, 2)
    return 0

def calculate_salary(self, employee: Employee, 
                     attendance: Attendance, 
                     bonus: float = 0, 
                     allowance: float = 0,
                     other_deductions: float = 0) -> SalaryDetail:
    """计算完整薪酬"""

    # 1. 计算应发工资
    overtime_pay = self.calculate_overtime_pay(
        employee.base_salary, 
        attendance.overtime_hours, 
        attendance.work_days
    )

    gross_salary = (employee.base_salary + overtime_pay + 
                   bonus + allowance)

    # 2. 计算社保和公积金
    social_security = self.calculate_social_security(gross_salary)
    housing_fund = self.calculate_housing_fund(gross_salary)

    # 3. 计算应纳税所得额
    taxable_income = (gross_salary - social_security - 
                     housing_fund - employee.tax_threshold)

    # 4. 计算个人所得税
    personal_tax = self.calculate_tax(taxable_income)

    # 5. 计算实发工资
    net_salary = (gross_salary - social_security - 
                 housing_fund - personal_tax - other_deductions)

    return SalaryDetail(
        emp_id=employee.emp_id,
        year=attendance.year,
        month=attendance.month,
        base_salary=employee.base_salary,
        overtime_pay=overtime_pay,
        bonus=bonus,
        allowance=allowance,
        social_security=social_security,
        housing_fund=housing_fund,
        personal_tax=personal_tax,
        other_deductions=other_deductions,
        net_salary=net_salary
    )

四、AI智能分析模块
这是系统的亮点所在。我们将使用机器学习算法来实现智能化的薪酬分析。

4.1 薪酬预测模型
python
from sklearn.ensemble import RandomForestRegressor
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import LabelEncoder
import warnings
warnings.filterwarnings('ignore')

class SalaryPredictor:
"""薪酬预测器"""

def __init__(self):
    self.model = RandomForestRegressor(
        n_estimators=100,
        max_depth=10,
        random_state=42
    )
    self.label_encoders = {}

def prepare_features(self, df: pd.DataFrame) -> pd.DataFrame:
    """准备特征数据"""
    # 对类别特征进行编码
    categorical_features = ['department', 'position']
    for feature in categorical_features:
        if feature not in self.label_encoders:
            self.label_encoders[feature] = LabelEncoder()
            df[feature + '_encoded'] = self.label_encoders[feature].fit_transform(df[feature])
        else:
            df[feature + '_encoded'] = self.label_encoders[feature].transform(df[feature])
        df.drop(feature, axis=1, inplace=True)

    # 计算工作年限
    if 'join_date' in df.columns:
        df['join_date'] = pd.to_datetime(df['join_date'])
        current_date = pd.Timestamp.now()
        df['work_years'] = (current_date - df['join_date']).dt.days / 365.25
        df.drop('join_date', axis=1, inplace=True)

    return df

def train(self, historical_data: pd.DataFrame):
    """训练模型"""
    # 准备数据
    features = historical_data.drop(['emp_id', 'net_salary', 'month', 'year'], axis=1)
    features = self.prepare_features(features)
    target = historical_data['net_salary']

    # 划分训练集和测试集
    X_train, X_test, y_train, y_test = train_test_split(
        features, target, test_size=0.2, random_state=42
    )

    # 训练模型
    self.model.fit(X_train, y_train)

    # 评估模型
    train_score = self.model.score(X_train, y_train)
    test_score = self.model.score(X_test, y_test)

    print(f"模型训练完成 - R²分数:")
    print(f"训练集: {train_score:.4f}")
    print(f"测试集: {test_score:.4f}")

    return train_score, test_score

def predict(self, employee_features: pd.DataFrame) -> np.ndarray:
    """预测薪酬"""
    features = self.prepare_features(employee_features.copy())
    predictions = self.model.predict(features)
    return predictions

def get_feature_importance(self, feature_names: list) -> pd.DataFrame:
    """获取特征重要性"""
    importance_df = pd.DataFrame({
        'feature': feature_names,
        'importance': self.model.feature_importances_
    }).sort_values('importance', ascending=False)
    return importance_df

4.2 异常检测
python
from sklearn.ensemble import IsolationForest

class SalaryAnomalyDetector:
"""薪酬异常检测器"""

def __init__(self, contamination=0.1):
    self.detector = IsolationForest(
        contamination=contamination,
        random_state=42
    )

def detect_anomalies(self, salary_data: pd.DataFrame) -> pd.DataFrame:
    """检测异常薪酬数据"""
    # 选择用于检测的特征
    features = ['base_salary', 'overtime_pay', 'bonus', 
               'allowance', 'net_salary']

    # 标准化处理
    from sklearn.preprocessing import StandardScaler
    scaler = StandardScaler()
    scaled_data = scaler.fit_transform(salary_data[features])

    # 检测异常
    predictions = self.detector.fit_predict(scaled_data)

    # 标记异常
    salary_data['is_anomaly'] = predictions == -1
    salary_data['anomaly_score'] = self.detector.decision_function(scaled_data)

    return salary_data

def explain_anomaly(self, anomaly_row: pd.Series) -> str:
    """解释异常原因"""
    explanations = []

    # 检查各项指标是否超出正常范围
    metrics = ['base_salary', 'overtime_pay', 'bonus', 'allowance']

    for metric in metrics:
        if metric in anomaly_row.index:
            value = anomaly_row[metric]
            # 这里可以结合历史数据判断是否异常
            if value > anomaly_row[f'{metric}_mean'] + 2 * anomaly_row[f'{metric}_std']:
                explanations.append(f"{metric}显著偏高")
            elif value < anomaly_row[f'{metric}_mean'] - 2 * anomaly_row[f'{metric}_std']:
                explanations.append(f"{metric}显著偏低")

    if not explanations:
        explanations.append("综合指标异常")

    return "；".join(explanations)

五、完整的工资计算系统
现在，让我们将这些模块整合成一个完整的系统：

python
class SalaryManagementSystem:
"""智能薪酬管理系统"""

def __init__(self):
    self.employees = {}  # 员工信息库
    self.salary_history = []  # 薪酬历史记录
    self.calculator = SalaryCalculator()
    self.predictor = SalaryPredictor()
    self.anomaly_detector = SalaryAnomalyDetector()

def add_employee(self, employee: Employee):
    """添加员工"""
    self.employees[employee.emp_id] = employee
    print(f"员工 {employee.name} 添加成功")

def process_monthly_salary(self, attendance_list: List[Attendance],
                           bonuses: Dict[str, float] = None,
                           allowances: Dict[str, float] = None) -> List[SalaryDetail]:
    """处理月度工资"""
    results = []

    for attendance in attendance_list:
        emp_id = attendance.emp_id
        if emp_id not in self.employees:
            print(f"警告: 未找到员工 {emp_id}")
            continue

        employee = self.employees[emp_id]
        bonus = bonuses.get(emp_id, 0) if bonuses else 0
        allowance = allowances.get(emp_id, 0) if allowances else 0

        salary_detail = self.calculator.calculate_salary(
            employee, attendance, bonus, allowance
        )

        results.append(salary_detail)
        self.salary_history.append(salary_detail)

        # 打印工资条
        self.print_payslip(salary_detail, employee)

    return results

def print_payslip(self, salary_detail: SalaryDetail, employee: Employee):
    """打印工资条"""
    print("\n" + "="*50)
    print(f"工资条 - {employee.name} ({employee.emp_id})")
    print(f"{salary_detail.year}年{salary_detail.month}月")
    print("="*50)
    print(f"基本工资: ¥{salary_detail.base_salary:,.2f}")
    print(f"加班费: ¥{salary_detail.overtime_pay:,.2f}")
    print(f"奖金: ¥{salary_detail.bonus:,.2f}")
    print(f"津贴: ¥{salary_detail.allowance:,.2f}")
    print("-"*50)
    print(f"社保扣款: ¥{salary_detail.social_security:,.2f}")
    print(f"公积金扣款: ¥{salary_detail.housing_fund:,.2f}")
    print(f"个人所得税: ¥{salary_detail.personal_tax:,.2f}")
    if salary_detail.other_deductions > 0:
        print(f"其他扣款: ¥{salary_detail.other_deductions:,.2f}")
    print("-"*50)
    print(f"实发工资: ¥{salary_detail.net_salary:,.2f}")
    print("="*50 + "\n")

def analyze_salary_trends(self, department: str = None) -> pd.DataFrame:
    """分析薪酬趋势"""
    if not self.salary_history:
        print("暂无薪酬数据")
        return None

    # 转换为DataFrame
    df = pd.DataFrame([vars(s) for s in self.salary_history])

    # 合并员工信息
    emp_df = pd.DataFrame([vars(e) for e in self.employees.values()])
    df = df.merge(emp_df, on='emp_id')

    if department:
        df = df[df['department'] == department]

    # 生成统计报表
    stats = df.groupby(['year', 'month']).agg({
        'net_salary': ['mean', 'median', 'sum'],
        'base_salary': 'mean',
        'bonus': 'mean'
    }).round(2)

    return stats

def ai_recommendations(self) -> Dict:
    """AI优化建议"""
    if len(self.salary_history) < 10:
        return {"status": "数据不足，需要更多历史数据"}

    df = pd.DataFrame([vars(s) for s in self.salary_history])

    recommendations = {}

    # 1. 分析薪酬分布
    salary_stats = df['net_salary'].describe()
    recommendations['salary_distribution'] = {
        '平均工资': f"¥{salary_stats['mean']:,.2f}",
        '中位数': f"¥{salary_stats['50%']:,.2f}",
        '标准差': f"¥{salary_stats['std']:,.2f}"
    }

    # 2. 检查加班费占比
    df['overtime_ratio'] = df['overtime_pay'] / df['net_salary']
    avg_overtime_ratio = df['overtime_ratio'].mean()

    if avg_overtime_ratio > 0.3:
        recommendations['overtime_warning'] = "加班费占比较高，建议评估人员配置或招聘新员工"
    elif avg_overtime_ratio > 0.2:
        recommendations['overtime_suggestion'] = "加班情况较为普遍，建议关注员工工作负荷"

    # 3. 税务优化建议
    avg_tax_rate = (df['personal_tax'] / df['net_salary']).mean()
    if avg_tax_rate > 0.15:
        recommendations['tax_suggestion'] = "建议关注税务筹划，合理利用专项附加扣除"

    return recommendations

六、实战演示
让我们通过一个完整的示例来展示系统的使用：

python

创建系统实例

sms = SalaryManagementSystem()

添加员工

employee1 = Employee(
emp_id="EMP001",
name="张三",
department="技术部",
position="高级工程师",
base_salary=15000,
join_date=datetime(2020, 1, 1),
bank_account="6217**1234"
)

employee2 = Employee(
emp_id="EMP002",
name="李四",
department="市场部",
position="市场经理",
base_salary=12000,
join_date=datetime(2021, 3, 15),
bank_account="6217**5678"
)

sms.add_employee(employee1)
sms.add_employee(employee2)

准备考勤数据

attendance_list = [
Attendance(emp_id="EMP001", year=2024, month=1,
work_days=22, leave_days=0, overtime_hours=15, absent_days=0),
Attendance(emp_id="EMP002", year=2024, month=1,
work_days=20, leave_days=2, overtime_hours=5, absent_days=0)
]

设置奖金

bonuses = {"EMP001": 3000, "EMP002": 2000}
allowances = {"EMP001": 500, "EMP002": 300}

计算工资

results = sms.process_monthly_salary(attendance_list, bonuses, allowances)

分析薪酬趋势

print("\n薪酬统计分析:")
trends = sms.analyze_salary_trends()
if trends is not None:
print(trends)

AI优化建议

print("\nAI优化建议:")
recommendations = sms.ai_recommendations()
for key, value in recommendations.items():
print(f"{key}: {value}")

异常检测示例

print("\n异常检测:")
if len(results) > 0:
salary_df = pd.DataFrame([vars(r) for r in results])
anomaly_result = sms.anomaly_detector.detect_anomalies(salary_df)
anomalies = anomaly_result[anomaly_result['is_anomaly'] == True]
if len(anomalies) > 0:
print(f"发现 {len(anomalies)} 个异常记录")
else:
print("未发现异常记录")