From ba7b7108fd7f148c1c3965f2d40ca29ed9a38781 Mon Sep 17 00:00:00 2001 From: adawdani <261673663+adawdani@users.noreply.github.com> Date: Mon, 18 May 2026 16:43:18 -0500 Subject: [PATCH] Add Tabula-py bank statement PDF parser for DCF integration --- bank_statement_parser.py | 298 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 298 insertions(+) create mode 100644 bank_statement_parser.py diff --git a/bank_statement_parser.py b/bank_statement_parser.py new file mode 100644 index 0000000..b3cfb25 --- /dev/null +++ b/bank_statement_parser.py @@ -0,0 +1,298 @@ +""" +Bank Statement PDF Parser using Tabula-py +========================================== + +This module extracts bank statement data from PDF files and converts them +to a structured format compatible with DCF valuation models. + +Author: Business Valuation DCF Team +""" + +import tabula +import pandas as pd +import os +from pathlib import Path +from typing import Optional, Tuple +import warnings + +warnings.filterwarnings('ignore') + + +class BankStatementParser: + """ + Parse bank statement PDFs and extract transaction data for financial analysis. + + Attributes: + pdf_path (str): Path to the bank statement PDF file + df (pd.DataFrame): Extracted transaction data + """ + + def __init__(self, pdf_path: str): + """ + Initialize the parser with a PDF file. + + Args: + pdf_path (str): Full path to the bank statement PDF + + Raises: + FileNotFoundError: If PDF file does not exist + """ + if not os.path.exists(pdf_path): + raise FileNotFoundError(f"PDF file not found: {pdf_path}") + + self.pdf_path = pdf_path + self.df = None + self.temp_csv = "temp_statement.csv" + + def extract_from_pdf(self, pages: str = 'all') -> pd.DataFrame: + """ + Extract transaction table from PDF using Tabula. + + Args: + pages (str): Pages to extract ('all', '1', '1-3', etc.) + + Returns: + pd.DataFrame: Extracted data as DataFrame + + Raises: + Exception: If PDF extraction fails + """ + try: + print(f"šŸ“„ Extracting tables from {self.pdf_path}...") + + # Convert PDF to CSV using Tabula + tabula.convert_into( + self.pdf_path, + self.temp_csv, + output_format="csv", + pages=pages, + multiple_tables=False + ) + + # Load CSV + self.df = pd.read_csv(self.temp_csv) + print(f"āœ… Successfully extracted {len(self.df)} transactions") + + return self.df + + except Exception as e: + print(f"āŒ Error extracting PDF: {e}") + raise + + def standardize_columns(self, + date_col: Optional[str] = None, + amount_col: Optional[str] = None, + description_col: Optional[str] = None) -> pd.DataFrame: + """ + Standardize column names and data types for DCF analysis. + + Args: + date_col (str): Name of the date column in the PDF + amount_col (str): Name of the amount/value column + description_col (str): Name of the description column + + Returns: + pd.DataFrame: Cleaned and standardized DataFrame + """ + if self.df is None: + raise ValueError("No data extracted. Call extract_from_pdf() first.") + + # Attempt auto-detection if columns not specified + if date_col is None: + date_col = self._detect_column(['Date', 'date', 'Transaction Date', 'Date of Transaction']) + + if amount_col is None: + amount_col = self._detect_column(['Amount', 'amount', 'Value', 'value', 'Debit', 'Credit']) + + if description_col is None: + description_col = self._detect_column(['Description', 'description', 'Details', 'Particulars']) + + # Rename columns + rename_dict = {} + if date_col: + rename_dict[date_col] = 'Date' + if amount_col: + rename_dict[amount_col] = 'Amount' + if description_col: + rename_dict[description_col] = 'Description' + + self.df.rename(columns=rename_dict, inplace=True) + + # Convert Date to datetime + if 'Date' in self.df.columns: + self.df['Date'] = pd.to_datetime(self.df['Date'], errors='coerce') + + # Convert Amount to float + if 'Amount' in self.df.columns: + self.df['Amount'] = pd.to_numeric(self.df['Amount'], errors='coerce') + + # Remove rows with NaN dates or amounts + self.df = self.df.dropna(subset=['Date', 'Amount']) + + print(f"āœ… Standardized {len(self.df)} valid transactions") + return self.df + + def _detect_column(self, column_options: list) -> Optional[str]: + """ + Auto-detect column name from possible variations. + + Args: + column_options (list): List of possible column names to search for + + Returns: + str or None: Found column name or None + """ + for col in column_options: + if col in self.df.columns: + return col + return None + + def calculate_cash_flows(self, + frequency: str = 'M', + include_positive: bool = True, + include_negative: bool = True) -> pd.Series: + """ + Calculate aggregated cash flows by time period. + + Args: + frequency (str): Frequency for aggregation + ('D'=Daily, 'W'=Weekly, 'M'=Monthly, 'Q'=Quarterly, 'Y'=Yearly) + include_positive (bool): Include positive cash flows (inflows) + include_negative (bool): Include negative cash flows (outflows) + + Returns: + pd.Series: Aggregated cash flows by period + """ + if self.df is None or 'Date' not in self.df.columns or 'Amount' not in self.df.columns: + raise ValueError("No valid data. Call extract_from_pdf() and standardize_columns() first.") + + df_copy = self.df.copy() + + # Filter by positive/negative + if include_positive and not include_negative: + df_copy = df_copy[df_copy['Amount'] > 0] + elif include_negative and not include_positive: + df_copy = df_copy[df_copy['Amount'] < 0] + + # Aggregate by period + cash_flows = df_copy.groupby(df_copy['Date'].dt.to_period(frequency))['Amount'].sum() + + return cash_flows + + def get_summary_stats(self) -> dict: + """ + Get summary statistics of the bank statement. + + Returns: + dict: Summary statistics including total inflows, outflows, net, etc. + """ + if self.df is None: + raise ValueError("No data extracted. Call extract_from_pdf() first.") + + inflows = self.df[self.df['Amount'] > 0]['Amount'].sum() + outflows = self.df[self.df['Amount'] < 0]['Amount'].sum() + net = inflows + outflows + + stats = { + 'Total Inflows': round(inflows, 2), + 'Total Outflows': round(outflows, 2), + 'Net Cash Flow': round(net, 2), + 'Transaction Count': len(self.df), + 'Period Start': self.df['Date'].min(), + 'Period End': self.df['Date'].max(), + 'Average Transaction': round(self.df['Amount'].mean(), 2) + } + + return stats + + def export_to_csv(self, output_path: str = 'statement_cleaned.csv') -> str: + """ + Export cleaned statement to CSV file. + + Args: + output_path (str): Output file path + + Returns: + str: Path to exported file + """ + if self.df is None: + raise ValueError("No data to export.") + + self.df.to_csv(output_path, index=False) + print(f"āœ… Exported to {output_path}") + return output_path + + def cleanup(self): + """Remove temporary files.""" + if os.path.exists(self.temp_csv): + os.remove(self.temp_csv) + print(f"āœ… Cleaned up temporary files") + + +def parse_bank_statement( + pdf_path: str, + date_col: Optional[str] = None, + amount_col: Optional[str] = None, + description_col: Optional[str] = None +) -> Tuple[pd.DataFrame, dict]: + """ + Convenience function to parse bank statement in one call. + + Args: + pdf_path (str): Path to bank statement PDF + date_col (str): Column name for dates + amount_col (str): Column name for amounts + description_col (str): Column name for descriptions + + Returns: + Tuple[pd.DataFrame, dict]: Cleaned DataFrame and summary statistics + """ + parser = BankStatementParser(pdf_path) + parser.extract_from_pdf() + parser.standardize_columns(date_col, amount_col, description_col) + + stats = parser.get_summary_stats() + + return parser.df, stats + + +if __name__ == "__main__": + # Example usage + print("=" * 60) + print("Bank Statement PDF Parser - Example") + print("=" * 60) + + # Replace with your PDF path + pdf_file = "sample_bank_statement.pdf" + + if os.path.exists(pdf_file): + parser = BankStatementParser(pdf_file) + + # Extract from PDF + df = parser.extract_from_pdf() + print("\nšŸ“Š Raw Data Sample:") + print(df.head()) + + # Standardize columns + df_clean = parser.standardize_columns() + print("\nšŸ“Š Cleaned Data Sample:") + print(df_clean.head()) + + # Get statistics + stats = parser.get_summary_stats() + print("\nšŸ“ˆ Summary Statistics:") + for key, value in stats.items(): + print(f" {key}: {value}") + + # Calculate monthly cash flows + monthly_cf = parser.calculate_cash_flows(frequency='M') + print("\nšŸ“… Monthly Cash Flows:") + print(monthly_cf) + + # Export cleaned data + parser.export_to_csv() + + # Cleanup + parser.cleanup() + else: + print(f"āš ļø Please provide a bank statement PDF at: {pdf_file}")