API Importers & ML Prediction

beancount-blue includes an extensible, secure banking API ingestion engine. It connects directly to bank and Open Banking APIs, stores state locally in compressed, version-controlled caches, and generates native Beancount directives.

It uniquely features a built-in Machine Learning Predictor using scikit-learn that learns from your existing Beancount books to automatically categorize payees and counter-accounts for incoming transactions.

---

Supported Bank Providers

1. Monzo (monzo)

• Authentication: OAuth 2.0 via Monzo Developer Portal (developers.monzo.com).
• Capabilities:
• Ingests transactions and balances from standard current accounts, joint accounts, and Monzo Flex.
• Automatically detects Monzo Pots and generates internal transfer postings to pot accounts (e.g. Assets:Monzo:Main -> Assets:Monzo:HolidayPot).
• Correctly resolves pending card authorizations and Mastercard clearing reversals using mastercard_lifecycle_id.
• Supports pocket money transfers for Young Monzo accounts.

2. Starling (starling)

• Authentication: Personal Access Token via Starling Developer Portal.
• Capabilities:
• Ingests primary accounts, spending spaces, and savings spaces.
• Automatically maps internal transfers between spaces.
• Supports multi-user joint accounts via user_map (mapping Starling user UUIDs to friendly names in transaction metadata).

3. TrueLayer Open Banking (truelayer)

• Authentication: OAuth 2.0 Open Banking client credentials via TrueLayer Console (console.truelayer.com).
• Capabilities:
• Connects to dozens of UK and European banks, credit cards, and institutions (Amex, Chase, Barclaycard, etc.).
• Synchronizes both bank accounts and credit cards with automated debit/credit sign standardization.

---

Configuration

Importers are configured through YAML files typically stored in your api_configs/ directory.

Example Configuration: Monzo (api_configs/monzo.yaml)

importer_name: monzo
client_id: "oauth2client_00000000000000"
client_secret: "mnzpub.secret_key_here"

# Cache & State Settings
cache_data: "data/monzo.json.gz"
units: 2

# Native Machine Learning Prediction
auto_predict: true
predict_ledger_path: "main.beancount"
predict_model_path: "data/monzo_model.joblib"
predict_min_confidence: 0.6
predict_retrain_days: 7.0

# Account Mapping (Anchor Account)
account_map:
  acc_123456789: "Assets:Current:Monzo"

Example Configuration: Starling (api_configs/starling.yaml)

importer_name: starling
personal_access_token: "your_starling_token_here"
cache_data: "data/starling.json.gz"

auto_predict: true
predict_ledger_path: "main.beancount"
predict_model_path: "data/starling_model.joblib"

account_map:
  "00000000-0000-0000-0000-000000000001": "Assets:Current:Starling"

user_map:
  "88888888-8888-8888-8888-888888888888": "Alice"

Example Configuration: TrueLayer (api_configs/truelayer.yaml)

importer_name: truelayer
client_id: "your_truelayer_client_id"
client_secret: "your_truelayer_client_secret"
cache_data: "data/truelayer.json.gz"

auto_predict: true
predict_ledger_path: "main.beancount"
predict_model_path: "data/truelayer_model.joblib"

account_map:
  "acc_amex_gold": "Liabilities:CreditCard:Amex"

---

Machine Learning Auto-Categorization

When auto_predict: true is enabled, the importer automatically learns from your historical ledger:

1. Training Heuristic: On extraction, it loads transactions from predict_ledger_path involving the accounts listed in account_map.
2. Feature Extraction & Classification: It extracts text features (narration, description, existing payee) using a TfidfVectorizer and fits a fast, lightweight online logistic regression model (SGDClassifier).
3. Model Persistence: The pipeline is serialized to predict_model_path using joblib.
4. Inference: When new transactions arrive from the bank API, the model predicts the most likely counter_account (e.g. Expenses:Groceries) and clean payee. Predictions with probability above predict_min_confidence are automatically injected into the imported entries.
5. Smart Retraining: The model is automatically retrained when your ledger file modification time (mtime) is newer than the saved model file, or if the model file is older than predict_retrain_days.

---

Command Line Interface (bean-blue-importer)

beancount-blue installs a dedicated CLI tool: bean-blue-importer.

Common Commands

# 1. Fetch latest data from bank API and save compressed cache
bean-blue-importer sync --settings api_configs/monzo.yaml

# 2. Output Beancount directives with ML predictions applied
bean-blue-importer beancount --settings api_configs/monzo.yaml > imported_entries.beancount

# 3. Force-retrain the ML categorization model
bean-blue-importer train --settings api_configs/monzo.yaml --ledger main.beancount

---

Python API Reference

beancount_blue.importer.delta_importer

beancount_blue.importer.delta_importer.APIImporter

Bases: BaseSettings

Base class for all API Importers.

This class manages configuration, API state caching, filtering, and machine learning predictions. Subclasses (like MonzoImporter or StarlingImporter) implement the refresh and extract logic.

Source code in beancount_blue/importer/delta_importer.py

