Banking Integration Guide

from fin_infra.banking import easy_banking

# Auto-detects provider and credentials from environment
banking = easy_banking()

# Use immediately
accounts = banking.accounts(access_token="user_token")
transactions = banking.transactions(
    access_token="user_token",
    start_date="2025-01-01",
    end_date="2025-01-31"
)

# Use Teller (default)
banking = easy_banking(provider="teller")

# Use Plaid
banking = easy_banking(provider="plaid")

from fin_infra.providers.banking.teller_client import TellerClient

# Teller with certificates
teller = TellerClient(
    cert_path="./teller_certificate.pem",
    key_path="./teller_private_key.pem",
    environment="sandbox"  # or "production"
)

TELLER_CERTIFICATE_PATH=./teller_certificate.pem  # Path to signed certificate
TELLER_PRIVATE_KEY_PATH=./teller_private_key.pem  # Path to private key
TELLER_ENVIRONMENT=sandbox                        # sandbox | production

PLAID_CLIENT_ID=your_client_id_here
PLAID_SECRET=your_secret_here
PLAID_ENVIRONMENT=sandbox  # sandbox | development | production

from fin_infra.banking import easy_banking

banking = easy_banking()

# Get all accounts for user
accounts = banking.accounts(access_token="user_access_token")

for account in accounts:
    print(f"{account.name}: {account.balance} {account.currency}")
    print(f"Type: {account.type}")  # checking, savings, credit_card, etc.
    print(f"Institution: {account.institution_name}")

from fin_infra.models import BankAccount

# Pydantic model with validation
account = BankAccount(
    id="acc_123",
    name="My Checking",
    official_name="Premium Checking Account",
    type="depository",
    subtype="checking",
    balance=5432.10,
    currency="USD",
    institution_id="ins_001",
    institution_name="Chase Bank"
)

from datetime import date, timedelta

banking = easy_banking()

# Last 30 days
end_date = date.today()
start_date = end_date - timedelta(days=30)

transactions = banking.transactions(
    access_token="user_access_token",
    start_date=start_date.isoformat(),
    end_date=end_date.isoformat()
)

for txn in transactions:
    print(f"{txn.date}: {txn.description} - ${txn.amount}")
    print(f"Category: {txn.category}")
    print(f"Merchant: {txn.merchant_name}")

from fin_infra.models import Transaction

txn = Transaction(
    id="txn_123",
    account_id="acc_123",
    amount=-45.67,  # Negative = debit, Positive = credit
    date="2025-01-15",
    description="STARBUCKS #12345",
    merchant_name="Starbucks",
    category="Food & Drink",
    pending=False,
    currency="USD"
)

# ⚠️ Handle PII carefully - see Security section
identity = banking.identity(access_token="user_access_token")

for owner in identity.owners:
    print(f"Name: {owner.name}")
    print(f"Email: {owner.email}")
    print(f"Phone: {owner.phone}")
    print(f"Address: {owner.address}")

# Lightweight call for balance updates only
balances = banking.balances(access_token="user_access_token")

for bal in balances:
    print(f"Account {bal.account_id}: ${bal.current}")
    print(f"Available: ${bal.available}")

from fastapi import FastAPI, Depends, HTTPException
from svc_infra.api.fastapi.ease import easy_service_app
from svc_infra.cache import init_cache
from fin_infra.banking import easy_banking

# Backend from svc-infra
app = easy_service_app(name="BankingAPI")
init_cache(url="redis://localhost")

# Financial provider from fin-infra
banking = easy_banking()

@app.get("/accounts")
async def get_accounts(access_token: str):
    """Fetch user's bank accounts"""
    try:
        accounts = banking.accounts(access_token=access_token)
        return {"accounts": accounts}
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))

@app.get("/transactions")
async def get_transactions(
    access_token: str,
    start_date: str,
    end_date: str
):
    """Fetch user's transactions"""
    txns = banking.transactions(
        access_token=access_token,
        start_date=start_date,
        end_date=end_date
    )
    return {"transactions": txns}

