WIP: Add tax calculator

src/investir/investir.py

@@ -6,6 +6,7 @@
 from .config import Config
 from .logging import setup_logging
 from .parser.factory import ParserFactory
+from .taxcalculator import TaxCalculator
 from .transaction import Acquisition, Disposal
 from .trhistory import TrHistory
 
@@ -93,6 +94,13 @@ def create_interest_command(subparser, parent_parser) -> None:
         parents=[parent_parser])
 
 
+def create_tax_command(subparser, parent_parser) -> None:
+    subparser.add_parser(
+        'tax',
+        help='show tax report',
+        parents=[parent_parser])
+
+
 def main() -> None:
     parser = argparse.ArgumentParser()
 
@@ -134,6 +142,7 @@ def main() -> None:
     create_dividends_command(subparser, parent_parser)
     create_transfers_command(subparser, parent_parser)
     create_interest_command(subparser, parent_parser)
+    create_tax_command(subparser, parent_parser)
 
     args = parser.parse_args()
 
@@ -175,3 +184,6 @@ def main() -> None:
         tr_hist.show_transfers(filters)
     elif args.command == 'interest':
         tr_hist.show_interest(filters)
+    elif args.command == 'tax':
+        calculator = TaxCalculator(tr_hist)
+        calculator.calculate_cgt_tax()

src/investir/taxcalculator.py

@@ -0,0 +1,126 @@
+import logging
+
+from abc import ABC, abstractmethod
+from collections import defaultdict
+from dataclasses import dataclass
+from decimal import Decimal
+from enum import Enum
+from datetime import timedelta
+from itertools import groupby
+
+from .transaction import Order, Acquisition, Disposal
+from .trhistory import TrHistory
+
+logger = logging.getLogger(__name__)
+
+
+class DisposalType(Enum):
+    SAME_DAY = 1
+    BED_AND_BREAKFAST = 2
+    SECTION_104 = 3
+
+
+@dataclass
+class CapitalTaxEvent:
+    acquisition_cost: Decimal
+    acquisition_fees: Decimal
+    disposal: Disposal
+    disposal_type: DisposalType
+
+
+class ShareMatcher(ABC):
+    @abstractmethod
+    def rule(self) -> DisposalType:
+        pass
+
+    @staticmethod
+    @abstractmethod
+    def match(ord1: Acquisition, ord2: Disposal) -> bool:
+        pass
+
+
+class SameDayMatcher(ShareMatcher):
+    def rule(self) -> DisposalType:
+        return DisposalType.SAME_DAY
+
+    @staticmethod
+    def match(ord1: Acquisition, ord2: Disposal) -> bool:
+        return ord1.date == ord2.date
+
+
+class BedAndBreakfastMatcher(ShareMatcher):
+    def rule(self) -> DisposalType:
+        return DisposalType.BED_AND_BREAKFAST
+
+    @staticmethod
+    def match(ord1: Acquisition, ord2: Disposal) -> bool:
+        return (ord2.date < ord1.date
+                <= ord2.date() + timedelta(days=30))
+
+
+class TaxCalculator:
+    def __init__(self, tr_hist: TrHistory) -> None:
+        self._tr_hist = tr_hist
+        self._acquisitions: dict[str, list[Acquisition]] = defaultdict(list)
+        self._disposals: dict[str, list[Disposal]] = defaultdict(list)
+        self._tax_events: dict[int, list[CapitalTaxEvent]] = defaultdict(list)
+
+    def calculate_cgt_tax(self) -> None:
+        self._prepare()
+
+        for ticker in self._disposals.keys():
+            self._match_shares(ticker, SameDayMatcher())
+
+        for tax_event in self._tax_events.values():
+            print(tax_event)
+
+    def _prepare(self) -> None:
+        groups = groupby(
+            self._tr_hist.orders(),
+            key=lambda o: (o.ticker, type(o), o.date))
+
+        for key, group in groups:
+            ticker = key[0]
+            pooled_order = Order.pool(*group)
+            if isinstance(pooled_order, Acquisition):
+                self._acquisitions[ticker].append(pooled_order)
+            elif isinstance(pooled_order, Disposal):
+                self._disposals[ticker].append(pooled_order)
+
+            # logging.debug('New order created: %s', pooled_order)
+
+    def _match_shares(self, ticker: str, matcher: ShareMatcher):
+        acquisits = self._acquisitions[ticker]
+        disposals = self._disposals[ticker]
+
+        a_idx = 0
+        d_idx = 0
+
+        while d_idx < len(disposals):
+            a = acquisits[a_idx]
+            d = disposals[d_idx]
+
+            if not matcher.match(a, d):
+                break
+
+            if a.quantity > d.quantity:
+                a_matched, a_remaining = a.split(d.quantity)
+                acquisits[a_idx] = a_remaining
+                d_matched = d
+                d_idx += 1
+            elif d.quantity > a.quantity:
+                d_matched, d_remaining = d.split(a.quantity)
+                disposals[d_idx] = d_remaining
+                a_matched = a
+                a_idx += 1
+            else:
+                del acquisits[a_idx]
+                del disposals[d_idx]
+
+            tax_event = CapitalTaxEvent(
+                a_matched.total_cost,
+                a_matched.fees,
+                d_matched,
+                matcher.rule())
+
+            self._tax_events[d.tax_year()].append(tax_event)