diff --git a/tests/unit/portfolio/test_tax_calculator.py b/tests/unit/portfolio/test_tax_calculator.py new file mode 100644 index 00000000..49774581 --- /dev/null +++ b/tests/unit/portfolio/test_tax_calculator.py @@ -0,0 +1,1214 @@ +"""Tests for Australian CGT Calculator. + +Issue #32: [PORT-31] Australian CGT calculator - 50% discount, tax reports + +Tests cover: +- CGT enums and dataclasses +- Acquisition and disposal recording +- FIFO cost basis matching +- 50% CGT discount for holdings >12 months +- Tax year calculations (July-June) +- Capital loss tracking and carry-forward +- Tax report generation +- Foreign currency conversions +- Edge cases and validation +""" + +import pytest +from datetime import date, timedelta +from decimal import Decimal + +from tradingagents.portfolio.tax_calculator import ( + CGTMethod, + AssetType, + CGTAssetAcquisition, + CGTDisposal, + CGTEvent, + TaxYearSummary, + AustralianCGTCalculator, +) + + +# ============================================================================== +# CGTMethod Enum Tests +# ============================================================================== + + +class TestCGTMethod: + """Tests for CGTMethod enum.""" + + def test_discount_value(self): + """Test DISCOUNT method value.""" + assert CGTMethod.DISCOUNT.value == "discount" + + def test_indexation_value(self): + """Test INDEXATION method value.""" + assert CGTMethod.INDEXATION.value == "indexation" + + def test_other_value(self): + """Test OTHER method value.""" + assert CGTMethod.OTHER.value == "other" + + def test_all_methods_exist(self): + """Test all expected methods exist.""" + methods = [m for m in CGTMethod] + assert len(methods) == 3 + assert CGTMethod.DISCOUNT in methods + assert CGTMethod.INDEXATION in methods + assert CGTMethod.OTHER in methods + + +# ============================================================================== +# AssetType Enum Tests +# ============================================================================== + + +class TestAssetType: + """Tests for AssetType enum.""" + + def test_shares_value(self): + """Test SHARES type value.""" + assert AssetType.SHARES.value == "shares" + + def test_foreign_shares_value(self): + """Test FOREIGN_SHARES type value.""" + assert AssetType.FOREIGN_SHARES.value == "foreign_shares" + + def test_etf_value(self): + """Test ETF type value.""" + assert AssetType.ETF.value == "etf" + + def test_cryptocurrency_value(self): + """Test CRYPTOCURRENCY type value.""" + assert AssetType.CRYPTOCURRENCY.value == "cryptocurrency" + + def test_property_value(self): + """Test PROPERTY type value.""" + assert AssetType.PROPERTY.value == "property" + + def test_collectables_value(self): + """Test COLLECTABLES type value.""" + assert AssetType.COLLECTABLES.value == "collectables" + + def test_other_value(self): + """Test OTHER type value.""" + assert AssetType.OTHER.value == "other" + + def test_all_types_exist(self): + """Test all expected asset types exist.""" + types = [t for t in AssetType] + assert len(types) == 7 + + +# ============================================================================== +# CGTAssetAcquisition Tests +# ============================================================================== + + +class TestCGTAssetAcquisition: + """Tests for CGTAssetAcquisition dataclass.""" + + def test_create_basic_acquisition(self): + """Test creating a basic AUD acquisition.""" + acq = CGTAssetAcquisition( + acquisition_date=date(2023, 1, 15), + quantity=Decimal("100"), + cost_per_unit=Decimal("95.50"), + total_cost_aud=Decimal("9550"), + ) + assert acq.acquisition_date == date(2023, 1, 15) + assert acq.quantity == Decimal("100") + assert acq.cost_per_unit == Decimal("95.50") + assert acq.total_cost_aud == Decimal("9550") + assert acq.currency == "AUD" + assert acq.exchange_rate is None + assert acq.incidental_costs == Decimal("0") + + def test_acquisition_with_incidental_costs(self): + """Test acquisition with brokerage fees.""" + acq = CGTAssetAcquisition( + acquisition_date=date(2023, 1, 15), + quantity=Decimal("100"), + cost_per_unit=Decimal("95.50"), + total_cost_aud=Decimal("9550"), + incidental_costs=Decimal("19.95"), + ) + assert acq.incidental_costs == Decimal("19.95") + assert acq.total_cost_base == Decimal("9569.95") + + def test_cost_base_per_unit(self): + """Test cost base per unit calculation.""" + acq = CGTAssetAcquisition( + acquisition_date=date(2023, 1, 15), + quantity=Decimal("100"), + cost_per_unit=Decimal("95.50"), + total_cost_aud=Decimal("9550"), + incidental_costs=Decimal("50"), + ) + # (9550 + 50) / 100 = 96.00 + assert acq.cost_base_per_unit == Decimal("96") + + def test_cost_base_per_unit_zero_quantity(self): + """Test cost base per unit with zero quantity.""" + acq = CGTAssetAcquisition( + acquisition_date=date(2023, 1, 15), + quantity=Decimal("0"), + cost_per_unit=Decimal("95.50"), + total_cost_aud=Decimal("0"), + ) + assert acq.cost_base_per_unit == Decimal("0") + + def test_foreign_acquisition(self): + """Test foreign currency acquisition.""" + acq = CGTAssetAcquisition( + acquisition_date=date(2023, 1, 15), + quantity=Decimal("50"), + cost_per_unit=Decimal("150"), # USD + total_cost_aud=Decimal("11250"), # After AUD conversion + currency="USD", + exchange_rate=Decimal("1.50"), + ) + assert acq.currency == "USD" + assert acq.exchange_rate == Decimal("1.50") + assert acq.total_cost_aud == Decimal("11250") + + +# ============================================================================== +# CGTDisposal Tests +# ============================================================================== + + +class TestCGTDisposal: + """Tests for CGTDisposal dataclass.""" + + def test_create_basic_disposal(self): + """Test creating a basic disposal.""" + disposal = CGTDisposal( + disposal_date=date(2024, 3, 20), + symbol="CBA.AX", + quantity=Decimal("50"), + proceeds_per_unit=Decimal("110.25"), + total_proceeds_aud=Decimal("5512.50"), + ) + assert disposal.disposal_date == date(2024, 3, 20) + assert disposal.symbol == "CBA.AX" + assert disposal.quantity == Decimal("50") + assert disposal.total_proceeds_aud == Decimal("5512.50") + + def test_net_proceeds_calculation(self): + """Test net proceeds after selling costs.""" + disposal = CGTDisposal( + disposal_date=date(2024, 3, 20), + symbol="CBA.AX", + quantity=Decimal("50"), + proceeds_per_unit=Decimal("110.25"), + total_proceeds_aud=Decimal("5512.50"), + incidental_costs=Decimal("19.95"), + ) + assert disposal.net_proceeds == Decimal("5492.55") + + +# ============================================================================== +# CGTEvent Tests +# ============================================================================== + + +class TestCGTEvent: + """Tests for CGTEvent dataclass.""" + + def test_is_gain(self): + """Test capital gain detection.""" + disposal = CGTDisposal( + disposal_date=date(2024, 3, 20), + symbol="CBA.AX", + quantity=Decimal("50"), + proceeds_per_unit=Decimal("110.25"), + total_proceeds_aud=Decimal("5512.50"), + ) + event = CGTEvent( + event_date=date(2024, 3, 20), + symbol="CBA.AX", + asset_type=AssetType.SHARES, + disposal=disposal, + gross_gain=Decimal("737.50"), + discount_eligible=True, + discount_amount=Decimal("368.75"), + net_gain=Decimal("368.75"), + holding_period_days=430, + cgt_method=CGTMethod.DISCOUNT, + tax_year=2024, + ) + assert event.is_gain is True + assert event.is_loss is False + + def test_is_loss(self): + """Test capital loss detection.""" + disposal = CGTDisposal( + disposal_date=date(2024, 3, 20), + symbol="CBA.AX", + quantity=Decimal("50"), + proceeds_per_unit=Decimal("80"), + total_proceeds_aud=Decimal("4000"), + ) + event = CGTEvent( + event_date=date(2024, 3, 20), + symbol="CBA.AX", + asset_type=AssetType.SHARES, + disposal=disposal, + gross_gain=Decimal("-775"), + discount_eligible=False, + discount_amount=Decimal("0"), + net_gain=Decimal("-775"), + holding_period_days=430, + cgt_method=CGTMethod.OTHER, + tax_year=2024, + ) + assert event.is_gain is False + assert event.is_loss is True + + +# ============================================================================== +# TaxYearSummary Tests +# ============================================================================== + + +class TestTaxYearSummary: + """Tests for TaxYearSummary dataclass.""" + + def test_num_events(self): + """Test event count property.""" + summary = TaxYearSummary( + tax_year=2024, + start_date=date(2023, 7, 1), + end_date=date(2024, 6, 30), + total_gains=Decimal("1000"), + total_losses=Decimal("200"), + losses_applied=Decimal("200"), + carried_forward_losses=Decimal("0"), + losses_to_carry=Decimal("0"), + net_capital_gain=Decimal("800"), + discounted_gains=Decimal("1000"), + discount_applied=Decimal("400"), + taxable_gain=Decimal("400"), + events=[], + ) + assert summary.num_events == 0 + + def test_num_gains_and_losses(self): + """Test gain and loss count properties.""" + disposal1 = CGTDisposal( + disposal_date=date(2024, 1, 15), + symbol="CBA.AX", + quantity=Decimal("50"), + proceeds_per_unit=Decimal("110"), + total_proceeds_aud=Decimal("5500"), + ) + event1 = CGTEvent( + event_date=date(2024, 1, 15), + symbol="CBA.AX", + asset_type=AssetType.SHARES, + disposal=disposal1, + gross_gain=Decimal("500"), + discount_eligible=True, + discount_amount=Decimal("250"), + net_gain=Decimal("250"), + holding_period_days=400, + cgt_method=CGTMethod.DISCOUNT, + tax_year=2024, + ) + disposal2 = CGTDisposal( + disposal_date=date(2024, 2, 15), + symbol="NAB.AX", + quantity=Decimal("50"), + proceeds_per_unit=Decimal("25"), + total_proceeds_aud=Decimal("1250"), + ) + event2 = CGTEvent( + event_date=date(2024, 2, 15), + symbol="NAB.AX", + asset_type=AssetType.SHARES, + disposal=disposal2, + gross_gain=Decimal("-250"), + discount_eligible=False, + discount_amount=Decimal("0"), + net_gain=Decimal("-250"), + holding_period_days=100, + cgt_method=CGTMethod.OTHER, + tax_year=2024, + ) + summary = TaxYearSummary( + tax_year=2024, + start_date=date(2023, 7, 1), + end_date=date(2024, 6, 30), + total_gains=Decimal("500"), + total_losses=Decimal("250"), + losses_applied=Decimal("250"), + carried_forward_losses=Decimal("0"), + losses_to_carry=Decimal("0"), + net_capital_gain=Decimal("250"), + discounted_gains=Decimal("500"), + discount_applied=Decimal("125"), + taxable_gain=Decimal("125"), + events=[event1, event2], + ) + assert summary.num_events == 2 + assert summary.num_gains == 1 + assert summary.num_losses == 1 + + +# ============================================================================== +# AustralianCGTCalculator - Tax Year Tests +# ============================================================================== + + +class TestTaxYearCalculations: + """Tests for Australian tax year calculations.""" + + def test_tax_year_july_date(self): + """Test tax year for July date (start of new FY).""" + # July 1, 2023 is in FY2023-24 (tax year 2024) + assert AustralianCGTCalculator.get_tax_year(date(2023, 7, 1)) == 2024 + + def test_tax_year_june_date(self): + """Test tax year for June date (end of FY).""" + # June 30, 2024 is in FY2023-24 (tax year 2024) + assert AustralianCGTCalculator.get_tax_year(date(2024, 6, 30)) == 2024 + + def test_tax_year_january_date(self): + """Test tax year for January date.""" + # January 15, 2024 is in FY2023-24 (tax year 2024) + assert AustralianCGTCalculator.get_tax_year(date(2024, 1, 15)) == 2024 + + def test_tax_year_december_date(self): + """Test tax year for December date.""" + # December 15, 2023 is in FY2023-24 (tax year 2024) + assert AustralianCGTCalculator.get_tax_year(date(2023, 12, 15)) == 2024 + + def test_tax_year_dates(self): + """Test getting tax year start and end dates.""" + start, end = AustralianCGTCalculator.get_tax_year_dates(2024) + assert start == date(2023, 7, 1) + assert end == date(2024, 6, 30) + + +# ============================================================================== +# AustralianCGTCalculator - Acquisition Tests +# ============================================================================== + + +class TestAcquisitions: + """Tests for asset acquisition recording.""" + + def test_add_basic_acquisition(self): + """Test adding a basic AUD acquisition.""" + calculator = AustralianCGTCalculator() + acq = calculator.add_acquisition( + symbol="CBA.AX", + acquisition_date=date(2023, 1, 15), + quantity=Decimal("100"), + cost_per_unit=Decimal("95.50"), + ) + assert acq.quantity == Decimal("100") + assert acq.cost_per_unit == Decimal("95.50") + assert acq.total_cost_aud == Decimal("9550") + assert calculator.get_holding_quantity("CBA.AX") == Decimal("100") + + def test_add_acquisition_with_brokerage(self): + """Test adding acquisition with incidental costs.""" + calculator = AustralianCGTCalculator() + acq = calculator.add_acquisition( + symbol="CBA.AX", + acquisition_date=date(2023, 1, 15), + quantity=Decimal("100"), + cost_per_unit=Decimal("95.50"), + incidental_costs=Decimal("19.95"), + ) + assert acq.incidental_costs == Decimal("19.95") + assert acq.total_cost_base == Decimal("9569.95") + + def test_add_foreign_acquisition(self): + """Test adding a foreign currency acquisition.""" + calculator = AustralianCGTCalculator() + acq = calculator.add_acquisition( + symbol="AAPL", + acquisition_date=date(2023, 1, 15), + quantity=Decimal("50"), + cost_per_unit=Decimal("150"), # USD + currency="USD", + exchange_rate=Decimal("1.50"), # AUD per USD + ) + assert acq.currency == "USD" + assert acq.exchange_rate == Decimal("1.50") + # 50 * 150 * 1.50 = 11250 + assert acq.total_cost_aud == Decimal("11250") + + def test_add_multiple_acquisitions_same_symbol(self): + """Test adding multiple parcels of same asset.""" + calculator = AustralianCGTCalculator() + calculator.add_acquisition( + symbol="CBA.AX", + acquisition_date=date(2023, 1, 15), + quantity=Decimal("100"), + cost_per_unit=Decimal("95.50"), + ) + calculator.add_acquisition( + symbol="CBA.AX", + acquisition_date=date(2023, 6, 20), + quantity=Decimal("50"), + cost_per_unit=Decimal("100"), + ) + assert calculator.get_holding_quantity("CBA.AX") == Decimal("150") + holdings = calculator.get_holdings("CBA.AX") + assert len(holdings) == 2 + # Should be sorted by acquisition date (FIFO) + assert holdings[0].acquisition_date == date(2023, 1, 15) + assert holdings[1].acquisition_date == date(2023, 6, 20) + + def test_acquisition_with_asset_type(self): + """Test setting asset type on acquisition.""" + calculator = AustralianCGTCalculator() + calculator.add_acquisition( + symbol="BTC", + acquisition_date=date(2023, 1, 15), + quantity=Decimal("0.5"), + cost_per_unit=Decimal("30000"), + asset_type=AssetType.CRYPTOCURRENCY, + ) + assert calculator.get_asset_type("BTC") == AssetType.CRYPTOCURRENCY + + def test_acquisition_invalid_quantity(self): + """Test acquisition with invalid quantity.""" + calculator = AustralianCGTCalculator() + with pytest.raises(ValueError, match="Quantity must be positive"): + calculator.add_acquisition( + symbol="CBA.AX", + acquisition_date=date(2023, 1, 15), + quantity=Decimal("0"), + cost_per_unit=Decimal("95.50"), + ) + + def test_acquisition_negative_quantity(self): + """Test acquisition with negative quantity.""" + calculator = AustralianCGTCalculator() + with pytest.raises(ValueError, match="Quantity must be positive"): + calculator.add_acquisition( + symbol="CBA.AX", + acquisition_date=date(2023, 1, 15), + quantity=Decimal("-10"), + cost_per_unit=Decimal("95.50"), + ) + + def test_foreign_acquisition_missing_exchange_rate(self): + """Test foreign acquisition without exchange rate.""" + calculator = AustralianCGTCalculator() + with pytest.raises(ValueError, match="Exchange rate required"): + calculator.add_acquisition( + symbol="AAPL", + acquisition_date=date(2023, 1, 15), + quantity=Decimal("50"), + cost_per_unit=Decimal("150"), + currency="USD", + ) + + +# ============================================================================== +# AustralianCGTCalculator - Disposal Tests +# ============================================================================== + + +class TestDisposals: + """Tests for asset disposal and CGT calculation.""" + + def test_basic_disposal_with_gain(self): + """Test basic disposal with capital gain.""" + calculator = AustralianCGTCalculator() + calculator.add_acquisition( + symbol="CBA.AX", + acquisition_date=date(2023, 1, 15), + quantity=Decimal("100"), + cost_per_unit=Decimal("95.50"), + ) + event = calculator.dispose( + symbol="CBA.AX", + disposal_date=date(2023, 6, 20), # Less than 12 months + quantity=Decimal("50"), + proceeds_per_unit=Decimal("110.25"), + ) + assert event.symbol == "CBA.AX" + assert event.disposal.quantity == Decimal("50") + # Proceeds: 50 * 110.25 = 5512.50 + # Cost: 50 * 95.50 = 4775 + # Gain: 5512.50 - 4775 = 737.50 + assert event.gross_gain == Decimal("737.50") + assert event.discount_eligible is False # Less than 12 months + assert event.net_gain == Decimal("737.50") + assert event.cgt_method == CGTMethod.OTHER + + def test_disposal_with_discount(self): + """Test disposal eligible for 50% CGT discount.""" + calculator = AustralianCGTCalculator() + calculator.add_acquisition( + symbol="CBA.AX", + acquisition_date=date(2023, 1, 15), + quantity=Decimal("100"), + cost_per_unit=Decimal("95.50"), + ) + event = calculator.dispose( + symbol="CBA.AX", + disposal_date=date(2024, 3, 20), # More than 12 months + quantity=Decimal("50"), + proceeds_per_unit=Decimal("110.25"), + ) + assert event.discount_eligible is True + assert event.gross_gain == Decimal("737.50") + assert event.discount_amount == Decimal("368.75") + assert event.net_gain == Decimal("368.75") + assert event.cgt_method == CGTMethod.DISCOUNT + + def test_disposal_with_loss_no_discount(self): + """Test disposal with capital loss (no discount on losses).""" + calculator = AustralianCGTCalculator() + calculator.add_acquisition( + symbol="CBA.AX", + acquisition_date=date(2023, 1, 15), + quantity=Decimal("100"), + cost_per_unit=Decimal("95.50"), + ) + event = calculator.dispose( + symbol="CBA.AX", + disposal_date=date(2024, 3, 20), # More than 12 months + quantity=Decimal("50"), + proceeds_per_unit=Decimal("80"), # Selling at loss + ) + # Proceeds: 50 * 80 = 4000 + # Cost: 50 * 95.50 = 4775 + # Loss: 4000 - 4775 = -775 + assert event.gross_gain == Decimal("-775.00") + assert event.discount_eligible is False # No discount on losses + assert event.discount_amount == Decimal("0.00") + assert event.net_gain == Decimal("-775.00") + assert event.is_loss is True + + def test_disposal_fifo_matching(self): + """Test FIFO cost basis matching.""" + calculator = AustralianCGTCalculator() + # First parcel at $90 + calculator.add_acquisition( + symbol="CBA.AX", + acquisition_date=date(2023, 1, 15), + quantity=Decimal("100"), + cost_per_unit=Decimal("90"), + ) + # Second parcel at $100 + calculator.add_acquisition( + symbol="CBA.AX", + acquisition_date=date(2023, 6, 20), + quantity=Decimal("100"), + cost_per_unit=Decimal("100"), + ) + # Dispose 150 shares - should use first 100 at $90, then 50 at $100 + event = calculator.dispose( + symbol="CBA.AX", + disposal_date=date(2023, 9, 15), + quantity=Decimal("150"), + proceeds_per_unit=Decimal("105"), + ) + # Proceeds: 150 * 105 = 15750 + # Cost: 100 * 90 + 50 * 100 = 9000 + 5000 = 14000 + # Gain: 15750 - 14000 = 1750 + assert event.gross_gain == Decimal("1750.00") + # Remaining should be 50 shares at $100 + assert calculator.get_holding_quantity("CBA.AX") == Decimal("50") + + def test_disposal_partial_parcel(self): + """Test partial disposal of a parcel.""" + calculator = AustralianCGTCalculator() + calculator.add_acquisition( + symbol="CBA.AX", + acquisition_date=date(2023, 1, 15), + quantity=Decimal("100"), + cost_per_unit=Decimal("95.50"), + ) + calculator.dispose( + symbol="CBA.AX", + disposal_date=date(2023, 6, 20), + quantity=Decimal("30"), + proceeds_per_unit=Decimal("100"), + ) + # Remaining should be 70 shares + assert calculator.get_holding_quantity("CBA.AX") == Decimal("70") + holdings = calculator.get_holdings("CBA.AX") + assert len(holdings) == 1 + assert holdings[0].quantity == Decimal("70") + + def test_disposal_insufficient_holdings(self): + """Test disposal with insufficient holdings.""" + calculator = AustralianCGTCalculator() + calculator.add_acquisition( + symbol="CBA.AX", + acquisition_date=date(2023, 1, 15), + quantity=Decimal("100"), + cost_per_unit=Decimal("95.50"), + ) + with pytest.raises(ValueError, match="Insufficient holdings"): + calculator.dispose( + symbol="CBA.AX", + disposal_date=date(2023, 6, 20), + quantity=Decimal("150"), + proceeds_per_unit=Decimal("100"), + ) + + def test_disposal_no_holdings(self): + """Test disposal with no holdings.""" + calculator = AustralianCGTCalculator() + with pytest.raises(ValueError, match="Insufficient holdings"): + calculator.dispose( + symbol="CBA.AX", + disposal_date=date(2023, 6, 20), + quantity=Decimal("50"), + proceeds_per_unit=Decimal("100"), + ) + + def test_disposal_with_selling_costs(self): + """Test disposal with incidental (selling) costs.""" + calculator = AustralianCGTCalculator() + calculator.add_acquisition( + symbol="CBA.AX", + acquisition_date=date(2023, 1, 15), + quantity=Decimal("100"), + cost_per_unit=Decimal("95.50"), + ) + event = calculator.dispose( + symbol="CBA.AX", + disposal_date=date(2023, 6, 20), + quantity=Decimal("50"), + proceeds_per_unit=Decimal("110"), + incidental_costs=Decimal("19.95"), + ) + # Net proceeds: 50 * 110 - 19.95 = 5480.05 + # Cost: 50 * 95.50 = 4775 + # Gain: 5480.05 - 4775 = 705.05 + assert event.gross_gain == Decimal("705.05") + + +# ============================================================================== +# AustralianCGTCalculator - Tax Year Summary Tests +# ============================================================================== + + +class TestTaxYearSummary: + """Tests for tax year summary calculations.""" + + def test_simple_summary_with_gains(self): + """Test summary with only gains.""" + calculator = AustralianCGTCalculator() + calculator.add_acquisition( + symbol="CBA.AX", + acquisition_date=date(2022, 7, 1), + quantity=Decimal("100"), + cost_per_unit=Decimal("90"), + ) + calculator.dispose( + symbol="CBA.AX", + disposal_date=date(2023, 10, 15), # FY2024, >12 months + quantity=Decimal("100"), + proceeds_per_unit=Decimal("110"), + ) + summary = calculator.calculate_tax_year_summary(2024) + # Gross gain: 100 * (110 - 90) = 2000 + assert summary.total_gains == Decimal("2000.00") + assert summary.total_losses == Decimal("0.00") + assert summary.discounted_gains == Decimal("2000.00") + # Discount: 2000 * 0.50 = 1000 + assert summary.discount_applied == Decimal("1000.00") + assert summary.taxable_gain == Decimal("1000.00") + + def test_summary_with_losses_applied(self): + """Test summary with losses applied to gains.""" + calculator = AustralianCGTCalculator() + # Gain asset + calculator.add_acquisition( + symbol="CBA.AX", + acquisition_date=date(2022, 7, 1), + quantity=Decimal("100"), + cost_per_unit=Decimal("90"), + ) + # Loss asset + calculator.add_acquisition( + symbol="NAB.AX", + acquisition_date=date(2022, 7, 1), + quantity=Decimal("100"), + cost_per_unit=Decimal("35"), + ) + # Dispose with gain + calculator.dispose( + symbol="CBA.AX", + disposal_date=date(2023, 10, 15), + quantity=Decimal("100"), + proceeds_per_unit=Decimal("110"), + ) + # Dispose with loss + calculator.dispose( + symbol="NAB.AX", + disposal_date=date(2023, 11, 15), + quantity=Decimal("100"), + proceeds_per_unit=Decimal("25"), + ) + summary = calculator.calculate_tax_year_summary(2024) + assert summary.total_gains == Decimal("2000.00") + assert summary.total_losses == Decimal("1000.00") + assert summary.losses_applied == Decimal("1000.00") + + def test_summary_with_carried_forward_losses(self): + """Test summary with carried forward losses.""" + calculator = AustralianCGTCalculator() + calculator.set_carried_forward_losses(Decimal("500")) + calculator.add_acquisition( + symbol="CBA.AX", + acquisition_date=date(2022, 7, 1), + quantity=Decimal("100"), + cost_per_unit=Decimal("90"), + ) + calculator.dispose( + symbol="CBA.AX", + disposal_date=date(2023, 10, 15), + quantity=Decimal("100"), + proceeds_per_unit=Decimal("110"), + ) + summary = calculator.calculate_tax_year_summary(2024) + assert summary.carried_forward_losses == Decimal("500.00") + # Gross gain 2000 - 500 carried loss = 1500 + # Then 50% discount = 750 taxable + assert summary.taxable_gain == Decimal("750.00") + + def test_summary_excess_losses_carry_forward(self): + """Test excess losses are carried forward.""" + calculator = AustralianCGTCalculator() + calculator.add_acquisition( + symbol="CBA.AX", + acquisition_date=date(2022, 7, 1), + quantity=Decimal("100"), + cost_per_unit=Decimal("90"), + ) + calculator.add_acquisition( + symbol="NAB.AX", + acquisition_date=date(2022, 7, 1), + quantity=Decimal("100"), + cost_per_unit=Decimal("50"), + ) + # Small gain + calculator.dispose( + symbol="CBA.AX", + disposal_date=date(2023, 10, 15), + quantity=Decimal("100"), + proceeds_per_unit=Decimal("95"), + ) + # Large loss + calculator.dispose( + symbol="NAB.AX", + disposal_date=date(2023, 11, 15), + quantity=Decimal("100"), + proceeds_per_unit=Decimal("20"), + ) + summary = calculator.calculate_tax_year_summary(2024) + assert summary.total_gains == Decimal("500.00") # 100 * (95-90) + assert summary.total_losses == Decimal("3000.00") # 100 * (50-20) + assert summary.taxable_gain == Decimal("0.00") + assert summary.losses_to_carry == Decimal("2500.00") # 3000 - 500 + + def test_summary_no_events(self): + """Test summary with no events for tax year.""" + calculator = AustralianCGTCalculator() + summary = calculator.calculate_tax_year_summary(2024) + assert summary.num_events == 0 + assert summary.total_gains == Decimal("0.00") + assert summary.total_losses == Decimal("0.00") + assert summary.taxable_gain == Decimal("0.00") + + +# ============================================================================== +# AustralianCGTCalculator - Tax Report Tests +# ============================================================================== + + +class TestTaxReport: + """Tests for tax report generation.""" + + def test_generate_basic_report(self): + """Test generating a basic tax report.""" + calculator = AustralianCGTCalculator() + calculator.add_acquisition( + symbol="CBA.AX", + acquisition_date=date(2022, 7, 1), + quantity=Decimal("100"), + cost_per_unit=Decimal("90"), + ) + calculator.dispose( + symbol="CBA.AX", + disposal_date=date(2023, 10, 15), + quantity=Decimal("100"), + proceeds_per_unit=Decimal("110"), + ) + report = calculator.generate_tax_report(2024) + assert report["tax_year"] == "FY2023-24" + assert report["period"]["start"] == "2023-07-01" + assert report["period"]["end"] == "2024-06-30" + assert "summary" in report + assert "statistics" in report + assert "transactions" in report + + def test_report_summary_values(self): + """Test report summary values.""" + calculator = AustralianCGTCalculator() + calculator.add_acquisition( + symbol="CBA.AX", + acquisition_date=date(2022, 7, 1), + quantity=Decimal("100"), + cost_per_unit=Decimal("90"), + ) + calculator.dispose( + symbol="CBA.AX", + disposal_date=date(2023, 10, 15), + quantity=Decimal("100"), + proceeds_per_unit=Decimal("110"), + ) + report = calculator.generate_tax_report(2024) + summary = report["summary"] + assert summary["total_capital_gains"] == "2000.00" + assert summary["total_capital_losses"] == "0.00" + assert summary["discount_method_gains"] == "2000.00" + assert summary["cgt_discount_applied"] == "1000.00" + assert summary["taxable_capital_gain"] == "1000.00" + + def test_report_without_details(self): + """Test report without transaction details.""" + calculator = AustralianCGTCalculator() + calculator.add_acquisition( + symbol="CBA.AX", + acquisition_date=date(2022, 7, 1), + quantity=Decimal("100"), + cost_per_unit=Decimal("90"), + ) + calculator.dispose( + symbol="CBA.AX", + disposal_date=date(2023, 10, 15), + quantity=Decimal("100"), + proceeds_per_unit=Decimal("110"), + ) + report = calculator.generate_tax_report(2024, include_details=False) + assert "transactions" not in report + + def test_report_statistics(self): + """Test report statistics.""" + calculator = AustralianCGTCalculator() + # Two gains, one loss + calculator.add_acquisition( + symbol="CBA.AX", + acquisition_date=date(2022, 7, 1), + quantity=Decimal("100"), + cost_per_unit=Decimal("90"), + ) + calculator.add_acquisition( + symbol="BHP.AX", + acquisition_date=date(2022, 7, 1), + quantity=Decimal("100"), + cost_per_unit=Decimal("40"), + ) + calculator.add_acquisition( + symbol="NAB.AX", + acquisition_date=date(2022, 7, 1), + quantity=Decimal("100"), + cost_per_unit=Decimal("35"), + ) + calculator.dispose( + symbol="CBA.AX", + disposal_date=date(2023, 10, 15), + quantity=Decimal("100"), + proceeds_per_unit=Decimal("110"), + ) + calculator.dispose( + symbol="BHP.AX", + disposal_date=date(2023, 10, 15), + quantity=Decimal("100"), + proceeds_per_unit=Decimal("50"), + ) + calculator.dispose( + symbol="NAB.AX", + disposal_date=date(2023, 10, 15), + quantity=Decimal("100"), + proceeds_per_unit=Decimal("25"), + ) + report = calculator.generate_tax_report(2024) + stats = report["statistics"] + assert stats["number_of_disposals"] == 3 + assert stats["number_of_gains"] == 2 + assert stats["number_of_losses"] == 1 + + +# ============================================================================== +# AustralianCGTCalculator - Unrealised Gains Tests +# ============================================================================== + + +class TestUnrealisedGains: + """Tests for unrealised gains calculation.""" + + def test_unrealised_gains_calculation(self): + """Test calculating unrealised gains.""" + calculator = AustralianCGTCalculator() + calculator.add_acquisition( + symbol="CBA.AX", + acquisition_date=date(2023, 1, 15), + quantity=Decimal("100"), + cost_per_unit=Decimal("90"), + ) + result = calculator.get_unrealised_gains({"CBA.AX": Decimal("110")}) + assert result["CBA.AX"]["quantity"] == Decimal("100") + assert result["CBA.AX"]["cost_base"] == Decimal("9000.00") + assert result["CBA.AX"]["market_value"] == Decimal("11000.00") + assert result["CBA.AX"]["unrealised_gain"] == Decimal("2000.00") + + def test_unrealised_loss_calculation(self): + """Test calculating unrealised losses.""" + calculator = AustralianCGTCalculator() + calculator.add_acquisition( + symbol="CBA.AX", + acquisition_date=date(2023, 1, 15), + quantity=Decimal("100"), + cost_per_unit=Decimal("90"), + ) + result = calculator.get_unrealised_gains({"CBA.AX": Decimal("80")}) + assert result["CBA.AX"]["unrealised_gain"] == Decimal("-1000.00") + + def test_unrealised_gains_missing_price(self): + """Test unrealised gains with missing price.""" + calculator = AustralianCGTCalculator() + calculator.add_acquisition( + symbol="CBA.AX", + acquisition_date=date(2023, 1, 15), + quantity=Decimal("100"), + cost_per_unit=Decimal("90"), + ) + result = calculator.get_unrealised_gains({}) + assert result["CBA.AX"]["market_value"] == Decimal("0.00") + assert result["CBA.AX"]["unrealised_gain"] == Decimal("0.00") + + def test_unrealised_gains_multiple_parcels(self): + """Test unrealised gains with multiple acquisition parcels.""" + calculator = AustralianCGTCalculator() + calculator.add_acquisition( + symbol="CBA.AX", + acquisition_date=date(2023, 1, 15), + quantity=Decimal("100"), + cost_per_unit=Decimal("90"), + ) + calculator.add_acquisition( + symbol="CBA.AX", + acquisition_date=date(2023, 6, 15), + quantity=Decimal("50"), + cost_per_unit=Decimal("100"), + ) + result = calculator.get_unrealised_gains({"CBA.AX": Decimal("110")}) + assert result["CBA.AX"]["quantity"] == Decimal("150") + # Cost: 100 * 90 + 50 * 100 = 14000 + assert result["CBA.AX"]["cost_base"] == Decimal("14000.00") + # Value: 150 * 110 = 16500 + assert result["CBA.AX"]["market_value"] == Decimal("16500.00") + assert result["CBA.AX"]["unrealised_gain"] == Decimal("2500.00") + + +# ============================================================================== +# AustralianCGTCalculator - Edge Cases Tests +# ============================================================================== + + +class TestEdgeCases: + """Tests for edge cases and special scenarios.""" + + def test_exactly_365_days_no_discount(self): + """Test holding exactly 365 days (not eligible for discount).""" + calculator = AustralianCGTCalculator() + acq_date = date(2023, 1, 15) + calculator.add_acquisition( + symbol="CBA.AX", + acquisition_date=acq_date, + quantity=Decimal("100"), + cost_per_unit=Decimal("90"), + ) + # Exactly 365 days later + disposal_date = acq_date + timedelta(days=365) + event = calculator.dispose( + symbol="CBA.AX", + disposal_date=disposal_date, + quantity=Decimal("100"), + proceeds_per_unit=Decimal("110"), + ) + # Must be MORE than 365 days for discount + assert event.discount_eligible is False + + def test_366_days_eligible_for_discount(self): + """Test holding 366 days (eligible for discount).""" + calculator = AustralianCGTCalculator() + acq_date = date(2023, 1, 15) + calculator.add_acquisition( + symbol="CBA.AX", + acquisition_date=acq_date, + quantity=Decimal("100"), + cost_per_unit=Decimal("90"), + ) + disposal_date = acq_date + timedelta(days=366) + event = calculator.dispose( + symbol="CBA.AX", + disposal_date=disposal_date, + quantity=Decimal("100"), + proceeds_per_unit=Decimal("110"), + ) + assert event.discount_eligible is True + + def test_zero_cost_acquisition(self): + """Test acquisition with zero cost (e.g., inheritance).""" + calculator = AustralianCGTCalculator() + acq = calculator.add_acquisition( + symbol="CBA.AX", + acquisition_date=date(2023, 1, 15), + quantity=Decimal("100"), + cost_per_unit=Decimal("0"), + ) + assert acq.total_cost_aud == Decimal("0") + event = calculator.dispose( + symbol="CBA.AX", + disposal_date=date(2024, 3, 20), + quantity=Decimal("100"), + proceeds_per_unit=Decimal("100"), + ) + # All proceeds are gain + assert event.gross_gain == Decimal("10000.00") + + def test_very_small_quantities(self): + """Test with very small quantities (fractional shares).""" + calculator = AustralianCGTCalculator() + calculator.add_acquisition( + symbol="BRK.A", + acquisition_date=date(2023, 1, 15), + quantity=Decimal("0.001"), + cost_per_unit=Decimal("500000"), + ) + event = calculator.dispose( + symbol="BRK.A", + disposal_date=date(2023, 6, 20), + quantity=Decimal("0.001"), + proceeds_per_unit=Decimal("510000"), + ) + assert event.gross_gain == Decimal("10.00") + + def test_clear_calculator(self): + """Test clearing all calculator data.""" + calculator = AustralianCGTCalculator() + calculator.add_acquisition( + symbol="CBA.AX", + acquisition_date=date(2023, 1, 15), + quantity=Decimal("100"), + cost_per_unit=Decimal("90"), + ) + calculator.set_carried_forward_losses(Decimal("1000")) + calculator.clear() + assert calculator.get_holding_quantity("CBA.AX") == Decimal("0") + assert calculator.get_carried_forward_losses() == Decimal("0") + assert len(calculator.get_events()) == 0 + + def test_set_and_get_asset_type(self): + """Test setting and getting asset type.""" + calculator = AustralianCGTCalculator() + calculator.set_asset_type("BTC", AssetType.CRYPTOCURRENCY) + assert calculator.get_asset_type("BTC") == AssetType.CRYPTOCURRENCY + # Default type for unknown asset + assert calculator.get_asset_type("UNKNOWN") == AssetType.SHARES + + def test_invalid_carried_forward_losses(self): + """Test setting invalid carried forward losses.""" + calculator = AustralianCGTCalculator() + with pytest.raises(ValueError, match="non-negative"): + calculator.set_carried_forward_losses(Decimal("-100")) + + def test_get_events_all(self): + """Test getting all events.""" + calculator = AustralianCGTCalculator() + calculator.add_acquisition( + symbol="CBA.AX", + acquisition_date=date(2022, 7, 1), + quantity=Decimal("100"), + cost_per_unit=Decimal("90"), + ) + calculator.dispose( + symbol="CBA.AX", + disposal_date=date(2023, 10, 15), + quantity=Decimal("50"), + proceeds_per_unit=Decimal("110"), + ) + calculator.dispose( + symbol="CBA.AX", + disposal_date=date(2024, 10, 15), + quantity=Decimal("50"), + proceeds_per_unit=Decimal("120"), + ) + all_events = calculator.get_events() + assert len(all_events) == 2 + + def test_get_events_filtered_by_year(self): + """Test getting events filtered by tax year.""" + calculator = AustralianCGTCalculator() + calculator.add_acquisition( + symbol="CBA.AX", + acquisition_date=date(2022, 7, 1), + quantity=Decimal("100"), + cost_per_unit=Decimal("90"), + ) + calculator.dispose( + symbol="CBA.AX", + disposal_date=date(2023, 10, 15), # FY2024 + quantity=Decimal("50"), + proceeds_per_unit=Decimal("110"), + ) + calculator.dispose( + symbol="CBA.AX", + disposal_date=date(2024, 10, 15), # FY2025 + quantity=Decimal("50"), + proceeds_per_unit=Decimal("120"), + ) + fy2024_events = calculator.get_events(tax_year=2024) + assert len(fy2024_events) == 1 + fy2025_events = calculator.get_events(tax_year=2025) + assert len(fy2025_events) == 1 + + +# ============================================================================== +# Module Import Tests +# ============================================================================== + + +class TestModuleImports: + """Tests for module imports.""" + + def test_import_from_portfolio_module(self): + """Test importing from portfolio module.""" + from tradingagents.portfolio import ( + CGTMethod, + AssetType, + CGTAssetAcquisition, + CGTDisposal, + CGTEvent, + TaxYearSummary, + AustralianCGTCalculator, + ) + assert CGTMethod is not None + assert AssetType is not None + assert CGTAssetAcquisition is not None + assert CGTDisposal is not None + assert CGTEvent is not None + assert TaxYearSummary is not None + assert AustralianCGTCalculator is not None + + def test_calculator_constants(self): + """Test calculator constants.""" + assert AustralianCGTCalculator.DISCOUNT_RATE == Decimal("0.50") + assert AustralianCGTCalculator.DISCOUNT_HOLDING_DAYS == 365 diff --git a/tradingagents/portfolio/__init__.py b/tradingagents/portfolio/__init__.py index 7f93138f..ce9f9ec1 100644 --- a/tradingagents/portfolio/__init__.py +++ b/tradingagents/portfolio/__init__.py @@ -1,6 +1,6 @@ -"""Portfolio module for portfolio state and performance management. +"""Portfolio module for portfolio state, performance, and tax management. -This module provides portfolio state tracking and performance metrics: +This module provides portfolio state tracking, performance metrics, and tax calculations: - Current holdings with cost basis and market values - Multi-currency cash balances - Real-time mark-to-market valuation @@ -9,19 +9,26 @@ This module provides portfolio state tracking and performance metrics: - Drawdown analysis - Trade statistics - Benchmark comparison +- Australian CGT calculations with 50% discount +- FIFO cost basis tracking +- Tax year reports Issue #29: [PORT-28] Portfolio state - holdings, cash, mark-to-market Issue #31: [PORT-30] Performance metrics - Sharpe, drawdown, returns +Issue #32: [PORT-31] Australian CGT calculator - 50% discount, tax reports Submodules: portfolio_state: Core portfolio state management performance: Performance metrics calculation + tax_calculator: Australian CGT calculations Classes: Enums: - Currency: Supported currencies (USD, EUR, GBP, etc.) - HoldingType: Type of holding (LONG, SHORT) - Period: Time period for performance calculations + - CGTMethod: CGT calculation method (discount, indexation, other) + - AssetType: Type of CGT asset (shares, ETF, crypto, etc.) Data Classes: - Holding: Individual holding/position in the portfolio @@ -31,10 +38,15 @@ Classes: - DrawdownInfo: Information about a drawdown period - TradeStats: Trade-level statistics - PerformanceMetrics: Complete performance metrics summary + - CGTAssetAcquisition: Record of an asset purchase (cost base parcel) + - CGTDisposal: Record of an asset sale + - CGTEvent: A CGT event (disposal with gain/loss calculation) + - TaxYearSummary: Summary of CGT for an Australian tax year Main Classes: - PortfolioState: Live portfolio state with mark-to-market updates - PerformanceCalculator: Calculator for performance metrics + - AustralianCGTCalculator: Australian CGT calculator with 50% discount Protocols: - PriceProvider: Protocol for price data providers @@ -101,6 +113,19 @@ from .performance import ( calculate_yearly_returns, ) +from .tax_calculator import ( + # Enums + CGTMethod, + AssetType, + # Data Classes + CGTAssetAcquisition, + CGTDisposal, + CGTEvent, + TaxYearSummary, + # Main Class + AustralianCGTCalculator, +) + __all__ = [ # Portfolio State Enums "Currency", @@ -128,4 +153,14 @@ __all__ = [ "calculate_rolling_returns", "calculate_monthly_returns", "calculate_yearly_returns", + # Tax Calculator Enums + "CGTMethod", + "AssetType", + # Tax Calculator Data Classes + "CGTAssetAcquisition", + "CGTDisposal", + "CGTEvent", + "TaxYearSummary", + # Tax Calculator Main Class + "AustralianCGTCalculator", ] diff --git a/tradingagents/portfolio/tax_calculator.py b/tradingagents/portfolio/tax_calculator.py new file mode 100644 index 00000000..48a68a51 --- /dev/null +++ b/tradingagents/portfolio/tax_calculator.py @@ -0,0 +1,769 @@ +"""Australian Capital Gains Tax Calculator. + +This module provides Australian CGT calculations including: +- CGT calculations with 50% discount for assets held >12 months +- Tax year reports (Australian financial year: July-June) +- Currency conversion for foreign assets +- Capital loss tracking and carry-forward +- FIFO cost basis calculations + +Issue #32: [PORT-31] Australian CGT calculator - 50% discount, tax reports + +Design Principles: + - ATO-compliant CGT calculations + - Australian financial year (July 1 - June 30) + - Support for various asset classes + - Comprehensive audit trail +""" + +from dataclasses import dataclass, field +from datetime import datetime, date, timedelta +from decimal import Decimal, ROUND_HALF_UP +from enum import Enum +from typing import Any, Dict, List, Optional, Tuple, Union + + +class CGTMethod(Enum): + """CGT calculation method.""" + DISCOUNT = "discount" # 50% discount for assets held >12 months + INDEXATION = "indexation" # Historical indexation method (pre-1999) + OTHER = "other" # No special treatment + + +class AssetType(Enum): + """Type of CGT asset.""" + SHARES = "shares" # Australian shares + FOREIGN_SHARES = "foreign_shares" # Foreign shares + ETF = "etf" # Exchange-traded funds + CRYPTOCURRENCY = "cryptocurrency" # Digital assets + PROPERTY = "property" # Real estate + COLLECTABLES = "collectables" # Art, jewellery, etc. + OTHER = "other" + + +@dataclass +class CGTAssetAcquisition: + """Record of an asset acquisition (cost base parcel). + + Attributes: + acquisition_date: When the asset was acquired + quantity: Number of units acquired + cost_per_unit: Cost per unit in acquisition currency + total_cost_aud: Total cost in AUD (after currency conversion) + currency: Currency of acquisition (for foreign assets) + exchange_rate: Exchange rate used for conversion (if foreign) + incidental_costs: Brokerage, legal fees, etc. + asset_id: Optional asset identifier + metadata: Additional acquisition data + """ + acquisition_date: date + quantity: Decimal + cost_per_unit: Decimal + total_cost_aud: Decimal + currency: str = "AUD" + exchange_rate: Optional[Decimal] = None + incidental_costs: Decimal = field(default_factory=lambda: Decimal("0")) + asset_id: Optional[str] = None + metadata: Dict[str, Any] = field(default_factory=dict) + + @property + def cost_base_per_unit(self) -> Decimal: + """Cost base per unit including incidental costs.""" + if self.quantity == 0: + return Decimal("0") + return (self.total_cost_aud + self.incidental_costs) / self.quantity + + @property + def total_cost_base(self) -> Decimal: + """Total cost base including incidental costs.""" + return self.total_cost_aud + self.incidental_costs + + +@dataclass +class CGTDisposal: + """Record of an asset disposal. + + Attributes: + disposal_date: When the asset was disposed + symbol: Asset symbol + quantity: Number of units disposed + proceeds_per_unit: Proceeds per unit in disposal currency + total_proceeds_aud: Total proceeds in AUD + currency: Currency of proceeds + exchange_rate: Exchange rate used for conversion + incidental_costs: Selling costs (brokerage, etc.) + matched_acquisitions: Acquisitions used for cost base (FIFO) + metadata: Additional disposal data + """ + disposal_date: date + symbol: str + quantity: Decimal + proceeds_per_unit: Decimal + total_proceeds_aud: Decimal + currency: str = "AUD" + exchange_rate: Optional[Decimal] = None + incidental_costs: Decimal = field(default_factory=lambda: Decimal("0")) + matched_acquisitions: List[Tuple[CGTAssetAcquisition, Decimal]] = field(default_factory=list) + metadata: Dict[str, Any] = field(default_factory=dict) + + @property + def net_proceeds(self) -> Decimal: + """Net proceeds after selling costs.""" + return self.total_proceeds_aud - self.incidental_costs + + +@dataclass +class CGTEvent: + """A CGT event (disposal of an asset). + + Attributes: + event_date: Date of the CGT event + symbol: Asset symbol + asset_type: Type of asset + disposal: Disposal record + gross_gain: Capital gain/loss before discount + discount_eligible: Whether 50% discount applies + discount_amount: Amount of discount (if eligible) + net_gain: Net capital gain/loss after discount + holding_period_days: Days between acquisition and disposal + cgt_method: CGT calculation method used + tax_year: Australian tax year (ending June 30) + metadata: Additional event data + """ + event_date: date + symbol: str + asset_type: AssetType + disposal: CGTDisposal + gross_gain: Decimal + discount_eligible: bool + discount_amount: Decimal + net_gain: Decimal + holding_period_days: int + cgt_method: CGTMethod + tax_year: int # Year ending (e.g., 2024 for FY2023-24) + metadata: Dict[str, Any] = field(default_factory=dict) + + @property + def is_gain(self) -> bool: + """Check if this is a capital gain.""" + return self.net_gain > 0 + + @property + def is_loss(self) -> bool: + """Check if this is a capital loss.""" + return self.net_gain < 0 + + +@dataclass +class TaxYearSummary: + """Summary of CGT events for an Australian tax year. + + Attributes: + tax_year: Australian tax year (year ending June 30) + start_date: Start of tax year (July 1) + end_date: End of tax year (June 30) + total_gains: Total capital gains (before applying losses) + total_losses: Total capital losses + losses_applied: Losses applied against gains + carried_forward_losses: Losses carried forward from previous years + losses_to_carry: Losses to carry forward to next year + net_capital_gain: Net capital gain after losses + discounted_gains: Total gains eligible for discount + discount_applied: Total discount amount applied + taxable_gain: Final taxable capital gain + events: List of CGT events in this year + metadata: Additional summary data + """ + tax_year: int + start_date: date + end_date: date + total_gains: Decimal + total_losses: Decimal + losses_applied: Decimal + carried_forward_losses: Decimal + losses_to_carry: Decimal + net_capital_gain: Decimal + discounted_gains: Decimal + discount_applied: Decimal + taxable_gain: Decimal + events: List[CGTEvent] = field(default_factory=list) + metadata: Dict[str, Any] = field(default_factory=dict) + + @property + def num_events(self) -> int: + """Number of CGT events in this tax year.""" + return len(self.events) + + @property + def num_gains(self) -> int: + """Number of gain events.""" + return sum(1 for e in self.events if e.is_gain) + + @property + def num_losses(self) -> int: + """Number of loss events.""" + return sum(1 for e in self.events if e.is_loss) + + +class AustralianCGTCalculator: + """Australian Capital Gains Tax calculator. + + Implements Australian CGT rules including: + - 50% discount for assets held more than 12 months + - FIFO (First In, First Out) cost base matching + - Capital loss tracking and carry-forward + - Australian financial year (July-June) + - Currency conversion for foreign assets + + Example: + >>> calculator = AustralianCGTCalculator() + >>> calculator.add_acquisition( + ... symbol="CBA.AX", + ... acquisition_date=date(2023, 1, 15), + ... quantity=Decimal("100"), + ... cost_per_unit=Decimal("95.50"), + ... ) + >>> event = calculator.dispose( + ... symbol="CBA.AX", + ... disposal_date=date(2024, 3, 20), + ... quantity=Decimal("50"), + ... proceeds_per_unit=Decimal("110.25"), + ... ) + >>> print(f"Net gain: ${event.net_gain}") + """ + + # CGT discount rate for eligible assets + DISCOUNT_RATE = Decimal("0.50") + + # Minimum holding period for discount (in days) + DISCOUNT_HOLDING_DAYS = 365 # > 12 months + + def __init__(self, base_currency: str = "AUD"): + """Initialize the CGT calculator. + + Args: + base_currency: Base currency for calculations (should be AUD) + """ + self.base_currency = base_currency + self._holdings: Dict[str, List[CGTAssetAcquisition]] = {} + self._asset_types: Dict[str, AssetType] = {} + self._events: List[CGTEvent] = [] + self._carried_forward_losses: Decimal = Decimal("0") + + @staticmethod + def get_tax_year(event_date: date) -> int: + """Get Australian tax year for a date. + + Australian tax year runs July 1 to June 30. + Returns the year in which the tax year ends. + + Args: + event_date: Date to check + + Returns: + Tax year (year ending) - e.g., 2024 for FY2023-24 + """ + if event_date.month >= 7: + # July onwards is next tax year + return event_date.year + 1 + else: + # January to June is current calendar year + return event_date.year + + @staticmethod + def get_tax_year_dates(tax_year: int) -> Tuple[date, date]: + """Get start and end dates for an Australian tax year. + + Args: + tax_year: Tax year (year ending) + + Returns: + Tuple of (start_date, end_date) + """ + start = date(tax_year - 1, 7, 1) + end = date(tax_year, 6, 30) + return start, end + + def set_asset_type(self, symbol: str, asset_type: AssetType) -> None: + """Set the asset type for a symbol. + + Args: + symbol: Asset symbol + asset_type: Type of asset + """ + self._asset_types[symbol] = asset_type + + def get_asset_type(self, symbol: str) -> AssetType: + """Get the asset type for a symbol. + + Args: + symbol: Asset symbol + + Returns: + Asset type (defaults to SHARES if not set) + """ + return self._asset_types.get(symbol, AssetType.SHARES) + + def add_acquisition( + self, + symbol: str, + acquisition_date: date, + quantity: Decimal, + cost_per_unit: Decimal, + currency: str = "AUD", + exchange_rate: Optional[Decimal] = None, + incidental_costs: Decimal = Decimal("0"), + asset_type: Optional[AssetType] = None, + metadata: Optional[Dict[str, Any]] = None, + ) -> CGTAssetAcquisition: + """Add an asset acquisition (purchase). + + Args: + symbol: Asset symbol + acquisition_date: Date of acquisition + quantity: Number of units acquired + cost_per_unit: Cost per unit in acquisition currency + currency: Currency of acquisition + exchange_rate: Exchange rate to AUD (for foreign assets) + incidental_costs: Brokerage and other acquisition costs + asset_type: Type of asset + metadata: Additional data + + Returns: + The created CGTAssetAcquisition record + """ + if quantity <= 0: + raise ValueError("Quantity must be positive") + if cost_per_unit < 0: + raise ValueError("Cost per unit must be non-negative") + + # Calculate total cost in AUD + total_cost = quantity * cost_per_unit + if currency != "AUD": + if exchange_rate is None: + raise ValueError("Exchange rate required for foreign currency acquisitions") + total_cost_aud = total_cost * exchange_rate + else: + total_cost_aud = total_cost + exchange_rate = Decimal("1") + + acquisition = CGTAssetAcquisition( + acquisition_date=acquisition_date, + quantity=quantity, + cost_per_unit=cost_per_unit, + total_cost_aud=total_cost_aud, + currency=currency, + exchange_rate=exchange_rate, + incidental_costs=incidental_costs, + metadata=metadata or {}, + ) + + if symbol not in self._holdings: + self._holdings[symbol] = [] + self._holdings[symbol].append(acquisition) + + # Sort by acquisition date (FIFO order) + self._holdings[symbol].sort(key=lambda x: x.acquisition_date) + + if asset_type: + self._asset_types[symbol] = asset_type + + return acquisition + + def get_holdings(self, symbol: str) -> List[CGTAssetAcquisition]: + """Get current holdings for a symbol. + + Args: + symbol: Asset symbol + + Returns: + List of acquisition parcels (in FIFO order) + """ + return self._holdings.get(symbol, []).copy() + + def get_holding_quantity(self, symbol: str) -> Decimal: + """Get total quantity held for a symbol. + + Args: + symbol: Asset symbol + + Returns: + Total quantity held + """ + holdings = self._holdings.get(symbol, []) + return sum(h.quantity for h in holdings) + + def dispose( + self, + symbol: str, + disposal_date: date, + quantity: Decimal, + proceeds_per_unit: Decimal, + currency: str = "AUD", + exchange_rate: Optional[Decimal] = None, + incidental_costs: Decimal = Decimal("0"), + metadata: Optional[Dict[str, Any]] = None, + ) -> CGTEvent: + """Dispose of an asset (sell). + + Uses FIFO matching to determine cost base. + + Args: + symbol: Asset symbol + disposal_date: Date of disposal + quantity: Number of units disposed + proceeds_per_unit: Proceeds per unit in disposal currency + currency: Currency of proceeds + exchange_rate: Exchange rate to AUD (for foreign currencies) + incidental_costs: Brokerage and other selling costs + metadata: Additional data + + Returns: + The CGT event for this disposal + + Raises: + ValueError: If insufficient holdings for disposal + """ + if quantity <= 0: + raise ValueError("Quantity must be positive") + + holdings = self._holdings.get(symbol, []) + total_held = sum(h.quantity for h in holdings) + + if quantity > total_held: + raise ValueError(f"Insufficient holdings: have {total_held}, trying to dispose {quantity}") + + # Calculate total proceeds in AUD + total_proceeds = quantity * proceeds_per_unit + if currency != "AUD": + if exchange_rate is None: + raise ValueError("Exchange rate required for foreign currency disposals") + total_proceeds_aud = total_proceeds * exchange_rate + else: + total_proceeds_aud = total_proceeds + exchange_rate = Decimal("1") + + # Match acquisitions using FIFO + remaining = quantity + matched: List[Tuple[CGTAssetAcquisition, Decimal]] = [] + total_cost_base = Decimal("0") + weighted_holding_days = 0 + + new_holdings = [] + for acquisition in holdings: + if remaining <= 0: + new_holdings.append(acquisition) + continue + + if acquisition.quantity <= remaining: + # Use entire parcel + matched.append((acquisition, acquisition.quantity)) + total_cost_base += acquisition.total_cost_base + days_held = (disposal_date - acquisition.acquisition_date).days + weighted_holding_days += days_held * int(acquisition.quantity) + remaining -= acquisition.quantity + else: + # Partial use of parcel + matched.append((acquisition, remaining)) + fraction = remaining / acquisition.quantity + total_cost_base += acquisition.total_cost_base * fraction + days_held = (disposal_date - acquisition.acquisition_date).days + weighted_holding_days += days_held * int(remaining) + + # Create remaining parcel + remaining_parcel = CGTAssetAcquisition( + acquisition_date=acquisition.acquisition_date, + quantity=acquisition.quantity - remaining, + cost_per_unit=acquisition.cost_per_unit, + total_cost_aud=acquisition.total_cost_aud * (Decimal("1") - fraction), + currency=acquisition.currency, + exchange_rate=acquisition.exchange_rate, + incidental_costs=acquisition.incidental_costs * (Decimal("1") - fraction), + asset_id=acquisition.asset_id, + metadata=acquisition.metadata, + ) + new_holdings.append(remaining_parcel) + remaining = Decimal("0") + + self._holdings[symbol] = new_holdings + + # Create disposal record + disposal = CGTDisposal( + disposal_date=disposal_date, + symbol=symbol, + quantity=quantity, + proceeds_per_unit=proceeds_per_unit, + total_proceeds_aud=total_proceeds_aud, + currency=currency, + exchange_rate=exchange_rate, + incidental_costs=incidental_costs, + matched_acquisitions=matched, + metadata=metadata or {}, + ) + + # Calculate gain/loss + net_proceeds = disposal.net_proceeds + gross_gain = net_proceeds - total_cost_base + + # Calculate weighted average holding period + # Handle fractional quantities by using float division with rounding + if quantity > 0: + avg_holding_days = int(weighted_holding_days / float(quantity)) + else: + avg_holding_days = 0 + + # Determine if discount eligible + discount_eligible = ( + gross_gain > 0 and + avg_holding_days > self.DISCOUNT_HOLDING_DAYS + ) + + # Calculate discount + if discount_eligible: + discount_amount = gross_gain * self.DISCOUNT_RATE + net_gain = gross_gain - discount_amount + cgt_method = CGTMethod.DISCOUNT + else: + discount_amount = Decimal("0") + net_gain = gross_gain + cgt_method = CGTMethod.OTHER + + # Create CGT event + event = CGTEvent( + event_date=disposal_date, + symbol=symbol, + asset_type=self.get_asset_type(symbol), + disposal=disposal, + gross_gain=gross_gain.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP), + discount_eligible=discount_eligible, + discount_amount=discount_amount.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP), + net_gain=net_gain.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP), + holding_period_days=avg_holding_days, + cgt_method=cgt_method, + tax_year=self.get_tax_year(disposal_date), + metadata=metadata or {}, + ) + + self._events.append(event) + return event + + def get_events(self, tax_year: Optional[int] = None) -> List[CGTEvent]: + """Get CGT events, optionally filtered by tax year. + + Args: + tax_year: Optional tax year to filter by + + Returns: + List of CGT events + """ + if tax_year is None: + return self._events.copy() + return [e for e in self._events if e.tax_year == tax_year] + + def set_carried_forward_losses(self, amount: Decimal) -> None: + """Set carried forward losses from previous years. + + Args: + amount: Loss amount to carry forward (should be positive) + """ + if amount < 0: + raise ValueError("Carried forward losses must be non-negative") + self._carried_forward_losses = amount + + def get_carried_forward_losses(self) -> Decimal: + """Get current carried forward losses.""" + return self._carried_forward_losses + + def calculate_tax_year_summary( + self, + tax_year: int, + apply_carried_losses: bool = True, + ) -> TaxYearSummary: + """Calculate CGT summary for a tax year. + + The ATO rules for applying capital losses: + 1. Capital losses must first be applied to capital gains + 2. Current year losses applied first, then carried forward losses + 3. Losses are applied to discount gains BEFORE the discount + 4. Excess losses are carried forward (never expire) + + Args: + tax_year: Australian tax year (year ending) + apply_carried_losses: Whether to apply carried forward losses + + Returns: + TaxYearSummary with complete CGT calculations + """ + events = self.get_events(tax_year) + start_date, end_date = self.get_tax_year_dates(tax_year) + + # Separate gains and losses + gains = [e for e in events if e.is_gain] + losses = [e for e in events if e.is_loss] + + # Calculate totals (using gross gain for gains before discount) + # Use Decimal("0") as start to ensure result is Decimal, not int + total_gains = sum((e.gross_gain for e in gains), Decimal("0")) + total_losses = abs(sum((e.net_gain for e in losses), Decimal("0"))) + + # Discounted vs non-discounted gains + discounted_gains = sum( + (e.gross_gain for e in gains if e.discount_eligible), Decimal("0") + ) + non_discounted_gains = sum( + (e.gross_gain for e in gains if not e.discount_eligible), Decimal("0") + ) + + # Available losses (current year + carried forward) + carried_forward = self._carried_forward_losses if apply_carried_losses else Decimal("0") + total_available_losses = total_losses + carried_forward + + # Apply losses to gains + # First, apply to non-discounted gains + losses_to_non_discounted = min(total_available_losses, non_discounted_gains) + remaining_losses = total_available_losses - losses_to_non_discounted + + # Then, apply remaining losses to discounted gains (before discount) + losses_to_discounted = min(remaining_losses, discounted_gains) + remaining_losses = remaining_losses - losses_to_discounted + + # Calculate net gains after losses + net_non_discounted = non_discounted_gains - losses_to_non_discounted + net_discounted_before_discount = discounted_gains - losses_to_discounted + + # Apply 50% discount to remaining discounted gains + discount_applied = net_discounted_before_discount * self.DISCOUNT_RATE + net_discounted_after_discount = net_discounted_before_discount - discount_applied + + # Final taxable gain + taxable_gain = net_non_discounted + net_discounted_after_discount + taxable_gain = max(Decimal("0"), taxable_gain) # Can't be negative + + # Calculate losses applied and to carry forward + losses_applied = losses_to_non_discounted + losses_to_discounted + losses_to_carry = remaining_losses + + # Net capital gain (before considering losses for report) + net_capital_gain = total_gains - total_losses + if apply_carried_losses: + net_capital_gain -= carried_forward + + return TaxYearSummary( + tax_year=tax_year, + start_date=start_date, + end_date=end_date, + total_gains=total_gains.quantize(Decimal("0.01")), + total_losses=total_losses.quantize(Decimal("0.01")), + losses_applied=losses_applied.quantize(Decimal("0.01")), + carried_forward_losses=carried_forward.quantize(Decimal("0.01")), + losses_to_carry=losses_to_carry.quantize(Decimal("0.01")), + net_capital_gain=net_capital_gain.quantize(Decimal("0.01")), + discounted_gains=discounted_gains.quantize(Decimal("0.01")), + discount_applied=discount_applied.quantize(Decimal("0.01")), + taxable_gain=taxable_gain.quantize(Decimal("0.01")), + events=events, + ) + + def generate_tax_report( + self, + tax_year: int, + include_details: bool = True, + ) -> Dict[str, Any]: + """Generate a tax report for the ATO. + + Args: + tax_year: Australian tax year (year ending) + include_details: Whether to include detailed transaction list + + Returns: + Dictionary with tax report data + """ + summary = self.calculate_tax_year_summary(tax_year) + + report = { + "tax_year": f"FY{tax_year - 1}-{str(tax_year)[-2:]}", + "period": { + "start": summary.start_date.isoformat(), + "end": summary.end_date.isoformat(), + }, + "summary": { + "total_capital_gains": str(summary.total_gains), + "total_capital_losses": str(summary.total_losses), + "net_capital_gain": str(summary.net_capital_gain), + "discount_method_gains": str(summary.discounted_gains), + "cgt_discount_applied": str(summary.discount_applied), + "prior_year_losses_applied": str(summary.carried_forward_losses), + "current_year_losses_applied": str(summary.losses_applied - summary.carried_forward_losses), + "losses_carried_forward": str(summary.losses_to_carry), + "taxable_capital_gain": str(summary.taxable_gain), + }, + "statistics": { + "number_of_disposals": summary.num_events, + "number_of_gains": summary.num_gains, + "number_of_losses": summary.num_losses, + }, + } + + if include_details: + report["transactions"] = [ + { + "date": event.event_date.isoformat(), + "symbol": event.symbol, + "asset_type": event.asset_type.value, + "quantity": str(event.disposal.quantity), + "proceeds": str(event.disposal.total_proceeds_aud), + "cost_base": str(sum( + acq.total_cost_base * (qty / acq.quantity) + for acq, qty in event.disposal.matched_acquisitions + )), + "gross_gain_loss": str(event.gross_gain), + "holding_period_days": event.holding_period_days, + "discount_eligible": event.discount_eligible, + "discount_amount": str(event.discount_amount), + "net_gain_loss": str(event.net_gain), + } + for event in summary.events + ] + + return report + + def get_unrealised_gains(self, current_prices: Dict[str, Decimal]) -> Dict[str, Dict[str, Decimal]]: + """Calculate unrealised gains for current holdings. + + Args: + current_prices: Dictionary of symbol -> current price in AUD + + Returns: + Dictionary of symbol -> {quantity, cost_base, market_value, unrealised_gain} + """ + result = {} + + for symbol, holdings in self._holdings.items(): + if not holdings: + continue + + total_quantity = sum(h.quantity for h in holdings) + total_cost_base = sum(h.total_cost_base for h in holdings) + + if symbol in current_prices: + market_value = total_quantity * current_prices[symbol] + unrealised_gain = market_value - total_cost_base + else: + market_value = Decimal("0") + unrealised_gain = Decimal("0") + + result[symbol] = { + "quantity": total_quantity, + "cost_base": total_cost_base.quantize(Decimal("0.01")), + "market_value": market_value.quantize(Decimal("0.01")), + "unrealised_gain": unrealised_gain.quantize(Decimal("0.01")), + } + + return result + + def clear(self) -> None: + """Clear all holdings, events, and carried losses.""" + self._holdings.clear() + self._asset_types.clear() + self._events.clear() + self._carried_forward_losses = Decimal("0")