fix: preserve FIFO cost basis through depot transfers

pp_terminal/commands/simulate_share_sell.py

@@ -27,7 +27,7 @@
 import typer
 from pandera.typing import DataFrame
 
-from pp_terminal.data.filters import filter_by_account_and_security, filter_by_security, filter_by_account
+from pp_terminal.data.filters import filter_by_security, filter_by_account
 from pp_terminal.domain.cost_basis import enrich_fifo_lots, finalize_sell_lots
 from pp_terminal.data.tax import load_prepaid_tax_data
 from pp_terminal.domain.sell_strategy import SellStrategy, FixedSharesStrategy, MinTaxStrategy
@@ -85,9 +85,14 @@ def prepare_share_sell_df(  # pylint: disable=too-many-arguments,too-many-positi
         raise InputError(f"No price data for: {', '.join(missing_prices)}")
 
     all_enriched = []
-    for (acc_id, sec_id, _currency), _shares_held in holdings.items():
+    seen_security_ids: set[str] = set()
+    for (_acc_id, sec_id, _currency), _shares_held in holdings.items():
+        if sec_id in seen_security_ids:
+            continue
+        seen_security_ids.add(sec_id)
+        # All accounts needed so cost_basis.py can resolve TRANSFER_OUT → TRANSFER_IN pairs.
         transactions = snapshot.securities_account_transactions.pipe(
-            filter_by_account_and_security, security_id=sec_id, account_id=acc_id
+            filter_by_security, security_id=sec_id
         )
         sale_price = price if price else latest_prices.loc[sec_id]
         enriched = enrich_fifo_lots(

pp_terminal/commands/view_transactions.py

@@ -0,0 +1,121 @@
+"""
+    Copyright (C) 2025-26 Dipl.-Ing. Christoph Massmann <chris@dev-investor.de>
+
+    This file is part of pp-terminal.
+
+    pp-terminal is free software: you can redistribute it and/or modify
+    it under the terms of the GNU General Public License as published by
+    the Free Software Foundation, either version 3 of the License, or
+    (at your option) any later version.
+
+    pp-terminal is distributed in the hope that it will be useful,
+    but WITHOUT ANY WARRANTY; without even the implied warranty of
+    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+    GNU General Public License for more details.
+
+    You should have received a copy of the GNU General Public License
+    along with pp-terminal. If not, see <http://www.gnu.org/licenses/>.
+"""
+
+from datetime import datetime
+from typing import cast
+
+import pandas as pd
+import typer
+from typing_extensions import Annotated
+
+from pp_terminal.data.filters import (
+    filter_by_account, filter_by_security, filter_by_type,
+    filter_earlier_than, filter_later_than,
+)
+from pp_terminal.domain.portfolio import Portfolio, get_security_by_id, resolve_security_id
+from pp_terminal.domain.schemas import TransactionType
+from pp_terminal.exceptions import InputError
+from pp_terminal.output.strategy import OutputStrategy, Console
+from pp_terminal.output.table_decorator import TableOptions
+from pp_terminal.utils.helper import footer
+
+app = typer.Typer()
+console = Console()
+
+_RESULT_COLUMNS = ['date', 'securityName', 'securityId', 'accountId', 'type', 'amount', 'shares', 'currency', 'fees', 'taxes']
+
+
+def prepare_transactions_df(  # pylint: disable=too-many-arguments,too-many-positional-arguments
+        portfolio: Portfolio,
+        security_id: str | None = None,
+        account_id: str | None = None,
+        from_date: datetime | None = None,
+        to_date: datetime | None = None,
+        transaction_types: list[TransactionType] | None = None,
+) -> pd.DataFrame:
+    df = portfolio.securities_account_transactions
+
+    if security_id:
+        df = df.pipe(filter_by_security, security_id=security_id)
+    if account_id:
+        df = df.pipe(filter_by_account, account_id=account_id)
+    if from_date:
+        df = df.pipe(filter_later_than, target_date=from_date)
+    if to_date:
+        df = df.pipe(filter_earlier_than, target_date=to_date)
+    if transaction_types:
+        df = df.pipe(filter_by_type, transaction_types=transaction_types)
+
+    if df.empty:
+        return pd.DataFrame(columns=_RESULT_COLUMNS)
+
+    df = df.reset_index().sort_values('date')
+    df['securityName'] = df['securityId'].map(
+        lambda sid: get_security_by_id(portfolio, sid).name if sid and sid in portfolio.securities.index else ''
+    )
+
+    return df[_RESULT_COLUMNS]
+
+
+def _parse_types(types: list[str] | None) -> list[TransactionType] | None:
+    if not types:
+        return None
+    result = []
+    for t in types:
+        try:
+            result.append(TransactionType[t.upper()])
+        except KeyError as e:
+            valid = [tt.name for tt in TransactionType]
+            raise InputError(f"Unknown transaction type '{t}'. Valid types: {', '.join(valid)}") from e
+    return result
+
+
+@app.command(name="transactions")
+def view_transactions(  # pylint: disable=too-many-arguments,too-many-positional-arguments
+        ctx: typer.Context,
+        security: Annotated[str | None, typer.Argument(help="Security ISIN or ID (defaults to all)")] = None,
+        account_id: Annotated[str | None, typer.Option("--account-id", "-a", help="Account ID filter")] = None,
+        from_date: Annotated[datetime | None, typer.Option("--from", formats=["%Y-%m-%d"], help="Start date (inclusive)")] = None,
+        to_date: Annotated[datetime | None, typer.Option("--to", formats=["%Y-%m-%d"], help="End date (inclusive)")] = None,
+        types: Annotated[list[str] | None, typer.Option("--type", help="Transaction type filter (e.g. BUY, SELL, DIVIDENDS)")] = None,
+) -> None:
+    """Show transactions with optional filters by security, account, date range, or type."""
+    portfolio = cast(Portfolio, ctx.obj.portfolio)
+    output = cast(OutputStrategy, ctx.obj.output)
+
+    security_id = None
+    if security:
+        try:
+            security_id = resolve_security_id(portfolio, security)
+        except InputError as e:
+            raise typer.BadParameter(str(e)) from e
+
+    transaction_types = _parse_types(types)
+
+    df = prepare_transactions_df(portfolio, security_id, account_id, from_date, to_date, transaction_types)
+
+    if df.empty:
+        console.print(output.empty_result())
+        return
+
+    console.print(*output.result_table(
+        df,
+        TableOptions(title="Transactions", show_index=False, show_total=False)
+    ))
+    console.print(output.text(footer()), style="dim")

pp_terminal/domain/cost_basis.py

@@ -19,6 +19,7 @@
 
 from datetime import datetime
 import logging
+from typing import Any
 
 import pandas as pd
 from pandera.typing import DataFrame
@@ -37,7 +38,43 @@ def _filter_purchase_transactions(transactions: DataFrame[TransactionSchema]) ->
     return TransactionSchema.validate(valid_purchases)
 
 
-def _get_remaining_lots_after_fifo_matching(transactions: DataFrame[TransactionSchema]) -> DataFrame[TaxLotSchema]:
+def _build_transfer_in_lookup(transfer_in_transactions: DataFrame[TransactionSchema]) -> dict[tuple[object, ...], str]:
+    lookup: dict[tuple[object, ...], str] = {}
+    for (date, dest_account_id, security_id), row in transfer_in_transactions.iterrows():
+        key = (date, security_id, round(float(row['shares']), 4))
+        lookup[key] = dest_account_id
+    return lookup
+
+
+def _consume_lots_fifo(
+        remaining_lots: list[dict[str, Any]],
+        account_id: str,
+        shares_to_match: float,
+        dest_account_id: str | None,
+) -> tuple[float, list[dict[str, Any]]]:
+    transferred_lots: list[dict[str, Any]] = []
+    for lot in remaining_lots:
+        if shares_to_match <= 0:
+            break
+        if lot['accountId'] != account_id:
+            continue
+        lot_shares = lot['shares']
+        shares_from_lot = min(shares_to_match, lot_shares)
+        new_shares = lot_shares - shares_from_lot
+        shares_to_match -= shares_from_lot
+
+        if dest_account_id and lot_shares > 0:
+            ratio = shares_from_lot / lot_shares
+            transferred_lots.append({**lot, 'accountId': dest_account_id, 'shares': shares_from_lot, 'fees': lot['fees'] * ratio})
+
+        if lot_shares > 0:
+            lot['fees'] = lot['fees'] * (new_shares / lot_shares)
+        lot['shares'] = new_shares
+
+    return shares_to_match, transferred_lots
+
+
+def _get_remaining_lots_after_fifo_matching(transactions: DataFrame[TransactionSchema]) -> DataFrame[TaxLotSchema]:  # pylint: disable=too-many-locals
     """
     Match all sell transactions to purchase lots using FIFO and return remaining lots.
 
@@ -52,6 +89,11 @@ def _get_remaining_lots_after_fifo_matching(transactions: DataFrame[TransactionS
         for schema validation and interface consistency, internally we use list of dicts
         for ~10x faster mutation during the matching algorithm. DataFrame .loc[] access
         has significant overhead that doesn't add value for sequential state updates.
+
+        Depot transfers (TRANSFER_OUT → TRANSFER_IN pairs, as recorded by Portfolio
+        Performance's "Wertpapierübertrag" feature) are handled by moving FIFO lots from the
+        source account to the destination account while preserving the original acquisition
+        cost.
     """
     lots = _filter_purchase_transactions(transactions)
     lots['purchasePrice'] = lots['amount'].abs() / lots['shares']  # save actual market price per share
@@ -62,40 +104,33 @@ def _get_remaining_lots_after_fifo_matching(transactions: DataFrame[TransactionS
         return TaxLotSchema.validate(lots)
 
     sell_transactions = transactions.pipe(filter_by_type, transaction_types=[TransactionType.SELL, TransactionType.DELIVERY_OUTBOUND])
-    if sell_transactions.empty:
-        return TaxLotSchema.validate(lots)
-
-    # Convert to list of dicts for fast mutation during FIFO matching
-    remaining_lots = lots.reset_index().to_dict('records')
-    sales_sorted = sell_transactions.sort_index(level='date')
+    transfer_out_transactions = transactions.pipe(filter_by_type, transaction_types=[TransactionType.TRANSFER_OUT])
+    transfer_in_transactions = transactions.pipe(filter_by_type, transaction_types=[TransactionType.TRANSFER_IN])
 
-    for (_date, account_id, _security_id), row in sales_sorted.iterrows():
-        shares_to_match = float(row['shares'])
+    transfer_in_lookup = _build_transfer_in_lookup(transfer_in_transactions) if not transfer_in_transactions.empty else {}
 
-        # Match against lots in FIFO order (only from same account)
-        for lot in remaining_lots:
-            if shares_to_match <= 0:
-                break
+    outgoing_frames = [f for f in [sell_transactions, transfer_out_transactions] if not f.empty]
+    if not outgoing_frames:
+        return TaxLotSchema.validate(lots)
+    all_outgoing = pd.concat(outgoing_frames).sort_index(level='date') if len(outgoing_frames) > 1 else outgoing_frames[0].sort_index(level='date')
 
-            if lot['accountId'] != account_id:
-                continue
+    remaining_lots = lots.reset_index().to_dict('records')
 
-            # Consume shares from this lot
-            lot_shares = lot['shares']
-            shares_from_lot = min(shares_to_match, lot_shares)
-            new_shares = lot_shares - shares_from_lot
-            shares_to_match -= shares_from_lot
+    for (_date, account_id, _security_id), row in all_outgoing.iterrows():
+        is_transfer_out = str(row['type']) == TransactionType.TRANSFER_OUT.name
+        shares_to_sell = float(row['shares'])
 
-            # Proportionally reduce fees based on remaining shares
-            if lot_shares > 0:
-                lot['fees'] = lot['fees'] * (new_shares / lot_shares)
+        dest_account_id = transfer_in_lookup.get((_date, _security_id, round(shares_to_sell, 4))) if is_transfer_out else None
+        if is_transfer_out and dest_account_id is None:
+            log.warning('No matching TRANSFER_IN for TRANSFER_OUT of %.4f shares of %s on %s — lots will be dropped', shares_to_sell, _security_id, _date)
 
-            lot['shares'] = new_shares
+        unmatched, transferred_lots = _consume_lots_fifo(remaining_lots, account_id, shares_to_sell, dest_account_id)
+        if is_transfer_out:
+            remaining_lots.extend(transferred_lots)
 
-        if shares_to_match > 0.0001:  # Allow small floating point errors
-            log.warning('Sale of %.8f shares for security %s could not be fully matched to purchase lots', shares_to_match, _security_id)
+        if unmatched > 0.0001:  # Allow small floating point errors
+            log.warning('Sale/transfer of %.8f shares for security %s could not be fully matched to purchase lots', unmatched, _security_id)
 
-    # Filter out exhausted lots and convert back to DataFrame
     remaining_lots = [lot for lot in remaining_lots if lot['shares'] > 0.0001]
     if not remaining_lots:
         return TaxLotSchema.empty()

pp_terminal/domain/portfolio.py

@@ -139,6 +139,16 @@ def get_securities_account_by_id(portfolio: Portfolio, account_id: str) -> Accou
     return Account(**account_data)
 
 
+def resolve_security_id(portfolio: Portfolio, security: str) -> str:
+    """Resolve an ISIN or securityId UUID to an internal securityId. Raises InputError if not found."""
+    isin_matches = portfolio.securities.index[portfolio.securities['isin'] == security]
+    if len(isin_matches) == 1:
+        return str(isin_matches[0])
+    if security in portfolio.securities.index:
+        return security
+    raise InputError(f"Security '{security}' not found by ID or ISIN")
+
+
 def get_security_by_id(portfolio: Portfolio, security_id: str) -> Security:
     if security_id not in portfolio.securities.index:
         raise InputError(f"Security '{security_id}' not found in portfolio")

pp_terminal/mcp_server.py

@@ -32,10 +32,11 @@
 from pp_terminal.commands.view_accounts import prepare_accounts_df
 from pp_terminal.commands.view_securities import prepare_securities_df
 from pp_terminal.commands.view_taxonomies import prepare_taxonomies_df
+from pp_terminal.commands.view_transactions import prepare_transactions_df
 from pp_terminal.data.filters import clean_for_display
 from pp_terminal.data.tax import load_prepaid_tax_data
 from pp_terminal.domain.portfolio import Portfolio
-from pp_terminal.domain.schemas import AccountType
+from pp_terminal.domain.schemas import AccountType, TransactionType
 from pp_terminal.data.pp_portfolio_builder import CachedPpPortfolioBuilder
 from pp_terminal.exceptions import InputError
 from pp_terminal.utils.cache import checksum
@@ -191,6 +192,48 @@ def query_accounts(
         df = prepare_accounts_df(portfolio, config, by_date, parsed_type)
         return _clean_records(df.reset_index())
 
+    @mcp.tool()
+    def query_transactions(
+        security: str | None = None,
+        account_id: str | None = None,
+        from_date: str | None = None,
+        to_date: str | None = None,
+        transaction_type: str | None = None,
+    ) -> list[dict[str, Any]]:
+        """List securities account transactions with optional filters.
+
+        Each row is one transaction showing: date, securityName, securityId, accountId,
+        type, amount, shares, currency, fees, taxes.
+
+        Args:
+            security: ISIN or securityId UUID (defaults to all securities)
+            account_id: Filter by account ID
+            from_date: Start date as ISO string, inclusive (e.g. '2023-01-01')
+            to_date: End date as ISO string, inclusive (e.g. '2023-12-31')
+            transaction_type: Transaction type filter (e.g. 'BUY', 'SELL', 'DIVIDENDS', 'TRANSFER_IN')
+        """
+        portfolio = _ensure_fresh_portfolio()
+
+        security_id = _resolve_security(portfolio, security) if security else None
+
+        parsed_types = None
+        if transaction_type:
+            try:
+                parsed_types = [TransactionType[transaction_type.upper()]]
+            except KeyError as e:
+                valid = [tt.name for tt in TransactionType]
+                raise InputError(f"Unknown transaction type '{transaction_type}'. Valid types: {', '.join(valid)}") from e
+
+        df = prepare_transactions_df(
+            portfolio,
+            security_id=security_id,
+            account_id=account_id,
+            from_date=datetime.fromisoformat(from_date) if from_date else None,
+            to_date=datetime.fromisoformat(to_date) if to_date else None,
+            transaction_types=parsed_types,
+        )
+        return _clean_records(df)
+
     @mcp.tool()
     def simulate_vap(
         year: int | None = None,

pyproject.toml

@@ -19,7 +19,7 @@ dependencies = [
     "mcp (>=1.0.0,<2.0.0)",
 ]
 scripts = { pp-terminal = "pp_terminal.main:app" }
-entry-points = { "pp_terminal.commands" = { "view.accounts" = "pp_terminal.commands.view_accounts:app", "view.securities" = "pp_terminal.commands.view_securities:app", "view.taxonomies" = "pp_terminal.commands.view_taxonomies:app", "simulate.vorabpauschale" = "pp_terminal.commands.simulate_vap:app", "simulate.interest" = "pp_terminal.commands.simulate_interest:app", "simulate.share-sell" = "pp_terminal.commands.simulate_share_sell:app", "validate" = "pp_terminal.commands.validate:app", "export" = "pp_terminal.commands.export:app" } }
+entry-points = { "pp_terminal.commands" = { "view.accounts" = "pp_terminal.commands.view_accounts:app", "view.securities" = "pp_terminal.commands.view_securities:app", "view.taxonomies" = "pp_terminal.commands.view_taxonomies:app", "view.transactions" = "pp_terminal.commands.view_transactions:app", "simulate.vorabpauschale" = "pp_terminal.commands.simulate_vap:app", "simulate.interest" = "pp_terminal.commands.simulate_interest:app", "simulate.share-sell" = "pp_terminal.commands.simulate_share_sell:app", "validate" = "pp_terminal.commands.validate:app", "export" = "pp_terminal.commands.export:app" } }
 dynamic = ["version"]
 
 

tests/commands/test_integration.py

@@ -90,6 +90,25 @@ def test_view_securities_csv_output(request: TopRequest) -> None:
     assert result.output == expected_output
 
 
+def test_view_transactions_csv_output(request: TopRequest) -> None:
+    runner = CliRunner()
+    fixtures_dir = request.path.parent.parent / 'fixtures'
+    xml_file = fixtures_dir / 'kommer.ids.xml'
+    golden_file = fixtures_dir / 'expected_view_transactions_kommer.csv'
+
+    result = runner.invoke(app, [
+        '--file', str(xml_file),
+        '--output', 'csv',
+        '--no-cache',
+        'view', 'transactions'
+    ])
+
+    assert result.exit_code == 0, f"Command failed with: {result.output}"
+
+    expected_output = Path(golden_file).read_text(encoding='utf-8')
+    assert result.output == expected_output
+
+
 def test_view_accounts_json_output(request: TopRequest) -> None:
     runner = CliRunner()
     fixtures_dir = request.path.parent.parent / 'fixtures'