from svc_infra.cache import cache_read, cache_write, resource

# Define cached resource
bank = resource("bank_account", "user_id")

@app.get("/accounts/{user_id}")
@bank.cache_read(ttl=300, suffix="accounts")  # 5 min cache
async def get_cached_accounts(user_id: str, access_token: str):
    """Cached account fetch"""
    return banking.accounts(access_token=access_token)

@app.post("/accounts/{user_id}/refresh")
@bank.cache_write(tags=["accounts"], recache=[...])
async def refresh_accounts(user_id: str, access_token: str):
    """Invalidate cache and refresh"""
    return banking.accounts(access_token=access_token)

from svc_infra.api.fastapi.auth.add import add_auth_users
from svc_infra.api.fastapi.auth.dependencies import get_current_user

# Add auth to app
add_auth_users(app, ...)

@app.get("/me/accounts")
async def my_accounts(user = Depends(get_current_user)):
    """Get current user's accounts"""
    # Retrieve access_token from user's encrypted storage
    access_token = decrypt_token(user.bank_token_encrypted)
    return banking.accounts(access_token=access_token)

from fin_infra.banking import add_banking

# ✅ Mount complete banking API with one call
banking_provider = add_banking(
    app,
    provider="teller",  # or "plaid" (optional, defaults to env)
    prefix="/banking"   # default: "/banking"
)

# Auto-generated routes (using svc-infra dual routers):
# POST   /banking/link            - Create Plaid Link token or Teller enrollment URL
# POST   /banking/exchange        - Exchange public token for access token
# GET    /banking/accounts        - Fetch user's bank accounts
# GET    /banking/transactions    - Fetch transactions with date range
# GET    /banking/balances        - Fetch account balances only
# GET    /banking/identity        - Fetch identity/owner information (PII)

# Landing page card automatically registered at /banking/docs
# OpenAPI schema available at /banking/openapi.json

from fastapi import FastAPI, HTTPException, Depends
from svc_infra.api.fastapi.ease import easy_service_app
from svc_infra.logging import setup_logging
from svc_infra.cache import init_cache
from svc_infra.obs import add_observability
from svc_infra.api.fastapi.auth.add import add_auth_users
from svc_infra.api.fastapi.auth.dependencies import get_current_user
from fin_infra.banking import add_banking, easy_banking

# 1. Setup logging (svc-infra)
setup_logging(level="INFO", fmt="json")

# 2. Create service app (svc-infra)
app = easy_service_app(
    name="FinanceAPI",
    release="production",
    api_version="v1"
)

# 3. Initialize cache (svc-infra)
init_cache(url="redis://localhost:6379", prefix="finapi", version="v1")

# 4. Add observability (svc-infra)
shutdown_obs = add_observability(
    app,
    metrics_path="/metrics",
    skip_metric_paths=["/health", "/metrics"]
)

# 5. Add auth (svc-infra)
add_auth_users(
    app,
    db_url="postgresql://user:pass@localhost/db",
    secret_key="your-secret-key",
    prefix="/auth"
)

# 6. Add banking (fin-infra) - One-liner!
banking_provider = add_banking(app, provider="teller", prefix="/banking")

# 7. Custom protected routes using banking provider
from svc_infra.cache import resource

bank_cache = resource("bank", "user_id")

@app.get("/me/accounts")
@bank_cache.cache_read(ttl=60, suffix="accounts")
async def my_accounts(user=Depends(get_current_user)):
    """Get current user's bank accounts (cached 60s)"""
    if not user.bank_access_token:
        raise HTTPException(status_code=404, detail="No bank connected")
    
    # Use banking provider from app.state or returned instance
    accounts = banking_provider.accounts(access_token=user.bank_access_token)
    return {"accounts": accounts}

