diff --git a/.gitignore b/.gitignore index e5cbb64..de9b34b 100644 --- a/.gitignore +++ b/.gitignore @@ -32,3 +32,7 @@ google-services.json # Android Profiling *.hprof + +# Python cache files +__pycache__/ +*.py[cod] diff --git a/README.md b/README.md index 812e029..eb4e4a3 100644 --- a/README.md +++ b/README.md @@ -1 +1,38 @@ -# https-github.com-TheAlgorithms-Python-blob-master-CONTRIBUTING.md-coding-style \ No newline at end of file +# Enhanced Idealization Platform + +This repository now contains a Python MVP for an enhanced idealization platform. +It helps users define success, model constraints, generate multiple candidate +futures, stress test them across scenarios, and receive ranked recommendations +with confidence, tradeoffs, and reasoning traces. + +## What it does + +- captures goals, criteria, constraints, and context in a structured request +- generates multiple candidate strategies from a baseline or explicit blueprints +- adapts scoring based on user profiles and weighted success criteria +- simulates candidate performance under future scenarios +- explains why one recommendation outranks another +- compares the current recommendation against previous runs + +## Project layout + +- `/enhanced_idealization/models.py` - data models and request parsing +- `/enhanced_idealization/engine.py` - candidate generation, simulation, scoring +- `/enhanced_idealization/reporting.py` - plain-text recommendation rendering +- `/enhanced_idealization/__main__.py` - command-line entry point +- `/examples/product_strategy.json` - sample input +- `/tests/test_engine.py` - automated tests + +## Run the sample + +```bash +cd /path/to/enhanced-idealization +python -m enhanced_idealization examples/product_strategy.json --profile enterprise_ops +``` + +## Run the tests + +```bash +cd /path/to/enhanced-idealization +python -m unittest discover -s tests +``` diff --git a/enhanced_idealization/__init__.py b/enhanced_idealization/__init__.py new file mode 100644 index 0000000..5c517ea --- /dev/null +++ b/enhanced_idealization/__init__.py @@ -0,0 +1,6 @@ +"""Enhanced idealization platform package.""" + +from .engine import EnhancedIdealizationEngine +from .models import IdealizationRequest, IdealizationResult + +__all__ = ["EnhancedIdealizationEngine", "IdealizationRequest", "IdealizationResult"] diff --git a/enhanced_idealization/__main__.py b/enhanced_idealization/__main__.py new file mode 100644 index 0000000..97a5d58 --- /dev/null +++ b/enhanced_idealization/__main__.py @@ -0,0 +1,32 @@ +from __future__ import annotations + +import argparse +import json +from pathlib import Path + +from .engine import EnhancedIdealizationEngine +from .models import IdealizationRequest +from .reporting import render_text_report + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description="Run the enhanced idealization platform.") + parser.add_argument("input_path", type=Path, help="Path to a JSON request file.") + parser.add_argument( + "--profile", + help="Optional profile name to apply to the recommendation process.", + ) + return parser + + +def main() -> None: + parser = build_parser() + args = parser.parse_args() + payload = json.loads(args.input_path.read_text(encoding="utf-8")) + request = IdealizationRequest.from_dict(payload) + result = EnhancedIdealizationEngine().recommend(request, profile_name=args.profile) + print(render_text_report(result)) + + +if __name__ == "__main__": + main() diff --git a/enhanced_idealization/engine.py b/enhanced_idealization/engine.py new file mode 100644 index 0000000..afc1092 --- /dev/null +++ b/enhanced_idealization/engine.py @@ -0,0 +1,399 @@ +from __future__ import annotations + +from statistics import mean, pstdev + +from .models import ( + Candidate, + CandidateBlueprint, + Constraint, + IdealizationRequest, + IdealizationResult, + Recommendation, + Scenario, + ScenarioAssessment, + UserProfile, + clamp, +) + +DEFAULT_STRATEGIES: dict[str, dict[str, float]] = { + "Balanced Horizon": { + "feasibility": 0.08, + "efficiency": 0.07, + "goal_alignment": 0.08, + "risk_resilience": 0.06, + "adaptability": 0.06, + "collaboration": 0.05, + }, + "Bold Leap": { + "innovation": 0.14, + "goal_alignment": 0.08, + "efficiency": 0.04, + "risk_resilience": -0.07, + "cost_efficiency": -0.06, + "feasibility": -0.03, + }, + "Resilient Core": { + "risk_resilience": 0.14, + "adaptability": 0.11, + "feasibility": 0.06, + "collaboration": 0.03, + "innovation": -0.02, + }, + "Lean Efficiency": { + "cost_efficiency": 0.14, + "efficiency": 0.12, + "feasibility": 0.05, + "collaboration": -0.02, + "innovation": -0.03, + }, + "Collaborative Orbit": { + "collaboration": 0.15, + "adaptability": 0.08, + "goal_alignment": 0.06, + "efficiency": 0.03, + "cost_efficiency": -0.01, + }, +} + +SCORE_WEIGHT_BASE = 0.55 +SCORE_WEIGHT_SCENARIO = 0.35 +SCORE_WEIGHT_FUTURE = 0.10 +LOCKED_CONSTRAINT_WEIGHT_BOOST = 0.15 +LOCKED_CONSTRAINT_PENALTY = 0.35 +UNLOCKED_CONSTRAINT_PENALTY = 0.15 +CONFIDENCE_MARGIN_SCALE = 1.5 +CONFIDENCE_MARGIN_BASELINE = 0.5 +CONFIDENCE_WEIGHT_BASE = 0.4 +CONFIDENCE_WEIGHT_STABILITY = 0.4 +CONFIDENCE_WEIGHT_EVIDENCE = 0.2 +DEFAULT_CONSTRAINT_MARGIN = 0.5 +DEFAULT_SCENARIO_SCORE = 0.5 +DEFAULT_EVIDENCE_SCORE = 0.5 + + +class EnhancedIdealizationEngine: + def recommend( + self, request: IdealizationRequest, profile_name: str | None = None + ) -> IdealizationResult: + profile = self._select_profile(request, profile_name) + criteria_weights = self._adapt_weights(request, profile) + candidates = self._build_candidates(request, profile) + + recommendations = [] + for candidate in candidates: + base_score, constraint_status = self._score_candidate( + candidate.metrics, request.constraints, criteria_weights + ) + scenario_assessments = [ + self._simulate(candidate, scenario, request.constraints, criteria_weights) + for scenario in request.scenarios + ] + scenario_score = ( + self._weighted_scenario_score(scenario_assessments, request.scenarios) + if scenario_assessments + else base_score + ) + future_readiness = mean( + candidate.metrics.get(metric, 0.0) + for metric in ("adaptability", "innovation", "collaboration") + ) + total_score = round( + (base_score * SCORE_WEIGHT_BASE) + + (scenario_score * SCORE_WEIGHT_SCENARIO) + + (future_readiness * SCORE_WEIGHT_FUTURE), + 4, + ) + confidence = self._compute_confidence( + candidate.metrics, request.constraints, scenario_assessments, candidate + ) + tradeoffs = self._build_tradeoffs(candidate.metrics) + explanation = self._build_explanation( + candidate.name, + total_score, + base_score, + scenario_score, + confidence, + tradeoffs, + constraint_status, + ) + reasoning_trace = self._build_reasoning_trace( + request, profile, candidate, constraint_status, scenario_assessments + ) + recommendations.append( + Recommendation( + rank=0, + candidate=candidate, + total_score=total_score, + base_score=base_score, + scenario_score=scenario_score, + confidence=confidence, + explanation=explanation, + tradeoffs=tradeoffs, + reasoning_trace=reasoning_trace, + scenario_assessments=scenario_assessments, + constraint_status=constraint_status, + ) + ) + + recommendations.sort(key=lambda item: (item.total_score, item.confidence), reverse=True) + ranked = [ + Recommendation( + rank=index + 1, + candidate=item.candidate, + total_score=item.total_score, + base_score=item.base_score, + scenario_score=item.scenario_score, + confidence=item.confidence, + explanation=item.explanation, + tradeoffs=item.tradeoffs, + reasoning_trace=item.reasoning_trace, + scenario_assessments=item.scenario_assessments, + constraint_status=item.constraint_status, + ) + for index, item in enumerate(recommendations) + ] + return IdealizationResult( + request_title=request.title, + applied_profile=profile.name if profile else None, + recommendations=ranked, + comparison_notes=self._compare_with_history(request, ranked), + criteria_weights=criteria_weights, + ) + + def _select_profile( + self, request: IdealizationRequest, profile_name: str | None + ) -> UserProfile | None: + if profile_name is None: + return request.profiles[0] if request.profiles else None + for profile in request.profiles: + if profile.name == profile_name: + return profile + raise ValueError(f"Unknown profile: {profile_name}") + + def _adapt_weights( + self, request: IdealizationRequest, profile: UserProfile | None + ) -> dict[str, float]: + raw_weights: dict[str, float] = {} + for criterion in request.criteria: + weight = criterion.weight + if profile: + weight += profile.weight_adjustments.get(criterion.name, 0.0) + if any( + constraint.locked and constraint.metric == criterion.name + for constraint in request.constraints + ): + weight += LOCKED_CONSTRAINT_WEIGHT_BOOST + raw_weights[criterion.name] = max(weight, 0.01) + + total = sum(raw_weights.values()) or 1.0 + return {name: round(weight / total, 4) for name, weight in raw_weights.items()} + + def _build_candidates( + self, request: IdealizationRequest, profile: UserProfile | None + ) -> list[Candidate]: + candidates: list[Candidate] = [] + + if request.candidate_blueprints: + for blueprint in request.candidate_blueprints: + metrics = self._apply_profile_bias(dict(blueprint.metrics), profile) + candidates.append( + Candidate( + name=blueprint.name, + description=blueprint.description, + metrics=metrics, + strategy="custom blueprint", + assumptions=blueprint.assumptions, + facts=blueprint.facts, + collaboration_notes=blueprint.collaboration_notes, + ) + ) + + for strategy, deltas in DEFAULT_STRATEGIES.items(): + metrics = {key: value for key, value in request.baseline_metrics.items()} + for metric, delta in deltas.items(): + metrics[metric] = clamp(metrics.get(metric, 0.5) + delta) + metrics = self._apply_profile_bias(metrics, profile) + assumptions = [ + "Baseline metrics realistically represent the current state.", + "The chosen strategy can be executed without adding hidden constraints.", + ] + facts = [ + f"Generated from the '{strategy}' strategy pattern.", + f"Objective focus: {request.objective}.", + ] + candidates.append( + Candidate( + name=strategy, + description=f"{strategy} prioritizes a distinct path toward {request.success_definition}.", + metrics=metrics, + strategy="generated strategy", + assumptions=assumptions, + facts=facts, + collaboration_notes=list(request.collaboration_notes), + ) + ) + + deduplicated: dict[str, Candidate] = {} + for candidate in candidates: + deduplicated[candidate.name] = candidate + return list(deduplicated.values()) + + def _apply_profile_bias( + self, metrics: dict[str, float], profile: UserProfile | None + ) -> dict[str, float]: + if profile is None: + return {key: clamp(value) for key, value in metrics.items()} + adjusted = dict(metrics) + for metric, delta in profile.metric_bias.items(): + adjusted[metric] = clamp(adjusted.get(metric, 0.5) + delta) + return {key: clamp(value) for key, value in adjusted.items()} + + def _score_candidate( + self, + metrics: dict[str, float], + constraints: list[Constraint], + criteria_weights: dict[str, float], + ) -> tuple[float, list[str]]: + base = sum(metrics.get(name, 0.0) * weight for name, weight in criteria_weights.items()) + status: list[str] = [] + penalty = 0.0 + for constraint in constraints: + if constraint.is_satisfied(metrics): + status.append(f"pass: {constraint.name}") + continue + severity = LOCKED_CONSTRAINT_PENALTY if constraint.locked else UNLOCKED_CONSTRAINT_PENALTY + penalty += severity + status.append(f"fail: {constraint.name}") + return round(clamp(base - penalty), 4), status + + def _simulate( + self, + candidate: Candidate, + scenario: Scenario, + constraints: list[Constraint], + criteria_weights: dict[str, float], + ) -> ScenarioAssessment: + adjusted = dict(candidate.metrics) + for metric, delta in scenario.metric_adjustments.items(): + adjusted[metric] = clamp(adjusted.get(metric, 0.5) + delta) + score, status = self._score_candidate(adjusted, constraints, criteria_weights) + failures = [item.replace("fail: ", "") for item in status if item.startswith("fail: ")] + notes = [scenario.narrative] if scenario.narrative else [] + if failures: + notes.append("Scenario exposes constraint pressure.") + return ScenarioAssessment( + scenario_name=scenario.name, + score=score, + adjusted_metrics=adjusted, + notes=notes, + constraint_failures=failures, + ) + + def _weighted_scenario_score( + self, assessments: list[ScenarioAssessment], scenarios: list[Scenario] + ) -> float: + total_weight = sum(scenario.probability for scenario in scenarios) or 1.0 + weighted = 0.0 + for assessment, scenario in zip(assessments, scenarios): + weighted += assessment.score * scenario.probability + return round(weighted / total_weight, 4) + + def _compute_confidence( + self, + metrics: dict[str, float], + constraints: list[Constraint], + assessments: list[ScenarioAssessment], + candidate: Candidate, + ) -> float: + margins = [constraint.margin(metrics) for constraint in constraints] or [DEFAULT_CONSTRAINT_MARGIN] + base = clamp( + mean(clamp(margin, -1.0, 1.0) for margin in margins) / CONFIDENCE_MARGIN_SCALE + + CONFIDENCE_MARGIN_BASELINE + ) + scenario_scores = [assessment.score for assessment in assessments] or [DEFAULT_SCENARIO_SCORE] + stability = 1.0 - clamp(pstdev(scenario_scores) if len(scenario_scores) > 1 else 0.0) + evidence_total = len(candidate.facts) + len(candidate.assumptions) + evidence = ( + len(candidate.facts) / evidence_total if evidence_total else DEFAULT_EVIDENCE_SCORE + ) + confidence = ( + (base * CONFIDENCE_WEIGHT_BASE) + + (stability * CONFIDENCE_WEIGHT_STABILITY) + + (evidence * CONFIDENCE_WEIGHT_EVIDENCE) + ) + return round(clamp(confidence), 4) + + def _build_tradeoffs(self, metrics: dict[str, float]) -> list[str]: + ordered = sorted(metrics.items(), key=lambda item: item[1], reverse=True) + names = [name for name, _ in ordered] + if not names: + return [ + "Strength data is not available yet.", + "Tradeoff data is not available yet.", + ] + if len(names) == 1: + return [ + f"Strength currently centers on {names[0]}.", + f"Tradeoff analysis needs additional metrics beyond {names[0]}.", + ] + strengths = names[:2] + weaknesses = names[-2:] + return [ + f"Strength concentrated in {strengths[0]} and {strengths[1]}.", + f"Tradeoff pressure remains around {weaknesses[0]} and {weaknesses[1]}.", + ] + + def _build_explanation( + self, + candidate_name: str, + total_score: float, + base_score: float, + scenario_score: float, + confidence: float, + tradeoffs: list[str], + constraint_status: list[str], + ) -> str: + constraint_summary = "all constraints satisfied" if all( + item.startswith("pass:") for item in constraint_status + ) else "some constraints require attention" + return ( + f"{candidate_name} scores {total_score:.3f} overall " + f"(base {base_score:.3f}, scenario {scenario_score:.3f}) with " + f"{confidence:.3f} confidence; {constraint_summary}. {tradeoffs[0]} {tradeoffs[1]}" + ) + + def _build_reasoning_trace( + self, + request: IdealizationRequest, + profile: UserProfile | None, + candidate: Candidate, + constraint_status: list[str], + assessments: list[ScenarioAssessment], + ) -> list[str]: + stressed = [assessment.scenario_name for assessment in assessments if assessment.constraint_failures] + trace = [ + f"Objective: {request.objective}", + f"Success definition: {request.success_definition}", + f"Candidate path: {candidate.name} ({candidate.strategy})", + f"Profile applied: {profile.name if profile else 'none'}", + f"Constraint status: {', '.join(constraint_status) if constraint_status else 'none'}", + ] + if stressed: + trace.append(f"Stress scenarios: {', '.join(stressed)}") + if candidate.collaboration_notes: + trace.append( + f"Collaboration inputs: {'; '.join(candidate.collaboration_notes)}" + ) + return trace + + def _compare_with_history( + self, request: IdealizationRequest, recommendations: list[Recommendation] + ) -> list[str]: + if not request.history or not recommendations: + return [] + previous_best = max(entry.score for entry in request.history) + current_best = recommendations[0].total_score + delta = round(current_best - previous_best, 4) + if delta >= 0: + return [f"Current top recommendation improves on prior best by {delta:.3f} points."] + return [f"Current top recommendation trails prior best by {abs(delta):.3f} points."] diff --git a/enhanced_idealization/models.py b/enhanced_idealization/models.py new file mode 100644 index 0000000..db71aae --- /dev/null +++ b/enhanced_idealization/models.py @@ -0,0 +1,209 @@ +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any + + +def clamp(value: float, lower: float = 0.0, upper: float = 1.0) -> float: + return max(lower, min(upper, value)) + + +@dataclass(frozen=True) +class Criterion: + name: str + weight: float + target: str = "maximize" + + @staticmethod + def from_dict(data: dict[str, Any]) -> "Criterion": + return Criterion( + name=data["name"], + weight=float(data.get("weight", 1.0)), + target=data.get("target", "maximize"), + ) + + +@dataclass(frozen=True) +class Constraint: + name: str + metric: str + minimum: float | None = None + maximum: float | None = None + locked: bool = False + + def is_satisfied(self, metrics: dict[str, float]) -> bool: + value = metrics.get(self.metric, 0.0) + if self.minimum is not None and value < self.minimum: + return False + if self.maximum is not None and value > self.maximum: + return False + return True + + def margin(self, metrics: dict[str, float]) -> float: + value = metrics.get(self.metric, 0.0) + lower_margin = value - self.minimum if self.minimum is not None else 1.0 + upper_margin = self.maximum - value if self.maximum is not None else 1.0 + return min(lower_margin, upper_margin) + + @staticmethod + def from_dict(data: dict[str, Any]) -> "Constraint": + return Constraint( + name=data["name"], + metric=data["metric"], + minimum=data.get("minimum"), + maximum=data.get("maximum"), + locked=bool(data.get("locked", False)), + ) + + +@dataclass(frozen=True) +class Scenario: + name: str + metric_adjustments: dict[str, float] = field(default_factory=dict) + probability: float = 1.0 + narrative: str = "" + + @staticmethod + def from_dict(data: dict[str, Any]) -> "Scenario": + return Scenario( + name=data["name"], + metric_adjustments={ + key: float(value) for key, value in data.get("metric_adjustments", {}).items() + }, + probability=float(data.get("probability", 1.0)), + narrative=data.get("narrative", ""), + ) + + +@dataclass(frozen=True) +class UserProfile: + name: str + weight_adjustments: dict[str, float] = field(default_factory=dict) + metric_bias: dict[str, float] = field(default_factory=dict) + + @staticmethod + def from_dict(data: dict[str, Any]) -> "UserProfile": + return UserProfile( + name=data["name"], + weight_adjustments={ + key: float(value) for key, value in data.get("weight_adjustments", {}).items() + }, + metric_bias={key: float(value) for key, value in data.get("metric_bias", {}).items()}, + ) + + +@dataclass(frozen=True) +class CandidateBlueprint: + name: str + description: str + metrics: dict[str, float] + assumptions: list[str] = field(default_factory=list) + facts: list[str] = field(default_factory=list) + collaboration_notes: list[str] = field(default_factory=list) + + @staticmethod + def from_dict(data: dict[str, Any]) -> "CandidateBlueprint": + return CandidateBlueprint( + name=data["name"], + description=data["description"], + metrics={key: clamp(float(value)) for key, value in data.get("metrics", {}).items()}, + assumptions=list(data.get("assumptions", [])), + facts=list(data.get("facts", [])), + collaboration_notes=list(data.get("collaboration_notes", [])), + ) + + +@dataclass(frozen=True) +class HistoryEntry: + label: str + score: float + summary: str = "" + + @staticmethod + def from_dict(data: dict[str, Any]) -> "HistoryEntry": + return HistoryEntry( + label=data["label"], + score=float(data["score"]), + summary=data.get("summary", ""), + ) + + +@dataclass(frozen=True) +class IdealizationRequest: + title: str + domain: str + objective: str + success_definition: str + baseline_metrics: dict[str, float] + criteria: list[Criterion] + constraints: list[Constraint] + scenarios: list[Scenario] + profiles: list[UserProfile] = field(default_factory=list) + candidate_blueprints: list[CandidateBlueprint] = field(default_factory=list) + collaboration_notes: list[str] = field(default_factory=list) + history: list[HistoryEntry] = field(default_factory=list) + + @staticmethod + def from_dict(data: dict[str, Any]) -> "IdealizationRequest": + return IdealizationRequest( + title=data["title"], + domain=data["domain"], + objective=data["objective"], + success_definition=data["success_definition"], + baseline_metrics={ + key: clamp(float(value)) for key, value in data.get("baseline_metrics", {}).items() + }, + criteria=[Criterion.from_dict(item) for item in data.get("criteria", [])], + constraints=[Constraint.from_dict(item) for item in data.get("constraints", [])], + scenarios=[Scenario.from_dict(item) for item in data.get("scenarios", [])], + profiles=[UserProfile.from_dict(item) for item in data.get("profiles", [])], + candidate_blueprints=[ + CandidateBlueprint.from_dict(item) for item in data.get("candidate_blueprints", []) + ], + collaboration_notes=list(data.get("collaboration_notes", [])), + history=[HistoryEntry.from_dict(item) for item in data.get("history", [])], + ) + + +@dataclass(frozen=True) +class Candidate: + name: str + description: str + metrics: dict[str, float] + strategy: str + assumptions: list[str] = field(default_factory=list) + facts: list[str] = field(default_factory=list) + collaboration_notes: list[str] = field(default_factory=list) + + +@dataclass(frozen=True) +class ScenarioAssessment: + scenario_name: str + score: float + adjusted_metrics: dict[str, float] + notes: list[str] + constraint_failures: list[str] + + +@dataclass(frozen=True) +class Recommendation: + rank: int + candidate: Candidate + total_score: float + base_score: float + scenario_score: float + confidence: float + explanation: str + tradeoffs: list[str] + reasoning_trace: list[str] + scenario_assessments: list[ScenarioAssessment] + constraint_status: list[str] + + +@dataclass(frozen=True) +class IdealizationResult: + request_title: str + applied_profile: str | None + recommendations: list[Recommendation] + comparison_notes: list[str] + criteria_weights: dict[str, float] diff --git a/enhanced_idealization/reporting.py b/enhanced_idealization/reporting.py new file mode 100644 index 0000000..b75cc96 --- /dev/null +++ b/enhanced_idealization/reporting.py @@ -0,0 +1,36 @@ +from __future__ import annotations + +from .models import IdealizationResult + + +def render_text_report(result: IdealizationResult) -> str: + lines = [ + f"Enhanced Idealization Report: {result.request_title}", + f"Applied profile: {result.applied_profile or 'none'}", + "Adaptive criteria weights:", + ] + for name, weight in sorted(result.criteria_weights.items()): + lines.append(f" - {name}: {weight:.3f}") + + if result.comparison_notes: + lines.append("Comparison:") + lines.extend(f" - {note}" for note in result.comparison_notes) + + lines.append("Recommendations:") + for recommendation in result.recommendations: + lines.append( + f"{recommendation.rank}. {recommendation.candidate.name} " + f"(score={recommendation.total_score:.3f}, confidence={recommendation.confidence:.3f})" + ) + lines.append(f" {recommendation.explanation}") + lines.append(" Reasoning trace:") + for item in recommendation.reasoning_trace: + lines.append(f" - {item}") + lines.append(" Scenario assessments:") + for assessment in recommendation.scenario_assessments: + failures = ", ".join(assessment.constraint_failures) if assessment.constraint_failures else "none" + lines.append( + f" - {assessment.scenario_name}: score={assessment.score:.3f}, " + f"constraint_failures={failures}" + ) + return "\n".join(lines) diff --git a/examples/product_strategy.json b/examples/product_strategy.json new file mode 100644 index 0000000..c073828 --- /dev/null +++ b/examples/product_strategy.json @@ -0,0 +1,135 @@ +{ + "title": "Immense Decision Intelligence Workspace", + "domain": "product strategy", + "objective": "Design a platform that generates idealized product futures for enterprise teams.", + "success_definition": "Deliver explainable, resilient, and high-alignment recommendations faster than manual planning.", + "baseline_metrics": { + "feasibility": 0.66, + "efficiency": 0.62, + "goal_alignment": 0.7, + "risk_resilience": 0.58, + "adaptability": 0.6, + "collaboration": 0.57, + "cost_efficiency": 0.59, + "innovation": 0.64 + }, + "criteria": [ + { "name": "feasibility", "weight": 0.17 }, + { "name": "efficiency", "weight": 0.14 }, + { "name": "goal_alignment", "weight": 0.22 }, + { "name": "risk_resilience", "weight": 0.16 }, + { "name": "adaptability", "weight": 0.11 }, + { "name": "collaboration", "weight": 0.08 }, + { "name": "cost_efficiency", "weight": 0.06 }, + { "name": "innovation", "weight": 0.06 } + ], + "constraints": [ + { + "name": "minimum feasibility", + "metric": "feasibility", + "minimum": 0.55, + "locked": true + }, + { + "name": "minimum resilience", + "metric": "risk_resilience", + "minimum": 0.5 + } + ], + "scenarios": [ + { + "name": "budget compression", + "probability": 0.45, + "metric_adjustments": { + "cost_efficiency": 0.08, + "innovation": -0.04, + "feasibility": -0.03 + }, + "narrative": "Budgets tighten and leaders want faster proof of value." + }, + { + "name": "rapid demand surge", + "probability": 0.35, + "metric_adjustments": { + "adaptability": 0.08, + "collaboration": 0.05, + "efficiency": -0.03 + }, + "narrative": "Adoption accelerates and the platform must scale across teams." + }, + { + "name": "compliance expansion", + "probability": 0.2, + "metric_adjustments": { + "risk_resilience": -0.06, + "feasibility": -0.02, + "goal_alignment": 0.03 + }, + "narrative": "Regulatory requirements grow and auditability becomes more important." + } + ], + "profiles": [ + { + "name": "enterprise_ops", + "weight_adjustments": { + "risk_resilience": 0.08, + "collaboration": 0.04, + "innovation": -0.02 + }, + "metric_bias": { + "risk_resilience": 0.05, + "collaboration": 0.04 + } + }, + { + "name": "innovation_lab", + "weight_adjustments": { + "innovation": 0.08, + "adaptability": 0.03, + "cost_efficiency": -0.03 + }, + "metric_bias": { + "innovation": 0.05, + "adaptability": 0.04 + } + } + ], + "candidate_blueprints": [ + { + "name": "Evidence Grid", + "description": "A measured platform centered on auditable recommendations and team consensus.", + "metrics": { + "feasibility": 0.74, + "efficiency": 0.63, + "goal_alignment": 0.79, + "risk_resilience": 0.73, + "adaptability": 0.67, + "collaboration": 0.76, + "cost_efficiency": 0.58, + "innovation": 0.61 + }, + "assumptions": [ + "Teams will share structured input consistently." + ], + "facts": [ + "The blueprint preserves full reasoning trails.", + "The blueprint prioritizes cross-team visibility." + ], + "collaboration_notes": [ + "Operations wants strong auditability.", + "Product leadership wants clear option comparison." + ] + } + ], + "collaboration_notes": [ + "Design wants visual tradeoff comparison.", + "Strategy wants the system to compare multiple futures at once." + ], + "history": [ + { + "label": "manual planning baseline", + "score": 0.69, + "summary": "Past process relied on manual workshops and slower tradeoff analysis." + } + ] +} diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..0797571 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,16 @@ +[build-system] +requires = ["setuptools>=68"] +build-backend = "setuptools.build_meta" + +[project] +name = "enhanced-idealization" +version = "0.1.0" +description = "A Python MVP for generating and evaluating idealized future-state options." +readme = "README.md" +requires-python = ">=3.11" +authors = [ + { name = "GitHub Copilot Task Agent" } +] + +[tool.setuptools] +packages = ["enhanced_idealization"] diff --git a/tests/test_engine.py b/tests/test_engine.py new file mode 100644 index 0000000..618bc53 --- /dev/null +++ b/tests/test_engine.py @@ -0,0 +1,186 @@ +from __future__ import annotations + +import unittest + +from enhanced_idealization.engine import EnhancedIdealizationEngine +from enhanced_idealization.models import IdealizationRequest + + +def build_request() -> IdealizationRequest: + return IdealizationRequest.from_dict( + { + "title": "Platform strategy", + "domain": "product", + "objective": "Recommend the best future-state platform direction.", + "success_definition": "Produce explainable, resilient, and scalable options.", + "baseline_metrics": { + "feasibility": 0.65, + "efficiency": 0.63, + "goal_alignment": 0.69, + "risk_resilience": 0.56, + "adaptability": 0.61, + "collaboration": 0.58, + "cost_efficiency": 0.57, + "innovation": 0.66, + }, + "criteria": [ + {"name": "feasibility", "weight": 0.17}, + {"name": "efficiency", "weight": 0.15}, + {"name": "goal_alignment", "weight": 0.2}, + {"name": "risk_resilience", "weight": 0.16}, + {"name": "adaptability", "weight": 0.11}, + {"name": "collaboration", "weight": 0.08}, + {"name": "cost_efficiency", "weight": 0.07}, + {"name": "innovation", "weight": 0.06}, + ], + "constraints": [ + { + "name": "minimum feasibility", + "metric": "feasibility", + "minimum": 0.55, + "locked": True, + }, + { + "name": "minimum resilience", + "metric": "risk_resilience", + "minimum": 0.5, + }, + ], + "scenarios": [ + { + "name": "scale pressure", + "probability": 0.5, + "metric_adjustments": { + "adaptability": 0.06, + "efficiency": -0.03, + "collaboration": 0.04, + }, + }, + { + "name": "budget squeeze", + "probability": 0.5, + "metric_adjustments": { + "cost_efficiency": 0.08, + "feasibility": -0.04, + "innovation": -0.05, + }, + }, + ], + "profiles": [ + { + "name": "enterprise", + "weight_adjustments": {"risk_resilience": 0.08, "collaboration": 0.03}, + "metric_bias": {"risk_resilience": 0.05, "collaboration": 0.03}, + }, + { + "name": "innovation", + "weight_adjustments": {"innovation": 0.08, "adaptability": 0.03}, + "metric_bias": {"innovation": 0.05, "adaptability": 0.04}, + }, + ], + "candidate_blueprints": [ + { + "name": "Safety Mesh", + "description": "Resilient operating model for regulated growth.", + "metrics": { + "feasibility": 0.75, + "efficiency": 0.61, + "goal_alignment": 0.73, + "risk_resilience": 0.8, + "adaptability": 0.65, + "collaboration": 0.7, + "cost_efficiency": 0.55, + "innovation": 0.58, + }, + "facts": ["Proven governance workflow."], + "assumptions": ["Regulated clients dominate adoption."], + }, + { + "name": "Vision Sprint", + "description": "Higher upside model with more innovation leverage.", + "metrics": { + "feasibility": 0.58, + "efficiency": 0.64, + "goal_alignment": 0.72, + "risk_resilience": 0.51, + "adaptability": 0.75, + "collaboration": 0.57, + "cost_efficiency": 0.54, + "innovation": 0.84, + }, + "facts": ["Strong prototype signal."], + "assumptions": ["Teams accept more delivery volatility."], + }, + ], + "history": [{"label": "previous state", "score": 0.67}], + } + ) + + +class EnhancedIdealizationEngineTests(unittest.TestCase): + def setUp(self) -> None: + self.engine = EnhancedIdealizationEngine() + self.request = build_request() + + def test_returns_ranked_recommendations(self) -> None: + result = self.engine.recommend(self.request, profile_name="enterprise") + self.assertGreaterEqual(len(result.recommendations), 5) + self.assertEqual(result.recommendations[0].rank, 1) + self.assertGreaterEqual( + result.recommendations[0].total_score, result.recommendations[-1].total_score + ) + self.assertTrue(result.comparison_notes) + + def test_personalization_changes_recommendation_weights(self) -> None: + enterprise_result = self.engine.recommend(self.request, profile_name="enterprise") + innovation_result = self.engine.recommend(self.request, profile_name="innovation") + self.assertNotEqual( + enterprise_result.criteria_weights["innovation"], + innovation_result.criteria_weights["innovation"], + ) + enterprise_scores = { + item.candidate.name: item.total_score for item in enterprise_result.recommendations + } + innovation_scores = { + item.candidate.name: item.total_score for item in innovation_result.recommendations + } + self.assertGreater( + innovation_scores["Vision Sprint"], + enterprise_scores["Vision Sprint"], + ) + + def test_scenario_stress_testing_is_included(self) -> None: + result = self.engine.recommend(self.request) + top = result.recommendations[0] + self.assertEqual(len(top.scenario_assessments), 2) + self.assertTrue(any("Objective:" in item for item in top.reasoning_trace)) + + def test_low_margin_candidates_reduce_confidence(self) -> None: + result = self.engine.recommend(self.request, profile_name="enterprise") + recommendation_by_name = { + item.candidate.name: item for item in result.recommendations + } + self.assertLess( + recommendation_by_name["Vision Sprint"].confidence, + recommendation_by_name["Safety Mesh"].confidence, + ) + + def test_tradeoff_builder_handles_sparse_metrics(self) -> None: + self.assertEqual( + self.engine._build_tradeoffs({}), + [ + "Strength data is not available yet.", + "Tradeoff data is not available yet.", + ], + ) + self.assertEqual( + self.engine._build_tradeoffs({"feasibility": 0.8}), + [ + "Strength currently centers on feasibility.", + "Tradeoff analysis needs additional metrics beyond feasibility.", + ], + ) + + +if __name__ == "__main__": + unittest.main()