tools-only Delay Analysis
Overview
Analyze construction schedule delays for project recovery and claims. Perform time impact analysis (TIA), identify concurrent delays, calculate delay damages, and prepare documentation for dispute resolution.
"Proper delay analysis is essential for fair resolution of construction disputes" — DDC Community
Delay Analysis Methods
┌─────────────────────────────────────────────────────────────────┐
│ DELAY ANALYSIS METHODS │
├─────────────────────────────────────────────────────────────────┤
│ │
│ As-Planned vs As-Built │ Time Impact Analysis (TIA) │
│ ───────────────────── │ ──────────────────────────── │
│ Compare original to │ Insert delay events into │
│ actual schedule │ schedule to measure impact │
│ │ │
│ Windows Analysis │ Collapsed As-Built │
│ ──────────────── │ ───────────────── │
│ Divide project into │ Remove delays from as-built │
│ time periods │ to find "but-for" completion │
│ │
└─────────────────────────────────────────────────────────────────┘
Technical Implementation
from dataclasses import dataclass, field
from typing import List, Dict, Optional, Tuple
from datetime import datetime, timedelta
from enum import Enum
from collections import defaultdict
class DelayType(Enum):
EXCUSABLE_COMPENSABLE = "excusable_compensable" # Owner caused - time + money
EXCUSABLE_NON_COMPENSABLE = "excusable_non_compensable" # Neither party - time only
NON_EXCUSABLE = "non_excusable" # Contractor caused - no relief
CONCURRENT = "concurrent" # Both parties - complex
class DelayCause(Enum):
OWNER_CHANGE = "owner_change"
LATE_INFORMATION = "late_information"
DIFFERING_CONDITIONS = "differing_conditions"
PERMIT_DELAY = "permit_delay"
WEATHER = "weather"
LABOR_SHORTAGE = "labor_shortage"
MATERIAL_DELAY = "material_delay"
SUBCONTRACTOR = "subcontractor"
COORDINATION = "coordination"
ACCESS = "access"
FORCE_MAJEURE = "force_majeure"
@dataclass
class DelayEvent:
id: str
description: str
cause: DelayCause
delay_type: DelayType
start_date: datetime
end_date: datetime
affected_activities: List[str]
responsible_party: str
documented: bool = True
supporting_docs: List[str] = field(default_factory=list)
calculated_impact: int = 0 # days
concurrent_with: List[str] = field(default_factory=list)
@dataclass
class ScheduleVersion:
version_id: str
version_type: str # baseline, update, as-built
data_date: datetime
completion_date: datetime
activities: Dict[str, Dict] # activity_id -> {start, finish, duration}
@dataclass
class WindowPeriod:
window_id: str
start_date: datetime
end_date: datetime
planned_progress: float
actual_progress: float
delay_days: int
delay_events: List[str]
responsible_parties: Dict[str, int] # party -> delay days
@dataclass
class DelayAnalysisReport:
project_name: str
analysis_date: datetime
original_completion: datetime
actual_completion: datetime
total_delay: int
excusable_delay: int
non_excusable_delay: int
concurrent_delay: int
delay_events: List[DelayEvent]
delay_by_cause: Dict[str, int]
delay_by_party: Dict[str, int]
recommended_extension: int
potential_damages: float
class DelayAnalyzer:
"""Analyze construction schedule delays."""
# Daily delay costs by project size
DEFAULT_DAILY_COSTS = {
"small": 5000, # < $10M
"medium": 15000, # $10M - $50M
"large": 40000, # $50M - $200M
"mega": 100000 # > $200M
}
def __init__(self, project_name: str, contract_completion: datetime):
self.project_name = project_name
self.contract_completion = contract_completion
self.delay_events: Dict[str, DelayEvent] = {}
self.schedule_versions: Dict[str, ScheduleVersion] = {}
self.window_periods: List[WindowPeriod] = []
self.daily_cost = self.DEFAULT_DAILY_COSTS["medium"]
def set_daily_delay_cost(self, cost: float):
"""Set daily delay cost for damages calculation."""
self.daily_cost = cost
def add_schedule_version(self, version_id: str, version_type: str,
data_date: datetime, completion_date: datetime,
activities: Dict[str, Dict]) -> ScheduleVersion:
"""Add schedule version for analysis."""
version = ScheduleVersion(
version_id=version_id,
version_type=version_type,
data_date=data_date,
completion_date=completion_date,
activities=activities
)
self.schedule_versions[version_id] = version
return version
def add_delay_event(self, id: str, description: str,
cause: DelayCause, delay_type: DelayType,
start_date: datetime, end_date: datetime,
affected_activities: List[str],
responsible_party: str,
supporting_docs: List[str] = None) -> DelayEvent:
"""Add delay event for analysis."""
event = DelayEvent(
id=id,
description=description,
cause=cause,
delay_type=delay_type,
start_date=start_date,
end_date=end_date,
affected_activities=affected_activities,
responsible_party=responsible_party,
supporting_docs=supporting_docs or []
)
self.delay_events[id] = event
return event
def perform_as_planned_vs_as_built(self) -> Dict:
"""Perform As-Planned vs As-Built analysis."""
baseline = self.schedule_versions.get("baseline")
as_built = self.schedule_versions.get("as_built")
if not baseline or not as_built:
raise ValueError("Need baseline and as-built schedules")
total_delay = (as_built.completion_date - baseline.completion_date).days
# Analyze each activity
activity_delays = []
for act_id, baseline_act in baseline.activities.items():
if act_id in as_built.activities:
as_built_act = as_built.activities[act_id]
baseline_finish = baseline_act['finish']
actual_finish = as_built_act['finish']
if isinstance(baseline_finish, str):
baseline_finish = datetime.fromisoformat(baseline_finish)
if isinstance(actual_finish, str):
actual_finish = datetime.fromisoformat(actual_finish)
delay = (actual_finish - baseline_finish).days
if delay > 0:
activity_delays.append({
'activity_id': act_id,
'planned_finish': baseline_finish,
'actual_finish': actual_finish,
'delay_days': delay
})
return {
'method': 'As-Planned vs As-Built',
'baseline_completion': baseline.completion_date,
'actual_completion': as_built.completion_date,
'total_delay': total_delay,
'activity_delays': sorted(activity_delays, key=lambda x: -x['delay_days'])
}
def perform_time_impact_analysis(self, delay_event_id: str) -> Dict:
"""Perform Time Impact Analysis for specific delay event."""
if delay_event_id not in self.delay_events:
raise ValueError(f"Delay event {delay_event_id} not found")
event = self.delay_events[delay_event_id]
# Find schedule version just before delay
pre_delay_schedule = None
for version in sorted(self.schedule_versions.values(),
key=lambda v: v.data_date, reverse=True):
if version.data_date < event.start_date:
pre_delay_schedule = version
break
if not pre_delay_schedule:
pre_delay_schedule = self.schedule_versions.get("baseline")
if not pre_delay_schedule:
raise ValueError("No pre-delay schedule found")
# Calculate impact
original_completion = pre_delay_schedule.completion_date
delay_duration = (event.end_date - event.start_date).days
# Check if delay is on critical path
critical_impact = False
for act_id in event.affected_activities:
if act_id in pre_delay_schedule.activities:
act = pre_delay_schedule.activities[act_id]
if act.get('is_critical', False):
critical_impact = True
break
if critical_impact:
impact_days = delay_duration
new_completion = original_completion + timedelta(days=delay_duration)
else:
# Need to check float
impact_days = max(0, delay_duration - 5) # Simplified - assume 5 days float
new_completion = original_completion + timedelta(days=impact_days)
event.calculated_impact = impact_days
return {
'method': 'Time Impact Analysis',
'delay_event': event.id,
'delay_description': event.description,
'delay_duration': delay_duration,
'critical_path_impact': critical_impact,
'schedule_impact_days': impact_days,
'original_completion': original_completion,
'impacted_completion': new_completion,
'delay_type': event.delay_type.value,
'responsible_party': event.responsible_party
}
def identify_concurrent_delays(self) -> List[Tuple[str, str, int]]:
"""Identify concurrent delay events."""
concurrent = []
events = list(self.delay_events.values())
for i, event1 in enumerate(events):
for event2 in events[i+1:]:
# Check for overlap
overlap_start = max(event1.start_date, event2.start_date)
overlap_end = min(event1.end_date, event2.end_date)
if overlap_start < overlap_end:
overlap_days = (overlap_end - overlap_start).days
concurrent.append((event1.id, event2.id, overlap_days))
event1.concurrent_with.append(event2.id)
event2.concurrent_with.append(event1.id)
return concurrent
def perform_windows_analysis(self, window_days: int = 30) -> List[WindowPeriod]:
"""Perform windows analysis by dividing project into periods."""
baseline = self.schedule_versions.get("baseline")
as_built = self.schedule_versions.get("as_built")
if not baseline or not as_built:
raise ValueError("Need baseline and as-built schedules")
windows = []
current_start = baseline.data_date
window_num = 1
while current_start < as_built.completion_date:
window_end = min(
current_start + timedelta(days=window_days),
as_built.completion_date
)
# Find delay events in this window
window_events = [
e.id for e in self.delay_events.values()
if e.start_date < window_end and e.end_date > current_start
]
# Calculate delay by party
party_delays = defaultdict(int)
for event_id in window_events:
event = self.delay_events[event_id]
overlap_start = max(event.start_date, current_start)
overlap_end = min(event.end_date, window_end)
days = (overlap_end - overlap_start).days
party_delays[event.responsible_party] += days
window = WindowPeriod(
window_id=f"W{window_num:02d}",
start_date=current_start,
end_date=window_end,
planned_progress=0.0, # Would calculate from schedule
actual_progress=0.0,
delay_days=sum(party_delays.values()),
delay_events=window_events,
responsible_parties=dict(party_delays)
)
windows.append(window)
current_start = window_end
window_num += 1
self.window_periods = windows
return windows
def calculate_delay_damages(self) -> Dict:
"""Calculate potential delay damages."""
# Summarize delays by type
excusable_compensable = 0
excusable_non_compensable = 0
non_excusable = 0
for event in self.delay_events.values():
impact = event.calculated_impact or (event.end_date - event.start_date).days
# Adjust for concurrency
if event.concurrent_with:
impact = impact // 2 # Simplified concurrency handling
if event.delay_type == DelayType.EXCUSABLE_COMPENSABLE:
excusable_compensable += impact
elif event.delay_type == DelayType.EXCUSABLE_NON_COMPENSABLE:
excusable_non_compensable += impact
elif event.delay_type == DelayType.NON_EXCUSABLE:
non_excusable += impact
# Calculate damages
contractor_damages = excusable_compensable * self.daily_cost
owner_ld = non_excusable * self.daily_cost
return {
'excusable_compensable_days': excusable_compensable,
'excusable_non_compensable_days': excusable_non_compensable,
'non_excusable_days': non_excusable,
'recommended_time_extension': excusable_compensable + excusable_non_compensable,
'contractor_delay_damages': contractor_damages,
'owner_liquidated_damages': owner_ld,
'daily_rate_used': self.daily_cost
}
def generate_analysis_report(self, actual_completion: datetime) -> DelayAnalysisReport:
"""Generate comprehensive delay analysis report."""
total_delay = (actual_completion - self.contract_completion).days
# Categorize delays
delay_by_cause = defaultdict(int)
delay_by_party = defaultdict(int)
excusable = 0
non_excusable = 0
concurrent = 0
for event in self.delay_events.values():
impact = event.calculated_impact or (event.end_date - event.start_date).days
delay_by_cause[event.cause.value] += impact
delay_by_party[event.responsible_party] += impact
if event.concurrent_with:
concurrent += impact // 2
elif event.delay_type in [DelayType.EXCUSABLE_COMPENSABLE,
DelayType.EXCUSABLE_NON_COMPENSABLE]:
excusable += impact
else:
non_excusable += impact
damages = self.calculate_delay_damages()
return DelayAnalysisReport(
project_name=self.project_name,
analysis_date=datetime.now(),
original_completion=self.contract_completion,
actual_completion=actual_completion,
total_delay=total_delay,
excusable_delay=excusable,
non_excusable_delay=non_excusable,
concurrent_delay=concurrent,
delay_events=list(self.delay_events.values()),
delay_by_cause=dict(delay_by_cause),
delay_by_party=dict(delay_by_party),
recommended_extension=damages['recommended_time_extension'],
potential_damages=damages['contractor_delay_damages']
)
def generate_report_markdown(self, report: DelayAnalysisReport) -> str:
"""Generate markdown report."""
lines = [
"# Delay Analysis Report",
"",
f"**Project:** {report.project_name}",
f"**Analysis Date:** {report.analysis_date.strftime('%Y-%m-%d')}",
"",
"## Schedule Summary",
"",
f"| Milestone | Date |",
f"|-----------|------|",
f"| Contract Completion | {report.original_completion.strftime('%Y-%m-%d')} |",
f"| Actual Completion | {report.actual_completion.strftime('%Y-%m-%d')} |",
f"| **Total Delay** | **{report.total_delay} days** |",
"",
"## Delay Classification",
"",
f"| Category | Days |",
f"|----------|------|",
f"| Excusable Delay | {report.excusable_delay} |",
f"| Non-Excusable Delay | {report.non_excusable_delay} |",
f"| Concurrent Delay | {report.concurrent_delay} |",
"",
"## Delay by Cause",
""
]
for cause, days in sorted(report.delay_by_cause.items(), key=lambda x: -x[1]):
lines.append(f"- **{cause}**: {days} days")
lines.extend([
"",
"## Delay by Responsible Party",
""
])
for party, days in sorted(report.delay_by_party.items(), key=lambda x: -x[1]):
lines.append(f"- **{party}**: {days} days")
lines.extend([
"",
"## Recommendations",
"",
f"- **Recommended Time Extension:** {report.recommended_extension} days",
f"- **Potential Delay Damages:** ${report.potential_damages:,.0f}",
""
])
return "\n".join(lines)
Quick Start
from datetime import datetime, timedelta
# Initialize analyzer
analyzer = DelayAnalyzer(
"Office Tower",
contract_completion=datetime(2024, 12, 31)
)
# Add schedule versions
analyzer.add_schedule_version(
"baseline", "baseline",
datetime(2024, 1, 1),
datetime(2024, 12, 31),
activities={"A100": {"finish": datetime(2024, 6, 30), "is_critical": True}}
)
analyzer.add_schedule_version(
"as_built", "as_built",
datetime(2025, 3, 15),
datetime(2025, 3, 15),
activities={"A100": {"finish": datetime(2024, 8, 15)}}
)
# Add delay events
analyzer.add_delay_event(
"D001",
"Owner-directed design change to HVAC system",
DelayCause.OWNER_CHANGE,
DelayType.EXCUSABLE_COMPENSABLE,
datetime(2024, 4, 1),
datetime(2024, 5, 15),
["A100", "A101"],
"Owner"
)
analyzer.add_delay_event(
"D002",
"Unexpected rock encountered in excavation",
DelayCause.DIFFERING_CONDITIONS,
DelayType.EXCUSABLE_COMPENSABLE,
datetime(2024, 3, 15),
datetime(2024, 4, 30),
["A050"],
"Owner"
)
# Perform analyses
tia = analyzer.perform_time_impact_analysis("D001")
print(f"TIA Impact: {tia['schedule_impact_days']} days")
concurrent = analyzer.identify_concurrent_delays()
print(f"Concurrent delays found: {len(concurrent)}")
# Generate report
report = analyzer.generate_analysis_report(datetime(2025, 3, 15))
print(analyzer.generate_report_markdown(report))
Requirements
pip install (no external dependencies)