class APIImporter[APIData: BaseModel](BaseSettings, metaclass=ABCMeta):
    """
    Base class for all API Importers.

    This class manages configuration, API state caching, filtering, and machine learning predictions.
    Subclasses (like MonzoImporter or StarlingImporter) implement the `refresh` and `extract` logic.
    """

    # Main type
    importer_name: str = Field(description="The unique name identifying this importer.")

    # Name of the importer
    name: str | None = Field(None, description="Name of this particular import configuration.")

    # Configure parameters
    min_date: date | None = Field(None, description="Only extract transactions on or after this date.")
    account_map: dict[str, str | AccountConfig] | None = Field(
        None, description="Mapping of API account IDs to Beancount account names or config objects."
    )
    cache_only: bool = Field(
        False, description="If True, skips the API refresh and only loads data from the local cache."
    )
    cache_data: str | None = Field(
        None, description="File path to the compressed tar.gz file where the API state is cached."
    )
    interactive_auth: bool = Field(
        False, exclude=True, description="Allow interactive CLI auth flows (like input() or local webservers)."
    )
    redirect_uri: str | None = Field(None, exclude=True, description="Dynamic redirect URI for OAuth callbacks")

    # Predictor options
    auto_predict: bool = Field(False, description="Enable the ML predictor to guess payees and counter_accounts.")
    predict_ledger_path: str | None = Field(
        None, description="Path to the main Beancount ledger file used as training data."
    )
    predict_model_path: str = Field(
        "predictor_model.joblib", description="File path where the trained ML model is stored/cached."
    )
    predict_anchor_accounts: list[str] | None = Field(
        None, description="List of anchor accounts to train on. Defaults to the values in `account_map`."
    )
    predict_skip_accounts: list[str] = Field(
        default_factory=list, description="List of counter-accounts to explicitly ignore when training the ML model."
    )
    predict_remap_accounts: dict[str, str] = Field(
        default_factory=dict,
        description="Mapping of accounts to rename during ML prediction training.",
    )
    predict_min_confidence: float = Field(
        0.5, description="The minimum confidence threshold (0.0 to 1.0) required to apply a prediction."
    )
    predict_retrain_days: float | None = Field(
        7.0,
        description="Force a retraining of the ML model if it is older than this many days.",
    )

    @property
    def anchor_accounts(self) -> list[str]:
        if self.account_map:
            return [v.name if isinstance(v, AccountConfig) else v for v in self.account_map.values()]
        return []

    def get_account_configs_by_name(self) -> dict[str, AccountConfig]:
        if not self.account_map:
            return {}
        return {
            (v.name if isinstance(v, AccountConfig) else v): (
                v
                if isinstance(v, AccountConfig)
                else AccountConfig(name=v, starting_balance=None, starting_date=None, currency=None)
            )
            for v in self.account_map.values()
        }

    def handle_callback(self, code: str, state: str) -> None:
        """Handle an OAuth/web callback for this importer."""
        raise NotImplementedError("This importer does not support web callbacks.")

    @classmethod
    def get_types(cls) -> type[APIData]:
        """Magic introspection to avoid 'config_type = ...' boilerplate"""
        for base in get_original_bases(cls):
            if not hasattr(base, "__pydantic_generic_metadata__"):
                continue
            if base.__pydantic_generic_metadata__.get("origin") is APIImporter:
                return base.__pydantic_generic_metadata__.get("args")[0]  # type: ignore
        raise ImporterConfigurationError(f"{cls.__name__} must inherit from APIImporter[APIData]")

    @abstractmethod
    def refresh(self, state: APIData) -> None:
        """Refresh the data from the API.

        state: mutable BaseModel for the state of the API connection and data.
        """

    @abstractmethod
    def extract(self, state: APIData) -> list[ImportedTransaction]:
        """Extract imported transactions from API data.

        state: BaseModel to extract the transactions from.
        """

    @abstractmethod
    def extract_available_balances(self, state: APIData) -> dict[str, tuple[Decimal, str]]:
        """Return a mapping of raw API account IDs to (available_balance, currency).
        Note: This represents the bank's available balance (often including pending transactions),
        not the cleared ledger balance.
        """

    def format_available_balances(self, state: APIData | ImporterState[APIData], acct: str | None = None) -> str | None:
        """Translates raw available balances into a human-readable string using account_map."""
        if isinstance(state, ImporterState):
            from typing import cast

            raw_balances = self.extract_available_balances(cast(APIData, state.data))
        else:
            raw_balances = self.extract_available_balances(state)
        if not raw_balances:
            return None

        lines: list[str] = []
        m = {k: (v.name if isinstance(v, AccountConfig) else v) for k, v in (self.account_map or {}).items()}
        sorted_keys = sorted(m.keys(), key=len, reverse=True)
        for raw_id, (amount, currency) in raw_balances.items():
            account_name = raw_id
            for k in sorted_keys:
                if account_name == k or account_name.startswith(k + ":"):
                    account_name = account_name.replace(k, m[k], 1)
                    break

            if acct:
                if account_name != acct:
                    continue

                return f"{amount:,.2f} {currency}"

            lines.append(f"{account_name}: {amount:,.2f} {currency}")

        if not lines:
            return None

        return "\n".join(sorted(lines))

    def load_data(self) -> ImporterState[APIData]:
        api_data_type = self.get_types()
        state_type = ImporterState[api_data_type]

        if self.cache_data:
            with load(self.cache_data, state_type, skip_save=self.cache_only) as state:
                if not self.cache_only:
                    log.info("Refreshing data from API, Cache only is %s", self.cache_only)
                    try:
                        self.refresh(state.data)
                        state.last_sync_error = None
                    except APIImporterRedirectRequired:
                        raise
                    except Exception as e:
                        log.exception("Error during API refresh")
                        state.last_sync_error = str(e)
                    finally:
                        state.last_sync_time = datetime.now()
                        try:
                            entries = self.extract(state.data)
                            if entries:
                                dates = [e.date for e in entries if getattr(e, "date", None)]
                                if dates:
                                    state.latest_transaction_date = max(dates)
                        except Exception as e:
                            log.debug(f"Could not extract dates for dashboard metadata: {e}")
                return state
        else:
            if self.cache_only:
                log.warning("No cache data path provided, but cache_only is set to True. Ignoring cache_only.")
            state = state_type(data=api_data_type())
            try:
                self.refresh(state.data)
                state.last_sync_error = None
            except APIImporterRedirectRequired:
                raise
            except Exception as e:
                log.exception("Error during API refresh")
                state.last_sync_error = str(e)
            finally:
                state.last_sync_time = datetime.now()
            return state

    def filter(self, data: list[ImportedTransaction]) -> list[ImportedTransaction]:
        """Filter imported transactions based on config.

        data: List of imported transactions.
        config: Configuration object.
        """
        if self.account_map:
            m = {k: (v.name if isinstance(v, AccountConfig) else v) for k, v in self.account_map.items()}
            sorted_keys = sorted(m.keys(), key=len, reverse=True)
            for e in data:
                for k in sorted_keys:
                    v = m[k]
                    if e.account == k or e.account.startswith(k + ":"):
                        e.account = e.account.replace(k, v, 1)
                        break
                if e.counter_account:
                    for k in sorted_keys:
                        v = m[k]
                        if e.counter_account == k or e.counter_account.startswith(k + ":"):
                            e.counter_account = e.counter_account.replace(k, v, 1)
                            break
        return data

    @final
    def beancount_load(self, existing: Entries | None = None) -> Entries:
        state = self.load_data()
        imported_entries = self.extract(state.data)
        imported_entries = self.filter(imported_entries)

        # ML Prediction logic
        if self.auto_predict:
            import time

            from beancount.loader import load_file

            from .predictor import TransactionPredictor

            predictor = TransactionPredictor(Path(self.predict_model_path))

            # Heuristic: Check if we need to retrain
            retrain = False
            if self.predict_ledger_path:
                ledger_path = Path(self.predict_ledger_path)
                if ledger_path.exists():
                    model_path = Path(self.predict_model_path)
                    if not model_path.exists():
                        log.info("Model missing. Retraining...")
                        retrain = True
                    else:
                        model_mtime = model_path.stat().st_mtime
                        if ledger_path.stat().st_mtime > model_mtime:
                            log.info("Ledger is newer than model. Retraining...")
                            retrain = True
                        elif self.predict_retrain_days is not None:
                            age_days = (time.time() - model_mtime) / 86400.0
                            if age_days > self.predict_retrain_days:
                                log.info(
                                    f"Model age ({age_days:.1f} days) exceeds threshold "
                                    f"({self.predict_retrain_days} days). Retraining..."
                                )
                                retrain = True

            if not retrain and not predictor.load():
                log.info("Model could not be loaded or is invalid. Retraining as fallback...")
                retrain = True

            if retrain and self.predict_ledger_path:
                ledger_path = Path(self.predict_ledger_path)
                if ledger_path.exists():
                    entries, _, _ = load_file(str(ledger_path))
                    anchors = self.predict_anchor_accounts or self.anchor_accounts
                    predictor.train(
                        entries,
                        anchors,
                        self.predict_skip_accounts,
                        self.predict_remap_accounts,
                        imported_entries=imported_entries,
                    )

            predictor.apply_predictions(imported_entries, min_confidence=self.predict_min_confidence)

        ret = imported_to_beancount(
            imported_entries, existing=existing, account_configs=self.get_account_configs_by_name()
        )
        log.info(f"Found {len(imported_entries)} entries, returning {len(ret)} entries when de-duplicated.")
        return self.filter_beancount(ret)

    def filter_beancount(self, entries: Entries) -> Entries:
        """Filter the final Beancount entries."""
        configs = self.get_account_configs_by_name()
        filtered: list[Directive] = []
        for e in entries:
            # Global min_date
            if self.min_date and getattr(e, "date", date.min) < self.min_date:
                continue

            # Per-account starting_date (min_date)
            drop = False
            if isinstance(e, Transaction):
                for p in e.postings:
                    conf = configs.get(p.account)
                    if conf and conf.starting_date and e.date < conf.starting_date:
                        drop = True
                        break
            elif isinstance(e, Balance):
                conf = configs.get(e.account)
                if conf and conf.starting_date and e.date < conf.starting_date:
                    drop = True

            if not drop:
                filtered.append(e)

        return filtered

