How Can I Refactor A Python Module To Avoid Repeated Calculation Logic [closed]

I have a Python module with multiple pure functions for calculating tax breakdowns, deductions, and regime comparison. Right now, some helper functions are shared, but the flow still feels a bit repetitive between building opportunity cards and comparing regimes.

What would be a cleaner way to structure this code so the calculation logic stays reusable, testable, and avoids duplication?

"""Deterministic FY 2025-26 (AY 2026-27) Indian income tax engine.  
  
Reads backend/data/tax_rules_2025_26.json for every slab/cap/rate used below.  
No Firestore/auth access here - this module is a pure function of  
(rules JSON, TaxProfilePayload), mirroring app/services/scheme_services.py.  
"""  
  
import json  
from functools import lru_cache  
from pathlib import Path  
  
from app.models.tax import (  
    DeductionLine,  
    PresumptiveResult,  
    RegimeBreakdown,  
    RegimeComparisonResult,  
    TaxOpportunity,  
    TaxProfilePayload,  
    TaxSummaryResponse,  
)  
  
DATA_PATH = Path(__file__).resolve().parent.parent.parent / "data" / "tax_rules_2025_26.json"  
  
SALARIED_LIKE_OCCUPATIONS = {"salaried", "pensioner"}  
PRESUMPTIVE_OCCUPATIONS = {"business", "professional"}  
  
SECTION_DESCRIPTIONS = {  
    "80C": "Section 80C lets you cut your taxable income by investing in things like PPF, EPF, ELSS mutual "  
    "funds, life insurance premiums, or your children's tuition fees.",  
    "80CCD1B": "An extra deduction just for putting money into the National Pension System (NPS) yourself, "  
    "on top of the regular 80C limit.",  
    "80D_self": "Premiums you pay for health insurance covering yourself, your spouse, or your children can be "  
    "deducted from your taxable income.",  
    "80D_parents": "Premiums you pay for your parents' health insurance are deducted separately from your own "  
    "80D limit - so you get two deductions, not one.",  
    "24B": "Interest paid on a home loan for the house you live in reduces your taxable income under Section "  
    "24(b), up to the yearly cap.",  
    "80E": "All the interest you pay on an education loan (for yourself, spouse, or children) can be deducted "  
    "- there's no upper limit, only a time limit of 8 years.",  
    "HRA": "If your salary includes House Rent Allowance and you pay rent, part of that HRA is tax-free based "  
    "on your rent, salary, and city.",  
    "80GG": "If you pay rent but your salary doesn't include HRA, you can still claim a deduction for rent "  
    "paid under Section 80GG.",  
    "80G": "Donations to eligible charities can be partly deducted from your taxable income.",  
    "80TTA_TTB": "Interest earned on your savings bank account is tax-free up to this limit.",  
    "80U": "A flat deduction is available if you have a certified disability - the amount depends on how "  
    "severe it is, not on actual expenses.",  
    "80DD": "A flat deduction is available if you support a dependent family member with a certified "  
    "disability - claimed separately from your own 80U, so both can apply at once.",  
}  
  
  
@lru_cache(maxsize=1)  
def load_rules() -> dict:  
    with open(DATA_PATH, encoding="utf-8") as f:  
        return json.load(f)  
  
  
def slab_tax(taxable_income: float, slabs: list) -> float:  
    if taxable_income is None or taxable_income <= 0:  
        return 0.0  
    tax = 0.0  
    for lo, hi, rate in slabs:  
        if taxable_income <= lo:  
            break  
        upper = hi if hi is not None else taxable_income  
        amount_in_slab = min(taxable_income, upper) - lo  
        if amount_in_slab > 0:  
            tax += amount_in_slab * rate  
    return tax  
  
  
def get_age_band(age: int | None) -> str:  
    if age is None:  
        return "general"  
    if age >= 80:  
        return "super_senior_80plus"  
    if age >= 60:  
        return "senior_60_80"  
    return "general"  
  
  
def compute_old_regime_gross_tax(taxable_income: float, age: int | None, rules: dict) -> float:  
    slabs = rules["old_regime"]["slabs"][get_age_band(age)]  
    return slab_tax(taxable_income, slabs)  
  
  
def compute_new_regime_gross_tax(taxable_income: float, rules: dict) -> float:  
    slabs = rules["new_regime"]["slabs"]["all"]  
    return slab_tax(taxable_income, slabs)  
  
  
