# ingestion/normalize.py
from pydantic import BaseModel
from datetime import datetime
from typing import Optional

class RawTransaction(BaseModel):
    external_id: str
    merchant: str
    amount: float
    currency: str
    timestamp: datetime
    category_hint: Optional[str] = None
    metadata: dict = {}

class NormalizedTransaction(BaseModel):
    id: str
    external_id: str
    merchant: str
    amount_cents: int
    currency: str
    posted_at: datetime
    category: str = "uncategorized"
    tags: list[str] = []
    source: str
    ingested_at: datetime = datetime.utcnow()

def normalize(raw: RawTransaction, source: str) -> NormalizedTransaction:
    return NormalizedTransaction(
        id=f"{source}_{raw.external_id}",
        external_id=raw.external_id,
        merchant=raw.merchant,
        amount_cents=int(raw.amount * 100),
        currency=raw.currency,
        posted_at=raw.timestamp,
        source=source
    )

# config/categories.yml
rules:
  - pattern: "aws|amazon web services"
    category: "cloud_infrastructure"
    confidence: 0.95
  - pattern: "stripe|paddle|paypal"
    category: "payment_processing"
    confidence: 0.90
  - pattern: "github|gitlab|bitbucket"
    category: "developer_tools"
    confidence: 0.92
  - pattern: "rent|lease|co-working"
    category: "office_expenses"
    confidence: 0.85
default:
  category: "uncategorized"
  route: "review_queue"

python

categorization/engine.py

import re import yaml

def load_rules(path: str) -> list[dict]: with open(path) as f: return yaml.safe_load(f)["rules"]

def classify(transaction: NormalizedTransaction, rules: list[dict]) -> NormalizedTransaction: merchant_lower = transaction.merchant.lower() for rule in rules: if re.search(rule["pattern"], merchant_lower): transaction.category = rule["category"] transaction.tags.append(f"rule:{rule['category']}") return transaction transaction.category = "uncategorized" transaction.tags.append("route:review_queue") return transaction

#### 3. Model Runway & Cash Flow
Runway is not static. It requires rolling projections based on burn rate, recurring revenue, and variable expenses. Use a simple but robust formula:

`Runway (months) = (Cash + Receivables) / (Monthly Burn + Variable Expense Run Rate)`

