change-order-analysis

変更注文の分析

概要

このスキルは、建設プロジェクト向けの機械学習ベースの変更注文分析を実装します。変更注文のコストを予測し、種類を分類し、過去のデータからパターンを特定して、承認プロセスを効率化します。

機能：

• 変更注文の分類
• コストへの影響の予測
• スケジュールへの影響の分析
• パターンの特定
• 根本原因の分析
• 承認ワークフローの最適化

クイックスタート

from dataclasses import dataclass, field
from datetime import date, datetime
from typing import List, Dict, Optional
from enum import Enum

class ChangeOrderType(Enum):
    DESIGN_CHANGE = "design_change"
    OWNER_REQUEST = "owner_request"
    FIELD_CONDITION = "field_condition"
    CODE_COMPLIANCE = "code_compliance"
    VALUE_ENGINEERING = "value_engineering"
    ERROR_OMISSION = "error_omission"
    SCOPE_CHANGE = "scope_change"

class ChangeOrderStatus(Enum):
    DRAFT = "draft"
    SUBMITTED = "submitted"
    UNDER_REVIEW = "under_review"
    APPROVED = "approved"
    REJECTED = "rejected"
    IMPLEMENTED = "implemented"

@dataclass
class ChangeOrder:
    co_number: str
    title: str
    description: str
    co_type: ChangeOrderType
    status: ChangeOrderStatus
    submitted_date: date
    requested_by: str
    cost_impact: float
    schedule_impact_days: int
    affected_elements: List[str] = field(default_factory=list)

def classify_change_order(description: str) -> ChangeOrderType:
    """Simple rule-based classification"""
    description_lower = description.lower()

    if any(word in description_lower for word in ['design', 'drawing', 'specification']):
        return ChangeOrderType.DESIGN_CHANGE
    elif any(word in description_lower for word in ['owner', 'client', 'request']):
        return ChangeOrderType.OWNER_REQUEST
    elif any(word in description_lower for word in ['site', 'field', 'condition', 'unforeseen']):
        return ChangeOrderType.FIELD_CONDITION
    elif any(word in description_lower for word in ['code', 'regulation', 'compliance']):
        return ChangeOrderType.CODE_COMPLIANCE
    elif any(word in description_lower for word in ['value', 'alternative', 'savings']):
        return ChangeOrderType.VALUE_ENGINEERING
    elif any(word in description_lower for word in ['error', 'omission', 'mistake']):
        return ChangeOrderType.ERROR_OMISSION
    else:
        return ChangeOrderType.SCOPE_CHANGE

# Example
co = ChangeOrder(
    co_number="CO-001",
    title="Additional structural reinforcement",
    description="Site conditions revealed weaker soil requiring additional foundation reinforcement",
    co_type=classify_change_order("Site conditions revealed weaker soil"),
    status=ChangeOrderStatus.SUBMITTED,
    submitted_date=date.today(),
    requested_by="Site Engineer",
    cost_impact=50000,
    schedule_impact_days=5
)
print(f"CO Type: {co.co_type.value}")

包括的な変更注文システム

変更注文の管理

from dataclasses import dataclass, field
from datetime import date, datetime, timedelta
from typing import List, Dict, Optional, Tuple
from enum import Enum
import pandas as pd
import numpy as np

class ImpactSeverity(Enum):
    MINOR = "minor"  # < 1% cost, < 1 week schedule
    MODERATE = "moderate"  # 1-5% cost, 1-4 weeks schedule
    MAJOR = "major"  # 5-10% cost, 1-3 months schedule
    CRITICAL = "critical"  # > 10% cost, > 3 months schedule

@dataclass
class CostBreakdown:
    labor: float = 0
    materials: float = 0
    equipment: float = 0
    subcontractor: float = 0
    overhead: float = 0
    profit: float = 0

    @property
    def total(self) -> float:
        return self.labor + self.materials + self.equipment + self.subcontractor + self.overhead + self.profit

@dataclass
class ScheduleImpact:
    direct_days: int
    ripple_days: int
    critical_path_affected: bool
    affected_activities: List[str] = field(default_factory=list)

    @property
    def total_days(self) -> int:
        return self.direct_days + self.ripple_days

@dataclass
class ChangeOrderDetail:
    co_id: str
    co_number: str
    title: str
    description: str
    justification: str

    # Classification
    co_type: ChangeOrderType
    initiated_by: str  # owner, contractor, designer, etc.
    responsibility: str  # who pays

    # Status
    status: ChangeOrderStatus
    submitted_date: date
    approved_date: Optional[date] = None
    implemented_date: Optional[date] = None

    # Impact
    cost_breakdown: CostBreakdown = field(default_factory=CostBreakdown)
    schedule_impact: ScheduleImpact = None
    severity: ImpactSeverity = ImpactSeverity.MINOR

    # Affected scope
    affected_elements: List[str] = field(default_factory=list)
    affected_drawings: List[str] = field(default_factory=list)
    affected_specs: List[str] = field(default_factory=list)

    # Supporting documents
    attachments: List[str] = field(default_factory=list)
    related_rfis: List[str] = field(default_factory=list)
    related_cos: List[str] = field(default_factory=list)

    # Approval
    approvals: List[Dict] = field(default_factory=list)
    comments: List[Dict] = field(default_factory=list)