def apply_87a_rebate(gross_tax: float, taxable_income: float, regime: str, regime_rules: dict) -> float:  
    threshold = regime_rules["rebate_87a"]["threshold_taxable_income"]  
    max_rebate = regime_rules["rebate_87a"]["max_rebate"]  
  
    if taxable_income <= threshold:  
        return min(gross_tax, max_rebate)  
  
    # Marginal relief only exists in the statute for the new regime (added  
    # alongside its higher rebate threshold). The old regime's 87A rebate is  
    # a hard cliff - cross ₹5L by ₹1 and full tax is due, no smoothing.  
    if regime == "new":  
        excess = taxable_income - threshold  
        if gross_tax > excess:  
            return max(gross_tax - excess, 0.0)  
    return 0.0  
  
  
def compute_cess(tax_after_rebate_and_surcharge: float, rules: dict) -> float:  
    return tax_after_rebate_and_surcharge * rules["cess_rate"]  
  
  
def compute_surcharge(tax_after_rebate: float, taxable_income: float, regime: str, rules: dict, tax_fn) -> float:  
    slabs = rules["surcharge"]["slabs"]  
    applicable = None  
    for lo, _hi, rate in slabs:  
        if taxable_income > lo:  
            applicable = (lo, rate)  
  
    if applicable is None:  
        return 0.0  
  
    lo, rate = applicable  
    if regime == "new":  
        rate = min(rate, rules["surcharge"]["new_regime_max_rate"])  
  
    surcharge = tax_after_rebate * rate  
  
    # Marginal relief at the surcharge threshold: total tax+surcharge can't  
    # grow faster than the income that crossed the threshold.  
    tax_at_threshold = tax_fn(lo)  
    max_total = tax_at_threshold + (taxable_income - lo)  
    if tax_after_rebate + surcharge > max_total:  
        surcharge = max(max_total - tax_after_rebate, 0.0)  
  
    return surcharge  
  
  
def compute_final_tax(taxable_income: float, regime: str, age: int | None, rules: dict) -> dict:  
    if regime == "old":  
        gross = compute_old_regime_gross_tax(taxable_income, age, rules)  
        regime_rules = rules["old_regime"]  
  
        def tax_fn(inc):  
            return compute_old_regime_gross_tax(inc, age, rules)  
    else:  
        gross = compute_new_regime_gross_tax(taxable_income, rules)  
        regime_rules = rules["new_regime"]  
  
        def tax_fn(inc):  
            return compute_new_regime_gross_tax(inc, rules)  
  
    rebate = apply_87a_rebate(gross, taxable_income, regime, regime_rules)  
    tax_after_rebate = max(gross - rebate, 0.0)  
    surcharge = compute_surcharge(tax_after_rebate, taxable_income, regime, rules, tax_fn)  
    cess = compute_cess(tax_after_rebate + surcharge, rules)  
    total = tax_after_rebate + surcharge + cess  
  
    return {  
        "taxableIncome": taxable_income,  
        "grossTax": gross,  
        "rebate": rebate,  
        "surcharge": surcharge,  
        "cess": cess,  
        "totalTax": total,  
    }  
  
  
def compute_hra_exemption(basic_salary: float, hra_received: float, rent_paid: float, is_metro: bool, rules: dict) -> float:  
    if not basic_salary or not rent_paid or not hra_received:  
        return 0.0  
    cfg = rules["hra"]  
    pct = cfg["metro_pct_of_salary"] if is_metro else cfg["non_metro_pct_of_salary"]  
    option_hra = hra_received  
    option_rent = max(rent_paid - cfg["rent_minus_income_pct"] * basic_salary, 0.0)  
    option_pct_salary = pct * basic_salary  
    return max(min(option_hra, option_rent, option_pct_salary), 0.0)  
  
  
def compute_80gg_deduction(rent_paid: float, total_income: float, rules: dict) -> float:  
    if not rent_paid:  
        return 0.0  
    cfg = rules["sections"]["80GG"]  
    option_annual_cap = cfg["annual_cap"]  
    option_pct_income = cfg["pct_of_income"] * (total_income or 0)  
    option_rent = max(rent_paid - cfg["rent_minus_income_pct"] * (total_income or 0), 0.0)  
    return max(min(option_annual_cap, option_pct_income, option_rent), 0.0)  
  
  
