Derive missing bank statement opening and closing balances
This commit is contained in:
@@ -203,6 +203,75 @@ def extract_text(path: str | Path) -> str:
|
||||
def page_of_line(text: str, position: int) -> int:
|
||||
return text[:position].count('\f')+1
|
||||
|
||||
|
||||
|
||||
def _chronological_row_indices(df: pd.DataFrame) -> list:
|
||||
"""Return transaction row indices in statement chronology.
|
||||
|
||||
Bank exports may be oldest-first or newest-first. Preserve the bank's row
|
||||
order within a date while reversing the full sequence only when the dated
|
||||
rows clearly run newest-to-oldest.
|
||||
"""
|
||||
if df.empty:
|
||||
return []
|
||||
dates = pd.to_datetime(df.get("transaction_date"), errors="coerce")
|
||||
valid = dates.dropna()
|
||||
descending = len(valid) >= 2 and valid.iloc[0] > valid.iloc[-1]
|
||||
indices = df.index.tolist()
|
||||
return list(reversed(indices)) if descending else indices
|
||||
|
||||
|
||||
def _derive_missing_statement_meta(df: pd.DataFrame, meta: StatementMeta) -> None:
|
||||
"""Populate balances/totals omitted by the printed statement.
|
||||
|
||||
Some bank statements, including HDFC's compact monthly export, begin with
|
||||
the first transaction and its post-transaction closing balance but do not
|
||||
print a separate opening balance. In that case:
|
||||
|
||||
opening = first_balance + first_debit - first_credit
|
||||
|
||||
The closing balance is the last chronological running balance. Explicit
|
||||
values extracted from the PDF are never overwritten.
|
||||
"""
|
||||
if df.empty:
|
||||
return
|
||||
|
||||
order = _chronological_row_indices(df)
|
||||
if not order:
|
||||
return
|
||||
|
||||
if meta.opening_balance is None:
|
||||
for idx in order:
|
||||
balance = df.at[idx, "balance"] if "balance" in df.columns else None
|
||||
if pd.isna(balance):
|
||||
continue
|
||||
debit = df.at[idx, "debit"] if "debit" in df.columns else None
|
||||
credit = df.at[idx, "credit"] if "credit" in df.columns else None
|
||||
debit_value = 0.0 if pd.isna(debit) else float(debit)
|
||||
credit_value = 0.0 if pd.isna(credit) else float(credit)
|
||||
if debit_value == 0.0 and credit_value == 0.0:
|
||||
continue
|
||||
meta.opening_balance = round(float(balance) + debit_value - credit_value, 2)
|
||||
break
|
||||
|
||||
if meta.closing_balance is None:
|
||||
for idx in reversed(order):
|
||||
balance = df.at[idx, "balance"] if "balance" in df.columns else None
|
||||
if pd.notna(balance):
|
||||
meta.closing_balance = round(float(balance), 2)
|
||||
break
|
||||
|
||||
|
||||
def _derive_missing_statement_totals(df: pd.DataFrame, meta: StatementMeta) -> None:
|
||||
"""Populate statement totals from the validated transaction population."""
|
||||
if df.empty:
|
||||
return
|
||||
if meta.total_debit is None:
|
||||
meta.total_debit = round(float(pd.to_numeric(df.get("debit"), errors="coerce").fillna(0).sum()), 2)
|
||||
if meta.total_credit is None:
|
||||
meta.total_credit = round(float(pd.to_numeric(df.get("credit"), errors="coerce").fillna(0).sum()), 2)
|
||||
|
||||
|
||||
def _apply_balance_delta_validation(df: pd.DataFrame, meta: StatementMeta) -> pd.DataFrame:
|
||||
"""Validate and, when necessary, correct debit/credit using balance movement.
|
||||
|
||||
@@ -224,10 +293,7 @@ def _apply_balance_delta_validation(df: pd.DataFrame, meta: StatementMeta) -> pd
|
||||
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()
|
||||
order = _chronological_row_indices(x)
|
||||
|
||||
previous_balance = meta.opening_balance
|
||||
for idx in order:
|
||||
@@ -285,7 +351,9 @@ def finalize(df: pd.DataFrame, meta: StatementMeta) -> pd.DataFrame:
|
||||
df[c] = pd.NaT
|
||||
else:
|
||||
df[c] = values.map(parse_flexible_date)
|
||||
_derive_missing_statement_meta(df, meta)
|
||||
df = _apply_balance_delta_validation(df, meta)
|
||||
_derive_missing_statement_totals(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)
|
||||
|
||||
Reference in New Issue
Block a user