diff --git a/jarca_ledger_mcp/application/__init__.py b/jarca_ledger_mcp/application/__init__.py index f256948..2af3e36 100644 --- a/jarca_ledger_mcp/application/__init__.py +++ b/jarca_ledger_mcp/application/__init__.py @@ -9,6 +9,7 @@ from . import categorization_services as _categorization_services from . import cross_transfer_service as _cross_transfer_service from . import imports_service as _imports_service from . import import_transfer_services as _import_transfer_services +from . import monthly_settlement_service as _monthly_settlement_service from . import reconciliation_service as _reconciliation_service from . import shared as shared from . import system_services as _system_services @@ -27,6 +28,7 @@ _EXPORT_MODULES = ( _cross_transfer_service, _imports_service, _import_transfer_services, + _monthly_settlement_service, _reconciliation_service, _system_services, _transactions_write_services, diff --git a/jarca_ledger_mcp/application/monthly_settlement_service.py b/jarca_ledger_mcp/application/monthly_settlement_service.py new file mode 100644 index 0000000..de8c83b --- /dev/null +++ b/jarca_ledger_mcp/application/monthly_settlement_service.py @@ -0,0 +1,113 @@ +from __future__ import annotations + +from .assets_service import get_asset_snapshot_report +from .imports_service import list_ledger_import_batches +from .reconciliation_service import get_balance_reconciliation_report +from .review_decision_service import get_review_operating_indicators +from .shared import Any, fail, normalize_month, normalize_snapshot_at, ok +from .transaction_report_services import get_budget_vs_actual_report, get_monthly_cashflow_report + + +def _month_end_date(month: str) -> str: + return normalize_snapshot_at(month, default_now=False)[:10] + + +def _budget_overrun_count(budget_report: dict[str, Any]) -> int: + return sum(1 for item in budget_report.get("items", []) if item.get("status") == "over") + + +def _unresolved_blocker_summary(review_report: dict[str, Any], reconciliation_report: dict[str, Any]) -> dict[str, Any]: + indicators = review_report.get("indicators", {}) if review_report.get("success") else {} + reconciliation_items = reconciliation_report.get("items", []) if reconciliation_report.get("success") else [] + reconciliation_attention_count = sum(1 for item in reconciliation_items if item.get("status") != "OK") + counts = { + "uncategorized_recommendation_backlog": int(indicators.get("uncategorized_recommendation_backlog") or 0), + "pending_review_count": int(indicators.get("pending_review_count") or 0), + "held_recheck_count": int(indicators.get("held_recheck_count") or 0), + "unreconciled_account_count": int(indicators.get("unreconciled_account_count") or 0), + "reconciliation_attention_count": reconciliation_attention_count, + } + return {"count": sum(counts.values()), "counts": counts} + + +def _source_coverage( + cashflow_report: dict[str, Any], + budget_report: dict[str, Any], + review_report: dict[str, Any], + reconciliation_report: dict[str, Any], + import_report: dict[str, Any], + asset_report: dict[str, Any], +) -> dict[str, Any]: + latest_import = (import_report.get("import_batches") or [None])[0] + return { + "cashflow": {"success": bool(cashflow_report.get("success")), "source": "transactions"}, + "budget": { + "success": bool(budget_report.get("success")), + "budget_count": len(budget_report.get("items", [])), + }, + "review": { + "success": bool(review_report.get("success")), + "indicators": review_report.get("indicators", {}) if review_report.get("success") else {}, + }, + "reconciliation": { + "success": bool(reconciliation_report.get("success")), + "item_count": int(reconciliation_report.get("count") or 0), + }, + "imports": { + "success": bool(import_report.get("success")), + "recent_batch_count": int(import_report.get("count") or 0), + "latest_imported_at": latest_import.get("imported_at") if isinstance(latest_import, dict) else "", + }, + "assets": { + "success": bool(asset_report.get("success")), + "basis_date": asset_report.get("basis_date", ""), + "source": asset_report.get("source", ""), + "is_closed_snapshot": bool(asset_report.get("is_closed_snapshot", False)), + "net_worth": asset_report.get("net_worth"), + }, + } + + +def get_monthly_settlement_summary(month: str = "", db_path: str | None = None) -> dict[str, Any]: + """Return the read-only monthly settlement summary foundation for dashboard/MCP consumers.""" + try: + selected_month = normalize_month(month) + basis_date = _month_end_date(selected_month) + except Exception as exc: + return fail("validation_error", str(exc)) + + cashflow_report = get_monthly_cashflow_report(selected_month, db_path=db_path) + if not cashflow_report.get("success"): + return cashflow_report + budget_report = get_budget_vs_actual_report(selected_month, db_path=db_path) + review_report = get_review_operating_indicators(db_path=db_path) + reconciliation_report = get_balance_reconciliation_report(db_path=db_path) + import_report = list_ledger_import_batches(limit=5, db_path=db_path) + asset_report = get_asset_snapshot_report(basis_date=basis_date, db_path=db_path) + + unresolved_blockers = _unresolved_blocker_summary(review_report, reconciliation_report) + return ok( + selected_month=selected_month, + month=selected_month, + basis_date=basis_date, + actual_income=int(cashflow_report.get("income") or 0), + actual_expense=int(cashflow_report.get("expenses") or 0), + net_cashflow=int(cashflow_report.get("net_cashflow") or 0), + savings_rate=cashflow_report.get("savings_rate"), + budget_overrun_count=_budget_overrun_count(budget_report), + unresolved_blocker_count=int(unresolved_blockers["count"]), + unresolved_blockers=unresolved_blockers, + source_coverage=_source_coverage( + cashflow_report, + budget_report, + review_report, + reconciliation_report, + import_report, + asset_report, + ), + reports={ + "cashflow": cashflow_report, + "budget": budget_report, + }, + read_only=True, + ) diff --git a/jarca_ledger_mcp/service.py b/jarca_ledger_mcp/service.py index 9847313..1788c2d 100644 --- a/jarca_ledger_mcp/service.py +++ b/jarca_ledger_mcp/service.py @@ -12,6 +12,7 @@ from .application import categorization_services as _categorization_services from .application import cross_transfer_service as _cross_transfer_service from .application import imports_service as _imports_service from .application import import_transfer_services as _import_transfer_services +from .application import monthly_settlement_service as _monthly_settlement_service from .application import reconciliation_service as _reconciliation_service from .application import review_decision_service as _review_decision_service from .application import shared as _shared @@ -34,6 +35,7 @@ _PROXY_MODULES = ( _transaction_report_services, _reconciliation_service, _review_decision_service, + _monthly_settlement_service, _imports_service, _import_transfer_services, _cross_transfer_service, diff --git a/tests/test_service.py b/tests/test_service.py index c4be9c8..f3e67c3 100644 --- a/tests/test_service.py +++ b/tests/test_service.py @@ -147,6 +147,38 @@ def test_budget_vs_actual_and_investable_cashflow(tmp_path): assert investable["investable_cashflow"] == 1_250_000 +def test_monthly_settlement_summary_read_model_foundation(tmp_path): + db_path = str(tmp_path / "ledger.sqlite3") + service.create_ledger_budget("2026-06", "식비", 500_000, db_path=db_path) + service.create_ledger_transaction("2026-06-01", "income", 4_000_000, category="수입", db_path=db_path) + service.create_ledger_transaction("2026-06-02", "expense", 550_000, category="식비", db_path=db_path) + service.create_ledger_transaction("2026-06-03", "expense", 150_000, category="", db_path=db_path) + + summary = service.get_monthly_settlement_summary("2026-06", db_path=db_path) + + assert summary["success"] is True + assert summary["selected_month"] == "2026-06" + assert summary["basis_date"] == "2026-06-30" + assert summary["actual_income"] == 4_000_000 + assert summary["actual_expense"] == 550_000 + assert summary["net_cashflow"] == 3_450_000 + assert summary["savings_rate"] == 0.8625 + assert summary["budget_overrun_count"] == 1 + assert summary["unresolved_blocker_count"] >= 1 + assert summary["unresolved_blockers"]["counts"]["uncategorized_recommendation_backlog"] == 1 + assert summary["source_coverage"]["assets"]["basis_date"] == "2026-06-30" + assert summary["read_only"] is True + + +def test_monthly_settlement_summary_rejects_invalid_month(tmp_path): + db_path = str(tmp_path / "ledger.sqlite3") + + summary = service.get_monthly_settlement_summary("2026-99", db_path=db_path) + + assert summary["success"] is False + assert summary["error_code"] == "validation_error" + + def test_duplicate_detection_and_soft_delete(tmp_path): db_path = str(tmp_path / "ledger.sqlite3") first = service.create_ledger_transaction("2026-06-10", "expense", 10_000, merchant="카페", account_id="card1", db_path=db_path)