Product Management AI Agent

# product/story_generator.py
from typing import List, Dict
import openai
from dataclasses import dataclass
import json

@dataclass
class UserStory:
    title: str
    description: str
    acceptance_criteria: List[str]
    priority: str
    effort: int  # Story points
    business_value: int  # 1-10
    dependencies: List[str]
    tags: List[str]

class AIStoryGenerator:
    def __init__(self, api_key: str):
        self.client = openai.OpenAI(api_key=api_key)
    
    def generate_story(self, feature_description: str, context: Dict) -> UserStory:
        """Generate user story from feature description"""
        
        prompt = f"""
You are a product manager creating a user story.

Feature: {feature_description}

Product Context:
- Target Users: {context.get('target_users', 'General users')}
- Product Type: {context.get('product_type', 'SaaS application')}
- Technical Stack: {context.get('tech_stack', 'Web application')}

Generate a user story in this JSON format:
{{
  "title": "As a [user type], I want [goal] so that [benefit]",
  "description": "Detailed description of the feature",
  "acceptance_criteria": [
    "Given [context], when [action], then [outcome]",
    "..."
  ],
  "priority": "high|medium|low",
  "effort": 1-13,  // Story points (Fibonacci)
  "business_value": 1-10,
  "dependencies": ["List of dependent stories or features"],
  "tags": ["Relevant tags"]
}}

Ensure acceptance criteria are specific, measurable, and testable.
"""
        
        response = self.client.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": "You are an expert product manager."},
                {"role": "user", "content": prompt}
            ],
            response_format={"type": "json_object"},
            temperature=0.7
        )
        
        story_data = json.loads(response.choices[0].message.content)
        
        return UserStory(
            title=story_data['title'],
            description=story_data['description'],
            acceptance_criteria=story_data['acceptance_criteria'],
            priority=story_data['priority'],
            effort=story_data['effort'],
            business_value=story_data['business_value'],
            dependencies=story_data.get('dependencies', []),
            tags=story_data.get('tags', [])
        )
    
    def generate_epic_breakdown(self, epic: str) -> List[UserStory]:
        """Break down an epic into individual user stories"""
        
        prompt = f"""
Break down this epic into 3-7 individual user stories:

Epic: {epic}

For each story, provide:
1. Title (user story format)
2. Description
3. 3-5 acceptance criteria
4. Priority
5. Estimated effort (story points)
6. Business value (1-10)
7. Dependencies
8. Tags

Return as JSON array.
"""
        
        response = self.client.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": "You are an expert product manager."},
                {"role": "user", "content": prompt}
            ],
            response_format={"type": "json_object"},
            temperature=0.7
        )
        
        data = json.loads(response.choices[0].message.content)
        
        return [
            UserStory(**story)
            for story in data.get('stories', [])
        ]
    
    def refine_story(self, story: UserStory, feedback: str) -> UserStory:
        """Refine story based on feedback"""
        
        prompt = f"""
Refine this user story based on feedback:

Original Story:
{json.dumps(story.__dict__, indent=2)}

Feedback: {feedback}

Provide improved version addressing the feedback.
"""
        
        response = self.client.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": "You are an expert product manager."},
                {"role": "user", "content": prompt}
            ],
            response_format={"type": "json_object"},
            temperature=0.7
        )
        
        refined_data = json.loads(response.choices[0].message.content)
        return UserStory(**refined_data)

# analytics/product_metrics.py
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from typing import Dict, List, Tuple
import psycopg2
from dataclasses import dataclass

@dataclass
class ProductMetrics:
    # Acquisition
    new_users: int
    activation_rate: float
    
    # Engagement
    dau: int  # Daily Active Users
    mau: int  # Monthly Active Users
    wau: int  # Weekly Active Users
    dau_mau_ratio: float  # Stickiness
    session_duration_avg: float
    sessions_per_user: float
    
    # Retention
    retention_day_1: float
    retention_day_7: float
    retention_day_30: float
    cohort_retention: Dict[str, List[float]]
    
    # Revenue
    mrr: float  # Monthly Recurring Revenue
    arr: float  # Annual Recurring Revenue
    arpu: float  # Average Revenue Per User
    ltv: float  # Lifetime Value
    cac: float  # Customer Acquisition Cost
    ltv_cac_ratio: float
    
    # Product
    feature_adoption: Dict[str, float]
    nps_score: float  # Net Promoter Score
    churn_rate: float

