Spaces:

MJ-Prod
/

Fiscal

Sleeping

Fiscal / plaid_client.py
csv Data_4
e7b9446 3 months ago
25 kB
"""
plaid_client.py — real Plaid + fixture fallback for trial mode
Real endpoints (get_financial_snapshot, get_chart_data, get_recurring_transactions)
hit real Plaid. Trial-mode variants (get_snapshot_from_fixtures, etc.) load from 
fixtures/raw_balances.json and fixtures/raw_transactions.json.
Called from app.py:
- Signed-in users → get_financial_snapshot / get_chart_data / get_recurring_transactions
- Trial users → get_snapshot_from_fixtures / get_chart_data_from_fixtures / get_recurring_from_fixtures
"""

import os
import json
from datetime import date, timedelta
from pathlib import Path
from collections import defaultdict
import re

from plaid.api import plaid_api
from plaid.configuration import Configuration
from plaid.api_client import ApiClient
from plaid.model.products import Products
from plaid.model.country_code import CountryCode
from plaid.exceptions import ApiException
from plaid.model.accounts_balance_get_request import AccountsBalanceGetRequest
from plaid.model.transactions_get_request import TransactionsGetRequest
from plaid.model.transactions_get_request_options import TransactionsGetRequestOptions
from plaid.model.link_token_create_request import LinkTokenCreateRequest
from plaid.model.link_token_create_request_user import LinkTokenCreateRequestUser
from plaid.model.item_public_token_exchange_request import ItemPublicTokenExchangeRequest


# ---------- Client setup ----------

def _make_client():
    env = os.environ.get("PLAID_ENV", "sandbox")
    host = {
        "sandbox":     "https://sandbox.plaid.com",
        "development": "https://development.plaid.com",
        "production":  "https://production.plaid.com",
    }[env]
    config = Configuration(
        host=host,
        api_key={
            "clientId": os.environ["PLAID_CLIENT_ID"],
            "secret":   os.environ["PLAID_SECRET"]
        }
    )
    return plaid_api.PlaidApi(ApiClient(config))

plaid_client = _make_client()


# ---------- Fixture loading (trial mode only) ----------

FIXTURES_DIR = Path(__file__).parent / "fixtures"

def _load_raw_balances():
    path = FIXTURES_DIR / "raw_balances.json"
    if not path.exists():
        raise FileNotFoundError(f"Fixture not found: {path}")
    return json.loads(path.read_text())

from datetime import date, timedelta

def _shift_fixture_to_today(fixture):
    """
    Slide all transaction dates forward so the newest transaction
    lands on today. Preserves relative gaps between transactions
    so recurring detection still works correctly.
    """
    txns = fixture.get('transactions', [])
    if not txns:
        return fixture

    # Find the latest date in the fixture
    def parse_date(d):
        return date.fromisoformat(d) if isinstance(d, str) else d

    latest = max(parse_date(t['date']) for t in txns)
    today = date.today()
    shift_days = (today - latest).days

    if shift_days <= 0:
        return fixture  # already current or in the future

    # Shift every transaction date forward by shift_days
    for t in txns:
        original = parse_date(t['date'])
        t['date'] = str(original + timedelta(days=shift_days))

    return fixture


def _load_raw_transactions():
    path = FIXTURES_DIR / "raw_transactions.json"
    if not path.exists():
        raise FileNotFoundError(f"Fixture not found: {path}")
    fixture = json.loads(path.read_text())
    return _shift_fixture_to_today(fixture)

# ---------- REAL: Balances ----------

def get_balances(access_token: str) -> str:
    request = AccountsBalanceGetRequest(access_token=access_token)
    response = plaid_client.accounts_balance_get(request)
    return _format_balances(response)


def _format_balances(response) -> str:
    """Shared formatter used by both real and fixture paths."""
    lines = ["USER'S CURRENT ACCOUNT BALANCES:"]
    for account in response['accounts']:
        name    = account['name']
        subtype = str(account['subtype'])
        current = account['balances']['current']

        if current is None:
            continue
        if subtype in ['mortgage', 'line of credit']:
            continue

        if subtype == 'credit card':
            limit     = account['balances']['limit'] or 0
            owing     = current
            available = limit - current if limit > 0 else 0
            lines.append(
                f"- {name} (Credit Card): "
                f"${owing:.2f} owing, ${available:.2f} available out of ${limit:.2f} limit"
            )
        elif subtype == 'checking':
            available = account['balances']['available'] or current
            lines.append(f"- {name} (Chequing): ${current:.2f} balance, ${available:.2f} available")
        elif subtype == 'savings':
            lines.append(f"- {name} (Savings): ${current:.2f}")
        elif subtype == 'rrsp':
            lines.append(f"- {name} (RRSP): ${current:.2f}")
        else:
            lines.append(f"- {name} ({subtype}): ${current:.2f}")

    return "\n".join(lines)


