refactor: improve knapsack selection accuracy with dynamic scaling and greedy budget filling

src/calculation_budget_constrains/service.py

@@ -1,6 +1,7 @@
+import math
 from collections import defaultdict
 import random
-from typing import Optional
+from typing import Optional, List, Dict, Tuple
 from uuid import UUID
 
 from sqlalchemy import Delete, Select
@@ -8,23 +9,12 @@ from sqlalchemy import Delete, Select
 from src.auth.service import CurrentUser
 from src.contribution_util import calculate_contribution_accurate
 from src.database.core import CollectorDbSession, DbSession
-# from src.scope_equipment.model import ScopeEquipment
-# from src.scope_equipment.service import get_by_scope_name
 from src.overhaul_activity.service import get_all_by_session_id, get_standard_scope_by_session_id
 
 # async def get_all_budget_constrains(
 #     *, db_session: DbSession, session_id: str, cost_threshold: float = 100000000
 # ):
 
-# At the module level, add this dictionary to store persistent EAF values
-from collections import defaultdict
-from uuid import UUID
-from typing import List, Dict, Tuple
-
-from src.database.core import CollectorDbSession, DbSession
-from src.overhaul_activity.service import get_standard_scope_by_session_id
-from src.contribution_util import calculate_contribution_accurate
-
 
 async def get_all_budget_constrains(
     *,
@@ -109,64 +99,87 @@ def calculate_asset_eaf_contributions(plant_result, eq_results):
 
 
 def greedy_selection(equipments: List[dict], budget: float) -> Tuple[List[dict], List[dict]]:
-    """Greedy fallback: select items by score until budget is used."""
+    """Greedy selection: select items by score until budget is used."""
     # Sort by priority_score descending
     equipments_sorted = sorted(equipments, key=lambda x: x["priority_score"], reverse=True)
-    total_cost = 0
+    current_cost_total = 0.0
     selected, excluded = [], []
     
     for eq in equipments_sorted:
-        if total_cost + eq["cost"] <= budget:
+        cost = eq.get("total_cost", 0.0)
+        if current_cost_total + cost <= budget:
             selected.append(eq)
-            total_cost += eq["cost"]
+            current_cost_total += cost
         else:
             excluded.append(eq)
     
     return selected, excluded
 
 
-def knapsack_selection(equipments: List[dict], budget: float, scale: int = 10_000_000) -> Tuple[List[dict], List[dict]]:
+def knapsack_selection(equipments: List[dict], budget: float, scale: Optional[float] = None) -> Tuple[List[dict], List[dict]]:
     """
     Select equipment optimally within budget using 0/1 knapsack DP.
-    Uses scaling + 1D DP optimization to avoid MemoryError.
+    Uses dynamic scaling + 1D DP optimization to ensure accuracy for very large numbers.
-    Falls back to greedy if W is too large.
     """
-    n = len(equipments)
+    if not equipments:
+        return [], []
 
-    # Scale costs + budget
+    # Pre-filter: strictly impossible items
-    costs = [int(eq["total_cost"] // scale) for eq in equipments]
+    eligible_items = []
-    values = [eq["priority_score"] for eq in equipments]
+    strictly_excluded = []
-    W = int(budget // scale)
+    for eq in equipments:
+        if eq["total_cost"] > budget:
+            strictly_excluded.append(eq)
+        else:
+            eligible_items.append(eq)
+
+    if not eligible_items:
+        return [], strictly_excluded
+
+    n = len(eligible_items)
+
+    # Dynamic scaling for big numbers: target higher W (10,000) for better precision
+    if scale is None:
+        target_W = 10000
+        scale = max(1.0, budget / target_W)
 
-    # Fallback if W is still too large
+    costs = [int(math.ceil(eq["total_cost"] / scale)) for eq in eligible_items]
-    if W > 1_000_000:
+    values = [eq["priority_score"] for eq in eligible_items]
-        print("too big")
+    W = int(budget // scale)
-        return greedy_selection(equipments, budget)
 
     # 1D DP array
     dp = [0.0] * (W + 1)
-    keep = [[False] * (W + 1) for _ in range(n)]  # track selection choices
+    # 2D table for backtracking
+    keep = [[False] * (W + 1) for _ in range(n)]
 
     for i in range(n):
         cost, value = costs[i], values[i]
+        # Skip if cost is zero but actual cost is greater than budget (handled by eligible_items filter)
         for w in range(W, cost - 1, -1):
-            if dp[w - cost] + value >= dp[w]:  # <= FIXED HERE
+            if dp[w - cost] + value >= dp[w]:
                 dp[w] = dp[w - cost] + value
                 keep[i][w] = True
 
-    # Backtrack to find selected items
+    # Backtrack
-    selected, excluded = [], []
+    selected = []
+    backtrack_excluded = []
     w = W
     for i in range(n - 1, -1, -1):
         if keep[i][w]:
-            selected.append(equipments[i])
+            selected.append(eligible_items[i])
             w -= costs[i]
         else:
-            excluded.append(equipments[i])
+            backtrack_excluded.append(eligible_items[i])
 
-    # Optional: fill leftover budget with zero-priority items
+    excluded = backtrack_excluded + strictly_excluded
-    remaining_budget = budget - sum(eq["total_cost"] for eq in selected)
+
+    # Precision correction: greedy fill leftover budget with actual values
+    current_total_spent = sum(eq["total_cost"] for eq in selected)
+    remaining_budget = budget - current_total_spent
+    
     if remaining_budget > 0:
+        # Sort by priority score DESC then cost ASC
+        excluded.sort(key=lambda x: (-x["priority_score"], x["total_cost"]))
         for eq in excluded[:]:
             if eq["total_cost"] <= remaining_budget:
                 selected.append(eq)