class ProductAnalytics:
    def __init__(self, db_connection: str):
        self.conn = psycopg2.connect(db_connection)
    
    def calculate_metrics(self, start_date: str, end_date: str) -> ProductMetrics:
        """Calculate all product metrics for date range"""
        
        # Acquisition metrics
        new_users = self._get_new_users(start_date, end_date)
        activation_rate = self._calculate_activation_rate(start_date, end_date)
        
        # Engagement metrics
        dau = self._get_dau(end_date)
        mau = self._get_mau(end_date)
        wau = self._get_wau(end_date)
        dau_mau_ratio = dau / mau if mau > 0 else 0
        
        session_stats = self._get_session_stats(start_date, end_date)
        
        # Retention metrics
        retention = self._calculate_retention(start_date)
        cohort_retention = self._calculate_cohort_retention()
        
        # Revenue metrics
        revenue_metrics = self._calculate_revenue_metrics(start_date, end_date)
        
        # Product metrics
        feature_adoption = self._calculate_feature_adoption(end_date)
        nps = self._calculate_nps(start_date, end_date)
        churn = self._calculate_churn_rate(start_date, end_date)
        
        return ProductMetrics(
            new_users=new_users,
            activation_rate=activation_rate,
            dau=dau,
            mau=mau,
            wau=wau,
            dau_mau_ratio=dau_mau_ratio,
            session_duration_avg=session_stats['avg_duration'],
            sessions_per_user=session_stats['sessions_per_user'],
            retention_day_1=retention['day_1'],
            retention_day_7=retention['day_7'],
            retention_day_30=retention['day_30'],
            cohort_retention=cohort_retention,
            mrr=revenue_metrics['mrr'],
            arr=revenue_metrics['arr'],
            arpu=revenue_metrics['arpu'],
            ltv=revenue_metrics['ltv'],
            cac=revenue_metrics['cac'],
            ltv_cac_ratio=revenue_metrics['ltv_cac_ratio'],
            feature_adoption=feature_adoption,
            nps_score=nps,
            churn_rate=churn
        )
    
    def _calculate_cohort_retention(self) -> Dict[str, List[float]]:
        """Calculate retention by cohort"""
        query = """
        WITH cohorts AS (
            SELECT 
                user_id,
                DATE_TRUNC('month', created_at) AS cohort_month
            FROM users
        ),
        user_activities AS (
            SELECT
                c.cohort_month,
                c.user_id,
                DATE_TRUNC('month', a.activity_date) AS activity_month,
                EXTRACT(MONTH FROM AGE(a.activity_date, c.cohort_month)) AS month_number
            FROM cohorts c
            LEFT JOIN user_activity a ON c.user_id = a.user_id
        )
        SELECT
            cohort_month,
            month_number,
            COUNT(DISTINCT user_id) AS active_users
        FROM user_activities
        GROUP BY cohort_month, month_number
        ORDER BY cohort_month, month_number
        """
        
        df = pd.read_sql(query, self.conn)
        
        # Pivot to cohort table
        cohort_table = df.pivot_table(
            index='cohort_month',
            columns='month_number',
            values='active_users'
        )
        
        # Calculate retention percentages
        cohort_retention = {}
        for cohort in cohort_table.index:
            cohort_size = cohort_table.loc[cohort, 0]
            retention_pct = (cohort_table.loc[cohort] / cohort_size * 100).tolist()
            cohort_retention[str(cohort)] = retention_pct
        
        return cohort_retention
    
    def _calculate_revenue_metrics(self, start_date: str, end_date: str) -> Dict:
        """Calculate all revenue-related metrics"""
        query = """
        WITH mrr_calc AS (
            SELECT SUM(subscription_amount) AS mrr
            FROM subscriptions
            WHERE status = 'active'
            AND DATE_TRUNC('month', current_period_start) = DATE_TRUNC('month', CURRENT_DATE)
        ),
        arpu_calc AS (
            SELECT 
                SUM(amount) / COUNT(DISTINCT user_id) AS arpu
            FROM transactions
            WHERE created_at BETWEEN %s AND %s
        ),
        ltv_calc AS (
            SELECT
                AVG(total_revenue / NULLIF(EXTRACT(MONTH FROM AGE(churn_date, created_at)), 0)) AS avg_monthly_value,
                AVG(EXTRACT(MONTH FROM AGE(COALESCE(churn_date, CURRENT_DATE), created_at))) AS avg_lifetime_months
            FROM users
        ),
        cac_calc AS (
            SELECT
                SUM(marketing_spend) / COUNT(DISTINCT user_id) AS cac
            FROM user_attribution
            WHERE created_at BETWEEN %s AND %s
        )
        SELECT
            m.mrr,
            m.mrr * 12 AS arr,
            a.arpu,
            l.avg_monthly_value * l.avg_lifetime_months AS ltv,
            c.cac
        FROM mrr_calc m
        CROSS JOIN arpu_calc a
        CROSS JOIN ltv_calc l
        CROSS JOIN cac_calc c
        """
        
        cursor = self.conn.cursor()
        cursor.execute(query, (start_date, end_date, start_date, end_date))
        result = cursor.fetchone()
        cursor.close()
        
        mrr, arr, arpu, ltv, cac = result
        
        return {
            'mrr': mrr or 0,
            'arr': arr or 0,
            'arpu': arpu or 0,
            'ltv': ltv or 0,
            'cac': cac or 0,
            'ltv_cac_ratio': (ltv / cac) if cac > 0 else 0
        }