# ---------- REAL: Transactions ----------

def get_transactions(access_token: str, days: int = 30) -> str:
    end_date = date.today()
    start_date = end_date - timedelta(days=days)

    request = TransactionsGetRequest(
        access_token=access_token,
        start_date=start_date,
        end_date=end_date,
        options=TransactionsGetRequestOptions(
            count=200,
            include_personal_finance_category=True
        )
    )
    response = plaid_client.transactions_get(request)
    return _format_transactions(response['transactions'], days)


def _format_transactions(transactions, days: int = 30) -> str:
    """Shared formatter used by both real and fixture paths."""
    end_date = date.today()
    week_start = end_date - timedelta(days=end_date.weekday())

    if not transactions:
        return "No transactions found in the last 30 days."

    today_total = 0.0
    week_total = 0.0
    month_total = 0.0
    income_total = 0.0
    category_totals: dict[str, float] = {}

    for txn in transactions:
        amount = txn['amount']
        name = txn['name']
        txn_date_val = txn['date']
        if isinstance(txn_date_val, str):
            txn_date = date.fromisoformat(txn_date_val)
        else:
            txn_date = txn_date_val
        category = txn.get('personal_finance_category', {}).get('primary', 'OTHER')

        if category in ['TRANSFER_IN', 'TRANSFER_OUT']:
            continue

        name_lower = name.lower()
        if category == 'LOAN_PAYMENTS' and any(w in name_lower for w in ['payroll', 'salary', 'direct dep', 'employer', 'wages']):
            income_total += amount
            continue

        if category == 'LOAN_PAYMENTS':
            continue

        if amount < 0:
            income_total += abs(amount)
            continue

        month_total += amount
        readable = category.replace("_", " ").title()
        category_totals[readable] = category_totals.get(readable, 0) + amount

        if txn_date >= week_start:
            week_total += amount

        if txn_date == end_date:
            today_total += amount

    lines = ["FINANCIAL SUMMARY (pre-calculated, do NOT recalculate):"]
    lines.append(f"Today's spending: ${today_total:.2f}")
    lines.append(f"This week's spending (since {week_start}): ${week_total:.2f}")
    lines.append(f"This month's spending (last {days} days): ${month_total:.2f}")

    if income_total > 0:
        lines.append(f"Income received (last {days} days): +${income_total:.2f}")

    lines.append(f"Average daily spending: ${month_total / max(days, 1):.2f}")

    lines.append("\nSPENDING BY CATEGORY:")
    for category, total in sorted(category_totals.items(), key=lambda x: -x[1]):
        pct = (total / month_total * 100) if month_total > 0 else 0
        lines.append(f"- {category}: ${total:.2f} ({pct:.0f}%)")

    lines.append("\nRECENT TRANSACTIONS (last 5):")
    count = 0
    for txn in transactions:
        if count >= 5:
            break
        amount = txn['amount']
        category = txn.get('personal_finance_category', {}).get('primary', 'OTHER')
        if amount <= 0 or category in ['TRANSFER_IN', 'TRANSFER_OUT', 'LOAN_PAYMENTS']:
            continue
        lines.append(f"  - {txn['date']} {txn['name']}: ${amount:.2f}")
        count += 1

    return "\n".join(lines)


# ---------- REAL: Combined snapshot ----------

def get_financial_snapshot(access_token: str) -> str:
    try:
        balances = get_balances(access_token)
        transactions = get_transactions(access_token)
        return f"{balances}\n\n{transactions}"
    except Exception as e:
        error_str = str(e)
        print(f"Plaid API Exception caught: {error_str}", flush=True)
        if "ITEM_LOGIN_REQUIRED" in error_str:
            return "BANK_REAUTH_REQUIRED"
        return "NO_BANK_DATA"


# ---------- REAL: Link tokens ----------

def create_update_link_token(user_id: str, access_token: str) -> str:
    request = LinkTokenCreateRequest(
        user=LinkTokenCreateRequestUser(client_user_id=user_id),
        client_name="Cheque Up",
        country_codes=[CountryCode("CA")],
        language="en",
        access_token=access_token,
        redirect_uri="https://chequeup.ca",
    )
    response = plaid_client.link_token_create(request)
    return response["link_token"]


