contractor-matching-ai

name	contractor-matching-ai
description	AI-powered contractor matching and selection for construction projects. Analyze contractor capabilities, past performance, certifications, and project requirements to recommend optimal matches.
homepage	https://datadrivenconstruction.io
metadata	{"openclaw":{"emoji":"🚀","os":["darwin","linux","win32"],"homepage":"https://datadrivenconstruction.io","requires":{"bins":"[Truncated]"}}}

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: 
    value: 
    completion_date: date
    work_type: 
    performance_rating:   
    on_time: 
    on_budget: 
    client_reference_available: 


 :
    contractor_id: 
    company_name: 
    legal_name: 
    registration_number: 
    size: ContractorSize
    founded_year: 
    employees_count: 

    
    specializations: [WorkCategory]
    equipment_owned: []
    max_project_value: 
    min_project_value: 

    
    certifications: []  
    licenses: []  

    
    completed_projects: 
    active_projects: 
    references: [ProjectReference] = field(default_factory=)

    
    safety_certifications: [] = field(default_factory=)
    incident_rate:  =   
    fatality_count:  = 
    lost_time_incidents:  = 

    
    annual_revenue:  = 
    credit_rating:  = 
    insurance_coverage:  = 
    bonding_capacity:  = 

    
    headquarters_region:  = 
    operating_regions: [] = field(default_factory=)
    willing_to_travel:  = 

    
    current_workload_pct:  =   
    earliest_availability: [date] = 

    
    historical_bid_data: [] = field(default_factory=)

     () -> :
        
          .references:
               

        ratings = [r.performance_rating  r  .references]
        on_time_rate = (  r  .references  r.on_time) / (.references)
        on_budget_rate = (  r  .references  r.on_budget) / (.references)

        
        avg_rating = (ratings) / (ratings) /  *   
        on_time_score = on_time_rate * 
        on_budget_score = on_budget_rate * 

         avg_rating *  + on_time_score *  + on_budget_score * 

     () -> :
        
        base_score = 

        
         .incident_rate > :
            base_score -= (, .incident_rate * )
         .fatality_count > :
            base_score -=   
         .lost_time_incidents > :
            base_score -= (, .lost_time_incidents * )

        
           .safety_certifications    .safety_certifications:
            base_score += 

         (, (, base_score))

     () -> :
        
          - .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 = {
                'performance': 0.25,
                'safety': 0.20,
                'price': 0.20,
                'capacity': 0.15,
                'experience': 0.10,
                'financial': 0.10
            }