# roadmap/prioritization.py
from typing import List, Dict
from dataclasses import dataclass
import pandas as pd

@dataclass
class Feature:
    id: str
    name: str
    description: str
    reach: int  # Number of users affected per quarter
    impact: float  # 0.25=minimal, 0.5=low, 1=medium, 2=high, 3=massive
    confidence: float  # 0.5=low, 0.8=medium, 1.0=high
    effort: int  # Person-months
    
    @property
    def rice_score(self) -> float:
        """Calculate RICE score: (Reach × Impact × Confidence) / Effort"""
        return (self.reach * self.impact * self.confidence) / self.effort

class RoadmapPrioritizer:
    def __init__(self):
        self.features: List[Feature] = []
    
    def add_feature(self, feature: Feature):
        """Add feature to roadmap"""
        self.features.append(feature)
    
    def prioritize_rice(self) -> pd.DataFrame:
        """Prioritize features using RICE framework"""
        data = []
        for feature in self.features:
            data.append({
                'id': feature.id,
                'name': feature.name,
                'reach': feature.reach,
                'impact': feature.impact,
                'confidence': feature.confidence,
                'effort': feature.effort,
                'rice_score': feature.rice_score
            })
        
        df = pd.DataFrame(data)
        df = df.sort_values('rice_score', ascending=False)
        df['rank'] = range(1, len(df) + 1)
        
        return df
    
    def prioritize_value_effort(self) -> pd.DataFrame:
        """2x2 matrix: Value vs Effort"""
        data = []
        for feature in self.features:
            value = feature.reach * feature.impact * feature.confidence
            
            # Categorize into quadrants
            if value > 1000 and feature.effort <= 3:
                quadrant = 'Quick Wins'
                priority = 1
            elif value > 1000 and feature.effort > 3:
                quadrant = 'Major Projects'
                priority = 2
            elif value <= 1000 and feature.effort <= 3:
                quadrant = 'Fill-ins'
                priority = 3
            else:
                quadrant = 'Time Sinks'
                priority = 4
            
            data.append({
                'id': feature.id,
                'name': feature.name,
                'value': value,
                'effort': feature.effort,
                'quadrant': quadrant,
                'priority': priority
            })
        
        df = pd.DataFrame(data)
        df = df.sort_values('priority')
        
        return df
    
    def generate_roadmap(self, quarters: int = 4) -> Dict[str, List[Feature]]:
        """Generate quarterly roadmap based on capacity"""
        # Sort by RICE score
        prioritized = self.prioritize_rice()
        
        # Team capacity (person-months per quarter)
        capacity_per_quarter = 12  # Adjust based on team size
        
        roadmap = {}
        current_quarter = 1
        remaining_capacity = capacity_per_quarter
        
        for _, row in prioritized.iterrows():
            feature = next(f for f in self.features if f.id == row['id'])
            
            if feature.effort <= remaining_capacity:
                quarter_key = f'Q{current_quarter}'
                if quarter_key not in roadmap:
                    roadmap[quarter_key] = []
                
                roadmap[quarter_key].append(feature)
                remaining_capacity -= feature.effort
            else:
                # Move to next quarter
                current_quarter += 1
                if current_quarter > quarters:
                    break
                
                quarter_key = f'Q{current_quarter}'
                roadmap[quarter_key] = [feature]
                remaining_capacity = capacity_per_quarter - feature.effort
        
        return roadmap

# experiments/ab_testing.py
import numpy as np
from scipy import stats
from typing import Dict, Tuple
from dataclasses import dataclass

@dataclass
class ABTestResult:
    control_conversion: float
    variant_conversion: float
    relative_improvement: float
    p_value: float
    is_significant: bool
    confidence_interval: Tuple[float, float]
    sample_size_control: int
    sample_size_variant: int
    statistical_power: float

