For schwab we can now use the awards centre report to calculate the s…

…tock price of the Stock Activities. (#104)

Also fixed a double conversion to gbp bug in the stock activity calculation

Co-authored-by: Ruslan Sayfutdinov <[email protected]>

.gitignore

@@ -2,6 +2,7 @@
 *.log
 .DS_Store
 .mypy_cache/
+.vscode
 __pycache__
 env
 dist

cgt_calc/args_parser.py

@@ -50,6 +50,13 @@ def create_parser() -> argparse.ArgumentParser:
         nargs="?",
         help="monthly GBP/USD prices from HMRC",
     )
+    parser.add_argument(
+        "--schwab-award",
+        type=str,
+        default=None,
+        nargs="?",
+        help="file containing schwab award data for stock prices",
+    )
     parser.add_argument(
         "--initial-prices",
         type=str,

cgt_calc/exceptions.py

@@ -68,6 +68,14 @@ def __init__(self, row: list[str], count: int, file: str):
         )
 
 
+class UnexpectedRowCountError(ParsingError):
+    """Unexpected row error."""
+
+    def __init__(self, count: int, file: str):
+        """Initialise."""
+        super().__init__(file, f"The following file doesn't have {count} rows:")
+
+
 class CalculatedAmountDiscrepancyError(InvalidTransactionError):
     """Calculated amount discrepancy error."""
 

cgt_calc/main.py

@@ -46,6 +46,14 @@
 LOGGER = logging.getLogger(__name__)
 
 
+def get_amount_or_fail(transaction: BrokerTransaction) -> Decimal:
+    """Return the transaction amount or throw an error."""
+    amount = transaction.amount
+    if amount is None:
+        raise AmountMissingError(transaction)
+    return amount
+
+
 class CapitalGainsCalculator:
     """Main calculator class."""
 