extract(state) abstractmethod

Extract imported transactions from API data.

state: BaseModel to extract the transactions from.

Source code in beancount_blue/importer/delta_importer.py

@abstractmethod
def extract(self, state: APIData) -> list[ImportedTransaction]:
    """Extract imported transactions from API data.

    state: BaseModel to extract the transactions from.
    """

extract_available_balances(state) abstractmethod

Return a mapping of raw API account IDs to (available_balance, currency). Note: This represents the bank's available balance (often including pending transactions), not the cleared ledger balance.

Source code in beancount_blue/importer/delta_importer.py

@abstractmethod
def extract_available_balances(self, state: APIData) -> dict[str, tuple[Decimal, str]]:
    """Return a mapping of raw API account IDs to (available_balance, currency).
    Note: This represents the bank's available balance (often including pending transactions),
    not the cleared ledger balance.
    """

filter(data)

Filter imported transactions based on config.

data: List of imported transactions. config: Configuration object.

Source code in beancount_blue/importer/delta_importer.py

def filter(self, data: list[ImportedTransaction]) -> list[ImportedTransaction]:
    """Filter imported transactions based on config.

    data: List of imported transactions.
    config: Configuration object.
    """
    if self.account_map:
        m = {k: (v.name if isinstance(v, AccountConfig) else v) for k, v in self.account_map.items()}
        sorted_keys = sorted(m.keys(), key=len, reverse=True)
        for e in data:
            for k in sorted_keys:
                v = m[k]
                if e.account == k or e.account.startswith(k + ":"):
                    e.account = e.account.replace(k, v, 1)
                    break
            if e.counter_account:
                for k in sorted_keys:
                    v = m[k]
                    if e.counter_account == k or e.counter_account.startswith(k + ":"):
                        e.counter_account = e.counter_account.replace(k, v, 1)
                        break
    return data

