feat: bank transactions with a Plaid like schema

README.md

@@ -142,5 +142,8 @@ context-use-data/
 | Google | Available | Searches, YouTube, Shopping, Lens, Discover | [Export your data](https://support.google.com/accounts/answer/3024190) |
 | Netflix | Available | Viewing Activity, Search History, Ratings, My List, Messages, Preferences | [Export your data](https://help.netflix.com/en/node/100624) |
 | Airbnb | Available | Wishlists, Search History, Reviews, Reservations, Messages | [Export your data](https://www.airbnb.com/help/article/2273) |
+| Amex | Available | Transactions | Manual CSV download from Amex |
+| Barclays | Available | Transactions | Manual CSV download from Barclays |
+| Revolut | Available | Transactions | Manual CSV download from Revolut |
 
 Want another provider? Contribute it by pointing your coding agent to the [Adding a Data Provider](docs/add-provider/AGENTS.md) guide.

context_use/cli/bank_setup.py

@@ -0,0 +1,192 @@
+"""Interactive column-mapping setup for the generic bank provider.
+
+Reads CSV headers from a zip archive and walks the user through mapping
+columns to the fields required by :class:`BankMapping`.
+"""
+
+from __future__ import annotations
+
+import csv
+import io
+import zipfile
+
+from context_use.cli import output as out
+from context_use.providers.bank.mapping import AmountColumns, BankMapping
+
+
+def _read_csv_headers(zip_path: str) -> tuple[str, list[str]] | None:
+    """Extract headers from the first CSV found inside *zip_path*.
+
+    Returns ``(csv_filename, headers)`` or ``None`` if no CSV is found.
+    """
+    with zipfile.ZipFile(zip_path) as zf:
+        for name in sorted(zf.namelist()):
+            if name.startswith("__MACOSX"):
+                continue
+            if name.lower().endswith(".csv"):
+                with zf.open(name) as f:
+                    reader = csv.reader(io.TextIOWrapper(f, encoding="utf-8"))
+                    try:
+                        headers = next(reader)
+                    except StopIteration:
+                        continue
+                    return name, [h.strip() for h in headers]
+    return None
+
+
+def _pick_column(headers: list[str], prompt_text: str) -> str | None:
+    """Let the user pick one column by number."""
+    choice = input(prompt_text).strip()
+    try:
+        idx = int(choice) - 1
+        if 0 <= idx < len(headers):
+            return headers[idx]
+    except ValueError:
+        pass
+    out.error("Invalid choice.")
+    return None
+
+
+def _print_columns(headers: list[str]) -> None:
+    for i, h in enumerate(headers, 1):
+        print(f"  {out.bold(str(i))}. {h}")
+    print()
+
+
+def _ask_amount_columns(headers: list[str]) -> AmountColumns | None:
+    """Ask whether amount is a single column or split into in/out."""
+    out.header("Amount columns")
+    print()
+    print(f"  {out.bold('1')}. Single column (e.g. +100 / -50)")
+    print(f"  {out.bold('2')}. Separate columns for money in / money out")
+    print()
+
+    mode = input("  Amount format? [1-2]: ").strip()
+
+    if mode == "1":
+        print()
+        out.info("Which column contains the transaction amount?")
+        print()
+        _print_columns(headers)
+        col = _pick_column(headers, f"  Amount column [1-{len(headers)}]: ")
+        if col is None:
+            return None
+        return AmountColumns(single=col)
+
+    if mode == "2":
+        print()
+        out.info("Which column contains money IN (credits/deposits)?")
+        print()
+        _print_columns(headers)
+        col_in = _pick_column(headers, f"  Money-in column [1-{len(headers)}]: ")
+        if col_in is None:
+            return None
+
+        print()
+        out.info("Which column contains money OUT (debits/payments)?")
+        print()
+        _print_columns(headers)
+        col_out = _pick_column(headers, f"  Money-out column [1-{len(headers)}]: ")
+        if col_out is None:
+            return None
+
+        return AmountColumns(money_in=col_in, money_out=col_out)
+
+    out.error("Invalid choice.")
+    return None
+
+
+def _ask_yes_no(prompt_text: str, *, default: bool = False) -> bool:
+    hint = "Y/n" if default else "y/N"
+    answer = input(f"  {prompt_text} [{hint}]: ").strip().lower()
+    if not answer:
+        return default
+    return answer in ("y", "yes")
+
+
+def run_bank_setup(zip_path: str) -> BankMapping | None:
+    """Run the full interactive bank CSV setup.
+
+    Returns a :class:`BankMapping` or ``None`` if the user aborts.
+    """
+    result = _read_csv_headers(zip_path)
+    if result is None:
+        out.error("No CSV files found in the archive.")
+        return None
+
+    csv_filename, headers = result
+    if not headers:
+        out.error(f"CSV file {csv_filename} has no columns.")
+        return None
+
+    out.header("Bank CSV setup")
+    out.kv("File", csv_filename)
+    print()
+
+    out.info("CSV columns found:")
+    print()
+    _print_columns(headers)
+
+    bank_name = input("  Bank name (e.g. Chase, HSBC): ").strip()
+    if not bank_name:
+        out.error("Bank name is required.")
+        return None
+
+    print()
+    out.info("Which column contains the transaction date?")
+    print()
+    _print_columns(headers)
+    date_col = _pick_column(headers, f"  Date column [1-{len(headers)}]: ")
+    if date_col is None:
+        return None
+
+    print()
+    amount = _ask_amount_columns(headers)
+    if amount is None:
+        return None
+
+    print()
+    out.info("Which column contains the transaction description?")
+    print()
+    _print_columns(headers)
+    desc_col = _pick_column(headers, f"  Description column [1-{len(headers)}]: ")
+    if desc_col is None:
+        return None
+
+    print()
+    is_credit_card = _ask_yes_no(
+        "Is this a credit card? (charges shown as positive amounts)"
+    )
+
+    print()
+    currency = input("  Currency code (e.g. GBP, USD, EUR): ").strip().upper()
+    if not currency:
+        out.error("Currency is required.")
+        return None
+
+    print()
+    out.header("Mapping summary")
+    out.kv("Bank", bank_name)
+    out.kv("Date column", date_col)
+    if amount.single:
+        out.kv("Amount column", amount.single)
+    else:
+        out.kv("Money-in column", amount.money_in)
+        out.kv("Money-out column", amount.money_out)
+    out.kv("Description column", desc_col)
+    out.kv("Credit card", "yes" if is_credit_card else "no")
+    out.kv("Currency", currency)
+    print()
+
+    if not _ask_yes_no("Proceed with this mapping?", default=True):
+        out.warn("Aborted.")
+        return None
+
+    return BankMapping(
+        bank_name=bank_name,
+        date_column=date_col,
+        amount=amount,
+        description_column=desc_col,
+        currency=currency,
+        is_credit_card=is_credit_card,
+    )

context_use/cli/commands/ingest.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import argparse
+from collections.abc import Callable
 from typing import TYPE_CHECKING
 
 from context_use.cli import output as out
@@ -14,6 +15,7 @@
 
 if TYPE_CHECKING:
     from context_use import ContextUse
+    from context_use.etl.core.pipe import Pipe
 
 
 class IngestCommand(ContextCommand):
@@ -38,13 +40,29 @@ async def run(
             return
         provider_str, zip_path = picked
 
+        pipe_factory: Callable[[type[Pipe]], Pipe] | None = None
+        if provider_str == "bank":
+            from context_use.cli.bank_setup import run_bank_setup
+            from context_use.providers.bank.generic_pipe import GenericBankPipe
+
+            mapping = run_bank_setup(zip_path)
+            if mapping is None:
+                return
+
+            def _bank_factory(pipe_cls: type[Pipe]) -> Pipe:
+                return GenericBankPipe(mapping=mapping)
+
+            pipe_factory = _bank_factory
+
         print()
         out.header(f"Ingesting {provider_str} archive")
         out.kv("File", zip_path)
         out.kv("Provider", provider_str)
         print()
 
-        result = await ctx.process_archive(provider_str, zip_path)
+        result = await ctx.process_archive(
+            provider_str, zip_path, pipe_factory=pipe_factory
+        )
 
         out.success("Archive processed")
         out.kv("Archive ID", result.archive_id)

context_use/core.py

@@ -3,6 +3,7 @@
 import logging
 import zipfile
 from collections import defaultdict
+from collections.abc import Callable
 from pathlib import PurePosixPath
 from typing import TYPE_CHECKING
 
@@ -35,6 +36,7 @@
     from datetime import date, datetime
     from typing import Any
 
+    from context_use.etl.core.pipe import Pipe
     from context_use.llm.litellm.clients import LiteLLMBase
     from context_use.storage.base import StorageBackend
     from context_use.store.base import Store
@@ -92,6 +94,7 @@ async def process_archive(
         self,
         provider: str,
         path: str,
+        pipe_factory: Callable[[type[Pipe]], Pipe] | None = None,
     ) -> PipelineResult:
         """Unzip, discover, and run ETL for the given archive.
 
@@ -151,7 +154,7 @@ async def process_archive(
         for task_model in task_models:
             try:
                 pipe_cls = provider_cfg.get_pipe(task_model.interaction_type)
-                pipe = pipe_cls()
+                pipe = pipe_factory(pipe_cls) if pipe_factory else pipe_cls()
                 count = await self._run_pipe(pipe, task_model)
 
                 task_model.status = EtlTaskStatus.COMPLETED.value

context_use/etl/payload/models.py

@@ -435,6 +435,33 @@ def _get_preview(self, provider: str | None) -> str | None:
         return parts
 
 
+class FibreTransaction(Create, _BaseFibreMixin):
+    fibreKind: Literal["Transaction"] = Field("Transaction", alias="fibre_kind")
+    object: Note  # type: ignore[reportIncompatibleVariableOverride, reportGeneralTypeIssues]
+    actor: Person | None = None  # type: ignore[reportIncompatibleVariableOverride]
+
+    def _get_preview(self, provider: str | None) -> str | None:
+        amount = str(self.object.name) if self.object.name else ""
+        desc = str(self.object.content) if self.object.content else ""
+        if amount.startswith("-"):
+            verb, prep = "Spent", "at"
+            display_amount = amount[1:]
+        elif amount.startswith("+"):
+            verb, prep = "Received", "from"
+            display_amount = amount[1:]
+        else:
+            verb, prep = "Transacted", "at"
+            display_amount = amount
+        parts = f"{verb} {display_amount}"
+        if desc:
+            parts += f" {prep} {desc}"
+        if self.actor and self.actor.name:
+            parts += f" (by {self.actor.name})"
+        if provider:
+            parts += f" via {provider}"
+        return parts
+
+
 # --- Discriminated unions ---
 
 FibreReactionByType = Annotated[
@@ -455,7 +482,8 @@ def _get_preview(self, provider: str | None) -> str | None:
     | FibreCollection
     | FibreSendMessage
     | FibreReceiveMessage
-    | FibreComment,
+    | FibreComment
+    | FibreTransaction,
     Field(discriminator="fibreKind"),
 ]
 
@@ -484,5 +512,6 @@ def _get_preview(self, provider: str | None) -> str | None:
 FibreComment.model_rebuild()
 FibreSearch.model_rebuild()
 FibreAddObjectToCollection.model_rebuild()
+FibreTransaction.model_rebuild()
 FibreFollowedBy.model_rebuild()
 FibreFollowing.model_rebuild()

context_use/providers/__init__.py

@@ -1,10 +1,14 @@
 from context_use.providers import (  # noqa: F401 — triggers provider registration
     airbnb,
+    amex,
+    bank,
+    barclays,
     chatgpt,
     claude,
     google,
     instagram,
     netflix,
+    revolut,
 )
 from context_use.providers.registry import (
     get_memory_config,

context_use/providers/amex/__init__.py

@@ -0,0 +1,8 @@
+from context_use.providers.amex import transactions
+from context_use.providers.registry import register_provider
+
+PROVIDER = "amex"
+
+register_provider(PROVIDER, modules=[transactions])
+
+__all__ = ["PROVIDER"]

context_use/providers/amex/transactions/__init__.py

@@ -0,0 +1,3 @@
+from context_use.providers.amex.transactions.pipe import AmexTransactionsPipe
+
+__all__ = ["AmexTransactionsPipe"]