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_per_day,
        'cost_per_sf': cost_per_sf,
        'labor_productivity': resources['labor_hours'] / total_duration,
    }

    # Warnings
    warnings = []
    if params['overtime_percentage'] > 30:
        warnings.append("High overtime may cause burnout and quality issues")
    if params['contingency_percentage'] < 8:
        warnings.append("Low contingency increases risk of budget overrun")
    if params['productivity_factor'] < 0.8:
        warnings.append("Low productivity factor may not be sustainable")

    # Compare to base scenario
    base_params = {p.name: p.base_value for p in self.parameters.values()}
    base_cost = self.calculate_cost(base_params)
    base_duration = self.calculate_duration(base_params)

    comparison = {
        'cost_change_pct': ((total_cost - base_cost) / base_cost) * 100,
        'cost_change_abs': total_cost - base_cost,
        'duration_change_pct': ((total_duration - base_duration) / base_duration) * 100,
        'duration_change_days': total_duration - base_duration,
    }

    result = ScenarioResult(
        scenario_id=scenario.id,
        scenario_name=scenario.name,
        total_cost=total_cost,
        total_duration=total_duration,
        resource_requirements=resources,
        risk_score=risk_score,
        key_metrics=key_metrics,
        warnings=warnings,
        comparison_to_base=comparison
    )

    self.results[scenario.id] = result
    return result

def run_sensitivity_analysis(self, parameter: str,
                              values: List[float] = None,
                              steps: int = 10) -> SensitivityResult:
    """Run sensitivity analysis on a single parameter."""
    if parameter not in self.parameters:
        raise ValueError(f"Unknown parameter: {parameter}")

    param = self.parameters[parameter]

    if values is None:
        min_val = param.min_value or param.base_value * 0.5
        max_val = param.max_value or param.base_value * 1.5
        values = np.linspace(min_val, max_val, steps).tolist()

    base_params = {p.name: p.base_value for p in self.parameters.values()}
    base_cost = self.calculate_cost(base_params)
    base_duration = self.calculate_duration(base_params)

    cost_impacts = []
    duration_impacts = []

    for val in values:
        test_params = base_params.copy()
        test_params[parameter] = val

        cost = self.calculate_cost(test_params)
        duration = self.calculate_duration(test_params)

        cost_impacts.append(((cost - base_cost) / base_cost) * 100)
        duration_impacts.append(((duration - base_duration) / base_duration) * 100)

    # Calculate sensitivity score (range of impact)
    cost_range = max(cost_impacts) - min(cost_impacts)
    duration_range = max(duration_impacts) - min(duration_impacts)
    sensitivity_score = (cost_range + duration_range) / 2

    return SensitivityResult(
        parameter=parameter,
        values_tested=values,
        cost_impacts=cost_impacts,
        duration_impacts=duration_impacts,
        sensitivity_score=sensitivity_score
    )

def compare_scenarios(self, scenario_ids: List[str] = None) -> pd.DataFrame:
    """Compare multiple scenarios side by side."""
    if scenario_ids is None:
        scenario_ids = list(self.scenarios.keys())

    data = []
    for sid in scenario_ids:
        if sid not in self.results:
            scenario = self.scenarios[sid]
            self.evaluate_scenario(scenario)

        result = self.results[sid]
        data.append({
            'Scenario': result.scenario_name,
            'Total Cost': f"${result.total_cost:,.0f}",
            'Duration (days)': result.total_duration,
            'Cost Change': f"{result.comparison_to_base['cost_change_pct']:+.1f}%",
            'Duration Change': f"{result.comparison_to_base['duration_change_days']:+.0f} days",
            'Risk Score': f"{result.risk_score:.0f}/100",
            'Cost/SF': f"${result.key_metrics['cost_per_sf']:.2f}",
        })

    return pd.DataFrame(data)