class ChangeOrderManager:
    """Manage project change orders"""

    def __init__(self, project_id: str, contract_value: float):
        self.project_id = project_id
        self.contract_value = contract_value
        self.change_orders: Dict[str, ChangeOrderDetail] = {}
        self.co_counter = 0

    def create_change_order(self, title: str, description: str,
                           co_type: ChangeOrderType,
                           initiated_by: str) -> ChangeOrderDetail:
        """Create new change order"""
        self.co_counter += 1
        co_id = f"CO-{self.project_id}-{self.co_counter:04d}"

        co = ChangeOrderDetail(
            co_id=co_id,
            co_number=f"CO-{self.co_counter:04d}",
            title=title,
            description=description,
            justification="",
            co_type=co_type,
            initiated_by=initiated_by,
            responsibility="TBD",
            status=ChangeOrderStatus.DRAFT,
            submitted_date=date.today()
        )

        self.change_orders[co_id] = co
        return co

    def update_cost(self, co_id: str, cost_breakdown: CostBreakdown):
        """Update change order cost"""
        co = self.change_orders.get(co_id)
        if co:
            co.cost_breakdown = cost_breakdown
            co.severity = self._calculate_severity(co)

    def update_schedule_impact(self, co_id: str, impact: ScheduleImpact):
        """Update schedule impact"""
        co = self.change_orders.get(co_id)
        if co:
            co.schedule_impact = impact
            co.severity = self._calculate_severity(co)

    def _calculate_severity(self, co: ChangeOrderDetail) -> ImpactSeverity:
        """Calculate change order severity"""
        cost_pct = co.cost_breakdown.total / self.contract_value * 100
        schedule_days = co.schedule_impact.total_days if co.schedule_impact else 0

        if cost_pct > 10 or schedule_days > 90:
            return ImpactSeverity.CRITICAL
        elif cost_pct > 5 or schedule_days > 30:
            return ImpactSeverity.MAJOR
        elif cost_pct > 1 or schedule_days > 7:
            return ImpactSeverity.MODERATE
        else:
            return ImpactSeverity.MINOR

    def submit_for_approval(self, co_id: str):
        """Submit change order for approval"""
        co = self.change_orders.get(co_id)
        if co and co.status == ChangeOrderStatus.DRAFT:
            co.status = ChangeOrderStatus.SUBMITTED
            co.submitted_date = date.today()

    def approve(self, co_id: str, approver: str, comments: str = ""):
        """Approve change order"""
        co = self.change_orders.get(co_id)
        if co:
            co.approvals.append({
                'approver': approver,
                'action': 'approved',
                'date': date.today().isoformat(),
                'comments': comments
            })
            co.status = ChangeOrderStatus.APPROVED
            co.approved_date = date.today()

    def get_summary(self) -> Dict:
        """Get change order summary"""
        if not self.change_orders:
            return {'message': 'No change orders'}

        total_cost = sum(co.cost_breakdown.total for co in self.change_orders.values())
        total_schedule = sum(
            co.schedule_impact.total_days if co.schedule_impact else 0
            for co in self.change_orders.values()
        )

        by_type = {}
        by_status = {}
        by_severity = {}

        for co in self.change_orders.values():
            t = co.co_type.value
            by_type[t] = by_type.get(t, 0) + co.cost_breakdown.total

            s = co.status.value
            by_status[s] = by_status.get(s, 0) + 1

            sev = co.severity.value
            by_severity[sev] = by_severity.get(sev, 0) + 1

        return {
            'total_change_orders': len(self.change_orders),
            'total_cost_impact': total_cost,
            'cost_impact_pct': total_cost / self.contract_value * 100,
            'total_schedule_impact_days': total_schedule,
            'by_type': by_type,
            'by_status': by_status,
            'by_severity': by_severity
        }

ML による分類と予測

from sklearn.ensemble import RandomForestClassifier, GradientBoostingRegressor
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import LabelEncoder
import pandas as pd
import numpy as np
import joblib

