"""年假折算算法 — 公司财年制 (4/1 — 次年 3/31) annual-leave proration. Two distinct quantities are produced for the same person on the same run, both shown to HR: - **应享额度 (entitlement)** — what the person would earn if they worked the FULL fiscal year. Segments span the entire 4/1 → 3/31. - **结算额度 (settlement)** — prorated to actual employed days (4/1 → termination_date, capped at fiscal year end). This is what HR subtracts used / adds carryover against to get the final balance. Both quantities pass through the same three-stage pipeline: 1. **Segmentation** — split the date range at every 社会工龄 N=10 / N=20 boundary AND every 司龄 S=k (k=1..8) boundary that falls inside the range. Each segment carries its own (legal_days, welfare_days) basis from the policy table. 2. **Per-segment proration** — segment_days / 365.25 × basis. Sum all segment values to get a raw total per leave type. 3. **Rounding + caps** — raw total is TRUNCATED to 2 decimals then rounded UP to the next 0.5 step. Then two caps run in order: first the per-type 基数 cap (a prorated result may never exceed the employee's own annual-leave basis), then the 15-day combined cap (if legal + welfare > 15 we hold legal and reduce welfare). Denominator note (2026-08-26, HR-confirmed). The denominator is the fixed average year 365.25, NOT the fiscal year's own length. It used to be a flat 365, which quietly overpaid every full-year employee in a fiscal year containing Feb 29: 366/365 × 10 = 10.0274 → 10.5 days for someone who simply worked the whole year. 365.25 alone does not fix that (366/365.25 × 10 = 10.0205, which still ceilings to 10.5) — the 基数 cap is what pulls it back to 10. The pair is load-bearing in both directions: 365.25 also keeps a NON-leap full year whole (365/365.25 × 10 = 9.9932 → 9.99 → 10.0), which a larger denominator would have shaved. The 基数 cap is taken as the MAXIMUM basis across the block's segments, per leave type. Not the closing segment's basis: an employee crossing 社会工龄 N=20 mid-year ends the year at welfare basis 0, and capping on that would retroactively confiscate the welfare days they had already earned in the earlier segments. Because a block's raw total can only exceed its max basis when the block spans more than 365.25 days, this cap is inert outside the leap-year case it exists for. The policy table (社会工龄 N × 司龄 S → days): N=[0,1) → legal 0, welfare 7 (only when S in [0,1)) N=[1,10) → legal 5, welfare = lookup by S (2..10) N=[10,20) → legal 10, welfare = lookup by S (0,0,0,0,1,2,3,4,5) N=[20,∞) → legal 15, welfare 0 (no welfare ever) Reference: company HR policy doc (uploaded screenshot 2026-05-14). """ from __future__ import annotations from dataclasses import dataclass from datetime import date, timedelta from decimal import ROUND_DOWN, ROUND_HALF_UP, Decimal from typing import Literal # Fiscal year window: April 1 — March 31 of the next calendar year. _FISCAL_YEAR_START_MONTH = 4 _FISCAL_YEAR_START_DAY = 1 #: Proration denominator — the fixed average year (365.25), HR-confirmed #: 2026-08-26. See the module docstring for why it is 365.25 and not the #: fiscal year's own 365/366 length. _PRORATION_DAYS_IN_YEAR = Decimal("365.25") #: Rendered into the per-segment formula strings HR reads back to #: employees, so it must track ``_PRORATION_DAYS_IN_YEAR`` exactly. _PRORATION_DENOMINATOR_LABEL = "365.25" #: Denominator for the tenure (工龄) year-fraction used to place segment #: boundaries. Deliberately still 365 and deliberately NOT the proration #: denominator: this one answers "has the N-th anniversary passed", which #: is a calendar question already pinned by ``_anniversary_on_or_after``. #: Moving it to 365.25 would nudge which side of a boundary a segment #: falls on — a different change, with no policy asking for it. _TENURE_DAYS_IN_YEAR = Decimal(365) _COMBINED_CAP_DAYS = Decimal("15.0") # ── Policy table ─────────────────────────────────────────────────── def _legal_days_for_n(years_of_social_tenure: Decimal) -> int: """Legal annual leave days given total social-tenure years N.""" if years_of_social_tenure < Decimal(1): return 0 if years_of_social_tenure < Decimal(10): return 5 if years_of_social_tenure < Decimal(20): return 10 return 15 def _welfare_days_for_n_s( years_of_social_tenure: Decimal, years_of_company_tenure: Decimal, ) -> int: """Welfare annual leave days given social tenure N and company tenure S. Mirrors the policy matrix. N ≥ 20 always returns 0 — the rule "一旦社会工龄跨入 N≥20 当日起:福利年假直接按 0 天执行" is enforced here, segments crossing this boundary correctly split at the crossing date. """ n = years_of_social_tenure s = years_of_company_tenure if n >= Decimal(20): return 0 if n < Decimal(1): # 0 < N < 1 row of the policy table only has a value for 0 < S < 1 # (which by definition matches when company tenure also hasn't # hit 1 yet). All other S buckets are unreachable: a person # with <1 year of total work history cannot also have ≥1 year # at the current company. return 7 if s < Decimal(1) else 0 if n < Decimal(10): # S bands: [0,1)=2 [1,2)=3 [2,3)=4 [3,4)=5 [4,5)=6 [5,6)=7 # [6,7)=8 [7,8)=9 [8,∞)=10 if s < Decimal(1): return 2 if s < Decimal(2): return 3 if s < Decimal(3): return 4 if s < Decimal(4): return 5 if s < Decimal(5): return 6 if s < Decimal(6): return 7 if s < Decimal(7): return 8 if s < Decimal(8): return 9 return 10 # 10 ≤ N < 20: legal jumps to 10, welfare resets and grows with S # past S=4. [0,4)=0 [4,5)=1 [5,6)=2 [6,7)=3 [7,8)=4 [8,∞)=5 if s < Decimal(4): return 0 if s < Decimal(5): return 1 if s < Decimal(6): return 2 if s < Decimal(7): return 3 if s < Decimal(8): return 4 return 5 # ── Date helpers ─────────────────────────────────────────────────── def fiscal_year_window(reference: date) -> tuple[date, date]: """Return (start, end) of the fiscal year containing ``reference``. Fiscal year runs Apr-1 → Mar-31 next year. ``reference`` may be inside the window or outside; the window returned is the one that contains it. """ if (reference.month, reference.day) >= ( _FISCAL_YEAR_START_MONTH, _FISCAL_YEAR_START_DAY, ): start_year = reference.year else: start_year = reference.year - 1 start = date(start_year, _FISCAL_YEAR_START_MONTH, _FISCAL_YEAR_START_DAY) end = date(start_year + 1, _FISCAL_YEAR_START_MONTH, _FISCAL_YEAR_START_DAY) - timedelta(days=1) return start, end def _years_between(start: date, end: date) -> Decimal: """Whole-day fractional years (end − start) / 365. Calendar-year ANSI semantics aren't used because the company rule speaks of "社会工龄满 N 年", which the calculator interprets as "N anniversaries have elapsed since the start date". A simple days/365 approximation suffices given inputs are days and the final answer rounds in 0.5-day steps. """ return Decimal((end - start).days) / _TENURE_DAYS_IN_YEAR def _anniversary_on_or_after(start: date, n: int, window_start: date) -> date | None: """Return the date when N years have elapsed since ``start``. Returns the calendar anniversary on/after ``window_start``, or ``None`` if the anniversary is before the window starts (already in effect) or absurdly far in the future. """ try: anniversary = date(start.year + n, start.month, start.day) except ValueError: # Leap-day birthday (Feb 29) when target year isn't a leap year — # roll back to Feb 28 so the anniversary always exists. anniversary = date(start.year + n, start.month, start.day - 1) if anniversary < window_start: return None return anniversary # ── Result shapes ────────────────────────────────────────────────── @dataclass(slots=True) class Segment: """One slice of the fiscal year over which (legal_basis, welfare_basis) is constant.""" index: int start_date: date end_date: date # inclusive days: int legal_basis: int # policy-table value at this segment welfare_basis: int legal_formula: str # e.g. "91/365×5" welfare_formula: str legal_raw_value: Decimal # segment's prorated contribution welfare_raw_value: Decimal def to_dict(self) -> dict: return { "index": self.index, "start_date": self.start_date.isoformat(), "end_date": self.end_date.isoformat(), "days": self.days, "legal_basis": self.legal_basis, "welfare_basis": self.welfare_basis, "legal_formula": self.legal_formula, "welfare_formula": self.welfare_formula, "legal_raw_value": str(self.legal_raw_value), "welfare_raw_value": str(self.welfare_raw_value), } @dataclass(slots=True) class Block: """A full computed block (entitlement OR settlement). ``raw`` = exact sum of segment proration values (4 decimal precision). ``rounded`` = after step-1 (2-decimal truncation) + step-2 (0.5 up). ``capped`` = after the 基数 cap AND the 15-day combined cap. The two caps carry separate flags because they mean different things to HR: ``basis_cap_applied`` says "this person's prorated result was pulled back to their own basis" (leap-year overshoot), ``cap_applied`` says "statutory + welfare together exceeded 15 so welfare was reduced". Folding them into one flag would put a wrong explanation in front of an employee disputing their settlement. """ label: Literal["entitlement", "settlement"] range_start: date range_end: date segments: list[Segment] legal_raw: Decimal welfare_raw: Decimal legal_rounded: Decimal welfare_rounded: Decimal legal_capped: Decimal welfare_capped: Decimal cap_applied: bool basis_cap_applied: bool @dataclass(slots=True) class CalculationResult: work_start_date: date hire_date: date termination_date: date fiscal_year_start: date fiscal_year_end: date entitlement: Block settlement: Block carry_legal_days: Decimal carry_welfare_days: Decimal used_legal_days: Decimal used_welfare_days: Decimal # 本期余额 — this fiscal year's own remaining balance: the prorated # settlement entitlement MINUS what's already been taken this year. # Deliberately excludes 上年结转 (carryover) so the UI reads # 本期余额 + 上年结转 = 总余额 (balance_*_days). # May go negative when used days exceed this period's accrual (the # excess having been drawn from carryover). period_balance_legal_days: Decimal period_balance_welfare_days: Decimal balance_legal_days: Decimal balance_welfare_days: Decimal # ── Rounding ─────────────────────────────────────────────────────── def truncate_to_2dp(value: Decimal) -> Decimal: """Step 1 — TRUNCATE to 2 decimal places (2026-08-26, was half-up). Truncating rather than rounding matters only in the narrow band just above a half-day step, because step 2 immediately ceilings to 0.5 anyway. ``10.5051`` is the whole difference: half-up made it ``10.51`` and then bought a full extra half-day (``11.0``), while truncation holds it at ``10.50`` → ``10.5``. In other words this step buys a 0.01 tolerance so an arithmetic crumb cannot cost the company half a day. Everywhere else the two agree. """ return value.quantize(Decimal("0.01"), rounding=ROUND_DOWN) def round_up_to_half_step(value: Decimal) -> Decimal: """Step 2 — round UP to the next 0.5 step. 0.14 → 0.5 0.50 → 0.5 0.52 → 1.0 1.00 → 1.0 1.01 → 1.5 """ if value <= Decimal(0): return Decimal("0.0") doubled = value * Decimal(2) # ceil int_part = int(doubled) if Decimal(int_part) < doubled: int_part += 1 return (Decimal(int_part) / Decimal(2)).quantize(Decimal("0.1")) def apply_rounding(raw: Decimal) -> Decimal: return round_up_to_half_step(truncate_to_2dp(raw)) def apply_basis_cap(value: Decimal, basis_cap: int) -> tuple[Decimal, bool]: """Clamp a rounded per-type total to the employee's own 年假基数. Returns ``(value, cap_applied)``. ``basis_cap`` is the MAXIMUM policy-table basis across the block's segments for this leave type — see the module docstring for why "maximum" and not "the basis at the end of the range". Runs BEFORE :func:`apply_cap`: this one only ever lowers a value, so the 15-day combined ceiling still holds afterwards. """ ceiling = Decimal(basis_cap).quantize(Decimal("0.1")) if value > ceiling: return ceiling, True return value, False def apply_cap(legal: Decimal, welfare: Decimal) -> tuple[Decimal, Decimal, bool]: """Apply the 15-day combined cap. Returns (legal, welfare_capped, cap_applied). Policy rule: ``legal + welfare ≤ 15`` total; statutory is kept and welfare is reduced when the sum exceeds the cap. Today legal maxes at 15 (N≥20) and welfare at 10 (N∈[1,10), S≥8) — under the current rule table the only triggering shape is "high legal + small welfare" where reducing welfare gets us back under 15 cleanly. Defensive guard for future rule changes: if a hypothetical future tier pushes legal ABOVE 15 on its own, we clamp legal to 15 first (and welfare to 0). Without this guard, ``_COMBINED_CAP_DAYS - legal`` would go negative and silently floor welfare at 0 — leaving legal at e.g. 18 days, "cap_applied=True" set, but the actual sum still violates the policy cap. The function's contract is "sum ≤ 15", so it must enforce that absolutely, not just for the welfare component. """ if legal >= _COMBINED_CAP_DAYS: if legal > _COMBINED_CAP_DAYS or welfare > Decimal(0): return ( _COMBINED_CAP_DAYS.quantize(Decimal("0.1")), Decimal("0.0"), True, ) return legal, welfare, False total = legal + welfare if total <= _COMBINED_CAP_DAYS: return legal, welfare, False new_welfare = (_COMBINED_CAP_DAYS - legal).quantize(Decimal("0.1")) return legal, new_welfare, True # ── Segmentation ─────────────────────────────────────────────────── def _segmentation_boundaries( work_start_date: date, hire_date: date, window_start: date, window_end: date, ) -> list[date]: """Collect all date breakpoints inside (window_start, window_end]. Breakpoints come from N anniversaries (10, 20) and S anniversaries (1..8). N=1 is also a breakpoint because the N<1 row of the policy table differs from N≥1. We do not split on S=0→1 explicitly when N<1, because that case is unreachable per the policy doc. """ breakpoints: set[date] = set() for n in (1, 10, 20): anni = _anniversary_on_or_after(work_start_date, n, window_start) if anni and window_start < anni <= window_end: breakpoints.add(anni) for s in range(1, 9): anni = _anniversary_on_or_after(hire_date, s, window_start) if anni and window_start < anni <= window_end: breakpoints.add(anni) return sorted(breakpoints) def _build_segments( work_start_date: date, hire_date: date, range_start: date, range_end: date, ) -> list[Segment]: """Split [range_start, range_end] at every policy breakpoint inside. Each segment carries the (legal_basis, welfare_basis) lookup as of its own ``start_date`` — boundaries are placed exactly so a segment is internally constant. """ if range_end < range_start: return [] boundaries = _segmentation_boundaries( work_start_date, hire_date, range_start, range_end, ) cuts = [range_start, *boundaries] # ensure ordered and unique; each cut starts a new segment segments: list[Segment] = [] for i, seg_start in enumerate(cuts): seg_end = (cuts[i + 1] - timedelta(days=1)) if i + 1 < len(cuts) else range_end if seg_end < seg_start: continue days = (seg_end - seg_start).days + 1 n_at_start = _years_between(work_start_date, seg_start) s_at_start = _years_between(hire_date, seg_start) legal = _legal_days_for_n(n_at_start) welfare = _welfare_days_for_n_s(n_at_start, s_at_start) legal_raw = (Decimal(days) / _PRORATION_DAYS_IN_YEAR) * Decimal(legal) welfare_raw = (Decimal(days) / _PRORATION_DAYS_IN_YEAR) * Decimal(welfare) segments.append(Segment( index=len(segments) + 1, start_date=seg_start, end_date=seg_end, days=days, legal_basis=legal, welfare_basis=welfare, legal_formula=f"{days}/{_PRORATION_DENOMINATOR_LABEL}×{legal}", welfare_formula=f"{days}/{_PRORATION_DENOMINATOR_LABEL}×{welfare}", legal_raw_value=legal_raw, welfare_raw_value=welfare_raw, )) return segments def _sum_block(segments: list[Segment]) -> tuple[Decimal, Decimal]: legal = sum((s.legal_raw_value for s in segments), Decimal(0)) welfare = sum((s.welfare_raw_value for s in segments), Decimal(0)) return legal, welfare def _build_block( label: Literal["entitlement", "settlement"], work_start_date: date, hire_date: date, range_start: date, range_end: date, ) -> Block: segments = _build_segments(work_start_date, hire_date, range_start, range_end) legal_raw, welfare_raw = _sum_block(segments) legal_rounded = apply_rounding(legal_raw) welfare_rounded = apply_rounding(welfare_raw) # 基数 cap first (only ever lowers), then the 15-day combined cap. # ``default=0`` covers a degenerate empty range — a settlement # window that closed before it opened has no basis to cap against, # and its totals are 0 anyway. # ``*_rounded`` deliberately keeps the PRE-cap value: the UI shows # 取整后 and 封顶后最终值 side by side, and collapsing them would hide # the very step HR has to explain to the employee. legal_basis_cap = max((s.legal_basis for s in segments), default=0) welfare_basis_cap = max((s.welfare_basis for s in segments), default=0) legal_after_basis, legal_basis_capped = apply_basis_cap( legal_rounded, legal_basis_cap, ) welfare_after_basis, welfare_basis_capped = apply_basis_cap( welfare_rounded, welfare_basis_cap, ) basis_cap_applied = legal_basis_capped or welfare_basis_capped legal_capped, welfare_capped, cap_applied = apply_cap( legal_after_basis, welfare_after_basis, ) return Block( label=label, range_start=range_start, range_end=range_end, segments=segments, legal_raw=legal_raw.quantize(Decimal("0.0001"), rounding=ROUND_HALF_UP), welfare_raw=welfare_raw.quantize(Decimal("0.0001"), rounding=ROUND_HALF_UP), legal_rounded=legal_rounded, welfare_rounded=welfare_rounded, legal_capped=legal_capped, welfare_capped=welfare_capped, cap_applied=cap_applied, basis_cap_applied=basis_cap_applied, ) # ── Public entry point ───────────────────────────────────────────── def calculate( *, work_start_date: date, hire_date: date, termination_date: date, carry_legal_days: Decimal, carry_welfare_days: Decimal, used_legal_days: Decimal, used_welfare_days: Decimal, ) -> CalculationResult: """Full 年假 calculation for one person. The fiscal year window is derived from ``termination_date`` — callers picking the calculator for an active employee should pass the projected exit date (or today). Inputs: - ``work_start_date`` — 参加工作日期 (社会工龄 起点) - ``hire_date`` — 入职日期 (司龄 起点) - ``termination_date``— 离职日期 - carry / used balances — HR-provided or from HRIS Returns both the full-year entitlement (assuming all-year service) and the prorated settlement (actual days employed within the fiscal year), plus the final unused-balance numbers. """ fy_start, fy_end = fiscal_year_window(termination_date) # Settlement window starts at fiscal year start (employee is # already-employed when fiscal year begins — onboarding mid-year is # an edge the rule doesn't yet special-case; HR can override # ``hire_date`` to the fiscal year start in that scenario). settlement_start = max(fy_start, hire_date) settlement_end = min(termination_date, fy_end) entitlement = _build_block("entitlement", work_start_date, hire_date, fy_start, fy_end) settlement = _build_block( "settlement", work_start_date, hire_date, settlement_start, settlement_end, ) period_balance_legal = ( settlement.legal_capped - used_legal_days ).quantize(Decimal("0.1")) period_balance_welfare = ( settlement.welfare_capped - used_welfare_days ).quantize(Decimal("0.1")) balance_legal = (period_balance_legal + carry_legal_days).quantize(Decimal("0.1")) balance_welfare = ( period_balance_welfare + carry_welfare_days ).quantize(Decimal("0.1")) return CalculationResult( work_start_date=work_start_date, hire_date=hire_date, termination_date=termination_date, fiscal_year_start=fy_start, fiscal_year_end=fy_end, entitlement=entitlement, settlement=settlement, carry_legal_days=carry_legal_days, carry_welfare_days=carry_welfare_days, used_legal_days=used_legal_days, used_welfare_days=used_welfare_days, period_balance_legal_days=period_balance_legal, period_balance_welfare_days=period_balance_welfare, balance_legal_days=balance_legal, balance_welfare_days=balance_welfare, )