class ABTestAnalyzer:
    def __init__(self, significance_level: float = 0.05):
        self.alpha = significance_level
    
    def analyze_test(self, 
                     control_conversions: int,
                     control_visitors: int,
                     variant_conversions: int,
                     variant_visitors: int) -> ABTestResult:
        """Analyze A/B test results"""
        
        # Calculate conversion rates
        control_rate = control_conversions / control_visitors
        variant_rate = variant_conversions / variant_visitors
        
        # Calculate relative improvement
        relative_improvement = (variant_rate - control_rate) / control_rate * 100
        
        # Two-proportion z-test
        p_value = self._two_proportion_ztest(
            control_conversions, control_visitors,
            variant_conversions, variant_visitors
        )
        
        # Statistical significance
        is_significant = p_value < self.alpha
        
        # Confidence interval
        ci = self._calculate_confidence_interval(
            variant_rate, control_rate,
            variant_visitors, control_visitors
        )
        
        # Statistical power
        power = self._calculate_power(
            control_rate, variant_rate,
            control_visitors, variant_visitors
        )
        
        return ABTestResult(
            control_conversion=control_rate,
            variant_conversion=variant_rate,
            relative_improvement=relative_improvement,
            p_value=p_value,
            is_significant=is_significant,
            confidence_interval=ci,
            sample_size_control=control_visitors,
            sample_size_variant=variant_visitors,
            statistical_power=power
        )
    
    def _two_proportion_ztest(self, 
                               control_conv: int, control_total: int,
                               variant_conv: int, variant_total: int) -> float:
        """Perform two-proportion z-test"""
        p1 = control_conv / control_total
        p2 = variant_conv / variant_total
        
        p_pool = (control_conv + variant_conv) / (control_total + variant_total)
        
        se = np.sqrt(p_pool * (1 - p_pool) * (1/control_total + 1/variant_total))
        z_score = (p2 - p1) / se
        
        p_value = 2 * (1 - stats.norm.cdf(abs(z_score)))
        
        return p_value
    
    def calculate_sample_size(self, 
                              baseline_rate: float,
                              mde: float,  # Minimum Detectable Effect
                              power: float = 0.8) -> int:
        """Calculate required sample size per variant"""
        alpha = self.alpha
        beta = 1 - power
        
        z_alpha = stats.norm.ppf(1 - alpha/2)
        z_beta = stats.norm.ppf(power)
        
        p1 = baseline_rate
        p2 = baseline_rate * (1 + mde)
        
        n = (z_alpha * np.sqrt(2 * p1 * (1-p1)) + 
             z_beta * np.sqrt(p1*(1-p1) + p2*(1-p2)))**2 / (p2-p1)**2
        
        return int(np.ceil(n))

# feedback/sentiment_analysis.py
from transformers import pipeline
from typing import List, Dict
import pandas as pd

class FeedbackAnalyzer:
    def __init__(self):
        self.sentiment_analyzer = pipeline(
            "sentiment-analysis",
            model="distilbert-base-uncased-finetuned-sst-2-english"
        )
        self.zero_shot_classifier = pipeline(
            "zero-shot-classification",
            model="facebook/bart-large-mnli"
        )
    
    def analyze_feedback(self, feedback_text: str) -> Dict:
        """Analyze user feedback"""
        
        # Sentiment analysis
        sentiment = self.sentiment_analyzer(feedback_text)[0]
        
        # Categorize feedback
        categories = [
            'bug report',
            'feature request',
            'usability issue',
            'performance complaint',
            'positive feedback',
            'question'
        ]
        
        classification = self.zero_shot_classifier(
            feedback_text,
            categories,
            multi_label=True
        )
        
        # Extract top categories
        top_categories = [
            {'category': label, 'score': score}
            for label, score in zip(classification['labels'], classification['scores'])
            if score > 0.5
        ]
        
        return {
            'text': feedback_text,
            'sentiment': sentiment['label'],
            'sentiment_score': sentiment['score'],
            'categories': top_categories
        }
    
    def aggregate_feedback(self, feedback_list: List[str]) -> pd.DataFrame:
        """Aggregate and analyze multiple feedback entries"""
        results = [self.analyze_feedback(fb) for fb in feedback_list]
        return pd.DataFrame(results)

{
  "maxTokens": 4000,
  "temperature": 0.5,
  "systemPrompt": "You are an AI-powered product management agent focused on data-driven decision making and strategic planning"
}