filter_beancount(entries)

Filter the final Beancount entries.

Source code in beancount_blue/importer/delta_importer.py

def filter_beancount(self, entries: Entries) -> Entries:
    """Filter the final Beancount entries."""
    configs = self.get_account_configs_by_name()
    filtered: list[Directive] = []
    for e in entries:
        # Global min_date
        if self.min_date and getattr(e, "date", date.min) < self.min_date:
            continue

        # Per-account starting_date (min_date)
        drop = False
        if isinstance(e, Transaction):
            for p in e.postings:
                conf = configs.get(p.account)
                if conf and conf.starting_date and e.date < conf.starting_date:
                    drop = True
                    break
        elif isinstance(e, Balance):
            conf = configs.get(e.account)
            if conf and conf.starting_date and e.date < conf.starting_date:
                drop = True

        if not drop:
            filtered.append(e)

    return filtered

format_available_balances(state, acct=None)

Translates raw available balances into a human-readable string using account_map.

Source code in beancount_blue/importer/delta_importer.py

def format_available_balances(self, state: APIData | ImporterState[APIData], acct: str | None = None) -> str | None:
    """Translates raw available balances into a human-readable string using account_map."""
    if isinstance(state, ImporterState):
        from typing import cast

        raw_balances = self.extract_available_balances(cast(APIData, state.data))
    else:
        raw_balances = self.extract_available_balances(state)
    if not raw_balances:
        return None

    lines: list[str] = []
    m = {k: (v.name if isinstance(v, AccountConfig) else v) for k, v in (self.account_map or {}).items()}
    sorted_keys = sorted(m.keys(), key=len, reverse=True)
    for raw_id, (amount, currency) in raw_balances.items():
        account_name = raw_id
        for k in sorted_keys:
            if account_name == k or account_name.startswith(k + ":"):
                account_name = account_name.replace(k, m[k], 1)
                break

        if acct:
            if account_name != acct:
                continue

            return f"{amount:,.2f} {currency}"

        lines.append(f"{account_name}: {amount:,.2f} {currency}")

    if not lines:
        return None

    return "\n".join(sorted(lines))

get_types() classmethod

Magic introspection to avoid 'config_type = ...' boilerplate

Source code in beancount_blue/importer/delta_importer.py

@classmethod
def get_types(cls) -> type[APIData]:
    """Magic introspection to avoid 'config_type = ...' boilerplate"""
    for base in get_original_bases(cls):
        if not hasattr(base, "__pydantic_generic_metadata__"):
            continue
        if base.__pydantic_generic_metadata__.get("origin") is APIImporter:
            return base.__pydantic_generic_metadata__.get("args")[0]  # type: ignore
    raise ImporterConfigurationError(f"{cls.__name__} must inherit from APIImporter[APIData]")

handle_callback(code, state)

Handle an OAuth/web callback for this importer.

Source code in beancount_blue/importer/delta_importer.py

def handle_callback(self, code: str, state: str) -> None:
    """Handle an OAuth/web callback for this importer."""
    raise NotImplementedError("This importer does not support web callbacks.")

refresh(state) abstractmethod

Refresh the data from the API.

state: mutable BaseModel for the state of the API connection and data.

Source code in beancount_blue/importer/delta_importer.py

@abstractmethod
def refresh(self, state: APIData) -> None:
    """Refresh the data from the API.

    state: mutable BaseModel for the state of the API connection and data.
    """

beancount_blue.importer.monzo

beancount_blue.importer.monzo.MonzoImporter

Bases: APIImporter[MonzoData]

Monzo API Setup Instructions

To automatically sync your Monzo account, you need to create an OAuth API client in the Monzo Developer portal.

1. Log in: Go to developers.monzo.com and log in with your email. You will receive a magic link in your email and a push notification in your Monzo app to approve the login.
2. Create a Client:
3. Click on "Clients" in the top navigation bar.
4. Click "New OAuth Client".
5. Give it a name (e.g., Beancount Sync).
6. Set the Confidentiality to Confidential.
7. Set the Redirect URLs to http://localhost:8000/callback (or your preferred local callback URL if you are using a custom auth flow).
8. Copy Credentials: Once created, copy the Client ID and Client Secret.
9. Configure Fava: Paste these values into the client_id and client_secret fields below.

Note: Monzo requires you to re-authenticate API access via the app every 90 days. If your sync fails with an authentication error, you may need to approve the connection in your Monzo app.

Source code in beancount_blue/importer/monzo.py