```python
# modeling/runway.py
from datetime import datetime, timedelta
from collections import defaultdict

def calculate_runway(transactions: list[NormalizedTransaction], cash_balance_cents: int) -> dict:
    monthly_revenue = defaultdict(int)
    monthly_expense = defaultdict(int)
    
    for tx in transactions:
        month_key = tx.posted_at.strftime("%Y-%m")
        if tx.amount_cents > 0:
            monthly_revenue[month_key] += tx.amount_cents
        else:
            monthly_expense[month_key] += abs(tx.amount_cents)
            
    # 3-month rolling average burn
    recent_months = sorted(monthly_expense.keys())[-3:]
    avg_monthly_burn = sum(monthly_expense[m] for m in recent_months) / len(recent_months) if recent_months else 0
    
    # 3-month rolling average revenue
    recent_rev_months = sorted(monthly_revenue.keys())[-3:]
    avg_monthly_rev = sum(monthly_revenue[m] for m in recent_rev_months) / len(recent_rev_months) if recent_rev_months else 0
    
    net_monthly_burn = avg_monthly_burn - avg_monthly_rev
    runway_months = (cash_balance_cents / net_monthly_burn) if net_monthly_burn > 0 else float('inf')
    
    return {
        "cash_balance_cents": cash_balance_cents,
        "avg_monthly_burn_cents": avg_monthly_burn,
        "avg_monthly_revenue_cents": avg_monthly_rev,
        "net_monthly_burn_cents": net_monthly_burn,
        "runway_months": round(runway_months, 2),
        "calculated_at": datetime.utcnow().isoformat()
    }

4. Reconcile & Alert

Schedule daily reconciliation via GitHub Actions or cron. Flag discrepancies, missing receipts, or runway thresholds.

# .github/workflows/finance-sync.yml
name: Daily Finance Reconciliation
on:
  schedule:
    - cron: "0 2 * * *"  # 2 AM UTC
jobs:
  sync:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.11"
      - name: Install dependencies
        run: pip install -r requirements.txt
      - name: Run ingestion & categorization
        env:
          STRIPE_API_KEY: ${{ secrets.STRIPE_API_KEY }}
          DB_URL: ${{ secrets.DB_URL }}
        run: python main.py --sync
      - name: Check runway threshold
        run: |
          RUNWAY=$(python -c "import json; print(json.load(open('output/runway.json'))['runway_months'])")
          if (( $(echo "$RUNWAY < 6" | bc -l) )); then
            echo "::warning::Runway below 6 months: ${RUNWAY} months"
          fi

5. Store & Version Configurations

Treat financial rules, categories, and alert thresholds as infrastructure. Store in Git. Enforce PR reviews for rule changes. This prevents silent drift and enables rollback.

Pitfall Guide

1. Commingling Personal and Business Transactions Mixing accounts corrupts categorization, breaks tax reporting, and pierces the corporate veil. Always route through a dedicated business entity and separate banking rails.

2. Ignoring Payment Processor Fees & Chargebacks Gross revenue ≠ cash received. Stripe/Paddle fees average 2.9% + $0.30. Chargebacks and refunds create negative cash flow that spreadsheets often miss. Always track net settlement amounts, not invoice totals.

3. Hardcoding Tax Rates Sales tax, VAT, and digital service taxes vary by jurisdiction, product type, and threshold. Hardcoded rates cause compliance failures. Use a tax engine (TaxJar, Avalara, or Stripe Tax) or maintain a versioned rate table with effective dates.

4. Skipping Idempotency & Deduplication Payment webhooks retry. Bank APIs paginate inconsistently. Without external ID tracking and idempotent inserts, transactions duplicate, inflating revenue and distorting runway. Always upsert on external_id.

5. Treating Revenue as Liquidity Recognized revenue ≠ available cash. SaaS deferred revenue, marketplace escrow, and net-30 payment terms create liquidity gaps. Model cash, not accrual, for runway calculations.

6. Overcomplicating the Pipeline Building a full ML categorization engine or real-time forecasting dashboard before establishing baseline ingestion creates maintenance overhead. Start deterministic, add intelligence only after 90 days of clean data.

7. No Audit Trail or Configuration Versioning Financial rules change. Without Git-tracked configs and append-only transaction logs, you cannot trace why a category shifted or why a tax calculation changed. Treat finance configs like infrastructure-as-code.

Production Bundle

Action Checklist

• Open a dedicated business checking account; zero personal transactions
• Configure webhooks for all payment processors and business cards
• Implement idempotent ingestion with external_id upserts
• Deploy deterministic categorization rules; quarantine low-confidence transactions
• Calculate rolling 3-month burn and runway; set 6-month alert threshold
• Schedule daily reconciliation via CI/CD; log all runs
• Export quarterly tax-ready reports with categorized expenses and VAT/sales tax breakdown
• Version control all financial configurations; require PR review for rule changes

Decision Matrix

Approach	Setup Time	Monthly Cost	Scalability	Control	Compliance Readiness
Manual Spreadsheet	2 hrs	$0	Low	High	Low
SaaS Accounting (QuickBooks/Xero)	6 hrs	$30–$80	Medium	Low	High
Engineered Pipeline	15–20 hrs	$10–$40 (infra)	High	High	High
Hybrid (SaaS + API Sync)	10 hrs	$40–$90	Medium-High	Medium	High

Choose engineered pipeline if you ship code daily, process >50 transactions/month, and require custom categorization or runway modeling. Choose hybrid if compliance deadlines are rigid and you lack engineering bandwidth for maintenance.

Configuration Template

# finance-config.yml
ledger:
  source: postgres
  schema: public
  table: transactions
  idempotency_key: external_id

ingestion:
  providers:
    - name: stripe
      webhook_secret_env: STRIPE_WEBHOOK_SECRET
      endpoint: /webhooks/stripe
    - name: plaid
      env_keys: PLAID_CLIENT_ID, PLAID_SECRET
      sync_interval: daily
  
categorization:
  rules_file: config/categories.yml
  confidence_threshold: 0.85
  review_queue_table: uncategorized_queue
  
modeling:
  runway:
    calculation_window_months: 3
    alert_threshold_months: 6
    notification_channels:
      - email
      - slack
  
compliance:
  tax_engine: stripe_tax
  export_format: csv
  retention_days: 2555  # 7 years

Quick Start Guide

1. Provision Infrastructure: Create a PostgreSQL instance (Supabase, Neon, or local Docker). Initialize transactions table with external_id as unique constraint.
2. Deploy Ingestion: Clone the pipeline repo, set environment variables for payment providers, and run python main.py --seed to pull historical transactions.
3. Configure Rules: Copy finance-config.yml and config/categories.yml. Adjust patterns to match your vendor stack. Run python categorize.py --dry-run to validate.
4. Schedule Reconciliation: Enable the GitHub Actions workflow or cron job. Verify daily logs and check output/runway.json.
5. Set Alerts: Configure Slack/email notifications for runway < 6 months, uncategorized queue > 50 items, or negative net burn. Test with a simulated low-cash scenario.

Finance for solo founders is not a bookkeeping exercise. It is a data system that determines whether your code ships or your runway expires. Engineer it with the same rigor as your product architecture, and solvency becomes a measurable, maintainable property of your startup.