bin/Plan.md

Finanzberatungs-Tool — Implementierungsplan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Ziel: Finanzberatungs-Tool als rootless-Podman-Pod (Postgres + FastAPI-App mit Web-GUI + Grafana) in ~/bin, plus Claude-Code-Beratungsumgebung in ~/fb.

Spec: /home/wlfb/fb/docs/superpowers/specs/2026-07-17-finanzberatung-design.md — bei Fragen dort nachlesen.

Architektur: Eine FastAPI-App liefert REST-API (für Claude Code, Bearer-API-Key) und server-gerenderte Web-GUI (Jinja2 + HTMX, Session-Login). PDF-Auszüge (Volksbank/Raiffeisen, HypoVereinsbank, DKB) werden durch deterministische pdfplumber-Parser mit Saldo-Plausibilisierung und Duplikat-Schutz importiert — immer mit Vorschau + Bestätigung. Eine reine Python-Szenario-Engine rechnet tagesgenaue Liquiditätsvorschauen (wiederkehrende Posten, Einmalposten, Kredite, Kürzungs-Modifikatoren). Grafana wird vollständig provisioniert und liest über SQL-Views.

Tech-Stack: Python 3.12, FastAPI, SQLAlchemy 2 + Alembic, Postgres 17 (Prod) / SQLite (Tests), pdfplumber, Jinja2 + HTMX, Grafana OSS, rootless Podman + systemd-User-Services.

Globale Vorgaben

• Arbeitsverzeichnis ist das Git-Repo /home/wlfb/bin; Tool-Code unter finance/. Phase 8 schreibt zusätzlich ins Repo /home/wlfb/fb (absolute Pfade, dort separat committen).
• Alle Container-Images mit gepinnten, unveränderlichen Tags (kein :latest); Vorbild: example_create_pod_langflow.sh.
• Ports: API/GUI 127.0.0.1:8096 → 8000, Grafana 127.0.0.1:8097 → 3000, Postgres nur pod-intern.
• Persistente Daten unter ~/.local/share/finance_pod/ (Bind-Mounts mit :Z), systemd-Units in ~/.config/systemd/user/.
• Geldbeträge immer Decimal (nie float), DB-Typ Numeric(12, 2); positive Beträge = Eingang, negative = Ausgang.
• Secrets nur in finance/.env (gitignored); niemals Secrets, Fixture-PDFs oder echte Kontodaten committen.
• Sprache der GUI und aller Nutzertexte: Deutsch.
• Vor jedem Commit: cd /home/wlfb/bin/finance && python -m pytest -q muss grün sein (bzw. geskippte Fixture-Tests sind ok).
• Commit-Messages enden mit Co-Authored-By:-Zeile gemäß Harness-Vorgabe.

---

Phase 1: Projekt-Gerüst

Task 1: Python-Projekt finance/ anlegen

Files:

• Create: finance/requirements.txt, finance/requirements-dev.txt, finance/pytest.ini, finance/.gitignore, finance/app/__init__.py, finance/app/config.py, finance/tests/__init__.py, finance/tests/test_config.py

Interfaces:

• Produces: app.config.get_settings() -> Settings mit Feldern database_url: str, api_key: str, gui_user: str, gui_password_hash: str, session_secret: str, inbox_dir: Path, uploads_dir: Path, warn_threshold: Decimal, horizon_days: int (Default 548 = 18 Monate).

• Step 1: Dateien anlegen

finance/requirements.txt:

fastapi==0.116.1
uvicorn==0.35.0
sqlalchemy==2.0.41
alembic==1.16.2
psycopg[binary]==3.2.9
jinja2==3.1.6
python-multipart==0.0.20
itsdangerous==2.2.0
pdfplumber==0.11.7

finance/requirements-dev.txt:

-r requirements.txt
pytest==8.4.1
httpx==0.28.1

Falls eine gepinnte Version bei pip install nicht auflösbar ist: auf die nächste verfügbare Patch-Version anheben und im Commit erwähnen.

finance/pytest.ini:

[pytest]
testpaths = tests

finance/.gitignore:

.env
__pycache__/
*.sqlite
tests/fixtures/*.pdf
.venv/

finance/app/config.py:

import os
from dataclasses import dataclass
from decimal import Decimal
from functools import lru_cache
from pathlib import Path


@dataclass(frozen=True)
class Settings:
    database_url: str
    api_key: str
    gui_user: str
    gui_password_hash: str
    session_secret: str
    inbox_dir: Path
    uploads_dir: Path
    warn_threshold: Decimal
    horizon_days: int


@lru_cache
def get_settings() -> Settings:
    e = os.environ.get
    return Settings(
        database_url=e("FB_DATABASE_URL", "sqlite:///./fb.sqlite"),
        api_key=e("FB_API_KEY", ""),
        gui_user=e("FB_GUI_USER", "admin"),
        gui_password_hash=e("FB_GUI_PASSWORD_HASH", ""),
        session_secret=e("FB_SESSION_SECRET", "dev-secret"),
        inbox_dir=Path(e("FB_INBOX_DIR", "./inbox")),
        uploads_dir=Path(e("FB_UPLOADS_DIR", "./uploads")),
        warn_threshold=Decimal(e("FB_WARN_THRESHOLD", "0")),
        horizon_days=int(e("FB_HORIZON_DAYS", "548")),
    )

finance/tests/test_config.py:

from app.config import get_settings


def test_defaults():
    s = get_settings()
    assert s.horizon_days == 548
    assert s.gui_user == "admin"

• Step 2: venv anlegen, Dependencies installieren, Test ausführen

cd /home/wlfb/bin/finance
python3 -m venv .venv && .venv/bin/pip install -r requirements-dev.txt
.venv/bin/python -m pytest -q

Erwartet: 1 passed.

• Step 3: Commit

cd /home/wlfb/bin && git add finance && git commit -m "feat: Projekt-Geruest finance/ mit Config und Test-Setup"

---

Phase 2: Datenmodell

Task 2: SQLAlchemy-Modelle + Alembic

Files:

• Create: finance/app/models/__init__.py, finance/app/models/tables.py, finance/app/db.py, finance/alembic.ini, finance/alembic/env.py, finance/alembic/script.py.mako (von alembic init), finance/tests/conftest.py, finance/tests/test_models.py

Interfaces:

• Produces: ORM-Klassen Account, Category, CategoryRule, Statement, Transaction, RecurringItem, PlannedItem, Loan, Scenario, ScenarioLoan, ScenarioModifier, ProjectionPoint, ProjectionResult (Import aus app.models.tables), app.db.get_session() (FastAPI-Dependency), app.db.Base.

• Statement.status ∈ {"draft", "confirmed", "error"}; Transaction.status ∈ {"draft", "confirmed"}; RecurringItem.rhythm ∈ {"monthly", "quarterly", "yearly"}; Loan.repayment_type ∈ {"annuity", "bullet"}; ScenarioModifier.target_type ∈ {"category", "recurring"}, kind ∈ {"percent", "absolute", "remove"}.

• Step 1: Failing Test schreiben

finance/tests/conftest.py:

import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import Session
from sqlalchemy.pool import StaticPool

from app.db import Base


@pytest.fixture
def db():
    engine = create_engine(
        "sqlite://", poolclass=StaticPool,
        connect_args={"check_same_thread": False},
    )
    Base.metadata.create_all(engine)
    with Session(engine) as session:
        yield session

finance/tests/test_models.py:

from datetime import date
from decimal import Decimal

from app.models.tables import Account, Statement, Transaction


def test_account_transaction_roundtrip(db):
    acc = Account(bank="DKB", iban="DE02120300000000202051", name="Giro", type="giro")
    db.add(acc)
    db.flush()
    st = Statement(filename="a.pdf", bank="DKB", account_id=acc.id,
                   period_start=date(2026, 6, 1), period_end=date(2026, 6, 30),
                   opening_balance=Decimal("100.00"), closing_balance=Decimal("50.00"),
                   status="draft")
    db.add(st)
    db.flush()
    tx = Transaction(account_id=acc.id, statement_id=st.id,
                     booking_date=date(2026, 6, 3), value_date=date(2026, 6, 3),
                     amount=Decimal("-50.00"), purpose="Miete Juni",
                     counterparty="Vermieter GmbH", status="draft",
                     dedup_hash="abc", is_duplicate=False)
    db.add(tx)
    db.commit()
    assert db.query(Transaction).one().amount == Decimal("-50.00")

• Step 2: Test ausführen — erwartet FAIL (ModuleNotFoundError: app.db)

• Step 3: Implementieren

finance/app/db.py:

from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, Session

from app.config import get_settings


class Base(DeclarativeBase):
    pass


_engine = None


def get_engine():
    global _engine
    if _engine is None:
        _engine = create_engine(get_settings().database_url)
    return _engine


def get_session():
    with Session(get_engine()) as session:
        yield session

finance/app/models/tables.py (vollständig):

from datetime import date, datetime
from decimal import Decimal

from sqlalchemy import Boolean, Date, DateTime, ForeignKey, Integer, Numeric, String, Text
from sqlalchemy.orm import Mapped, mapped_column

from app.db import Base

MONEY = Numeric(12, 2)


class Account(Base):
    __tablename__ = "accounts"
    id: Mapped[int] = mapped_column(primary_key=True)
    bank: Mapped[str] = mapped_column(String(50))
    iban: Mapped[str] = mapped_column(String(34), unique=True)
    name: Mapped[str] = mapped_column(String(100))
    type: Mapped[str] = mapped_column(String(20), default="giro")


class Category(Base):
    __tablename__ = "categories"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100), unique=True)


class CategoryRule(Base):
    __tablename__ = "category_rules"
    id: Mapped[int] = mapped_column(primary_key=True)
    pattern: Mapped[str] = mapped_column(String(200))
    category_id: Mapped[int] = mapped_column(ForeignKey("categories.id"))
    priority: Mapped[int] = mapped_column(Integer, default=100)


class Statement(Base):
    __tablename__ = "statements"
    id: Mapped[int] = mapped_column(primary_key=True)
    filename: Mapped[str] = mapped_column(String(255))
    bank: Mapped[str] = mapped_column(String(50))
    account_id: Mapped[int | None] = mapped_column(ForeignKey("accounts.id"), nullable=True)
    period_start: Mapped[date | None] = mapped_column(Date, nullable=True)
    period_end: Mapped[date | None] = mapped_column(Date, nullable=True)
    opening_balance: Mapped[Decimal | None] = mapped_column(MONEY, nullable=True)
    closing_balance: Mapped[Decimal | None] = mapped_column(MONEY, nullable=True)
    status: Mapped[str] = mapped_column(String(20), default="draft")
    error_message: Mapped[str | None] = mapped_column(Text, nullable=True)


class Transaction(Base):
    __tablename__ = "transactions"
    id: Mapped[int] = mapped_column(primary_key=True)
    account_id: Mapped[int] = mapped_column(ForeignKey("accounts.id"))
    statement_id: Mapped[int | None] = mapped_column(ForeignKey("statements.id"), nullable=True)
    booking_date: Mapped[date] = mapped_column(Date)
    value_date: Mapped[date | None] = mapped_column(Date, nullable=True)
    amount: Mapped[Decimal] = mapped_column(MONEY)
    purpose: Mapped[str] = mapped_column(Text, default="")
    counterparty: Mapped[str] = mapped_column(String(200), default="")
    category_id: Mapped[int | None] = mapped_column(ForeignKey("categories.id"), nullable=True)
    status: Mapped[str] = mapped_column(String(20), default="draft")
    dedup_hash: Mapped[str] = mapped_column(String(64), index=True)
    is_duplicate: Mapped[bool] = mapped_column(Boolean, default=False)


class RecurringItem(Base):
    __tablename__ = "recurring_items"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(200))
    amount: Mapped[Decimal] = mapped_column(MONEY)
    rhythm: Mapped[str] = mapped_column(String(20))
    due_day: Mapped[int] = mapped_column(Integer)
    start_date: Mapped[date | None] = mapped_column(Date, nullable=True)
    end_date: Mapped[date | None] = mapped_column(Date, nullable=True)
    category_id: Mapped[int | None] = mapped_column(ForeignKey("categories.id"), nullable=True)


class PlannedItem(Base):
    __tablename__ = "planned_items"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(200))
    amount: Mapped[Decimal] = mapped_column(MONEY)
    due: Mapped[date] = mapped_column(Date)
    category_id: Mapped[int | None] = mapped_column(ForeignKey("categories.id"), nullable=True)


class Loan(Base):
    __tablename__ = "loans"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(200))
    principal: Mapped[Decimal] = mapped_column(MONEY)
    annual_rate_pct: Mapped[Decimal] = mapped_column(Numeric(5, 2))
    term_months: Mapped[int] = mapped_column(Integer)
    payout_date: Mapped[date] = mapped_column(Date)
    repayment_type: Mapped[str] = mapped_column(String(20), default="annuity")


class Scenario(Base):
    __tablename__ = "scenarios"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(200), unique=True)
    description: Mapped[str] = mapped_column(Text, default="")
    include_recurring: Mapped[bool] = mapped_column(Boolean, default=True)
    include_planned: Mapped[bool] = mapped_column(Boolean, default=True)


class ScenarioLoan(Base):
    __tablename__ = "scenario_loans"
    scenario_id: Mapped[int] = mapped_column(ForeignKey("scenarios.id"), primary_key=True)
    loan_id: Mapped[int] = mapped_column(ForeignKey("loans.id"), primary_key=True)


class ScenarioModifier(Base):
    __tablename__ = "scenario_modifiers"
    id: Mapped[int] = mapped_column(primary_key=True)
    scenario_id: Mapped[int] = mapped_column(ForeignKey("scenarios.id"))
    target_type: Mapped[str] = mapped_column(String(20))
    target_id: Mapped[int] = mapped_column(Integer)
    kind: Mapped[str] = mapped_column(String(20))
    value: Mapped[Decimal] = mapped_column(MONEY, default=Decimal("0"))


class ProjectionPoint(Base):
    __tablename__ = "projection_points"
    id: Mapped[int] = mapped_column(primary_key=True)
    scenario_id: Mapped[int] = mapped_column(ForeignKey("scenarios.id"), index=True)
    day: Mapped[date] = mapped_column(Date)
    balance: Mapped[Decimal] = mapped_column(MONEY)


class ProjectionResult(Base):
    __tablename__ = "projection_results"
    scenario_id: Mapped[int] = mapped_column(ForeignKey("scenarios.id"), primary_key=True)
    computed_at: Mapped[datetime] = mapped_column(DateTime)
    low_point_date: Mapped[date] = mapped_column(Date)
    low_point_balance: Mapped[Decimal] = mapped_column(MONEY)
    below_zero_date: Mapped[date | None] = mapped_column(Date, nullable=True)
    below_threshold_date: Mapped[date | None] = mapped_column(Date, nullable=True)

finance/app/models/__init__.py:

from app.models import tables  # noqa: F401

• Step 4: Test ausführen — erwartet PASS (.venv/bin/python -m pytest -q)

• Step 5: Alembic initialisieren

cd /home/wlfb/bin/finance
.venv/bin/alembic init alembic

In alembic/env.py ergänzen (vor run_migrations_offline):

from app.config import get_settings
from app.db import Base
import app.models  # noqa: F401

config.set_main_option("sqlalchemy.url", get_settings().database_url)
target_metadata = Base.metadata

Dann initiale Migration erzeugen und lokal gegen SQLite prüfen:

FB_DATABASE_URL=sqlite:///./mig_test.sqlite .venv/bin/alembic revision --autogenerate -m "initial schema"
FB_DATABASE_URL=sqlite:///./mig_test.sqlite .venv/bin/alembic upgrade head
rm mig_test.sqlite

Erwartet: Migration in alembic/versions/ erzeugt, upgrade head läuft fehlerfrei.

• Step 6: Commit — git add finance && git commit -m "feat: Datenmodell (SQLAlchemy) und Alembic-Migration"

---

Phase 3: Szenario-Engine (reines Python, keine DB)

Task 3: Kredit-Mathematik

Files:

• Create: finance/app/engine/__init__.py, finance/app/engine/loans.py, finance/tests/test_loans.py

Interfaces:

• Produces: add_months(d: date, months: int) -> date; annuity_payment(principal: Decimal, annual_rate_pct: Decimal, term_months: int) -> Decimal; loan_schedule(principal, annual_rate_pct, term_months, payout_date, repayment_type) -> list[Installment] mit Installment(due: date, payment: Decimal, interest: Decimal, principal: Decimal, remaining: Decimal); payment ist negativ (Ausgabe).

• Step 1: Failing Tests schreiben

finance/tests/test_loans.py:

from datetime import date
from decimal import Decimal

from app.engine.loans import add_months, annuity_payment, loan_schedule


def test_add_months_clamps_month_end():
    assert add_months(date(2026, 1, 31), 1) == date(2026, 2, 28)


def test_annuity_payment():
    # 10.000 EUR, 6 % p.a., 48 Monate -> 234,85 EUR Monatsrate
    assert annuity_payment(Decimal("10000"), Decimal("6"), 48) == Decimal("234.85")


def test_annuity_schedule_amortizes_fully():
    plan = loan_schedule(Decimal("10000"), Decimal("6"), 48, date(2026, 8, 1), "annuity")
    assert len(plan) == 48
    assert plan[-1].remaining == Decimal("0.00")
    assert plan[0].due == date(2026, 9, 1)
    assert plan[0].interest == Decimal("50.00")  # 10000 * 0,5 %
    assert plan[0].payment < 0


def test_bullet_loan():
    plan = loan_schedule(Decimal("10000"), Decimal("6"), 12, date(2026, 8, 1), "bullet")
    assert all(i.principal == 0 for i in plan[:-1])
    assert plan[-1].principal == Decimal("10000")
    assert plan[-1].remaining == Decimal("0.00")

• Step 2: Ausführen — erwartet FAIL (Modul fehlt)

• Step 3: Implementieren

finance/app/engine/loans.py:

import calendar
from dataclasses import dataclass
from datetime import date
from decimal import ROUND_HALF_UP, Decimal

CENT = Decimal("0.01")


@dataclass(frozen=True)
class Installment:
    due: date
    payment: Decimal
    interest: Decimal
    principal: Decimal
    remaining: Decimal


def add_months(d: date, months: int) -> date:
    y, m0 = divmod(d.year * 12 + d.month - 1 + months, 12)
    m = m0 + 1
    return date(y, m, min(d.day, calendar.monthrange(y, m)[1]))


def annuity_payment(principal: Decimal, annual_rate_pct: Decimal, term_months: int) -> Decimal:
    i = annual_rate_pct / Decimal(100) / Decimal(12)
    if i == 0:
        return (principal / term_months).quantize(CENT, ROUND_HALF_UP)
    q = (Decimal(1) + i) ** term_months
    return (principal * i * q / (q - Decimal(1))).quantize(CENT, ROUND_HALF_UP)


def loan_schedule(principal: Decimal, annual_rate_pct: Decimal, term_months: int,
                  payout_date: date, repayment_type: str) -> list[Installment]:
    i = annual_rate_pct / Decimal(100) / Decimal(12)
    plan: list[Installment] = []
    remaining = principal
    if repayment_type == "bullet":
        for n in range(1, term_months + 1):
            interest = (remaining * i).quantize(CENT, ROUND_HALF_UP)
            principal_part = remaining if n == term_months else Decimal("0")
            remaining = remaining - principal_part
            plan.append(Installment(add_months(payout_date, n),
                                    -(interest + principal_part), interest,
                                    principal_part, remaining.quantize(CENT)))
        return plan
    rate = annuity_payment(principal, annual_rate_pct, term_months)
    for n in range(1, term_months + 1):
        interest = (remaining * i).quantize(CENT, ROUND_HALF_UP)
        principal_part = rate - interest
        if n == term_months or principal_part > remaining:
            principal_part = remaining
        remaining = remaining - principal_part
        plan.append(Installment(add_months(payout_date, n),
                                -(interest + principal_part), interest,
                                principal_part, remaining.quantize(CENT)))
    return plan

• Step 4: Ausführen — erwartet PASS

• Step 5: Commit — feat: Kredit-Mathematik (Annuitaet, endfaellig)

Task 4: Wiederholungs-Termine (Recurrence)

Files:

• Create: finance/app/engine/recurrence.py, finance/tests/test_recurrence.py

Interfaces:

• Produces: occurrences(rhythm: str, due_day: int, window_start: date, window_end: date, item_start: date | None = None, item_end: date | None = None) -> list[date].

• Step 1: Failing Tests

finance/tests/test_recurrence.py:

from datetime import date

from app.engine.recurrence import occurrences


def test_monthly_clamps_short_months():
    got = occurrences("monthly", 31, date(2026, 1, 1), date(2026, 3, 31))
    assert got == [date(2026, 1, 31), date(2026, 2, 28), date(2026, 3, 31)]


def test_quarterly_respects_item_bounds():
    got = occurrences("quarterly", 15, date(2026, 1, 1), date(2026, 12, 31),
                      item_start=date(2026, 4, 1), item_end=date(2026, 10, 31))
    assert got == [date(2026, 4, 15), date(2026, 7, 15), date(2026, 10, 15)]


def test_yearly():
    got = occurrences("yearly", 1, date(2026, 1, 1), date(2028, 12, 31))
    assert got == [date(2026, 1, 1), date(2027, 1, 1), date(2028, 1, 1)]

• Step 2: Ausführen — FAIL

• Step 3: Implementieren

finance/app/engine/recurrence.py:

import calendar
from datetime import date

STEP = {"monthly": 1, "quarterly": 3, "yearly": 12}


def _clamp(y: int, m: int, day: int) -> date:
    return date(y, m, min(day, calendar.monthrange(y, m)[1]))


def occurrences(rhythm: str, due_day: int, window_start: date, window_end: date,
                item_start: date | None = None, item_end: date | None = None) -> list[date]:
    step = STEP[rhythm]
    lo = max(window_start, item_start) if item_start else window_start
    hi = min(window_end, item_end) if item_end else window_end
    anchor = item_start or window_start
    d = _clamp(anchor.year, anchor.month, due_day)
    out: list[date] = []
    while d <= hi:
        if d >= lo:
            out.append(d)
        total = d.year * 12 + d.month - 1 + step
        y, m0 = divmod(total, 12)
        d = _clamp(y, m0 + 1, due_day)
    return out

• Step 4: Ausführen — PASS · Step 5: Commit — feat: Recurrence-Terminberechnung

Task 5: Projektion, Modifikatoren, Szenario-Zahlungsstrom

Files:

• Create: finance/app/engine/projection.py, finance/app/engine/scenario.py, finance/tests/test_projection.py, finance/tests/test_scenario_engine.py

Interfaces:

• Produces:

  • projection.project(start_balance: Decimal, start_date: date, cashflows: list[tuple[date, Decimal]], horizon_days: int, threshold: Decimal = Decimal("0")) -> Projection mit Projection(series: list[tuple[date, Decimal]], low_point: tuple[date, Decimal], first_below_zero: date | None, first_below_threshold: date | None).
  • scenario.PlainRecurring(name, amount, rhythm, due_day, start_date, end_date, category_id, id) / PlainPlanned(name, amount, due, category_id) / PlainModifier(target_type, target_id, kind, value) (Dataclasses).
  • scenario.build_cashflows(recurring: list[PlainRecurring], planned: list[PlainPlanned], loan_schedules: list[list[Installment]], loan_payouts: list[tuple[date, Decimal]], modifiers: list[PlainModifier], window_start: date, window_end: date) -> list[tuple[date, Decimal]].

• Step 1: Failing Tests

finance/tests/test_projection.py:

from datetime import date
from decimal import Decimal

from app.engine.projection import project


def test_low_point_and_threshold():
    flows = [(date(2026, 8, 1), Decimal("-150")), (date(2026, 8, 10), Decimal("200"))]
    p = project(Decimal("100"), date(2026, 7, 31), flows, horizon_days=15,
                threshold=Decimal("60"))
    assert p.low_point == (date(2026, 8, 1), Decimal("-50"))
    assert p.first_below_zero == date(2026, 8, 1)
    assert p.first_below_threshold == date(2026, 8, 1)
    assert p.series[-1][1] == Decimal("150")


def test_no_crossing():
    p = project(Decimal("100"), date(2026, 7, 31), [], horizon_days=5)
    assert p.first_below_zero is None
    assert p.low_point[1] == Decimal("100")

finance/tests/test_scenario_engine.py:

from datetime import date
from decimal import Decimal

from app.engine.loans import loan_schedule
from app.engine.scenario import (PlainModifier, PlainPlanned, PlainRecurring,
                                 build_cashflows)

W = (date(2026, 8, 1), date(2026, 10, 31))


def test_recurring_with_percent_cut():
    rec = [PlainRecurring(id=1, name="Marketing", amount=Decimal("-1000"),
                          rhythm="monthly", due_day=5, start_date=None,
                          end_date=None, category_id=7)]
    mods = [PlainModifier(target_type="category", target_id=7,
                          kind="percent", value=Decimal("50"))]
    flows = build_cashflows(rec, [], [], [], mods, *W)
    assert flows == [(date(2026, 8, 5), Decimal("-500.00")),
                     (date(2026, 9, 5), Decimal("-500.00")),
                     (date(2026, 10, 5), Decimal("-500.00"))]


def test_remove_modifier_and_planned_and_loan():
    rec = [PlainRecurring(id=2, name="Abo", amount=Decimal("-50"),
                          rhythm="monthly", due_day=1, start_date=None,
                          end_date=None, category_id=None)]
    mods = [PlainModifier(target_type="recurring", target_id=2,
                          kind="remove", value=Decimal("0"))]
    planned = [PlainPlanned(name="Steuer", amount=Decimal("-2000"),
                            due=date(2026, 9, 15), category_id=None)]
    sched = loan_schedule(Decimal("10000"), Decimal("6"), 48, date(2026, 8, 1), "annuity")
    flows = build_cashflows(rec, planned, [sched],
                            [(date(2026, 8, 1), Decimal("10000"))], mods, *W)
    days = [d for d, _ in flows]
    assert date(2026, 8, 1) in days            # Kredit-Auszahlung
    assert date(2026, 9, 15) in days           # Einmalposten
    assert all(a != Decimal("-50") for _, a in flows)  # Abo entfernt

• Step 2: Ausführen — FAIL

• Step 3: Implementieren

finance/app/engine/projection.py:

from collections import defaultdict
from dataclasses import dataclass
from datetime import date, timedelta
from decimal import Decimal


@dataclass(frozen=True)
class Projection:
    series: list[tuple[date, Decimal]]
    low_point: tuple[date, Decimal]
    first_below_zero: date | None
    first_below_threshold: date | None


def project(start_balance: Decimal, start_date: date,
            cashflows: list[tuple[date, Decimal]], horizon_days: int,
            threshold: Decimal = Decimal("0")) -> Projection:
    end = start_date + timedelta(days=horizon_days)
    by_day: dict[date, Decimal] = defaultdict(lambda: Decimal("0"))
    for d, amount in cashflows:
        if start_date < d <= end:
            by_day[d] += amount
    series: list[tuple[date, Decimal]] = []
    balance = start_balance
    low = (start_date, start_balance)
    below0: date | None = None
    below_t: date | None = None
    d = start_date + timedelta(days=1)
    while d <= end:
        balance += by_day.get(d, Decimal("0"))
        series.append((d, balance))
        if balance < low[1]:
            low = (d, balance)
        if below0 is None and balance < 0:
            below0 = d
        if below_t is None and balance < threshold:
            below_t = d
        d += timedelta(days=1)
    return Projection(series, low, below0, below_t)

finance/app/engine/scenario.py:

from dataclasses import dataclass
from datetime import date
from decimal import ROUND_HALF_UP, Decimal

from app.engine.loans import Installment
from app.engine.recurrence import occurrences

CENT = Decimal("0.01")


@dataclass(frozen=True)
class PlainRecurring:
    id: int
    name: str
    amount: Decimal
    rhythm: str
    due_day: int
    start_date: date | None
    end_date: date | None
    category_id: int | None


@dataclass(frozen=True)
class PlainPlanned:
    name: str
    amount: Decimal
    due: date
    category_id: int | None


@dataclass(frozen=True)
class PlainModifier:
    target_type: str
    target_id: int
    kind: str
    value: Decimal


def _modified(amount: Decimal, category_id: int | None, recurring_id: int | None,
              modifiers: list[PlainModifier]) -> Decimal | None:
    for m in modifiers:
        hit = ((m.target_type == "category" and category_id == m.target_id)
               or (m.target_type == "recurring" and recurring_id == m.target_id))
        if not hit:
            continue
        if m.kind == "remove":
            return None
        if m.kind == "percent":
            amount = (amount * (Decimal(100) - m.value) / Decimal(100)).quantize(CENT, ROUND_HALF_UP)
        elif m.kind == "absolute":
            if amount < 0:
                amount = min(Decimal("0"), amount + m.value)
            else:
                amount = max(Decimal("0"), amount - m.value)
    return amount


def build_cashflows(recurring: list[PlainRecurring], planned: list[PlainPlanned],
                    loan_schedules: list[list[Installment]],
                    loan_payouts: list[tuple[date, Decimal]],
                    modifiers: list[PlainModifier],
                    window_start: date, window_end: date) -> list[tuple[date, Decimal]]:
    flows: list[tuple[date, Decimal]] = []
    for r in recurring:
        amount = _modified(r.amount, r.category_id, r.id, modifiers)
        if amount is None:
            continue
        for d in occurrences(r.rhythm, r.due_day, window_start, window_end,
                             r.start_date, r.end_date):
            flows.append((d, amount))
    for p in planned:
        amount = _modified(p.amount, p.category_id, None, modifiers)
        if amount is not None and window_start <= p.due <= window_end:
            flows.append((p.due, amount))
    for d, payout in loan_payouts:
        if window_start <= d <= window_end:
            flows.append((d, payout))
    for sched in loan_schedules:
        for inst in sched:
            if window_start <= inst.due <= window_end:
                flows.append((inst.due, inst.payment))
    return sorted(flows)

• Step 4: Ausführen — PASS · Step 5: Commit — feat: Szenario-Engine (Projektion, Modifikatoren, Zahlungsstrom)

---

Phase 4: PDF-Parser

Task 6: Parser-Basis, Bank-Erkennung, Validierung

Files:

• Create: finance/app/parsers/__init__.py, finance/app/parsers/base.py, finance/app/parsers/detect.py, finance/app/parsers/validate.py, finance/tests/test_parser_base.py

Interfaces:

• Produces:

  • base.ParsedTransaction(booking_date: date, value_date: date | None, amount: Decimal, purpose: str, counterparty: str); base.ParsedStatement(bank: str, iban: str | None, period_start, period_end, opening_balance, closing_balance, transactions: list[ParsedTransaction]); base.ParserError(Exception); base.parse_german_amount(s: str) -> Decimal; base.parse_german_date(s: str, default_year: int | None = None) -> date.
  • detect.detect_bank(text: str) -> str | None mit Rückgabe "vr" | "hvb" | "dkb" | None.
  • validate.balance_difference(p: ParsedStatement) -> Decimal (0 = plausibel); validate.dedup_hash(account_id: int, booking_date: date, amount: Decimal, purpose: str) -> str.

• Step 1: Failing Tests

finance/tests/test_parser_base.py:

from datetime import date
from decimal import Decimal

from app.parsers.base import (ParsedStatement, ParsedTransaction,
                              parse_german_amount, parse_german_date)
from app.parsers.detect import detect_bank
from app.parsers.validate import balance_difference, dedup_hash


def test_parse_german_amount():
    assert parse_german_amount("1.234,56") == Decimal("1234.56")
    assert parse_german_amount("1.234,56-") == Decimal("-1234.56")
    assert parse_german_amount("12,00 S") == Decimal("-12.00")
    assert parse_german_amount("12,00 H") == Decimal("12.00")


def test_parse_german_date():
    assert parse_german_date("03.06.2026") == date(2026, 6, 3)
    assert parse_german_date("03.06.", default_year=2026) == date(2026, 6, 3)


def test_detect_bank():
    assert detect_bank("... Volksbank Musterstadt eG ...") == "vr"
    assert detect_bank("... Raiffeisenbank ...") == "vr"
    assert detect_bank("... HypoVereinsbank ...") == "hvb"
    assert detect_bank("... UniCredit Bank ...") == "hvb"
    assert detect_bank("... Deutsche Kreditbank AG ...") == "dkb"
    assert detect_bank("... DKB ...") == "dkb"
    assert detect_bank("irgendwas") is None


def _stmt(txs, opening, closing):
    return ParsedStatement(bank="vr", iban=None, period_start=None,
                           period_end=None, opening_balance=opening,
                           closing_balance=closing, transactions=txs)


def test_balance_difference():
    txs = [ParsedTransaction(date(2026, 6, 1), None, Decimal("-50"), "Miete", "V"),
           ParsedTransaction(date(2026, 6, 2), None, Decimal("20"), "Gutschrift", "K")]
    assert balance_difference(_stmt(txs, Decimal("100"), Decimal("70"))) == Decimal("0")
    assert balance_difference(_stmt(txs, Decimal("100"), Decimal("75"))) == Decimal("5")


def test_dedup_hash_normalizes_whitespace():
    a = dedup_hash(1, date(2026, 6, 1), Decimal("-50.00"), "Miete  Juni")
    b = dedup_hash(1, date(2026, 6, 1), Decimal("-50.00"), "Miete Juni")
    assert a == b and len(a) == 64

• Step 2: Ausführen — FAIL

• Step 3: Implementieren

finance/app/parsers/base.py:

import re
from dataclasses import dataclass, field
from datetime import date
from decimal import Decimal


class ParserError(Exception):
    pass


@dataclass(frozen=True)
class ParsedTransaction:
    booking_date: date
    value_date: date | None
    amount: Decimal
    purpose: str
    counterparty: str


@dataclass
class ParsedStatement:
    bank: str
    iban: str | None
    period_start: date | None
    period_end: date | None
    opening_balance: Decimal | None
    closing_balance: Decimal | None
    transactions: list[ParsedTransaction] = field(default_factory=list)


def parse_german_amount(s: str) -> Decimal:
    s = s.strip().replace(" ", "").replace(" ", "")
    neg = s.endswith("-") or s.startswith("-") or s.endswith("S")
    s = s.strip("+-SH")
    value = Decimal(s.replace(".", "").replace(",", "."))
    return -value if neg else value


def parse_german_date(s: str, default_year: int | None = None) -> date:
    m = re.fullmatch(r"(\d{2})\.(\d{2})\.(\d{4})?", s.strip())
    if not m:
        raise ParserError(f"Kein Datum: {s!r}")
    year = int(m.group(3)) if m.group(3) else default_year
    if year is None:
        raise ParserError(f"Jahr fehlt: {s!r}")
    return date(year, int(m.group(2)), int(m.group(1)))

finance/app/parsers/detect.py:

MARKERS = [
    ("vr", ["Volksbank", "Raiffeisenbank", "VR-Bank", "VR Bank"]),
    ("hvb", ["HypoVereinsbank", "UniCredit"]),
    ("dkb", ["Deutsche Kreditbank", "DKB"]),
]


def detect_bank(text: str) -> str | None:
    for bank, needles in MARKERS:
        if any(n in text for n in needles):
            return bank
    return None

finance/app/parsers/validate.py:

import hashlib
from datetime import date
from decimal import Decimal

from app.parsers.base import ParsedStatement


def balance_difference(p: ParsedStatement) -> Decimal:
    total = sum((t.amount for t in p.transactions), Decimal("0"))
    return (p.opening_balance + total - p.closing_balance).copy_abs()


def dedup_hash(account_id: int, booking_date: date, amount: Decimal, purpose: str) -> str:
    raw = f"{account_id}|{booking_date.isoformat()}|{amount:.2f}|{' '.join(purpose.split())}"
    return hashlib.sha256(raw.encode()).hexdigest()

• Step 4: Ausführen — PASS · Step 5: Commit — feat: Parser-Basis, Bank-Erkennung, Saldo-/Duplikat-Pruefung

Task 7: Bank-Parser VR, HVB, DKB (fixture-getrieben)

Files:

• Create: finance/app/parsers/vr.py, finance/app/parsers/hvb.py, finance/app/parsers/dkb.py, finance/app/parsers/registry.py, finance/tests/test_bank_parsers.py, finance/tests/fixtures/README.md

Interfaces:

• Produces: je Modul parse(path: Path) -> ParsedStatement; registry.PARSERS: dict[str, Callable[[Path], ParsedStatement]] mit Schlüsseln "vr", "hvb", "dkb"; registry.parse_pdf(path: Path) -> ParsedStatement (extrahiert Text, erkennt Bank via detect_bank, ruft Parser; ParserError bei unbekannter Bank).

Wichtig: Die echten PDF-Layouts liegen erst mit den Fixtures vor (Nutzer legt pro Bank mindestens ein echtes PDF unter finance/tests/fixtures/ ab: vr_beispiel.pdf, hvb_beispiel.pdf, dkb_beispiel.pdf — gitignored). Die Fixture-Tests skippen, solange die Datei fehlt. Die Regex-Konstanten am Dateianfang jedes Parsers sind der einzige anzupassende Teil; Arbeitsschleife pro Bank: Test laufen lassen → bei Fehlschlag mit dem Debug-Skript den Text ansehen → Regex anpassen → wiederholen, bis die Saldo-Prüfung aufgeht.

• Step 1: Fixture-README und Tests schreiben

finance/tests/fixtures/README.md:

# Beispiel-Auszüge (nicht committen!)

Hier pro Bank mindestens einen echten PDF-Kontoauszug ablegen:
- `vr_beispiel.pdf`  (Volksbank/Raiffeisenbank)
- `hvb_beispiel.pdf` (HypoVereinsbank)
- `dkb_beispiel.pdf` (DKB)

Beträge dürfen verfremdet sein, das Layout muss echt sein.
PDFs sind via .gitignore vom Repo ausgeschlossen.

finance/tests/test_bank_parsers.py:

from decimal import Decimal
from pathlib import Path

import pytest

from app.parsers.registry import parse_pdf
from app.parsers.validate import balance_difference

FIXTURES = Path(__file__).parent / "fixtures"
CASES = [("vr_beispiel.pdf", "vr"), ("hvb_beispiel.pdf", "hvb"),
         ("dkb_beispiel.pdf", "dkb")]


@pytest.mark.parametrize("filename,bank", CASES)
def test_parse_fixture(filename, bank):
    path = FIXTURES / filename
    if not path.exists():
        pytest.skip(f"Fixture {filename} fehlt")
    stmt = parse_pdf(path)
    assert stmt.bank == bank
    assert stmt.opening_balance is not None
    assert stmt.closing_balance is not None
    assert len(stmt.transactions) > 0
    assert balance_difference(stmt) == Decimal("0")

• Step 2: Parser-Gerüst implementieren (gemeinsames Muster)

finance/app/parsers/vr.py (HVB/DKB analog — gleiche Struktur, eigene Konstanten; Dateien hvb.py/dkb.py identisch aufbauen, nur BANK = "hvb" bzw. "dkb"):

"""Parser für Volksbank/Raiffeisenbank-Kontoauszüge.

Die REGEX-Konstanten unten sind das Einzige, was beim Abgleich mit den
echten Fixture-PDFs angepasst werden muss.
"""
import re
from datetime import date
from pathlib import Path

import pdfplumber

from app.parsers.base import (ParsedStatement, ParsedTransaction, ParserError,
                              parse_german_amount, parse_german_date)

BANK = "vr"
RE_IBAN = re.compile(r"\b([A-Z]{2}\d{2}(?:\s?\d{4}){4,5}(?:\s?\d{1,2})?)\b")
# "Kontoauszug 6/2026" oder "vom 01.06.2026 bis 30.06.2026"
RE_PERIOD = re.compile(r"vom\s+(\d{2}\.\d{2}\.\d{4})\s+bis\s+(\d{2}\.\d{2}\.\d{4})")
# "alter Kontostand 1.234,56 H" / "neuer Kontostand 987,65 H"
RE_OPENING = re.compile(r"alter\s+Kontostand\s+([\d.,]+\s*[SH-]?)", re.I)
RE_CLOSING = re.compile(r"neuer\s+Kontostand\s+([\d.,]+\s*[SH-]?)", re.I)
# Buchungszeile: "01.06. 01.06. <Text> 1.234,56 S"
RE_TX = re.compile(r"^(\d{2}\.\d{2}\.)\s+(\d{2}\.\d{2}\.)\s+(.*?)\s+([\d.,]+\s*[SH-])$")


def parse(path: Path) -> ParsedStatement:
    with pdfplumber.open(path) as pdf:
        text = "\n".join((page.extract_text() or "") for page in pdf.pages)
    period = RE_PERIOD.search(text)
    if not period:
        raise ParserError("VR: Abrechnungszeitraum nicht gefunden")
    period_start = parse_german_date(period.group(1))
    period_end = parse_german_date(period.group(2))
    opening = RE_OPENING.search(text)
    closing = RE_CLOSING.search(text)
    if not opening or not closing:
        raise ParserError("VR: Anfangs-/Endsaldo nicht gefunden")
    iban_m = RE_IBAN.search(text)
    txs: list[ParsedTransaction] = []
    pending: dict | None = None
    for line in text.splitlines():
        m = RE_TX.match(line.strip())
        if m:
            if pending:
                txs.append(_finish(pending))
            pending = {
                "booking": parse_german_date(m.group(1), period_end.year),
                "value": parse_german_date(m.group(2), period_end.year),
                "head": m.group(3).strip(),
                "amount": parse_german_amount(m.group(4)),
                "extra": [],
            }
        elif pending and line.strip() and not RE_CLOSING.search(line):
            pending["extra"].append(line.strip())
    if pending:
        txs.append(_finish(pending))
    return ParsedStatement(bank=BANK,
                           iban=iban_m.group(1).replace(" ", "") if iban_m else None,
                           period_start=period_start, period_end=period_end,
                           opening_balance=parse_german_amount(opening.group(1)),
                           closing_balance=parse_german_amount(closing.group(1)),
                           transactions=txs)


def _finish(p: dict) -> ParsedTransaction:
    counterparty = p["extra"][0] if p["extra"] else ""
    purpose = " ".join([p["head"], *p["extra"][1:]]).strip()
    return ParsedTransaction(booking_date=p["booking"], value_date=p["value"],
                             amount=p["amount"], purpose=purpose,
                             counterparty=counterparty)

finance/app/parsers/registry.py:

from pathlib import Path
from typing import Callable

import pdfplumber

from app.parsers import dkb, hvb, vr
from app.parsers.base import ParsedStatement, ParserError
from app.parsers.detect import detect_bank

PARSERS: dict[str, Callable[[Path], ParsedStatement]] = {
    "vr": vr.parse, "hvb": hvb.parse, "dkb": dkb.parse,
}


def parse_pdf(path: Path) -> ParsedStatement:
    with pdfplumber.open(path) as pdf:
        first = pdf.pages[0].extract_text() or ""
    bank = detect_bank(first)
    if bank is None:
        raise ParserError("Bank nicht erkannt")
    return PARSERS[bank](path)

• Step 3: Debug-Hilfsskript anlegen — finance/scripts/dump_pdf_text.py:

"""PDF-Text anzeigen, um Parser-Regexe abzugleichen: python scripts/dump_pdf_text.py <pdf>"""
import sys

import pdfplumber

with pdfplumber.open(sys.argv[1]) as pdf:
    for i, page in enumerate(pdf.pages, 1):
        print(f"--- Seite {i} ---")
        print(page.extract_text() or "(kein Text)")

• Step 4: Fixture-Abgleich je Bank — Für jede vorhandene Fixture: .venv/bin/python -m pytest tests/test_bank_parsers.py -v ausführen; bei FAIL mit dump_pdf_text.py den echten Text ansehen und nur die RE_*-Konstanten (und falls nötig die Salden-Begriffe, z. B. „Kontostand am…") des jeweiligen Parsers anpassen, bis der Test inkl. balance_difference == 0 grün ist. Fehlt eine Fixture noch, bleibt der Test geskippt — dann diesen Zustand committen und den Abgleich nachholen, sobald die Datei da ist.

• Step 5: Alle Tests ausführen — PASS/SKIP · Step 6: Commit — feat: PDF-Parser VR/HVB/DKB mit Registry und Fixture-Tests

---

Phase 5: API

Task 8: FastAPI-App, Auth (API-Key + Session-Login)

Files:

• Create: finance/app/main.py, finance/app/auth.py, finance/tests/test_auth.py, finance/tests/conftest.py (erweitern)

Interfaces:

• Produces: app.main.app (FastAPI); auth.hash_password(pw: str) -> str (Format salt$hex, PBKDF2-SHA256); auth.verify_password(pw: str, stored: str) -> bool; Dependency auth.require_auth (akzeptiert Authorization: Bearer <FB_API_KEY> oder gültiges Session-Cookie fb_session, sonst 401); Routen POST /login (Form: username, password; setzt Cookie, Redirect /), GET /login (Formular), POST /logout.

• Consumes: get_settings, get_session.

• Step 1: conftest erweitern + Failing Tests

In finance/tests/conftest.py ergänzen:

from fastapi.testclient import TestClient

from app.db import get_session


@pytest.fixture
def client(db, monkeypatch):
    monkeypatch.setenv("FB_API_KEY", "test-key")
    from app.config import get_settings
    get_settings.cache_clear()
    from app.auth import hash_password
    monkeypatch.setenv("FB_GUI_PASSWORD_HASH", hash_password("geheim"))
    get_settings.cache_clear()
    from app.main import app
    app.dependency_overrides[get_session] = lambda: iter([db])
    with TestClient(app) as c:
        yield c
    app.dependency_overrides.clear()
    get_settings.cache_clear()

finance/tests/test_auth.py:

def test_api_requires_key(client):
    assert client.get("/api/accounts").status_code == 401
    r = client.get("/api/accounts", headers={"Authorization": "Bearer test-key"})
    assert r.status_code == 200


def test_login_flow(client):
    r = client.post("/login", data={"username": "admin", "password": "falsch"},
                    follow_redirects=False)
    assert r.status_code == 401
    r = client.post("/login", data={"username": "admin", "password": "geheim"},
                    follow_redirects=False)
    assert r.status_code == 303
    assert client.get("/api/accounts").status_code == 200  # Cookie reicht

• Step 2: Ausführen — FAIL

• Step 3: Implementieren

finance/app/auth.py:

import hashlib
import hmac
import secrets

from fastapi import HTTPException, Request
from itsdangerous import BadSignature, TimestampSigner

from app.config import get_settings

COOKIE = "fb_session"
MAX_AGE = 60 * 60 * 12  # 12 h


def hash_password(pw: str, salt: str | None = None) -> str:
    salt = salt or secrets.token_hex(16)
    digest = hashlib.pbkdf2_hmac("sha256", pw.encode(), salt.encode(), 200_000).hex()
    return f"{salt}${digest}"


def verify_password(pw: str, stored: str) -> bool:
    try:
        salt, digest = stored.split("$", 1)
    except ValueError:
        return False
    return hmac.compare_digest(hash_password(pw, salt).split("$", 1)[1], digest)


def _signer() -> TimestampSigner:
    return TimestampSigner(get_settings().session_secret)


def make_session_token() -> str:
    return _signer().sign(b"gui").decode()


def session_valid(token: str | None) -> bool:
    if not token:
        return False
    try:
        _signer().unsign(token, max_age=MAX_AGE)
        return True
    except BadSignature:
        return False


def require_auth(request: Request) -> None:
    settings = get_settings()
    header = request.headers.get("authorization", "")
    if settings.api_key and header == f"Bearer {settings.api_key}":
        return
    if session_valid(request.cookies.get(COOKIE)):
        return
    raise HTTPException(status_code=401, detail="Nicht angemeldet")

finance/app/main.py:

import hmac

from fastapi import Depends, FastAPI, Form, Request
from fastapi.responses import HTMLResponse, RedirectResponse

from app import auth
from app.config import get_settings

app = FastAPI(title="Finanzberatungs-Tool")


@app.get("/login", response_class=HTMLResponse)
def login_form():
    return """<form method=post action=/login>
      <input name=username placeholder=Benutzer>
      <input name=password type=password placeholder=Passwort>
      <button>Anmelden</button></form>"""


@app.post("/login")
def login(username: str = Form(...), password: str = Form(...)):
    s = get_settings()
    if not (hmac.compare_digest(username, s.gui_user)
            and auth.verify_password(password, s.gui_password_hash)):
        return HTMLResponse("Login fehlgeschlagen", status_code=401)
    resp = RedirectResponse("/", status_code=303)
    resp.set_cookie(auth.COOKIE, auth.make_session_token(), httponly=True,
                    max_age=auth.MAX_AGE, samesite="lax")
    return resp


@app.post("/logout")
def logout():
    resp = RedirectResponse("/login", status_code=303)
    resp.delete_cookie(auth.COOKIE)
    return resp


@app.get("/api/accounts", dependencies=[Depends(auth.require_auth)])
def _placeholder_accounts():  # wird in Task 9 durch Router ersetzt
    return []

• Step 4: Ausführen — PASS · Step 5: Commit — feat: FastAPI-App mit API-Key- und Session-Auth

Task 9: CRUD — Konten, Buchungen, Kategorien, Regeln

Files:

• Create: finance/app/routers/__init__.py, finance/app/routers/accounts.py, finance/app/routers/transactions.py, finance/app/routers/categories.py, finance/app/services/__init__.py, finance/app/services/categorize.py, finance/app/services/balances.py, finance/tests/test_crud_api.py
• Modify: finance/app/main.py (Platzhalter-Route entfernen, Router einbinden)

Interfaces:

• Produces (alle unter Depends(require_auth)):

  • GET/POST /api/accounts, GET /api/accounts/{id} — Felder wie ORM; Antwort enthält zusätzlich balance (aus services.balances.account_balance).
  • GET /api/transactions (Filter: account_id, date_from, date_to, category_id, q, status Default confirmed, limit Default 200, offset), POST /api/transactions (manuelle Buchung; Status confirmed, dedup_hash via validate.dedup_hash; bei vorhandenem Hash 409 außer force=true), PATCH /api/transactions/{id} (nur category_id).
  • GET/POST /api/categories, GET/POST /api/category-rules, DELETE /api/category-rules/{id}.
  • services.categorize.apply_rules(session, transactions) -> int (setzt category_id per Regel, Rückgabe: Anzahl kategorisiert; Regel-Match: rule.pattern.lower() in (purpose + " " + counterparty).lower(), Regeln nach priority aufsteigend, erster Treffer gewinnt).
  • services.balances.account_balance(session, account) -> Decimal (Endsaldo des neuesten bestätigten Statements + Summe bestätigter Buchungen mit booking_date > period_end; ohne Statement: Summe aller bestätigten Buchungen). services.balances.total_balance(session) -> Decimal.

• Step 1: Failing Tests — finance/tests/test_crud_api.py:

from datetime import date
from decimal import Decimal

from app.models.tables import Account, Statement, Transaction
from app.services.balances import account_balance

H = {"Authorization": "Bearer test-key"}


def test_account_crud_and_balance(client, db):
    r = client.post("/api/accounts", headers=H, json={
        "bank": "DKB", "iban": "DE02120300000000202051",
        "name": "Giro", "type": "giro"})
    assert r.status_code == 201
    acc_id = r.json()["id"]
    db.add(Statement(filename="a.pdf", bank="dkb", account_id=acc_id,
                     period_start=date(2026, 6, 1), period_end=date(2026, 6, 30),
                     opening_balance=Decimal("0"), closing_balance=Decimal("100.00"),
                     status="confirmed"))
    db.add(Transaction(account_id=acc_id, booking_date=date(2026, 7, 2),
                       amount=Decimal("-30.00"), purpose="Bar", status="confirmed",
                       dedup_hash="x1"))
    db.commit()
    acc = db.get(Account, acc_id)
    assert account_balance(db, acc) == Decimal("70.00")


def test_manual_transaction_dedup(client):
    acc = client.post("/api/accounts", headers=H, json={
        "bank": "DKB", "iban": "DE99", "name": "G", "type": "giro"}).json()
    payload = {"account_id": acc["id"], "booking_date": "2026-07-01",
               "amount": "-10.00", "purpose": "Kaffee", "counterparty": ""}
    assert client.post("/api/transactions", headers=H, json=payload).status_code == 201
    assert client.post("/api/transactions", headers=H, json=payload).status_code == 409
    payload["force"] = True
    assert client.post("/api/transactions", headers=H, json=payload).status_code == 201


def test_rules_categorize(client):
    cat = client.post("/api/categories", headers=H, json={"name": "Energie"}).json()
    client.post("/api/category-rules", headers=H,
                json={"pattern": "stadtwerke", "category_id": cat["id"]})
    acc = client.post("/api/accounts", headers=H, json={
        "bank": "DKB", "iban": "DE98", "name": "G2", "type": "giro"}).json()
    client.post("/api/transactions", headers=H, json={
        "account_id": acc["id"], "booking_date": "2026-07-03",
        "amount": "-80.00", "purpose": "STADTWERKE Abschlag", "counterparty": ""})
    txs = client.get("/api/transactions", headers=H,
                     params={"account_id": acc["id"]}).json()
    assert txs[0]["category_id"] == cat["id"]

• Step 2: Ausführen — FAIL

• Step 3: Implementieren — Router mit Pydantic-Schemas (BaseModel je Endpoint, model_config = ConfigDict(from_attributes=True)), Services:

finance/app/services/categorize.py:

from sqlalchemy import select
from sqlalchemy.orm import Session

from app.models.tables import CategoryRule, Transaction


def apply_rules(session: Session, transactions: list[Transaction]) -> int:
    rules = session.execute(
        select(CategoryRule).order_by(CategoryRule.priority)).scalars().all()
    hits = 0
    for tx in transactions:
        if tx.category_id is not None:
            continue
        haystack = f"{tx.purpose} {tx.counterparty}".lower()
        for rule in rules:
            if rule.pattern.lower() in haystack:
                tx.category_id = rule.category_id
                hits += 1
                break
    return hits

finance/app/services/balances.py:

from decimal import Decimal

from sqlalchemy import func, select
from sqlalchemy.orm import Session

from app.models.tables import Account, Statement, Transaction


def account_balance(session: Session, account: Account) -> Decimal:
    stmt = session.execute(
        select(Statement).where(Statement.account_id == account.id,
                                Statement.status == "confirmed")
        .order_by(Statement.period_end.desc())).scalars().first()
    q = select(func.coalesce(func.sum(Transaction.amount), 0)).where(
        Transaction.account_id == account.id, Transaction.status == "confirmed")
    if stmt is not None:
        q = q.where(Transaction.booking_date > stmt.period_end)
        base = stmt.closing_balance
    else:
        base = Decimal("0")
    return Decimal(base) + Decimal(session.execute(q).scalar_one())


def total_balance(session: Session) -> Decimal:
    accounts = session.execute(select(Account)).scalars().all()
    return sum((account_balance(session, a) for a in accounts), Decimal("0"))

Router-Muster (finance/app/routers/accounts.py, transactions/categories analog — vollständige CRUD-Logik gemäß Interface-Block, manuelle Buchung ruft validate.dedup_hash + apply_rules):

from decimal import Decimal

from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel, ConfigDict
from sqlalchemy import select
from sqlalchemy.orm import Session

from app.auth import require_auth
from app.db import get_session
from app.models.tables import Account
from app.services.balances import account_balance

router = APIRouter(prefix="/api/accounts", tags=["accounts"],
                   dependencies=[Depends(require_auth)])


class AccountIn(BaseModel):
    bank: str
    iban: str
    name: str
    type: str = "giro"


class AccountOut(AccountIn):
    model_config = ConfigDict(from_attributes=True)
    id: int
    balance: Decimal = Decimal("0")


@router.get("", response_model=list[AccountOut])
def list_accounts(session: Session = Depends(get_session)):
    out = []
    for acc in session.execute(select(Account)).scalars():
        item = AccountOut.model_validate(acc)
        item.balance = account_balance(session, acc)
        out.append(item)
    return out


@router.post("", response_model=AccountOut, status_code=201)
def create_account(data: AccountIn, session: Session = Depends(get_session)):
    if session.execute(select(Account).where(Account.iban == data.iban)).scalar():
        raise HTTPException(409, "IBAN existiert bereits")
    acc = Account(**data.model_dump())
    session.add(acc)
    session.commit()
    session.refresh(acc)
    return AccountOut.model_validate(acc)

In main.py: Platzhalter-Route _placeholder_accounts löschen, stattdessen app.include_router(accounts.router) etc.

• Step 4: Ausführen — PASS · Step 5: Commit — feat: CRUD-API Konten/Buchungen/Kategorien/Regeln

Task 10: Import-API (Upload → Vorschau → Bestätigen)

Files:

• Create: finance/app/routers/imports.py, finance/app/services/importer.py, finance/tests/test_import_api.py

Interfaces:

• Produces:

  • services.importer.process_pdf(session, path: Path) -> Statement — Ablauf: registry.parse_pdf; Konto per IBAN suchen, sonst Account(bank=<bank>, iban=<iban oder "UNBEKANNT-"+filename>, name=iban, type="giro") anlegen; Statement anlegen; bei ParserError → status="error" + error_message; bei balance_difference != 0 → status="error", error_message=f"Saldo-Differenz {diff} EUR"; sonst Draft-Transaktionen anlegen (status="draft", dedup_hash, is_duplicate=True, wenn Hash schon als confirmed existiert), apply_rules auf die Drafts, status="draft". PDF wird nach uploads_dir verschoben (bei Fehler: bleibt in Inbox).
  • POST /api/imports/upload (multipart file) → speichert nach inbox_dir, ruft process_pdf, gibt Statement zurück (201).
  • POST /api/imports/scan-inbox → verarbeitet alle *.pdf in inbox_dir, Rückgabe Liste.
  • GET /api/imports → Statements mit Status; GET /api/imports/{id}/preview → {statement, transactions: [...], balance_ok: bool, duplicates: int}.
  • POST /api/imports/{id}/confirm → alle Drafts mit is_duplicate=False auf confirmed, Statement auf confirmed; DELETE /api/imports/{id} → Statement + zugehörige Drafts löschen.

• Consumes: registry.parse_pdf, validate.*, categorize.apply_rules.

• Step 1: Failing Tests — finance/tests/test_import_api.py: Da echte PDFs in Unit-Tests fehlen, wird registry.parse_pdf gemockt:

from datetime import date
from decimal import Decimal
from pathlib import Path

import pytest

from app.parsers.base import ParsedStatement, ParsedTransaction

H = {"Authorization": "Bearer test-key"}

FAKE = ParsedStatement(
    bank="dkb", iban="DE02120300000000202051",
    period_start=date(2026, 6, 1), period_end=date(2026, 6, 30),
    opening_balance=Decimal("100.00"), closing_balance=Decimal("40.00"),
    transactions=[ParsedTransaction(date(2026, 6, 3), date(2026, 6, 3),
                                    Decimal("-60.00"), "Miete Juni", "Vermieter")])


@pytest.fixture
def fake_parse(monkeypatch):
    monkeypatch.setattr("app.services.importer.parse_pdf", lambda p: FAKE)


def _upload(client, name="auszug.pdf"):
    return client.post("/api/imports/upload", headers=H,
                       files={"file": (name, b"%PDF-fake", "application/pdf")})


def test_upload_preview_confirm(client, fake_parse, tmp_path, monkeypatch):
    monkeypatch.setenv("FB_INBOX_DIR", str(tmp_path / "inbox"))
    monkeypatch.setenv("FB_UPLOADS_DIR", str(tmp_path / "uploads"))
    from app.config import get_settings
    get_settings.cache_clear()
    r = _upload(client)
    assert r.status_code == 201
    sid = r.json()["id"]
    prev = client.get(f"/api/imports/{sid}/preview", headers=H).json()
    assert prev["balance_ok"] is True
    assert len(prev["transactions"]) == 1
    assert client.post(f"/api/imports/{sid}/confirm", headers=H).status_code == 200
    txs = client.get("/api/transactions", headers=H).json()
    assert len(txs) == 1 and txs[0]["status"] == "confirmed"
    # Zweiter Import desselben Auszugs: alles Duplikate
    sid2 = _upload(client, "auszug2.pdf").json()["id"]
    prev2 = client.get(f"/api/imports/{sid2}/preview", headers=H).json()
    assert prev2["duplicates"] == 1
    client.post(f"/api/imports/{sid2}/confirm", headers=H)
    assert len(client.get("/api/transactions", headers=H).json()) == 1

• Step 2: Ausführen — FAIL · Step 3: Implementieren gemäß Interface-Block (Importer-Service + Router; parse_pdf in importer.py importieren als from app.parsers.registry import parse_pdf, damit der Test-Monkeypatch greift).

• Step 4: Ausführen — PASS · Step 5: Commit — feat: Import-API mit Vorschau, Saldo-Pruefung und Duplikat-Schutz

Task 11: Planungs- & Szenario-API

Files:

• Create: finance/app/routers/planning.py, finance/app/routers/scenarios.py, finance/app/services/projection_service.py, finance/app/services/suggestions.py, finance/tests/test_planning_api.py

Interfaces:

• Produces:

  • CRUD: GET/POST /api/recurring, PATCH/DELETE /api/recurring/{id}; analog /api/planned, /api/loans; GET /api/loans/{id}/schedule → Tilgungsplan (Installments als JSON).
  • GET /api/recurring/suggestions → services.suggestions.suggest_recurring(session) -> list[dict]: bestätigte Buchungen gruppiert nach (account_id, counterparty, amount); Gruppen mit ≥ 3 aufeinanderfolgenden Monaten ⇒ Vorschlag {name, amount, rhythm: "monthly", due_day (Median der Tage), category_id}; bereits als RecurringItem vorhandene (gleicher Name + Betrag) werden ausgelassen.
  • Szenarien: GET/POST /api/scenarios, PATCH/DELETE /api/scenarios/{id}, POST /api/scenarios/{id}/loans/{loan_id}, DELETE dito, POST /api/scenarios/{id}/modifiers + DELETE /api/scenarios/{id}/modifiers/{mod_id}.
  • POST /api/scenarios/{id}/project?horizon_days=548 → services.projection_service.run_projection(session, scenario, horizon_days, start_date) -> ProjectionResult: Startsaldo = total_balance; ORM → Plain-Dataclasses; build_cashflows + project (threshold = settings.warn_threshold); alte ProjectionPoints des Szenarios löschen, neue Serie + ProjectionResult (upsert) speichern; Response {low_point_date, low_point_balance, below_zero_date, below_threshold_date, series: [{day, balance}, ...]}. start_date Parameter default: heutiges Datum (date.today() nur hier, nicht in der Engine).

• Step 1: Failing Test — finance/tests/test_planning_api.py:

from datetime import date
from decimal import Decimal

from app.models.tables import Account, Statement

H = {"Authorization": "Bearer test-key"}


def _seed_balance(db, amount="1000.00"):
    acc = Account(bank="dkb", iban="DE01", name="G", type="giro")
    db.add(acc)
    db.flush()
    db.add(Statement(filename="s.pdf", bank="dkb", account_id=acc.id,
                     period_start=date(2026, 6, 1), period_end=date(2026, 6, 30),
                     opening_balance=Decimal("0"), closing_balance=Decimal(amount),
                     status="confirmed"))
    db.commit()


def test_scenario_projection(client, db):
    _seed_balance(db)
    client.post("/api/recurring", headers=H, json={
        "name": "Miete", "amount": "-600.00", "rhythm": "monthly", "due_day": 1})
    sc = client.post("/api/scenarios", headers=H, json={"name": "Basis"}).json()
    r = client.post(f"/api/scenarios/{sc['id']}/project", headers=H,
                    params={"horizon_days": 92, "start_date": "2026-07-15"})
    assert r.status_code == 200
    body = r.json()
    # 3 Mietzahlungen (01.08., 01.09., 01.10.): 400 -> -200 -> -800
    assert Decimal(body["low_point_balance"]) == Decimal("-800.00")
    assert body["below_zero_date"] == "2026-09-01"
    assert len(body["series"]) == 92


def test_loan_in_scenario_keeps_balance_positive(client, db):
    _seed_balance(db)
    client.post("/api/recurring", headers=H, json={
        "name": "Miete", "amount": "-600.00", "rhythm": "monthly", "due_day": 1})
    loan = client.post("/api/loans", headers=H, json={
        "name": "K1", "principal": "5000.00", "annual_rate_pct": "6.0",
        "term_months": 48, "payout_date": "2026-07-20",
        "repayment_type": "annuity"}).json()
    sc = client.post("/api/scenarios", headers=H, json={"name": "Kredit"}).json()
    client.post(f"/api/scenarios/{sc['id']}/loans/{loan['id']}", headers=H)
    body = client.post(f"/api/scenarios/{sc['id']}/project", headers=H,
                       params={"horizon_days": 92, "start_date": "2026-07-15"}).json()
    assert Decimal(body["low_point_balance"]) > Decimal("0")

• Step 2: Ausführen — FAIL · Step 3: Implementieren gemäß Interface-Block. Kern von projection_service.run_projection:

from datetime import date, timedelta
from datetime import datetime, timezone
from decimal import Decimal

from sqlalchemy import delete, select
from sqlalchemy.orm import Session

from app.config import get_settings
from app.engine.loans import loan_schedule
from app.engine.projection import project
from app.engine.scenario import (PlainModifier, PlainPlanned, PlainRecurring,
                                 build_cashflows)
from app.models.tables import (Loan, PlannedItem, ProjectionPoint,
                               ProjectionResult, RecurringItem, Scenario,
                               ScenarioLoan, ScenarioModifier)
from app.services.balances import total_balance


def run_projection(session: Session, scenario: Scenario, horizon_days: int,
                   start_date: date) -> ProjectionResult:
    end = start_date + timedelta(days=horizon_days)
    recurring = []
    if scenario.include_recurring:
        recurring = [PlainRecurring(r.id, r.name, Decimal(r.amount), r.rhythm,
                                    r.due_day, r.start_date, r.end_date, r.category_id)
                     for r in session.execute(select(RecurringItem)).scalars()]
    planned = []
    if scenario.include_planned:
        planned = [PlainPlanned(p.name, Decimal(p.amount), p.due, p.category_id)
                   for p in session.execute(select(PlannedItem)).scalars()]
    loans = session.execute(
        select(Loan).join(ScenarioLoan, ScenarioLoan.loan_id == Loan.id)
        .where(ScenarioLoan.scenario_id == scenario.id)).scalars().all()
    schedules = [loan_schedule(Decimal(l.principal), Decimal(l.annual_rate_pct),
                               l.term_months, l.payout_date, l.repayment_type)
                 for l in loans]
    payouts = [(l.payout_date, Decimal(l.principal)) for l in loans]
    modifiers = [PlainModifier(m.target_type, m.target_id, m.kind, Decimal(m.value))
                 for m in session.execute(select(ScenarioModifier).where(
                     ScenarioModifier.scenario_id == scenario.id)).scalars()]
    flows = build_cashflows(recurring, planned, schedules, payouts, modifiers,
                            start_date, end)
    proj = project(total_balance(session), start_date, flows, horizon_days,
                   threshold=get_settings().warn_threshold)
    session.execute(delete(ProjectionPoint).where(
        ProjectionPoint.scenario_id == scenario.id))
    for d, bal in proj.series:
        session.add(ProjectionPoint(scenario_id=scenario.id, day=d, balance=bal))
    result = session.get(ProjectionResult, scenario.id) or ProjectionResult(
        scenario_id=scenario.id, computed_at=datetime.now(timezone.utc),
        low_point_date=proj.low_point[0], low_point_balance=proj.low_point[1])
    result.computed_at = datetime.now(timezone.utc)
    result.low_point_date, result.low_point_balance = proj.low_point
    result.below_zero_date = proj.first_below_zero
    result.below_threshold_date = proj.first_below_threshold
    session.add(result)
    session.commit()
    return result

• Step 4: Ausführen — PASS · Step 5: Commit — feat: Planungs- und Szenario-API mit Projektion und Vorschlaegen

---

Phase 6: Web-GUI

Task 12: Basis-Layout, Übersicht, Buchungen

Files:

• Create: finance/app/routers/gui.py, finance/app/templates/base.html, finance/app/templates/index.html, finance/app/templates/transactions.html, finance/app/static/htmx.min.js, finance/app/static/style.css, finance/tests/test_gui.py
• Modify: finance/app/main.py (StaticFiles mounten, gui-Router einbinden, Jinja2Templates)

Interfaces:

• Produces: HTML-Routen GET / (Übersicht), GET /buchungen — beide leiten ohne gültige Session auf /login um (302), API-Key gilt hier nicht. GUI-Formulare posten auf die bestehenden /api/...-Endpunkte (Cookie-Auth greift dort).

• Login-Formular aus Task 8 wird durch Template login.html ersetzt (gleiche Routen).

• Step 1: htmx vendoren

mkdir -p /home/wlfb/bin/finance/app/static
curl -fsSL https://unpkg.com/htmx.org@2.0.6/dist/htmx.min.js \
  -o /home/wlfb/bin/finance/app/static/htmx.min.js

• Step 2: Failing Test — finance/tests/test_gui.py:

def test_pages_require_login(client):
    for path in ("/", "/buchungen", "/import", "/planung"):
        r = client.get(path, follow_redirects=False)
        assert r.status_code in (302, 303), path
        assert r.headers["location"] == "/login"


def test_pages_render_after_login(client):
    client.post("/login", data={"username": "admin", "password": "geheim"})
    for path in ("/", "/buchungen"):
        r = client.get(path)
        assert r.status_code == 200
        assert "Finanzberatung" in r.text

(/import und /planung kommen in Task 13 — Test deckt sie schon ab; bis dahin dürfen die Routen minimal existieren und nur das Template-Gerüst rendern.)

• Step 3: Implementieren — base.html mit Navigation (Übersicht / Import / Buchungen / Planung / Grafana-Link auf http://<host>:8097, Logout-Button), <script src="/static/htmx.min.js"></script>, style.css schlicht (max-width, Tabellenstil, rote Negativbeträge .neg { color: #b00 }). index.html: Gesamtsaldo + Tabelle Konto→Saldo (services.balances), Liste der nächsten 10 anstehenden Posten (occurrences der RecurringItems + PlannedItems der nächsten 30 Tage), Warnbox falls neueste ProjectionResult.below_threshold_date gesetzt, <iframe> auf Grafana-Dashboard. transactions.html: Filterformular (GET-Parameter an eigene Route, serverseitig gefiltert über dieselbe Query-Logik wie /api/transactions), Tabelle, Inline-<select> je Zeile für Kategorie (HTMX hx-patch auf /api/transactions/{id}), Formular „Buchung manuell erfassen" (HTMX-Post auf /api/transactions), Button „Regel daraus erzeugen" (Formular vorbefüllt mit counterparty als Pattern, Post auf /api/category-rules). GUI-Router-Muster:

from fastapi import APIRouter, Depends, Request
from fastapi.responses import RedirectResponse
from fastapi.templating import Jinja2Templates
from sqlalchemy.orm import Session

from app.auth import COOKIE, session_valid
from app.db import get_session

templates = Jinja2Templates(directory="app/templates")
router = APIRouter()


def gui_session(request: Request):
    if not session_valid(request.cookies.get(COOKIE)):
        raise _redirect()


def _redirect():
    from fastapi import HTTPException
    exc = HTTPException(302)
    exc.headers = {"Location": "/login"}
    return exc


@router.get("/", dependencies=[Depends(gui_session)])
def index(request: Request, session: Session = Depends(get_session)):
    ...  # Salden + anstehende Posten laden, templates.TemplateResponse("index.html", {...})

• Step 4: Ausführen — PASS · Step 5: Commit — feat: Web-GUI Basis, Uebersicht, Buchungsliste

Task 13: Import-Seite (Drag-and-Drop) und Planungs-Seite

Files:

• Create: finance/app/templates/import.html, finance/app/templates/planning.html, finance/app/templates/_preview_table.html
• Modify: finance/app/routers/gui.py (Routen GET /import, GET /planung)

Interfaces:

• Consumes: Import-API (Task 10), Planungs-API (Task 11).

• Step 1: Implementieren

  • import.html: Drop-Zone (<div id="dropzone">) mit kleinem Inline-Script: dragover/drop-Events, FormData + fetch("/api/imports/upload", {method: "POST", body: fd}) (Cookie-Auth), danach htmx.trigger("#imports", "refresh"). Importliste als HTMX-Fragment (hx-get-Polling alle 5 s), Status-Badges (draft = „wartet auf Bestätigung", error = rot mit error_message, confirmed = grün). Je Draft: aufklappbare Vorschau (_preview_table.html: Buchungstabelle, Saldo-Status, Duplikat-Markierung) mit Buttons „Übernehmen" (hx-post /api/imports/{id}/confirm) und „Verwerfen" (hx-delete).
  • planning.html: vier Abschnitte (wiederkehrende Posten inkl. „Vorschläge übernehmen"-Liste aus /api/recurring/suggestions, Einmalposten, Kredite mit Tilgungsplan-Aufklapper, Szenarien). Szenario-Abschnitt: Liste + Formular (Name, Checkboxen include_recurring/include_planned, Kredite zuordnen, Modifikatoren hinzufügen: Ziel-Dropdown Kategorie/Posten, Art percent/absolute/remove, Wert), Button „Durchrechnen" (hx-post /api/scenarios/{id}/project), Ergebnisanzeige (Tiefpunkt mit Datum, Unterschreitungs-Daten, Hinweis „Kurven in Grafana ansehen").

• Step 2: Alle Tests ausführen — PASS (Test aus Task 12 deckt die neuen Routen ab)

• Step 3: Manuelle Prüfung — App lokal starten und durchklicken:

cd /home/wlfb/bin/finance
FB_API_KEY=dev FB_GUI_PASSWORD_HASH=$(.venv/bin/python -c \
  "from app.auth import hash_password; print(hash_password('dev'))") \
  .venv/bin/uvicorn app.main:app --port 8000

Login mit admin/dev, alle vier Seiten öffnen, eine Test-PDF in die Drop-Zone ziehen (Fehlerfall „Bank nicht erkannt" ist ok und muss sauber angezeigt werden).

• Step 4: Commit — feat: Import-Seite mit Drag-and-Drop und Planungs-Seite

---

Phase 7: Grafana, Views, Container, Pod-Skript

Task 14: SQL-Views + Grafana-Provisionierung

Files:

• Create: finance/app/models/views.py, finance/grafana/provisioning/datasources/postgres.yaml, finance/grafana/provisioning/dashboards/provider.yaml, finance/grafana/dashboards/finanzen.json
• Modify: finance/app/main.py (Views bei Startup anlegen, nur Postgres)

Interfaces:

• Produces: Views v_balance_history(day, account, balance) (Fensterfunktion, kumulierte bestätigte Buchungen je Konto), v_balance_total(day, balance), v_monthly_by_category(month, category, total), v_projection(scenario, day, balance).

• Step 1: Views implementieren — finance/app/models/views.py:

from sqlalchemy import text
from sqlalchemy.engine import Engine

VIEWS: dict[str, str] = {
    "v_balance_history": """
        SELECT t.booking_date AS day, a.name AS account,
               SUM(SUM(t.amount)) OVER (PARTITION BY a.id
                                        ORDER BY t.booking_date) AS balance
        FROM transactions t JOIN accounts a ON a.id = t.account_id
        WHERE t.status = 'confirmed'
        GROUP BY a.id, a.name, t.booking_date""",
    "v_balance_total": """
        SELECT booking_date AS day,
               SUM(SUM(amount)) OVER (ORDER BY booking_date) AS balance
        FROM transactions WHERE status = 'confirmed' GROUP BY booking_date""",
    "v_monthly_by_category": """
        SELECT date_trunc('month', t.booking_date) AS month,
               COALESCE(c.name, 'unkategorisiert') AS category,
               SUM(t.amount) AS total
        FROM transactions t LEFT JOIN categories c ON c.id = t.category_id
        WHERE t.status = 'confirmed' GROUP BY 1, 2""",
    "v_projection": """
        SELECT s.name AS scenario, p.day, p.balance
        FROM projection_points p JOIN scenarios s ON s.id = p.scenario_id""",
}


def create_views(engine: Engine) -> None:
    if engine.dialect.name != "postgresql":
        return
    with engine.begin() as conn:
        for name, body in VIEWS.items():
            conn.execute(text(f"CREATE OR REPLACE VIEW {name} AS {body}"))

In main.py via Lifespan-Handler aufrufen (create_views(get_engine())).

• Step 2: Grafana-Provisionierung — datasources/postgres.yaml:

apiVersion: 1
datasources:
  - name: FinanzDB
    type: postgres
    access: proxy
    url: localhost:5432
    user: finance_read
    jsonData:
      database: finance
      sslmode: disable
    secureJsonData:
      password: ${FINANCE_READ_PASSWORD}

dashboards/provider.yaml:

apiVersion: 1
providers:
  - name: finanzen
    type: file
    options:
      path: /var/lib/grafana/dashboards

finance/grafana/dashboards/finanzen.json: ein Dashboard „Finanzen" mit vier Panels (Datenquelle FinanzDB, rawSql-Queries):

1. Kontostand-Verlauf (timeseries): SELECT day AS time, account AS metric, balance AS value FROM v_balance_history ORDER BY day
2. Gesamt-Saldo (timeseries): SELECT day AS time, balance AS value FROM v_balance_total ORDER BY day
3. Monat nach Kategorie (barchart, gestapelt): SELECT month AS time, category AS metric, total AS value FROM v_monthly_by_category ORDER BY month
4. Szenario-Vergleich (timeseries, Threshold-Linien bei 0 und Warnschwelle): SELECT day AS time, scenario AS metric, balance AS value FROM v_projection ORDER BY day JSON per Hand aus einem minimalen Grafana-Dashboard-Export aufbauen (schemaVersion aktuell, uid: "finanzen"); endgültige Feinheiten (Stacking, Thresholds) beim Smoke-Test in Task 16 in der Grafana-UI justieren und JSON re-exportieren.

• Step 3: Tests laufen lassen — PASS (Views werden auf SQLite übersprungen) · Step 4: Commit — feat: SQL-Views und Grafana-Provisionierung

Task 15: Containerfile

Files:

• Create: finance/Containerfile, finance/entrypoint.sh

• Step 1: Schreiben

finance/Containerfile:

FROM docker.io/library/python:3.12.11-slim

WORKDIR /srv/finance
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY app/ app/
COPY alembic/ alembic/
COPY alembic.ini .
COPY entrypoint.sh /entrypoint.sh
RUN chmod +x /entrypoint.sh
EXPOSE 8000
ENTRYPOINT ["/entrypoint.sh"]

(Falls python:3.12.11-slim nicht existiert: aktuelle 3.12-Patch-Version nachschlagen und pinnen.)

finance/entrypoint.sh:

#!/bin/sh
set -e
alembic upgrade head
exec uvicorn app.main:app --host 0.0.0.0 --port 8000

• Step 2: Build testen

cd /home/wlfb/bin/finance && podman build -t localhost/finance-api:0.1.0 .

Erwartet: Build erfolgreich.

• Step 3: Commit — feat: Containerfile und Entrypoint

Task 16: Pod-Skript create_pod_finance.sh

Files:

• Create: create_pod_finance.sh (im Repo-Root /home/wlfb/bin, ausführbar)

Struktur — exakt dem Muster von example_create_pod_langflow.sh folgen (Reihenfolge: Variablen → systemd-Stop → Verzeichnisse → Pod → Container → podman generate systemd --name --new --files → Pod stoppen/entfernen → systemctl --user enable --now → Readiness-Check):

• Step 1: Skript schreiben — Eckdaten:

POD_NAME='finance_pod'
DB_CTR_NAME='finance-db_ctr'
API_CTR_NAME='finance-api_ctr'
GRAFANA_CTR_NAME='finance-grafana_ctr'

POSTGRES_IMAGE='docker.io/library/postgres:17.10'
GRAFANA_IMAGE='docker.io/grafana/grafana-oss:12.1.0'   # vor Nutzung Tag verifizieren: podman pull
API_IMAGE='localhost/finance-api:0.1.0'

HOST_LOCAL_IP='127.0.0.1'
API_HOST_PORT='8096'; API_CONTAINER_PORT='8000'
GRAFANA_HOST_PORT='8097'; GRAFANA_CONTAINER_PORT='3000'

BIND_DIR="$HOME/.local/share/$POD_NAME"
# Unterverzeichnisse: postgres-data, grafana-data, inbox, uploads
ENV_FILE="$HOME/bin/finance/.env"

Besonderheiten gegenüber dem Beispiel:

1. .env-Erzeugung beim ersten Lauf (vor den Containern):

if [ ! -f "$ENV_FILE" ]; then
  GUI_PASSWORD=$(openssl rand -base64 12)
  cat > "$ENV_FILE" <<EOF
POSTGRES_PASSWORD=$(openssl rand -hex 16)
FINANCE_READ_PASSWORD=$(openssl rand -hex 16)
FB_API_KEY=$(openssl rand -hex 32)
FB_SESSION_SECRET=$(openssl rand -hex 32)
FB_GUI_USER=admin
FB_GUI_PASSWORD_HASH=$(python3 -c "import hashlib,secrets;s=secrets.token_hex(16);print(s+'\$'+hashlib.pbkdf2_hmac('sha256',b'$GUI_PASSWORD',s.encode(),200000).hex())")
GRAFANA_ADMIN_PASSWORD=$(openssl rand -base64 12)
EOF
  chmod 600 "$ENV_FILE"
  echo "NEU ERZEUGT: GUI-Login admin / $GUI_PASSWORD  (jetzt notieren!)"
  echo "Grafana-Login admin / siehe GRAFANA_ADMIN_PASSWORD in $ENV_FILE"
fi
set -a; . "$ENV_FILE"; set +a

2. Image-Build: podman build -t "$API_IMAGE" "$HOME/bin/finance" vor Pod-Erstellung.
3. Container-Starts:
   • Postgres: -e POSTGRES_USER=finance -e POSTGRES_PASSWORD="$POSTGRES_PASSWORD" -e POSTGRES_DB=finance -v "$BIND_DIR/postgres-data:/var/lib/postgresql/data:Z"; danach pg_isready-Warteschleife wie im Beispiel; dann Read-only-User idempotent anlegen:

podman exec "$DB_CTR_NAME" psql -U finance -d finance -c \
  "DO \$\$ BEGIN IF NOT EXISTS (SELECT FROM pg_roles WHERE rolname='finance_read')
   THEN CREATE ROLE finance_read LOGIN PASSWORD '$FINANCE_READ_PASSWORD'; END IF; END \$\$;
   GRANT CONNECT ON DATABASE finance TO finance_read;
   GRANT USAGE ON SCHEMA public TO finance_read;
   GRANT SELECT ON ALL TABLES IN SCHEMA public TO finance_read;
   ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT SELECT ON TABLES TO finance_read;"

• API: -e FB_DATABASE_URL="postgresql+psycopg://finance:$POSTGRES_PASSWORD@localhost:5432/finance" -e FB_API_KEY -e FB_SESSION_SECRET -e FB_GUI_USER -e FB_GUI_PASSWORD_HASH -e FB_INBOX_DIR=/data/inbox -e FB_UPLOADS_DIR=/data/uploads -v "$BIND_DIR/inbox:/data/inbox:Z" -v "$BIND_DIR/uploads:/data/uploads:Z"
• Grafana: -e GF_SECURITY_ADMIN_PASSWORD="$GRAFANA_ADMIN_PASSWORD" -e FINANCE_READ_PASSWORD -v "$HOME/bin/finance/grafana/provisioning:/etc/grafana/provisioning:Z,ro" -v "$HOME/bin/finance/grafana/dashboards:/var/lib/grafana/dashboards:Z,ro" -v "$BIND_DIR/grafana-data:/var/lib/grafana:Z"

4. Readiness-Check am Ende: curl auf http://127.0.0.1:8096/login und http://127.0.0.1:8097/api/health.

• Step 2: Ausführen und verifizieren

chmod +x /home/wlfb/bin/create_pod_finance.sh && /home/wlfb/bin/create_pod_finance.sh
systemctl --user status pod-finance_pod.service --no-pager
curl -s -o /dev/null -w '%{http_code}\n' http://127.0.0.1:8096/login   # 200
curl -s -o /dev/null -w '%{http_code}\n' http://127.0.0.1:8097/api/health  # 200

• Step 3: End-to-End-Smoke-Test — Im Browser: GUI-Login mit dem ausgegebenen Passwort, ein Fixture-PDF per Drag-and-Drop importieren, Vorschau bestätigen, Buchungen prüfen; Grafana öffnen, Dashboard „Finanzen" zeigt Daten; ein Basis-Szenario anlegen und durchrechnen. API mit Key testen:

. /home/wlfb/bin/finance/.env
curl -s -H "Authorization: Bearer $FB_API_KEY" http://127.0.0.1:8096/api/accounts

Grafana-Panel-Feinschliff (Task 14 Step 2) jetzt erledigen, JSON re-exportieren und einchecken.

• Step 4: Commit — feat: Pod-Skript finance_pod mit systemd-Service und Secrets-Bootstrap

---

Phase 8: Claude-Code-Umgebung in /home/wlfb/fb

Task 17: CLAUDE.md und Skills

Files (alle im Repo /home/wlfb/fb — dort committen!):

• Create: /home/wlfb/fb/CLAUDE.md, /home/wlfb/fb/.claude/skills/finanz-api/SKILL.md, /home/wlfb/fb/.claude/skills/finanzberatung/SKILL.md, /home/wlfb/fb/.claude/skills/auszug-import/SKILL.md

• Step 1: CLAUDE.md schreiben — Inhalt:

# Finanzberatungs-Umgebung

Beratung und Auswertung laufen auf Deutsch.

## Finanzberatungs-Tool
- API + Web-GUI: http://127.0.0.1:8096 (OpenAPI: /docs), Grafana: http://127.0.0.1:8097
- API-Key: `FB_API_KEY` in `/home/wlfb/bin/finance/.env` (Header `Authorization: Bearer <key>`)
- Quellcode/Betrieb: Repo `/home/wlfb/bin` (Pod `finance_pod`,
  `systemctl --user status pod-finance_pod.service`)
- Import-Inbox (PDFs hier ablegen): `~/.local/share/finance_pod/inbox/`

## Grundregeln
- Echte (bestätigte) Buchungen niemals ändern oder löschen — Korrekturen nur
  als Kategorie-Anpassung; Zukunftsplanung ausschließlich über Szenarien.
- Importe immer über die Vorschau bestätigen lassen, nie blind übernehmen.
- Empfehlungen (Kredit, Kürzungen) stets mit durchgerechneten Szenarien und
  Zahlen (Tiefpunkt, Unterschreitungs-Datum, Ratenhöhe) begründen.
- Für API-Nutzung den Skill `finanz-api` verwenden.

• Step 2: Skills schreiben — jeweils SKILL.md mit Frontmatter (name, description mit „Use when …"-Trigger):
  • finanz-api: Beschreibung aller Endpunkt-Gruppen (accounts, transactions, categories, category-rules, imports, recurring, planned, loans, scenarios, project) mit je einem konkreten curl-Beispiel, Muster:

KEY=$(grep '^FB_API_KEY=' /home/wlfb/bin/finance/.env | cut -d= -f2)
curl -s -H "Authorization: Bearer $KEY" http://127.0.0.1:8096/api/accounts | jq

• finanzberatung: Workflow — (1) Lage erfassen: Konten/Salden, unkategorisierte Buchungen, gepflegte wiederkehrende Posten prüfen (ggf. /api/recurring/suggestions durchgehen), (2) Basis-Szenario anlegen/durchrechnen, (3) Fragestellung in Varianten übersetzen (Kredit-Parameter, Kürzungs-Modifikatoren je Kategorie), (4) Varianten durchrechnen, (5) Vergleichstabelle (Tiefpunkt, Unterschreitungsdaten, Gesamtkosten Kredit aus Tilgungsplan) und Empfehlung formulieren; Hinweis auf Grafana-Dashboard „Szenario-Vergleich" für den Nutzer.

• auszug-import: Workflow — GET /api/imports auf error/draft prüfen; bei Drafts Vorschau dem Nutzer zusammenfassen und nach Bestätigung confirm aufrufen; bei Parser-Fehlern: PDF aus Inbox mit finance/scripts/dump_pdf_text.py ansehen, Regex-Konstanten im passenden Parser (/home/wlfb/bin/finance/app/parsers/) fixen, Tests laufen lassen, Image neu bauen + create_pod_finance.sh erneut ausführen.

• Step 3: Commit im fb-Repo

cd /home/wlfb/fb && git add CLAUDE.md .claude/skills && \
  git commit -m "feat: Beratungsumgebung (CLAUDE.md, Skills finanz-api/finanzberatung/auszug-import)"

---

Abschluss-Checkliste

• Alle Tests grün: cd /home/wlfb/bin/finance && .venv/bin/python -m pytest -q (Fixture-Tests ggf. SKIP, sobald PDFs da sind: PASS Pflicht)
• pod-finance_pod.service aktiv und enabled; GUI, API (mit Key) und Grafana erreichbar (Smoke-Test aus Task 16 Step 3 komplett durchlaufen)
• Kein Secret, keine PDF, keine echten Kontodaten in git log -p beider Repos
• Beide Repos committet (~/bin: Tool + Plan-Häkchen, ~/fb: Umgebung)