Azure DevOps Demo Models · main
modelops_demo_runner.py

tools/modelops_demo_runner.py · 15,623 bytes · read-only public source view
from __future__ import annotations
import argparse
import csv
import json
import math
from pathlib import Path
from typing import Any


def as_float(values: dict[str, Any], key: str, default: float) -> float:
    try:
        return float(values.get(key, default))
    except (TypeError, ValueError):
        return default


def as_int(values: dict[str, Any], key: str, default: int) -> int:
    try:
        return int(float(values.get(key, default)))
    except (TypeError, ValueError):
        return default


def write_csv(path: Path, rows: list[dict[str, Any]]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    fieldnames = list(rows[0].keys()) if rows else ['message']
    with path.open('w', encoding='utf-8', newline='') as handle:
        writer = csv.DictWriter(handle, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(rows or [{'message': 'No rows produced'}])


def write_json(path: Path, value: Any) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(value, indent=2) + '\n', encoding='utf-8')


def life_reserve(p: dict[str, Any]) -> tuple[list[dict[str, Any]], dict[str, Any]]:
    policies = as_int(p, 'policies', 100000)
    premium = as_float(p, 'averagePremium', 1250)
    claim_rate = as_float(p, 'claimRate', 0.012)
    avg_claim = as_float(p, 'averageClaim', 85000)
    expense = as_float(p, 'expenseRatio', 0.18)
    discount = as_float(p, 'discountRate', 0.035)
    years = as_int(p, 'projectionYears', 20)
    scenario = str(p.get('scenario', 'Base'))
    if scenario == 'Adverse mortality': claim_rate *= 1.25
    if scenario == 'Low interest': discount = max(0.005, discount - 0.02)
    if scenario == 'Combined stress':
        claim_rate *= 1.25
        discount = max(0.005, discount - 0.02)
    rows = []
    reserve = 0.0
    for year in range(1, years + 1):
        active = max(0, round(policies * (0.985 ** (year - 1))))
        premiums = active * premium
        claims = active * claim_rate * avg_claim
        expenses = premiums * expense
        net = premiums - claims - expenses
        reserve = max(0.0, reserve * (1 + discount) - net)
        rows.append({'year': year, 'active_policies': active, 'premiums': round(premiums, 2), 'claims': round(claims, 2), 'expenses': round(expenses, 2), 'reserve': round(reserve, 2)})
    return rows, {'scenario': scenario, 'endingReserve': round(reserve, 2), 'years': years, 'status': 'Completed'}


def pension_funding(p: dict[str, Any]) -> tuple[list[dict[str, Any]], dict[str, Any]]:
    assets = as_float(p, 'openingAssets', 8_500_000_000)
    liability = as_float(p, 'openingLiability', 9_200_000_000)
    contributions = as_float(p, 'annualContributions', 520_000_000)
    benefits = as_float(p, 'annualBenefits', 610_000_000)
    ret = as_float(p, 'investmentReturn', 0.055)
    growth = as_float(p, 'liabilityGrowth', 0.038)
    years = as_int(p, 'projectionYears', 30)
    scenario = str(p.get('scenario', 'Reference'))
    if scenario == 'Low return': ret -= 0.025
    if scenario == 'Longevity stress': growth += 0.012
    if scenario == 'Contribution increase': contributions *= 1.15
    rows=[]
    for year in range(1, years + 1):
        assets = assets * (1 + ret) + contributions - benefits
        liability *= 1 + growth
        ratio = assets / liability if liability else 0
        rows.append({'year':year,'assets':round(assets,2),'liability':round(liability,2),'funded_ratio':round(ratio,4),'contributions':round(contributions,2),'benefits':round(benefits,2)})
        benefits *= 1.025
        contributions *= 1.022
    return rows, {'scenario':scenario,'endingFundedRatio':rows[-1]['funded_ratio'],'endingAssets':rows[-1]['assets'],'endingLiability':rows[-1]['liability'],'status':'Completed'}


def credit_stress(p: dict[str, Any]) -> tuple[list[dict[str, Any]], dict[str, Any]]:
    exposure=as_float(p,'totalExposure',2_500_000_000)
    pd=as_float(p,'probabilityDefault',0.018)
    lgd=as_float(p,'lossGivenDefault',0.42)
    multiplier=as_float(p,'stressMultiplier',1.75)
    years=as_int(p,'years',5)
    scenario=str(p.get('scenario','Baseline'))
    scenario_factor={'Baseline':1.0,'Moderate recession':1.35,'Severe recession':2.0,'Rate shock':1.55}.get(scenario,1.0)
    rows=[]
    total=0.0
    for year in range(1,years+1):
        year_pd=min(1.0,pd*(1+0.05*(year-1))*scenario_factor)
        expected=exposure*year_pd*lgd*(multiplier if scenario != 'Baseline' else 1.0)
        total+=expected
        rows.append({'year':year,'exposure':round(exposure,2),'probability_default':round(year_pd,6),'loss_given_default':round(lgd,4),'expected_loss':round(expected,2)})
        exposure*=0.96
    return rows, {'scenario':scenario,'cumulativeExpectedLoss':round(total,2),'lossToOpeningExposure':round(total/as_float(p,'totalExposure',2_500_000_000),6),'status':'Completed'}


def public_benefit(p: dict[str, Any]) -> tuple[list[dict[str, Any]], dict[str, Any]]:
    people=as_int(p,'startingBeneficiaries',1_450_000)
    monthly=as_float(p,'averageMonthlyBenefit',780)
    growth=as_float(p,'populationGrowth',0.021)
    indexation=as_float(p,'indexationRate',0.024)
    years=as_int(p,'projectionYears',15)
    scenario=str(p.get('policyScenario','Current policy'))
    if scenario == 'Expanded eligibility': growth += 0.012
    if scenario == 'Reduced indexation': indexation = max(0,indexation-0.012)
    if scenario == 'Economic downturn': growth += 0.02
    rows=[]
    cumulative=0.0
    for year in range(1,years+1):
        annual=people*monthly*12
        cumulative+=annual
        rows.append({'year':year,'beneficiaries':round(people),'monthly_benefit':round(monthly,2),'annual_cost':round(annual,2),'cumulative_cost':round(cumulative,2)})
        people*=1+growth
        monthly*=1+indexation
    return rows, {'scenario':scenario,'endingBeneficiaries':rows[-1]['beneficiaries'],'finalAnnualCost':rows[-1]['annual_cost'],'cumulativeCost':round(cumulative,2),'status':'Completed'}


def hospital_capacity(p: dict[str, Any]) -> tuple[list[dict[str, Any]], dict[str, Any]]:
    beds=as_int(p,'licensedBeds',540)
    occupancy=as_float(p,'currentOccupancy',0.91)
    admissions=as_int(p,'monthlyAdmissions',4800)
    los=as_float(p,'averageLengthStay',5.2)
    growth=as_float(p,'demandGrowth',0.032)
    months=as_int(p,'forecastMonths',24)
    scenario=str(p.get('scenario','Current plan'))
    if scenario == 'Winter surge': growth += 0.08
    if scenario == 'Staffing shortage': beds=round(beds*0.9)
    if scenario == 'Capacity expansion': beds=round(beds*1.12)
    rows=[]
    max_occ=0.0
    for month in range(1,months+1):
        seasonal=1+0.08*math.sin((month-1)*math.pi/6)
        projected_admissions=admissions*((1+growth)**(month/12))*seasonal
        required_bed_days=projected_admissions*los
        available=beds*30.4375
        occ=min(1.5,required_bed_days/available)
        wait=max(0,(occ-0.85)*80)
        max_occ=max(max_occ,occ)
        rows.append({'month':month,'admissions':round(projected_admissions),'licensed_beds':beds,'occupancy':round(occ,4),'wait_time_index':round(wait,2)})
    return rows, {'scenario':scenario,'peakOccupancy':round(max_occ,4),'monthsAbove95Percent':sum(1 for r in rows if r['occupancy']>0.95),'status':'Completed'}


def energy_demand(p: dict[str, Any]) -> tuple[list[dict[str, Any]], dict[str, Any]]:
    peak=as_float(p,'basePeakMw',21400)
    growth=as_float(p,'annualGrowth',0.016)
    uplift=as_float(p,'electrificationUplift',0.008)
    weather=as_float(p,'weatherFactor',1.03)
    years=as_int(p,'projectionYears',12)
    scenario=str(p.get('scenario','Reference'))
    if scenario == 'Rapid electrification': uplift += 0.02
    if scenario == 'Conservation': growth -= 0.012
    if scenario == 'Extreme weather': weather *= 1.08
    rows=[]
    for year in range(1,years+1):
        demand=peak*((1+growth+uplift)**year)*weather
        rows.append({'year':year,'peak_demand_mw':round(demand,2),'reserve_requirement_mw':round(demand*1.15,2),'annual_growth':round(growth+uplift,5),'weather_factor':round(weather,4)})
    return rows, {'scenario':scenario,'endingPeakMw':rows[-1]['peak_demand_mw'],'capacityRequirementMw':rows[-1]['reserve_requirement_mw'],'status':'Completed'}


def inventory_reorder(p: dict[str, Any]) -> tuple[list[dict[str, Any]], dict[str, Any]]:
    annual=as_int(p,'annualDemand',85000)
    order=as_float(p,'orderCost',240)
    unit=as_float(p,'unitCost',18.75)
    hold=as_float(p,'holdingRate',0.22)
    lead=as_int(p,'leadTimeDays',18)
    sigma=as_float(p,'dailyDemandStdDev',42)
    service=str(p.get('serviceLevel','95%'))
    z={'90%':1.282,'95%':1.645,'97.5%':1.96,'99%':2.326}.get(service,1.645)
    holding_cost=max(0.0001,unit*hold)
    eoq=math.sqrt(2*annual*order/holding_cost)
    daily=annual/365
    safety=z*sigma*math.sqrt(lead)
    reorder=daily*lead+safety
    rows=[{'sku':'DEMO-001','annual_demand':annual,'economic_order_quantity':round(eoq,2),'safety_stock':round(safety,2),'reorder_point':round(reorder,2),'service_level':service}]
    annual_ordering=(annual/eoq)*order
    annual_holding=(eoq/2+safety)*holding_cost
    return rows, {'serviceLevel':service,'economicOrderQuantity':round(eoq,2),'reorderPoint':round(reorder,2),'estimatedAnnualInventoryCost':round(annual_ordering+annual_holding,2),'status':'Completed'}


def retail_markdown(p: dict[str, Any]) -> tuple[list[dict[str, Any]], dict[str, Any]]:
    units=as_int(p,'baseWeeklyUnits',12500)
    price=as_float(p,'listPrice',79.99)
    cost=as_float(p,'unitCost',31.5)
    elasticity=as_float(p,'priceElasticity',1.45)
    markdown=as_float(p,'markdownPercent',0.2)
    weeks=as_int(p,'weeks',13)
    campaign=str(p.get('campaign','Seasonal'))
    if campaign == 'Clearance': markdown=max(markdown,0.35)
    if campaign == 'Loyalty event': units=round(units*1.08)
    if campaign == 'No promotion': markdown=0.0
    rows=[]
    total_revenue=0.0
    total_margin=0.0
    for week in range(1,weeks+1):
        selling=price*(1-markdown)
        demand=units*(1+elasticity*markdown)*(0.99**(week-1))
        revenue=demand*selling
        margin=demand*(selling-cost)
        total_revenue+=revenue
        total_margin+=margin
        rows.append({'week':week,'markdown':round(markdown,4),'selling_price':round(selling,2),'units':round(demand),'revenue':round(revenue,2),'gross_margin':round(margin,2)})
    return rows, {'campaign':campaign,'totalRevenue':round(total_revenue,2),'totalGrossMargin':round(total_margin,2),'marginRate':round(total_margin/total_revenue,4) if total_revenue else 0,'status':'Completed'}


def vba_assessment(p: dict[str, Any]) -> tuple[list[dict[str, Any]], dict[str, Any]]:
    routines=as_int(p,'routineCount',42)
    external=as_int(p,'externalCalls',6)
    globals_count=as_int(p,'globalVariables',18)
    handlers=as_int(p,'errorHandlers',24)
    coverage=as_float(p,'testCoverage',0.15)
    target=str(p.get('targetPlatform','.NET'))
    score=min(100,round(20 + routines*0.6 + external*3 + globals_count*0.7 + max(0,0.5-coverage)*30))
    categories=[('Core calculations',max(1,round(routines*0.45)),'Automate tests then extract'),('Workbook integration',max(1,round(routines*0.25)),'Replace worksheet coupling'),('External dependencies',external,'Introduce adapters'),('Reporting and formatting',max(1,routines-round(routines*0.70)),'Rebuild as output services')]
    rows=[{'category':name,'routine_count':count,'recommended_action':action,'target_platform':target} for name,count,action in categories]
    return rows, {'targetPlatform':target,'complexityScore':score,'riskBand':'High' if score>=70 else 'Medium' if score>=40 else 'Low','estimatedMigrationSprints':max(2,math.ceil(routines/12)+math.ceil(external/3)),'status':'Completed'}


def enrollment_forecast(p: dict[str, Any]) -> tuple[list[dict[str, Any]], dict[str, Any]]:
    enrollment=as_int(p,'startingEnrollment',28600)
    apps=as_int(p,'annualApplications',52000)
    admit=as_float(p,'admitRate',0.48)
    yield_rate=as_float(p,'yieldRate',0.29)
    retention=as_float(p,'retentionRate',0.88)
    years=as_int(p,'projectionYears',8)
    scenario=str(p.get('scenario','Reference'))
    if scenario == 'International growth': apps=round(apps*1.08)
    if scenario == 'Domestic decline': apps=round(apps*0.94)
    if scenario == 'Retention improvement': retention=min(0.98,retention+0.035)
    rows=[]
    for year in range(1,years+1):
        new_students=apps*admit*yield_rate
        continuing=enrollment*retention
        enrollment=new_students+continuing
        rows.append({'year':year,'applications':round(apps),'new_students':round(new_students),'continuing_students':round(continuing),'total_enrollment':round(enrollment),'retention_rate':round(retention,4)})
        apps*=1.012
    return rows, {'scenario':scenario,'endingEnrollment':rows[-1]['total_enrollment'],'changeFromOpening':rows[-1]['total_enrollment']-as_int(p,'startingEnrollment',28600),'status':'Completed'}


ENGINES = {
    'life-reserve': life_reserve,
    'pension-funding': pension_funding,
    'credit-stress': credit_stress,
    'public-benefit': public_benefit,
    'hospital-capacity': hospital_capacity,
    'energy-demand': energy_demand,
    'inventory-reorder': inventory_reorder,
    'retail-markdown': retail_markdown,
    'vba-assessment': vba_assessment,
    'enrollment-forecast': enrollment_forecast,
}


def find_model(repo: Path, model_key: str) -> tuple[dict[str, Any], dict[str, Any]]:
    catalog=json.loads((repo/'catalog.json').read_text(encoding='utf-8'))
    entry=next((x for x in catalog['models'] if x['modelKey']==model_key),None)
    if entry is None:
        raise ValueError(f'Unknown model key: {model_key}')
    manifest=json.loads((repo/entry['path']/'modelops.json').read_text(encoding='utf-8'))
    return entry,manifest


def run(repo: Path, model_key: str, config_path: Path, output: Path) -> dict[str, Any]:
    entry,manifest=find_model(repo,model_key)
    config=json.loads(config_path.read_text(encoding='utf-8'))
    parameters=config.get('parameters',config)
    engine=ENGINES.get(entry['engineKey'])
    if engine is None:
        raise ValueError(f"No safe engine for {entry['engineKey']}")
    rows,summary=engine(parameters)
    output.mkdir(parents=True,exist_ok=True)
    csv_output=next((x for x in manifest['outputs'] if x['contentType']=='text/csv'),manifest['outputs'][0])
    json_output=next((x for x in manifest['outputs'] if x['contentType']=='application/json'),manifest['outputs'][-1])
    csv_name=Path(csv_output['pathTemplate']).name
    json_name=Path(json_output['pathTemplate']).name
    write_csv(output/csv_name,rows)
    summary.update({'modelKey':model_key,'modelName':entry['displayName'],'version':entry['version'],'engineKey':entry['engineKey']})
    write_json(output/json_name,summary)
    (output/'run.log').write_text(f"Model: {entry['displayName']}\nVersion: {entry['version']}\nEngine: {entry['engineKey']}\nStatus: Succeeded\n",encoding='utf-8')
    return {'modelKey':model_key,'outputDirectory':str(output),'files':[csv_name,json_name,'run.log'],'summary':summary}


def main() -> int:
    parser=argparse.ArgumentParser()
    parser.add_argument('--repo',required=True)
    parser.add_argument('--model',required=True)
    parser.add_argument('--config',required=True)
    parser.add_argument('--output',required=True)
    args=parser.parse_args()
    result=run(Path(args.repo).resolve(),args.model,Path(args.config).resolve(),Path(args.output).resolve())
    print(json.dumps(result,indent=2))
    return 0


if __name__=='__main__':
    raise SystemExit(main())