feat: SubstitutionEngine — deterministic swap candidates with compensation hints

app/services/recipe/substitution_engine.py

@@ -0,0 +1,126 @@
+"""
+SubstitutionEngine — deterministic ingredient swap candidates with compensation hints.
+
+Powered by:
+  - substitution_pairs table (derived from lishuyang/recipepairs)
+  - ingredient_profiles functional metadata (USDA FDC)
+"""
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass, field
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from app.db.store import Store
+
+# Compensation threshold — if |delta| exceeds this, surface a hint
+_FAT_THRESHOLD = 5.0       # grams per 100g
+_GLUTAMATE_THRESHOLD = 1.0 # mg per 100g
+_MOISTURE_THRESHOLD = 15.0 # grams per 100g
+
+_RICHNESS_COMPENSATORS = ["olive oil", "coconut oil", "butter", "shortening", "full-fat coconut milk"]
+_DEPTH_COMPENSATORS    = ["nutritional yeast", "soy sauce", "miso", "mushroom powder",
+                           "better than bouillon not-beef", "smoked paprika"]
+_MOISTURE_BINDERS      = ["cornstarch", "flour", "arrowroot", "breadcrumbs"]
+
+
+@dataclass(frozen=True)
+class CompensationHint:
+    ingredient: str
+    reason: str
+    element: str
+
+
+@dataclass(frozen=True)
+class SubstitutionSwap:
+    original_name: str
+    substitute_name: str
+    constraint_label: str
+    fat_delta: float
+    moisture_delta: float
+    glutamate_delta: float
+    protein_delta: float
+    occurrence_count: int
+    compensation_hints: list[dict] = field(default_factory=list)
+    explanation: str = ""
+
+
+class SubstitutionEngine:
+    def __init__(self, store: "Store") -> None:
+        self._store = store
+
+    def find_substitutes(
+        self,
+        ingredient_name: str,
+        constraint: str,
+    ) -> list[SubstitutionSwap]:
+        rows = self._store._fetch_all("""
+            SELECT substitute_name, constraint_label,
+                   fat_delta, moisture_delta, glutamate_delta, protein_delta,
+                   occurrence_count, compensation_hints
+            FROM substitution_pairs
+            WHERE original_name = ? AND constraint_label = ?
+            ORDER BY occurrence_count DESC
+        """, (ingredient_name.lower(), constraint))
+
+        return [self._row_to_swap(ingredient_name, row) for row in rows]
+
+    def _row_to_swap(self, original: str, row: dict) -> SubstitutionSwap:
+        hints = self._build_hints(row)
+        explanation = self._build_explanation(original, row, hints)
+        return SubstitutionSwap(
+            original_name=original,
+            substitute_name=row["substitute_name"],
+            constraint_label=row["constraint_label"],
+            fat_delta=row.get("fat_delta") or 0.0,
+            moisture_delta=row.get("moisture_delta") or 0.0,
+            glutamate_delta=row.get("glutamate_delta") or 0.0,
+            protein_delta=row.get("protein_delta") or 0.0,
+            occurrence_count=row.get("occurrence_count") or 1,
+            compensation_hints=[{"ingredient": h.ingredient, "reason": h.reason, "element": h.element} for h in hints],
+            explanation=explanation,
+        )
+
+    def _build_hints(self, row: dict) -> list[CompensationHint]:
+        hints = []
+        fat_delta = row.get("fat_delta") or 0.0
+        glutamate_delta = row.get("glutamate_delta") or 0.0
+        moisture_delta = row.get("moisture_delta") or 0.0
+
+        if fat_delta < -_FAT_THRESHOLD:
+            missing = abs(fat_delta)
+            sugg = _RICHNESS_COMPENSATORS[0]
+            hints.append(CompensationHint(
+                ingredient=sugg,
+                reason=f"substitute has ~{missing:.0f}g/100g less fat — add {sugg} to restore Richness",
+                element="Richness",
+            ))
+
+        if glutamate_delta < -_GLUTAMATE_THRESHOLD:
+            sugg = _DEPTH_COMPENSATORS[0]
+            hints.append(CompensationHint(
+                ingredient=sugg,
+                reason=f"substitute is lower in umami — add {sugg} to restore Depth",
+                element="Depth",
+            ))
+
+        if moisture_delta > _MOISTURE_THRESHOLD:
+            sugg = _MOISTURE_BINDERS[0]
+            hints.append(CompensationHint(
+                ingredient=sugg,
+                reason=f"substitute adds ~{moisture_delta:.0f}g/100g more moisture — add {sugg} to maintain Structure",
+                element="Structure",
+            ))
+
+        return hints
+
+    def _build_explanation(
+        self, original: str, row: dict, hints: list[CompensationHint]
+    ) -> str:
+        sub = row["substitute_name"]
+        count = row.get("occurrence_count") or 1
+        base = f"Replace {original} with {sub} (seen in {count} recipes)."
+        if hints:
+            base += " To compensate: " + "; ".join(h.reason for h in hints) + "."
+        return base

tests/services/recipe/test_substitution_engine.py

@@ -0,0 +1,45 @@
+import json, pytest
+from tests.services.recipe.test_element_classifier import store_with_profiles
+
+
+@pytest.fixture
+def store_with_subs(store_with_profiles):
+    store_with_profiles.conn.execute("""
+        INSERT INTO substitution_pairs
+          (original_name, substitute_name, constraint_label,
+           fat_delta, moisture_delta, glutamate_delta, occurrence_count)
+        VALUES (?,?,?,?,?,?,?)
+    """, ("butter", "coconut oil", "vegan", -1.0, 0.0, 0.0, 15))
+    store_with_profiles.conn.execute("""
+        INSERT INTO substitution_pairs
+          (original_name, substitute_name, constraint_label,
+           fat_delta, moisture_delta, glutamate_delta, occurrence_count)
+        VALUES (?,?,?,?,?,?,?)
+    """, ("ground beef", "lentils", "vegan", -15.0, 10.0, -2.0, 45))
+    store_with_profiles.conn.commit()
+    return store_with_profiles
+
+
+def test_find_substitutes_for_constraint(store_with_subs):
+    from app.services.recipe.substitution_engine import SubstitutionEngine
+    engine = SubstitutionEngine(store_with_subs)
+    results = engine.find_substitutes("butter", constraint="vegan")
+    assert len(results) > 0
+    assert results[0].substitute_name == "coconut oil"
+
+
+def test_compensation_hints_for_large_delta(store_with_subs):
+    from app.services.recipe.substitution_engine import SubstitutionEngine
+    engine = SubstitutionEngine(store_with_subs)
+    results = engine.find_substitutes("ground beef", constraint="vegan")
+    assert len(results) > 0
+    swap = results[0]
+    # Fat delta is -15g — should suggest a Richness compensation
+    assert any(h["element"] == "Richness" for h in swap.compensation_hints)
+
+
+def test_no_substitutes_returns_empty(store_with_subs):
+    from app.services.recipe.substitution_engine import SubstitutionEngine
+    engine = SubstitutionEngine(store_with_subs)
+    results = engine.find_substitutes("unobtainium", constraint="vegan")
+    assert results == []