class MonzoImporter(APIImporter[MonzoData]):
    """
    ### Monzo API Setup Instructions

    To automatically sync your Monzo account, you need to create an OAuth API client in the Monzo Developer portal.

    1. **Log in:** Go to [developers.monzo.com](https://developers.monzo.com/) and log in with your email. You will receive a magic link in your email and a push notification in your Monzo app to approve the login.
    2. **Create a Client:**
       - Click on **"Clients"** in the top navigation bar.
       - Click **"New OAuth Client"**.
       - Give it a name (e.g., `Beancount Sync`).
       - Set the **Confidentiality** to `Confidential`.
       - Set the **Redirect URLs** to `http://localhost:8000/callback` (or your preferred local callback URL if you are using a custom auth flow).
    3. **Copy Credentials:** Once created, copy the **Client ID** and **Client Secret**.
    4. **Configure Fava:** Paste these values into the `client_id` and `client_secret` fields below.

    *Note: Monzo requires you to re-authenticate API access via the app every 90 days. If your sync fails with an authentication error, you may need to approve the connection in your Monzo app.*
    """

    importer_name: Literal["monzo"] = "monzo"  # pyright: ignore[reportIncompatibleVariableOverride]

    units: int = 2
    client_id: str = Field(description="Monzo Client ID")
    client_secret: SecretStr = Field(description="Monzo Client Secret")

    @final
    @override
    def refresh(self, state: MonzoData) -> None:
        state.refresh(self.client_id, self.client_secret.get_secret_value(), interactive_auth=self.interactive_auth)

    @final
    @override
    def extract_available_balances(self, state: MonzoData) -> dict[str, tuple[Decimal, str]]:
        res: dict[str, tuple[Decimal, str]] = {}
        for account_id, account_data in state.accounts.items():
            if not account_data.balances:
                continue
            if account_data.account.closed:
                continue
            if account_data.account.type == "uk_rewards":
                continue
            highest_ts = max(account_data.balances.keys())
            bal = account_data.balances[highest_ts]

            amount = Decimal(bal.balance) / pow(10, self.units)
            res[account_id] = (amount, bal.currency)
        return res

    @final
    @override
    def extract(self, state: MonzoData) -> list[ImportedTransaction]:
        return state.extract(units=self.units)

beancount_blue.importer.starling

beancount_blue.importer.starling.StarlingImporter

Bases: APIImporter[StarlingData]

Starling API Setup Instructions

To sync your Starling account, you need to generate a Personal Access Token from the Starling Developer portal.

1. Log in: Go to developer.starlingbank.com and create a developer account if you haven't already.
2. Connect your Bank Account: Follow the prompts to link your actual Starling Bank account to your developer account. You will need the Starling app on your phone to approve this.
3. Create a Token:
4. Navigate to "Personal Access Tokens" in the developer dashboard.
5. Click "Create Token".
6. Give it a name (e.g., Beancount Fava Sync).
7. Ensure you grant it read-only scopes for account, balance, and transaction data. Do not grant payment or write scopes.
8. Copy the Token: Once generated, copy the token immediately. You will not be able to see it again.
9. Configure Fava: Paste this token into the personal_access_token field below.

Source code in beancount_blue/importer/starling.py

