Add formula-driven bank analysis workbook and party grouping

This commit is contained in:
A R R R Associates
2026-07-14 19:01:40 +05:30
parent 57e326ff68
commit e2690884eb
+396 -86
View File
@@ -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