class ContractorMatchingEngine:
    

     ():
        .contractors: [, ContractorProfile] = {}
        .vectorizer = TfidfVectorizer(ngram_range=(, ))
        .scaler = MinMaxScaler()

     ():
        
        .contractors[profile.contractor_id] = profile

     () -> []:
        
        eligible = ._filter_eligible(requirements)

          eligible:
             []

        scored = []
         contractor  eligible:
            score, breakdown = ._calculate_match_score(contractor, requirements)
            scored.append({
                : contractor.contractor_id,
                : contractor.company_name,
                : score,
                : breakdown,
                : contractor
            })

        
        scored.sort(key= x: x[], reverse=)

         scored[:top_n]

     () -> [ContractorProfile]:
        
        eligible = []

         contractor  .contractors.values():
            
             req.region   contractor.operating_regions:
                  contractor.willing_to_travel:
                    

            
            contractor_cats = (contractor.specializations)
            required_cats = (req.work_categories)
              required_cats.intersection(contractor_cats):
                

            
             req.estimated_value > contractor.max_project_value:
                
             req.estimated_value < contractor.min_project_value:
                

            
            contractor_certs = (c[]  c  contractor.certifications
                                   c.get(, date.) >= date.today())
              (req.required_certifications).issubset(contractor_certs):
                

            
            contractor_licenses = (l[]  l  contractor.licenses
                                      l.get(, date.) >= date.today())
              (req.required_licenses).issubset(contractor_licenses):
                

            
             contractor.current_workload_pct >= :  
                

            
             contractor.earliest_availability  contractor.earliest_availability > req.start_date:
                

            eligible.append(contractor)

         eligible

     () -> [, ]:
        
        weights = req.priority_weights
        breakdown = {}

        
        breakdown[] = contractor.calculate_performance_score()

        
        breakdown[] = contractor.calculate_safety_score()

        
        breakdown[] = ._calculate_price_score(contractor, req)

        
        breakdown[] = contractor.get_capacity_score()

        
        breakdown[] = ._calculate_experience_score(contractor, req)

        
        breakdown[] = ._calculate_financial_score(contractor, req)

        
        total = (
            breakdown[key] * weights.get(key, )
             key  breakdown
        )

         total, breakdown

     () -> :
        
          contractor.historical_bid_data:
               

        
        similar_bids = [
            bid  bid  contractor.historical_bid_data
             bid.get(, ) *  <= req.estimated_value <= bid.get(, ) * 
        ]

          similar_bids:
             

        
        variances = [bid.get(, )  bid  similar_bids]
        avg_variance = (variances) / (variances)

        
        
         avg_variance <= -:
               
         avg_variance <= :
              - avg_variance  
         avg_variance <= :
              - avg_variance  
        :
             (,  - avg_variance)  

     () -> :
        
          contractor.references:
               

        relevant_projects = []
         ref  contractor.references:
            
            :
                work_cat = WorkCategory(ref.work_type)
                 work_cat  req.work_categories:
                    relevant_projects.append(ref)
             ValueError:
                

          relevant_projects:
             

        
        recent_relevant = [
            p  p  relevant_projects
             (date.today() - p.completion_date).days <=  *   
        ]

        count_score = (, (relevant_projects) * )
        recency_score = (, (recent_relevant) * )

        
        values = [p.value  p  relevant_projects]
        avg_value = (values) / (values)
        value_ratio = (req.estimated_value, avg_value) / (req.estimated_value, avg_value)
        value_score = value_ratio * 

         count_score + recency_score + value_score

     () -> :
        
        score =   

        
         contractor.bonding_capacity >= req.estimated_value:
            score += 
         contractor.bonding_capacity >= req.estimated_value * :
            score += 

        
         contractor.insurance_coverage >= req.estimated_value:
            score += 
         contractor.insurance_coverage >= req.estimated_value * :
            score += 

        
        credit_scores = {: , : , : , : , : , : -}
        score += credit_scores.get(contractor.credit_rating, )

         (, (, score))

     () -> pd.DataFrame:
        
        data = []

         cid  contractor_ids:
            contractor = .contractors.get(cid)
              contractor:
                

            score, breakdown = ._calculate_match_score(contractor, req)

            row = {
                : contractor.company_name,
                : ,
                : ,
                : ,
                : ,
                : ,
                : ,
                : ,
                : contractor.active_projects,
                : 
            }
            data.append(row)

         pd.DataFrame(data)

Bid Analysis and Prediction

from sklearn.ensemble import RandomForestRegressor
from sklearn.model_selection import train_test_split
import numpy as np

