Add manual tax calculator for rule-based tax analysis and integrate with matching engine
This commit is contained in:
bolade
@@ -0,0 +1,583 @@
|
||||
"""
|
||||
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)
|
||||
+125
-11
@@ -5,14 +5,17 @@ from services.ai_matcher import AIMatcher
|
||||
from services.ai_rules import AIRulesEngine
|
||||
from services.feedback_logger import FeedbackLogger
|
||||
from services.llm_tax_analyzer import LLMTaxAnalyzer
|
||||
from services.manual_tax_calculator import ManualTaxCalculator
|
||||
|
||||
|
||||
class MatchingEngine:
|
||||
def __init__(self):
|
||||
def __init__(self, use_manual_tax_calculator: bool = False):
|
||||
self.ai_matcher = AIMatcher()
|
||||
self.rules_engine = AIRulesEngine()
|
||||
self.feedback_logger = FeedbackLogger()
|
||||
self.llm_tax_analyzer = LLMTaxAnalyzer()
|
||||
self.manual_tax_calculator = ManualTaxCalculator()
|
||||
self.use_manual_tax_calculator = use_manual_tax_calculator
|
||||
|
||||
def process_matching(
|
||||
self,
|
||||
@@ -42,19 +45,28 @@ class MatchingEngine:
|
||||
match.confidence_score = 1.0
|
||||
match.match_reason += " (Auto-approved by rules)"
|
||||
|
||||
# Now apply LLM-based tax analysis in a SINGLE batch call
|
||||
try:
|
||||
enhanced_matches = self.llm_tax_analyzer.analyze_and_apply_tax_rules_batch(
|
||||
# Apply tax analysis - use manual calculator or LLM based on configuration
|
||||
if self.use_manual_tax_calculator:
|
||||
# Use deterministic rule-based calculator
|
||||
enhanced_matches = self._apply_manual_tax_analysis(
|
||||
ai_matches, user_location
|
||||
)
|
||||
except Exception as e:
|
||||
# If batch LLM analysis fails, log it and continue with matches as-is
|
||||
import logging
|
||||
else:
|
||||
# Use LLM-based tax analysis in a SINGLE batch call
|
||||
try:
|
||||
enhanced_matches = (
|
||||
self.llm_tax_analyzer.analyze_and_apply_tax_rules_batch(
|
||||
ai_matches, user_location
|
||||
)
|
||||
)
|
||||
except Exception as e:
|
||||
# If batch LLM analysis fails, log it and continue with matches as-is
|
||||
import logging
|
||||
|
||||
logging.error(f"Batch LLM tax analysis failed: {str(e)}")
|
||||
for match in ai_matches:
|
||||
match.match_reason += " (Note: Advanced tax analysis unavailable)"
|
||||
enhanced_matches = ai_matches
|
||||
logging.error(f"Batch LLM tax analysis failed: {str(e)}")
|
||||
for match in ai_matches:
|
||||
match.match_reason += " (Note: Advanced tax analysis unavailable)"
|
||||
enhanced_matches = ai_matches
|
||||
|
||||
return enhanced_matches
|
||||
|
||||
@@ -155,6 +167,108 @@ class MatchingEngine:
|
||||
|
||||
return match
|
||||
|
||||
def _apply_manual_tax_analysis(
|
||||
self, matches: List[Match], user_location: str = "ON"
|
||||
) -> List[Match]:
|
||||
"""
|
||||
Apply deterministic rule-based tax analysis to all matches
|
||||
No LLM calls - pure business logic for consistent results
|
||||
"""
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
logger.info(
|
||||
f"Applying manual tax analysis to {len(matches)} matches using rule-based calculator"
|
||||
)
|
||||
|
||||
enhanced_matches = []
|
||||
|
||||
for match in matches:
|
||||
try:
|
||||
# Get comprehensive tax analysis from manual calculator
|
||||
tax_analysis = self.manual_tax_calculator.calculate_tax_analysis(
|
||||
match.receipt, match.transaction, user_location
|
||||
)
|
||||
|
||||
# Store the complete tax analysis
|
||||
match.tax_analysis = tax_analysis
|
||||
|
||||
# Apply confidence adjustments
|
||||
confidence_adj = tax_analysis.get("confidence_adjustment", {})
|
||||
|
||||
# Boost confidence if tax rules validate the match
|
||||
boost = confidence_adj.get("boost", 0.0)
|
||||
if boost > 0:
|
||||
match.confidence_score = min(1.0, match.confidence_score + boost)
|
||||
match.match_reason += f" (Tax validated: +{boost:.2f})"
|
||||
|
||||
# Reduce confidence if tax issues detected
|
||||
reduce = confidence_adj.get("reduce", 0.0)
|
||||
if reduce > 0:
|
||||
match.confidence_score = max(0.0, match.confidence_score - reduce)
|
||||
match.match_reason += f" (Tax issues: -{reduce:.2f})"
|
||||
|
||||
# Add flags for manual review
|
||||
review_flags = []
|
||||
|
||||
# Sales tax issues
|
||||
sales_tax = tax_analysis.get("sales_tax", {})
|
||||
if sales_tax.get("requires_review"):
|
||||
if sales_tax.get("is_international"):
|
||||
review_flags.append("International Transaction - FX Review")
|
||||
else:
|
||||
discrepancy_pct = sales_tax.get("discrepancy_percentage", 0)
|
||||
review_flags.append(
|
||||
f"Sales Tax Discrepancy: {discrepancy_pct:.1f}%"
|
||||
)
|
||||
|
||||
# FX issues
|
||||
fx = tax_analysis.get("foreign_exchange", {})
|
||||
if fx.get("currency_mismatch"):
|
||||
review_flags.append(
|
||||
f"FX: {fx['receipt_currency']} → {fx['transaction_currency']} (${fx['discrepancy']:.2f})"
|
||||
)
|
||||
|
||||
# Capital asset depreciation
|
||||
depreciation = tax_analysis.get("depreciation", {})
|
||||
if depreciation.get("is_capital_asset"):
|
||||
cca_class = depreciation.get("cca_class", "Unknown")
|
||||
year1_cca = depreciation.get("cca_depreciation", {}).get(
|
||||
"year_1_depreciation", 0
|
||||
)
|
||||
review_flags.append(
|
||||
f"Capital Asset ({cca_class}) - Year 1 CCA: ${year1_cca:.2f}"
|
||||
)
|
||||
|
||||
# Meals & entertainment
|
||||
meals_ent = tax_analysis.get("meals_entertainment", {})
|
||||
if meals_ent.get("is_meals_entertainment"):
|
||||
tax_deduction = meals_ent.get("tax_deduction_amount", 0)
|
||||
accounting_deduction = meals_ent.get(
|
||||
"accounting_deduction_amount", 0
|
||||
)
|
||||
review_flags.append(
|
||||
f"M&E: Tax ${tax_deduction:.2f} (50%), Accounting ${accounting_deduction:.2f} (100%)"
|
||||
)
|
||||
|
||||
# Add review flags to match reason
|
||||
if review_flags:
|
||||
match.match_reason += " | " + "; ".join(review_flags)
|
||||
|
||||
enhanced_matches.append(match)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
f"Manual tax analysis failed for match: {str(e)}", exc_info=True
|
||||
)
|
||||
match.match_reason += " (Tax analysis failed)"
|
||||
enhanced_matches.append(match)
|
||||
|
||||
logger.info(
|
||||
f"Manual tax analysis completed for {len(enhanced_matches)} matches"
|
||||
)
|
||||
return enhanced_matches
|
||||
|
||||
def approve_match(self, match: Match, user_id: str):
|
||||
# Log the approval
|
||||
self.feedback_logger.log_override(
|
||||
|
||||
Reference in New Issue
Block a user