code/self_healing_finetuning/config/profile_loader.py

from __future__ import annotations

import json
from pathlib import Path
from typing import Any

from n21.settings import CONFIG_ROOT


PROFILE_SCHEMA_VERSION = "shft_strategy_profile_v1"


def profile_id(asset_class: str, role: str) -> str:
    return f"{asset_class}_{role}".lower().replace("-", "_")


def profile_path(asset_class: str, role: str) -> Path:
    return CONFIG_ROOT / "profiles" / f"{profile_id(asset_class, role)}.json"


def load_strategy_profile(asset_class: str, role: str, path: Path | None = None) -> dict[str, Any]:
    resolved = path or profile_path(asset_class, role)
    if not resolved.exists():
        raise FileNotFoundError(f"strategy profile not found: {resolved}")
    profile = json.loads(resolved.read_text(encoding="utf-8-sig"))
    errors = validate_strategy_profile(profile, asset_class=asset_class, role=role)
    if errors:
        raise ValueError(f"invalid strategy profile {resolved}: {'; '.join(errors)}")
    profile["profile_path"] = str(resolved)
    return profile


def validate_strategy_profile(profile: dict[str, Any], *, asset_class: str | None = None, role: str | None = None) -> list[str]:
    errors: list[str] = []
    if profile.get("schema_version") != PROFILE_SCHEMA_VERSION:
        errors.append(f"schema_version must be {PROFILE_SCHEMA_VERSION}")
    if asset_class and profile.get("asset_class") != asset_class:
        errors.append(f"asset_class must be {asset_class}")
    if role and profile.get("role") != role:
        errors.append(f"role must be {role}")
    for key in (
        "critical_failure_buckets",
        "forbidden_behaviors",
        "promotion_priorities",
        "training_modes",
        "strategy_rules",
    ):
        if not isinstance(profile.get(key), list) or not profile[key]:
            errors.append(f"{key} must be a non-empty list")
    weights = profile.get("promotion_impact_weights")
    if not isinstance(weights, dict):
        errors.append("promotion_impact_weights must be an object")
    else:
        for key in ("critical_regression", "pairwise_loss", "human_failure", "aggregate_shortfall"):
            if key not in weights:
                errors.append(f"promotion_impact_weights.{key} must be present")
    return errors