gitea-impl-agent/jarca-ledger-mcp
1
0
Fork 0
forked from jsfamily/jarca-ledger-mcp
Code Pull Requests Activity

feat: complete dcy-015 monthly settlement dashboard

Browse Source 
Merge dcy-015 monthly settlement dashboard delivery after local and GitHub Actions verification.
This commit is contained in:
genichin
2026-07-12 20:16:36 +09:00
committed by GitHub
parent a4a913cdd6
commit 835728793e
171 changed files with 12231 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dashboard/app.py
+234 -2
View File
@@ -953,6 +953,236 @@ def category_rule_preview_dialog(rule: dict[str, Any], db_path: str) -> None:
else:
st.info("이 규칙에 부합하는 거래가 없습니다.")
def page_monthly_settlement(db_path: str) -> None:
render_page_header(
"월정산",
"선택한 월을 기준으로 월말 결산 상태와 미해결 블로커를 빠르게 점검합니다.",
kicker="🧾 월말 점검",
pills=["read-model 기반", "마감 전 점검", "fixture-safe"],
)
query_month = st.query_params.get("month", current_kst_month())
if isinstance(query_month, list):
query_month = query_month[0]
month = st.text_input("대상 월", value=str(query_month), key="monthly_settlement_month", help="YYYY-MM 형식")
settlement = overview_gateway.get_monthly_settlement_summary(month, db_path=db_path)
if not settlement.get("success"):
st.error(settlement.get("error", settlement))
st.caption(f"입력한 월: {month}")
return
selected_month = settlement.get("month") or settlement.get("selected_month") or month
basis_date = settlement.get("basis_date") or f"{selected_month}-01"
source_coverage = settlement.get("source_coverage", {}) if isinstance(settlement.get("source_coverage"), dict) else {}
unresolved_blockers = settlement.get("unresolved_blockers", {}) if isinstance(settlement.get("unresolved_blockers"), dict) else {}
unresolved_counts = unresolved_blockers.get("counts", {}) if isinstance(unresolved_blockers.get("counts"), dict) else {}
review_queue_res = overview_gateway.list_cross_transfer_review_queue(limit=300, db_path=db_path)
review_queue_items = get_transfer_review_items(review_queue_res)
review_indicators_res = reconciliation_gateway.get_review_operating_indicators(db_path=db_path)
review_indicators = review_indicators_res.get("indicators", {}) if review_indicators_res.get("success") else {}
reconciliation_res = reconciliation_gateway.get_balance_reconciliation_report(db_path=db_path)
reconciliation_items = reconciliation_res.get("items", []) if reconciliation_res.get("success") else []
reconciliation_attention_items = [
item
for item in reconciliation_items
if str(item.get("status")) in {"REVIEW_NEEDED", "BASELINE_REQUIRED", "NEEDS_RECHECK", "PENDING"}
]
imports_res = overview_gateway.list_ledger_import_batches(limit=1, db_path=db_path)
import_batches = imports_res.get("import_batches", []) if imports_res.get("success") else []
asset_snapshot = overview_gateway.get_asset_snapshot_report(basis_date=basis_date, db_path=db_path)
read_only_stamp = datetime.now(ZoneInfo("Asia/Seoul")).strftime("%Y-%m-%d %H:%M:%S KST")
render_section_label("📘", "월정산 셸")
st.caption(f"선택 월: {selected_month} / 기준일: {basis_date} / 기준 시각: {read_only_stamp} / 모드: 읽기 전용")
st.markdown(
"<span style='display:inline-block; padding:0.15rem 0.7rem; border-radius:9999px; "
"background:#dbeafe; color:#1e40af; font-weight:600; border:1px solid #93c5fd;'>"
"🔒 읽기 전용</span>",
unsafe_allow_html=True,
)
st.info("거래/예산 수정은 각 운영 화면(거래내역, 요약)에서 처리하세요.")
st.markdown("- [거래내역으로 이동](?page=거래내역)")
st.markdown("- [예산 확인으로 이동](?page=요약)")
st.markdown("월간 핵심 KPI")
st.caption("월간 핵심 지표를 통해 수입·지출·잔고 흐름을 점검하세요.")
k1, k2, k3, k4 = st.columns(4)
k1.metric("실제 수입", format_krw(settlement.get("actual_income", 0)))
k2.metric("실제 지출", format_krw(settlement.get("actual_expense", 0)))
k3.metric("순현금흐름", format_krw(settlement.get("net_cashflow", 0)))
k4.metric("저축률", format_percent(settlement.get("savings_rate")))
st.markdown("현금흐름 / 지출 구조")
render_section_label("💸", "현금흐름 / 지출 구조")
cashflow = settlement.get("reports", {}).get("cashflow", {}) if isinstance(settlement.get("reports"), dict) else {}
if cashflow:
c1, c2 = st.columns(2)
by_type = cashflow.get("by_type", {}) or {}
by_type_df = pd.DataFrame(
[{"구분": str(key), "금액": int(value)} for key, value in by_type.items()]
)
if not by_type_df.empty:
by_type_df["금액"] = by_type_df["금액"].map(format_krw)
by_type_df = by_type_df.sort_values("금액", ascending=False)
c1.dataframe(by_type_df, width="stretch", height=170)
by_kind = cashflow.get("by_cashflow_kind", {}) or {}
by_kind_df = pd.DataFrame(
[{"현금흐름 구분": str(key), "금액": int(value)} for key, value in by_kind.items()]
)
if not by_kind_df.empty:
by_kind_df["금액"] = by_kind_df["금액"].map(format_krw)
c2.dataframe(by_kind_df, width="stretch", height=170)
else:
st.info("해당 월 현금흐름 정보가 아직 없습니다.")
st.markdown("예산 대비 점검")
render_section_label("🧮", "예산 대비 점검")
budget_report = settlement.get("reports", {}).get("budget", {}) if isinstance(settlement.get("reports"), dict) else {}
b1, b2, b3 = st.columns(3)
b1.metric("예산 항목 수", f"{int(budget_report.get('count', 0))}건")
b2.metric("월 예산", format_krw(budget_report.get("total_budget", 0)))
b3.metric("예산 초과 항목", f"{int(settlement.get('budget_overrun_count', 0))}건")
budget_items = budget_report.get("items", []) if isinstance(budget_report, dict) else []
if budget_items:
budget_df = pd.DataFrame(budget_items)
budget_df["예산"] = budget_df["budget"].map(format_krw)
budget_df["실제"] = budget_df["actual"].map(format_krw)
budget_df["잔여"] = budget_df["remaining"].map(format_krw)
budget_df["사용률"] = budget_df["usage_rate"].map(format_percent)
st.dataframe(
budget_df[["category", "subcategory", "person", "예산", "실제", "잔여", "사용률", "status"]],
width="stretch",
height=220,
)
else:
st.info(f"{selected_month} 예산 항목이 없습니다.")
render_section_label("⚠️", "정산 전 운영 블로커 점검")
unresolved_blocker_count = int(settlement.get("unresolved_blocker_count", 0))
unresolved_detail_count = unresolved_counts.get("unresolved_count", unresolved_blocker_count)
st.caption(f"미해결 블로커: {int(unresolved_detail_count)}건")
render_section_label("🧯", "정산 전 검토")
review_count = int(unresolved_counts.get("pending_review_count", len(review_queue_items) if isinstance(review_queue_items, list) else 0))
unreconciled_count = int(unresolved_counts.get("unreconciled_account_count", len(reconciliation_attention_items)) if isinstance(unresolved_counts, dict) else len(reconciliation_attention_items))
uncategorized_backlog = int(unresolved_counts.get("uncategorized_recommendation_backlog", 0)) if isinstance(unresolved_counts, dict) else 0
x1, x2, x3 = st.columns(3)
x1.metric("미처리 review", f"{review_count}건")
x2.metric("미대조 계좌", f"{unreconciled_count}건")
x3.metric("미분류 추천 대기", f"{uncategorized_backlog}건")
if review_queue_items:
st.markdown("확인필요 큐 최근 항목")
for item in review_queue_items[:5]:
tx_date = item.get("transaction_date") or item.get("reviewed_at") or "-"
merchant = item.get("merchant") or item.get("counterparty") or "-"
amount = format_krw(item.get("amount") if item.get("amount") is not None else item.get("difference"))
status = item.get("review_status") or item.get("status") or "pending"
st.markdown(f"- {tx_date} · {merchant} · {amount} · {status}")
else:
st.caption("확인필요 큐는 비어 있습니다.")
st.markdown("- [확인필요 큐로 이동](?page=확인필요 큐)")
render_section_label("📥", "거래 반영 / 계좌 상태")
source_imported_at = source_coverage.get("imports", {}).get("latest_imported_at")
t1, t2, t3 = st.columns(3)
t1.metric("최근 Import 건수", f"{int(source_coverage.get('imports', {}).get('recent_batch_count', 0))}건")
t2.metric("최근 Import 시각", source_imported_at or "없음")
t3.metric("보류 후 처리", str(review_indicators.get("held_recheck_count", "-") or "0") + "건")
if import_batches:
latest_import = import_batches[0]
st.caption(f"최근 Import: 출처 {latest_import.get('source', '알 수 없음')} / 수집일 {latest_import.get('imported_at', '미기록')}")
else:
st.info("최근 Import 배치가 없습니다.")
st.markdown("- [계좌 화면에서 잔액대조 확인](?page=계좌)")
st.markdown("- [Import 화면에서 이력 확인](?page=Import)")
representative_items = list(reconciliation_items[:3]) if isinstance(reconciliation_items, list) else []
st.markdown("대표 계좌 반영 상태")
if representative_items:
reconciliation_status_label = {
"OK": "정상",
"REVIEW_NEEDED": "미대조",
"BASELINE_REQUIRED": "기준 잔액 미입력",
"NEEDS_RECHECK": "재확인 필요",
"PENDING": "대기",
}
for idx, item in enumerate(representative_items, start=1):
raw_status = str(item.get("status", "-") or "-")
synced_at = item.get("last_synced_at") or item.get("baseline_at") or item.get("calculation_basis_at") or "-"
baseline_balance = item.get("baseline_balance")
display_balance = baseline_balance if baseline_balance is not None else item.get("calculated_balance")
item_line = (
f"- [{idx}] 계좌명: {item.get('account_name', '-')}, "
f"기관: {item.get('institution', item.get('bank', '미기록'))}, "
f"유형: {item.get('account_type', item.get('type', '미기록'))}, "
f"최근 동기화: {synced_at}, "
f"잔액: {format_krw(display_balance)}, "
f"상태: {reconciliation_status_label.get(raw_status, raw_status)}"
)
st.markdown(
item_line,
)
else:
st.caption("잔액대조 대상 계좌가 현재 없습니다.")
st.markdown("- [미분류 페이지](?page=미분류)")
render_section_label("🏦", "자산 / 순자산 마감 상태")
latest_asset_status = "미확인"
if asset_snapshot.get("success"):
latest_asset_status = "마감 완료" if bool(asset_snapshot.get("is_closed_snapshot")) else "미마감"
warnings = asset_snapshot.get("warnings") if isinstance(asset_snapshot, dict) else []
missing_valuation_items = asset_snapshot.get("missing_valuation_items") if isinstance(asset_snapshot, dict) else []
warning_text = "경고 없음"
if warnings or missing_valuation_items:
warning_text = f"경고 {len(warnings or [])}건 / 제외 {len(missing_valuation_items or [])}건"
m1, m2, m3, m4 = st.columns(4)
m1.metric("자산 합계", format_krw(asset_snapshot.get("total_assets") if isinstance(asset_snapshot, dict) else source_coverage.get("assets", {}).get("total_assets")))
m2.metric("총부채", format_krw(asset_snapshot.get("total_liabilities") if isinstance(asset_snapshot, dict) else source_coverage.get("assets", {}).get("total_liabilities")))
m3.metric("순자산", format_krw(
asset_snapshot.get("net_worth")
if isinstance(asset_snapshot, dict)
else source_coverage.get("assets", {}).get("net_worth")
))
m4.metric("자산 마감 상태", latest_asset_status)
st.caption(f"자산 상세: {warning_text}")
st.caption(f"마감 기준일: {asset_snapshot.get('basis_date', basis_date)}")
st.markdown("- [자산/마감 화면에서 스냅샷 확인](?page=자산/마감)")
render_section_label("📜", "월정산 결과 / Evidence")
st.markdown("### 읽기 전용 근거 요약")
evidence_close_status = "마감 완료" if bool(asset_snapshot.get("is_closed_snapshot")) else "미마감"
import_summary = source_coverage.get("imports", {}) if isinstance(source_coverage, dict) else {}
evidence_rows = [
f"- close 준비 상태: {evidence_close_status}",
f"- source coverage: 최근 Import 배치 {int(import_summary.get('recent_batch_count', 0))}건",
f"- 최근 Import: {import_summary.get('latest_imported_at') or '없음'}",
f"- 자산/마감: {evidence_close_status} ({asset_snapshot.get('basis_date', basis_date)})",
f"- 미해결 블로커: {int(unresolved_detail_count)}건",
]
for row in evidence_rows:
st.markdown(row)
st.markdown("### 다음 액션")
st.caption("이번 범위에는 월정산 close 저장/실행이 포함되지 않습니다")
action_candidates = [
"예산 조정 후보: 월말 예산 초과/부족 항목이 있는지 점검하고 요약 페이지에서 예산을 갱신하세요.",
"분류/검토 규칙 후보: 미분류 추천 및 확인필요 큐에서 규칙 후보를 먼저 점검하세요.",
"대조/Import/백업 후속 작업: 계좌 잔액대조, 최근 Import 이력, 백업 최신화 상태를 함께 확인하세요.",
"수기 자산/마감 화면에서 기준일·평가 상태를 정합성 확인 후 운영 승인 여부를 결정하세요.",
]
for action in action_candidates:
st.markdown(f"- {action}")
st.divider()
st.markdown("- [요약 페이지로 이동](?page=요약)")
def page_overview(db_path: str) -> None:
health = overview_gateway.health_check(db_path)
render_page_header(
@@ -1839,7 +2069,7 @@ def main() -> None:
db_path = get_db_path()
inject_app_css()
pages = ["요약", "자산/마감", "거래내역", "확인필요 큐", "미분류", "분류 규칙", "카테고리", "계좌", "Import"]
pages = ["월정산", "요약", "자산/마감", "거래내역", "확인필요 큐", "미분류", "분류 규칙", "카테고리", "계좌", "Import"]
query_page = st.query_params.get("page", "요약")
page = query_page if query_page in pages else "요약"
st.session_state["page"] = page
@@ -1853,7 +2083,9 @@ def main() -> None:
clear_cache()
st.rerun()
if page == "요약":
if page == "월정산":
page_monthly_settlement(db_path)
elif page == "요약":
page_overview(db_path)
elif page == "자산/마감":
page_assets(db_path)