Recurring Charge Monitor

Detect price hikes in your bank transactions with Python and SQLite

Welcome to another official UVF IT build guide. Today, we are creating a minimal clone of Spendalyst, a tool that helps users spot when their recurring subscriptions quietly increase in price. Instead of relying on complex AI or bank APIs, we will build a local-first solution that analyzes a CSV of transactions to identify patterns and flag anomalies.

This project is perfect for beginners who want to understand data processing, pattern matching, and simple web interfaces. By the end of this 45-minute session, you will have a working Python application that reads your transaction history, groups recurring charges, and alerts you if a bill costs more than it did last month.

Before You Start

Recurring Charge Monitor Architecture
LOCAL RUNTIME POST /upload Insert Data Analyze Query History Browser Uploads CSV Flask App Python Server SQLite transactions.db Analysis Engine Pattern Matcher
1

Phase 1: Scaffold the Project

Create a new directory for your project and initialize a virtual environment. This keeps your dependencies isolated and prevents conflicts with other Python projects on your machine. We will name the directory recurring-monitor to keep things organized.

We will use Flask for the web interface and pandas for data manipulation. Installing these packages now ensures that the code in the following phases runs without import errors. Run the commands below to set up the environment.

mkdir recurring-monitor
cd recurring-monitor
python3 -m venv venv
source venv/bin/activate  # On Windows use: venv\Scripts\activate
pip install flask pandas
2

Phase 2: Data Model & Database

Define the structure for our transactions. We need to store the date, description, and amount to allow for historical comparison. This data will reside in a lightweight SQLite database, which requires no server setup and is perfect for local-first applications.

Create a simple SQLite database to hold this data. We will use a single table called transactions. The init_db function below ensures the table exists and creates it if it doesn't, making the script idempotent and safe to run multiple times.

database.py

import sqlite3
import os

DB_NAME = 'transactions.db'

def get_connection():
    """Create and return a database connection."""
    conn = sqlite3.connect(DB_NAME)
    conn.row_factory = sqlite3.Row
    return conn