def create_link_token(user_id: str) -> str:
    request = LinkTokenCreateRequest(
        user=LinkTokenCreateRequestUser(client_user_id=user_id),
        client_name="Cheque Up",
        products=[Products("transactions")],
        country_codes=[CountryCode("CA")],
        language="en",
        redirect_uri="https://chequeup.ca",
    )
    response = plaid_client.link_token_create(request)
    return response["link_token"]


def exchange_public_token(public_token: str) -> str:
    request = ItemPublicTokenExchangeRequest(public_token=public_token)
    response = plaid_client.item_public_token_exchange(request)
    return response["access_token"]


# ---------- REAL: Chart data ----------

def get_chart_data(access_token: str) -> dict:
    """Returns burn-down velocity chart data based on chequing account."""
    end_date = date.today()
    start_date = end_date.replace(day=1)

    try:
        balance_req = AccountsBalanceGetRequest(access_token=access_token)
        balance_resp = plaid_client.accounts_balance_get(balance_req)
        balance_data = balance_resp
    except Exception as e:
        print(f"Balance error in chart_data: {e}", flush=True)
        balance_data = {'accounts': []}

    try:
        request = TransactionsGetRequest(
            access_token=access_token,
            start_date=start_date,
            end_date=end_date,
            options=TransactionsGetRequestOptions(count=300, include_personal_finance_category=True)
        )
        response = plaid_client.transactions_get(request)
        transactions = response['transactions']
    except Exception as e:
        print(f"Transaction error in chart_data: {e}", flush=True)
        transactions = []

    return _build_chart_data(balance_data, transactions, start_date, end_date)


def _build_chart_data(balance_data, transactions, start_date, end_date) -> dict:
    """Shared chart builder used by both real and fixture paths."""
    accounts = []
    chequing_balance = 0.0

    for account in balance_data['accounts']:
        current = account['balances']['current']
        if current is None:
            continue
        subtype = str(account['subtype'])

        if subtype == 'checking':
            chequing_balance = current
        if subtype in ['mortgage', 'line of credit']:
            continue

        accounts.append({
            "name": account['name'],
            "type": subtype,
            "balance": round(current, 2),
        })

    burndown = []
    income_total = 0.0
    expense_total = 0.0
    daily_net: dict[str, float] = {}

    for txn in transactions:
        amount = txn['amount']
        txn_date = str(txn['date'])
        category = txn.get('personal_finance_category', {}).get('primary', 'OTHER')

        if category in ['TRANSFER_IN', 'TRANSFER_OUT']:
            continue

        name_lower = txn['name'].lower()
        if category == 'LOAN_PAYMENTS' and any(w in name_lower for w in ['payroll', 'salary', 'direct dep', 'employer', 'wages']):
            income_total += amount
            daily_net[txn_date] = daily_net.get(txn_date, 0) - amount
            continue

        if amount < 0:
            income_total += abs(amount)
            daily_net[txn_date] = daily_net.get(txn_date, 0) + abs(amount)
        else:
            expense_total += amount
            daily_net[txn_date] = daily_net.get(txn_date, 0) - amount

    daily_balances: dict[str, float] = {}
    running = chequing_balance
    current = end_date
    while current >= start_date:
        day_str = str(current)
        daily_balances[day_str] = running
        net_today = daily_net.get(day_str, 0)
        running = running - net_today
        current -= timedelta(days=1)

    current = start_date
    while current <= end_date:
        day_str = str(current)
        balance = daily_balances.get(day_str, 0)
        day_spend = 0
        for txn in transactions:
            if str(txn['date']) == day_str and txn['amount'] > 0:
                cat = txn.get('personal_finance_category', {}).get('primary', 'OTHER')
                if cat not in ['TRANSFER_IN', 'TRANSFER_OUT']:
                    day_spend += txn['amount']

        burndown.append({
            "date": current.strftime("%b %d"),
            "balance": round(balance, 2),
            "spent": round(day_spend, 2),
        })
        current += timedelta(days=1)

    return {
        "burndown": burndown,
        "income_total": round(income_total, 2),
        "expense_total": round(expense_total, 2),
        "accounts": accounts,
    }


# ---------- REAL: Recurring transactions ----------

def get_recurring_transactions(access_token: str) -> dict:
    """Analyze 6 months of transactions to find recurring patterns."""
    end_date = date.today()
    start_date = end_date - timedelta(days=180)

    try:
        all_transactions = []
        offset = 0
        while True:
            request = TransactionsGetRequest(
                access_token=access_token,
                start_date=start_date,
                end_date=end_date,
                options=TransactionsGetRequestOptions(
                    count=500,
                    offset=offset,
                    include_personal_finance_category=True
                )
            )
            response = plaid_client.transactions_get(request)
            all_transactions.extend(response['transactions'])
            if len(all_transactions) >= response['total_transactions']:
                break
            offset = len(all_transactions)

        return _detect_recurring(all_transactions, start_date, end_date)

    except Exception as e:
        error_str = str(e)
        print(f"Recurring transaction error: {error_str}", flush=True)
        not_ready = 'PRODUCT_NOT_READY' in error_str or 'not yet ready' in error_str
        return {
            'recurring_expenses': [],
            'recurring_income': [],
            'projected_events': [],
            'analysis_period': None,
            'not_ready': not_ready,
        }


