216 lines
11 KiB
Python
216 lines
11 KiB
Python
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")
|