@app.get("/me/transactions")
@bank_cache.cache_read(ttl=300, suffix="transactions")  # 5 min cache
async def my_transactions(
    user=Depends(get_current_user),
    start_date: str = "2025-01-01",
    end_date: str = "2025-01-31"
):
    """Get current user's transactions (cached 5min)"""
    if not user.bank_access_token:
        raise HTTPException(status_code=404, detail="No bank connected")
    
    txns = banking_provider.transactions(
        access_token=user.bank_access_token,
        start_date=start_date,
        end_date=end_date
    )
    return {"transactions": txns, "count": len(txns)}

# 8. Cleanup on shutdown
@app.on_event("shutdown")
async def shutdown_event():
    shutdown_obs()  # Cleanup observability resources

# Set environment variables
export SQL_URL="postgresql://user:pass@localhost/db"
export REDIS_URL="redis://localhost:6379"
export TELLER_CERT_PATH="/path/to/certificate.pem"
export TELLER_KEY_PATH="/path/to/private_key.key"
export APP_SECRET_KEY="your-secret-key"

# Start server
uvicorn main:app --reload

# API available at:
# - Docs: http://localhost:8000/docs
# - Banking card: http://localhost:8000/banking/docs
# - Auth endpoints: http://localhost:8000/auth/*
# - Banking endpoints: http://localhost:8000/banking/*
# - Custom endpoints: http://localhost:8000/me/accounts

from fastapi import FastAPI
from fin_infra.banking import add_banking

app = FastAPI(title="Banking API")

# One-liner setup
add_banking(app, provider="teller")

# That's it! 6 endpoints ready to use:
# POST /banking/link
# POST /banking/exchange  
# GET  /banking/accounts
# GET  /banking/transactions
# GET  /banking/balances
# GET  /banking/identity

from fin_infra.banking import easy_banking

# Initialize provider
banking = easy_banking(provider="teller")

# Use directly in scripts, background jobs, etc.
access_token = "user_access_token"

accounts = banking.accounts(access_token=access_token)
for acc in accounts:
    print(f"{acc.name}: ${acc.balance}")

transactions = banking.transactions(
    access_token=access_token,
    start_date="2025-01-01",
    end_date="2025-01-31"
)
print(f"Found {len(transactions)} transactions")

from svc_infra.jobs.easy import easy_jobs
from fin_infra.banking import easy_banking

# Setup jobs (svc-infra)
worker, scheduler = easy_jobs(app, redis_url="redis://localhost:6379")

# Banking provider
banking = easy_banking()

@worker.task
async def sync_bank_data(user_id: str, access_token: str):
    """Background job to sync bank data"""
    accounts = banking.accounts(access_token=access_token)
    transactions = banking.transactions(
        access_token=access_token,
        start_date="2025-01-01",
        end_date="2025-01-31"
    )
    
    # Store in database (using svc-infra DB utilities)
    await db.save_accounts(user_id, accounts)
    await db.save_transactions(user_id, transactions)
    
    return {"accounts": len(accounts), "transactions": len(transactions)}

# Schedule daily sync
@scheduler.scheduled_job("cron", hour=2, minute=0)  # 2 AM daily
async def daily_bank_sync():
    """Sync all users' bank data"""
    users = await db.get_users_with_banks()
    for user in users:
        await sync_bank_data.kiq(user.id, user.bank_access_token)

from svc_infra.security import encrypt_field, decrypt_field

# Encrypt before storing
encrypted_token = encrypt_field(access_token, key=app_encryption_key)
await db.execute(
    "INSERT INTO user_tokens (user_id, token_encrypted) VALUES (?, ?)",
    (user_id, encrypted_token)
)

# Decrypt when using
encrypted = await db.fetchone("SELECT token_encrypted FROM user_tokens WHERE user_id = ?", (user_id,))
access_token = decrypt_field(encrypted[0], key=app_encryption_key)

# Never store plaintext tokens/PII
await db.execute(
    "INSERT INTO users (user_id, access_token, ssn) VALUES (?, ?, ?)",
    (user_id, access_token, ssn)  # ❌ SECURITY VIOLATION
)

from svc_infra.data.add import add_data_lifecycle

# Auto-expire PII per retention policies
add_data_lifecycle(
    app,
    retention_policies={
        "transactions": 90,  # days
        "identity": 30
    },
    anonymization_rules={...}
)

