311 lines
13 KiB
Python
311 lines
13 KiB
Python
"""
|
||
賞与計算サービス (Bonus Calculation Service)
|
||
"""
|
||
import json
|
||
import os
|
||
import functools
|
||
from decimal import Decimal, ROUND_HALF_UP, ROUND_DOWN
|
||
from datetime import date
|
||
from typing import Dict, Any
|
||
from ...core.database import get_connection
|
||
|
||
_BONUS_TAX_RATES_PATH = os.path.join(
|
||
os.path.dirname(os.path.dirname(__file__)),
|
||
'config', 'bonus_tax_rates.json'
|
||
)
|
||
|
||
|
||
@functools.lru_cache(maxsize=1)
|
||
def _load_bonus_tax_rates():
|
||
"""賞与算出率表をJSONファイルから読み込む（キャッシュ付き）"""
|
||
with open(_BONUS_TAX_RATES_PATH, encoding='utf-8') as f:
|
||
return json.load(f)
|
||
|
||
|
||
class BonusCalculationService:
|
||
"""賞与計算サービス"""
|
||
|
||
@staticmethod
|
||
def get_insurance_rate(rate_type: str, target_date: date, calculation_target: str = '賞与') -> Dict[str, Any]:
|
||
"""保険料率を取得（賞与用、年度がない場合は前年度を参照）"""
|
||
with get_connection() as conn:
|
||
with conn.cursor() as cur:
|
||
# 翌月払いのため、新年度料率は5月支払分（4月分）から適用
|
||
# rate_year=XXXX は XXXX年5月支払〜(XXXX+1)年4月支払に適用
|
||
# 1〜4月支払は前年度（year-1）の料率を使用
|
||
target_year = target_date.year if target_date.month >= 5 else target_date.year - 1
|
||
cur.execute(
|
||
"""
|
||
SELECT * FROM insurance_rates
|
||
WHERE rate_type = %s
|
||
AND rate_year = %s
|
||
AND calculation_target = %s
|
||
ORDER BY rate_id DESC
|
||
LIMIT 1
|
||
""",
|
||
(rate_type, target_year, calculation_target)
|
||
)
|
||
result = cur.fetchone()
|
||
|
||
# 現在の年度でデータが見つからない場合は前年度を検索（最大3年遡る）
|
||
if not result:
|
||
for year_offset in range(1, 4):
|
||
previous_year = target_year - year_offset
|
||
cur.execute(
|
||
"""
|
||
SELECT * FROM insurance_rates
|
||
WHERE rate_type = %s
|
||
AND rate_year = %s
|
||
AND calculation_target = %s
|
||
ORDER BY rate_id DESC
|
||
LIMIT 1
|
||
""",
|
||
(rate_type, previous_year, calculation_target)
|
||
)
|
||
result = cur.fetchone()
|
||
if result:
|
||
break
|
||
|
||
return result
|
||
|
||
@staticmethod
|
||
def get_insurance_limit(setting_type: str, target_date: date) -> Decimal:
|
||
"""社会保険料上限を取得"""
|
||
with get_connection() as conn:
|
||
with conn.cursor() as cur:
|
||
cur.execute(
|
||
"""
|
||
SELECT limit_amount FROM insurance_limit_settings
|
||
WHERE setting_type = %s
|
||
AND valid_from <= %s
|
||
AND (valid_to IS NULL OR valid_to >= %s)
|
||
ORDER BY valid_from DESC
|
||
LIMIT 1
|
||
""",
|
||
(setting_type, target_date, target_date)
|
||
)
|
||
result = cur.fetchone()
|
||
return Decimal(str(result["limit_amount"])) if result else Decimal("0")
|
||
|
||
@staticmethod
|
||
def get_previous_salary_avg(employee_id: int, payment_date: date) -> Decimal:
|
||
"""前月の社会保険料等控除後の給与を取得（賞与所得税計算用）
|
||
|
||
税法上は「前月分の社会保険料等控除後の給与」= 直前1ヶ月分のみを使用する。
|
||
ステータス問わず最新の月次給与レコードを参照する。
|
||
"""
|
||
with get_connection() as conn:
|
||
with conn.cursor() as cur:
|
||
# 賞与支払日より前の直近1ヶ月分の給与を取得（ステータス問わず）
|
||
cur.execute(
|
||
"""
|
||
SELECT total_payment - commute_allowance - health_insurance - care_insurance - pension_insurance - employment_insurance - COALESCE(child_support, 0) as net_salary
|
||
FROM monthly_payroll
|
||
WHERE employee_id = %s
|
||
AND payment_date < %s
|
||
ORDER BY payment_date DESC
|
||
LIMIT 1
|
||
""",
|
||
(employee_id, payment_date)
|
||
)
|
||
result = cur.fetchone()
|
||
return Decimal(str(result["net_salary"])) if result and result["net_salary"] else Decimal("0")
|
||
|
||
@staticmethod
|
||
def calculate_bonus_income_tax(
|
||
bonus_amount: Decimal,
|
||
previous_salary_avg: Decimal,
|
||
dependents_count: int
|
||
) -> Decimal:
|
||
"""
|
||
賞与所得税を計算
|
||
国税庁「賞与に対する源泉徴収税額の算出率の表」甲欄を使用
|
||
bonus_amount: 社会保険料等控除後の賞与額
|
||
previous_salary_avg: 前月の社会保険料等控除後の給与額
|
||
"""
|
||
if previous_salary_avg == Decimal("0"):
|
||
# 前月給与がない場合: 月額表方式が必要だがフォールバックとして10.21%
|
||
return (bonus_amount * Decimal("0.1021")).quantize(Decimal("1"), rounding=ROUND_HALF_UP)
|
||
|
||
table = _load_bonus_tax_rates()
|
||
salary_k = float(previous_salary_avg) / 1000 # 千円単位に変換
|
||
dep_idx = min(int(dependents_count), 7)
|
||
|
||
rate_percent = Decimal("10.21") # フォールバック
|
||
for entry in table['entries']:
|
||
rng = entry['ranges'][dep_idx]
|
||
from_k = rng[0]
|
||
to_k = rng[1]
|
||
if to_k is None:
|
||
# 上限なし（最高税率帯）
|
||
if salary_k >= from_k:
|
||
rate_percent = Decimal(str(entry['rate']))
|
||
break
|
||
elif from_k <= salary_k < to_k:
|
||
rate_percent = Decimal(str(entry['rate']))
|
||
break
|
||
|
||
rate = rate_percent / Decimal("100")
|
||
return (bonus_amount * rate).quantize(Decimal("1"), rounding=ROUND_HALF_UP)
|
||
|
||
@staticmethod
|
||
def calculate_bonus(
|
||
employee_id: int,
|
||
bonus_year: int,
|
||
bonus_month: int,
|
||
bonus_type: str,
|
||
payment_date: date,
|
||
base_bonus: Decimal,
|
||
performance_bonus: Decimal = Decimal("0"),
|
||
other_bonus: Decimal = Decimal("0"),
|
||
other_deduction: Decimal = Decimal("0"),
|
||
calculated_by: str = "system"
|
||
) -> Dict[str, Any]:
|
||
"""賞与を計算"""
|
||
|
||
# 総支給額
|
||
total_bonus = base_bonus + performance_bonus + other_bonus
|
||
|
||
# 社会保険料上限を取得（賞与専用の標準賞与額上限を使用）
|
||
health_insurance_limit = BonusCalculationService.get_insurance_limit(
|
||
"健康保険標準賞与額上限", payment_date
|
||
)
|
||
pension_insurance_limit = BonusCalculationService.get_insurance_limit(
|
||
"厚生年金標準賞与額上限", payment_date
|
||
)
|
||
|
||
# 賞与の標準賞与額（上限適用）
|
||
health_standard_bonus = min(total_bonus, health_insurance_limit) if health_insurance_limit > 0 else total_bonus
|
||
pension_standard_bonus = min(total_bonus, pension_insurance_limit) if pension_insurance_limit > 0 else total_bonus
|
||
|
||
# 従業員情報を取得（年齢計算のため）
|
||
with get_connection() as conn:
|
||
with conn.cursor() as cur:
|
||
cur.execute(
|
||
"SELECT birth_date FROM employees WHERE employee_id = %s",
|
||
(employee_id,)
|
||
)
|
||
employee_info = cur.fetchone()
|
||
|
||
# 年齢を計算（介護保険は40歳以上65歳未満が対象）
|
||
age = 0
|
||
if employee_info and employee_info["birth_date"]:
|
||
birth_date = employee_info["birth_date"]
|
||
# より正確な年齢計算
|
||
age = payment_date.year - birth_date.year
|
||
# 誕生日がまだ来ていない場合は-1
|
||
if (payment_date.month, payment_date.day) < (birth_date.month, birth_date.day):
|
||
age -= 1
|
||
|
||
# 健康保険（賞与額の1/1000で計算）
|
||
health_insurance_rate = BonusCalculationService.get_insurance_rate("健康保険", payment_date)
|
||
health_insurance = Decimal("0")
|
||
if health_insurance_rate:
|
||
health_insurance = (
|
||
health_standard_bonus * Decimal(str(health_insurance_rate["employee_rate"]))
|
||
).quantize(Decimal("1"), rounding=ROUND_DOWN)
|
||
|
||
# 介護保険（40歳以上65歳未満が対象）
|
||
care_insurance = Decimal("0")
|
||
if 40 <= age < 65:
|
||
care_insurance_rate = BonusCalculationService.get_insurance_rate("介護保険", payment_date)
|
||
if care_insurance_rate:
|
||
care_insurance = (
|
||
health_standard_bonus * Decimal(str(care_insurance_rate["employee_rate"]))
|
||
).quantize(Decimal("1"), rounding=ROUND_DOWN)
|
||
|
||
# 厚生年金
|
||
pension_insurance_rate = BonusCalculationService.get_insurance_rate("厚生年金", payment_date)
|
||
pension_insurance = Decimal("0")
|
||
if pension_insurance_rate:
|
||
pension_insurance = (
|
||
pension_standard_bonus * Decimal(str(pension_insurance_rate["employee_rate"]))
|
||
).quantize(Decimal("1"), rounding=ROUND_DOWN)
|
||
|
||
# 雇用保険
|
||
employment_insurance_rate = BonusCalculationService.get_insurance_rate("雇用保険", payment_date)
|
||
employment_insurance = Decimal("0")
|
||
if employment_insurance_rate:
|
||
employment_insurance = (
|
||
total_bonus * Decimal(str(employment_insurance_rate["employee_rate"]))
|
||
).quantize(Decimal("1"), rounding=ROUND_DOWN)
|
||
|
||
# 子ども・子育て支援金（令和8年5月支給分以降のみ適用）
|
||
# 賞与は bonus_month でなく実際の支給日（payment_date）で判定する
|
||
child_support = Decimal("0")
|
||
is_child_support_applicable = (
|
||
(payment_date.year == 2026 and payment_date.month >= 5) or payment_date.year >= 2027
|
||
)
|
||
if is_child_support_applicable:
|
||
child_support_rate = BonusCalculationService.get_insurance_rate("子ども・子育て支援金", payment_date)
|
||
if child_support_rate:
|
||
child_support = (
|
||
total_bonus * Decimal(str(child_support_rate["employee_rate"]))
|
||
).quantize(Decimal("1"), rounding=ROUND_DOWN)
|
||
print(f"[DEBUG] 賞与子育て支援金: 賞与総額={total_bonus} × 料率={child_support_rate['employee_rate']} = {child_support}")
|
||
|
||
# 所得税の計算（賞与特有の計算）
|
||
from ..calculation.service import PayrollCalculationService
|
||
dependents_count = PayrollCalculationService.get_active_dependents_count(employee_id, payment_date)
|
||
previous_salary_avg = BonusCalculationService.get_previous_salary_avg(employee_id, payment_date)
|
||
|
||
# 課税対象額
|
||
taxable_bonus = (
|
||
total_bonus -
|
||
health_insurance -
|
||
care_insurance -
|
||
pension_insurance -
|
||
employment_insurance -
|
||
child_support
|
||
)
|
||
|
||
income_tax = BonusCalculationService.calculate_bonus_income_tax(
|
||
taxable_bonus,
|
||
previous_salary_avg,
|
||
dependents_count
|
||
)
|
||
|
||
# 総控除額
|
||
total_deduction = (
|
||
health_insurance +
|
||
care_insurance +
|
||
pension_insurance +
|
||
employment_insurance +
|
||
child_support +
|
||
income_tax +
|
||
other_deduction
|
||
)
|
||
|
||
# 差引支給額
|
||
net_bonus = total_bonus - total_deduction
|
||
|
||
return {
|
||
"employee_id": employee_id,
|
||
"bonus_year": bonus_year,
|
||
"bonus_month": bonus_month,
|
||
"bonus_type": bonus_type,
|
||
"payment_date": payment_date,
|
||
"status": "calculated",
|
||
|
||
# 支給項目
|
||
"base_bonus": base_bonus,
|
||
"performance_bonus": performance_bonus,
|
||
"other_bonus": other_bonus,
|
||
"total_bonus": total_bonus,
|
||
|
||
# 控除項目
|
||
"health_insurance": health_insurance,
|
||
"care_insurance": care_insurance,
|
||
"pension_insurance": pension_insurance,
|
||
"employment_insurance": employment_insurance,
|
||
"child_support": child_support,
|
||
"income_tax": income_tax,
|
||
"other_deduction": other_deduction,
|
||
"total_deduction": total_deduction,
|
||
|
||
# 差引支給額
|
||
"net_bonus": net_bonus,
|
||
|
||
"calculated_by": calculated_by
|
||
}
|