def compute_presumptive_44ad(turnover: float, cash_pct: float, rules: dict) -> dict:  
    cfg = rules["presumptive_44ad"]  
    turnover = turnover or 0  
    cash_pct = cash_pct or 0  
    cash_amount = turnover * (cash_pct / 100)  
    digital_amount = turnover - cash_amount  
    high_digital = cash_pct <= cfg["cash_receipt_threshold_pct"] * 100  
    cap = cfg["turnover_cap_high_digital"] if high_digital else cfg["turnover_cap_normal"]  
    presumptive_income = cash_amount * cfg["cash_rate"] + digital_amount * cfg["digital_rate"]  
    return {  
        "eligible": turnover <= cap,  
        "cashAmount": cash_amount,  
        "digitalAmount": digital_amount,  
        "presumptiveIncome": presumptive_income,  
        "applicableCap": cap,  
    }  
  
  
def compute_presumptive_44ada(receipts: float, cash_pct: float, rules: dict) -> dict:  
    cfg = rules["presumptive_44ada"]  
    receipts = receipts or 0  
    cash_pct = cash_pct or 0  
    cash_amount = receipts * (cash_pct / 100)  
    digital_amount = receipts - cash_amount  
    high_digital = cash_pct <= cfg["cash_receipt_threshold_pct"] * 100  
    cap = cfg["receipts_cap_high_digital"] if high_digital else cfg["receipts_cap_normal"]  
    presumptive_income = receipts * cfg["presumptive_rate"]  
    return {  
        "eligible": receipts <= cap,  
        "cashAmount": cash_amount,  
        "digitalAmount": digital_amount,  
        "presumptiveIncome": presumptive_income,  
        "applicableCap": cap,  
    }  
  
  
def _presumptive_income(profile: TaxProfilePayload, rules: dict) -> float | None:  
    turnover = profile.turnoverOrReceipts or 0  
    cash_pct = profile.cashReceiptsPct or 0  
    if profile.occupation == "business":  
        result = compute_presumptive_44ad(turnover, cash_pct, rules)  
    elif profile.occupation == "professional":  
        result = compute_presumptive_44ada(turnover, cash_pct, rules)  
    else:  
        return None  
    return result["presumptiveIncome"] if result["eligible"] else None  
  
  
def _gross_total_income(profile: TaxProfilePayload, rules: dict) -> float:  
    if profile.occupation in PRESUMPTIVE_OCCUPATIONS and profile.optPresumptive:  
        presumptive_income = _presumptive_income(profile, rules)  
        if presumptive_income is not None:  
            return presumptive_income  
    return profile.annualIncome or 0  
  
  
def compute_deductions_breakdown(profile: TaxProfilePayload, rules: dict) -> list[dict]:  
    sections = rules["sections"]  
    lines: list[dict] = []  
  
    cap = sections["80C"]["cap"]  
    claimed = min(profile.investment80C or 0, cap)  
    lines.append({"section": "80C", "label": sections["80C"]["label"], "claimed": claimed, "cap": cap, "headroom": max(cap - claimed, 0)})  
  
    cap = sections["80CCD1B"]["cap"]  
    claimed = min(profile.nps80CCD1B or 0, cap)  
    lines.append({"section": "80CCD1B", "label": sections["80CCD1B"]["label"], "claimed": claimed, "cap": cap, "headroom": max(cap - claimed, 0)})  
  
    cap = sections["80D_self"]["senior_cap"] if profile.selfOrFamilySenior else sections["80D_self"]["cap"]  
    claimed = min(profile.insuranceSelf or 0, cap)  
    lines.append({"section": "80D_self", "label": sections["80D_self"]["label"], "claimed": claimed, "cap": cap, "headroom": max(cap - claimed, 0)})  
  
    cap = sections["80D_parents"]["senior_cap"] if profile.parentsSenior else sections["80D_parents"]["cap"]  
    claimed = min(profile.insuranceParents or 0, cap)  
    lines.append({"section": "80D_parents", "label": sections["80D_parents"]["label"], "claimed": claimed, "cap": cap, "headroom": max(cap - claimed, 0)})  
  
    cap = sections["24B"]["cap"]  
    claimed = min(profile.homeLoanInterest or 0, cap)  
    lines.append({"section": "24B", "label": sections["24B"]["label"], "claimed": claimed, "cap": cap, "headroom": max(cap - claimed, 0)})  
  
    claimed = profile.educationLoanInterest or 0  
    lines.append({"section": "80E", "label": sections["80E"]["label"], "claimed": claimed, "cap": None, "headroom": 0.0})  
  
    is_salaried = profile.occupation in SALARIED_LIKE_OCCUPATIONS  
    if is_salaried and profile.hasHRA and profile.rentPaid:  
        exemption = compute_hra_exemption(profile.basicSalary or 0, profile.hraReceived or 0, profile.rentPaid or 0, bool(profile.isMetro), rules)  
        lines.append({"section": "HRA", "label": "House Rent Allowance exemption", "claimed": exemption, "cap": exemption, "headroom": 0.0})  
    elif profile.rentPaid and not profile.hasHRA:  
        cap = sections["80GG"]["annual_cap"]  
        claimed = compute_80gg_deduction(profile.rentPaid or 0, profile.annualIncome or 0, rules)  
        lines.append({"section": "80GG", "label": sections["80GG"]["label"], "claimed": claimed, "cap": cap, "headroom": max(cap - claimed, 0)})  
  
    rate = sections["80G"]["deduction_rate"]  
    claimed = (profile.donation80G or 0) * rate  
    lines.append({"section": "80G", "label": sections["80G"]["label"], "claimed": claimed, "cap": None, "headroom": 0.0})  
  
    cap = sections["80TTA_TTB"]["senior_cap"] if get_age_band(profile.age) != "general" else sections["80TTA_TTB"]["cap"]  
    claimed = min(profile.savingsInterest or 0, cap)  
    lines.append({"section": "80TTA_TTB", "label": sections["80TTA_TTB"]["label"], "claimed": claimed, "cap": cap, "headroom": max(cap - claimed, 0)})  
  
    if profile.hasDisability:  
        cap = sections["80U"]["severe_cap"] if profile.disabilitySeverity == "severe" else sections["80U"]["normal_cap"]  
        lines.append({"section": "80U", "label": sections["80U"]["label"], "claimed": cap, "cap": cap, "headroom": 0.0})  
  
    if profile.supportsDisabledDependent:  
        cap = sections["80DD"]["severe_cap"] if profile.dependentDisabilitySeverity == "severe" else sections["80DD"]["normal_cap"]  
        lines.append({"section": "80DD", "label": sections["80DD"]["label"], "claimed": cap, "cap": cap, "headroom": 0.0})  
  
    return lines  
  
  