from svc_infra.api.fastapi.middlewares.rate_limit import SimpleRateLimit

# Protect banking endpoints
app.add_middleware(
    SimpleRateLimit,
    limit=100,  # requests
    window=60,  # seconds
    paths=["/api/v1/banking/*"]
)

from svc_infra.http import with_retries

# Auto-retry on 429/500 errors
@with_retries(max_attempts=3, backoff=2.0)
async def fetch_accounts_with_retry(access_token: str):
    return banking.accounts(access_token=access_token)

# Acceptance tests auto-load .env via conftest.py
# For manual scripts:
from dotenv import load_dotenv
load_dotenv()

import logging

# Enable debug logging
logging.basicConfig(level=logging.DEBUG)
logging.getLogger("httpx").setLevel(logging.DEBUG)

# See all HTTP requests/responses
banking = easy_banking()
accounts = banking.accounts(access_token="...")  # Watch console for details

from fin_infra.banking import easy_banking

# Try Teller first, fallback to Plaid
try:
    banking = easy_banking(provider="teller")
except ValueError:
    banking = easy_banking(provider="plaid")

import os

env = os.getenv("APP_ENV", "development")

if env == "production":
    os.environ["TELLER_ENVIRONMENT"] = "production"
    os.environ["PLAID_ENVIRONMENT"] = "production"
else:
    os.environ["TELLER_ENVIRONMENT"] = "sandbox"
    os.environ["PLAID_ENVIRONMENT"] = "sandbox"

banking = easy_banking()  # Auto-uses correct environment

