Ein- und Auslagerungen als Wasserfall- bzw. Brueckendiagramm

Bisher waren es zwei Linien mit der Anzahl der Vorgaenge. Gemeint war eine
Bruecke: die Bestandslinie, und schwebende gruene/rote Balken, die jeden ihrer
Spruenge erklaeren.

Damit Balken und Linie ueberhaupt zusammen in ein Bild duerfen, zaehlen die
Balken jetzt Mengen in Artikeleinheiten statt Vorgaengen - beides liegt so auf
derselben Achse. Zwei Skalen in einem Diagramm bleiben ausgeschlossen.

Neuer Endpunkt GET /dashboard/flow liefert je Abschnitt Anfangsbestand, Zugang
und Abgang aus einer Abfrage. Getrennt abgefragt koennten die Abschnitte von
timeline und activity um eine Schrittweite auseinanderliegen - dann stuende ein
Balken neben dem Sprung, den er erklaert.

/dashboard/activity bleibt fuer externe Zugriffe erhalten.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

backend/app/routers/dashboard.py

@@ -38,6 +38,7 @@ from ..schemas import (
     DashboardLayoutOut,
     DashboardStats,
     ExpirySplit,
+    FlowPoint,
     TimelinePoint,
 )
 from ..services.conversion import article_unit
