Add basic support for cash merger transactions (Schwab only) (#532)

* Add simpler cash merger transactions (schwab support only)

* Fix warning

---------

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

cgt_calc/main.py

@@ -206,7 +206,7 @@ def convert_to_hmrc_transactions(
             ]:
                 new_balance += get_amount_or_fail(transaction)
                 self.add_acquisition(portfolio, acquisition_list, transaction)
-            elif transaction.action is ActionType.SELL:
+            elif transaction.action in [ActionType.SELL, ActionType.CASH_MERGER]:
                 amount = get_amount_or_fail(transaction)
                 new_balance += amount
                 self.add_disposal(portfolio, disposal_list, transaction)

cgt_calc/model.py

@@ -42,6 +42,7 @@ class ActionType(Enum):
     REINVEST_DIVIDENDS = 13
     WIRE_FUNDS_RECEIVED = 14
     STOCK_SPLIT = 15
+    CASH_MERGER = 16
 
 
 @dataclass

cgt_calc/parsers/schwab.py

@@ -127,6 +127,9 @@ def action_from_str(label: str) -> ActionType:
     if label == "Stock Split":
         return ActionType.STOCK_SPLIT
 
+    if label in ["Cash Merger", "Cash Merger Adj"]:
+        return ActionType.CASH_MERGER
+
     raise ParsingError("schwab transactions", f"Unknown action: {label}")
 
 
@@ -224,6 +227,40 @@ def create(
         return transaction
 
 
+def _unify_schwab_cash_merger_trxs(
+    transactions: list[SchwabTransaction],
+) -> list[SchwabTransaction]:
+    filtered: list[SchwabTransaction] = []
+    for transaction in transactions:
+        if transaction.raw_action == "Cash Merger Adj":
+            assert (
+                len(filtered) > 0
+            ), "Cash Merger Adj must be precedeed by a Cash Merger transaction"
+            assert filtered[-1].raw_action == "Cash Merger"
+            assert filtered[-1].description == transaction.description
+            assert filtered[-1].symbol == transaction.symbol
+            assert filtered[-1].date == transaction.date
+            assert filtered[-1].quantity is None
+            assert filtered[-1].price is None
+            assert filtered[-1].amount is not None
+            assert transaction.amount is None
+            assert transaction.quantity is not None
+            # the quantity is negative but
+            # because we store it as a 'sell' we need it positive
+            filtered[-1].quantity = -1 * transaction.quantity
+            filtered[-1].price = filtered[-1].amount / filtered[-1].quantity
+            filtered[-1].fees += transaction.fees
+            print(
+                "WARNING: Cash Merger support is not complete and doesn't cover the "
+                "cases when shares are received aside from cash,  "
+                "please review this transaction carefully: "
+                f"{filtered[-1]}"
+            )
+        else:
+            filtered.append(transaction)
+    return filtered
+
+
 def read_schwab_transactions(
     transactions_file: str, schwab_award_transactions_file: str | None
 ) -> list[BrokerTransaction]:
@@ -252,6 +289,7 @@ def read_schwab_transactions(
                 )
                 for row in lines
             ]
+            transactions = _unify_schwab_cash_merger_trxs(transactions)
             transactions.reverse()
             return list(transactions)
     except FileNotFoundError:

tests/test_data/schwab_cash_merger/expected_output.txt

@@ -0,0 +1,30 @@
+WARNING: No schwab award file provided
+WARNING: Cash Merger support is not complete and doesn't cover the cases when shares are received aside from cash,  please review this transaction carefully: SchwabTransaction(date=datetime.date(2021, 3, 2), action=<ActionType.CASH_MERGER: 16>, symbol='FOO', description='FOO INC', quantity=Decimal('100'), price=Decimal('10'), fees=Decimal('0'), amount=Decimal('1000'), currency='USD', broker='Charles Schwab')
+INFO: No schwab Equity Award JSON file provided
+INFO: No trading212 folder provided
+INFO: No mssb folder provided
+INFO: No sharesight file provided
+INFO: No raw file provided
+First pass completed
+Final portfolio:
+Final balance:
+  Charles Schwab: 1000.00 (USD)
+Dividends: £0.00
+Dividend taxes: £0.00
+Interest: £0.00
+Disposal proceeds: £711.69
+
+
+Second pass completed
+Portfolio at the end of 2020/2021 tax year:
+For tax year 2020/2021:
+Number of disposals: 1
+Disposal proceeds: £711.69
+Allowable costs: £1783.50
+Capital gain: £0.00
+Capital loss: £1071.81
+Total capital gain: £-1071.81
+Taxable capital gain: £0
+
+Generate calculations report
+All done!

tests/test_data/schwab_cash_merger/transactions.csv

@@ -0,0 +1,5 @@
+Date,Action,Symbol,Description,Price,Quantity,Fees & Comm,Amount
+03/02/2021,Cash Merger,FOO,FOO INC,,,,$1000
+03/02/2021,Cash Merger Adj,FOO,FOO INC,,-100,,
+03/02/2021,Buy,FOO,FOO INC,$25,100,$6,-$2506
+03/01/2016,MoneyLink Transfer,,Tfr BANK,,,,$2506.00

tests/test_schwab_cash_merger.py

@@ -0,0 +1,31 @@
+"""Test Schwab Cash Merger support."""
+
+from pathlib import Path
+import subprocess
+import sys
+
+
+def test_run_with_schwab_cash_merger_files() -> None:
+    """Runs the script and verifies it doesn't fail."""
+    cmd = [
+        sys.executable,
+        "-m",
+        "cgt_calc.main",
+        "--year",
+        "2020",
+        "--schwab",
+        "tests/test_data/schwab_cash_merger/transactions.csv",
+        "--no-pdflatex",
+    ]
+    result = subprocess.run(cmd, check=True, capture_output=True)
+    assert result.stderr == b"", "Run with example files generated errors"
+    expected_file = (
+        Path("tests") / "test_data" / "schwab_cash_merger" / "expected_output.txt"
+    )
+    expected = expected_file.read_text()
+    cmd_str = " ".join([param if param else "''" for param in cmd])
+    assert result.stdout.decode("utf-8") == expected, (
+        "Run with example files generated unexpected outputs, "
+        "if you added new features update the test with:\n"
+        f"{cmd_str} > {expected_file}"
+    )