schedule-cost-link

What this skill does

# Schedule-Cost Linker

## Business Case

### Problem Statement
Integrating schedule and cost requires:
- Linking activities to budget items
- Creating cost-loaded schedules
- Generating cash flow forecasts
- Tracking earned value metrics

### Solution
Systematic linkage between schedule activities and cost data to enable integrated project control.

## Technical Implementation

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


class LoadingMethod(Enum):
    UNIFORM = "uniform"          # Even distribution
    FRONT_LOADED = "front_loaded"
    BACK_LOADED = "back_loaded"
    BELL_CURVE = "bell_curve"


@dataclass
class ScheduleActivity:
    activity_id: str
    name: str
    start_date: date
    finish_date: date
    duration: int
    percent_complete: float = 0


@dataclass
class CostItem:
    cost_code: str
    description: str
    budgeted_cost: float
    labor_cost: float
    material_cost: float
    equipment_cost: float


@dataclass
class ActivityCostLink:
    activity_id: str
    cost_code: str
    budgeted_cost: float
    loading_method: LoadingMethod


@dataclass
class EarnedValueMetrics:
    data_date: date
    bcws: float  # Budgeted Cost of Work Scheduled (PV)
    bcwp: float  # Budgeted Cost of Work Performed (EV)
    acwp: float  # Actual Cost of Work Performed (AC)
    sv: float    # Schedule Variance
    cv: float    # Cost Variance
    spi: float   # Schedule Performance Index
    cpi: float   # Cost Performance Index
    eac: float   # Estimate at Completion
    etc: float   # Estimate to Complete
    vac: float   # Variance at Completion


class ScheduleCostLinker:
    """Link schedule activities to cost items."""

    def __init__(self, project_name: str, budget_at_completion: float):
        self.project_name = project_name
        self.bac = budget_at_completion
        self.activities: Dict[str, ScheduleActivity] = {}
        self.cost_items: Dict[str, CostItem] = {}
        self.links: List[ActivityCostLink] = []
        self.actual_costs: Dict[str, float] = {}  # activity_id -> actual cost

    def add_activity(self,
                     activity_id: str,
                     name: str,
                     start_date: date,
                     finish_date: date,
                     percent_complete: float = 0):
        """Add schedule activity."""

        duration = (finish_date - start_date).days + 1

        self.activities[activity_id] = ScheduleActivity(
            activity_id=activity_id,
            name=name,
            start_date=start_date,
            finish_date=finish_date,
            duration=duration,
            percent_complete=percent_complete
        )

    def add_cost_item(self,
                      cost_code: str,
                      description: str,
                      budgeted_cost: float,
                      labor_pct: float = 0.4,
                      material_pct: float = 0.5,
                      equipment_pct: float = 0.1):
        """Add cost item."""

        self.cost_items[cost_code] = CostItem(
            cost_code=cost_code,
            description=description,
            budgeted_cost=budgeted_cost,
            labor_cost=budgeted_cost * labor_pct,
            material_cost=budgeted_cost * material_pct,
            equipment_cost=budgeted_cost * equipment_pct
        )

    def link_activity_cost(self,
                           activity_id: str,
                           cost_code: str,
                           loading_method: LoadingMethod = LoadingMethod.UNIFORM):
        """Link activity to cost item."""

        if activity_id not in self.activities:
            return

        cost_item = self.cost_items.get(cost_code)
        budgeted = cost_item.budgeted_cost if cost_item else 0

        self.links.append(ActivityCostLink(
            activity_id=activity_id,
            cost_code=cost_code,
            budgeted_cost=budgeted,
            loading_method=loading_method
        ))

    def record_actual_cost(self, activity_id: str, actual_cost: float):
        """Record actual cost for activity."""
        self.actual_costs[activity_id] = actual_cost

    def _distribute_cost(self,
                          cost: float,
                          start_date: date,
                          duration: int,
                          method: LoadingMethod) -> Dict[date, float]:
        """Distribute cost over activity duration."""

        daily_costs = {}

        if duration <= 0:
            return {start_date: cost}

        if method == LoadingMethod.UNIFORM:
            daily = cost / duration
            for i in range(duration):
                daily_costs[start_date + timedelta(days=i)] = daily

        elif method == LoadingMethod.FRONT_LOADED:
            total_weight = sum(range(duration, 0, -1))
            for i in range(duration):
                weight = (duration - i) / total_weight
                daily_costs[start_date + timedelta(days=i)] = cost * weight

        elif method == LoadingMethod.BACK_LOADED:
            total_weight = sum(range(1, duration + 1))
            for i in range(duration):
                weight = (i + 1) / total_weight
                daily_costs[start_date + timedelta(days=i)] = cost * weight

        elif method == LoadingMethod.BELL_CURVE:
            # Simplified bell curve
            mid = duration / 2
            for i in range(duration):
                distance = abs(i - mid)
                weight = 1 - (distance / mid) * 0.5
                daily_costs[start_date + timedelta(days=i)] = cost * weight / duration

        return daily_costs

    def generate_cost_loaded_schedule(self) -> pd.DataFrame:
        """Generate cost-loaded schedule."""

        data = []

        for link in self.links:
            activity = self.activities.get(link.activity_id)
            cost_item = self.cost_items.get(link.cost_code)

            if activity and cost_item:
                data.append({
                    'Activity ID': activity.activity_id,
                    'Activity Name': activity.name,
                    'Cost Code': link.cost_code,
                    'Description': cost_item.description,
                    'Start': activity.start_date,
                    'Finish': activity.finish_date,
                    'Duration': activity.duration,
                    'Budget': link.budgeted_cost,
                    '% Complete': activity.percent_complete,
                    'Earned Value': link.budgeted_cost * activity.percent_complete / 100,
                    'Loading': link.loading_method.value
                })

        return pd.DataFrame(data)

    def generate_cash_flow(self,
                           project_start: date = None,
                           project_end: date = None) -> pd.DataFrame:
        """Generate cash flow curve."""

        if not self.links:
            return pd.DataFrame()

        # Get date range
        if project_start is None:
            project_start = min(self.activities[l.activity_id].start_date for l in self.links)
        if project_end is None:
            project_end = max(self.activities[l.activity_id].finish_date for l in self.links)

        # Aggregate daily costs
        daily_totals = defaultdict(float)

        for link in self.links:
            activity = self.activities.get(link.activity_id)
            if not activity:
                continue

            daily_costs = self._distribute_cost(
                link.budgeted_cost,
                activity.start_date,
                activity.duration,
                link.loading_method
            )

            for day, cost in daily_costs.items():
                daily_totals[day] += cost

        # Build cash flow data
        data = []
        cumulative = 0
        current = project_start

        while current <= proje