Fix investment schema compatibility

AGENTS.md

@@ -23,6 +23,10 @@ Use four-space indentation, double-quoted strings, and Ruff's 88-character line
 
 Use pytest and favor focused unit tests with row dictionaries or temporary SQLite databases. Integration tests are opt-in: set `MONEYWIZ_TEST_DB_PATH` to an absolute path to a disposable MoneyWiz database, then run `.venv/bin/python -m pytest tests`. They skip when the variable is unset. Do not use a live personal database. CI runs unit tests, Ruff, and mypy; no minimum coverage threshold is configured.
 
+When automatic schema resolution is added or migrated, preserve caller-supplied aliases and converters, resolve inherited effective specifications rather than own-class defaults, and cover explicit aliases, inherited fields, partial consumer customization, and record construction. Restrict automatic row-evidence resolution to consumers that require it.
+
+Treat record identity admission as part of recoverable row processing: preflight every ID and GID collision before changing either lookup map, then perform insertion inside the row-recovery boundary. Cover duplicate identities, unchanged first-record lookups, and later valid-row recovery.
+
 ## Commit & Pull Request Guidelines
 
 Recent commits use short subjects such as `Fix schema profile validation and record construction`; capitalization and prefixes vary. Write a brief imperative subject that names the change. In pull requests, explain behavior changes, list the checks run, and link a relevant issue when one exists. Include a CLI example when changing user-visible shell behavior.

README.md

