Rakseeibra/arrr-erp
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add delta-based bank extraction and summary reconciliations

Browse Source
This commit is contained in:
A R R R Associates
2026-07-16 22:53:06 +05:30
parent d9ae1519a4
commit 46536bf96e
2 changed files with 132 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/modules/bank_statement_analyzer/analyzer.py
+52 -1
View File
@@ -587,6 +587,44 @@ def _safe_sheet_name(value: str) -> str:
return re.sub(r"[][\\:*?/]", "_", value)[:31]
def _write_summary_reconciliation(
worksheet,
start_row: int,
opening_balance: float,
statement_closing: float,
money_format,
text_format,
warning_format,
):
"""Add a formula-driven opening-to-closing reconciliation block."""
title_row = start_row
opening_row = start_row + 1
credit_row = start_row + 2
debit_row = start_row + 3
computed_row = start_row + 4
statement_row = start_row + 5
difference_row = start_row + 6
status_row = start_row + 7
worksheet.write(title_row, 0, "Bank Balance Reconciliation", text_format)
worksheet.write(opening_row, 0, "Opening Balance", text_format)
worksheet.write_number(opening_row, 1, float(opening_balance or 0.0), money_format)
worksheet.write(credit_row, 0, "Add: Total Credits", text_format)
worksheet.write_formula(credit_row, 1, "=SUM(tblTransactions[credit])", money_format, 0)
worksheet.write(debit_row, 0, "Less: Total Debits", text_format)
worksheet.write_formula(debit_row, 1, "=SUM(tblTransactions[debit])", money_format, 0)
worksheet.write(computed_row, 0, "Computed Closing Balance", text_format)
worksheet.write_formula(computed_row, 1, f"=B{opening_row + 1}+B{credit_row + 1}-B{debit_row + 1}", money_format, float(opening_balance or 0.0))
worksheet.write(statement_row, 0, "Statement Closing Balance", text_format)
worksheet.write_number(statement_row, 1, float(statement_closing or 0.0), money_format)
worksheet.write(difference_row, 0, "Difference", text_format)
worksheet.write_formula(difference_row, 1, f"=B{computed_row + 1}-B{statement_row + 1}", money_format, 0)
worksheet.write(status_row, 0, "Reconciliation Status", text_format)
worksheet.write_formula(status_row, 1, f'=IF(ABS(B{difference_row + 1})<=0.01,"Reconciled","RECONCILIATION ERROR")', text_format, "Reconciled")
worksheet.conditional_format(difference_row, 1, difference_row, 1, {"type": "cell", "criteria": "not between", "minimum": -0.01, "maximum": 0.01, "format": warning_format})
worksheet.conditional_format(status_row, 1, status_row, 1, {"type": "text", "criteria": "containing", "value": "ERROR", "format": warning_format})
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)
@@ -781,6 +819,10 @@ def export_excel(output, metas, all_df, unique_df, financial_year="", selected_b
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")
if len(cat_df):
ws.autofilter(0, 0, len(cat_df), len(cat_df.columns) - 1)
_write_summary_reconciliation(ws, len(cat_df) + 3, opening, statement_closing, money, text_fmt, warning_fmt)
# 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"):
@@ -797,6 +839,10 @@ def export_excel(output, metas, all_df, unique_df, financial_year="", selected_b
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")
if len(party_df):
ws.autofilter(0, 0, len(party_df), len(party_df.columns) - 1)
_write_summary_reconciliation(ws, len(party_df) + 3, opening, statement_closing, money, text_fmt, warning_fmt)
# 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"):
@@ -811,6 +857,9 @@ def export_excel(output, metas, all_df, unique_df, financial_year="", selected_b
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, "")
if len(pairs):
ws.autofilter(0, 0, len(pairs), len(pairs.columns) - 1)
_write_summary_reconciliation(ws, len(pairs) + 3, opening, statement_closing, money, text_fmt, warning_fmt)
# Formula-driven bank movement Trial Balance.
trial = pd.DataFrame({
@@ -880,7 +929,9 @@ def export_excel(output, metas, all_df, unique_df, financial_year="", selected_b
worksheet.freeze_panes(1, 0)
worksheet.set_row(0, 22, header)
max_col = max(0, worksheet.dim_colmax)
if worksheet.dim_rowmax >= 0 and name not in {"Transaction Classification"}:
if worksheet.dim_rowmax >= 0 and name not in {
"Transaction Classification", "Category Summary", "Party Summary", "Category Party Summary"
}:
worksheet.autofilter(0, 0, worksheet.dim_rowmax, max_col)
worksheet.set_column(0, max_col, 18)
for name in ("Category Summary", "Party Summary", "Category Party Summary", "Trial Balance"):
app/modules/bank_statement_analyzer/parsers/base.py
+80 -3
View File
@@ -29,7 +29,11 @@ class StatementMeta:
STANDARD_COLUMNS = [
"transaction_date", "value_date", "narration", "reference_no",
"debit", "credit", "balance", "bank_name", "customer_name",
"account_number", "source_file", "source_page", "parser_name"
"account_number", "source_file", "source_page", "parser_name",
# Internal extraction-audit fields. These are retained for diagnostics but
# are intentionally omitted from the client-facing workbook.
"printed_debit", "printed_credit", "balance_delta", "movement_difference",
"correction_applied", "correction_reason", "extraction_confidence",
]
def amount(v):
@@ -64,6 +68,76 @@ def extract_text(path: str|Path) -> str:
def page_of_line(text: str, position: int) -> int:
return text[:position].count('\f')+1
def _apply_balance_delta_validation(df: pd.DataFrame, meta: StatementMeta) -> pd.DataFrame:
"""Validate and, when necessary, correct debit/credit using balance movement.
Printed PDF columns remain the first extraction source. The running-balance
delta is the independent accounting control. Where the printed movement and
the balance delta disagree, the delta determines the corrected side and
amount. This common routine is used by every bank parser through ``finalize``.
"""
if df.empty:
return df
tolerance = 0.01
x = df.copy()
x["printed_debit"] = pd.to_numeric(x.get("debit"), errors="coerce")
x["printed_credit"] = pd.to_numeric(x.get("credit"), errors="coerce")
x["balance_delta"] = pd.NA
x["movement_difference"] = pd.NA
x["correction_applied"] = False
x["correction_reason"] = ""
x["extraction_confidence"] = "Printed columns"
dated = pd.to_datetime(x.get("transaction_date"), errors="coerce")
valid_dates = dated.dropna()
descending = len(valid_dates) >= 2 and valid_dates.iloc[0] > valid_dates.iloc[-1]
order = list(reversed(x.index.tolist())) if descending else x.index.tolist()
previous_balance = meta.opening_balance
for idx in order:
current_balance = x.at[idx, "balance"]
if pd.isna(current_balance):
x.at[idx, "extraction_confidence"] = "Review - balance unavailable"
continue
current_balance = float(current_balance)
if previous_balance is None or pd.isna(previous_balance):
previous_balance = current_balance
x.at[idx, "extraction_confidence"] = "Printed columns - no opening delta"
continue
delta = round(current_balance - float(previous_balance), 2)
debit = float(x.at[idx, "debit"]) if pd.notna(x.at[idx, "debit"]) else 0.0
credit = float(x.at[idx, "credit"]) if pd.notna(x.at[idx, "credit"]) else 0.0
printed_movement = round(credit - debit, 2)
movement_difference = round(delta - printed_movement, 2)
x.at[idx, "balance_delta"] = delta
x.at[idx, "movement_difference"] = movement_difference
if abs(movement_difference) <= tolerance:
x.at[idx, "extraction_confidence"] = "100% - printed movement matches delta"
elif abs(delta) > tolerance:
corrected_debit = round(abs(delta), 2) if delta < 0 else 0.0
corrected_credit = round(delta, 2) if delta > 0 else 0.0
x.at[idx, "debit"] = corrected_debit
x.at[idx, "credit"] = corrected_credit
x.at[idx, "correction_applied"] = True
x.at[idx, "correction_reason"] = "Debit/credit corrected from running-balance delta"
if abs(abs(printed_movement) - abs(delta)) <= tolerance:
x.at[idx, "extraction_confidence"] = "99% - amount matched, side corrected by delta"
elif debit == 0.0 and credit == 0.0:
x.at[idx, "extraction_confidence"] = "98% - missing movement derived from delta"
else:
x.at[idx, "extraction_confidence"] = "Review - printed movement replaced by delta"
else:
x.at[idx, "extraction_confidence"] = "Review - zero balance movement"
previous_balance = current_balance
return x
def finalize(df: pd.DataFrame, meta: StatementMeta) -> pd.DataFrame:
if df is None or df.empty:
return pd.DataFrame(columns=STANDARD_COLUMNS)
@@ -76,8 +150,11 @@ def finalize(df: pd.DataFrame, meta: StatementMeta) -> pd.DataFrame:
df[c] = pd.NaT
else:
df[c] = values.map(parse_flexible_date)
df['narration']=df.get('narration','').fillna('').map(norm)
df['reference_no']=df.get('reference_no','').fillna('').map(norm)
df = _apply_balance_delta_validation(df, meta)
narration_values = df['narration'] if 'narration' in df.columns else pd.Series('', index=df.index)
reference_values = df['reference_no'] if 'reference_no' in df.columns else pd.Series('', index=df.index)
df['narration']=narration_values.fillna('').map(norm)
df['reference_no']=reference_values.fillna('').map(norm)
df['bank_name']=meta.bank_name
df['customer_name']=meta.customer_name
df['account_number']=meta.account_number