class StarlingImporter(APIImporter[StarlingData]):
    """
    ### Starling API Setup Instructions

    To sync your Starling account, you need to generate a Personal Access Token from the Starling Developer portal.

    1. **Log in:** Go to [developer.starlingbank.com](https://developer.starlingbank.com/) and create a developer account if you haven't already.
    2. **Connect your Bank Account:** Follow the prompts to link your actual Starling Bank account to your developer account. You will need the Starling app on your phone to approve this.
    3. **Create a Token:**
       - Navigate to **"Personal Access Tokens"** in the developer dashboard.
       - Click **"Create Token"**.
       - Give it a name (e.g., `Beancount Fava Sync`).
       - Ensure you grant it **read-only** scopes for `account`, `balance`, and `transaction` data. Do not grant payment or write scopes.
    4. **Copy the Token:** Once generated, copy the token immediately. You will not be able to see it again.
    5. **Configure Fava:** Paste this token into the `personal_access_token` field below.
    """

    importer_name: Literal["starling"] = "starling"  # pyright: ignore[reportIncompatibleVariableOverride]

    personal_access_token: SecretStr = Field(..., description="Starling Personal Access Token")
    # For API access updating
    since_date: str | None = Field(None, description="Fetch transactions since this date.")
    spending_category_map: dict[str, str] = Field(
        default_factory=dict, description="Map of spending categories to Beancount account names."
    )
    faster_payments_map: dict[str, str] = Field(
        default_factory=dict, description="Map of faster payment identifiers to Beancount account names."
    )
    user_map: dict[str, str] = Field(default_factory=dict, description="Map of user UIDs to names.")

    def _get_counter_account(self, item: FeedItem, account_name: str) -> str | None:
        # spending_category_map = self.spending_category_map
        # faster_payments_map = self.faster_payments_map

        # # Try faster payments
        # if (
        #     item.source in [FeedItemSource.FASTER_PAYMENTS_IN, FeedItemSource.FASTER_PAYMENTS_OUT]
        #     and item.counterPartySubEntityIdentifier
        #     and item.counterPartySubEntitySubIdentifier
        # ):
        #     acct = f"{item.counterPartySubEntityIdentifier}-{item.counterPartySubEntitySubIdentifier}"
        #     if acct in faster_payments_map:
        #         return faster_payments_map[acct]

        # Try internal transfers
        if item.source == FeedItemSource.INTERNAL_TRANSFER:
            return f"{account_name}:{cleanup_string(item.counterPartyName)}"

        # # Try On Us Pay Me
        # if item.source == FeedItemSource.ON_US_PAY_ME:
        #     return "Assets:ZeroSumTransfer"

        # # Try spending category
        # if item.spendingCategory and item.spendingCategory in spending_category_map:
        #     return spending_category_map[item.spendingCategory]

        # # Default category
        # if item.spendingCategory:
        #     cp = spending_category_map.get("DEFAULT", "Expenses:Unknown:<CATEGORY>")
        #     return cp.replace("<CATEGORY>", cleanup_string(item.spendingCategory))

        return None

    @final
    @override
    def refresh(self, state: StarlingData) -> None:
        headers = {"Authorization": f"Bearer {self.personal_access_token.get_secret_value()}"}
        with httpx.Client(headers=headers, base_url=BASE_URL) as client:
            # Get accounts
            accounts_response = client.get("/api/v2/accounts")
            _ = accounts_response.raise_for_status()
            accounts = AccountsResponse.model_validate(accounts_response.json()).accounts

            now_utc = datetime.now(UTC)
            max_window_cutoff = now_utc - timedelta(days=364)

            for account in accounts:
                state.accounts[account.accountUid] = account

                # Get balance
                balance_response = client.get(f"/api/v2/accounts/{account.accountUid}/balance")
                _ = balance_response.raise_for_status()
                bal = BalanceResponse.model_validate(balance_response.json())
                state.balances[account.accountUid] = bal.effectiveBalance

                # Get spaces
                space_response = client.get(f"/api/v2/account/{account.accountUid}/spaces")
                _ = space_response.raise_for_status()
                spaces = SpacesResponse.model_validate(space_response.json())
                state.account_spending_spaces[account.accountUid] = spaces.spendingSpaces
                state.account_savings_spaces[account.accountUid] = spaces.savingsGoals

                # Get list of categories
                categories = (
                    [account.defaultCategory]
                    + list(s.spaceUid for s in state.account_spending_spaces[account.accountUid])
                    + list(s.savingsGoalUid for s in state.account_savings_spaces[account.accountUid])
                )

                # Get feed items for default category and all spaces
                for category_uid in categories:
                    category_items = [item for item in state.feed_items.values() if item.categoryUid == category_uid]

                    if self.since_date:
                        try:
                            d = date.fromisoformat(self.since_date)
                            target_since = datetime.combine(d, datetime.min.time(), tzinfo=UTC)
                        except ValueError:
                            target_since = datetime.fromisoformat(self.since_date)
                            if target_since.tzinfo is None:
                                target_since = target_since.replace(tzinfo=UTC)
                    elif category_items:
                        latest_time = max(
                            item.updatedAt if item.updatedAt.tzinfo else item.updatedAt.replace(tzinfo=UTC)
                            for item in category_items
                        )
                        target_since = latest_time - timedelta(days=7)
                    else:
                        target_since = max_window_cutoff

                    # Historical chunking if target_since is older than 364 days
                    current_start = target_since
                    while current_start < max_window_cutoff:
                        current_end = min(current_start + timedelta(days=364), max_window_cutoff)
                        params = {
                            "minTransactionTimestamp": current_start.strftime("%Y-%m-%dT%H:%M:%S.000Z"),
                            "maxTransactionTimestamp": current_end.strftime("%Y-%m-%dT%H:%M:%S.000Z"),
                        }
                        hist_response = client.get(
                            f"/api/v2/feed/account/{account.accountUid}/category/{category_uid}/transactions-between",
                            params=params,
                            timeout=30,
                        )
                        hist_response.raise_for_status()
                        for item in FeedItemsResponse.model_validate(hist_response.json()).feedItems:
                            state.feed_items[item.feedItemUid] = item

                        current_start = current_end

                    # Fetch active window / latest changes using changesSince
                    changes_since_dt = max(target_since, max_window_cutoff)
                    params = {"changesSince": changes_since_dt.strftime("%Y-%m-%dT%H:%M:%S.000Z")}

                    feed_response = client.get(
                        f"/api/v2/feed/account/{account.accountUid}/category/{category_uid}",
                        params=params,
                        timeout=30,
                    )
                    feed_response.raise_for_status()
                    feed_items = FeedItemsResponse.model_validate(feed_response.json()).feedItems

                    for item in feed_items:
                        state.feed_items[item.feedItemUid] = item

    @final
    @override
    def extract_available_balances(self, state: StarlingData) -> dict[str, tuple[Decimal, str]]:
        res: dict[str, tuple[Decimal, str]] = {}
        for accountUid, bal in state.balances.items():
            amount = Decimal(bal.minorUnits) / 100
            res[str(accountUid) + ":Main"] = (amount, bal.currency)

            for savingSpace in state.account_savings_spaces[accountUid]:
                if savingSpace.totalSaved:
                    res[str(accountUid) + ":" + savingSpace.name] = (
                        Decimal(savingSpace.totalSaved.minorUnits / 100),
                        savingSpace.totalSaved.currency,
                    )

            for spendingSpace in state.account_spending_spaces[accountUid]:
                res[str(accountUid) + ":" + spendingSpace.name] = (
                    Decimal(spendingSpace.balance.minorUnits / 100),
                    spendingSpace.balance.currency,
                )

        return res

    @final
    @override
    def extract(self, state: StarlingData) -> list[ImportedTransaction]:
        transactions: list[ImportedTransaction] = []

        # TODO: Friendly mapping for import process
        user_map = {UUID(k): v for k, v in self.user_map.items()}

        accounts: set[UUID] = set()
        category_map: dict[UUID, str] = {}
        for accountUid, account in state.accounts.items():
            category_map[account.defaultCategory] = str(accountUid) + ":Main"
            accounts.add(account.defaultCategory)
            for space in state.account_spending_spaces[accountUid]:
                category_map[space.spaceUid] = str(accountUid) + ":" + cleanup_string(space.name)
            for space in state.account_savings_spaces[accountUid]:
                category_map[space.savingsGoalUid] = str(accountUid) + ":" + cleanup_string(space.name)

        for item in state.feed_items.values():
            account_name = category_map.get(item.categoryUid)
            if not account_name:
                log.warning(f"Could not find beancount account name for account {item.categoryUid}")
                continue

            amount = Decimal(item.amount.minorUnits) / 100
            if item.direction == Direction.OUT:
                amount = -amount

            # Zero amount for declined, reversed, refunded
            if (
                item.status == FeedItemStatus.DECLINED
                or item.status == FeedItemStatus.REVERSED
                or item.status == FeedItemStatus.REFUNDED
            ):
                amount = Decimal(0)

            # Skip internal transfers for non-primary accounts
            if item.source == FeedItemSource.INTERNAL_TRANSFER and item.categoryUid not in accounts:
                continue

            # Determine counter account for internal transfers
            counter_account: str | None = None
            if item.source == FeedItemSource.INTERNAL_TRANSFER and item.counterPartyUid:
                counter_account = category_map.get(item.counterPartyUid)
                if not counter_account:
                    log.warning(f"Could not find counter account name for account {item.counterPartyUid}")
                    continue

            # counter_account = self._get_counter_account(item, account_name)

            meta: dict[str, Any] = {
                "__source__": item.model_dump_json(indent=2),
            }
            if item.settlementTime and item.transactionTime.date() != item.settlementTime.date():
                meta["transaction_date"] = item.transactionTime.date().isoformat()
            if item.transactingApplicationUserUid:
                user = user_map.get(item.transactingApplicationUserUid)
                if user:
                    meta["user"] = user
            if item.userNote:
                meta["note"] = item.userNote

            transactions.append(
                ImportedTransaction(
                    id=str(item.feedItemUid),
                    date=item.settlementTime.date() if item.settlementTime else item.transactionTime.date(),
                    settled=item.status == FeedItemStatus.SETTLED,
                    amount=amount,
                    currency=item.amount.currency,
                    account=account_name,
                    counter_account=counter_account,
                    narration=f"{item.counterPartyName} {item.reference or ''}".strip(),
                    payee=item.counterPartyName,
                    category=item.spendingCategory,
                    meta=meta,
                )
            )
        return transactions

