AI Contractor Matching

Overview

This skill implements AI-powered contractor matching for construction projects. Analyze project requirements against contractor capabilities, track historical performance, and generate recommendations based on multiple criteria.

Matching Criteria:

• Technical capabilities & expertise
• Past performance scores
• Certifications & licenses
• Geographic availability
• Capacity & current workload
• Pricing competitiveness
• Safety records

Quick Start

from dataclasses import dataclass, field
from typing import List, Dict, Optional
from datetime import date
import numpy as np

@dataclass
class Contractor:
    contractor_id: str
    name: str
    specializations: List[str]
    certifications: List[str]
    performance_score: float  # 0-100
    safety_score: float  # 0-100
    regions: List[str]
    capacity_available: float  # 0-100 percentage
    avg_bid_variance: float  # % above/below average

@dataclass
class ProjectRequirement:
    project_id: str
    work_types: List[str]
    required_certs: List[str]
    region: str
    estimated_value: float
    priority: str  # cost, quality, speed, safety

def match_contractors(project: ProjectRequirement,
                     contractors: List[Contractor],
                     top_n: int = 5) -> List[Dict]:
    """Simple contractor matching"""
    scores = []

    for c in contractors:
        # Check basic eligibility
        if project.region not in c.regions:
            continue

        work_match = len(set(project.work_types) & set(c.specializations))
        if work_match == 0:
            continue

        cert_match = len(set(project.required_certs) & set(c.certifications))
        if cert_match < len(project.required_certs):
            continue

        # Calculate score based on priority
        if project.priority == 'quality':
            score = c.performance_score * 0.6 + (100 - abs(c.avg_bid_variance)) * 0.2 + c.capacity_available * 0.2
        elif project.priority == 'cost':
            score = (100 - c.avg_bid_variance) * 0.5 + c.performance_score * 0.3 + c.capacity_available * 0.2
        elif project.priority == 'safety':
            score = c.safety_score * 0.6 + c.performance_score * 0.3 + c.capacity_available * 0.1
        else:  # speed
            score = c.capacity_available * 0.5 + c.performance_score * 0.3 + c.safety_score * 0.2

        scores.append({
            'contractor': c,
            'score': score,
            'work_match': work_match / len(project.work_types),
            'cert_match': cert_match / len(project.required_certs) if project.required_certs else 1.0
        })

    # Sort and return top matches
    scores.sort(key=lambda x: x['score'], reverse=True)
    return scores[:top_n]

# Example
contractors = [
    Contractor("C001", "ABC Builders", ["concrete", "structural"], ["ISO9001", "OHSAS18001"],
              85, 90, ["Moscow", "SPB"], 60, -5),
    Contractor("C002", "XYZ Construction", ["concrete", "finishing"], ["ISO9001"],
              78, 85, ["Moscow"], 80, 10),
]

project = ProjectRequirement("P001", ["concrete"], ["ISO9001"], "Moscow", 1000000, "quality")
matches = match_contractors(project, contractors)
for m in matches:
    print(f"{m['contractor'].name}: Score {m['score']:.1f}")

Comprehensive Matching System

Contractor Profile Management

from dataclasses import dataclass, field
from typing import List, Dict, Optional, Tuple
from datetime import date, datetime
from enum import Enum
import numpy as np
from sklearn.preprocessing import MinMaxScaler

class ContractorSize(Enum):
    MICRO = "micro"  # < 10 employees
    SMALL = "small"  # 10-50 employees
    MEDIUM = "medium"  # 50-250 employees
    LARGE = "large"  # > 250 employees

class WorkCategory(Enum):
    GENERAL = "general_contractor"
    CONCRETE = "concrete"
    STRUCTURAL_STEEL = "structural_steel"
    MEP = "mep"
    ELECTRICAL = "electrical"
    PLUMBING = "plumbing"
    HVAC = "hvac"
    FINISHING = "finishing"
    FACADE = "facade"
    ROOFING = "roofing"
    EXCAVATION = "excavation"
    FOUNDATION = "foundation"
    LANDSCAPING = "landscaping"
    DEMOLITION = "demolition"

@dataclass
class ProjectReference:
    project_name: str
    client: str
    value: float
    completion_date: date
    work_type: str
    performance_rating: float  # 1-5
    on_time: bool
    on_budget: bool
    client_reference_available: bool

@dataclass
class ContractorProfile:
    contractor_id: str
    company_name: str
    legal_name: str
    registration_number: str
    size: ContractorSize
    founded_year: int
    employees_count: int

    # Capabilities
    specializations: List[WorkCategory]
    equipment_owned: List[str]
    max_project_value: float
    min_project_value: float

    # Certifications
    certifications: List[Dict]  # {name, issuer, valid_until}
    licenses: List[Dict]  # {type, number, region, valid_until}

    # Performance
    completed_projects: int
    active_projects: int
    references: List[ProjectReference] = field(default_factory=list)

    # Safety
    safety_certifications: List[str] = field(default_factory=list)
    incident_rate: float = 0.0  # incidents per 1000 work hours
    fatality_count: int = 0
    lost_time_incidents: int = 0

    # Financial
    annual_revenue: float = 0
    credit_rating: str = ""
    insurance_coverage: float = 0
    bonding_capacity: float = 0

    # Geographic
    headquarters_region: str = ""
    operating_regions: List[str] = field(default_factory=list)
    willing_to_travel: bool = False

    # Current status
    current_workload_pct: float = 0  # 0-100
    earliest_availability: Optional[date] = None

    # Pricing
    historical_bid_data: List[Dict] = field(default_factory=list)

    def calculate_performance_score(self) -> float:
        """Calculate overall performance score"""
        if not self.references:
            return 50.0  # Default for new contractors

        ratings = [r.performance_rating for r in self.references]
        on_time_rate = sum(1 for r in self.references if r.on_time) / len(self.references)
        on_budget_rate = sum(1 for r in self.references if r.on_budget) / len(self.references)

        # Weighted average
        avg_rating = sum(ratings) / len(ratings) / 5 * 100  # Normalize to 0-100
        on_time_score = on_time_rate * 100
        on_budget_score = on_budget_rate * 100

        return avg_rating * 0.5 + on_time_score * 0.3 + on_budget_score * 0.2

    def calculate_safety_score(self) -> float:
        """Calculate safety score"""
        base_score = 100

        # Deductions
        if self.incident_rate > 0:
            base_score -= min(30, self.incident_rate * 10)
        if self.fatality_count > 0:
            base_score -= 50  # Major deduction for fatalities
        if self.lost_time_incidents > 0:
            base_score -= min(20, self.lost_time_incidents * 2)

        # Bonuses for certifications
        if 'ISO45001' in self.safety_certifications or 'OHSAS18001' in self.safety_certifications:
            base_score += 10

        return max(0, min(100, base_score))

    def get_capacity_score(self) -> float:
        """Calculate available capacity score"""
        return 100 - self.current_workload_pct

AI Matching Engine

from sklearn.metrics.pairwise import cosine_similarity
from sklearn.feature_extraction.text import TfidfVectorizer
import pandas as pd

@dataclass
class ProjectRequirements:
    project_id: str
    project_name: str
    work_categories: List[WorkCategory]
    required_certifications: List[str]
    required_licenses: List[str]
    region: str
    estimated_value: float
    start_date: date
    duration_months: int
    priority_weights: Dict[str, float] = field(default_factory=dict)
    special_requirements: List[str] = field(default_factory=list)

    def __post_init__(self):
        if not self.priority_weights:
            self.priority_weights = {
                'perform