finances

finances

Personal finances for the native Succès domain — local-first CAD budgeting.

Accounts, categories, transactions, subscriptions, budgets and savings goals live beside habits/notes in succes.db. Money is stored as integer cents. Remote sync is wired later; mutations still go through the operation log.

Classes

SuccesFinancesStore

SuccesFinancesStore(db_path: str | Path | None = None)

Bases: SuccesContinuityStore

Accounts, cashflow, subscriptions, budgets and savings goals.

Source code in src/diapason/succes/finances.py

def __init__(self, db_path: str | Path | None = None) -> None:
    super().__init__(db_path)
    with self._connect() as conn:
        conn.executescript(_FINANCES_SCHEMA)
        self._seed_defaults(conn)
        conn.commit()

Methods:

materialize_due_subscriptions

materialize_due_subscriptions(
    *, on_date: str | None = None
) -> dict[str, Any]

Create expense transactions for subscriptions due on or before on_date.

Source code in src/diapason/succes/finances.py

def materialize_due_subscriptions(
    self, *, on_date: str | None = None
) -> dict[str, Any]:
    """Create expense transactions for subscriptions due on or before on_date."""
    today = _validate_iso_date(on_date) if on_date else date.today().isoformat()
    created: list[dict[str, Any]] = []
    with self._transaction() as conn:
        rows = conn.execute(
            """SELECT * FROM succes_subscriptions
               WHERE deleted_at_ms IS NULL AND active=1 AND next_due_date<=?
               ORDER BY next_due_date""",
            (today,),
        ).fetchall()
        for row in rows:
            sub = self._subscription_dict(row)
            account_id = sub["accountId"]
            if not account_id:
                # Use first non-archived account.
                acct = conn.execute(
                    """SELECT id FROM succes_accounts
                       WHERE deleted_at_ms IS NULL AND archived=0
                       ORDER BY name LIMIT 1"""
                ).fetchone()
                if acct is None:
                    continue
                account_id = acct["id"]
            txn_id = str(uuid.uuid4())
            stamp = now_ms()
            conn.execute(
                """INSERT INTO succes_transactions
                   (id, account_id, category_id, txn_type, amount_cents, currency,
                    txn_date, payee, notes, transfer_account_id, subscription_id,
                    import_hash, updated_at_ms)
                   VALUES (?, ?, ?, 'expense', ?, 'CAD', ?, ?, ?, '', ?, '', ?)""",
                (
                    txn_id,
                    account_id,
                    sub["categoryId"],
                    int(round(sub["amount"] * 100)),
                    sub["nextDueDate"],
                    sub["name"],
                    f"Abonnement ({sub['cadence']})",
                    sub["id"],
                    stamp,
                ),
            )
            next_due = sub["nextDueDate"]
            # Advance until after today (catch up missed periods).
            guard = 0
            while next_due <= today and guard < 36:
                next_due = _advance_due(next_due, sub["cadence"])
                guard += 1
            conn.execute(
                "UPDATE succes_subscriptions SET next_due_date=?, updated_at_ms=? "
                "WHERE id=?",
                (next_due, stamp, sub["id"]),
            )
            created.append(self._load_txn(conn, txn_id) or {})
    return {"created": created, "count": len(created)}

import_csv

import_csv(
    csv_text: str,
    *,
    account_id: str,
    mapping: Mapping[str, str] | None = None,
) -> dict[str, Any]

Import bank CSV. mapping keys: date, amount, payee, notes, type, category.

Source code in src/diapason/succes/finances.py

def import_csv(
    self,
    csv_text: str,
    *,
    account_id: str,
    mapping: Mapping[str, str] | None = None,
) -> dict[str, Any]:
    """Import bank CSV. mapping keys: date, amount, payee, notes, type, category."""
    with self._connect() as conn:
        if self._load_account(conn, account_id) is None:
            raise SuccesNotFound("Ce compte n'existe pas.")
        categories = {
            row["name"].casefold(): self._category_dict(row)
            for row in conn.execute(
                "SELECT * FROM succes_finance_categories WHERE deleted_at_ms IS "
                "NULL"
            ).fetchall()
        }

    map_keys = {
        "date": (mapping or {}).get("date") or "date",
        "amount": (mapping or {}).get("amount") or "amount",
        "payee": (mapping or {}).get("payee") or "description",
        "notes": (mapping or {}).get("notes") or "notes",
        "type": (mapping or {}).get("type") or "type",
        "category": (mapping or {}).get("category") or "category",
    }

    reader = csv.DictReader(io.StringIO(csv_text))
    if not reader.fieldnames:
        raise SuccesError("Le fichier CSV est vide ou sans en-têtes.")

    created = 0
    skipped = 0
    errors: list[str] = []

    with self._transaction() as conn:
        for index, row in enumerate(reader, start=2):
            try:
                raw_date = str(row.get(map_keys["date"]) or "").strip()
                raw_amount = str(row.get(map_keys["amount"]) or "").strip()
                if not raw_date or not raw_amount:
                    skipped += 1
                    continue
                # Accept YYYY-MM-DD or DD/MM/YYYY
                if "/" in raw_date:
                    parts = raw_date.split("/")
                    if len(parts) == 3:
                        raw_date = (
                            f"{parts[2]}-{parts[1].zfill(2)}-{parts[0].zfill(2)}"
                        )
                txn_date = _validate_iso_date(raw_date)
                signed = float(
                    raw_amount.replace(",", ".").replace("$", "").replace(" ", "")
                )
                txn_type = str(row.get(map_keys["type"]) or "").strip().lower()
                if txn_type not in TXN_TYPES:
                    txn_type = "expense" if signed < 0 else "income"
                amount_cents = int(round(abs(signed) * 100))
                payee = str(row.get(map_keys["payee"]) or "").strip()[:160]
                notes = str(row.get(map_keys["notes"]) or "").strip()[:2000]
                cat_name = str(row.get(map_keys["category"]) or "").strip()
                category_id = ""
                if cat_name:
                    match = categories.get(cat_name.casefold())
                    if match:
                        category_id = match["id"]

                digest = hashlib.sha256(
                    f"{account_id}|{txn_date}|{amount_cents}|{payee}|{txn_type}".encode()
                ).hexdigest()
                exists = conn.execute(
                    """SELECT id FROM succes_transactions
                       WHERE import_hash=? AND deleted_at_ms IS NULL""",
                    (digest,),
                ).fetchone()
                if exists:
                    skipped += 1
                    continue

                txn_id = str(uuid.uuid4())
                stamp = now_ms()
                conn.execute(
                    """INSERT INTO succes_transactions
                       (id, account_id, category_id, txn_type, amount_cents, currency,
                        txn_date, payee, notes, transfer_account_id, subscription_id,
                        import_hash, updated_at_ms)
                       VALUES (?, ?, ?, ?, ?, 'CAD', ?, ?, ?, '', '', ?, ?)""",
                    (
                        txn_id,
                        account_id,
                        category_id,
                        txn_type,
                        amount_cents,
                        txn_date,
                        payee,
                        notes,
                        digest,
                        stamp,
                    ),
                )
                created += 1
            except Exception as exc:  # noqa: BLE001 — collect row errors
                errors.append(f"Ligne {index}: {exc}")
                if len(errors) >= 20:
                    break

    return {"created": created, "skipped": skipped, "errors": errors}