Upload strategy_core.py

strategy_core.py

@@ -0,0 +1,271 @@
+"""
+Strategic Sandbox - Core Logic Module
+Data models and simulation engine for strategy evaluation
+"""
+
+import json
+from typing import List, Dict, Any, Optional
+from dataclasses import dataclass, asdict
+import pandas as pd
+
+
+@dataclass
+class Goal:
+    """Strategic goal definition with main metric"""
+    text: str
+    metric: str
+    baseline: float
+    target: float
+    horizon: str
+    unit: str = "%"
+
+
+@dataclass
+class Arena:
+    """Market arena definition"""
+    market: str
+    category: str
+    competitors: List[str]
+    target_audience: str = ""
+
+
+@dataclass
+class Insight:
+    """Market or consumer insight"""
+    id: str
+    text: str
+    evidence: List[str]
+
+
+@dataclass
+class Hypothesis:
+    """Testable hypothesis"""
+    id: str
+    text: str
+    based_on: List[str]  # insight IDs
+    metric: str
+    expected_change: float
+
+
+@dataclass
+class Move:
+    """Strategic move/action"""
+    id: str
+    text: str
+    linked_hypothesis: str
+    impact: float  # 0-1
+    fit: float  # 0-1
+    risk: float  # 0-1
+    cost: float
+
+
+@dataclass
+class Metric:
+    """Success metric"""
+    id: str
+    text: str
+    baseline: float
+    target: float
+    unit: str
+
+
+class Strategy:
+    """Complete strategy model"""
+
+    def __init__(self):
+        self.goal: Optional[Goal] = None
+        self.arena: Optional[Arena] = None
+        self.insights: List[Insight] = []
+        self.hypotheses: List[Hypothesis] = []
+        self.moves: List[Move] = []
+        self.metrics: List[Metric] = []
+
+    def to_dict(self) -> Dict[str, Any]:
+        """Convert strategy to dictionary"""
+        return {
+            "goal": asdict(self.goal) if self.goal else None,
+            "arena": asdict(self.arena) if self.arena else None,
+            "insights": [asdict(i) for i in self.insights],
+            "hypotheses": [asdict(h) for h in self.hypotheses],
+            "moves": [asdict(m) for m in self.moves],
+            "metrics": [asdict(m) for m in self.metrics]
+        }
+
+    def to_json(self, filepath: str):
+        """Save strategy to JSON file"""
+        with open(filepath, 'w', encoding='utf-8') as f:
+            json.dump(self.to_dict(), f, indent=2, ensure_ascii=False)
+
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> 'Strategy':
+        """Load strategy from dictionary"""
+        strategy = cls()
+
+        if data.get("goal"):
+            strategy.goal = Goal(**data["goal"])
+        if data.get("arena"):
+            strategy.arena = Arena(**data["arena"])
+
+        strategy.insights = [Insight(**i) for i in data.get("insights", [])]
+        strategy.hypotheses = [Hypothesis(**h) for h in data.get("hypotheses", [])]
+        strategy.moves = [Move(**m) for m in data.get("moves", [])]
+        strategy.metrics = [Metric(**m) for m in data.get("metrics", [])]
+
+        return strategy
+
+    @classmethod
+    def from_json(cls, filepath: str) -> 'Strategy':
+        """Load strategy from JSON file"""
+        with open(filepath, 'r', encoding='utf-8') as f:
+            data = json.load(f)
+        return cls.from_dict(data)
+
+
+class SimulationEngine:
+    """Strategy simulation and scoring engine"""
+
+    @staticmethod
+    def calculate_move_score(move: Move) -> float:
+        """
+        Calculate move score using formula:
+        score = (impact × fit) × (1 - risk) / cost
+        """
+        if move.cost == 0:
+            return 0
+
+        score = (move.impact * move.fit) * (1 - move.risk) / (move.cost / 100000)
+        return round(score, 4)
+
+    @staticmethod
+    def simulate_strategy(strategy: Strategy) -> Dict[str, Any]:
+        """
+        Run simulation on complete strategy
+        Returns scores, rankings, and forecasts
+        """
+        results = {
+            "move_scores": [],
+            "total_impact": 0,
+            "metric_forecasts": [],
+            "recommendations": []
+        }
+
+        # Calculate scores for each move
+        for move in strategy.moves:
+            score = SimulationEngine.calculate_move_score(move)
+            results["move_scores"].append({
+                "id": move.id,
+                "text": move.text,
+                "score": score,
+                "impact": move.impact,
+                "fit": move.fit,
+                "risk": move.risk,
+                "cost": move.cost,
+                "linked_hypothesis": move.linked_hypothesis
+            })
+
+        # Sort by score
+        results["move_scores"].sort(key=lambda x: x["score"], reverse=True)
+
+        # Calculate total impact
+        total_score = sum(m["score"] for m in results["move_scores"])
+        results["total_impact"] = round(total_score, 4)
+
+        # Forecast main metric (from Goal) first
+        if strategy.goal:
+            linked_moves = []
+            linked_hypotheses = []
+            for move in strategy.moves:
+                for hyp in strategy.hypotheses:
+                    if hyp.id == move.linked_hypothesis and hyp.metric == strategy.goal.metric:
+                        linked_moves.append(move)
+                        if hyp.id not in linked_hypotheses:
+                            linked_hypotheses.append(hyp.id)
+
+            # Calculate forecast and contribution breakdown
+            moves_breakdown = []
+            for move in linked_moves:
+                move_score = SimulationEngine.calculate_move_score(move)
+                moves_breakdown.append({
+                    "id": move.id,
+                    "text": move.text,
+                    "score": move_score,
+                    "hypothesis": move.linked_hypothesis
+                })
+
+            moves_score = sum(m["score"] for m in moves_breakdown)
+            forecast = strategy.goal.baseline * (1 + moves_score)
+
+            results["metric_forecasts"].append({
+                "id": strategy.goal.metric,
+                "text": f"{strategy.goal.text} (MAIN GOAL)",
+                "baseline": strategy.goal.baseline,
+                "target": strategy.goal.target,
+                "forecast": round(forecast, 2),
+                "unit": strategy.goal.unit,
+                "gap_to_target": round(strategy.goal.target - forecast, 2),
+                "linked_moves": moves_breakdown,
+                "linked_hypotheses": linked_hypotheses,
+                "is_main": True
+            })
+
+        # Forecast supporting metrics
+        for metric in strategy.metrics:
+            # Find moves linked to this metric through hypotheses
+            linked_moves = []
+            linked_hypotheses = []
+            for move in strategy.moves:
+                for hyp in strategy.hypotheses:
+                    if hyp.id == move.linked_hypothesis and hyp.metric == metric.id:
+                        linked_moves.append(move)
+                        if hyp.id not in linked_hypotheses:
+                            linked_hypotheses.append(hyp.id)
+
+            # Calculate forecast and contribution breakdown
+            moves_breakdown = []
+            for move in linked_moves:
+                move_score = SimulationEngine.calculate_move_score(move)
+                moves_breakdown.append({
+                    "id": move.id,
+                    "text": move.text,
+                    "score": move_score,
+                    "hypothesis": move.linked_hypothesis
+                })
+
+            moves_score = sum(m["score"] for m in moves_breakdown)
+            forecast = metric.baseline * (1 + moves_score)
+
+            results["metric_forecasts"].append({
+                "id": metric.id,
+                "text": metric.text,
+                "baseline": metric.baseline,
+                "target": metric.target,
+                "forecast": round(forecast, 2),
+                "unit": metric.unit,
+                "gap_to_target": round(metric.target - forecast, 2),
+                "linked_moves": moves_breakdown,
+                "linked_hypotheses": linked_hypotheses,
+                "is_main": False
+            })
+
+        # Generate recommendations
+        if results["move_scores"]:
+            top_move = results["move_scores"][0]
+            if top_move["risk"] > 0.7:
+                results["recommendations"].append(f"⚠️ Top move '{top_move['text']}' has high risk ({top_move['risk']})")
+
+            high_cost_moves = [m for m in results["move_scores"] if m["cost"] > 100000]
+            if high_cost_moves:
+                results["recommendations"].append(f"💰 {len(high_cost_moves)} move(s) exceed 100k budget")
+
+        return results
+
+    @staticmethod
+    def create_results_dataframe(results: Dict[str, Any]) -> pd.DataFrame:
+        """Convert simulation results to pandas DataFrame"""
+        if not results.get("move_scores"):
+            return pd.DataFrame()
+
+        df = pd.DataFrame(results["move_scores"])
+        df = df[["id", "text", "score", "impact", "fit", "risk", "cost"]]
+        df.columns = ["ID", "Move", "Score", "Impact", "Fit", "Risk", "Cost"]
+        return df