def _detect_recurring(all_transactions, start_date, end_date) -> dict:
    """Shared recurring detector used by both real and fixture paths."""
    merchant_groups: dict[str, list] = defaultdict(list)

    for txn in all_transactions:
        name = txn['name']
        amount = txn['amount']
        txn_date = txn['date']
        category = txn.get('personal_finance_category', {}).get('primary', 'OTHER')

        if category in ['TRANSFER_IN', 'TRANSFER_OUT']:
            continue

        normalized = re.sub(r'[0-9#]+', '', name).strip()
        normalized = re.sub(r'\s+', ' ', normalized).upper()

        merchant_groups[normalized].append({
            'date': str(txn_date),
            'amount': amount,
            'name': name,
            'category': category,
        })

    recurring = []
    income = []

    for merchant, txns in merchant_groups.items():
        if len(txns) < 3:
            continue

        txns.sort(key=lambda x: x['date'])
        amounts = [t['amount'] for t in txns]
        avg_amount = sum(amounts) / len(amounts)
        if avg_amount == 0:
            continue
        variance = max(amounts) - min(amounts)
        if variance / abs(avg_amount) >= 0.15:
            continue

        dates = [date.fromisoformat(t['date']) for t in txns]
        gaps = [(dates[i+1] - dates[i]).days for i in range(len(dates)-1)]
        avg_gap = sum(gaps) / len(gaps) if gaps else 0

        if 5 <= avg_gap <= 10:
            frequency = 'weekly'
            freq_days = 7
        elif 12 <= avg_gap <= 17:
            frequency = 'biweekly'
            freq_days = 14
        elif 25 <= avg_gap <= 35:
            frequency = 'monthly'
            freq_days = 30
        else:
            continue

        months_present = set()
        for t in txns:
            d = date.fromisoformat(t['date'])
            months_present.add(f"{d.year}-{d.month:02d}")

        consecutive = len(months_present)
        is_income = avg_amount < 0

        entry = {
            'merchant': txns[0]['name'],
            'normalized_name': merchant,
            'amount': round(abs(avg_amount), 2),
            'frequency': frequency,
            'frequency_days': freq_days,
            'category': txns[0]['category'],
            'consecutive_months': consecutive,
            'last_date': txns[-1]['date'],
            'is_income': is_income,
            'history': [{'date': t['date'], 'amount': t['amount']} for t in txns],
        }

        if consecutive >= 5:
            entry['status'] = 'confirmed'
        elif consecutive >= 3:
            entry['status'] = 'likely'
        else:
            entry['status'] = 'possible'

        if is_income:
            income.append(entry)
        else:
            recurring.append(entry)

    recurring.sort(key=lambda x: -x['amount'])
    income.sort(key=lambda x: -x['amount'])

    projected_events = []
    today = date.today()

    for item in recurring + income:
        last = date.fromisoformat(item['last_date'])
        freq = item['frequency_days']
        next_date = last + timedelta(days=freq)
        while next_date <= today + timedelta(days=90):
            if next_date >= today:
                projected_events.append({
                    'date': str(next_date),
                    'merchant': item['merchant'],
                    'amount': item['amount'],
                    'is_income': item['is_income'],
                    'frequency': item['frequency'],
                    'status': item['status'],
                    'category': item['category'],
                })
            next_date += timedelta(days=freq)

    projected_events.sort(key=lambda x: x['date'])

    return {
        'recurring_expenses': recurring,
        'recurring_income': income,
        'projected_events': projected_events,
        'total_transactions': len(all_transactions),
        'analysis_period': {
            'start': str(start_date),
            'end': str(end_date),
            'total_transactions': len(all_transactions),
        }
    }


# ============================================================================
# TRIAL MODE: fixture-based variants
# Called by /trial/* endpoints in app.py — no Plaid API calls.
# ============================================================================

def get_snapshot_from_fixtures() -> str:
    """Trial mode: build financial snapshot from fixture data."""
    try:
        balances = _load_raw_balances()
        transactions_data = _load_raw_transactions()
        balance_text = _format_balances(balances)
        transaction_text = _format_transactions(transactions_data['transactions'])
        return f"{balance_text}\n\n{transaction_text}"
    except Exception as e:
        print(f"Fixture snapshot error: {e}", flush=True)
        return "NO_BANK_DATA"


