from __future__ import annotations from decimal import Decimal, InvalidOperation from pathlib import Path from sqlite3 import Connection from jarvis_finance.audit.log import record_audit_event from jarvis_finance.quality.alerts import create_alert from .common import ImportResult, clean, create_import_session, require_text, source_hash, stable_id, utc_now from .dry_run import read_csv_rows from .row_hash import compute_row_hash REQUIRED_COLUMNS = { "transaction_type", "platform_name", "account_name", "trade_date", "quantity", "currency_original", } QUANTITY_REQUIRED_TYPES = {"buy", "partial_sell", "full_sell", "initial_position_snapshot"} SNAPSHOT_TYPES = {"initial_position_snapshot", "initial_cash_snapshot"} def _decimal(value: object) -> Decimal | None: text = clean(value) if text == "": return None try: return Decimal(text) except InvalidOperation as exc: raise ValueError(f"invalid decimal: {text}") from exc def _lookup_account_id(conn: Connection, platform_name: str, account_name: str) -> str | None: row = conn.execute( """ SELECT a.account_id FROM accounts a JOIN platforms p ON p.platform_id = a.platform_id WHERE p.name = ? AND a.account_name = ? """, (platform_name, account_name), ).fetchone() return row["account_id"] if row else None def _lookup_instrument_id(conn: Connection, row: dict[str, str]) -> str | None: isin = clean(row.get("isin")).upper() ticker = clean(row.get("ticker")).upper() name = clean(row.get("name")) if isin: found = conn.execute("SELECT instrument_id FROM instruments WHERE isin = ? LIMIT 1", (isin,)).fetchone() if found: return found["instrument_id"] if ticker: found = conn.execute( "SELECT instrument_id FROM instruments WHERE ticker = ? AND name = ? LIMIT 1", (ticker, name), ).fetchone() if found: return found["instrument_id"] return None def _existing_transaction(conn: Connection, external_transaction_id: str, row_hash: str) -> bool: if external_transaction_id: row = conn.execute( "SELECT 1 FROM transactions WHERE external_transaction_id = ? LIMIT 1", (external_transaction_id,), ).fetchone() if row: return True row = conn.execute("SELECT 1 FROM transactions WHERE row_hash = ? LIMIT 1", (row_hash,)).fetchone() return row is not None def _validate_row(conn: Connection, row: dict[str, str], line: int) -> dict[str, object]: transaction_type = require_text(row, "transaction_type", line).lower() platform_name = require_text(row, "platform_name", line) account_name = require_text(row, "account_name", line) trade_date = require_text(row, "trade_date", line) currency = require_text(row, "currency_original", line).upper() account_id = _lookup_account_id(conn, platform_name, account_name) if not account_id: raise ValueError(f"line {line}: account not found for {platform_name}/{account_name}") instrument_id = _lookup_instrument_id(conn, row) if transaction_type != "initial_cash_snapshot" and not instrument_id: raise ValueError(f"line {line}: instrument not found") quantity = _decimal(row.get("quantity")) if transaction_type in QUANTITY_REQUIRED_TYPES and (quantity is None or quantity <= 0): raise ValueError(f"line {line}: quantity must be > 0 for {transaction_type}") fx_rate = _decimal(row.get("fx_rate_to_chf")) fx_status = "ok" quality_status = "ok" if currency == "CHF": fx_rate = Decimal("1") if fx_rate is None else fx_rate elif fx_rate is None: fx_status = "missing" quality_status = "incomplete" notes = clean(row.get("notes")) if transaction_type in SNAPSHOT_TYPES and "Initial snapshot" not in notes: notes = (notes + " | " if notes else "") + "Initial snapshot" return { "transaction_type": transaction_type, "account_id": account_id, "instrument_id": instrument_id, "trade_date": trade_date, "settlement_date": clean(row.get("settlement_date")) or None, "quantity": quantity, "price_original": _decimal(row.get("price_original")), "gross_amount_original": _decimal(row.get("gross_amount_original")), "fee_original": _decimal(row.get("fee_original")) or Decimal("0"), "tax_original": _decimal(row.get("tax_original")) or Decimal("0"), "net_amount_original": _decimal(row.get("net_amount_original")), "currency_original": currency, "fx_rate_to_chf": fx_rate, "fx_source": clean(row.get("fx_source")) or None, "fx_status": fx_status, "external_transaction_id": clean(row.get("external_transaction_id")), "quality_status": quality_status, "notes": notes, "row_hash": compute_row_hash(row), } def import_transactions_csv( conn: Connection, path: str | Path, *, commit: bool, source_filename: str | None = None, ) -> ImportResult: rows, errors = read_csv_rows(path, REQUIRED_COLUMNS) source_filename = source_filename or Path(path).name rows_new = rows_existing = 0 validated: list[dict[str, object]] = [] seen_hashes: set[str] = set() for line, row in enumerate(rows, start=2): try: item = _validate_row(conn, row, line) row_hash = str(item["row_hash"]) if row_hash in seen_hashes: raise ValueError(f"line {line}: duplicate row hash in source file") seen_hashes.add(row_hash) if _existing_transaction(conn, str(item["external_transaction_id"]), row_hash): rows_existing += 1 else: rows_new += 1 validated.append(item) except ValueError as exc: errors.append(str(exc)) status = "failed" if errors else ("committed" if commit else "dry_run_ok") if commit and not errors: now = utc_now() for item in validated: if _existing_transaction(conn, str(item["external_transaction_id"]), str(item["row_hash"])): continue transaction_id = stable_id("txn", item["external_transaction_id"] or item["row_hash"]) gross = item["gross_amount_original"] fee = item["fee_original"] tax = item["tax_original"] net = item["net_amount_original"] fx_rate = item["fx_rate_to_chf"] gross_chf = Decimal(str(gross)) * Decimal(str(fx_rate)) if gross is not None and fx_rate is not None else None fee_chf = Decimal(str(fee)) * Decimal(str(fx_rate)) if fee is not None and fx_rate is not None else None tax_chf = Decimal(str(tax)) * Decimal(str(fx_rate)) if tax is not None and fx_rate is not None else None net_chf = Decimal(str(net)) * Decimal(str(fx_rate)) if net is not None and fx_rate is not None else None conn.execute( """ INSERT INTO transactions( transaction_id, transaction_type, account_id, instrument_id, trade_date, settlement_date, quantity, price_original, gross_amount_original, fee_original, tax_original, net_amount_original, currency_original, fx_rate_to_chf, fx_source, fx_status, gross_amount_chf, fee_chf, tax_chf, net_amount_chf, source_type, source_id, external_transaction_id, row_hash, is_confirmed, quality_status, notes, created_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 1, ?, ?, ?) """, ( transaction_id, item["transaction_type"], item["account_id"], item["instrument_id"], item["trade_date"], item["settlement_date"], str(item["quantity"]) if item["quantity"] is not None else None, str(item["price_original"]) if item["price_original"] is not None else None, str(gross) if gross is not None else None, str(fee) if fee is not None else None, str(tax) if tax is not None else None, str(net) if net is not None else None, item["currency_original"], str(fx_rate) if fx_rate is not None else None, item["fx_source"], item["fx_status"], str(gross_chf) if gross_chf is not None else None, str(fee_chf) if fee_chf is not None else None, str(tax_chf) if tax_chf is not None else None, str(net_chf) if net_chf is not None else None, "csv", source_filename, item["external_transaction_id"], item["row_hash"], item["quality_status"], item["notes"], now, ), ) record_audit_event( conn, source="csv_import", action=str(item["transaction_type"]), entity_type="transaction", entity_id=transaction_id, new_values={"transaction_type": item["transaction_type"], "row_hash": item["row_hash"]}, user_text_note=str(item["notes"] or "CSV import"), confirmed=True, created_by="importer", ) if item["fx_status"] == "missing": create_alert( conn, priority="kritisch", category="fx", entity_type="transaction", entity_id=transaction_id, rule_id="missing_fx", message="FX-Kurs fehlt für importierte Fremdwährungstransaktion.", evidence={"currency": item["currency_original"], "trade_date": item["trade_date"]}, ) session_id = create_import_session( conn, import_type="transactions", source_filename=source_filename, file_hash=source_hash(path), status=status, rows_total=len(rows), rows_imported=rows_new if commit and not errors else 0, rows_failed=len(errors), errors=errors, notes="dry run" if not commit else None, ) return ImportResult(session_id, "transactions", status, len(rows), rows_new if not errors else 0, rows_existing, len(errors), errors)