def compute_taxable_income_old_with_deductions(profile: TaxProfilePayload, rules: dict) -> tuple[float, list[dict]]:  
    gross_total_income = _gross_total_income(profile, rules)  
    is_salaried = profile.occupation in SALARIED_LIKE_OCCUPATIONS  
    standard_deduction = rules["old_regime"]["standard_deduction"] if is_salaried else 0  
    lines = compute_deductions_breakdown(profile, rules)  
    total_deductions = sum(line["claimed"] for line in lines)  
    taxable = max(gross_total_income - standard_deduction - total_deductions, 0.0)  
    return taxable, lines  
  
  
def compute_taxable_income_old_without_deductions(profile: TaxProfilePayload, rules: dict) -> float:  
    gross_total_income = _gross_total_income(profile, rules)  
    is_salaried = profile.occupation in SALARIED_LIKE_OCCUPATIONS  
    standard_deduction = rules["old_regime"]["standard_deduction"] if is_salaried else 0  
    return max(gross_total_income - standard_deduction, 0.0)  
  
  
def compute_taxable_income_new(profile: TaxProfilePayload, rules: dict) -> float:  
    gross_total_income = _gross_total_income(profile, rules)  
    is_salaried = profile.occupation in SALARIED_LIKE_OCCUPATIONS  
    standard_deduction = rules["new_regime"]["standard_deduction"] if is_salaried else 0  
    return max(gross_total_income - standard_deduction, 0.0)  
  
  
def build_opportunities(profile: TaxProfilePayload, rules: dict | None = None) -> list[TaxOpportunity]:  
    rules = rules or load_rules()  
    taxable_income, lines = compute_taxable_income_old_with_deductions(profile, rules)  
    current_tax = compute_final_tax(taxable_income, "old", profile.age, rules)["totalTax"]  
  
    opportunities = []  
    for line in lines:  
        headroom = line["headroom"] or 0.0  
        if headroom > 0:  
            taxable_if_maxed = max(taxable_income - headroom, 0.0)  
            tax_if_maxed = compute_final_tax(taxable_if_maxed, "old", profile.age, rules)["totalTax"]  
            benefit = max(current_tax - tax_if_maxed, 0.0)  
        else:  
            benefit = 0.0  
  
        opportunities.append(  
            TaxOpportunity(  
                section=line["section"],  
                title=line["label"],  
                description=SECTION_DESCRIPTIONS.get(line["section"], line["label"]),  
                cap=line["cap"],  
                claimed=line["claimed"],  
                headroom=headroom,  
                potentialBenefit=benefit,  
                regimeScope="old_regime_only",  
            )  
        )  
    return opportunities  
  
  