class ChangeOrderPredictor:
    """ML-based change order classification and cost prediction"""

    def __init__(self):
        self.type_classifier = None
        self.cost_predictor = None
        self.schedule_predictor = None
        self.vectorizer = TfidfVectorizer(max_features=500, ngram_range=(1, 2))
        self.type_encoder = LabelEncoder()
        self.is_trained = False

    def train(self, historical_data: pd.DataFrame):
        """Train models on historical change order data

        Expected columns:
        - description: text description
        - co_type: change order type
        - cost_impact: cost in dollars
        - schedule_impact: days of delay
        - contract_value: original contract value
        - project_phase: phase when CO was raised
        - affected_elements_count: number of affected elements
        """
        # Prepare text features
        text_features = self.vectorizer.fit_transform(historical_data['description'])

        # Prepare numeric features
        numeric_features = historical_data[[
            'contract_value', 'affected_elements_count'
        ]].values

        # Combine features
        X = np.hstack([text_features.toarray(), numeric_features])

        # Train type classifier
        y_type = self.type_encoder.fit_transform(historical_data['co_type'])
        X_train, X_test, y_train, y_test = train_test_split(X, y_type, test_size=0.2)

        self.type_classifier = RandomForestClassifier(n_estimators=100, random_state=42)
        self.type_classifier.fit(X_train, y_train)

        type_accuracy = self.type_classifier.score(X_test, y_test)

        # Train cost predictor
        y_cost = historical_data['cost_impact'].values
        self.cost_predictor = GradientBoostingRegressor(n_estimators=100, random_state=42)
        self.cost_predictor.fit(X, y_cost)

        # Train schedule predictor
        y_schedule = historical_data['schedule_impact'].values
        self.schedule_predictor = GradientBoostingRegressor(n_estimators=100, random_state=42)
        self.schedule_predictor.fit(X, y_schedule)

        self.is_trained = True

        return {
            'type_classifier_accuracy': type_accuracy,
            'models_trained': True
        }

    def predict(self, description: str, contract_value: float,
               affected_elements_count: int = 1) -> Dict:
        """Predict change order type and impacts"""
        if not self.is_trained:
            return {'error': 'Models not trained'}

        # Prepare features
        text_features = self.vectorizer.transform([description])
        numeric_features = np.array([[contract_value, affected_elements_count]])
        X = np.hstack([text_features.toarray(), numeric_features])

        # Predict type
        type_probs = self.type_classifier.predict_proba(X)[0]
        type_idx = np.argmax(type_probs)
        predicted_type = self.type_encoder.inverse_transform([type_idx])[0]

        # Predict cost
        predicted_cost = self.cost_predictor.predict(X)[0]

        # Predict schedule
        predicted_schedule = self.schedule_predictor.predict(X)[0]

        return {
            'predicted_type': predicted_type,
            'type_confidence': float(type_probs[type_idx]),
            'type_probabilities': {
                self.type_encoder.inverse_transform([i])[0]: float(p)
                for i, p in enumerate(type_probs)
            },
            'predicted_cost': float(max(0, predicted_cost)),
            'predicted_schedule_days': int(max(0, predicted_schedule)),
            'cost_as_pct_contract': float(predicted_cost / contract_value * 100)
        }

    def save_models(self, path: str):
        """Save trained models"""
        joblib.dump({
            'type_classifier': self.type_classifier,
            'cost_predictor': self.cost_predictor,
            'schedule_predictor': self.schedule_predictor,
            'vectorizer': self.vectorizer,
            'type_encoder': self.type_encoder
        }, path)

    def load_models(self, path: str):
        """Load trained models"""
        data = joblib.load(path)
        self.type_classifier = data['type_classifier']
        self.cost_predictor = data['cost_predictor']
        self.schedule_predictor = data['schedule_predictor']
        self.vectorizer = data['vectorizer']
        self.type_encoder = data['type_encoder']
        self.is_trained = True

パターン分析

from collections import defaultdict
from typing import List, Dict
import pandas as pd

