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/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,13 @@
 from context_use.providers import (  # noqa: F401 — triggers provider registration
     airbnb,
+    amex,
+    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"]

context_use/providers/amex/transactions/pipe.py

@@ -0,0 +1,61 @@
+from __future__ import annotations
+
+import csv
+import io
+import json
+from collections.abc import Iterator
+
+from context_use.providers.amex.transactions.schemas import Model
+from context_use.providers.bank.pipe import _BankTransactionPipe
+from context_use.providers.bank.record import BankTransactionRecord
+from context_use.providers.bank.transaction_types import FALLBACK_INTERACTION_TYPE
+from context_use.providers.registry import declare_interaction
+from context_use.providers.types import InteractionConfig
+from context_use.storage.base import StorageBackend
+
+PROVIDER = "amex"
+
+
+class AmexTransactionsPipe(_BankTransactionPipe):
+    provider = PROVIDER
+    interaction_type = FALLBACK_INTERACTION_TYPE
+    archive_path_pattern = "*/amex/*.csv"
+    display_name = "Amex"
+    currency = "GBP"
+
+    def extract_file(
+        self,
+        source_uri: str,
+        storage: StorageBackend,
+    ) -> Iterator[BankTransactionRecord]:
+        stream = storage.open_stream(source_uri)
+        try:
+            reader = csv.DictReader(io.TextIOWrapper(stream, encoding="utf-8"))
+            for raw_row in reader:
+                row = Model.model_validate(raw_row)
+                if row.CR == "CR":
+                    amount = f"+{row.Amount}"
+                    transaction_type = "Payment"
+                else:
+                    amount = f"-{row.Amount}"
+                    transaction_type = "Card Payment"
+                foreign_amount = row.Foreign_Spend_Amount or None
+                foreign_currency = row.Foreign_Spend_Currency or None
+                yield BankTransactionRecord(
+                    date=row.Process_Date,
+                    authorized_date=row.Transaction_Date,
+                    amount=amount,
+                    currency=self.currency,
+                    description=row.Description,
+                    merchant_name=row.Description,
+                    account_owner=row.Cardmember,
+                    transaction_type=transaction_type,
+                    foreign_amount=foreign_amount,
+                    foreign_currency=foreign_currency,
+                    source=json.dumps(raw_row),
+                )
+        finally:
+            stream.close()
+
+
+declare_interaction(InteractionConfig(pipe=AmexTransactionsPipe))

context_use/providers/amex/transactions/schema.json

@@ -0,0 +1,40 @@
+{
+  "$schema": "http://json-schema.org/draft-07/schema#",
+  "type": "object",
+  "properties": {
+    "Cardmember": {
+      "type": "string"
+    },
+    "Transaction Date": {
+      "type": "string"
+    },
+    "Process Date": {
+      "type": "string"
+    },
+    "Description": {
+      "type": "string"
+    },
+    "Foreign Spend Amount": {
+      "type": "string"
+    },
+    "Foreign Spend Currency": {
+      "type": "string"
+    },
+    "Amount": {
+      "type": "string"
+    },
+    "CR": {
+      "type": "string"
+    }
+  },
+  "required": [
+    "Amount",
+    "CR",
+    "Cardmember",
+    "Description",
+    "Foreign Spend Amount",
+    "Foreign Spend Currency",
+    "Process Date",
+    "Transaction Date"
+  ]
+}

context_use/providers/amex/transactions/schemas.py

@@ -0,0 +1,18 @@
+# generated by datamodel-codegen:
+#   filename:  schema.json
+#   timestamp: 2026-03-21T18:32:40+00:00
+
+from __future__ import annotations
+
+from pydantic import BaseModel, Field
+
+
+class Model(BaseModel):
+    Cardmember: str
+    Transaction_Date: str = Field(..., alias="Transaction Date")
+    Process_Date: str = Field(..., alias="Process Date")
+    Description: str
+    Foreign_Spend_Amount: str = Field(..., alias="Foreign Spend Amount")
+    Foreign_Spend_Currency: str = Field(..., alias="Foreign Spend Currency")
+    Amount: str
+    CR: str

context_use/providers/bank/pipe.py

@@ -0,0 +1,59 @@
+from __future__ import annotations
+
+from datetime import UTC, datetime
+
+from context_use.activitystreams.actors import Person
+from context_use.activitystreams.objects import Note
+from context_use.etl.core.pipe import Pipe
+from context_use.etl.core.types import ThreadRow
+from context_use.etl.payload.models import (
+    CURRENT_THREAD_PAYLOAD_VERSION,
+    FibreTransaction,
+)
+from context_use.models.etl_task import EtlTask
+from context_use.providers.bank.record import BankTransactionRecord
+from context_use.providers.bank.transaction_types import normalize_transaction_type
+
+
+def _parse_date(value: str) -> datetime:
+    for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d"):
+        try:
+            return datetime.strptime(value, fmt).replace(tzinfo=UTC)
+        except ValueError:
+            continue
+    raise ValueError(f"Cannot parse date: {value!r}")
+
+
+class _BankTransactionPipe(Pipe[BankTransactionRecord]):
+    archive_version = 1
+    record_schema = BankTransactionRecord
+    display_name: str
+
+    def transform(
+        self,
+        record: BankTransactionRecord,
+        task: EtlTask,
+    ) -> ThreadRow:
+        published = _parse_date(record.date)
+        note = Note(  # type: ignore[reportCallIssue]
+            name=record.amount,
+            content=record.merchant_name or record.description,
+            published=published,
+        )
+        actor = Person(name=record.account_owner) if record.account_owner else None  # type: ignore[reportCallIssue]
+        payload = FibreTransaction(  # type: ignore[reportCallIssue]
+            object=note,
+            published=published,
+            actor=actor,
+        )
+
+        return ThreadRow(
+            unique_key=payload.unique_key(),
+            provider=self.provider,
+            interaction_type=normalize_transaction_type(record.transaction_type),
+            preview=payload.get_preview(self.display_name) or "",
+            payload=payload.to_dict(),
+            version=CURRENT_THREAD_PAYLOAD_VERSION,
+            asat=published,
+            source=record.source,
+        )

context_use/providers/bank/record.py

@@ -0,0 +1,19 @@
+from __future__ import annotations
+
+from pydantic import BaseModel
+
+
+class BankTransactionRecord(BaseModel):
+    date: str
+    authorized_date: str | None = None
+    amount: str
+    currency: str
+    description: str
+    merchant_name: str | None = None
+    transaction_type: str | None = None
+    payment_channel: str | None = None
+    pending: bool = False
+    account_owner: str | None = None
+    foreign_amount: str | None = None
+    foreign_currency: str | None = None
+    source: str | None = None

context_use/providers/bank/transaction_types.py

@@ -0,0 +1,22 @@
+from __future__ import annotations
+
+_TRANSACTION_TYPE_MAP: dict[str, str] = {
+    "Card Payment": "purchase",
+    "Card Purchase": "purchase",
+    "Direct Debit": "direct_debit",
+    "Standing Order": "standing_order",
+    "Transfer": "transfer",
+    "Bank Transfer": "transfer",
+    "Cash Withdrawal": "cash",
+    "Cheque": "cheque",
+    "Interest": "interest",
+    "Payment": "payment",
+}
+
+FALLBACK_INTERACTION_TYPE = "transaction"
+
+
+def normalize_transaction_type(raw: str | None) -> str:
+    if raw is None:
+        return FALLBACK_INTERACTION_TYPE
+    return _TRANSACTION_TYPE_MAP.get(raw, FALLBACK_INTERACTION_TYPE)

context_use/providers/barclays/__init__.py

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

context_use/providers/barclays/transactions/__init__.py

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

context_use/providers/barclays/transactions/pipe.py

@@ -0,0 +1,74 @@
+from __future__ import annotations
+
+import csv
+import io
+import json
+import re
+from collections.abc import Iterator
+
+from context_use.providers.bank.pipe import _BankTransactionPipe
+from context_use.providers.bank.record import BankTransactionRecord
+from context_use.providers.bank.transaction_types import FALLBACK_INTERACTION_TYPE
+from context_use.providers.barclays.transactions.schemas import Model
+from context_use.providers.registry import declare_interaction
+from context_use.providers.types import InteractionConfig
+from context_use.storage.base import StorageBackend
+
+PROVIDER = "barclays"
+
+_TYPE_PREFIXES = [
+    "Direct Debit",
+    "Card Purchase",
+    "Standing Order",
+    "Bank Transfer",
+    "Cash Withdrawal",
+    "Cheque",
+    "Interest",
+]
+
+_TYPE_PATTERN = re.compile(
+    r"^(" + "|".join(re.escape(p) for p in _TYPE_PREFIXES) + r")\b"
+)
+
+
+def _extract_transaction_type(description: str) -> str | None:
+    m = _TYPE_PATTERN.match(description)
+    return m.group(1) if m else None
+
+
+class BarclaysTransactionsPipe(_BankTransactionPipe):
+    provider = PROVIDER
+    interaction_type = FALLBACK_INTERACTION_TYPE
+    archive_path_pattern = "*/barclays/*.csv"
+    display_name = "Barclays"
+    currency = "GBP"
+
+    def extract_file(
+        self,
+        source_uri: str,
+        storage: StorageBackend,
+    ) -> Iterator[BankTransactionRecord]:
+        stream = storage.open_stream(source_uri)
+        try:
+            reader = csv.DictReader(io.TextIOWrapper(stream, encoding="utf-8"))
+            for raw_row in reader:
+                row = Model.model_validate(raw_row)
+                if row.Money_Out:
+                    amount = f"-{row.Money_Out}"
+                elif row.Money_In:
+                    amount = f"+{row.Money_In}"
+                else:
+                    continue
+                yield BankTransactionRecord(
+                    date=row.Date,
+                    amount=amount,
+                    currency=self.currency,
+                    description=row.Description,
+                    transaction_type=_extract_transaction_type(row.Description),
+                    source=json.dumps(raw_row),
+                )
+        finally:
+            stream.close()
+
+
+declare_interaction(InteractionConfig(pipe=BarclaysTransactionsPipe))