scenario-planner

建設プロジェクトのシナリオプランナー

概要

さまざまなプロジェクトシナリオをモデル化し、コスト、スケジュール、リソースへの影響を把握します。代替案を比較し、意思決定を最適化して、緊急事態に備えます。

ビジネスケース

建設に関する意思決定では、トレードオフを理解する必要があります。

• 設計の代替案：どの選択肢が最も費用対効果に優れていますか？
• スケジュールの短縮：工期を短縮すると、どのくらいのコストがかかりますか？
• リソースの選択肢：自社で対応する場合と、下請け業者に委託する場合のどちらがよいですか？
• リスクシナリオ：資材の価格が20%上昇した場合はどうなりますか？

技術的な実装

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)

クイックスタート

# 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)

依存関係

pip install pandas numpy