diff --git a/Plan.md b/Plan.md new file mode 100644 index 0000000..20175c4 --- /dev/null +++ b/Plan.md @@ -0,0 +1,2031 @@ +# 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`: +```ini +[pytest] +testpaths = tests +``` + +`finance/.gitignore`: +``` +.env +__pycache__/ +*.sqlite +tests/fixtures/*.pdf +.venv/ +``` + +`finance/app/config.py`: +```python +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`: +```python +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** + +```bash +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** + +```bash +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`: +```python +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`: +```python +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`: +```python +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): +```python +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`: +```python +from app.models import tables # noqa: F401 +``` + +- [ ] **Step 4: Test ausführen — erwartet PASS** (`.venv/bin/python -m pytest -q`) + +- [ ] **Step 5: Alembic initialisieren** + +```bash +cd /home/wlfb/bin/finance +.venv/bin/alembic init alembic +``` +In `alembic/env.py` ergänzen (vor `run_migrations_offline`): +```python +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: +```bash +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`: +```python +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`: +```python +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`: +```python +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`: +```python +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`: +```python +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`: +```python +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`: +```python +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`: +```python +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`: +```python +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`: +```python +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`: +```python +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`: +```python +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`: +```markdown +# 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`: +```python +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"`): +```python +"""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. 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`: +```python +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`: +```python +"""PDF-Text anzeigen, um Parser-Regexe abzugleichen: python scripts/dump_pdf_text.py """ +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 ` **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: +```python +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`: +```python +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`: +```python +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`: +```python +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 """
+ + +
""" + + +@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`: +```python +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`: +```python +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`: +```python +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`): +```python +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=, iban=, 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: +```python +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 `ProjectionPoint`s 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`: +```python +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`: +```python +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** +```bash +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`: +```python +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://:8097`, Logout-Button), ``, `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, `