"""Tax analysis Gradio interface components. This module provides UI components for tax impact analysis that can be integrated into the main Gradio application. """ from datetime import date from decimal import Decimal from typing import List, Dict, Any, Tuple, Optional import logging from dateutil import parser as date_parser from backend.tax.calculator import TaxCalculator from backend.tax.optimizer import TaxOptimizer from backend.tax.models import ( TaxFilingStatus, CostBasisMethod, ) logger = logging.getLogger(__name__) def create_tax_analysis( holdings: List[Dict[str, Any]], filing_status: str = "single", annual_income: float = 75000.0, cost_basis_method: str = "fifo", user_cost_basis: Optional[List[Dict[str, Any]]] = None, ) -> Dict[str, Any]: """Create comprehensive tax analysis for portfolio holdings. Args: holdings: List of portfolio holdings with ticker, quantity, market_value filing_status: Tax filing status (single, married_joint, married_separate, head_of_household) annual_income: Annual taxable income cost_basis_method: Cost basis method (fifo, lifo, hifo, average) user_cost_basis: Optional user-provided cost basis data with ticker, shares, purchase_price, purchase_date Returns: Dictionary containing tax analysis results """ try: # Convert filing status string to enum filing_status_map = { "single": TaxFilingStatus.SINGLE, "married_joint": TaxFilingStatus.MARRIED_JOINT, "married_separate": TaxFilingStatus.MARRIED_SEPARATE, "head_of_household": TaxFilingStatus.HEAD_OF_HOUSEHOLD, } filing_status_enum = filing_status_map.get(filing_status, TaxFilingStatus.SINGLE) # Convert cost basis method string to enum method_map = { "fifo": CostBasisMethod.FIFO, "lifo": CostBasisMethod.LIFO, "hifo": CostBasisMethod.HIFO, "average": CostBasisMethod.AVERAGE, } method_enum = method_map.get(cost_basis_method, CostBasisMethod.FIFO) # Initialize tax calculator tax_calc = TaxCalculator( filing_status=filing_status_enum, taxable_income=Decimal(str(annual_income)), tax_year=2024, ) # Convert user cost basis to dictionary for easy lookup user_basis_map = {} if user_cost_basis: for entry in user_cost_basis: ticker = entry.get("ticker", "").strip().upper() if ticker: user_basis_map[ticker] = entry # Track whether we used any user-provided data used_user_data = False # Create tax lots from holdings using user data or simulated data current_prices = {} for holding in holdings: ticker = holding.get("ticker", "") if not ticker: continue # Get current price current_price = holding.get("current_price", 0) or holding.get("market_value", 0) / max( holding.get("quantity", 1), 1 ) current_prices[ticker] = Decimal(str(current_price)) quantity = Decimal(str(holding.get("quantity", 0))) if quantity <= 0: continue # Use user-provided cost basis if available if ticker in user_basis_map: user_entry = user_basis_map[ticker] try: # Parse purchase price purchase_price = Decimal(str(user_entry.get("purchase_price", 0))) if purchase_price <= 0: raise ValueError("Invalid purchase price") # Parse purchase date date_str = user_entry.get("purchase_date", "") if isinstance(date_str, str) and date_str.strip(): acquisition_date = date_parser.parse(date_str).date() else: raise ValueError("Invalid purchase date") # Use user-specified shares if available user_shares = user_entry.get("shares") if user_shares is not None: quantity = Decimal(str(user_shares)) tax_calc.add_tax_lot( ticker=ticker, acquisition_date=acquisition_date, quantity=quantity, cost_basis_per_share=purchase_price, ) used_user_data = True except (ValueError, AttributeError) as e: logger.warning(f"Invalid user cost basis for {ticker}: {e}. Using simulated data.") # Fall back to simulated data sample_cost_basis = current_prices[ticker] * Decimal("0.8") acquisition_date = date.today().replace(year=date.today().year - 1) tax_calc.add_tax_lot( ticker=ticker, acquisition_date=acquisition_date, quantity=quantity, cost_basis_per_share=sample_cost_basis, ) else: # Use simulated data (20% below current price for demonstration) sample_cost_basis = current_prices[ticker] * Decimal("0.8") acquisition_date = date.today().replace(year=date.today().year - 1) tax_calc.add_tax_lot( ticker=ticker, acquisition_date=acquisition_date, quantity=quantity, cost_basis_per_share=sample_cost_basis, ) # Initialize tax optimizer tax_optimizer = TaxOptimizer(tax_calc) # Calculate unrealised gains/losses total_unrealised_gains = Decimal("0") total_unrealised_losses = Decimal("0") unrealised_by_ticker = {} for ticker, price in current_prices.items(): gain, qty = tax_calc.get_unrealised_gains(ticker, price) if gain > 0: total_unrealised_gains += gain elif gain < 0: total_unrealised_losses += abs(gain) unrealised_by_ticker[ticker] = { "quantity": float(qty), "unrealised_gain_loss": float(gain), "current_value": float(qty * price), } # Identify tax-loss harvesting opportunities harvesting_opportunities = tax_optimizer.identify_tax_loss_harvesting_opportunities( current_prices=current_prices, min_loss_threshold=Decimal("100"), min_loss_percentage=Decimal("0.05"), ) # Calculate potential tax rates lt_rate = tax_calc.get_capital_gains_rate(is_long_term=True) st_rate = tax_calc.get_capital_gains_rate(is_long_term=False) # Generate comprehensive report report = { "filing_status": filing_status, "annual_income": annual_income, "cost_basis_method": cost_basis_method, "used_user_data": used_user_data, "long_term_rate": float(lt_rate * 100), "short_term_rate": float(st_rate * 100), "total_unrealised_gains": float(total_unrealised_gains), "total_unrealised_losses": float(total_unrealised_losses), "net_unrealised": float(total_unrealised_gains - total_unrealised_losses), "unrealised_by_ticker": unrealised_by_ticker, "harvesting_opportunities": [ { "ticker": opp.ticker, "unrealised_loss": float(opp.unrealised_loss), "loss_percentage": float(opp.loss_percentage * 100), "quantity": float(opp.quantity), "is_long_term": opp.is_long_term, "estimated_tax_savings": float(opp.estimated_tax_savings), "recommendation": opp.recommended_action, } for opp in harvesting_opportunities ], } return report except Exception as e: logger.error(f"Error in tax analysis: {e}", exc_info=True) return { "error": str(e), "filing_status": filing_status, "annual_income": annual_income, "cost_basis_method": cost_basis_method, } def format_tax_analysis_output(report: Dict[str, Any]) -> str: """Format tax analysis report as markdown. Args: report: Tax analysis report dictionary Returns: Formatted markdown string """ if "error" in report: return f"""# Tax Impact Analysis **Error**: {report['error']} Please ensure you have run a portfolio analysis first. """ md = f"""# Tax Impact Analysis **IMPORTANT DISCLAIMER**: This analysis is for informational purposes only and does not constitute tax advice. Tax laws are complex and subject to change. Please consult a qualified tax professional before making any investment decisions based on this analysis. --- ### Tax Settings - **Filing Status**: {report['filing_status'].replace('_', ' ').title()} - **Annual Income**: ${report['annual_income']:,.2f} - **Cost Basis Method**: {report['cost_basis_method'].upper()} ### Your Tax Rates (2024-2025) - **Long-Term Capital Gains**: {report['long_term_rate']:.1f}% (holdings > 365 days) - **Short-Term Capital Gains**: {report['short_term_rate']:.1f}% (holdings ≤ 365 days) ### Current Position Summary - **Total Unrealised Gains**: ${report['total_unrealised_gains']:,.2f} - **Total Unrealised Losses**: ${report['total_unrealised_losses']:,.2f} - **Net Unrealised Position**: ${report['net_unrealised']:,.2f} """ # Estimated tax liability if all positions were sold today estimated_tax_on_gains = ( report['total_unrealised_gains'] * report['long_term_rate'] / 100 ) md += f""" ### Estimated Tax Liability If you were to sell all positions today (assuming long-term holdings): - **Estimated Tax**: ${estimated_tax_on_gains:,.2f} *Note: Actual tax may vary based on holding periods and loss offsets.* --- """ # Tax-loss harvesting opportunities opportunities = report.get('harvesting_opportunities', []) if opportunities: total_potential_savings = sum(opp['estimated_tax_savings'] for opp in opportunities) md += f"""### Tax-Loss Harvesting Opportunities **{len(opportunities)} opportunities identified** with potential tax savings of **${total_potential_savings:,.2f}** """ for i, opp in enumerate(opportunities[:5], 1): # Show top 5 term_type = "Long-term" if opp['is_long_term'] else "Short-term" md += f""" #### {i}. {opp['ticker']} - **Unrealised Loss**: ${abs(opp['unrealised_loss']):,.2f} ({abs(opp['loss_percentage']):.1f}%) - **Quantity**: {opp['quantity']:.2f} shares - **Type**: {term_type} - **Potential Tax Savings**: ${opp['estimated_tax_savings']:,.2f} **Recommendation**: {opp['recommendation']} """ if len(opportunities) > 5: md += f"\n*Plus {len(opportunities) - 5} additional opportunities...*\n" else: md += """### Tax-Loss Harvesting Opportunities No significant tax-loss harvesting opportunities identified at this time. """ # Cost basis method explanation md += f""" --- ### Cost Basis Method: {report['cost_basis_method'].upper()} """ method_explanations = { "fifo": """**First In, First Out (FIFO)**: Shares purchased first are sold first. This is the default method used by most brokerages.""", "lifo": """**Last In, First Out (LIFO)**: Most recently purchased shares are sold first. Can be useful for minimising gains if recent purchases were at higher prices.""", "hifo": """**Highest In, First Out (HIFO)**: Shares with highest cost basis are sold first. Optimal for minimising capital gains and current tax liability.""", "average": """**Average Cost**: Uses average cost of all shares. Commonly used for mutual funds and some ETFs.""", } md += method_explanations.get(report['cost_basis_method'], "") md += """ --- ### Key Tax Concepts **Long-Term vs Short-Term Capital Gains** - **Long-term**: Holdings sold after more than 365 days qualify for preferential rates (0%, 15%, or 20%) - **Short-term**: Holdings sold within 365 days are taxed as ordinary income (up to 37%) **Wash Sale Rule** - If you sell a security at a loss and buy substantially identical security within 30 days before or after (61-day window), the loss is disallowed - The disallowed loss is added to the cost basis of the replacement shares **Tax-Loss Harvesting** - Strategy of selling securities at a loss to offset capital gains - Can offset up to $3,000 of ordinary income per year - Excess losses can be carried forward to future years **Cost Basis Methods** - FIFO: Required for some securities, generally results in higher gains - LIFO: Can defer gains if recent purchases were at higher prices - HIFO: Optimal for tax minimisation but requires specific lot identification - Average: Simplest but less flexible for tax optimisation --- **Next Steps**: 1. Review your holding periods - consider waiting for long-term treatment if close to 365 days 2. If you have realised gains this year, consider harvesting losses to offset 3. Review wash sale implications before selling and repurchasing 4. Consider tax-efficient rebalancing strategies 5. Consult your tax advisor before implementing any strategy """ # Add disclaimer based on whether user data was used if report.get('used_user_data', False): md += """ --- **Data Source**: This analysis uses your provided purchase information for accurate tax calculations. """ else: md += """ --- **Data Source**: This analysis assumes current holdings were purchased at prices 20% below current market prices for demonstration purposes. For accurate analysis, use the 'Load Holdings' button and enter your actual purchase prices and dates. """ return md def compare_cost_basis_methods_output( holdings: List[Dict[str, Any]], filing_status: str, annual_income: float, ) -> str: """Compare tax implications across different cost basis methods. Args: holdings: List of portfolio holdings filing_status: Tax filing status annual_income: Annual taxable income Returns: Formatted markdown comparison """ methods = ["fifo", "lifo", "hifo", "average"] results = {} for method in methods: report = create_tax_analysis(holdings, filing_status, annual_income, method) if "error" not in report: results[method] = { "net_unrealised": report["net_unrealised"], "long_term_rate": report["long_term_rate"], "short_term_rate": report["short_term_rate"], } if not results: return "Unable to compare cost basis methods. Please run portfolio analysis first." md = """# Cost Basis Method Comparison Compare how different cost basis methods would affect your tax liability: | Method | Description | Typical Use Case | |--------|-------------|------------------| | **FIFO** | First In, First Out | Default method; sells oldest shares first | | **LIFO** | Last In, First Out | Sells newest shares first; may reduce gains | | **HIFO** | Highest In, First Out | Sells highest-cost shares first; minimises gains | | **Average** | Average Cost | Simple method for mutual funds/ETFs | **Note**: While different methods affect which shares are sold, your current unrealised position remains the same. The method matters when you actually sell shares. For this portfolio, all methods show similar unrealised positions since we're not executing any sales. To see the real impact of cost basis methods, you would need to: 1. Select specific shares to sell 2. Compare the resulting capital gains under each method 3. Calculate the tax liability for each scenario This feature requires transaction history data to provide meaningful comparisons. """ return md