def compare_regimes(profile: TaxProfilePayload, rules: dict | None = None) -> RegimeComparisonResult:  
    rules = rules or load_rules()  
  
    taxable_old_with, lines = compute_taxable_income_old_with_deductions(profile, rules)  
    taxable_old_without = compute_taxable_income_old_without_deductions(profile, rules)  
    taxable_new = compute_taxable_income_new(profile, rules)  
  
    tax_old_with = compute_final_tax(taxable_old_with, "old", profile.age, rules)  
    tax_old_without = compute_final_tax(taxable_old_without, "old", profile.age, rules)  
    tax_new = compute_final_tax(taxable_new, "new", profile.age, rules)  
  
    old_with_deductions = RegimeBreakdown(  
        regime="old_with_deductions",  
        taxableIncome=taxable_old_with,  
        deductionsBreakdown=[DeductionLine(section=l["section"], label=l["label"], amount=l["claimed"]) for l in lines],  
        grossTax=tax_old_with["grossTax"],  
        rebate=tax_old_with["rebate"],  
        surcharge=tax_old_with["surcharge"],  
        cess=tax_old_with["cess"],  
        totalTax=tax_old_with["totalTax"],  
    )  
    old_without_deductions = RegimeBreakdown(  
        regime="old_without_deductions",  
        taxableIncome=taxable_old_without,  
        deductionsBreakdown=[],  
        grossTax=tax_old_without["grossTax"],  
        rebate=tax_old_without["rebate"],  
        surcharge=tax_old_without["surcharge"],  
        cess=tax_old_without["cess"],  
        totalTax=tax_old_without["totalTax"],  
    )  
    new_regime = RegimeBreakdown(  
        regime="new",  
        taxableIncome=taxable_new,  
        deductionsBreakdown=[],  
        grossTax=tax_new["grossTax"],  
        rebate=tax_new["rebate"],  
        surcharge=tax_new["surcharge"],  
        cess=tax_new["cess"],  
        totalTax=tax_new["totalTax"],  
    )  
  
    recommended = "old" if old_with_deductions.totalTax <= new_regime.totalTax else "new"  
    savings_amount = abs(old_with_deductions.totalTax - new_regime.totalTax)  
  
    return RegimeComparisonResult(  
        oldWithDeductions=old_with_deductions,  
        oldWithoutDeductions=old_without_deductions,  
        newRegime=new_regime,  
        recommended=recommended,  
        savingsAmount=savings_amount,  
    )  
  
  
def compute_full_summary(profile: TaxProfilePayload) -> TaxSummaryResponse:  
    rules = load_rules()  
    opportunities = build_opportunities(profile, rules)  
    total_potential_benefit = sum(o.potentialBenefit for o in opportunities)  
    regime_comparison = compare_regimes(profile, rules)  
    return TaxSummaryResponse(  
        opportunities=opportunities,  
        totalPotentialBenefit=total_potential_benefit,  
        regimeComparison=regime_comparison,  
    )  
  
  
def compute_presumptive_result(business_type: str, turnover_or_receipts: float, cash_receipts_pct: float) -> PresumptiveResult:  
    rules = load_rules()  
    if business_type == "business":  
        result = compute_presumptive_44ad(turnover_or_receipts, cash_receipts_pct, rules)  
    else:  
        result = compute_presumptive_44ada(turnover_or_receipts, cash_receipts_pct, rules)  
  
    presumptive_income = result["presumptiveIncome"] if result["eligible"] else 0.0  
    tax_breakdown = compute_final_tax(presumptive_income, "old", None, rules)  
  
    return PresumptiveResult(  
        eligible=result["eligible"],  
        reason=None if result["eligible"] else "Turnover/receipts exceed the presumptive scheme's threshold.",  
        cashAmount=result["cashAmount"],  
        digitalAmount=result["digitalAmount"],  
        presumptiveIncome=presumptive_income,  
        applicableCap=result["applicableCap"],  
        taxOnPresumptiveIncome=RegimeBreakdown(  
            regime="old",  
            taxableIncome=tax_breakdown["taxableIncome"],  
            deductionsBreakdown=[],  
            grossTax=tax_breakdown["grossTax"],  
            rebate=tax_breakdown["rebate"],  
            surcharge=tax_breakdown["surcharge"],  
            cess=tax_breakdown["cess"],  
            totalTax=tax_breakdown["totalTax"],  
        ),  
    )