def find_optimal_scenario(self, objective: str = 'cost',
                          constraints: Dict[str, tuple] = None) -> Scenario:
    """Find optimal scenario given objective and constraints."""
    valid_results = []

    for sid, result in self.results.items():
        # Check constraints
        if constraints:
            meets_constraints = True
            if 'max_cost' in constraints and result.total_cost > constraints['max_cost']:
                meets_constraints = False
            if 'max_duration' in constraints and result.total_duration > constraints['max_duration']:
                meets_constraints = False
            if 'max_risk' in constraints and result.risk_score > constraints['max_risk']:
                meets_constraints = False

            if not meets_constraints:
                continue

        valid_results.append((sid, result))

    if not valid_results:
        return None

    # Sort by objective
    if objective == 'cost':
        valid_results.sort(key=lambda x: x[1].total_cost)
    elif objective == 'duration':
        valid_results.sort(key=lambda x: x[1].total_duration)
    elif objective == 'risk':
        valid_results.sort(key=lambda x: x[1].risk_score)
    elif objective == 'balanced':
        # Normalize and combine metrics
        valid_results.sort(key=lambda x: (
            x[1].total_cost / 1000000 +
            x[1].total_duration / 365 +
            x[1].risk_score / 100
        ))

    return self.scenarios[valid_results[0][0]]

def generate_report(self) -> str:
    """Generate scenario comparison report."""
    lines = ["# Scenario Analysis Report", ""]
    lines.append(f"**Project:** {self.base_project.get('name', 'Project')}")
    lines.append(f"**Generated:** {datetime.now().strftime('%Y-%m-%d %H:%M')}")
    lines.append(f"**Scenarios Analyzed:** {len(self.scenarios)}")
    lines.append("")

    # Comparison table
    lines.append("## Scenario Comparison")
    comparison = self.compare_scenarios()
    lines.append(comparison.to_markdown(index=False))
    lines.append("")

    # Best scenarios
    lines.append("## Optimal Scenarios")

    best_cost = self.find_optimal_scenario('cost')
    if best_cost:
        lines.append(f"- **Lowest Cost:** {best_cost.name}")

    best_duration = self.find_optimal_scenario('duration')
    if best_duration:
        lines.append(f"- **Shortest Duration:** {best_duration.name}")

    best_balanced = self.find_optimal_scenario('balanced')
    if best_balanced:
        lines.append(f"- **Best Balanced:** {best_balanced.name}")

    lines.append("")

    # Detailed results
    lines.append("## Detailed Results")
    for sid, result in self.results.items():
        lines.append(f"\n### {result.scenario_name}")
        lines.append(f"- **Cost:** ${result.total_cost:,.0f} ({result.comparison_to_base['cost_change_pct']:+.1f}%)")
        lines.append(f"- **Duration:** {result.total_duration} days ({result.comparison_to_base['duration_change_days']:+.0f})")
        lines.append(f"- **Risk Score:** {result.risk_score:.0f}/100")

        if result.warnings:
            lines.append("- **Warnings:**")
            for w in result.warnings:
                lines.append(f"  - ⚠️ {w}")

    return "\n".join(lines)

Quick Start

Define base project

base_project = { 'name': 'Office Building', 'base_cost': 5000000, 'base_duration': 365, 'gross_area': 50000 }

Initialize planner

planner = ConstructionScenarioPlanner(base_project)

Create scenarios

baseline = planner.create_scenario( "Baseline", "Standard approach with default parameters", {} )

accelerated = planner.create_scenario( "Accelerated Schedule", "Faster completion with overtime and larger crew", { 'overtime_percentage': 25, 'crew_size': 15, 'work_days_per_week': 6 } )

cost_optimized = planner.create_scenario( "Cost Optimized", "Lower cost with reduced contingency and smaller crew", { 'contingency_percentage': 7, 'crew_size': 8, 'subcontractor_markup': 12 } )

Evaluate all scenarios

for scenario in planner.scenarios.values(): result = planner.evaluate_scenario(scenario) print(f"{result.scenario_name}: ${result.total_cost:,.0f}, {result.total_duration} days")

Compare scenarios

comparison = planner.compare_scenarios() print(comparison)

Run sensitivity analysis

sensitivity = planner.run_sensitivity_analysis('material_escalation') print(f"Material escalation sensitivity: {sensitivity.sensitivity_score:.1f}")

Generate report

report = planner.generate_report() print(report)

Dependencies

pip install pandas numpy