""" Manual Tax Calculator - Rule-based tax calculations without LLM Implements the four core tax rules based on rules.py specifications """ import logging from typing import Any, Dict, Optional, Tuple from schemas import Receipt, Transaction logger = logging.getLogger(__name__) class ManualTaxCalculator: """ Deterministic tax calculator based on explicit rules from rules.py No LLM calls - pure business logic for accurate, consistent tax calculations """ # Provincial tax rates for Canada PROVINCIAL_TAX_RATES = { "ON": {"rate": 0.13, "name": "HST", "type": "Harmonized"}, "QC": {"rate": 0.14975, "name": "QST + GST", "type": "Combined"}, "BC": {"rate": 0.12, "name": "PST + GST", "type": "Combined"}, "AB": {"rate": 0.05, "name": "GST", "type": "Federal only"}, "SK": {"rate": 0.11, "name": "PST + GST", "type": "Combined"}, "MB": {"rate": 0.12, "name": "PST + GST", "type": "Combined"}, "NS": {"rate": 0.15, "name": "HST", "type": "Harmonized"}, "NB": {"rate": 0.15, "name": "HST", "type": "Harmonized"}, "NL": {"rate": 0.15, "name": "HST", "type": "Harmonized"}, "PE": {"rate": 0.15, "name": "HST", "type": "Harmonized"}, "NT": {"rate": 0.05, "name": "GST", "type": "Federal only"}, "NU": {"rate": 0.05, "name": "GST", "type": "Federal only"}, "YT": {"rate": 0.05, "name": "GST", "type": "Federal only"}, } # CCA rates by asset class (Canada Revenue Agency rates) CCA_RATES = { "vehicles": {"rate": 0.30, "class": "Class 10", "description": "Vehicles"}, "computer_equipment": { "rate": 0.55, "class": "Class 50", "description": "Computer Equipment", }, "furniture": { "rate": 0.20, "class": "Class 8", "description": "Furniture & Fixtures", }, "buildings": {"rate": 0.04, "class": "Class 1", "description": "Buildings"}, "machinery": { "rate": 0.20, "class": "Class 8", "description": "Machinery & Equipment", }, } # Capital asset threshold CAPITAL_ASSET_THRESHOLD = 500.00 # Meals & Entertainment categories MEALS_ENTERTAINMENT_KEYWORDS = [ "restaurant", "cafe", "coffee", "dining", "food", "meal", "catering", "entertainment", "bar", "pub", "bistro", "eatery", ] # Capital asset keywords CAPITAL_ASSET_KEYWORDS = { "vehicles": ["vehicle", "car", "truck", "van", "automobile", "suv"], "computer_equipment": [ "computer", "laptop", "desktop", "server", "tablet", "monitor", "printer", "scanner", ], "furniture": [ "furniture", "desk", "chair", "table", "cabinet", "bookshelf", "sofa", ], "buildings": ["building", "property", "real estate", "office space"], "machinery": ["machinery", "equipment", "tool", "industrial"], } def calculate_tax_analysis( self, receipt: Receipt, transaction: Transaction, user_location: str = "ON" ) -> Dict[str, Any]: """ Calculate comprehensive tax analysis for a receipt-transaction match Returns: Dict containing: - sales_tax: Sales tax calculation and validation - foreign_exchange: FX analysis and discrepancies - depreciation: Capital asset depreciation details - meals_entertainment: M&E deduction calculations - confidence_adjustment: Confidence boost/reduction """ analysis = {} # 1. Sales Tax Rule analysis["sales_tax"] = self._calculate_sales_tax( receipt, transaction, user_location ) # 2. Foreign Exchange Rule analysis["foreign_exchange"] = self._calculate_foreign_exchange( receipt, transaction ) # 3. Depreciation Rule analysis["depreciation"] = self._calculate_depreciation(receipt, user_location) # 4. Meals & Entertainment Rule analysis["meals_entertainment"] = self._calculate_meals_entertainment(receipt) # Calculate confidence adjustments analysis["confidence_adjustment"] = self._calculate_confidence_adjustment( analysis ) # Calculate final tax amount analysis["final_tax_amount"] = analysis["sales_tax"]["calculated_tax"] return analysis def _calculate_sales_tax( self, receipt: Receipt, transaction: Transaction, user_location: str ) -> Dict[str, Any]: """ Rule 1: Sales Tax Calculation - Priority: shipping address > billing address > user location - Different country: no Canadian tax - Missing location: default to user location """ # Determine the applicable location for tax receipt_location, location_source = self._determine_receipt_location( receipt, user_location ) # Check if international transaction is_international = self._is_international_transaction( receipt_location, user_location ) if is_international: return { "applicable_province": None, "applicable_rate": 0.0, "tax_name": "N/A", "calculated_tax": 0.0, "stated_tax": receipt.tax, "discrepancy": abs(receipt.tax - 0.0), "reason": f"International transaction - no Canadian tax applied. Receipt location: {receipt_location}", "requires_review": True, "location_source": location_source, "is_international": True, } # Get tax rate for the applicable province tax_info = self.PROVINCIAL_TAX_RATES.get( receipt_location, self.PROVINCIAL_TAX_RATES.get(user_location) ) # Calculate expected tax based on receipt amount # Tax should be calculated on pre-tax amount pre_tax_amount = receipt.amount - receipt.tax calculated_tax = round(pre_tax_amount * tax_info["rate"], 2) # Calculate discrepancy discrepancy = abs(receipt.tax - calculated_tax) discrepancy_percentage = ( (discrepancy / receipt.tax * 100) if receipt.tax > 0 else 0 ) # Determine if review is needed (>5% discrepancy) requires_review = discrepancy_percentage > 5.0 return { "applicable_province": receipt_location, "applicable_rate": tax_info["rate"], "tax_name": tax_info["name"], "calculated_tax": calculated_tax, "stated_tax": receipt.tax, "discrepancy": discrepancy, "discrepancy_percentage": round(discrepancy_percentage, 2), "reason": f"Tax calculated for {receipt_location} ({tax_info['name']}) - {location_source}", "requires_review": requires_review, "location_source": location_source, "is_international": False, } def _calculate_foreign_exchange( self, receipt: Receipt, transaction: Transaction ) -> Dict[str, Any]: """ Rule 2: Foreign Exchange Handling - Flag currency mismatches - Don't auto-fetch rates - Manual review required """ currency_mismatch = receipt.currency != transaction.currency if not currency_mismatch: return { "currency_mismatch": False, "receipt_currency": receipt.currency, "transaction_currency": transaction.currency, "requires_manual_review": False, "reason": "Currencies match - no FX adjustment needed", } # Calculate discrepancy discrepancy = abs(receipt.amount - transaction.amount) # Check if transaction has FX rate has_fx_rate = transaction.fx_rate is not None and transaction.fx_rate > 0 if has_fx_rate: expected_amount = round(receipt.amount * transaction.fx_rate, 2) calculated_discrepancy = abs(transaction.amount - expected_amount) else: expected_amount = None calculated_discrepancy = None return { "currency_mismatch": True, "receipt_currency": receipt.currency, "transaction_currency": transaction.currency, "receipt_amount": receipt.amount, "transaction_amount": transaction.amount, "discrepancy": discrepancy, "fx_rate": transaction.fx_rate, "expected_amount": expected_amount, "calculated_discrepancy": calculated_discrepancy, "requires_manual_review": True, "reason": f"Currency mismatch detected: {receipt.currency} → {transaction.currency}. Manual review required.", } def _calculate_depreciation( self, receipt: Receipt, user_location: str ) -> Dict[str, Any]: """ Rule 3: Depreciation Calculation - Always based on USER location (not receipt location) - Threshold: $500+ - Two methods: Straight-Line (accounting) and CCA (tax) """ # Check if this is a capital asset is_capital_asset = receipt.amount >= self.CAPITAL_ASSET_THRESHOLD asset_class = None cca_info = None if is_capital_asset: # Identify asset class from category and description asset_class = self._identify_asset_class(receipt) if asset_class: cca_info = self.CCA_RATES.get(asset_class) if not is_capital_asset or not asset_class: return { "is_capital_asset": False, "reason": f"Not a capital asset (Amount: ${receipt.amount:.2f}, Threshold: ${self.CAPITAL_ASSET_THRESHOLD:.2f})", } # Calculate straight-line depreciation (accounting) # Default: 5-year useful life, 10% residual value useful_life_years = 5 residual_percentage = 0.10 residual_value = receipt.amount * residual_percentage annual_straight_line = (receipt.amount - residual_value) / useful_life_years # Calculate CCA depreciation (tax - declining balance) cca_rate = cca_info["rate"] year1_cca = receipt.amount * cca_rate year2_cca = (receipt.amount - year1_cca) * cca_rate return { "is_capital_asset": True, "asset_class": asset_class, "cca_class": cca_info["class"], "cca_description": cca_info["description"], "asset_cost": receipt.amount, "user_location": user_location, "straight_line_depreciation": { "method": "Straight-Line (Accounting)", "useful_life_years": useful_life_years, "residual_value": round(residual_value, 2), "annual_depreciation": round(annual_straight_line, 2), }, "cca_depreciation": { "method": "CCA Declining Balance (Tax)", "cca_rate": cca_rate, "year_1_depreciation": round(year1_cca, 2), "year_2_depreciation": round(year2_cca, 2), }, "reason": f"Capital asset identified: {cca_info['description']} - Depreciation calculated based on user location ({user_location})", } def _calculate_meals_entertainment(self, receipt: Receipt) -> Dict[str, Any]: """ Rule 4: Meals & Entertainment Deductions - Tax: 50% of meal cost + 100% of sales tax - Accounting: 100% of meal cost + 100% of sales tax """ # Check if this is meals & entertainment is_meals_entertainment = self._is_meals_entertainment(receipt) if not is_meals_entertainment: return { "is_meals_entertainment": False, "reason": "Not classified as meals & entertainment", } # Calculate pre-tax meal amount meal_amount = receipt.amount - receipt.tax sales_tax = receipt.tax # Tax deduction: 50% of meal + 100% of tax tax_deduction = (meal_amount * 0.50) + sales_tax # Accounting deduction: 100% of meal + 100% of tax accounting_deduction = meal_amount + sales_tax return { "is_meals_entertainment": True, "meal_amount": round(meal_amount, 2), "sales_tax": round(sales_tax, 2), "total_receipt": round(receipt.amount, 2), "tax_deduction_amount": round(tax_deduction, 2), "tax_deduction_percentage": 50.0, "accounting_deduction_amount": round(accounting_deduction, 2), "accounting_deduction_percentage": 100.0, "reason": "Meals & Entertainment: 50% deductible for tax purposes, 100% for accounting", "breakdown": { "meal_cost": round(meal_amount, 2), "tax_50_percent": round(meal_amount * 0.50, 2), "full_sales_tax": round(sales_tax, 2), }, } def _calculate_confidence_adjustment( self, analysis: Dict[str, Any] ) -> Dict[str, float]: """ Calculate confidence boost/reduction based on tax analysis """ boost = 0.0 reduce = 0.0 # Sales tax analysis sales_tax = analysis.get("sales_tax", {}) if sales_tax.get("requires_review"): reduce += 0.05 else: # Small discrepancy is good discrepancy_pct = sales_tax.get("discrepancy_percentage", 0) if discrepancy_pct < 2.0: boost += 0.05 # Foreign exchange fx = analysis.get("foreign_exchange", {}) if fx.get("currency_mismatch"): reduce += 0.10 # FX always requires review # Depreciation - capital assets need review depreciation = analysis.get("depreciation", {}) if depreciation.get("is_capital_asset"): reduce += 0.05 return {"boost": round(boost, 2), "reduce": round(reduce, 2)} def _determine_receipt_location( self, receipt: Receipt, user_location: str ) -> Tuple[str, str]: """ Determine the applicable location for tax calculation Priority: shipping address > billing address > user location Returns: (province_code, source_description) """ # Check shipping address first if receipt.shipping_address: province = self._extract_province_from_address(receipt.shipping_address) if province: return province, "shipping address" # Check billing address if receipt.billing_address: province = self._extract_province_from_address(receipt.billing_address) if province: return province, "billing address" # Default to user location return user_location, "user location (default)" def _extract_province_from_address(self, address: str) -> Optional[str]: """ Extract Canadian province code from address string """ if not address: return None address_upper = address.upper() # Check for province codes for province_code in self.PROVINCIAL_TAX_RATES.keys(): if province_code in address_upper: return province_code # Check for full province names province_names = { "ONTARIO": "ON", "QUEBEC": "QC", "BRITISH COLUMBIA": "BC", "ALBERTA": "AB", "SASKATCHEWAN": "SK", "MANITOBA": "MB", "NOVA SCOTIA": "NS", "NEW BRUNSWICK": "NB", "NEWFOUNDLAND": "NL", "PRINCE EDWARD ISLAND": "PE", "NORTHWEST TERRITORIES": "NT", "NUNAVUT": "NU", "YUKON": "YT", } for full_name, code in province_names.items(): if full_name in address_upper: return code return None def _is_international_transaction( self, receipt_location: str, user_location: str ) -> bool: """ Check if this is an international transaction (receipt from outside Canada when user is in Canada, or vice versa) """ # If receipt location is not a Canadian province, it's international is_canadian = receipt_location in self.PROVINCIAL_TAX_RATES # For now, assume user_location is always Canadian # In future, add support for other countries return not is_canadian def _identify_asset_class(self, receipt: Receipt) -> Optional[str]: """ Identify the asset class from receipt category and description """ search_text = ( f"{receipt.category} {receipt.description} {receipt.vendor}".lower() ) for asset_class, keywords in self.CAPITAL_ASSET_KEYWORDS.items(): for keyword in keywords: if keyword in search_text: return asset_class return None def _is_meals_entertainment(self, receipt: Receipt) -> bool: """ Check if receipt is for meals & entertainment """ # Check explicit flag first if ( hasattr(receipt, "is_meals_entertainment") and receipt.is_meals_entertainment ): return True # Check category and description search_text = ( f"{receipt.category} {receipt.description} {receipt.vendor}".lower() ) for keyword in self.MEALS_ENTERTAINMENT_KEYWORDS: if keyword in search_text: return True return False def format_analysis_summary(self, analysis: Dict[str, Any]) -> str: """ Format the tax analysis into a human-readable summary """ lines = ["=== Tax Analysis Summary ===", ""] # Sales Tax st = analysis.get("sales_tax", {}) lines.append("1. SALES TAX:") if st.get("is_international"): lines.append(f" - {st['reason']}") lines.append(" - ⚠️ Review Required: International Transaction") else: lines.append(f" - Province: {st.get('applicable_province', 'N/A')}") lines.append( f" - Tax Rate: {st.get('applicable_rate', 0) * 100:.2f}% ({st.get('tax_name', 'N/A')})" ) lines.append(f" - Calculated Tax: ${st.get('calculated_tax', 0):.2f}") lines.append(f" - Stated Tax: ${st.get('stated_tax', 0):.2f}") lines.append( f" - Discrepancy: ${st.get('discrepancy', 0):.2f} ({st.get('discrepancy_percentage', 0):.1f}%)" ) if st.get("requires_review"): lines.append(" - ⚠️ Review Required: Tax discrepancy > 5%") lines.append("") # Foreign Exchange fx = analysis.get("foreign_exchange", {}) lines.append("2. FOREIGN EXCHANGE:") if fx.get("currency_mismatch"): lines.append( f" - Currency Mismatch: {fx['receipt_currency']} → {fx['transaction_currency']}" ) lines.append(f" - Receipt Amount: ${fx['receipt_amount']:.2f}") lines.append(f" - Transaction Amount: ${fx['transaction_amount']:.2f}") lines.append(f" - Discrepancy: ${fx['discrepancy']:.2f}") lines.append(" - ⚠️ Manual Review Required") else: lines.append(" - No currency mismatch") lines.append("") # Depreciation dep = analysis.get("depreciation", {}) lines.append("3. DEPRECIATION:") if dep.get("is_capital_asset"): lines.append(f" - Capital Asset: Yes ({dep['cca_description']})") lines.append(f" - Asset Cost: ${dep['asset_cost']:.2f}") lines.append( f" - CCA Class: {dep['cca_class']} ({dep['cca_depreciation']['cca_rate'] * 100:.0f}%)" ) lines.append( f" - Year 1 CCA: ${dep['cca_depreciation']['year_1_depreciation']:.2f}" ) lines.append( f" - Annual Straight-Line: ${dep['straight_line_depreciation']['annual_depreciation']:.2f}" ) else: lines.append(" - Not a capital asset") lines.append("") # Meals & Entertainment me = analysis.get("meals_entertainment", {}) lines.append("4. MEALS & ENTERTAINMENT:") if me.get("is_meals_entertainment"): lines.append(" - Type: Meals & Entertainment Expense") lines.append(f" - Meal Amount: ${me['meal_amount']:.2f}") lines.append(f" - Sales Tax: ${me['sales_tax']:.2f}") lines.append(f" - Tax Deduction (50%): ${me['tax_deduction_amount']:.2f}") lines.append( f" - Accounting Deduction (100%): ${me['accounting_deduction_amount']:.2f}" ) else: lines.append(" - Not a meals & entertainment expense") lines.append("") # Confidence Adjustment conf = analysis.get("confidence_adjustment", {}) lines.append("CONFIDENCE ADJUSTMENT:") lines.append(f" - Boost: +{conf.get('boost', 0):.2f}") lines.append(f" - Reduce: -{conf.get('reduce', 0):.2f}") return "\n".join(lines)