@@ -75,6 +83,7 @@ def add_acquisition(
         """Add new acquisition to the given list."""
         symbol = transaction.symbol
         quantity = transaction.quantity
+        price = transaction.price
         if symbol is None:
             raise SymbolMissingError(transaction)
         if quantity is None or quantity <= 0:
@@ -84,29 +93,22 @@ def add_acquisition(
             portfolio[symbol] += quantity
         else:
             portfolio[symbol] = quantity
+
         # Add to acquisition_list to apply same day rule
         if transaction.action in [ActionType.STOCK_ACTIVITY, ActionType.SPIN_OFF]:
-            stock_price_gbp = None
-
-            if transaction.price is not None and transaction.currency is not None:
-                stock_price_gbp = self.converter.to_gbp(
-                    transaction.price, transaction.currency, transaction.date
-                )
-            else:
-                stock_price_gbp = self.initial_prices.get(transaction.date, symbol)
-
-            amount = quantity * stock_price_gbp
+            if price is None:
+                price = self.initial_prices.get(transaction.date, symbol)
+            amount = round_decimal(quantity * price, 2)
         else:
-            if transaction.amount is None:
-                raise AmountMissingError(transaction)
-            if transaction.price is None:
+            if price is None:
                 raise PriceMissingError(transaction)
-            calculated_amount = round_decimal(
-                quantity * transaction.price + transaction.fees, 2
-            )
-            if transaction.amount != -calculated_amount:
+
+            amount = get_amount_or_fail(transaction)
+            calculated_amount = round_decimal(quantity * price + transaction.fees, 2)
+            if amount != -calculated_amount:
                 raise CalculatedAmountDiscrepancyError(transaction, -calculated_amount)
-            amount = -transaction.amount
+            amount = -amount
+
         add_to_list(
             acquisition_list,
             transaction.date,
@@ -143,14 +145,13 @@ def add_disposal(
         if portfolio[symbol] == 0:
             del portfolio[symbol]
         # Add to disposal_list to apply same day rule
-        if transaction.amount is None:
-            raise AmountMissingError(transaction)
-        if transaction.price is None:
+
+        amount = get_amount_or_fail(transaction)
+        price = transaction.price
+
+        if price is None:
             raise PriceMissingError(transaction)
-        amount = transaction.amount
-        calculated_amount = round_decimal(
-            quantity * transaction.price - transaction.fees, 2
-        )
+        calculated_amount = round_decimal(quantity * price - transaction.fees, 2)
         if amount != calculated_amount:
             raise CalculatedAmountDiscrepancyError(transaction, calculated_amount)
         add_to_list(
@@ -180,28 +181,20 @@ def convert_to_hmrc_transactions(
         for i, transaction in enumerate(transactions):
             new_balance = balance[(transaction.broker, transaction.currency)]
             if transaction.action is ActionType.TRANSFER:
-                if transaction.amount is None:
-                    raise AmountMissingError(transaction)
-                new_balance += transaction.amount
+                new_balance += get_amount_or_fail(transaction)
             elif transaction.action is ActionType.BUY:
-                if transaction.amount is None:
-                    raise AmountMissingError(transaction)
-                new_balance += transaction.amount
+                new_balance += get_amount_or_fail(transaction)
                 self.add_acquisition(portfolio, acquisition_list, transaction)
             elif transaction.action is ActionType.SELL:
-                if transaction.amount is None:
-                    raise AmountMissingError(transaction)
-                new_balance += transaction.amount
+                amount = get_amount_or_fail(transaction)
+                new_balance += amount
                 self.add_disposal(portfolio, disposal_list, transaction)
                 if self.date_in_tax_year(transaction.date):
-                    total_sells += self.converter.to_gbp_for(
-                        transaction.amount, transaction
-                    )
+                    total_sells += self.converter.to_gbp_for(amount, transaction)
             elif transaction.action is ActionType.FEE:
-                if transaction.amount is None:
-                    raise AmountMissingError(transaction)
-                new_balance += transaction.amount
-                transaction.fees = -transaction.amount
+                amount = get_amount_or_fail(transaction)
+                new_balance += amount
+                transaction.fees = -amount
                 transaction.quantity = Decimal(0)
                 gbp_fees = self.converter.to_gbp_for(transaction.fees, transaction)
                 if transaction.symbol is None:
@@ -217,29 +210,20 @@ def convert_to_hmrc_transactions(
             elif transaction.action in [ActionType.STOCK_ACTIVITY, ActionType.SPIN_OFF]:
                 self.add_acquisition(portfolio, acquisition_list, transaction)
             elif transaction.action in [ActionType.DIVIDEND, ActionType.CAPITAL_GAIN]:
-                if transaction.amount is None:
-                    raise AmountMissingError(transaction)
-                new_balance += transaction.amount
+                amount = get_amount_or_fail(transaction)
+                new_balance += amount
                 if self.date_in_tax_year(transaction.date):
-                    dividends += self.converter.to_gbp_for(
-                        transaction.amount, transaction
-                    )
+                    dividends += self.converter.to_gbp_for(amount, transaction)
             elif transaction.action in [ActionType.TAX, ActionType.ADJUSTMENT]:
-                if transaction.amount is None:
-                    raise AmountMissingError(transaction)
-                new_balance += transaction.amount
+                amount = get_amount_or_fail(transaction)
+                new_balance += amount
                 if self.date_in_tax_year(transaction.date):
-                    dividends_tax += self.converter.to_gbp_for(
-                        transaction.amount, transaction
-                    )
+                    dividends_tax += self.converter.to_gbp_for(amount, transaction)
             elif transaction.action is ActionType.INTEREST:
-                if transaction.amount is None:
-                    raise AmountMissingError(transaction)
-                new_balance += transaction.amount
+                amount = get_amount_or_fail(transaction)
+                new_balance += amount
                 if self.date_in_tax_year(transaction.date):
-                    interest += self.converter.to_gbp_for(
-                        transaction.amount, transaction
-                    )
+                    interest += self.converter.to_gbp_for(amount, transaction)
             else:
                 raise InvalidTransactionError(
                     transaction, f"Action not processed({transaction.action})"
@@ -656,7 +640,7 @@ def main() -> int:
 
     # Read data from input files
     broker_transactions = read_broker_transactions(
-        args.schwab, args.trading212, args.mssb
+        args.schwab, args.schwab_award, args.trading212, args.mssb
     )
     converter = CurrencyConverter(read_gbp_prices_history(args.gbp_history))
     initial_prices = InitialPrices(read_initial_prices(args.initial_prices))

cgt_calc/model.py

@@ -125,6 +125,10 @@ def taxable_gain(self) -> Decimal:
         assert self.capital_gain_allowance is not None
         return max(Decimal(0), self.total_gain() - self.capital_gain_allowance)
 
+    def __repr__(self) -> str:
+        """Return string representation."""
+        return f"<CalculationEntry: {str(self)}>"
+
     def __str__(self) -> str:
         """Return string representation."""
         out = f"Portfolio at the end of {self.tax_year}/{self.tax_year + 1} tax year:\n"

cgt_calc/parsers/__init__.py

@@ -42,13 +42,16 @@ def __str__(self) -> str:
 
 def read_broker_transactions(
     schwab_transactions_file: str | None,
+    schwab_awards_transactions_file: str | None,
     trading212_transactions_folder: str | None,
     mssb_transactions_folder: str | None,
 ) -> list[BrokerTransaction]:
     """Read transactions for all brokers."""
     transactions = []
     if schwab_transactions_file is not None:
-        transactions += read_schwab_transactions(schwab_transactions_file)
+        transactions += read_schwab_transactions(
+            schwab_transactions_file, schwab_awards_transactions_file
+        )
     else:
         print("WARNING: No schwab file provided")