Add Tabula-py bank statement PDF parser for DCF integration

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}")