@@ -86,7 +86,13 @@ profile.validate()
 ```
 
 `SchemaProfileResolver(db_path, baseline=None).resolve()` enriches a profile
-with database-specific tag-join table details. When no baseline is provided, it
+with database-specific tag-join table details and investment column choices.
+Explicit baseline alias overrides are preserved; automatic detection applies to
+investment fields that retain the default aliases.
+Holdings and buy/sell transactions are resolved separately using physical
+columns and entity-specific row evidence. If populated investment entities have
+no safe column choice, resolution raises `UnsupportedInvestmentSchemaError`
+before records are loaded. When no baseline is provided, it
 uses `DEFAULT_SCHEMA_PROFILE`. The resolved profile can be passed to `MoneywizApi` or the lower-level
 `DatabaseAccessor`:
 
@@ -101,6 +107,10 @@ a raw row, reports field-resolution failures together, and creates the model
 instance. Record
 classes receive mapped field values and do not depend on `SchemaProfile`.
 
+Managers skip malformed rows and report them through `manager.load_errors`,
+which returns a copy of the skipped records' IDs, model names, and error types.
+Unsupported database schemas stop loading instead of being treated as row errors.
+
 ## Tests
 
 Run unit tests without a database:

src/moneywiz_api/__init__.py

@@ -3,11 +3,13 @@
 from moneywiz_api.schema.schema_profile_resolver import (
     DEFAULT_SCHEMA_PROFILE,
     SchemaProfileResolver,
+    UnsupportedInvestmentSchemaError,
 )
 
 __all__ = [
     "MoneywizApi",
     "SchemaProfile",
     "SchemaProfileResolver",
+    "UnsupportedInvestmentSchemaError",
     "DEFAULT_SCHEMA_PROFILE",
 ]

src/moneywiz_api/database_accessor.py

@@ -2,7 +2,7 @@
 from collections import defaultdict
 from decimal import Decimal
 from pathlib import Path
-from typing import Any, Dict, List, Tuple
+from typing import Any, Callable, Dict, List, Tuple, cast
 
 from moneywiz_api.schema.raw_data_handler import RawDataHandler as RDH
 from moneywiz_api.model.record import Record
@@ -59,6 +59,13 @@ def ent_for(self, typename: str) -> ENT_ID:
         assert ent_id is not None, f"Unknown typename {typename}"
         return ent_id
 
+    def _table_exists(self, table_name: str) -> bool:
+        row = self._con.execute(
+            "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?",
+            (table_name,),
+        ).fetchone()
+        return row is not None
+
     def query_objects(self, typenames: List[str]) -> List[Any]:
         cur = self._con.cursor()
         res = cur.execute(
@@ -70,7 +77,12 @@ def query_objects(self, typenames: List[str]) -> List[Any]:
         )
         return res.fetchall()
 
-    def get_record(self, pk_id: ID, constructor: type[Record] = Record):
+    def _construct_record(self, row, constructor: Callable):
+        if isinstance(constructor, type) and issubclass(constructor, Record):
+            return self.schema_profile.create_record(row, constructor)
+        return cast(Callable, constructor)(row)
+
+    def get_record(self, pk_id: ID, constructor: Callable = Record):
         cur = self._con.cursor()
         res = cur.execute(
             """
@@ -80,9 +92,9 @@ def get_record(self, pk_id: ID, constructor: type[Record] = Record):
             [pk_id],
         )
 
-        return self.schema_profile.create_record(res.fetchone(), constructor)
+        return self._construct_record(res.fetchone(), constructor)
 
-    def get_record_by_gid(self, gid: GID, constructor: type[Record] = Record):
+    def get_record_by_gid(self, gid: GID, constructor: Callable = Record):
         cur = self._con.cursor()
         res = cur.execute(
             """
@@ -92,10 +104,12 @@ def get_record_by_gid(self, gid: GID, constructor: type[Record] = Record):
             [gid],
         )
 
-        return self.schema_profile.create_record(res.fetchone(), constructor)
+        return self._construct_record(res.fetchone(), constructor)
 
     def get_category_assignment(self) -> Dict[ID, List[Tuple[ID, Decimal]]]:
         transaction_map: Dict[ID, List[Tuple[ID, Decimal]]] = defaultdict(list)
+        if not self._table_exists("ZCATEGORYASSIGMENT"):
+            return transaction_map
         cur = self._con.cursor()
         res = cur.execute(
             """
@@ -111,6 +125,8 @@ def get_category_assignment(self) -> Dict[ID, List[Tuple[ID, Decimal]]]:
 
     def get_refund_maps(self) -> Dict[ID, ID]:
         refund_to_withdraw: Dict[ID, ID] = {}
+        if not self._table_exists("ZWITHDRAWREFUNDTRANSACTIONLINK"):
+            return refund_to_withdraw
         cur = self._con.cursor()
         res = cur.execute(
             """

src/moneywiz_api/managers/record_manager.py

@@ -1,17 +1,21 @@
 from abc import ABC, abstractmethod
+import logging
 from typing import Dict, Generic, TypeVar
 
 from moneywiz_api.database_accessor import DatabaseAccessor
 from moneywiz_api.model.record import Record
+from moneywiz_api.schema.schema_profile_resolver import UnsupportedInvestmentSchemaError
 from moneywiz_api.types import GID, ID
 
 T = TypeVar("T", bound=Record)
+logger = logging.getLogger(__name__)
 
 
 class RecordManager(ABC, Generic[T]):
     def __init__(self) -> None:
         self._records: Dict[ID, T] = {}
         self._gid_to_id: Dict[GID, ID] = {}
+        self._load_errors: list[tuple[ID, str, str]] = []
 
     @property
     @abstractmethod
@@ -23,22 +27,36 @@ def load(self, db_accessor: DatabaseAccessor) -> None:
 
         for record in records:
             typename = db_accessor.typename_for(record["Z_ENT"])
-            assert typename in self.ents, (
-                f"Unknown typename {typename} for record {record}"
-            )
-
-            model_cls = self.ents[typename]
-            obj = db_accessor.schema_profile.create_record(record, model_cls)
-            obj.validate()
-            self.add(obj)
+            if typename in self.ents:
+                try:
+                    obj = db_accessor.schema_profile.create_record(
+                        record, self.ents[typename]
+                    )
+                    obj.validate()
+                    self.add(obj)
+                except UnsupportedInvestmentSchemaError:
+                    raise
+                except (AssertionError, KeyError, RuntimeError, ValueError) as exc:
+                    record_id = record.get("Z_PK")
+                    detail = type(exc).__name__
+                    self._load_errors.append((record_id, typename, detail))
+                    logger.debug(
+                        "Skipping unreadable %s record %s: %s",
+                        typename,
+                        record_id,
+                        detail,
+                    )
+                    continue
 
     def add(self, record: T) -> None:
-        self._records[record.id] = record
+        if record.id in self._records:
+            raise RuntimeError(f"Duplicate id {record.id}")
         if record.gid in self._gid_to_id:
             raise RuntimeError(
                 f"Duplicate gid for {record}, existing record Id {self._gid_to_id[record.gid]}"
             )
 
+        self._records[record.id] = record
         self._gid_to_id[record.gid] = record.id
 
     def get(self, record_id: ID) -> T | None:
@@ -53,5 +71,10 @@ def get_by_gid(self, gid: GID) -> T | None:
     def records(self) -> Dict[ID, T]:
         return self._records
 
+    @property
+    def load_errors(self) -> list[tuple[ID, str, str]]:
+        """Return records skipped during best-effort read parsing."""
+        return list(self._load_errors)
+
     def __repr__(self):
         return "\n".join(f"{key}: {value}" for key, value in self.records().items())

src/moneywiz_api/model/transaction.py

@@ -336,6 +336,8 @@ def __init__(self, row, field_values):
         super().__init__(row, field_values)
 
         # Fixes
+        if self.original_exchange_rate == 0:
+            raise ValueError("cannot read a transfer with a zero exchange rate")
         if (
             self.recipient_amount in (None, Decimal(0))
             and self.original_amount is not None
@@ -344,6 +346,11 @@ def __init__(self, row, field_values):
             self.recipient_amount = -self.original_amount * self.original_exchange_rate
         if self.recipient_amount is not None:
             self.recipient_amount = abs(self.recipient_amount)
+        if self.original_amount == 0 and self.recipient_amount not in (
+            None,
+            Decimal(0),
+        ):
+            self.original_amount = self.amount
 
     def validate(self) -> None:
         super().validate()

src/moneywiz_api/schema/raw_data_handler.py

@@ -30,10 +30,13 @@ def get_decimal(raw_value: Any) -> Decimal:
     @staticmethod
     def filter_row(row: Dict[str, Any]) -> Dict[str, Any]:
         copy = {k: v for k, v in row.items()}
-        del copy["ZMANUALHISTORICALPRICESPERSHARE"]
-        del copy["ZIMPORTLINKIDARRAY2"]
-        del copy["ZIMPORTLINKIDARRAY"]
-        del copy["ZBANKLOGOPRIMARYCOLOR"]
+        for key in (
+            "ZMANUALHISTORICALPRICESPERSHARE",
+            "ZIMPORTLINKIDARRAY2",
+            "ZIMPORTLINKIDARRAY",
+            "ZBANKLOGOPRIMARYCOLOR",
+        ):
+            copy.pop(key, None)
         return {
             k: v
             for k, v in copy.items()