def get_chart_data_from_fixtures() -> dict:
    """Trial mode: build chart data from fixture data."""
    end_date = date.today()
    start_date = end_date.replace(day=1)
    try:
        balances = _load_raw_balances()
        transactions_data = _load_raw_transactions()
        return _build_chart_data(balances, transactions_data['transactions'], start_date, end_date)
    except Exception as e:
        print(f"Fixture chart_data error: {e}", flush=True)
        return {"burndown": [], "income_total": 0, "expense_total": 0, "accounts": []}


def get_recurring_from_fixtures() -> dict:
    """Trial mode: detect recurring patterns from fixture data."""
    end_date = date.today()
    start_date = end_date - timedelta(days=180)
    try:
        transactions_data = _load_raw_transactions()
        return _detect_recurring(transactions_data['transactions'], start_date, end_date)
    except Exception as e:
        print(f"Fixture recurring error: {e}", flush=True)
        return {
            'recurring_expenses': [],
            'recurring_income': [],
            'projected_events': [],
            'analysis_period': None,
        }


def _build_synthetic_balances_from_transactions(transactions: list) -> dict:
    """
    CSV mode: infer account balances from transaction history.
    We don't have real balances (CSV doesn't include them), so we estimate:
    - For each account_id in the transactions, sum credits - debits to get an inferred balance
    - Use latest transaction as reference point
    
    Returns a Plaid-shaped response so downstream formatters work unchanged.
    """
    from collections import defaultdict
    
    account_totals = defaultdict(float)
    account_names = {}
    latest_dates = {}
    
    for txn in transactions:
        acc_id = txn.get('account_id', 'csv_account_default')
        amount = txn.get('amount', 0)
        # Plaid convention: positive = expense (money out), negative = income (money in)
        # So current balance = sum of (-amount) = money in minus money out
        account_totals[acc_id] += (-amount)
        
        # Track a display name — use account_id or default
        if acc_id not in account_names:
            account_names[acc_id] = f"Uploaded Account {acc_id[-4:]}" if len(acc_id) > 4 else "Uploaded Account"
        
        # Track latest date
        txn_date = txn.get('date', '')
        if acc_id not in latest_dates or txn_date > latest_dates[acc_id]:
            latest_dates[acc_id] = txn_date
    
    # Build a Plaid-shaped accounts response
    accounts = []
    for acc_id, running_total in account_totals.items():
        accounts.append({
            'account_id': acc_id,
            'name': account_names[acc_id],
            'subtype': 'checking',  # assume chequing for CSV uploads
            'balances': {
                'current': round(running_total, 2),
                'available': round(running_total, 2),
                'limit': None,
            }
        })
    
    if not accounts:
        # Fallback: single generic account
        accounts.append({
            'account_id': 'csv_account_default',
            'name': 'Uploaded Account',
            'subtype': 'checking',
            'balances': {'current': 0, 'available': 0, 'limit': None}
        })
    
    return {'accounts': accounts}
 
 
def get_snapshot_from_csv(transactions: list) -> str:
    """CSV mode: build financial snapshot from user-uploaded transactions."""
    try:
        balances = _build_synthetic_balances_from_transactions(transactions)
        balance_text = _format_balances(balances)
        transaction_text = _format_transactions(transactions)
        return f"{balance_text}\n\n{transaction_text}"
    except Exception as e:
        print(f"CSV snapshot error: {e}", flush=True)
        return "NO_BANK_DATA"
 
 
def get_chart_data_from_csv(transactions: list) -> dict:
    """CSV mode: build chart data from user-uploaded transactions."""
    end_date = date.today()
    start_date = end_date.replace(day=1)
    try:
        balances = _build_synthetic_balances_from_transactions(transactions)
        return _build_chart_data(balances, transactions, start_date, end_date)
    except Exception as e:
        print(f"CSV chart_data error: {e}", flush=True)
        return {"burndown": [], "income_total": 0, "expense_total": 0, "accounts": []}
 
 
def get_recurring_from_csv(transactions: list) -> dict:
    """CSV mode: detect recurring patterns from user-uploaded transactions."""
    end_date = date.today()
    start_date = end_date - timedelta(days=180)
    try:
        return _detect_recurring(transactions, start_date, end_date)
    except Exception as e:
        print(f"CSV recurring error: {e}", flush=True)
        return {
            'recurring_expenses': [],
            'recurring_income': [],
            'projected_events': [],
            'analysis_period': None,
        }