Scenario Planner for Construction

Overview

Model different project scenarios to understand their impacts on cost, schedule, and resources. Compare alternatives, optimize decisions, and prepare for contingencies.

Business Case

Construction decisions require understanding trade-offs:

• Design Alternatives: Which option is most cost-effective?
• Schedule Compression: What's the cost of accelerating?
• Resource Options: In-house vs. subcontractor?
• Risk Scenarios: What if materials increase 20%?

Technical Implementation

from dataclasses import dataclass, field
from typing import List, Dict, Any, Optional, Callable
from datetime import datetime, timedelta
import pandas as pd
import numpy as np
from copy import deepcopy

@dataclass
class ScenarioParameter:
    name: str
    base_value: float
    unit: str
    min_value: Optional[float] = None
    max_value: Optional[float] = None
    description: str = ""

@dataclass
class Scenario:
    id: str
    name: str
    description: str
    parameters: Dict[str, float]
    created_at: datetime = field(default_factory=datetime.now)

@dataclass
class ScenarioResult:
    scenario_id: str
    scenario_name: str
    total_cost: float
    total_duration: int  # days
    resource_requirements: Dict[str, float]
    risk_score: float
    key_metrics: Dict[str, float]
    warnings: List[str]
    comparison_to_base: Dict[str, float]

@dataclass
class SensitivityResult:
    parameter: str
    values_tested: List[float]
    cost_impacts: List[float]
    duration_impacts: List[float]
    sensitivity_score: float

class ConstructionScenarioPlanner:
    """Scenario planning and what-if analysis for construction."""

    def __init__(self, base_project: Dict):
        self.base_project = base_project
        self.parameters: Dict[str, ScenarioParameter] = {}
        self.scenarios: Dict[str, Scenario] = {}
        self.results: Dict[str, ScenarioResult] = {}
        self.cost_model: Optional[Callable] = None
        self.duration_model: Optional[Callable] = None
        self._setup_default_parameters()

    def _setup_default_parameters(self):
        """Setup common construction scenario parameters."""
        default_params = [
            ScenarioParameter("labor_rate", 75, "$/hr", 50, 150, "Average labor rate"),
            ScenarioParameter("material_escalation", 0, "%", -10, 30, "Material cost change"),
            ScenarioParameter("productivity_factor", 1.0, "x", 0.5, 1.5, "Labor productivity multiplier"),
            ScenarioParameter("overtime_percentage", 0, "%", 0, 50, "Overtime work percentage"),
            ScenarioParameter("crew_size", 10, "workers", 5, 50, "Average crew size"),
            ScenarioParameter("work_days_per_week", 5, "days", 5, 7, "Working days per week"),
            ScenarioParameter("contingency_percentage", 10, "%", 5, 25, "Cost contingency"),
            ScenarioParameter("weather_delay_days", 0, "days", 0, 60, "Expected weather delays"),
            ScenarioParameter("permit_delay_days", 0, "days", 0, 90, "Expected permit delays"),
            ScenarioParameter("subcontractor_markup", 15, "%", 10, 30, "Subcontractor markup"),
        ]

        for param in default_params:
            self.parameters[param.name] = param

    def add_parameter(self, param: ScenarioParameter):
        """Add custom parameter."""
        self.parameters[param.name] = param

    def set_cost_model(self, model: Callable):
        """Set custom cost calculation model."""
        self.cost_model = model

    def set_duration_model(self, model: Callable):
        """Set custom duration calculation model."""
        self.duration_model = model

    def create_scenario(self, name: str, description: str,
                       parameter_changes: Dict[str, float]) -> Scenario:
        """Create a new scenario with parameter modifications."""
        # Start with base values
        params = {p.name: p.base_value for p in self.parameters.values()}

        # Apply changes
        for param_name, value in parameter_changes.items():
            if param_name in params:
                params[param_name] = value
            else:
                raise ValueError(f"Unknown parameter: {param_name}")

        scenario = Scenario(
            id=f"SCN-{len(self.scenarios) + 1:03d}",
            name=name,
            description=description,
            parameters=params
        )

        self.scenarios[scenario.id] = scenario
        return scenario

    def calculate_cost(self, params: Dict[str, float]) -> float:
        """Calculate total project cost based on parameters."""
        if self.cost_model:
            return self.cost_model(self.base_project, params)

        # Default cost model
        base_cost = self.base_project.get('base_cost', 1000000)

        # Labor adjustments
        labor_factor = params['labor_rate'] / 75  # Normalized to base rate
        productivity_impact = 1 / params['productivity_factor']
        overtime_premium = 1 + (params['overtime_percentage'] / 100 * 0.5)

        labor_cost = base_cost * 0.4 * labor_factor * productivity_impact * overtime_premium

        # Material adjustments
        material_cost = base_cost * 0.35 * (1 + params['material_escalation'] / 100)

        # Equipment and other
        equipment_cost = base_cost * 0.15

        # Subcontractor
        sub_cost = base_cost * 0.1 * (1 + params['subcontractor_markup'] / 100)

        subtotal = labor_cost + material_cost + equipment_cost + sub_cost

        # Contingency
        total = subtotal * (1 + params['contingency_percentage'] / 100)

        return total

    def calculate_duration(self, params: Dict[str, float]) -> int:
        """Calculate project duration based on parameters."""
        if self.duration_model:
            return self.duration_model(self.base_project, params)

        # Default duration model
        base_duration = self.base_project.get('base_duration', 365)

        # Crew size impact
        crew_factor = 10 / params['crew_size']  # Inverse relationship

        # Productivity impact
        productivity_factor = 1 / params['productivity_factor']

        # Work days impact
        workday_factor = 5 / params['work_days_per_week']

        # Overtime compression
        overtime_compression = 1 - (params['overtime_percentage'] / 100 * 0.3)

        calculated_duration = base_duration * crew_factor * productivity_factor * workday_factor * overtime_compression

        # Add delays
        delays = params['weather_delay_days'] + params['permit_delay_days']

        return int(calculated_duration + delays)

    def evaluate_scenario(self, scenario: Scenario) -> ScenarioResult:
        """Evaluate a scenario and calculate results."""
        params = scenario.parameters

        total_cost = self.calculate_cost(params)
        total_duration = self.calculate_duration(params)

        # Calculate resource requirements
        resources = {
            'labor_hours': total_duration * params['crew_size'] * 8 * (params['work_days_per_week'] / 5),
            'peak_workers': params['crew_size'] * (1 + params['overtime_percentage'] / 100 * 0.5),
            'overtime_hours': total_duration * params['crew_size'] * 8 * params['overtime_percentage'] / 100,
        }

        # Calculate risk score (0-100)
        risk_factors = [
            params['overtime_percentage'] / 50 * 20,  # High overtime = higher risk
            (1 - params['productivity_factor']) * 20 if params['productivity_factor'] < 1 else 0,
            params['material_escalation'] / 30 * 15 if params['material_escalation'] > 0 else 0,
            (25 - params['contingency_percentage']) / 20 * 15,  # Low contingency = higher risk
        ]
        risk_score = min(sum(risk_factors), 100)

        # Key metrics
        cost_per_day = total_cost / total_duration
        cost_per_sf = total_cost / self.base_project.get('gross_area', 50000)

        key_metrics = {
            'cost_per_day': cost_