beancount_blue.importer.truelayer

beancount_blue.importer.truelayer.TrueLayerImporter

Bases: APIImporter[TrueLayerData]

TrueLayer API Setup Instructions

TrueLayer allows you to connect to dozens of UK/EU banks (like Amex, Chase, Barclaycard) using Open Banking.

1. Create an Account: Go to console.truelayer.com and create a free developer account.
2. Create an Application:
3. In the TrueLayer Console, create a new application.
4. Go to "App Settings".
5. Under "Redirect URIs", add http://localhost:8000/callback (or your preferred local callback URL).
6. Enable Data API: Ensure the "Data" (Open Banking) product is enabled for your application.
7. Copy Credentials: Go to "App settings" -> "Credentials" and copy your Client ID and Client Secret.
8. Configure Fava: Paste these into the client_id and client_secret fields below.

Note: To link a specific bank account, you will typically need to complete TrueLayer's Auth Link flow in a browser to authorize the connection and get a valid access_token.

Source code in beancount_blue/importer/truelayer.py

class TrueLayerImporter(APIImporter[TrueLayerData]):
    """
    ### TrueLayer API Setup Instructions

    TrueLayer allows you to connect to dozens of UK/EU banks (like Amex, Chase, Barclaycard) using Open Banking.

    1. **Create an Account:** Go to [console.truelayer.com](https://console.truelayer.com/) and create a free developer account.
    2. **Create an Application:**
       - In the TrueLayer Console, create a new application.
       - Go to **"App Settings"**.
       - Under **"Redirect URIs"**, add `http://localhost:8000/callback` (or your preferred local callback URL).
    3. **Enable Data API:** Ensure the **"Data"** (Open Banking) product is enabled for your application.
    4. **Copy Credentials:** Go to **"App settings" -> "Credentials"** and copy your **Client ID** and **Client Secret**.
    5. **Configure Fava:** Paste these into the `client_id` and `client_secret` fields below.

    *Note: To link a specific bank account, you will typically need to complete TrueLayer's Auth Link flow in a browser to authorize the connection and get a valid `access_token`.*
    """

    importer_name: Literal["truelayer"] = "truelayer"  # pyright: ignore[reportIncompatibleVariableOverride]

    client_id: str = Field(description="Truelayer Client ID")
    client_secret: SecretStr = Field(description="Truelayer Client Secret")

    units: int = 2

    @final
    @override
    def handle_callback(self, code: str, state: str) -> None:
        if not self.cache_data:
            from beancount_blue.importer.delta_importer import ImporterConfigurationError

            raise ImporterConfigurationError("No cache_data path configured for callback.")

        api_data_type = self.get_types()
        from beancount_blue.importer.delta_importer import ImporterState

        state_type = ImporterState[api_data_type]
        from beancount_blue.importer.utils import load

        with load(self.cache_data, state_type) as importer_state:
            api = TrueLayerAPI(
                client_id=self.client_id,
                client_secret=self.client_secret.get_secret_value(),
                state=importer_state.data,
                interactive_auth=False,
                redirect_uri=self.redirect_uri,
            )

            # Exchange Code for Token
            token = cast(
                dict[str, Any],
                api.client.fetch_token(  # type: ignore[reportUnknownMemberType]
                    api.TOKEN_ENDPOINT,
                    grant_type="authorization_code",
                    code=code,
                    redirect_uri=self.redirect_uri,
                ),
            )
            api.update_token_callback(token)

    @final
    @override
    def refresh(self, state: TrueLayerData) -> None:
        is_first_run = state.token is None

        api = TrueLayerAPI(
            self.client_id,
            self.client_secret.get_secret_value(),
            state,
            interactive_auth=self.interactive_auth,
            redirect_uri=self.redirect_uri,
        )
        api.ensure_authorized()

        # 1. Accounts
        state.raw_accounts = api.get_accounts()
        for account in state.raw_accounts:
            if account.account_id not in state.accounts:
                state.accounts[account.account_id] = TrueLayerAccountConfig(
                    account_id=account.account_id,
                    name=account.display_name,
                    liability=False,
                )

        # 2. Cards
        state.raw_cards = api.get_cards()
        for card in state.raw_cards:
            if card.account_id not in state.cards:
                state.cards[card.account_id] = TrueLayerAccountConfig(
                    account_id=card.account_id,
                    name=card.display_name,
                    liability=card.card_type == "CREDIT",
                )

        # 3. Transactions & Balances
        state.raw_transactions = {}
        state.raw_pending_transactions = {}
        state.raw_balances = {}

        for type_ in ACCOUNT_TYPES:
            # type_ is "accounts" or "cards"
            config_map = state.accounts if type_ == "accounts" else state.cards
            # We need to map the string type_ to the Literal expected by get_transactions
            # This cast is safe because ACCOUNT_TYPES is defined as ("accounts", "cards")
            api_type = type_

            for account_config in config_map.values():
                if not account_config.enabled:
                    continue

                aid = account_config.account_id
                from_date = datetime.datetime.fromtimestamp(account_config.from_date, datetime.UTC)
                if not is_first_run:
                    from_date = max(from_date, datetime.datetime.now(datetime.UTC) - datetime.timedelta(days=89))
                to_date = datetime.datetime.now(datetime.UTC)

                state.raw_transactions[aid] = api.get_transactions(aid, api_type, from_date, to_date)
                state.raw_pending_transactions[aid] = api.get_transactions(
                    aid, api_type, from_date, to_date, pending=True
                )

                bal = api.get_balance(aid, api_type)
                if bal:
                    state.raw_balances[aid] = bal

    @final
    @override
    def extract_available_balances(self, state: TrueLayerData) -> dict[str, tuple[Decimal, str]]:
        res: dict[str, tuple[Decimal, str]] = {}
        for account_id, bal in state.raw_balances.items():
            # Use available balance, fallback to current if available is missing
            amount_val = bal.available if bal.available is not None else bal.current
            if amount_val is not None:
                res[account_id] = (Decimal(str(amount_val)), bal.currency)
        return res

    @final
    @override
    def extract(self, state: TrueLayerData) -> list[ImportedTransaction]:
        entries: list[ImportedTransaction] = []

        for config_map in [state.accounts, state.cards]:
            for account_config in config_map.values():
                if not account_config.enabled:
                    continue

                aid = account_config.account_id

                # Settled
                for txn in state.raw_transactions.get(aid, []):
                    entries.append(self._transform_transaction(txn, aid))

                # Pending
                for txn in state.raw_pending_transactions.get(aid, []):
                    t = self._transform_transaction(txn, aid)
                    t.settled = False
                    entries.append(t)

        return entries

    def _transform_transaction(self, txn: TrueLayerTransaction, account_id: str) -> ImportedTransaction:
        # Standardize amount: DEBIT should be negative (money out), CREDIT positive (money in)
        # We use abs() to handle cases where the provider might have already signed the amount.
        val = currency_to_decimal(txn.amount)
        amount = -abs(val) if txn.transaction_type == "DEBIT" else abs(val)

        date = dateutil.parser.parse(txn.timestamp).date()

        payee = txn.merchant_name or txn.meta.get("provider_merchant_name")

        # Robust ID generation
        if txn.transaction_id:
            txn_id = txn.transaction_id
        else:
            # Hash the content if no ID is provided
            content_str = f"{date.isoformat()}|{amount}|{txn.description}|{txn.currency}"
            txn_id = hashlib.sha256(content_str.encode("utf-8")).hexdigest()

        return ImportedTransaction(
            id=txn_id,
            date=date,
            settled=True,
            amount=amount,
            currency=txn.currency,
            account=account_id,
            narration=txn.description,
            payee=payee,
            category=txn.transaction_category,
            meta={
                "type": "truelayer",
                "category": txn.transaction_category or "",
                "classification": ",".join(txn.transaction_classification),
            },
        )