From 7c36b7d276a3a7b4faa2a43b1d09f6f4e7b36922 Mon Sep 17 00:00:00 2001 From: Cizz22 Date: Wed, 15 Apr 2026 13:25:56 +0700 Subject: [PATCH] refactor: improve knapsack selection accuracy with dynamic scaling and greedy budget filling --- src/calculation_budget_constrains/service.py | 85 +++++++++++--------- 1 file changed, 49 insertions(+), 36 deletions(-) diff --git a/src/calculation_budget_constrains/service.py b/src/calculation_budget_constrains/service.py index d16d7b5..faa1c24 100644 --- a/src/calculation_budget_constrains/service.py +++ b/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. - Falls back to greedy if W is too large. + Uses dynamic scaling + 1D DP optimization to ensure accuracy for very large numbers. """ - n = len(equipments) + if not equipments: + return [], [] - # Scale costs + budget - costs = [int(eq["total_cost"] // scale) for eq in equipments] - values = [eq["priority_score"] for eq in equipments] - W = int(budget // scale) + # Pre-filter: strictly impossible items + eligible_items = [] + strictly_excluded = [] + 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 - if W > 1_000_000: - print("too big") - return greedy_selection(equipments, budget) + costs = [int(math.ceil(eq["total_cost"] / scale)) for eq in eligible_items] + values = [eq["priority_score"] for eq in eligible_items] + W = int(budget // scale) # 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 - selected, excluded = [], [] + # Backtrack + 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 - remaining_budget = budget - sum(eq["total_cost"] for eq in selected) + excluded = backtrack_excluded + strictly_excluded + + # 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)