class ChangeOrderAnalyzer:
    """Analyze patterns in change orders"""

    def __init__(self, change_orders: List[ChangeOrderDetail]):
        self.cos = change_orders
        self.df = self._to_dataframe()

    def _to_dataframe(self) -> pd.DataFrame:
        """Convert change orders to DataFrame"""
        data = []
        for co in self.cos:
            data.append({
                'co_id': co.co_id,
                'co_type': co.co_type.value,
                'initiated_by': co.initiated_by,
                'cost': co.cost_breakdown.total,
                'schedule_days': co.schedule_impact.total_days if co.schedule_impact else 0,
                'submitted_date': co.submitted_date,
                'affected_elements': len(co.affected_elements),
                'severity': co.severity.value
            })
        return pd.DataFrame(data)

    def analyze_by_type(self) -> Dict:
        """Analyze change orders by type"""
        if self.df.empty:
            return {}

        analysis = {}
        for co_type in self.df['co_type'].unique():
            type_df = self.df[self.df['co_type'] == co_type]
            analysis[co_type] = {
                'count': len(type_df),
                'total_cost': type_df['cost'].sum(),
                'avg_cost': type_df['cost'].mean(),
                'total_schedule_days': type_df['schedule_days'].sum(),
                'avg_schedule_days': type_df['schedule_days'].mean()
            }

        return analysis

    def analyze_trends(self) -> Dict:
        """Analyze trends over time"""
        if self.df.empty:
            return {}

        self.df['month'] = pd.to_datetime(self.df['submitted_date']).dt.to_period('M')

        monthly = self.df.groupby('month').agg({
            'co_id': 'count',
            'cost': 'sum',
            'schedule_days': 'sum'
        }).rename(columns={'co_id': 'count'})

        return {
            'monthly_trend': monthly.to_dict(),
            'peak_month': monthly['count'].idxmax().strftime('%Y-%m'),
            'total_cost_trend': 'increasing' if monthly['cost'].is_monotonic_increasing else
                               'decreasing' if monthly['cost'].is_monotonic_decreasing else 'variable'
        }

    def identify_root_causes(self) -> List[Dict]:
        """Identify common root causes"""
        if self.df.empty:
            return []

        # Analyze by initiator and type combination
        causes = self.df.groupby(['initiated_by', 'co_type']).agg({
            'co_id': 'count',
            'cost': 'sum'
        }).reset_index()

        causes = causes.sort_values('cost', ascending=False)

        return [
            {
                'initiator': row['initiated_by'],
                'type': row['co_type'],
                'frequency': row['co_id'],
                'total_cost': row['cost'],
                'recommendation': self._get_recommendation(row['initiated_by'], row['co_type'])
            }
            for _, row in causes.head(10).iterrows()
        ]

    def _get_recommendation(self, initiator: str, co_type: str) -> str:
        """Generate recommendation based on pattern"""
        recommendations = {
            ('designer', 'design_change'): 'Improve design review process and BIM coordination',
            ('designer', 'error_omission'): 'Implement design quality checks and clash detection',
            ('owner', 'owner_request'): 'Define scope more clearly during planning phase',
            ('owner', 'scope_change'): 'Conduct thorough requirements gathering',
            ('contractor', 'field_condition'): 'Enhance site investigation before construction',
            ('contractor', 'value_engineering'): 'Include VE sessions earlier in project'
        }

        return recommendations.get(
            (initiator.lower(), co_type),
            'Review process and implement preventive measures'
        )

    def calculate_risk_score(self) -> float:
        """Calculate overall change order risk score"""
        if self.df.empty:
            return 0

        # Factors:
        # - Frequency of COs
        # - Cost impact severity
        # - Schedule impact severity
        # - Trend direction

        co_rate = len(self.df) / 12  # COs per month (assuming 12 month project)
        avg_cost_impact = self.df['cost'].mean()
        avg_schedule_impact = self.df['schedule_days'].mean()

        # Normalize and weight
        freq_score = min(1, co_rate / 10) * 30  # Up to 30 points
        cost_score = min(1, avg_cost_impact / 50000) * 40  # Up to 40 points
        schedule_score = min(1, avg_schedule_impact / 30) * 30  # Up to 30 points

        return freq_score + cost_score + schedule_score

    def generate_report(self, output_path: str) -> str:
        """Generate comprehensive analysis report"""
        with pd.ExcelWriter(output_path, engine='openpyxl') as writer:
            # Summary
            summary = pd.DataFrame([{
                'Total COs': len(self.cos),
                'Total Cost Impact': self.df['cost'].sum(),
                'Total Schedule Impact (days)': self.df['schedule_days'].sum(),
                'Risk Score': self.calculate_risk_score()
            }])
            summary.to_excel(writer, sheet_name='Summary', index=False)

            # By type
            pd.DataFrame(self.analyze_by_type()).T.to_excel(
                writer, sheet_name='By_Type'
            )

            # Root causes
            pd.DataFrame(self.identify_root_causes()).to_excel(
                writer, sheet_name='Root_Causes', index=False
            )

            # All COs
            self.df.to_excel(writer, sheet_name='All_COs', index=False)

        return output_path

クイックリファレンス

変更注文の種類	典型的な原因	防止策
設計変更	不完全な設計	BIM による調整、設計レビュー
発注者の要望	要件の変更	明確なスコープの定義
現場の状況	予測できない現場の問題	綿密な現場調査
法令遵守	規制の変更	早期の法令レビュー
バリューエンジニアリング	コスト削減の機会	VE ワークショップ
エラー／記載漏れ	設計上のミス	QA/QC プロセス

リソース

• AIA A201：契約の一般条件
• AGC ConsensusDocs：変更注文の管理
• DDC ウェブサイト：https://datadrivenconstruction.io

次のステップ

• 変更注文書類の処理については document-classification-nlp を参照してください
• プロジェクトのリスク分析については risk-assessment-ml を参照してください
• コスト見積もりについては cost-prediction を参照してください