feat(#16): Strang-basierte Klassifikation mit Ampel

- Migration: strang-Feld in ketten (scripts/migrate_strang.py)
- core/ampel.py: Ampel-Definition, Status-Mapping, Kontrollfragen
- API: strang+ampel in GET /api/ketten, /api/ketten/{id}
- API: ampel in GET /api/vorlagen/{id} (wenn in Kette)
- API: GET /api/ampel/definition (Legende für Frontend)
- API: nach_strang in GET /api/stats/dashboard
- Migration auf 6354 Ketten ausgeführt

backend/src/tracker/api/models.py

@@ -84,6 +84,7 @@ class VorlageDetail(BaseModel):
     referenzen_eingehend: list[ReferenzOut] = []
     kette_id: int | None = None
     umsetzungsbewertungen: list[UmsetzungsBewertung] = []
+    ampel: dict | None = None
 
 
 class KettenGliedOut(BaseModel):
@@ -102,6 +103,8 @@ class KetteKurz(BaseModel):
     letzte_aktivitaet: date | None = None
     vertagungen_count: int = 0
     glieder_count: int = 0
+    strang: str | None = None
+    ampel: dict | None = None
 
 
 class KetteDetail(BaseModel):
@@ -117,6 +120,8 @@ class KetteDetail(BaseModel):
     glieder: list[KettenGliedOut] = []
     antragsteller: list[ParteiOut] = []
     graph: dict | None = None
+    strang: str | None = None
+    ampel: dict | None = None
 
 
 class PaginatedVorlagen(BaseModel):

backend/src/tracker/api/routes/ampel.py

@@ -0,0 +1,14 @@
+from __future__ import annotations
+"""API routes for Ampel definitions."""
+
+from fastapi import APIRouter
+
+from tracker.core.ampel import get_ampel_definition
+
+router = APIRouter(prefix="/ampel", tags=["Ampel"])
+
+
+@router.get("/definition")
+def ampel_definition():
+    """Gibt die komplette Strang-Definition zurück (für Legende im Frontend)."""
+    return get_ampel_definition()

backend/src/tracker/api/routes/ketten.py

@@ -11,6 +11,7 @@ from tracker.api.models import (
     ParteiOut,
     VorlageKurz,
 )
+from tracker.core.ampel import get_ampel, get_ampel_kompakt
 from tracker.core.graph import get_kette_graph
 from tracker.db.session import get_connection
 
@@ -88,6 +89,7 @@ def list_ketten(
     rows = conn.execute(
         f"""SELECT k.id, k.typ, k.thema, k.status, k.status_seit,
                    k.letzte_aktivitaet, k.vertagungen_count, k.ursprung_id,
+                   k.strang,
                    v.aktenzeichen, v.typ as v_typ, v.betreff, v.datum_eingang,
                    v.ist_verwaltungsvorlage,
                    (SELECT COUNT(*) FROM ketten_glieder kg WHERE kg.kette_id = k.id) as glieder_count
@@ -117,6 +119,8 @@ def list_ketten(
             letzte_aktivitaet=r["letzte_aktivitaet"],
             vertagungen_count=r["vertagungen_count"],
             glieder_count=r["glieder_count"],
+            strang=r["strang"],
+            ampel=get_ampel_kompakt(r["strang"] or "", r["status"] or ""),
         )
         for r in rows
     ]
@@ -130,6 +134,7 @@ def get_kette(kette_id: int, conn=Depends(_db)):
     row = conn.execute(
         """SELECT k.id, k.typ, k.thema, k.status, k.status_seit,
                   k.letzte_aktivitaet, k.vertagungen_count, k.begruendung, k.ursprung_id,
+                  k.strang,
                   v.aktenzeichen, v.typ as v_typ, v.betreff, v.datum_eingang,
                   v.ist_verwaltungsvorlage
            FROM ketten k
@@ -183,6 +188,9 @@ def get_kette(kette_id: int, conn=Depends(_db)):
     # Graph/Perlenschnur data
     graph = get_kette_graph(conn, kette_id)
 
+    strang = row["strang"]
+    ampel_data = get_ampel(strang or "", row["status"] or "")
+
     return KetteDetail(
         id=row["id"],
         ursprung=VorlageKurz(
@@ -203,4 +211,6 @@ def get_kette(kette_id: int, conn=Depends(_db)):
         glieder=glieder,
         antragsteller=antragsteller,
         graph=graph,
+        strang=strang,
+        ampel=ampel_data,
     )

backend/src/tracker/api/routes/stats.py

@@ -198,6 +198,13 @@ def get_dashboard_stats(
     abgelehnt = _k_count("abgelehnt")
     total_bewertet = umgesetzt + teilweise + versandet + beschlossen + abgelehnt
 
+    # Aufschlüsselung nach Strang
+    strang_rows = conn.execute(f"""
+        SELECT strang, COUNT(*) as c FROM ketten k
+        {k_where + (' AND' if k_where else 'WHERE')} strang IS NOT NULL
+        GROUP BY strang ORDER BY c DESC
+    """.replace("WHERE  AND", "WHERE"), k_params).fetchall()
+
     return {
         "vorlagen_total": vorlagen_total,
         "ketten_total": ketten_total,
@@ -212,6 +219,7 @@ def get_dashboard_stats(
             "abgelehnt": abgelehnt,
             "total_bewertet": total_bewertet,
         },
+        "nach_strang": [{"strang": r["strang"], "anzahl": r["c"]} for r in strang_rows],
     }
 
 

backend/src/tracker/api/routes/vorlagen.py

@@ -281,12 +281,22 @@ def get_vorlage(vorlage_id: int, conn=Depends(_db)):
     # Referenzen
     refs = get_references_for_vorlage(conn, vorlage_id)
 
-    # Kette-Zugehörigkeit
+    # Kette-Zugehörigkeit + Ampel
+    from tracker.core.ampel import get_ampel
     kette_row = conn.execute(
         "SELECT kette_id FROM ketten_glieder WHERE vorlage_id = ? LIMIT 1",
         (vorlage_id,),
     ).fetchone()
 
+    kette_ampel = None
+    if kette_row:
+        kette_info = conn.execute(
+            "SELECT strang, status FROM ketten WHERE id = ?",
+            (kette_row["kette_id"],),
+        ).fetchone()
+        if kette_info and kette_info["strang"]:
+            kette_ampel = get_ampel(kette_info["strang"], kette_info["status"] or "")
+
     # KI-Zusammenfassung
     ki_row = conn.execute(
         "SELECT anmerkungen FROM ki_bewertungen WHERE vorlage_id = ? AND typ = 'zusammenfassung' LIMIT 1",
@@ -340,4 +350,5 @@ def get_vorlage(vorlage_id: int, conn=Depends(_db)):
         kette_id=kette_row["kette_id"] if kette_row else None,
         ki_zusammenfassung=ki_zusammenfassung,
         umsetzungsbewertungen=umsetzungsbewertungen,
+        ampel=kette_ampel,
     )

backend/src/tracker/core/ampel.py

@@ -0,0 +1,236 @@
+"""Ampel-Darstellungsschicht: Strang-basierte Klassifikation mit Ampel-Visualisierung.
+
+Dies ist eine reine Darstellungsschicht über der bestehenden Status-Engine (core/status.py).
+Die Status-Engine bleibt unverändert — die Ampel mappt deren Ergebnisse auf visuelle Schritte.
+"""
+
+from __future__ import annotations
+
+# Definiert die Zustände pro Strang in Reihenfolge
+STRANG_ZUSTAENDE: dict[str, list[dict]] = {
+    "antrag": [
+        {"id": "eingereicht", "label": "Eingereicht", "endfarbe": None},
+        {"id": "in_beratung", "label": "In Beratung", "endfarbe": None},
+        {"id": "beschlossen", "label": "Beschlossen", "endfarbe": "gelb"},
+        {"id": "umgesetzt", "label": "Umgesetzt", "endfarbe": "gruen"},
+    ],
+    "beschlussvorlage": [
+        {"id": "vorgelegt", "label": "Vorgelegt", "endfarbe": None},
+        {"id": "in_beratung", "label": "In Beratung", "endfarbe": None},
+        {"id": "beschlossen", "label": "Beschlossen", "endfarbe": "gelb"},
+        {"id": "umgesetzt", "label": "Umgesetzt", "endfarbe": "gruen"},
+    ],
+    "anfrage": [
+        {"id": "angefragt", "label": "Angefragt", "endfarbe": "gelb"},
+        {"id": "beantwortet", "label": "Beantwortet", "endfarbe": "gruen"},
+    ],
+    "mitteilung": [
+        {"id": "vorgelegt", "label": "Vorgelegt", "endfarbe": None},
+        {"id": "zur_kenntnis", "label": "Zur Kenntnis genommen", "endfarbe": "grau"},
+    ],
+}
+
+# Endstatus die von der Hauptreihe abzweigen
+ABZWEIGUNGEN: dict[str, dict] = {
+    "abgelehnt": {"label": "Abgelehnt", "farbe": "rot"},
+    "versandet": {"label": "Versandet", "farbe": "rot"},
+    "teilweise_umgesetzt": {"label": "Teilweise umgesetzt", "farbe": "amber"},
+    "verwiesen": {"label": "Verwiesen", "farbe": "gelb"},
+    "zurueckgezogen": {"label": "Zurückgezogen", "farbe": "grau"},
+}
+
+# Labels für Stränge
+STRANG_LABELS: dict[str, str] = {
+    "antrag": "Antrag",
+    "beschlussvorlage": "Beschlussvorlage",
+    "anfrage": "Anfrage",
+    "mitteilung": "Mitteilung",
+    "sonstig": "Sonstig",
+}
+
+# Kontrollfragen pro Strang
+KONTROLLFRAGEN: dict[str, str | None] = {
+    "antrag": "Hat die Verwaltung umgesetzt?",
+    "beschlussvorlage": "Wurde so umgesetzt wie beschlossen?",
+    "anfrage": "Wurde befriedigend geantwortet?",
+    "mitteilung": None,
+}
+
+# Mapping: DB-Status → (Schritt-ID der letzten erreichten Position, ist Abzweigung?)
+# Für jeden Strang kann das Mapping unterschiedlich sein.
+# Wir definieren ein generisches Mapping und strang-spezifische Overrides.
+
+_STATUS_TO_SCHRITT: dict[str, dict[str, tuple[str, bool]]] = {
+    "antrag": {
+        "eingereicht": ("eingereicht", False),
+        "in_beratung": ("in_beratung", False),
+        "vertagt": ("in_beratung", False),  # Vertagt = noch in Beratung
+        "beschlossen": ("beschlossen", False),
+        "umgesetzt": ("umgesetzt", False),
+        "versandet": ("beschlossen", True),
+        "abgelehnt": ("in_beratung", True),
+        "teilweise_umgesetzt": ("beschlossen", True),
+        "verwiesen": ("in_beratung", True),
+        "zurückgezogen": ("in_beratung", True),
+        "zurueckgezogen": ("in_beratung", True),
+        "abgewiegelt": ("beschlossen", True),
+        "offen": ("in_beratung", False),
+    },
+    "beschlussvorlage": {
+        "eingereicht": ("vorgelegt", False),
+        "vorgelegt": ("vorgelegt", False),
+        "in_beratung": ("in_beratung", False),
+        "vertagt": ("in_beratung", False),
+        "beschlossen": ("beschlossen", False),
+        "umgesetzt": ("umgesetzt", False),
+        "versandet": ("beschlossen", True),
+        "abgelehnt": ("in_beratung", True),
+        "teilweise_umgesetzt": ("beschlossen", True),
+        "verwiesen": ("in_beratung", True),
+        "zurückgezogen": ("in_beratung", True),
+        "zurueckgezogen": ("in_beratung", True),
+        "abgewiegelt": ("beschlossen", True),
+        "offen": ("in_beratung", False),
+    },
+    "anfrage": {
+        "angefragt": ("angefragt", False),
+        "beantwortet": ("beantwortet", False),
+        "offen": ("angefragt", False),
+        "abgewiegelt": ("angefragt", True),
+        "versandet": ("angefragt", True),
+        "zurückgezogen": ("angefragt", True),
+        "zurueckgezogen": ("angefragt", True),
+    },
+    "mitteilung": {
+        "vorgelegt": ("vorgelegt", False),
+        "zur_kenntnis": ("zur_kenntnis", False),
+        "eingereicht": ("vorgelegt", False),
+        "beantwortet": ("zur_kenntnis", False),  # Mapped to equivalent
+        "beschlossen": ("zur_kenntnis", False),
+    },
+}
+
+
+def _normalize_abzweigung_id(status: str) -> str:
+    """Normalize status string to ABZWEIGUNGEN key."""
+    mapping = {
+        "zurückgezogen": "zurueckgezogen",
+        "abgewiegelt": "versandet",  # Abgewiegelt → same visual as versandet
+    }
+    return mapping.get(status, status)
+
+
+def get_ampel(strang: str, aktueller_status: str) -> dict | None:
+    """Gibt die Ampel-Daten für Frontend zurück.
+
+    Returns None if strang is unknown or 'sonstig'.
+    """
+    if not strang or strang not in STRANG_ZUSTAENDE:
+        return None
+
+    schritte_def = STRANG_ZUSTAENDE[strang]
+    status_map = _STATUS_TO_SCHRITT.get(strang, {})
+
+    # Determine position and whether it's a branch-off
+    mapping = status_map.get(aktueller_status or "")
+    if mapping is None:
+        # Unknown status — show first step as active
+        aktiver_schritt_id = schritte_def[0]["id"]
+        ist_abzweigung = False
+    else:
+        aktiver_schritt_id, ist_abzweigung = mapping
+
+    # Find index of active step
+    schritt_ids = [s["id"] for s in schritte_def]
+    try:
+        aktiver_idx = schritt_ids.index(aktiver_schritt_id)
+    except ValueError:
+        aktiver_idx = 0
+
+    # Build schritte list
+    schritte = []
+    for i, s in enumerate(schritte_def):
+        if ist_abzweigung:
+            # Bei Abzweigung: alle bis aktiver_idx sind "erreicht", keiner ist "aktiv"
+            erreicht = i <= aktiver_idx
+            aktiv = False
+        else:
+            erreicht = i <= aktiver_idx
+            aktiv = i == aktiver_idx
+
+        # Farbe bestimmen
+        if aktiv and s["endfarbe"]:
+            farbe = s["endfarbe"]
+        elif aktiv:
+            farbe = "blau"  # Aktiver Schritt ohne spezielle Endfarbe
+        elif erreicht:
+            farbe = "grau"  # Bereits durchlaufen
+        else:
+            farbe = "grau"  # Noch nicht erreicht
+
+        schritte.append({
+            "id": s["id"],
+            "label": s["label"],
+            "aktiv": aktiv,
+            "erreicht": erreicht,
+            "farbe": farbe,
+        })
+
+    # Abzweigung
+    abzweigung = None
+    if ist_abzweigung:
+        norm_status = _normalize_abzweigung_id(aktueller_status or "")
+        if norm_status in ABZWEIGUNGEN:
+            abzw = ABZWEIGUNGEN[norm_status]
+            abzweigung = {
+                "id": norm_status,
+                "label": abzw["label"],
+                "farbe": abzw["farbe"],
+            }
+
+    return {
+        "strang": strang,
+        "strang_label": STRANG_LABELS.get(strang, strang.capitalize()),
+        "kontrollfrage": KONTROLLFRAGEN.get(strang),
+        "schritte": schritte,
+        "abzweigung": abzweigung,
+    }
+
+
+def get_ampel_kompakt(strang: str, aktueller_status: str) -> dict | None:
+    """Kompakte Ampel-Version für Listen: nur aktueller Schritt + Farbe."""
+    ampel = get_ampel(strang, aktueller_status)
+    if not ampel:
+        return None
+
+    if ampel["abzweigung"]:
+        return {
+            "schritt": ampel["abzweigung"]["label"],
+            "farbe": ampel["abzweigung"]["farbe"],
+            "ist_abzweigung": True,
+        }
+
+    aktiver = next((s for s in ampel["schritte"] if s["aktiv"]), None)
+    if aktiver:
+        return {
+            "schritt": aktiver["label"],
+            "farbe": aktiver["farbe"],
+            "ist_abzweigung": False,
+        }
+
+    return None
+
+
+def get_ampel_definition() -> dict:
+    """Gibt die komplette Strang-Definition zurück (für Legende im Frontend)."""
+    return {
+        "straenge": {
+            strang: {
+                "label": STRANG_LABELS.get(strang, strang.capitalize()),
+                "kontrollfrage": KONTROLLFRAGEN.get(strang),
+                "schritte": schritte,
+            }
+            for strang, schritte in STRANG_ZUSTAENDE.items()
+        },
+        "abzweigungen": ABZWEIGUNGEN,
+    }

backend/src/tracker/main.py

@@ -9,7 +9,7 @@ from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import FileResponse
 from fastapi.staticfiles import StaticFiles
 
-from tracker.api.routes import abstimmungen, bewertung, fraktionen, ketten, orte, stats, vorlagen
+from tracker.api.routes import abstimmungen, ampel, bewertung, fraktionen, ketten, orte, stats, vorlagen
 
 app = FastAPI(
     title="Antragstracker Hagen",
@@ -31,6 +31,7 @@ app.include_router(abstimmungen.router, prefix="/api")
 app.include_router(orte.router, prefix="/api")
 app.include_router(fraktionen.router, prefix="/api")
 app.include_router(bewertung.router, prefix="/api")
+app.include_router(ampel.router, prefix="/api")
 
 
 @app.get("/api/health")

scripts/migrate_strang.py

@@ -0,0 +1,64 @@
+#!/usr/bin/env python3
+"""Migration: Add 'strang' column to ketten and populate based on Ursprungs-Vorlage typ."""
+
+import sqlite3
+import sys
+from pathlib import Path
+
+DB_PATH = Path(__file__).parent.parent / "data" / "tracker.db"
+
+
+VORLAGE_TYP_TO_STRANG = {
+    "antrag": "antrag",
+    "anfrage": "anfrage",
+    "beschlussvorlage": "beschlussvorlage",
+    "mitteilungsvorlage": "mitteilung",
+}
+
+
+def migrate(db_path: Path = DB_PATH):
+    conn = sqlite3.connect(str(db_path))
+    conn.row_factory = sqlite3.Row
+
+    # Check if column already exists
+    cols = [r[1] for r in conn.execute("PRAGMA table_info(ketten)").fetchall()]
+    if "strang" not in cols:
+        print("Adding 'strang' column to ketten...")
+        conn.execute("ALTER TABLE ketten ADD COLUMN strang TEXT")
+        conn.commit()
+        print("Column added.")
+    else:
+        print("Column 'strang' already exists.")
+
+    # Populate strang based on Ursprungs-Vorlage typ
+    rows = conn.execute("""
+        SELECT k.id, v.typ as vorlage_typ
+        FROM ketten k
+        LEFT JOIN vorlagen v ON k.ursprung_id = v.id
+        WHERE k.strang IS NULL
+    """).fetchall()
+
+    if not rows:
+        print("All ketten already have strang set.")
+        conn.close()
+        return
+
+    print(f"Updating {len(rows)} ketten...")
+    counts = {}
+    for r in rows:
+        vorlage_typ = r["vorlage_typ"] or ""
+        strang = VORLAGE_TYP_TO_STRANG.get(vorlage_typ, "sonstig")
+        conn.execute("UPDATE ketten SET strang = ? WHERE id = ?", (strang, r["id"]))
+        counts[strang] = counts.get(strang, 0) + 1
+
+    conn.commit()
+    conn.close()
+
+    print("Done. Counts per strang:")
+    for strang, count in sorted(counts.items()):
+        print(f"  {strang}: {count}")
+
+
+if __name__ == "__main__":
+    db = Path(sys.argv[1]) if len(sys.argv) > 1 else DB_PATH
+    migrate(db)