class BidPredictor:
    """Predict contractor bid prices"""

    def __init__(self):
        self.model = RandomForestRegressor(n_estimators=100, random_state=42)
        self.is_trained = False

    def train(self, historical_bids: pd.DataFrame):
        """Train bid prediction model

        Expected columns:
        - contractor_size, work_type, region, project_value
        - contractor_performance, contractor_workload
        - winning_bid, contractor_bid
        """
        features = ['project_value', 'contractor_performance',
                   'contractor_workload', 'duration_months']

        # One-hot encode categorical
        df = pd.get_dummies(historical_bids,
                           columns=['contractor_size', 'work_type', 'region'])

        # Features that exist
        X_cols = [c for c in df.columns if c not in ['winning_bid', 'contractor_bid']]
        X = df[X_cols]
        y = df['contractor_bid']

        self.feature_columns = X_cols
        self.model.fit(X, y)
        self.is_trained = 

     () -> :
        
          .is_trained:
            
            base = project.estimated_value
            variance = np.random.uniform(-, )
             {
                : base * ( + variance),
                : ,
                : (-, )
            }

        
        features = {
            : project.estimated_value,
            : contractor.calculate_performance_score(),
            : contractor.current_workload_pct,
            : project.duration_months,
            : ,
            : 
        }

        
         cat  project.work_categories:
            features[] = 

        
        X = pd.DataFrame([features]).reindex(columns=.feature_columns, fill_value=)

        prediction = .model.predict(X)[]

        
         {
            : prediction,
            : ,
            : (-, ),
            : project.estimated_value,
            : (prediction - project.estimated_value) / project.estimated_value * 
        }


 :
    

     ():
        .engine = matching_engine
        .predictor = BidPredictor()

     () -> pd.DataFrame:
        
        results = []

         bid  bids:
            contractor = .engine.contractors.get(bid[])
              contractor:
                

            
            match_score, breakdown = .engine._calculate_match_score(
                contractor, project
            )

            
            avg_bid = (b[]  b  bids) / (bids)
            price_deviation = (bid[] - avg_bid) / avg_bid * 

             price_deviation <= -:
                price_score =   
             price_deviation <= :
                price_score =  - price_deviation
             price_deviation <= :
                price_score =  - price_deviation
            :
                price_score = (,  - price_deviation)

            
            eval_score = match_score *  + price_score * 

            results.append({
                : bid[],
                : contractor.company_name,
                : bid[],
                : ,
                : match_score,
                : price_score,
                : eval_score,
                : breakdown[],
                : breakdown[],
                : ._get_recommendation(eval_score, price_deviation)
            })

        df = pd.DataFrame(results)
        df = df.sort_values(, ascending=)

         df

     () -> :
        
         eval_score >=   price_dev <= :
             
         eval_score >= :
             
         eval_score >= :
             
         price_dev > :
             
        :
             

Contractor Recommendation Report

def generate_recommendation_report(engine: ContractorMatchingEngine,
                                   project: ProjectRequirements,
                                   output_path: str) -> str:
    """Generate contractor recommendation report"""
    matches = engine.find_matches(project, top_n=10)

    with pd.ExcelWriter(output_path, engine='openpyxl') as writer:
        # Summary
        summary_data = [{
            'Project': project.project_name,
            'Estimated Value': project.estimated_value,
            'Work Categories': ', '.join(c.value for c in project.work_categories),
            'Region': project.region,
            'Start Date': project.start_date.isoformat(),
            'Duration': f"{project.duration_months} months",
            'Contractors Found': len(matches)
        }]
        pd.DataFrame(summary_data).to_excel(writer, sheet_name='Summary', index=False)

        # Rankings
        ranking_data = []
        for i, match in enumerate(matches, 1):
            ranking_data.append({
                'Rank': i,
                'Contractor': match['company_name'],
                'Total Score': f"{match['total_score']:.1f}",
                : ,
                : ,
                : ,
                : ,
                : ,
                : 
            })
        pd.DataFrame(ranking_data).to_excel(writer, sheet_name=, index=)

        
         i,   (matches[:], ):
            profile = []
            profile_data = [{
                : , : profile.company_name
            }, {
                : , : profile.size.value
            }, {
                : , : profile.employees_count
            }, {
                : , : profile.completed_projects
            }, {
                : , : profile.active_projects
            }, {
                : , : 
            }, {
                : , : 
            }, {
                : , : profile.lost_time_incidents
            }]
            pd.DataFrame(profile_data).to_excel(
                writer, sheet_name=, index=
            )

     output_path

Quick Reference

Criterion	Weight Range	Data Sources
Performance	20-30%	Project references, ratings
Safety	15-25%	OSHA records, certifications
Price	15-25%	Historical bids
Capacity	10-20%	Current workload
Experience	10-15%	Similar projects
Financial	10-15%	Credit rating, bonding

Resources

• DDC Website: https://datadrivenconstruction.io
• Construction contractor databases: BuildingConnected, PlanHub

Next Steps

• See risk-assessment-ml for contractor risk analysis
• See document-classification-nlp for proposal analysis
• See open-construction-estimate for bid validation