from __future__ import annotations import json import hashlib from .account_valuation_service import _public_candidate from .shared import ( ASSET_BUCKETS, ASSET_BUCKET_ORDER, Any, LIABILITY_BUCKETS, current_kst_date, date, db, fail, new_id, normalize_date, normalize_liquidity_bucket, now_iso, ok, row_to_dict, sqlite3, ) def _valuation_policy() -> dict[str, Any]: return { "authority_sources": ["accounts.last_balance", "manual_assets"], "snapshot_role": "historical_record_layer", "formula": "asset_buckets - loan", "asset_buckets": ASSET_BUCKETS, "liability_buckets": LIABILITY_BUCKETS, "missing_valuation_policy": "exclude_without_transaction_inference", } def _evidence_state_hash(item: dict[str, Any]) -> str: """Bind idempotency to evidence state without exposing sensitive text.""" fields = ( "source_type", "source_id", "name", "amount", "currency", "liquidity_bucket", "authority_source", "valuation_id", "valuation_date", "verification_status", "source_note", "memo", "active", "carried_forward", ) payload = {key: item.get(key) for key in fields} return hashlib.sha256( json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode() ).hexdigest() def _account_valuation_items(conn: sqlite3.Connection, basis_date: str = "", selection_policy: str = "exact") -> tuple[list[dict[str, Any]], list[dict[str, Any]], list[dict[str, Any]]]: rows = conn.execute("SELECT * FROM accounts WHERE active=1 ORDER BY name").fetchall() included: list[dict[str, Any]] = [] excluded: list[dict[str, Any]] = [] invalid: list[dict[str, Any]] = [] for row in rows: account = row_to_dict(row) account_type = str(account.get("type") or "other").lower() raw_bucket = str(account.get("liquidity_bucket") or "other") try: bucket = normalize_liquidity_bucket(raw_bucket) except ValueError: bucket = "other" invalid.append({"id": account.get("id", ""), "name": account.get("name", ""), "liquidity_bucket": raw_bucket}) if account_type == "loan": bucket = "loan" base = { "id": account.get("id", ""), "source_id": account.get("id", ""), "source_type": "account", "authority_source": "accounts.last_balance", "name": account.get("name", ""), "type": account.get("type", ""), "liquidity_bucket": bucket, "valuation_date": account.get("last_synced_at", ""), "last_synced_at": account.get("last_synced_at", ""), "source_note": account.get("memo", ""), "memo": account.get("memo", ""), } balance = account.get("last_balance") if basis_date: valuation_rows = conn.execute( "SELECT * FROM account_valuations WHERE account_id=? AND currency=? ORDER BY basis_date DESC, recorded_at DESC, id DESC", (account["id"], account.get("currency") or "KRW"), ).fetchall() leaves = [v for v in valuation_rows if conn.execute("SELECT 1 FROM account_valuations WHERE supersedes_id=?", (v["id"],)).fetchone() is None] if account_type == "investment": # Older imports recorded investment balances as cash_balance; # keep that as a compatibility fallback after the canonical # total_value/balance purposes. purposes = ("total_value", "balance", "cash_balance") elif account_type == "loan": purposes = ("loan_balance", "loan", "balance") else: purposes = ("cash_balance", "balance") selected = None selected_purpose = None candidate_rows = [] for purpose in purposes: scoped = [v for v in leaves if v["valuation_kind"] == purpose] if selection_policy == "exact": verified = [v for v in scoped if v["verification_status"] == "verified" and v["basis_date"] == basis_date] else: eligible = [v for v in scoped if v["verification_status"] == "verified" and v["basis_date"] <= basis_date] latest = max((v["basis_date"] for v in eligible), default="") verified = [v for v in eligible if v["basis_date"] == latest] if len(verified) > 1: invalid.append({**base, "reason": "ambiguous_valuation", "valuation_kind": purpose}) verified = [] break if verified: selected, selected_purpose = verified[0], purpose break candidate_rows.extend(v for v in scoped if v["verification_status"] != "verified" and v["basis_date"] <= basis_date and (v["basis_date"] == basis_date if selection_policy == "exact" else True)) if selected is None: excluded.append({**base, "reason": "missing_verified_exact_valuation" if selection_policy == "exact" else "missing_verified_valuation", "unverified_candidates": [_public_candidate(v) for v in candidate_rows]}) continue valuation = selected if account_type == "loan" and ( selected_purpose not in {"loan", "loan_balance"} or not valuation["representative_liability_definition"] ): excluded.append({ **base, "reason": "missing_representative_liability_definition", "valuation_id": valuation["id"], "valuation_kind": selected_purpose, "valuation_date": valuation["basis_date"], "verification_status": valuation["verification_status"], }) continue balance = valuation["amount"] base.update({"valuation_id": valuation["id"], "verification_status": "verified", "valuation_kind": selected_purpose, "valuation_date": valuation["basis_date"], "currency": valuation["currency"], "carried_forward": valuation["basis_date"] != basis_date}) if candidate_rows: base["unverified_candidates"] = [_public_candidate(v) for v in candidate_rows if v["basis_date"] <= basis_date] if balance is None: excluded.append({**base, "reason": "missing_last_balance"}) continue amount = int(balance) included.append({**base, "amount": amount, "last_balance": amount}) return included, excluded, invalid def _manual_asset_valuation_items(conn: sqlite3.Connection, active_only: bool = True, basis_date: str = "", selection_policy: str = "exact") -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: where = "WHERE active=1" if active_only else "" rows = conn.execute(f"SELECT * FROM manual_assets {where} ORDER BY name").fetchall() included: list[dict[str, Any]] = [] excluded: list[dict[str, Any]] = [] for row in rows: asset = row_to_dict(row) if (asset.get("currency") or "KRW").upper() != "KRW": excluded.append({"id": asset.get("id", ""), "source_id": asset.get("id", ""), "source_type": "manual_asset", "reason": "non_krw_without_fx", "currency": asset.get("currency")}) continue try: bucket = normalize_liquidity_bucket(asset.get("liquidity_bucket")) except ValueError: bucket = "other" base = { "id": asset.get("id", ""), "source_id": asset.get("id", ""), "source_type": "manual_asset", "authority_source": "manual_assets", "name": asset.get("name", ""), "liquidity_bucket": bucket, "valuation_date": asset.get("valuation_date", ""), "source_note": asset.get("source_note", ""), "memo": asset.get("memo", ""), "active": asset.get("active", True), } value = asset.get("valuation_amount") if basis_date: valuation_date = str(asset.get("valuation_date") or "")[:10] if not valuation_date or valuation_date > basis_date or (selection_policy == "exact" and valuation_date != basis_date): excluded.append({**base, "reason": "missing_exact_manual_asset_valuation" if selection_policy == "exact" else "future_or_unknown_manual_asset_valuation"}) continue base["carried_forward"] = valuation_date != basis_date if value is None: excluded.append({**base, "reason": "missing_valuation_amount"}) continue included.append({**base, "amount": int(value), "valuation_amount": int(value)}) return included, excluded def _snapshot_rows_for_latest_run(conn: sqlite3.Connection, basis_date: str) -> tuple[list[sqlite3.Row], dict[str, Any]]: rows = conn.execute("SELECT rowid AS _rowid, * FROM asset_snapshots WHERE basis_date=? ORDER BY created_at ASC, name ASC", (basis_date,)).fetchall() run = conn.execute("SELECT * FROM asset_snapshot_runs WHERE basis_date=? ORDER BY rowid DESC LIMIT 1", (basis_date,)).fetchone() if not run: return rows, {"snapshot_run_id": "", "is_correction": False, "correction_of_run_id": "", "correction_reason": "", "total_snapshot_rows_for_basis_date": len(rows)} if rows else {} selected = [row for row in rows if row["snapshot_run_id"] == run["id"]] metadata = {"snapshot_run_id": run["id"], "is_correction": bool(run["correction_of_run_id"]), "correction_of_run_id": run["correction_of_run_id"] or "", "correction_reason": run["correction_reason"] or "", "close_status": run["close_status"], "selection_policy": run["selection_policy"], "approved_by": run["approved_by"], "approved_at": run["approved_at"], "approval_reason": run["approval_reason"], "approval_reference": run["approval_reference"], "applied_evidence": json.loads(run["applied_evidence"] or "[]"), "excluded_evidence": json.loads(run["excluded_evidence"] or "[]"), "carried_evidence": json.loads(run["carried_evidence"] or "[]"), "closed_at": run["created_at"] or "", "total_snapshot_rows_for_basis_date": len(rows)} return selected, metadata def _snapshot_items(conn: sqlite3.Connection, basis_date: str) -> list[dict[str, Any]]: rows, _metadata = _snapshot_rows_for_latest_run(conn, basis_date) items: list[dict[str, Any]] = [] for row in rows: snap = row_to_dict(row) try: bucket = normalize_liquidity_bucket(snap.get("liquidity_bucket")) except ValueError: bucket = "other" items.append( { "id": snap.get("id", ""), "source_id": snap.get("source_id", ""), "source_type": snap.get("source_type", ""), "authority_source": "asset_snapshots", "valuation_source": snap.get("valuation_source", ""), "name": snap.get("name", ""), "liquidity_bucket": bucket, "amount": int(snap.get("amount") or 0), "valuation_date": snap.get("basis_date", ""), "source_note": snap.get("source_note", ""), "memo": snap.get("memo", ""), "snapshot_run_id": snap.get("snapshot_run_id", ""), "is_correction": bool(snap.get("is_correction", False)), "correction_of_run_id": snap.get("correction_of_run_id", ""), "correction_reason": snap.get("correction_reason", ""), } ) return items def _summarize_valuation_items(included_items: list[dict[str, Any]], excluded_items: list[dict[str, Any]] | None = None) -> dict[str, Any]: bucket_totals = {bucket: 0 for bucket in ASSET_BUCKET_ORDER} bucket_counts = {bucket: 0 for bucket in ASSET_BUCKET_ORDER} for item in included_items: bucket = normalize_liquidity_bucket(item.get("liquidity_bucket")) amount = int(item.get("amount") or 0) bucket_counts[bucket] += 1 bucket_totals[bucket] += amount total_assets = sum(bucket_totals[bucket] for bucket in ASSET_BUCKETS) total_liabilities = sum(abs(bucket_totals[bucket]) for bucket in LIABILITY_BUCKETS) net_worth = total_assets - total_liabilities buckets = [] for bucket in ASSET_BUCKET_ORDER: amount = bucket_totals[bucket] denominator = total_liabilities if bucket in LIABILITY_BUCKETS else total_assets buckets.append( { "liquidity_bucket": bucket, "amount": amount, "account_count": sum(1 for item in included_items if item.get("source_type") == "account" and item.get("liquidity_bucket") == bucket), "manual_asset_count": sum(1 for item in included_items if item.get("source_type") == "manual_asset" and item.get("liquidity_bucket") == bucket), "item_count": bucket_counts[bucket], "asset_share": round(amount / total_assets, 4) if bucket not in LIABILITY_BUCKETS and total_assets else None, "liability_share": round(abs(amount) / total_liabilities, 4) if bucket in LIABILITY_BUCKETS and total_liabilities else None, "share": round(abs(amount) / denominator, 4) if denominator else None, } ) return { "buckets": buckets, "bucket_totals": bucket_totals, "total_assets": total_assets, "total_liabilities": total_liabilities, "net_worth": net_worth, "included_items": included_items, "excluded_items": excluded_items or [], "valuation_policy": _valuation_policy(), } def create_manual_asset(name: str, valuation_amount: int | None, valuation_date: str, liquidity_bucket: str, source_note: str = "", memo: str = "", currency: str = "KRW", active: bool = True, db_path: str | None = None) -> dict[str, Any]: if not name.strip(): return fail("validation_error", "manual asset name is required") try: bucket = normalize_liquidity_bucket(liquidity_bucket) amount = int(valuation_amount) if valuation_amount is not None else None vdate = normalize_date(valuation_date, default_today=True) except Exception as e: return fail("validation_error", str(e)) ts = now_iso() asset_id = new_id("asset") with db(db_path) as conn: conn.execute( """INSERT INTO manual_assets(id,name,liquidity_bucket,valuation_amount,valuation_date,currency,source_note,memo,active,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?,?,?,?)""", (asset_id, name.strip(), bucket, amount, vdate, currency, source_note, memo, int(active), ts, ts), ) row = conn.execute("SELECT * FROM manual_assets WHERE id=?", (asset_id,)).fetchone() return ok(manual_asset={**row_to_dict(row), "authority_source": "manual_assets"}) def list_manual_assets(active_only: bool = True, db_path: str | None = None) -> dict[str, Any]: where = "WHERE active=1" if active_only else "" with db(db_path) as conn: rows = conn.execute(f"SELECT * FROM manual_assets {where} ORDER BY name").fetchall() assets = [{**row_to_dict(row), "authority_source": "manual_assets"} for row in rows] return ok(manual_assets=assets, count=len(assets), valuation_policy=_valuation_policy()) def update_manual_asset(asset_id: str, name: str | None = None, valuation_amount: int | None = None, valuation_date: str | None = None, liquidity_bucket: str | None = None, source_note: str | None = None, memo: str | None = None, currency: str | None = None, active: bool | None = None, db_path: str | None = None) -> dict[str, Any]: allowed_updates: list[str] = [] params: list[Any] = [] try: if name is not None: if not name.strip(): return fail("validation_error", "manual asset name is required") allowed_updates.append("name=?") params.append(name.strip()) if valuation_amount is not None: allowed_updates.append("valuation_amount=?") params.append(int(valuation_amount)) if valuation_date is not None: allowed_updates.append("valuation_date=?") params.append(normalize_date(valuation_date, default_today=True)) if liquidity_bucket is not None: allowed_updates.append("liquidity_bucket=?") params.append(normalize_liquidity_bucket(liquidity_bucket)) if source_note is not None: allowed_updates.append("source_note=?") params.append(source_note) if memo is not None: allowed_updates.append("memo=?") params.append(memo) if currency is not None: allowed_updates.append("currency=?") params.append(currency or "KRW") if active is not None: allowed_updates.append("active=?") params.append(int(bool(active))) except Exception as e: return fail("validation_error", str(e)) if not allowed_updates: return fail("validation_error", "no fields to update") allowed_updates.append("updated_at=?") params.extend([now_iso(), asset_id]) with db(db_path) as conn: cur = conn.execute(f"UPDATE manual_assets SET {', '.join(allowed_updates)} WHERE id=?", params) if cur.rowcount == 0: return fail("not_found", "manual asset not found") row = conn.execute("SELECT * FROM manual_assets WHERE id=?", (asset_id,)).fetchone() return ok(manual_asset={**row_to_dict(row), "authority_source": "manual_assets"}) def deactivate_manual_asset(asset_id: str, hard_delete: bool = False, db_path: str | None = None) -> dict[str, Any]: with db(db_path) as conn: if hard_delete: cur = conn.execute("DELETE FROM manual_assets WHERE id=?", (asset_id,)) else: cur = conn.execute("UPDATE manual_assets SET active=0, updated_at=? WHERE id=?", (now_iso(), asset_id)) if cur.rowcount == 0: return fail("not_found", "manual asset not found") row = None if hard_delete else conn.execute("SELECT * FROM manual_assets WHERE id=?", (asset_id,)).fetchone() payload = {"manual_asset_id": asset_id, "deactivated": True, "hard_delete": hard_delete} if row is not None: payload["manual_asset"] = {**row_to_dict(row), "authority_source": "manual_assets"} return ok(**payload) def _snapshot_preview_from_conn(conn: sqlite3.Connection, basis: str, selection_policy: str = "exact") -> dict[str, Any]: account_items, account_excluded, invalid_bucket_accounts = _account_valuation_items(conn, basis, selection_policy) manual_items, manual_excluded = _manual_asset_valuation_items(conn, basis_date=basis, selection_policy=selection_policy) included = account_items + manual_items excluded = account_excluded + manual_excluded summary = _summarize_valuation_items(included, excluded) existing_rows = conn.execute("SELECT COUNT(*) FROM asset_snapshots WHERE basis_date=?", (basis,)).fetchone()[0] warnings: list[dict[str, Any]] = [] for item in excluded: warnings.append({"code": item.get("reason", "excluded_item"), "source_type": item.get("source_type", ""), "source_id": item.get("source_id", ""), "currency": item.get("currency", "KRW")}) if item.get("unverified_candidates"): warnings.append({"code": "unverified_candidate", "source_type": item.get("source_type", ""), "source_id": item.get("source_id", ""), "candidates": [{k: candidate.get(k) for k in ("id", "basis_date", "currency", "valuation_kind", "verification_status")} for candidate in item["unverified_candidates"]]}) for item in included: item_date = str(item.get("valuation_date") or "")[:10] if item_date and item_date < basis: warnings.append({"code": "stale_valuation", "source_type": item.get("source_type", ""), "source_id": item.get("source_id", ""), "name": item.get("name", ""), "valuation_date": item_date, "basis_date": basis}) for item in invalid_bucket_accounts: warnings.append({"code": "invalid_liquidity_bucket", **item}) if existing_rows: warnings.append({"code": "existing_snapshot_rows", "basis_date": basis, "snapshot_row_count": int(existing_rows), "policy": "append_only_correction_required"}) return { "basis_date": basis, "source": "live_authority", "is_closed_snapshot": False, "snapshot_count": 0, "existing_snapshot_row_count": int(existing_rows), "would_be_correction": bool(existing_rows), "warnings": warnings, **summary, } def preview_asset_snapshot_close(basis_date: str = "", selection_policy: str = "exact", db_path: str | None = None) -> dict[str, Any]: if selection_policy not in {"exact", "as_of"}: return fail("validation_error", "selection_policy must be exact or as_of") try: basis = normalize_date(basis_date, default_today=True) if basis_date else current_kst_date().isoformat() except Exception as e: return fail("validation_error", str(e)) with db(db_path) as conn: preview = _snapshot_preview_from_conn(conn, basis, selection_policy) preview["selection_policy"] = selection_policy blocking = [w for w in preview["warnings"] if w.get("code", "").startswith(("missing_", "future_", "unknown_", "unverified", "ambiguous_", "stale_"))] preview["close_status"] = "complete_exact" if selection_policy == "exact" and not blocking else "incomplete_draft" return ok(**preview) def close_asset_snapshot(basis_date: str = "", source_note: str = "", memo: str = "", correction_reason: str = "", selection_policy: str = "exact", idempotency_key: str = "", allow_missing_exclusions: bool = False, allow_as_of_carry_forward: bool = False, approved_by: str = "", approval_reason: str = "", approval_reference: str = "", db_path: str | None = None) -> dict[str, Any]: if selection_policy not in {"exact", "as_of"}: return fail("validation_error", "selection_policy must be exact or as_of") try: basis = normalize_date(basis_date, default_today=True) if basis_date else current_kst_date().isoformat() except Exception as e: return fail("validation_error", str(e)) ts = now_iso() run_id = new_id("snaprun") with db(db_path) as conn: conn.execute("BEGIN IMMEDIATE") preview = _snapshot_preview_from_conn(conn, basis, selection_policy) missing_exclusions = [w for w in preview["warnings"] if w.get("code", "").startswith(("missing_", "future_", "unknown_", "unverified", "ambiguous_"))] carried_items = [item for item in preview["included_items"] if item.get("carried_forward")] conditional = bool(missing_exclusions or carried_items) fingerprint_payload = {"basis_date": basis, "source_note": source_note, "memo": memo, "correction_reason": correction_reason, "selection_policy": selection_policy, "allow_missing_exclusions": allow_missing_exclusions, "allow_as_of_carry_forward": allow_as_of_carry_forward, "approved_by": approved_by, "approval_reason": approval_reason, "approval_reference": approval_reference, "included": [{**{k: i.get(k) for k in ("source_type", "source_id", "amount", "currency", "valuation_id", "valuation_date", "verification_status")}, "state_hash": _evidence_state_hash(i)} for i in preview["included_items"]], "excluded": [{**{k: i.get(k) for k in ("source_type", "source_id", "reason", "valuation_date", "verification_status")}, "state_hash": _evidence_state_hash(i)} for i in preview["excluded_items"]]} fingerprint = hashlib.sha256(json.dumps(fingerprint_payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode()).hexdigest() if missing_exclusions and (not allow_missing_exclusions or not approved_by.strip() or not approval_reason.strip() or not approval_reference.strip()): return fail("conditional_approval_required", "missing exclusions require explicit approval metadata and allow_missing_exclusions") if carried_items and (not allow_as_of_carry_forward or not approved_by.strip() or not approval_reason.strip() or not approval_reference.strip()): return fail("conditional_approval_required", "conditional close requires explicit approval metadata and flags") if idempotency_key: existing_run = conn.execute("SELECT * FROM asset_snapshot_runs WHERE idempotency_key=?", (idempotency_key,)).fetchone() if existing_run: if existing_run["request_fingerprint"] == fingerprint: return ok(reused=True, snapshot_run_id=existing_run["id"], close_status=existing_run["close_status"], basis_date=basis) return fail("idempotency_conflict", "snapshot close idempotency payload differs") included = preview["included_items"] latest_rows, latest_meta = _snapshot_rows_for_latest_run(conn, basis) # A conditional close may legitimately contain zero snapshot rows. # The run record, not row count, is the authoritative close marker. is_correction = bool(latest_meta.get("snapshot_run_id")) if is_correction and not correction_reason.strip(): return fail("validation_error", "correction_reason is required when closing a basis_date that already has a snapshot run", basis_date=basis, existing_snapshot_row_count=preview["existing_snapshot_row_count"]) correction_of_run_id = latest_meta.get("snapshot_run_id", "") snapshot_ids: list[str] = [] for item in included: snapshot_id = new_id("snap") row_note = source_note or item.get("source_note", "") or "month-end close" row_memo = memo or item.get("memo", "") conn.execute( """INSERT INTO asset_snapshots(id,basis_date,source_type,source_id,name,amount,currency,liquidity_bucket,valuation_source,source_note,memo,snapshot_run_id,correction_of_run_id,correction_reason,is_correction,closed_at,valuation_id,selection_policy,verification_status,evidence_kind,exclusion_reason,created_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", ( snapshot_id, basis, item.get("source_type", ""), item.get("source_id", item.get("id", "")), item.get("name", ""), int(item.get("amount") or 0), item.get("currency", "KRW"), normalize_liquidity_bucket(item.get("liquidity_bucket")), item.get("authority_source", item.get("source_type", "")), row_note, row_memo, run_id, correction_of_run_id, correction_reason.strip(), int(is_correction), ts, item.get("valuation_id", ""), selection_policy, item.get("verification_status", "unverified"), "carried_forward" if item.get("carried_forward") else "applied", "", ts, ), ) snapshot_ids.append(snapshot_id) summary = _summarize_valuation_items(included, []) applied_evidence = [{k: item.get(k) for k in ("source_type", "source_id", "amount", "currency", "valuation_id", "valuation_date", "verification_status")} for item in included] excluded_evidence = [{k: item.get(k) for k in ("source_type", "source_id", "reason", "currency")} for item in preview.get("excluded_items", [])] carried_evidence = [{k: item.get(k) for k in ("source_type", "source_id", "amount", "currency", "valuation_id", "valuation_date", "verification_status")} for item in included if item.get("carried_forward")] try: conn.execute("INSERT INTO asset_snapshot_runs(id,basis_date,close_status,selection_policy,idempotency_key,approved_by,approved_at,approval_reason,approval_reference,correction_of_run_id,correction_reason,applied_evidence,excluded_evidence,carried_evidence,request_fingerprint,created_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)", (run_id, basis, "conditional" if conditional else "complete_exact", selection_policy, idempotency_key, approved_by, ts if approved_by else "", approval_reason, approval_reference, correction_of_run_id, correction_reason.strip(), json.dumps(applied_evidence, ensure_ascii=False, sort_keys=True), json.dumps(excluded_evidence, ensure_ascii=False, sort_keys=True), json.dumps(carried_evidence, ensure_ascii=False, sort_keys=True), fingerprint, ts)) except sqlite3.IntegrityError as exc: conn.rollback() return fail("snapshot_conflict", "another snapshot close already owns this basis date", basis_date=basis, detail=str(exc)) return ok( basis_date=basis, source="snapshot", is_closed_snapshot=True, snapshot_run_id=run_id, correction_of_run_id=correction_of_run_id, correction_reason=correction_reason.strip(), is_correction=is_correction, closed_at=ts, close_status="conditional" if conditional else "complete_exact", selection_policy=selection_policy, inserted_count=len(snapshot_ids), inserted_snapshot_ids=snapshot_ids, warnings=preview["warnings"], **summary, ) def create_asset_snapshot(basis_date: str, source_type: str, source_id: str, name: str, amount: int | None, liquidity_bucket: str, valuation_source: str = "", source_note: str = "", memo: str = "", currency: str = "KRW", db_path: str | None = None) -> dict[str, Any]: if source_type not in {"account", "manual_asset"}: return fail("validation_error", "source_type must be one of ['account', 'manual_asset']") if not name.strip(): return fail("validation_error", "snapshot item name is required") try: basis = normalize_date(basis_date, default_today=True) bucket = normalize_liquidity_bucket(liquidity_bucket) amount_int = int(amount) if amount is not None else None except Exception as e: return fail("validation_error", str(e)) ts = now_iso() snapshot_id = new_id("snap") with db(db_path) as conn: conn.execute( """INSERT INTO asset_snapshots(id,basis_date,source_type,source_id,name,amount,currency,liquidity_bucket,valuation_source,source_note,memo,created_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?)""", (snapshot_id, basis, source_type, source_id, name.strip(), amount_int, currency, bucket, valuation_source or source_type, source_note, memo, ts), ) row = conn.execute("SELECT * FROM asset_snapshots WHERE id=?", (snapshot_id,)).fetchone() return ok(asset_snapshot=row_to_dict(row)) def get_asset_snapshot_report(basis_date: str = "", selection_policy: str = "exact", db_path: str | None = None) -> dict[str, Any]: if selection_policy not in {"exact", "as_of"}: return fail("validation_error", "selection_policy must be exact or as_of") basis = normalize_date(basis_date, default_today=True) if basis_date else current_kst_date().isoformat() with db(db_path) as conn: rows, close_metadata = _snapshot_rows_for_latest_run(conn, basis) snapshots = _snapshot_items(conn, basis) if rows else [] if close_metadata: summary = _summarize_valuation_items(snapshots, []) return ok( basis_date=basis, source="asset_snapshots", is_closed_snapshot=True, snapshot_count=len(snapshots), close_metadata=close_metadata, selection_policy=close_metadata.get("selection_policy", selection_policy), close_status=close_metadata.get("close_status", "complete_exact"), **summary, ) account_items, account_excluded, invalid_bucket_accounts = _account_valuation_items(conn, basis, selection_policy) manual_items, manual_excluded = _manual_asset_valuation_items(conn, basis_date=basis, selection_policy=selection_policy) included = account_items + manual_items excluded = account_excluded + manual_excluded summary = _summarize_valuation_items(included, excluded) return ok( basis_date=basis, source="live_authority", selection_policy=selection_policy, close_status="incomplete_draft" if excluded else "complete_exact", is_closed_snapshot=False, snapshot_count=0, invalid_bucket_accounts=invalid_bucket_accounts, **summary, ) def get_net_worth_timeseries(start_date: str = "", end_date: str = "", months: int = 12, db_path: str | None = None) -> dict[str, Any]: try: months_int = max(1, min(int(months), 120)) end_norm = normalize_date(end_date, default_today=True) if end_date else current_kst_date().isoformat() if start_date: start_norm = normalize_date(start_date) else: end_dt = date.fromisoformat(end_norm) start_month = end_dt.month - months_int + 1 year = end_dt.year + (start_month - 1) // 12 month = (start_month - 1) % 12 + 1 start_norm = date(year, month, 1).isoformat() except Exception as e: return fail("validation_error", str(e)) with db(db_path) as conn: basis_rows = conn.execute( """SELECT DISTINCT basis_date FROM asset_snapshots WHERE basis_date>=? AND basis_date<=? ORDER BY basis_date ASC""", (start_norm, end_norm), ).fetchall() points = [] for row in basis_rows: basis = row["basis_date"] items = _snapshot_items(conn, basis) summary = _summarize_valuation_items(items, []) points.append( { "basis_date": basis, "total_assets": summary["total_assets"], "total_liabilities": summary["total_liabilities"], "net_worth": summary["net_worth"], "source": "asset_snapshots", "item_count": len(items), } ) return ok(start_date=start_norm, end_date=end_norm, months=months_int, points=points, count=len(points), valuation_policy=_valuation_policy()) def get_asset_liquidity_report(db_path: str | None = None) -> dict[str, Any]: """Summarize assets/liabilities from shared valuation authority sources. Existing REQ-005 keys are preserved. When manual_assets is empty, totals and source remain equivalent to the original accounts.last_balance behavior. """ with db(db_path) as conn: account_items, account_excluded, invalid_bucket_accounts = _account_valuation_items(conn) manual_items, manual_excluded = _manual_asset_valuation_items(conn) included = account_items + manual_items excluded = account_excluded + manual_excluded summary = _summarize_valuation_items(included, excluded) source_accounts = [item for item in included if item.get("source_type") == "account"] source = "accounts.last_balance" if not manual_items and not manual_excluded else "accounts.last_balance+manual_assets" return ok( buckets=summary["buckets"], accounts=source_accounts, manual_assets=manual_items, total_assets=summary["total_assets"], total_liabilities=summary["total_liabilities"], net_worth=summary["net_worth"], missing_balance_accounts=[ {"id": item.get("id", ""), "name": item.get("name", ""), "liquidity_bucket": item.get("liquidity_bucket", "")} for item in account_excluded ], missing_valuation_items=excluded, invalid_bucket_accounts=invalid_bucket_accounts, manual_asset_count=len(manual_items), included_items=included, excluded_items=excluded, valuation_policy=summary["valuation_policy"], source=source, )