from __future__ import annotations from datetime import date from .shared import ( Any, BALANCE_RECONCILIATION_REFERENCE_RATIO, BALANCE_RECONCILIATION_TOLERANCE_KRW, RECONCILIATION_TARGET_STATUSES, read_db, read_operation, fail, normalize_date, normalize_snapshot_at, now_iso, ok, row_to_dict, sqlite3, ) from .review_decision_service import _fetch_latest_decisions def _transaction_signed_amount(tx_type: str, amount: int) -> int: if tx_type in {"income", "refund", "adjustment"}: return int(amount) if tx_type in {"expense", "transfer"}: return -int(amount) return 0 def _account_balance_row(conn: sqlite3.Connection, account_id: str, baseline_at: str, opening_evidence: bool = False) -> dict[str, Any]: date_cutoff = normalize_date(baseline_at[:10]) rows = conn.execute( """SELECT type, amount, balance_after, transaction_date, transaction_datetime, created_at FROM transactions WHERE status='active' AND account_id=? AND transaction_date<=? ORDER BY transaction_date ASC, transaction_datetime ASC, created_at ASC""", (account_id, date_cutoff), ).fetchall() signed_total = sum(_transaction_signed_amount(row["type"], int(row["amount"] or 0)) for row in rows) latest_balance_row = next((row for row in reversed(rows) if row["balance_after"] is not None), None) if latest_balance_row is not None: calculated_balance = int(latest_balance_row["balance_after"]) basis = "latest_transaction_balance_after" basis_at = latest_balance_row["transaction_datetime"] or latest_balance_row["transaction_date"] else: calculated_balance = int(signed_total) basis = "net_transaction_delta_without_opening_balance" basis_at = baseline_at has_transaction_coverage = bool( rows and rows[-1]["transaction_date"] == date_cutoff and all(row["balance_after"] is not None for row in rows) ) return { "calculated_balance": calculated_balance, "calculation_basis": basis, "calculation_basis_at": basis_at, "transaction_count": len(rows), "signed_transaction_total": int(signed_total), "calculation_complete": has_transaction_coverage and ( opening_evidence or (len(rows) == 1 and rows[0]["transaction_date"] == date_cutoff) ), } @read_operation def get_balance_reconciliation_report(account_id: str = "", baseline_balance: int | None = None, baseline_at: str = "", source_note: str = "", memo: str = "", tolerance: int = BALANCE_RECONCILIATION_TOLERANCE_KRW, db_path: str | None = None) -> dict[str, Any]: """Preview manual/partial-import account balance reconciliation without mutating transactions.""" try: baseline_at_norm = normalize_snapshot_at(baseline_at.strip(), default_now=False) if baseline_at and baseline_at.strip() else None tolerance_int = int(tolerance) if tolerance_int < 0: raise ValueError("tolerance must be non-negative") baseline_balance_int = int(baseline_balance) if baseline_balance is not None else None except Exception as e: return fail("validation_error", str(e)) with read_db(db_path) as conn: clauses = ["active=1"] params: list[Any] = [] if account_id: clauses.append("id=?") params.append(account_id) else: placeholders = ",".join("?" for _ in RECONCILIATION_TARGET_STATUSES) clauses.append(f"statement_status IN ({placeholders})") params.extend(sorted(RECONCILIATION_TARGET_STATUSES)) accounts = conn.execute(f"SELECT * FROM accounts WHERE {' AND '.join(clauses)} ORDER BY name", params).fetchall() account_targets = [("reconciliation_account", account["id"]) for account in accounts] latest_decisions = _fetch_latest_decisions(conn, account_targets) session_rows = conn.execute("SELECT * FROM reconciliation_sessions ORDER BY basis_date DESC, updated_at DESC").fetchall() latest_sessions: dict[str, dict[str, Any]] = {} for row in session_rows: session = row_to_dict(row) latest_sessions.setdefault(session["account_id"], session) items: list[dict[str, Any]] = [] for account in accounts: effective_baseline = baseline_balance_int if baseline_balance_int is not None else account["last_balance"] effective_at = baseline_at_norm or account["last_synced_at"] or now_iso() valuation = None ambiguous_valuation = False valuation_rows = conn.execute( """SELECT * FROM account_valuations v WHERE v.account_id=? AND v.verification_status='verified' AND UPPER(v.currency)=UPPER(?) AND v.basis_date<=? AND NOT EXISTS ( SELECT 1 FROM account_valuations child WHERE child.supersedes_id=v.id ) ORDER BY v.basis_date DESC, v.recorded_at DESC, v.id DESC""", (account["id"], account["currency"] or "KRW", effective_at[:10]), ).fetchall() purposes = ( ("loan_balance", "loan", "balance") if account["type"] == "loan" else (("total_value", "balance", "cash_balance") if account["type"] == "investment" else ("cash_balance", "balance")) ) transaction_rows = conn.execute( """SELECT transaction_date FROM transactions WHERE status='active' AND account_id=? AND transaction_date<=? ORDER BY transaction_date ASC, transaction_datetime ASC, created_at ASC""", (account["id"], effective_at[:10]), ).fetchall() first_transaction_date = transaction_rows[0]["transaction_date"] if transaction_rows else None has_opening_evidence = bool( first_transaction_date and any( row["basis_date"] < first_transaction_date and row["valuation_kind"] in purposes and ( account["type"] != "loan" or row["representative_liability_definition"] in {"principal_only", "principal_interest_fees"} ) for row in valuation_rows ) ) calc = _account_balance_row(conn, account["id"], effective_at, has_opening_evidence) for purpose in purposes: scoped = [ row for row in valuation_rows if row["valuation_kind"] == purpose and ( account["type"] != "loan" or row["representative_liability_definition"] in {"principal_only", "principal_interest_fees"} ) ] if not scoped: continue latest_basis = max(row["basis_date"] for row in scoped) latest = [row for row in scoped if row["basis_date"] == latest_basis] if len(latest) != 1: ambiguous_valuation = True else: valuation = latest[0] break if effective_baseline is None: difference = None ratio = None status = "BASELINE_REQUIRED" else: difference = int(effective_baseline) - int(calc["calculated_balance"]) ratio = abs(difference) / abs(int(effective_baseline)) if int(effective_baseline) else None status = "OK" if abs(difference) <= tolerance_int else "REVIEW_NEEDED" if account["type"] == "investment" and ambiguous_valuation: status = "AMBIGUOUS_VALUATION" difference = None elif account["type"] == "investment" and valuation is None: status = "VALUATION_UNAVAILABLE" difference = None elif not calc["calculation_complete"]: status = "INCOMPLETE_CALCULATION" difference = None elif account["type"] == "investment" and valuation is not None: status = "VALUATION_MANAGED" difference = None items.append({ "account_id": account["id"], "account_name": account["name"], "statement_status": account["statement_status"], "is_reconciliation_target": account["statement_status"] in RECONCILIATION_TARGET_STATUSES, "baseline_balance": effective_baseline, "baseline_at": effective_at, "calculated_balance": calc["calculated_balance"], "difference": difference, "tolerance": tolerance_int, "reference_ratio_threshold": BALANCE_RECONCILIATION_REFERENCE_RATIO, "difference_ratio": ratio, "status": status, "external_confirmed_value": int(valuation["amount"]) if valuation is not None else None, "valuation_id": valuation["id"] if valuation is not None else "", "valuation_kind": valuation["valuation_kind"] if valuation is not None else "", "provenance": { "valuation_date": valuation["basis_date"] if valuation is not None else "", "age_days": (date.fromisoformat(effective_at[:10]) - date.fromisoformat(valuation["basis_date"])).days if valuation is not None else None, "carried": bool(valuation is not None and valuation["basis_date"] != effective_at[:10]), "verification_status": valuation["verification_status"] if valuation is not None else "unavailable", "evidence_kind": "account_valuation" if valuation is not None else "none", "warnings": [status] if status not in {"OK", "REVIEW_NEEDED", "VALUATION_MANAGED"} else [], "exclusions": [], "snapshot_run_id": "", "close_status": "", "selection_policy": "as_of", }, "recommended_decisions": ["계좌이동", "실지출", "실수입", "투자이동", "분석제외", "확인필요"] if status == "REVIEW_NEEDED" else [], "reconciliation_session": latest_sessions.get(account["id"]), "review_status": latest_decisions.get(("reconciliation_account", account["id"]), {}).get("status", "pending"), "review_decision": latest_decisions.get(("reconciliation_account", account["id"])), "source_note": source_note, "memo": memo, **calc, }) return ok(items=items, count=len(items), tolerance=tolerance_int, baseline_at=baseline_at_norm, routing="preview_only_manual_confirmation_required")