diff --git a/app/modules/bank_statement_analyzer/analyzer.py b/app/modules/bank_statement_analyzer/analyzer.py index 97838eb..601eeb9 100644 --- a/app/modules/bank_statement_analyzer/analyzer.py +++ b/app/modules/bank_statement_analyzer/analyzer.py @@ -1,6 +1,7 @@ from __future__ import annotations from pathlib import Path +from difflib import SequenceMatcher import re import pandas as pd @@ -15,8 +16,12 @@ ANALYSIS_COLUMNS = [ "transaction_date", "value_date", "narration", "debit", "credit", "balance", "reference_no", "bank_name", "customer_name", "account_number", "source_file", "source_page", "parser_name", "mode", "amount", "direction", "narration_key", - "exact_key", "exact_duplicate", "possible_key", "possible_duplicate", "category", - "counterparty", "review_note", + "exact_key", "exact_duplicate", "possible_key", "possible_duplicate", + "duplicate_group_id", "duplicate_reason", "duplicate_confidence", + "transfer_bank_code", "transfer_reference", "transfer_comment", + "auto_party", "party_match_method", "party_match_confidence", + "auto_category", "auto_nature", "auto_group", "review_note", + "category", "counterparty", ] @@ -69,6 +74,89 @@ def _direction(row) -> str: return "Debit" if pd.notna(row.get("debit")) and float(row.get("debit") or 0) > 0 else "Credit" +TRANSFER_PATTERNS = { + "NEFT": re.compile(r"(?i)\bNEFT(?:/|\s+)([^/\s]+)?(?:/|\s+)([^/\s]+)?(?:/|\s+)([^\n]+)"), + "RTGS": re.compile(r"(?i)\bRTGS(?:/|\s+)([^/\s]+)?(?:/|\s+)([^/\s]+)?(?:/|\s+)([^\n]+)"), + "IMPS": re.compile(r"(?i)\bIMPS(?:/(?:P2A|P2P))?(?:/|\s+)([^/\s]+)?(?:/|\s+)([^/\s]+)?(?:/|\s+)([^\n]+)"), +} + + +def _normalize_party(value: str) -> str: + text = _text(value).upper() + text = re.sub(r"\b(?:PVT\.?\s*LTD\.?|PRIVATE\s+LIMITED)\b", "PRIVATE LIMITED", text) + text = re.sub(r"\bLTD\.?\b", "LIMITED", text) + text = re.sub(r"\bM/S\.?\b", "", text) + text = re.sub(r"[^A-Z0-9& ]+", " ", text) + text = re.sub(r"\b(?:MR|MRS|MS|MISS|SHRI|SRI|SMT)\b", "", text) + return re.sub(r"\s+", " ", text).strip()[:120] + + +def _extract_transfer_details(narration: str) -> tuple[str, str, str]: + raw = _text(narration) + upper = raw.upper() + for mode, pattern in TRANSFER_PATTERNS.items(): + if mode not in upper: + continue + match = pattern.search(raw) + if match: + parts = [(_text(v) if v else "") for v in match.groups()] + bank_code = parts[0][:30] + reference = parts[1][:80] + comment = parts[2][:240] + comment = re.sub(r"(?i)/?BRANCH\s*:\s*.*$", "", comment).strip(" /.-") + return bank_code, reference, comment + return "", "", "" + + +def _party_group_key(value: str) -> str: + return _normalize_party(value) + + +def _group_similar_parties(values: list[str]) -> dict[str, tuple[str, str, float]]: + canonical: list[str] = [] + result: dict[str, tuple[str, str, float]] = {} + for original in values: + key = _party_group_key(original) or "UNIDENTIFIED" + if key in result: + continue + best = None + best_score = 0.0 + for existing in canonical: + score = SequenceMatcher(None, key, existing).ratio() + tokens_a = set(key.split()) + tokens_b = set(existing.split()) + containment = min(len(tokens_a & tokens_b) / max(1, len(tokens_a)), len(tokens_a & tokens_b) / max(1, len(tokens_b))) + if key in existing or existing in key: + score = max(score, 0.96 if min(len(key), len(existing)) >= 6 else score) + score = max(score, containment) + if score > best_score: + best, best_score = existing, score + if best and best_score >= 0.94: + result[key] = (best.title(), "Fuzzy/partial", round(best_score * 100, 1)) + else: + canonical.append(key) + result[key] = (key.title(), "Normalized exact", 100.0) + return result + + +def _apply_party_grouping(df: pd.DataFrame) -> pd.DataFrame: + if df.empty: + return df + x = df.copy() + mapping = _group_similar_parties([str(v or "") for v in x["auto_party"].tolist()]) + grouped, methods, scores = [], [], [] + for value in x["auto_party"].tolist(): + key = _party_group_key(str(value or "")) or "UNIDENTIFIED" + party, method, score = mapping.get(key, (str(value or "Unidentified"), "Original", 0.0)) + grouped.append(party) + methods.append(method) + scores.append(score) + x["auto_party"] = grouped + x["party_match_method"] = methods + x["party_match_confidence"] = scores + return x + + def _extract_counterparty(narration: str, direction: str) -> str: raw = _text(narration) upper = raw.upper() @@ -193,14 +281,33 @@ def enrich(df, classification_enabled: bool = True): lambda r: f"{r.transaction_date}|{r.direction}|{r.amount:.2f}|{r.narration_key}", axis=1 ) x["possible_duplicate"] = x.duplicated("possible_key", keep=False) & ~x["exact_duplicate"] + x["duplicate_group_id"] = x.groupby("exact_key", dropna=False).ngroup().map(lambda n: f"EX-{n+1:06d}") + x.loc[~x["exact_duplicate"], "duplicate_group_id"] = "" + x["duplicate_reason"] = x.apply(lambda r: "Exact same date/value/narration/balance" if bool(r.get("exact_duplicate")) else ("Same date/direction/amount/normalized narration" if bool(r.get("possible_duplicate")) else ""), axis=1) + x["duplicate_confidence"] = x.apply(lambda r: 100.0 if bool(r.get("exact_duplicate")) else (85.0 if bool(r.get("possible_duplicate")) else 0.0), axis=1) + transfer = x["narration"].map(_extract_transfer_details) + x["transfer_bank_code"] = transfer.map(lambda t: t[0]) + x["transfer_reference"] = transfer.map(lambda t: t[1]) + x["transfer_comment"] = transfer.map(lambda t: t[2]) if classification_enabled: classified = x.apply(_classify_row, axis=1, result_type="expand") - x["category"] = classified[0] - x["counterparty"] = classified[1] + x["auto_category"] = classified[0] + x["auto_party"] = classified[1] + x = _apply_party_grouping(x) + x["auto_nature"] = x["direction"].map(lambda d: "Payment" if d == "Debit" else "Receipt") + x["auto_group"] = x["auto_category"].map(lambda c: "Contra / Balance Sheet" if c in {"Self Transfer / Contra", "Loan Receipt", "Loan Repayment", "Capital Introduction", "Drawings"} else ("Income / Receipt" if any(k in c for k in ("Receipt", "Deposit", "Interest Received", "Settlement")) else "Expense / Payment")) + x["category"] = x["auto_category"] + x["counterparty"] = x["auto_party"] x["review_note"] = x.apply(_review_note, axis=1) else: - x["category"] = "Classification disabled" - x["counterparty"] = x["narration"].map(lambda value: _text(value)[:120] or "Unidentified") + x["auto_category"] = "Classification disabled" + x["auto_party"] = x["narration"].map(lambda value: _text(value)[:120] or "Unidentified") + x["party_match_method"] = "Original" + x["party_match_confidence"] = 0.0 + x["auto_nature"] = x["direction"].map(lambda d: "Payment" if d == "Debit" else "Receipt") + x["auto_group"] = "Unclassified" + x["category"] = x["auto_category"] + x["counterparty"] = x["auto_party"] x["review_note"] = x.apply(_review_note, axis=1) return _ensure_analysis_columns(x) @@ -340,13 +447,29 @@ def draft_financials(df, metas): def _workbook_columns(df): preferred = [ - "transaction_date", "value_date", "narration", "counterparty", "category", "mode", "direction", - "debit", "credit", "balance", "reference_no", "bank_name", "customer_name", "account_number", - "source_file", "source_page", "parser_name", "exact_duplicate", "possible_duplicate", "review_note", + "transaction_date", "value_date", "narration", "transfer_bank_code", "transfer_reference", + "transfer_comment", "auto_party", "party_match_method", "party_match_confidence", + "auto_category", "auto_nature", "auto_group", "mode", "direction", "debit", "credit", + "balance", "reference_no", "bank_name", "customer_name", "account_number", "source_file", + "source_page", "parser_name", "exact_duplicate", "possible_duplicate", "duplicate_group_id", + "duplicate_reason", "duplicate_confidence", "review_note", ] return [column for column in preferred if column in df.columns] +def _excel_col(index: int) -> str: + result = "" + n = index + 1 + while n: + n, remainder = divmod(n - 1, 26) + result = chr(65 + remainder) + result + return result + + +def _safe_sheet_name(value: str) -> str: + return re.sub(r"[][\\:*?/]", "_", value)[:31] + + def export_excel(output, metas, all_df, unique_df, financial_year="", selected_bank="auto", classification_enabled=True): all_df = _ensure_analysis_columns(all_df) unique_df = _ensure_analysis_columns(unique_df) @@ -354,92 +477,279 @@ def export_excel(output, metas, all_df, unique_df, financial_year="", selected_b customer = next((meta.customer_name for meta in metas if meta.customer_name), "") account = next((meta.account_number for meta in metas if meta.account_number), "") banks = ", ".join(sorted({meta.bank_name for meta in metas})) - exact_count = _duplicate_count(all_df, "exact_duplicate") - possible_count = _duplicate_count(all_df, "possible_duplicate") - opening = next((meta.opening_balance for meta in metas if meta.opening_balance is not None), None) - closing = next((meta.closing_balance for meta in reversed(metas) if meta.closing_balance is not None), None) - dashboard = pd.DataFrame( - [ - ["Customer / Account Holder", customer, "Extracted from statement or user override"], - ["Account Number", account, "Extracted from statement or user override"], - ["Bank(s)", banks, "Detected/selected parser result"], - ["Financial Year", financial_year or "Not specified", "Optional user input"], - ["Statements Uploaded", len(metas), "PDF files processed"], - ["Rows Extracted", len(all_df), "Before exact duplicate removal"], - ["Exact Duplicate Rows", exact_count, "Includes overlapping statement rows"], - ["Possible Duplicate Rows", possible_count, "Requires manual review"], - ["Unique Transactions", len(unique_df), "Used for summaries"], - ["Opening Balance", opening, "From first available statement summary"], - ["Total Debit (Unique)", float(unique_df.debit.sum()) if not unique_df.empty else 0, "After exact duplicate removal"], - ["Total Credit (Unique)", float(unique_df.credit.sum()) if not unique_df.empty else 0, "After exact duplicate removal"], - ["Closing Balance", closing, "From last available statement summary"], - ["Classification", "Enabled" if classification_enabled else "Disabled", "Generic narration-based classification"], - ], - columns=["Metric", "Value", "Remarks"], - ) - all_export = all_df[_workbook_columns(all_df)].copy() if not all_df.empty else all_df.copy() - unique_export = unique_df[_workbook_columns(unique_df)].copy() if not unique_df.empty else unique_df.copy() - review_items = unique_export[unique_export["review_note"].fillna("").ne("")].copy() if not unique_export.empty else unique_export.copy() - notes = pd.DataFrame( - { - "Assumption / Method": [ - "Extraction method", - "Debit/Credit identification", - "Exact duplicate detection", - "Possible duplicate detection", - "Bank selection", - "Classification", - "Important limitation", - "Files analysed", - ], - "Details": [ - "Transactions are extracted using the selected bank parser or automatic parser detection.", - "Debit and credit fields are taken from the bank-specific parser and reconciled to statement summaries where available.", - "Exact duplicates use transaction date, value date, debit, credit, balance and normalized narration; the first occurrence is retained.", - "Possible duplicates use the same date, direction, amount and normalized narration and are not removed automatically.", - f"Selected option: {selected_bank or 'auto'}. Manual selection still validates that the PDF matches the selected bank format.", - "Categories, counterparties and review notes are generic narration-based indicators and must be verified against books and source records.", - "This workbook is an analysis aid, not a substitute for ledger, GST, inventory, receivable/payable, cash-book and supporting records.", - "; ".join(meta.source_file for meta in metas), - ], - } - ) + opening = next((meta.opening_balance for meta in metas if meta.opening_balance is not None), 0.0) or 0.0 + statement_closing = next((meta.closing_balance for meta in reversed(metas) if meta.closing_balance is not None), 0.0) or 0.0 - with pd.ExcelWriter(output, engine="xlsxwriter", datetime_format="dd-mmm-yyyy") as writer: - dashboard.to_excel(writer, sheet_name="Dashboard", index=False) + tx = unique_df[_workbook_columns(unique_df)].copy() if not unique_df.empty else pd.DataFrame(columns=_workbook_columns(unique_df)) + tx.insert(0, "row_id", range(1, len(tx) + 1)) + tx["manual_party"] = "" + tx["final_party"] = tx.get("auto_party", "") + tx["manual_category"] = "" + tx["final_category"] = tx.get("auto_category", "") + tx["manual_nature"] = "" + tx["final_nature"] = tx.get("auto_nature", "") + tx["manual_group"] = "" + tx["final_group"] = tx.get("auto_group", "") + tx["manual_review_note"] = "" + + raw_export = all_df[_workbook_columns(all_df)].copy() if not all_df.empty else all_df.copy() + exact_export = raw_export[all_df["exact_duplicate"].fillna(False).astype(bool).to_numpy()].copy() if not all_df.empty else raw_export.copy() + possible_export = raw_export[all_df["possible_duplicate"].fillna(False).astype(bool).to_numpy()].copy() if not all_df.empty else raw_export.copy() + + categories = sorted({str(v) for v in tx.get("auto_category", pd.Series(dtype=str)).dropna() if str(v).strip()} | { + "Advance Paid", "Advance Received", "Bank Charges", "Capital Introduction", "Cash Deposit / Cash Sales", + "Cash Withdrawal", "Customer / Business Receipt", "Drawings", "Fuel Expense", "Interest Paid", + "Interest Received", "Loan Receipt", "Loan Repayment", "Purchase / Supplier Payment", "Refund / Reversal", + "Rent Expense", "Salary Payment", "Self Transfer / Contra", "Tax / TDS Payment", "Unclassified / Review", + }) + parties = sorted({str(v) for v in tx.get("auto_party", pd.Series(dtype=str)).dropna() if str(v).strip()}) + natures = ["Receipt", "Payment", "Contra", "Loan", "Capital", "Refund", "Unclassified"] + groups = ["Income / Receipt", "Expense / Payment", "Contra / Balance Sheet", "Unclassified"] + + notes = pd.DataFrame({ + "Assumption / Method": [ + "Python analysis engine", "Excel reporting engine", "Date parsing", "Duplicate detection", + "Party grouping", "Transfer comments", "Manual overrides", "Trial balance limitation", "Files analysed", + ], + "Details": [ + "PDF extraction, transaction reconstruction, duplicate detection, references, comments and automatic classification are performed in Python.", + "Dashboard, category, party, category-party, monthly and trial-balance outputs are formula-driven and recalculate after manual overrides.", + "Bank parsers accept multiple supported date layouts and validate transaction dates against statement periods where available.", + "Exact and possible duplicate flags are Python-generated. Formula summaries exclude exact duplicates because Transaction Classification contains one retained row per exact group.", + "Party names are normalized using regex cleanup and conservative fuzzy/partial matching. Review suggested groupings before finalisation.", + "NEFT, RTGS and IMPS bank codes, references and narration comments are preserved in dedicated columns.", + "Enter corrections only in Manual Party, Manual Category, Manual Nature, Manual Group and Manual Review Note. Final columns use Excel formulas.", + "The Trial Balance is a bank-movement working paper, not a final accounting trial balance. Verify with ledgers, invoices, GST, loans, capital and supporting records.", + "; ".join(meta.source_file for meta in metas), + ], + }) + + with pd.ExcelWriter(output, engine="xlsxwriter", datetime_format="dd-mmm-yyyy", engine_kwargs={"options": {"strings_to_formulas": True}}) as writer: + workbook = writer.book + workbook.set_calc_mode("auto") + header = workbook.add_format({"bold": True, "bg_color": "#1F4E78", "font_color": "white", "border": 1}) + subheader = workbook.add_format({"bold": True, "bg_color": "#D9EAF7", "border": 1}) + money = workbook.add_format({"num_format": "#,##0.00;[Red](#,##0.00)", "border": 1}) + integer = workbook.add_format({"num_format": "0", "border": 1}) + percent = workbook.add_format({"num_format": "0.0%", "border": 1}) + date_format = workbook.add_format({"num_format": "dd-mmm-yyyy", "border": 1}) + input_fmt = workbook.add_format({"bg_color": "#FFF2CC", "border": 1}) + formula_fmt = workbook.add_format({"bg_color": "#E2F0D9", "border": 1}) + text_fmt = workbook.add_format({"border": 1}) + warning_fmt = workbook.add_format({"bg_color": "#FCE4D6", "font_color": "#9C0006", "border": 1}) + + # Raw and Python-analysis sheets. recon.to_excel(writer, sheet_name="Statement Reconciliation", index=False) - all_export.to_excel(writer, sheet_name="All Extracted Rows", index=False) - unique_export.to_excel(writer, sheet_name="Unique Transactions", index=False) - all_export[all_df["exact_duplicate"].fillna(False).astype(bool).to_numpy()].to_excel(writer, sheet_name="Exact Duplicates", index=False) - all_export[all_df["possible_duplicate"].fillna(False).astype(bool).to_numpy()].to_excel(writer, sheet_name="Possible Duplicates", index=False) - monthly_summary(unique_df).to_excel(writer, sheet_name="Monthly Summary", index=False) - mode_summary(unique_df).to_excel(writer, sheet_name="Mode Summary", index=False) - category_summary(unique_df).to_excel(writer, sheet_name="Category Summary", index=False) - party_summary(unique_df).to_excel(writer, sheet_name="Party Summary", index=False) - review_items.to_excel(writer, sheet_name="Review Items", index=False) - draft_financials(unique_df, metas).to_excel(writer, sheet_name="Draft Financials", index=False) + raw_export.to_excel(writer, sheet_name="All Extracted Rows", index=False) + exact_export.to_excel(writer, sheet_name="Exact Duplicates", index=False) + possible_export.to_excel(writer, sheet_name="Possible Duplicates", index=False) duplicate_summary(all_df).to_excel(writer, sheet_name="Duplicate Summary", index=False) notes.to_excel(writer, sheet_name="Assumptions", index=False) - workbook = writer.book - header = workbook.add_format({"bold": True, "bg_color": "#1F4E78", "font_color": "white", "border": 1}) - money = workbook.add_format({"num_format": "#,##0.00"}) - date_format = workbook.add_format({"num_format": "dd-mmm-yyyy"}) + # Masters first so validation ranges exist. + max_master = max(len(categories), len(parties), len(natures), len(groups), 1) + masters = pd.DataFrame({ + "Categories": categories + [""] * (max_master - len(categories)), + "Parties": parties + [""] * (max_master - len(parties)), + "Natures": natures + [""] * (max_master - len(natures)), + "Groups": groups + [""] * (max_master - len(groups)), + }) + masters.to_excel(writer, sheet_name="Masters", index=False) + workbook.define_name("CategoryList", f"=Masters!$A$2:$A${len(categories)+1}") + workbook.define_name("PartyList", f"=Masters!$B$2:$B${max(2, len(parties)+1)}") + workbook.define_name("NatureList", f"=Masters!$C$2:$C${len(natures)+1}") + workbook.define_name("GroupList", f"=Masters!$D$2:$D${len(groups)+1}") + + # Transaction classification table with manual inputs and formula-driven final fields. + tx.to_excel(writer, sheet_name="Transaction Classification", index=False) + ws_tx = writer.sheets["Transaction Classification"] + columns = list(tx.columns) + col_index = {name: i for i, name in enumerate(columns)} + for row in range(1, len(tx) + 1): + excel_row = row + 1 + for manual in ("manual_party", "manual_category", "manual_nature", "manual_group", "manual_review_note"): + ws_tx.write_blank(row, col_index[manual], None, input_fmt) + formulas = { + "final_party": f'=IF({ _excel_col(col_index["manual_party"]) }{excel_row}<>"",{ _excel_col(col_index["manual_party"]) }{excel_row},{ _excel_col(col_index["auto_party"]) }{excel_row})', + "final_category": f'=IF({ _excel_col(col_index["manual_category"]) }{excel_row}<>"",{ _excel_col(col_index["manual_category"]) }{excel_row},{ _excel_col(col_index["auto_category"]) }{excel_row})', + "final_nature": f'=IF({ _excel_col(col_index["manual_nature"]) }{excel_row}<>"",{ _excel_col(col_index["manual_nature"]) }{excel_row},{ _excel_col(col_index["auto_nature"]) }{excel_row})', + "final_group": f'=IF({ _excel_col(col_index["manual_group"]) }{excel_row}<>"",{ _excel_col(col_index["manual_group"]) }{excel_row},{ _excel_col(col_index["auto_group"]) }{excel_row})', + } + cached = { + "final_party": str(tx.iloc[row-1].get("auto_party", "")), + "final_category": str(tx.iloc[row-1].get("auto_category", "")), + "final_nature": str(tx.iloc[row-1].get("auto_nature", "")), + "final_group": str(tx.iloc[row-1].get("auto_group", "")), + } + for name, formula in formulas.items(): + ws_tx.write_formula(row, col_index[name], formula, formula_fmt, cached[name]) + last_row = max(2, len(tx) + 1) + ws_tx.data_validation(1, col_index["manual_category"], last_row-1, col_index["manual_category"], {"validate": "list", "source": "=CategoryList"}) + ws_tx.data_validation(1, col_index["manual_party"], last_row-1, col_index["manual_party"], {"validate": "list", "source": "=PartyList"}) + ws_tx.data_validation(1, col_index["manual_nature"], last_row-1, col_index["manual_nature"], {"validate": "list", "source": "=NatureList"}) + ws_tx.data_validation(1, col_index["manual_group"], last_row-1, col_index["manual_group"], {"validate": "list", "source": "=GroupList"}) + if len(tx): + ws_tx.add_table(0, 0, len(tx), len(columns)-1, {"name": "tblTransactions", "columns": [{"header": c} for c in columns], "style": "Table Style Medium 2"}) + ws_tx.freeze_panes(1, 4) + ws_tx.set_column(col_index["narration"], col_index["narration"], 60) + for c in ("transfer_comment", "auto_party", "manual_party", "final_party", "auto_category", "manual_category", "final_category", "review_note", "manual_review_note"): + ws_tx.set_column(col_index[c], col_index[c], 28) + for c in ("debit", "credit", "balance"): + ws_tx.set_column(col_index[c], col_index[c], 15, money) + for c in ("transaction_date", "value_date"): + ws_tx.set_column(col_index[c], col_index[c], 13, date_format) + + # Formula-driven Dashboard. + exact_excel_col = _excel_col(list(raw_export.columns).index("exact_duplicate")) if "exact_duplicate" in raw_export.columns else "A" + possible_excel_col = _excel_col(list(raw_export.columns).index("possible_duplicate")) if "possible_duplicate" in raw_export.columns else "A" + dashboard_rows = [ + ("Customer / Account Holder", customer, "Extracted or override"), + ("Account Number", account, "Extracted or override"), + ("Bank(s)", banks, "Detected/selected parser"), + ("Financial Year", financial_year or "Not specified", "User input"), + ("Statements Uploaded", len(metas), "PDF files processed"), + ("Transactions Retained", "=ROWS(tblTransactions[row_id])", "Exact-deduplicated transaction table"), + ("Exact Duplicate Rows", f'=COUNTIF(\'All Extracted Rows\'!${exact_excel_col}:${exact_excel_col},TRUE)', "Python detected"), + ("Possible Duplicate Rows", f'=COUNTIF(\'All Extracted Rows\'!${possible_excel_col}:${possible_excel_col},TRUE)', "Python detected"), + ("Opening Signed Balance", opening, "CR positive / DR negative"), + ("Total Debit", "=SUM(tblTransactions[debit])", "Formula-driven"), + ("Total Credit", "=SUM(tblTransactions[credit])", "Formula-driven"), + ("Computed Closing Balance", "=B10+B12-B11", "Opening + Credits - Debits"), + ("Statement Closing Balance", statement_closing, "Extracted from statement"), + ("Closing Difference", "=B13-B14", "Should be zero"), + ("Review Items", '=COUNTIF(tblTransactions[review_note],"?*")+COUNTIF(tblTransactions[manual_review_note],"?*")', "Formula-driven"), + ] + pd.DataFrame(dashboard_rows, columns=["Metric", "Value", "Remarks"]).to_excel(writer, sheet_name="Dashboard", index=False) + ws_dash = writer.sheets["Dashboard"] + ws_dash.set_column("A:A", 32) + ws_dash.set_column("B:B", 24) + ws_dash.set_column("C:C", 45) + ws_dash.conditional_format("B15", {"type": "cell", "criteria": "!=", "value": 0, "format": warning_fmt}) + + # Category Summary formulas. + cat_df = pd.DataFrame({"Category": categories}) + cat_df["Debit"] = "" + cat_df["Credit"] = "" + cat_df["Net Credit/(Debit)"] = "" + cat_df["Transaction Count"] = "" + cat_df["View Transactions"] = "" + cat_df.to_excel(writer, sheet_name="Category Summary", index=False) + ws = writer.sheets["Category Summary"] + for r in range(1, len(cat_df)+1): + er = r+1 + ws.write_formula(r, 1, f'=SUMIFS(tblTransactions[debit],tblTransactions[final_category],A{er})', money, 0) + ws.write_formula(r, 2, f'=SUMIFS(tblTransactions[credit],tblTransactions[final_category],A{er})', money, 0) + ws.write_formula(r, 3, f'=C{er}-B{er}', money, 0) + ws.write_formula(r, 4, f'=COUNTIF(tblTransactions[final_category],A{er})', integer, 0) + ws.write_formula(r, 5, '=HYPERLINK("#\'Transaction Classification\'!A1","View / Filter")', text_fmt, "View / Filter") + + # Party Summary formulas. + party_df = pd.DataFrame({"Party / Counterparty": parties}) + for c in ("Debit", "Credit", "Net Credit/(Debit)", "Transaction Count", "First Date", "Last Date", "View Transactions"): + party_df[c] = "" + party_df.to_excel(writer, sheet_name="Party Summary", index=False) + ws = writer.sheets["Party Summary"] + for r in range(1, len(party_df)+1): + er = r+1 + ws.write_formula(r, 1, f'=SUMIFS(tblTransactions[debit],tblTransactions[final_party],A{er})', money, 0) + ws.write_formula(r, 2, f'=SUMIFS(tblTransactions[credit],tblTransactions[final_party],A{er})', money, 0) + ws.write_formula(r, 3, f'=C{er}-B{er}', money, 0) + ws.write_formula(r, 4, f'=COUNTIF(tblTransactions[final_party],A{er})', integer, 0) + ws.write_formula(r, 5, f'=IFERROR(MINIFS(tblTransactions[transaction_date],tblTransactions[final_party],A{er}),"")', date_format, "") + ws.write_formula(r, 6, f'=IFERROR(MAXIFS(tblTransactions[transaction_date],tblTransactions[final_party],A{er}),"")', date_format, "") + ws.write_formula(r, 7, '=HYPERLINK("#\'Transaction Classification\'!A1","View / Filter")', text_fmt, "View / Filter") + + # Category + Party formula-driven summary and trial balance detail. + pairs = tx[["auto_category", "auto_party"]].drop_duplicates().rename(columns={"auto_category": "Main Category", "auto_party": "Party / Counterparty"}) if len(tx) else pd.DataFrame(columns=["Main Category", "Party / Counterparty"]) + for c in ("Debit", "Credit", "Net Credit/(Debit)", "Transaction Count", "Source Modes", "Review"): + pairs[c] = "" + pairs.to_excel(writer, sheet_name="Category Party Summary", index=False) + ws = writer.sheets["Category Party Summary"] + for r in range(1, len(pairs)+1): + er = r+1 + ws.write_formula(r, 2, f'=SUMIFS(tblTransactions[debit],tblTransactions[final_category],A{er},tblTransactions[final_party],B{er})', money, 0) + ws.write_formula(r, 3, f'=SUMIFS(tblTransactions[credit],tblTransactions[final_category],A{er},tblTransactions[final_party],B{er})', money, 0) + ws.write_formula(r, 4, f'=D{er}-C{er}', money, 0) + ws.write_formula(r, 5, f'=COUNTIFS(tblTransactions[final_category],A{er},tblTransactions[final_party],B{er})', integer, 0) + ws.write_formula(r, 6, f'=IF(E{er}>0,"See Transaction Classification","")', text_fmt, "See Transaction Classification") + ws.write_formula(r, 7, f'=IF(OR(A{er}="Unclassified / Review",A{er}="Other Bank Receipt / Review",A{er}="Other Bank Payment / Review"),"Review","")', text_fmt, "") + + # Formula-driven bank movement Trial Balance. + trial = pd.DataFrame({ + "Main Category": ["Opening Balance"] + list(pairs["Main Category"]) + ["Total Period Movement", "Computed Closing Balance", "Statement Closing Balance", "Difference"], + "Party / Counterparty": ["Bank Account"] + list(pairs["Party / Counterparty"]) + ["", "Bank Account", "Bank Account", "Reconciliation"], + "Debit": [""] * (len(pairs)+5), + "Credit": [""] * (len(pairs)+5), + "Net Credit/(Debit)": [""] * (len(pairs)+5), + "Reference": [""] * (len(pairs)+5), + }) + trial.to_excel(writer, sheet_name="Trial Balance", index=False) + ws = writer.sheets["Trial Balance"] + ws.write_number(1, 4, opening, money) + ws.write(1, 5, "Extracted statement opening signed balance", text_fmt) + for i in range(len(pairs)): + r = i + 2 + er = r + 1 + ws.write_formula(r, 2, f'=SUMIFS(tblTransactions[debit],tblTransactions[final_category],A{er},tblTransactions[final_party],B{er})', money, 0) + ws.write_formula(r, 3, f'=SUMIFS(tblTransactions[credit],tblTransactions[final_category],A{er},tblTransactions[final_party],B{er})', money, 0) + ws.write_formula(r, 4, f'=D{er}-C{er}', money, 0) + ws.write_formula(r, 5, '=HYPERLINK("#\'Transaction Classification\'!A1","View / Filter")', text_fmt, "View / Filter") + total_row = len(pairs) + 2 + computed_row = total_row + 1 + statement_row = total_row + 2 + diff_row = total_row + 3 + ws.write_formula(total_row, 2, f'=SUM(C3:C{total_row})', money, 0) + ws.write_formula(total_row, 3, f'=SUM(D3:D{total_row})', money, 0) + ws.write_formula(total_row, 4, f'=D{total_row+1}-C{total_row+1}', money, 0) + ws.write_formula(computed_row, 4, f'=E2+E{total_row+1}', money, opening) + ws.write_number(statement_row, 4, statement_closing, money) + ws.write_formula(diff_row, 4, f'=E{computed_row+1}-E{statement_row+1}', money, 0) + ws.conditional_format(diff_row, 4, diff_row, 4, {"type": "cell", "criteria": "!=", "value": 0, "format": warning_fmt}) + + # Formula-driven monthly summary. + dates = pd.to_datetime(tx.get("transaction_date", pd.Series(dtype="datetime64[ns]")), errors="coerce").dropna() + months = pd.period_range(dates.min().to_period("M"), dates.max().to_period("M"), freq="M") if not dates.empty else [] + monthly = pd.DataFrame({"Month": [p.to_timestamp() for p in months]}) + for c in ("Opening Balance", "Debit", "Credit", "Net Movement", "Computed Closing", "Transaction Count"): + monthly[c] = "" + monthly.to_excel(writer, sheet_name="Monthly Summary", index=False) + ws = writer.sheets["Monthly Summary"] + for r in range(1, len(monthly)+1): + er = r+1 + if r == 1: + ws.write_number(r, 1, opening, money) + else: + ws.write_formula(r, 1, f'=E{er-1}', money, 0) + ws.write_formula(r, 2, f'=SUMIFS(tblTransactions[debit],tblTransactions[transaction_date],">="&A{er},tblTransactions[transaction_date],"<"&EDATE(A{er},1))', money, 0) + ws.write_formula(r, 3, f'=SUMIFS(tblTransactions[credit],tblTransactions[transaction_date],">="&A{er},tblTransactions[transaction_date],"<"&EDATE(A{er},1))', money, 0) + ws.write_formula(r, 4, f'=D{er}-C{er}', money, 0) + ws.write_formula(r, 5, f'=B{er}+E{er}', money, 0) + ws.write_formula(r, 6, f'=COUNTIFS(tblTransactions[transaction_date],">="&A{er},tblTransactions[transaction_date],"<"&EDATE(A{er},1))', integer, 0) + ws.write_datetime(r, 0, pd.Timestamp(monthly.iloc[r-1]["Month"]).to_pydatetime(), workbook.add_format({"num_format": "mmm-yyyy", "border": 1})) + + # Transfer Comments and Review Items derived from Python analysis. + transfer_cols = [c for c in ["transaction_date", "mode", "transfer_bank_code", "transfer_reference", "transfer_comment", "auto_party", "auto_category", "debit", "credit", "review_note"] if c in tx.columns] + tx.loc[tx.get("transfer_comment", pd.Series(index=tx.index, dtype=str)).fillna("").ne(""), transfer_cols].to_excel(writer, sheet_name="Transfer Comments", index=False) + review_mask = tx.get("review_note", pd.Series(index=tx.index, dtype=str)).fillna("").ne("") + tx.loc[review_mask].to_excel(writer, sheet_name="Review Items", index=False) + + # Retain existing mode and draft sheets for backward compatibility. + mode_summary(unique_df).to_excel(writer, sheet_name="Mode Summary", index=False) + draft_financials(unique_df, metas).to_excel(writer, sheet_name="Draft Financials", index=False) + + # Shared formatting. for name, worksheet in writer.sheets.items(): worksheet.freeze_panes(1, 0) worksheet.set_row(0, 22, header) max_col = max(0, worksheet.dim_colmax) - if worksheet.dim_rowmax >= 0: + if worksheet.dim_rowmax >= 0 and name not in {"Transaction Classification"}: worksheet.autofilter(0, 0, worksheet.dim_rowmax, max_col) worksheet.set_column(0, max_col, 18) - writer.sheets["Dashboard"].set_column("A:A", 34) - writer.sheets["Dashboard"].set_column("B:B", 28) - writer.sheets["Dashboard"].set_column("C:C", 44) - for sheet in ("All Extracted Rows", "Unique Transactions", "Exact Duplicates", "Possible Duplicates", "Review Items"): - ws = writer.sheets[sheet] - ws.set_column("A:B", 13, date_format) - ws.set_column("C:C", 58) - ws.set_column("D:E", 30) - for sheet in ("Category Summary", "Party Summary", "Monthly Summary", "Mode Summary", "Draft Financials"): - writer.sheets[sheet].set_column(1, max(1, writer.sheets[sheet].dim_colmax), 18, money) + for name in ("Category Summary", "Party Summary", "Category Party Summary", "Trial Balance"): + writer.sheets[name].set_column("A:B", 30) + writer.sheets["Trial Balance"].set_column("C:E", 18, money) + writer.sheets["Assumptions"].set_column("A:A", 28) + writer.sheets["Assumptions"].set_column("B:B", 90) + writer.sheets["Masters"].set_column("A:D", 34) + writer.sheets["Masters"].hide() return output