### Transaction
```python
from fin_infra.models.transactions import Transaction

class Transaction:
    transaction_id: str
    account_id: str
    amount: Decimal
    date: date
    name: str
    merchant_name: str | None
    category: list[str]
    pending: bool
    iso_currency_code: str

from fastapi import FastAPI, Request
from fin_infra.banking.webhooks import verify_plaid_webhook

app = FastAPI()

@app.post("/webhooks/plaid")
async def plaid_webhook(request: Request):
    payload = await request.json()
    
    # Verify webhook signature
    if not verify_plaid_webhook(request.headers, payload):
        return {"error": "Invalid signature"}
    
    webhook_type = payload.get("webhook_type")
    webhook_code = payload.get("webhook_code")
    
    if webhook_type == "TRANSACTIONS":
        if webhook_code == "INITIAL_UPDATE":
            # Initial transaction data available
            pass
        elif webhook_code == "DEFAULT_UPDATE":
            # New transaction data available
            pass
        elif webhook_code == "HISTORICAL_UPDATE":
            # Historical transaction data available
            pass
    
    return {"status": "received"}

from fin_infra.banking.exceptions import (
    BankingProviderError,
    InvalidCredentialsError,
    ItemLoginRequiredError,
    RateLimitError
)

try:
    accounts = await banking.get_accounts(access_token)
except ItemLoginRequiredError:
    # User needs to re-authenticate with their bank
    link_token = await banking.create_link_token(
        user_id="user_123",
        access_token=access_token,  # Update mode
    )
except RateLimitError:
    # Implement exponential backoff
    pass
except BankingProviderError as e:
    # Handle general provider errors
    print(f"Error: {e.message}")

from fin_infra.banking import add_banking

# Mount banking API
banking = add_banking(app)

# Client makes request with filters:
# GET /banking/transactions?
#   access_token=<token>
#   &start_date=2025-01-01
#   &end_date=2025-01-31
#   &min_amount=10.00
#   &max_amount=100.00
#   &merchant_name=starbucks
#   &category=Food & Drink
#   &account_id=acc_123
#   &exclude_pending=true
#   &sort_by=date
#   &sort_order=desc
#   &page=1
#   &page_size=50

from fastapi import FastAPI, Query
from fin_infra.banking import easy_banking

banking = easy_banking()

@app.get("/transactions/filtered")
async def get_filtered_transactions(
    access_token: str = Query(...),
    start_date: str = Query(...),
    end_date: str = Query(...),
    min_amount: float | None = Query(None),
    category: str | None = Query(None),
):
    """Fetch transactions with filters"""
    txns = banking.transactions(
        access_token=access_token,
        start_date=start_date,
        end_date=end_date,
        min_amount=min_amount,
        category=category,
    )
    return {"transactions": txns, "count": len(txns)}

GET /banking/transactions?min_amount=500&sort_by=amount&sort_order=desc

GET /banking/transactions?merchant_name=amazon&start_date=2025-01-01

GET /banking/transactions?category=Food & Drink&start_date=2025-01-01&end_date=2025-01-31

GET /banking/transactions?exclude_pending=true&sort_by=date

from fin_infra.banking.history import track_balance, get_balance_history, BalanceHistory
from datetime import date

# Record a balance snapshot
track_balance(
    account_id="acc_123",
    balance=5432.10,
    date=date.today(),
    currency="USD"
)

# Retrieve balance history
history = get_balance_history(
    account_id="acc_123",
    start_date=date(2025, 1, 1),
    end_date=date(2025, 1, 31)
)

for snapshot in history:
    print(f"{snapshot.date}: ${snapshot.balance}")

from pydantic import BaseModel
from datetime import date

class BalanceHistory(BaseModel):
    account_id: str         # Account identifier
    balance: float          # Account balance at snapshot time
    date: date             # Snapshot date
    currency: str          # Currency code (USD, EUR, etc.)

# GET /banking/balance-history?account_id=acc_123&start_date=2025-01-01&end_date=2025-01-31

# Response:
{
  "account_id": "acc_123",
  "history": [
    {"date": "2025-01-01", "balance": 5000.00, "currency": "USD"},
    {"date": "2025-01-02", "balance": 4950.00, "currency": "USD"},
    {"date": "2025-01-03", "balance": 5100.00, "currency": "USD"},
    ...
  ],
  "count": 31
}

# Track balances across all accounts daily
for account in accounts:
    track_balance(
        account_id=account.id,
        balance=account.balance,
        date=date.today(),
        currency=account.currency
    )

# Generate net worth chart
total_history = {}
for account in accounts:
    history = get_balance_history(account.id, start_date, end_date)
    for snapshot in history:
        total_history[snapshot.date] = total_history.get(snapshot.date, 0) + snapshot.balance

# Identify accounts with declining balances
history = get_balance_history("acc_123", start_date, end_date)
trend = "increasing" if history[-1].balance > history[0].balance else "decreasing"
change_pct = ((history[-1].balance - history[0].balance) / history[0].balance) * 100

# Calculate average daily balance for interest/fee calculations
history = get_balance_history("acc_123", start_date, end_date)
avg_balance = sum(h.balance for h in history) / len(history)

# Use svc-infra SQL for persistent storage
from svc_infra.db.sql import add_sql_db

# Add to your database models:
class BalanceHistoryModel(Base):
    __tablename__ = "balance_history"
    
    id = Column(Integer, primary_key=True)
    account_id = Column(String, index=True)
    balance = Column(Float)
    date = Column(Date, index=True)
    currency = Column(String)
    
    __table_args__ = (
        Index('idx_account_date', 'account_id', 'date'),
    )

from svc_infra.jobs import easy_jobs

# Schedule daily balance tracking
worker, scheduler = easy_jobs(app)

@scheduler.scheduled_job('cron', hour=2)  # 2 AM daily
async def track_daily_balances():
    """Record balance snapshots for all users"""
    for user in users:
        accounts = banking.accounts(user.access_token)
        for account in accounts:
            track_balance(
                account_id=account.id,
                balance=account.balance,
                date=date.today(),
                currency=account.currency
            )

import pytest
from fin_infra.banking import easy_banking

@pytest.mark.asyncio
async def test_get_accounts():
    banking = easy_banking()
    
    # Use sandbox credentials
    access_token = "access-sandbox-xxx"
    accounts = await banking.get_accounts(access_token)
    
    assert len(accounts) > 0
    assert accounts[0].account_id is not None