Fintech Data Ingestion: Technical Blueprint

This blueprint provides the structural and operational requirements for building a production-grade data ingestion pipeline for the WealthView application. It focuses on the transition from third-party formats (OFX/QFX, CSV) to a normalized, query-optimized internal model.

1. Domain Model (Pydantic / Internal Schemas)

To ensure robust data parsing, use strictly typed schemas for normalization.

1.1 The Account Schema

Represents a standardized financial account across institutions (Checking, Savings, Investment, Credit).

from enum import Enum
from pydantic import BaseModel, Field
from typing import Optional
from datetime import datetime

class AccountType(str, Enum):
    brokerage = "brokerage"
    ira = "ira"
    401k = "401k"
    roth = "roth"
    bank = "bank"

class NormalizedAccount(BaseModel):
    account_id: str = Field(..., description="Unique internal UUID")
    institution_name: str
    official_name: Optional[str]
    type: AccountType
    balance_current: float
    iso_currency_code: str = "USD"
    last_synced: datetime = Field(default_factory=datetime.utcnow)

1.2 The Transaction Schema

Normalized schema for bank and brokerage transactions parsed from OFX or CSV.

class TransactionType(str, Enum):
    buy = "buy"
    sell = "sell"
    dividend = "dividend"
    deposit = "deposit"
    withdrawal = "withdrawal"

class NormalizedTransaction(BaseModel):
    transaction_id: str
    account_id: str
    date: datetime
    amount: float
    symbol: Optional[str]
    quantity: Optional[float]
    description: str
    type: TransactionType

2. Ingestion Pipeline Architecture

2.1 The "Idempotent Sync" Pattern (Deduplication)

WealthView explicitly avoids complex aggregator APIs (like Plaid) in favor of direct OFX/CSV imports. To prevent duplicate transactions during repeated uploads, implement a multi-key SHA-256 hash check:

Generate Import Hash: hash(date + amount + normalized_description). For OFX, the institution-provided FITID can be utilized.
Upsert Logic: Before inserting a batch, query the import_hash column for existing hashes in the account.
Silent Drops: Any transaction whose hash already exists is silently dropped, ensuring safe idempotency.

2.2 Normalization Logic (The "Mapper" Layer)

Different brokerages format their CSV exports differently (e.g., Fidelity vs Vanguard). Your mapper must:

Brokerage-Specific Parsers: Implement the strategy pattern to utilize specific parsers based on the source file format natively.
OFX4J: Use standard libraries to extract standard fields from OFX/QFX XML structures reliably.

3. Operational Resilience

3.1 Error Handling Protocols

Error Code	Strategy	User Action
`INVALID_FILE_FORMAT`	Abort Import	Prompt user to upload correct CSV/OFX
`CONFLICTING_MANUAL_OVERRIDE`	Log Warning	Notify user that a manual holding override exists

3.2 Security and Compliance

File Parsing Sandboxing: Ensure XML parsers explicitly disable External Entity Resolution (XXE) to prevent XML injection attacks via malicious OFX files.
Tenant Isolation: Ensure all import jobs strictly validate the JWT's tenant_id before modifying any account records.

4. RAG Implementation Hook

For an agent building WealthView, the prompt should be:

"Using the FintechDataIngestionBlueprint, implement a Java Spring Boot service that utilizes OFX4J to parse an uploaded OFX file, normalizes the transactions, generates SHA-256 deduplication hashes, and persists the new records to the database under the user's tenant_id."