@@ -388,6 +389,82 @@ def timeline(
     return punkte
 
 
+@router.get("/flow", response_model=list[FlowPoint])
+def flow(
+    days: int = 30,
+    product_id: int | None = None,
+    db: Session = Depends(get_db),
+    _: User = Depends(get_current_user),
+) -> list[FlowPoint]:
+    """Wasserfall: Anfangsbestand je Abschnitt plus Zu- und Abgang.
+
+    Balken und Bestandslinie kommen bewusst aus *einer* Abfrage. Getrennt
+    abgefragt könnten die Abschnitte der beiden Antworten um eine Schrittweite
+    auseinanderliegen – dann stünde ein Balken neben dem Sprung, den er erklärt.
+
+    Anders als :func:`activity` zählt das hier keine Vorgänge, sondern Mengen in
+    Artikeleinheiten: Nur so liegen Balken und Linie auf derselben Achse.
+    """
+    days = max(1, min(days, 730))
+    jetzt, schritt, anzahl = _eimer(days)
+
+    produkte = db.query(Product)
+    if product_id is not None:
+        produkte = produkte.filter(Product.id == product_id)
+    produkte = produkte.all()
+    if not produkte:
+        return []
+
+    faktoren = {p.id: (article_unit(p)[0] or 1.0) for p in produkte}
+    bestand = {p.id: current_stock(db, p.id) for p in produkte}
+
+    beginn = jetzt - schritt * anzahl
+    bewegungen = db.query(Movement).filter(Movement.created_at >= beginn)
+    if product_id is not None:
+        bewegungen = bewegungen.filter(Movement.product_id == product_id)
+
+    sekunden = schritt.total_seconds()
+    delta: dict[int, dict[int, float]] = defaultdict(lambda: defaultdict(float))
+    ein: dict[int, float] = defaultdict(float)
+    aus: dict[int, float] = defaultdict(float)
+    for m in bewegungen.all():
+        if m.product_id not in faktoren:
+            continue
+        index = int((jetzt - _utc(m.created_at)).total_seconds() // sekunden)
+        if not 0 <= index < anzahl:
+            continue
+        menge = _signiert(m)
+        delta[index][m.product_id] += menge
+        # Korrekturen tragen ihr Vorzeichen bereits – deshalb nach Vorzeichen
+        # einsortieren und nicht nach Bewegungsart.
+        einheiten = menge / faktoren[m.product_id]
+        if einheiten >= 0:
+            ein[index] += einheiten
+        else:
+            aus[index] -= einheiten
+
+    def summe() -> float:
+        return round(sum(bestand[pid] / faktoren[pid] for pid in bestand), 3)
+
+    # Rückwärts durch die Abschnitte: Der Endbestand des Abschnitts ist bekannt,
+    # der Anfangsbestand ergibt sich daraus, dass die Bewegungen herausfallen.
+    punkte: list[FlowPoint] = []
+    for i in range(anzahl):
+        for pid, wert in delta.get(i, {}).items():
+            bestand[pid] = bestand.get(pid, 0.0) - wert
+        punkte.append(
+            FlowPoint(
+                at=jetzt - schritt * i,
+                opening=summe(),
+                checked_in=round(ein.get(i, 0.0), 3),
+                checked_out=round(aus.get(i, 0.0), 3),
+            )
+        )
+
+    punkte.reverse()
+    return punkte
+
+
 @router.get("/activity", response_model=list[ActivityPoint])
 def activity(
     days: int = 30,

backend/app/schemas.py

@@ -441,3 +441,17 @@ class ActivityPoint(BaseModel):
     at: datetime
     checked_in: int
     checked_out: int
+
+
+class FlowPoint(BaseModel):
+    """Ein Abschnitt eines Wasserfalls: Anfangsbestand, Zu- und Abgang.
+
+    Alle Werte sind Artikeleinheiten und liegen damit auf derselben Skala wie
+    die Bestandslinie – nur so darf beides in ein Diagramm.
+    ``opening + checked_in - checked_out`` ergibt den Endbestand.
+    """
+
+    at: datetime
+    opening: float
+    checked_in: float
+    checked_out: float

backend/tests/test_dashboard.py

@@ -134,6 +134,44 @@ def test_timeline_kann_auf_ein_produkt_eingegrenzt_werden(db, rice, user):
     assert nur_reis[-1].article_units == 2
 
 
+def test_flow_liefert_eine_geschlossene_bruecke(db, rice, user):
+    """Anfangsbestand + Zugang - Abgang muss den Endbestand ergeben."""
+    check_in(db, rice, 5, "package", None, None, user)
+    check_out(db, rice, 2, "package", user)
+    db.commit()
+
+    abschnitte = dashboard.flow(days=3, product_id=None, db=db, _=user)
+    juengster = abschnitte[-1]
+    assert juengster.opening == 0          # vorher war das Lager leer
+    assert juengster.checked_in == 5       # Artikeleinheiten, nicht Vorgaenge
+    assert juengster.checked_out == 2
+    assert juengster.opening + juengster.checked_in - juengster.checked_out == 3
+
+
+def test_flow_schliesst_ohne_luecke_an_den_vorherigen_abschnitt_an(db, rice, user):
+    """Der Endbestand eines Abschnitts ist der Anfangsbestand des naechsten –
+    sonst haette die Bruecke einen Sprung, den kein Balken erklaert."""
+    check_in(db, rice, 4, "package", None, None, user)
+    db.commit()
+
+    abschnitte = dashboard.flow(days=7, product_id=None, db=db, _=user)
+    for vorher, nachher in zip(abschnitte, abschnitte[1:]):
+        schluss = vorher.opening + vorher.checked_in - vorher.checked_out
+        assert round(schluss, 3) == round(nachher.opening, 3)
+
+
+def test_flow_kann_auf_ein_produkt_eingegrenzt_werden(db, rice, user):
+    anderes = Product(name="Zucker", base_unit=BaseUnit.gram, package_size=1000)
+    db.add(anderes)
+    db.commit()
+    check_in(db, rice, 2, "package", None, None, user)
+    check_in(db, anderes, 5, "package", None, None, user)
+    db.commit()
+
+    nur_reis = dashboard.flow(days=2, product_id=rice.id, db=db, _=user)
+    assert nur_reis[-1].checked_in == 2
+
+
 def test_activity_zaehlt_ein_und_auslagerungen(db, rice, user):
     check_in(db, rice, 2, "package", None, None, user)
     check_in(db, rice, 1, "package", None, None, user)