def init_db():
    """Initialize the database and create the transactions table if it doesn't exist."""
    conn = get_connection()
    cursor = conn.cursor()
    
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS transactions (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            date TEXT NOT NULL,
            description TEXT NOT NULL,
            amount REAL NOT NULL
        )
    ''')
    
    # Add an index on description to speed up grouping queries
    cursor.execute('''
        CREATE INDEX IF NOT EXISTS idx_description 
        ON transactions(description)
    ''')
    
    conn.commit()
    conn.close()
    print(f"Database initialized: {DB_NAME}")

if __name__ == '__main__':
    init_db()
3

Phase 3: Core Logic - Pattern Detection

This is the heart of the clone. We need to identify recurring charges by grouping transactions with similar descriptions. Since bank descriptions can vary slightly (e.g., "NETFLIX.COM" vs "NETFLIX"), we normalize the text by converting it to uppercase and stripping whitespace.

We will calculate the average price for each group over the last 3 months. If the latest charge is significantly higher than the average (defined here as more than 5% higher), we flag it as a 'price hike'. This logic uses pandas to handle the date filtering and grouping efficiently.

analyzer.py

import pandas as pd
from datetime import datetime, timedelta
from database import get_connection


def analyze_recurring_charges(threshold_percent=5.0):
    """
    Analyze transactions to find recurring charges and detect price hikes.
    
    Args:
        threshold_percent: The percentage increase required to flag a price hike.
    
    Returns:
        A list of dictionaries containing recurring charge details.
    """
    conn = get_connection()
    
    # Fetch all transactions
    query = """
        SELECT date, description, amount 
        FROM transactions
        ORDER BY date DESC
    """
    df = pd.read_sql_query(query, conn)
    conn.close()
    
    if df.empty:
        return []
    
    # Convert date strings to datetime objects
    df['date'] = pd.to_datetime(df['date'])
    
    # Normalize description for grouping (uppercase, strip)
    df['normalized_desc'] = df['description'].str.upper().str.strip()
    
    # Define the lookback window (3 months)
    now = datetime.now()
    three_months_ago = now - timedelta(days=90)
    
    # Filter to the last 3 months for average calculation
    recent_df = df[df['date'] >= three_months_ago].copy()
    
    # Group by normalized description
    groups = recent_df.groupby('normalized_desc')
    
    results = []
    
    for name, group in groups:
        if len(group) < 2: 
            # Skip if it's not recurring (needs at least 2 occurrences)
            continue
            
        # Get the most recent transaction
        latest = group.iloc[0]
        
        # Calculate average of previous transactions (excluding the latest)
        previous_transactions = group.iloc[1:]
        if previous_transactions.empty:
            continue
            
        avg_price = previous_transactions['amount'].mean()
        latest_price = latest['amount']
        
        # Calculate percentage change
        if avg_price == 0:
            continue
            
        percent_change = ((latest_price - avg_price) / avg_price) * 100
        
        # Flag if price hike exceeds threshold
        is_price_hike = percent_change > threshold_percent
        
        results.append({
            'description': name,
            'latest_date': latest['date'].strftime('%Y-%m-%d'),
            'latest_amount': latest_price,
            'average_amount': round(avg_price, 2),
            'percent_change': round(percent_change, 2),
            'is_price_hike': is_price_hike,
            'occurrences': len(group)
        })
    
    # Sort by percent change descending to show biggest hikes first
    results.sort(key=lambda x: x['percent_change'], reverse=True)
    
    return results
4

Phase 4: API & UI

Now we wire everything together. We create a Flask application that handles the CSV upload, parses the data using pandas, and stores it in our SQLite database. This endpoint acts as the bridge between your browser and the analysis engine.

The second part of this phase is the frontend. We build a minimal HTML template with a file input and a results table. When the user uploads a CSV, the page will display the recurring charges and use CSS classes to highlight any 'price hikes' in red, making them easy to spot.

app.py

import pandas as pd
from flask import Flask, render_template, request, jsonify
from database import init_db, save_transactions
from analyzer import detect_recurring

app = Flask(__name__)

@app.route('/')
def index():
    return render_template('index.html')

@app.route('/upload', methods=['POST'])
def upload():
    if 'file' not in request.files:
        return jsonify({'error': 'No file part'}), 400
    
    file = request.files['file']
    if file.filename == '':
        return jsonify({'error': 'No selected file'}), 400

    # Read CSV into DataFrame
    try:
        df = pd.read_csv(file)
    except Exception as e:
        return jsonify({'error': f'Could not read CSV: {str(e)}'}), 400

    # Validate columns
    if not all(col in df.columns for col in ['date', 'description', 'amount']):
        return jsonify({'error': 'CSV must contain columns: date, description, amount'}), 400

    # Save to database
    records = df.to_dict('records')
    save_transactions(records)

    # Run analysis
    results = detect_recurring()
    
    return jsonify({'status': 'success', 'results': results})

if __name__ == '__main__':
    init_db()
    app.run(debug=True, port=5000)

templates/index.html

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>Recurring Charge Monitor</title>
    <style>
        body { font-family: Arial, sans-serif; max-width: 800px; margin: 0 auto; padding: 20px; }
        .upload-area { margin-bottom: 20px; padding: 15px; border: 1px solid #ccc; border-radius: 5px; }
        table { width: 100%; border-collapse: collapse; }
        th, td { padding: 10px; text-align: left; border-bottom: 1px solid #ddd; }
        .hike { background-color: #ffe6e6; color: #cc0000; font-weight: bold; }
        .normal { color: #333; }
        #status { margin-top: 10px; color: #0066cc; }
    </style>
</head>
<body>
    <h1>Recurring Charge Monitor</h1>
    <p>Upload a CSV with columns: <code>date</code>, <code>description</code>, <code>amount</code></p>

    <div class="upload-area">
        <input type="file" id="csvFile" accept=".csv">
        <button onclick="uploadCSV()">Analyze Transactions</button>
    </div>

    <div id="status"></div>

    <table id="resultsTable" style="display:none;">
        <thead>
            <tr>
                <th>Description</th>
                <th>Previous Avg</th>
                <th>Latest Charge</th>
                <th>Status</th>
            </tr>
        </thead>
        <tbody id="resultsBody"></tbody>
    </table>

    <script>
        function uploadCSV() {
            const fileInput = document.getElementById('csvFile');
            const statusDiv = document.getElementById('status');
            const table = document.getElementById('resultsTable');
            const tbody = document.getElementById('resultsBody');

            if (!fileInput.files[0]) {
                statusDiv.textContent = 'Please select a CSV file.';
                return;
            }

            const formData = new FormData();
            formData.append('file', fileInput.files[0]);

            statusDiv.textContent = 'Uploading and analyzing...';

            fetch('/upload', {
                method: 'POST',
                body: formData
            })
            .then(response => response.json())
            .then(data => {
                if (data.error) {
                    statusDiv.textContent = 'Error: ' + data.error;
                    return;
                }

                statusDiv.textContent = 'Analysis complete.';
                tbody.innerHTML = '';

                data.results.forEach(item => {
                    const tr = document.createElement('tr');
                    const isHike = item['status'] === 'hike';
                    
                    tr.innerHTML = `
                        <td>${item['description']}</td>
                        <td>$${item['previous_avg'].toFixed(2)}</td>
                        <td>$${item['latest_amount'].toFixed(2)}</td>
                        <td class="${isHike ? 'hike' : 'normal'}">${isHike ? 'PRICE HIKE' : 'Normal'}</td>
                    `;
                    tbody.appendChild(tr);
                });

                table.style.display = 'table';
            })
            .catch(error => {
                statusDiv.textContent = 'Upload failed: ' + error;
            });
        }
    </script>
</body>
</html>
5

Phase 5: Run & Verify

With the code in place, it's time to see it in action. Start the Flask development server. This will initialize the database and make your application available in the browser.

Open your browser and navigate to the local address. Upload a sample CSV file that contains recurring charges (like Netflix or Spotify) with at least one charge that is higher than the previous ones to verify the detection logic works correctly.

python app.py

Verifying the Stack

CheckCommand / Why it matters
Server starts without errorsRun python app.py and check the terminal for 'Running on http://127.0.0.1:5000'.
CSV upload worksUpload a sample CSV in the browser and ensure the page refreshes with results.
Recurring charges detectedLook for a list of items like 'Netflix' or 'Spotify' in the results table.
Price hike flaggedVerify that a charge with a higher amount than previous months is highlighted in red.

Next Steps

You now have a working recurring charge monitor! To extend this, you could add email notifications when a price hike is detected, or integrate with a bank API to fetch transactions automatically. Happy coding!