Spaces:

alie354
/

test

Runtime error

test / community_intelligence_layer.py
HF Space Deployer
test
5ff2222 over 1 year ago
37 kB
#!/usr/bin/env python3
"""
Community Intelligence Abstraction Layer
Enables cross-company learning while preserving privacy
"""
import os
import json
import hashlib
import secrets
from typing import Dict, List, Any, Optional, Tuple
from datetime import datetime, timedelta
from sqlalchemy import create_engine, Column, String, Integer, DateTime, Text, JSON, Boolean, Float, Index, func
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
from sqlalchemy.dialects.postgresql import UUID
import uuid
from collections import defaultdict, Counter
import statistics
=====================================================================
ANONYMIZED COMMUNITY DATABASE MODELS
=====================================================================
CommunityBase = declarative_base()
class AnonymizedCompanyProfile(CommunityBase):
tablename = "community_company_profiles"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)

# Anonymous identifiers
profile_hash = Column(String(64), unique=True, nullable=False)
cohort_id = Column(String(50), index=True)

# Generalized demographics (safe to aggregate)
stage_category = Column(String(50), index=True)  # idea, early, growth, scale
industry_category = Column(String(50), index=True)  # tech, commerce, services
team_size_range = Column(String(20), index=True)  # 1-3, 4-10, 11-25
funding_category = Column(String(30), index=True)  # unfunded, pre-seed, seed, series-a+
geography_region = Column(String(50), index=True)  # north_america, europe, asia

# Business model patterns
business_model_type = Column(String(50))  # subscription, marketplace, ecommerce
revenue_range = Column(String(30))  # 0-1k, 1k-5k, 5k-10k, 10k+

# Success indicators (anonymized)
success_score = Column(Float)  # Composite success metric 0-100
growth_trajectory = Column(String(30))  # declining, stable, growing, accelerating

# Behavioral patterns
decision_speed = Column(String(30))  # fast, moderate, deliberate
risk_tolerance = Column(String(30))  # conservative, moderate, aggressive
learning_style = Column(String(30))  # hands_on, research_heavy, community_driven

# Metadata
data_points_contributing = Column(Integer, default=1)
last_updated = Column(DateTime, default=datetime.utcnow)
created_at = Column(DateTime, default=datetime.utcnow)
class CommunityToolPattern(CommunityBase):
tablename = "community_tool_patterns"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)

# Tool information (public data)
tool_name = Column(String(255), index=True)
tool_category = Column(String(100), index=True)
alternative_tools = Column(JSON)  # What it competes with

# Context (anonymized)
cohort_context = Column(String(100), index=True)  # "early_tech_1-5_team"
stage_category = Column(String(50), index=True)
industry_category = Column(String(50), index=True)
team_size_range = Column(String(20), index=True)

# Adoption metrics
adoption_rate = Column(Float)  # % of similar companies that chose this
sample_size = Column(Integer)  # How many companies this represents
confidence_level = Column(String(20))  # low, medium, high based on sample size

# Satisfaction metrics
avg_satisfaction_score = Column(Float)  # 1-10 average
satisfaction_distribution = Column(JSON)  # {"1-3": 5%, "4-6": 15%, "7-10": 80%}
retention_rate = Column(Float)  # % still using after 6 months

# Usage patterns
typical_use_cases = Column(JSON)  # ["project_management", "documentation"]
integration_patterns = Column(JSON)  # What other tools used alongside
switching_patterns = Column(JSON)  # What tools people switch to/from

# Decision factors
selection_reasons = Column(JSON)  # Why people choose this tool
rejection_reasons = Column(JSON)  # Why people don't choose this tool
price_sensitivity = Column(String(30))  # high, medium, low

# Timing patterns
adoption_timeline_days = Column(Integer)  # Days from consideration to adoption
typical_team_size_when_adopted = Column(String(20))
typical_revenue_when_adopted = Column(String(30))

# Success correlation
companies_using_success_rate = Column(Float)  # Success rate of companies using this tool

# Metadata
last_updated = Column(DateTime, default=datetime.utcnow)
created_at = Column(DateTime, default=datetime.utcnow)
class CommunityDecisionPattern(CommunityBase):
tablename = "community_decision_patterns"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)

# Decision identification
decision_category = Column(String(100), index=True)  # hiring, pricing, product, etc.
decision_type = Column(String(100), index=True)  # specific decision within category
decision_context = Column(String(200))  # "first_hire_pre_seed_tech"

# Company context
cohort_context = Column(String(100), index=True)
stage_category = Column(String(50), index=True)
industry_category = Column(String(50), index=True)
team_size_range = Column(String(20), index=True)

# Decision options and outcomes
decision_options = Column(JSON)  # Available choices
chosen_option = Column(String(200))  # What was actually chosen
alternative_options = Column(JSON)  # What else was considered

# Success metrics
success_rate = Column(Float)  # % of times this decision led to positive outcomes
avg_time_to_outcome_days = Column(Integer)
outcome_satisfaction = Column(Float)  # 1-10 how happy with decision
would_repeat_rate = Column(Float)  # % who would make same decision again

# Context factors that influenced success
success_factors = Column(JSON)  # What made successful decisions work
failure_factors = Column(JSON)  # What caused unsuccessful decisions to fail
critical_timing_factors = Column(JSON)  # When this decision should be made

# Resource requirements
typical_budget_range = Column(String(50))
typical_time_investment = Column(String(50))
team_members_typically_involved = Column(JSON)

# Follow-up patterns
common_next_decisions = Column(JSON)  # What decisions typically follow this one
dependencies = Column(JSON)  # What other decisions this depends on

# Sample statistics
sample_size = Column(Integer)
confidence_level = Column(String(20))

# Metadata
last_updated = Column(DateTime, default=datetime.utcnow)
created_at = Column(DateTime, default=datetime.utcnow)
class CommunityJourneyPattern(CommunityBase):
tablename = "community_journey_patterns"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)

# Journey identification
journey_type = Column(String(100), index=True)  # "idea_to_mvp", "mvp_to_pmf", etc.
cohort_context = Column(String(100), index=True)

# Sequence patterns
typical_step_sequence = Column(JSON)  # Ordered list of typical steps
critical_milestones = Column(JSON)  # Must-hit milestones
optional_steps = Column(JSON)  # Steps that can be skipped
common_variations = Column(JSON)  # Alternative paths that also work

# Timing patterns
total_duration_days_p50 = Column(Integer)  # Median time
total_duration_days_p90 = Column(Integer)  # 90th percentile time
step_durations = Column(JSON)  # Time for each individual step

# Success patterns
success_rate_by_path = Column(JSON)  # Success rate for different sequences
accelerating_factors = Column(JSON)  # What speeds up the journey
blocking_factors = Column(JSON)  # What slows down the journey

# Resource patterns
typical_team_size_evolution = Column(JSON)  # How team grows during journey
typical_funding_pattern = Column(JSON)  # When/how much funding typically raised
key_tools_by_stage = Column(JSON)  # What tools are adopted when

# Outcome patterns
final_outcomes_distribution = Column(JSON)  # % success, pivot, failure
key_metrics_evolution = Column(JSON)  # How key metrics typically evolve

# Sample statistics
sample_size = Column(Integer)
confidence_level = Column(String(20))

# Metadata
last_updated = Column(DateTime, default=datetime.utcnow)
created_at = Column(DateTime, default=datetime.utcnow)
class CommunityMarketIntelligence(CommunityBase):
tablename = "community_market_intelligence"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)

# Market context
industry_category = Column(String(50), index=True)
geography_region = Column(String(50), index=True)
time_period = Column(String(50))  # "2024_q1", "2024_q2"

# Trend data
trend_type = Column(String(100), index=True)  # "tool_adoption", "funding_patterns"
trend_data = Column(JSON)  # The actual trend information

# Market insights
emerging_patterns = Column(JSON)  # New patterns being observed
declining_patterns = Column(JSON)  # Patterns becoming less common
seasonal_factors = Column(JSON)  # Time-based variations

# Benchmarks
industry_benchmarks = Column(JSON)  # Key metrics by industry
stage_benchmarks = Column(JSON)  # Key metrics by stage

# Sample statistics
sample_size = Column(Integer)
confidence_level = Column(String(20))

# Metadata
last_updated = Column(DateTime, default=datetime.utcnow)
created_at = Column(DateTime, default=datetime.utcnow)
=====================================================================
ANONYMIZATION ENGINE
=====================================================================
class AnonymizationEngine:
def init(self):
self.anonymization_rules = self._load_anonymization_rules()
def _load_anonymization_rules(self) -> Dict[str, Any]:
    """Load rules for anonymizing different data types"""
    return {
        "stage_mapping": {
            "idea": "idea",
            "pre-seed": "early", 
            "seed": "early",
            "series-a": "growth",
            "series-b": "growth", 
            "series-c": "scale",
            "series-d": "scale"
        },
        "industry_mapping": {
            "saas": "tech",
            "fintech": "tech",
            "healthtech": "tech",
            "ai": "tech",
            "ml": "tech",
            "edtech": "tech",
            "ecommerce": "commerce",
            "marketplace": "commerce",
            "retail": "commerce",
            "consulting": "services",
            "agency": "services"
        },
        "team_size_ranges": {
            (1, 3): "1-3",
            (4, 10): "4-10", 
            (11, 25): "11-25",
            (26, 50): "26-50",
            (51, 100): "51-100",
            (101, 1000): "100+"
        },
        "revenue_ranges": {
            (0, 1000): "0-1k",
            (1001, 5000): "1k-5k",
            (5001, 10000): "5k-10k",
            (10001, 25000): "10k-25k",
            (25001, 50000): "25k-50k",
            (50001, 100000): "50k-100k",
            (100001, float('inf')): "100k+"
        },
        "geography_mapping": {
            "united_states": "north_america",
            "canada": "north_america",
            "mexico": "north_america",
            "united_kingdom": "europe",
            "germany": "europe",
            "france": "europe",
            "netherlands": "europe",
            "spain": "europe",
            "italy": "europe",
            "china": "asia",
            "japan": "asia",
            "south_korea": "asia",
            "singapore": "asia",
            "india": "asia"
        }
    }

def anonymize_company_profile(self, company_data: Dict[str, Any]) -> Dict[str, Any]:
    """Convert company data to anonymous profile"""
    
    # Extract key attributes
    stage = company_data.get("stage", "unknown")
    industry = company_data.get("industry", "unknown")
    team_size = company_data.get("team_size", 1)
    
    anonymized = {
        "stage_category": self.anonymization_rules["stage_mapping"].get(stage, "unknown"),
        "industry_category": self.anonymization_rules["industry_mapping"].get(industry, "other"),
        "team_size_range": self._get_team_size_range(team_size),
        "funding_category": self._get_funding_category(company_data.get("funding_raised", "0")),
        "geography_region": self._get_geography_region(company_data.get("location", "unknown")),
        "business_model_type": self._generalize_business_model(company_data.get("business_model", "")),
        "revenue_range": self._get_revenue_range(company_data.get("revenue", 0))
    }
    
    # Generate cohort ID for similar companies
    anonymized["cohort_id"] = self._generate_cohort_id(anonymized)
    
    # Generate unique hash for this profile
    anonymized["profile_hash"] = self._create_hash(anonymized)
    
    return anonymized

def anonymize_tool_decision(self, tool_data: Dict[str, Any], company_context: Dict[str, Any]) -> Dict[str, Any]:
    """Convert tool decision to anonymous pattern"""
    
    company_anon = self.anonymize_company_profile(company_context)
    
    return {
        "tool_name": tool_data.get("tool_name"),  # Tools are public info
        "tool_category": tool_data.get("tool_category"),
        "cohort_context": company_anon["cohort_id"],
        "stage_category": company_anon["stage_category"],
        "industry_category": company_anon["industry_category"],
        "team_size_range": company_anon["team_size_range"],
        "satisfaction_score": tool_data.get("satisfaction_score", 5),
        "use_cases": self._categorize_use_cases(tool_data.get("use_cases", [])),
        "selection_reasons": self._categorize_reasons(tool_data.get("why_chosen", "")),
        "alternatives_considered": tool_data.get("alternatives_evaluated", [])
    }

def anonymize_decision(self, decision_data: Dict[str, Any], company_context: Dict[str, Any]) -> Dict[str, Any]:
    """Convert business decision to anonymous pattern"""
    
    company_anon = self.anonymize_company_profile(company_context)
    
    return {
        "decision_category": self._categorize_decision(decision_data.get("decision_type", "")),
        "decision_type": decision_data.get("decision_type"),
        "cohort_context": company_anon["cohort_id"],
        "stage_category": company_anon["stage_category"],
        "industry_category": company_anon["industry_category"],
        "team_size_range": company_anon["team_size_range"],
        "chosen_option": decision_data.get("decision_title", ""),
        "alternative_options": decision_data.get("alternatives_considered", []),
        "implementation_success": decision_data.get("implementation_status") == "completed",
        "outcome_satisfaction": self._estimate_satisfaction(decision_data),
        "time_to_outcome_days": self._calculate_implementation_time(decision_data)
    }

def _get_team_size_range(self, team_size: int) -> str:
    for (min_size, max_size), range_str in self.anonymization_rules["team_size_ranges"].items():
        if min_size <= team_size <= max_size:
            return range_str
    return "unknown"

def _get_revenue_range(self, revenue: float) -> str:
    for (min_rev, max_rev), range_str in self.anonymization_rules["revenue_ranges"].items():
        if min_rev <= revenue <= max_rev:
            return range_str
    return "unknown"

def _get_funding_category(self, funding_str: str) -> str:
    if not funding_str or funding_str == "0":
        return "unfunded"
    elif "pre-seed" in funding_str.lower():
        return "pre-seed"
    elif "seed" in funding_str.lower() and "series" not in funding_str.lower():
        return "seed"
    else:
        return "series-a+"

def _get_geography_region(self, location: str) -> str:
    location_lower = location.lower()
    for country, region in self.anonymization_rules["geography_mapping"].items():
        if country in location_lower:
            return region
    return "other"

def _generalize_business_model(self, business_model: str) -> str:
    if not business_model:
        return "unknown"
    
    model_lower = business_model.lower()
    if any(word in model_lower for word in ["subscription", "saas", "recurring"]):
        return "subscription"
    elif any(word in model_lower for word in ["marketplace", "platform", "commission"]):
        return "marketplace"
    elif any(word in model_lower for word in ["ecommerce", "retail", "selling"]):
        return "ecommerce"
    elif any(word in model_lower for word in ["consulting", "service", "agency"]):
        return "services"
    else:
        return "other"

def _categorize_decision(self, decision_type: str) -> str:
    decision_lower = decision_type.lower()
    if any(word in decision_lower for word in ["hire", "hiring", "team", "employee"]):
        return "hiring"
    elif any(word in decision_lower for word in ["tool", "software", "platform"]):
        return "tools"
    elif any(word in decision_lower for word in ["price", "pricing", "monetiz"]):
        return "pricing"
    elif any(word in decision_lower for word in ["product", "feature", "development"]):
        return "product"
    elif any(word in decision_lower for word in ["marketing", "growth", "acquisition"]):
        return "marketing"
    elif any(word in decision_lower for word in ["funding", "investment", "raise"]):
        return "funding"
    else:
        return "strategy"

def _categorize_use_cases(self, use_cases: List[str]) -> List[str]:
    categories = []
    for use_case in use_cases:
        use_case_lower = str(use_case).lower()
        if any(word in use_case_lower for word in ["project", "task", "management"]):
            categories.append("project_management")
        elif any(word in use_case_lower for word in ["communication", "chat", "message"]):
            categories.append("communication")
        elif any(word in use_case_lower for word in ["document", "note", "wiki"]):
            categories.append("documentation")
        elif any(word in use_case_lower for word in ["design", "prototype", "visual"]):
            categories.append("design")
        elif any(word in use_case_lower for word in ["analytics", "data", "metrics"]):
            categories.append("analytics")
    return list(set(categories))

def _categorize_reasons(self, reasons: str) -> List[str]:
    if not reasons:
        return []
    
    categories = []
    reasons_lower = reasons.lower()
    
    if any(word in reasons_lower for word in ["cost", "cheap", "affordable", "free"]):
        categories.append("cost_effective")
    if any(word in reasons_lower for word in ["easy", "simple", "intuitive", "user-friendly"]):
        categories.append("ease_of_use")
    if any(word in reasons_lower for word in ["feature", "functionality", "powerful"]):
        categories.append("features")
    if any(word in reasons_lower for word in ["integration", "api", "connect"]):
        categories.append("integration")
    if any(word in reasons_lower for word in ["team", "collaboration", "sharing"]):
        categories.append("collaboration")
    if any(word in reasons_lower for word in ["recommended", "popular", "everyone"]):
        categories.append("social_proof")
    
    return categories

def _estimate_satisfaction(self, decision_data: Dict) -> float:
    # Simple heuristic based on implementation status and feedback
    if decision_data.get("implementation_status") == "completed":
        return 8.0
    elif decision_data.get("implementation_status") == "in_progress":
        return 7.0
    else:
        return 6.0

def _calculate_implementation_time(self, decision_data: Dict) -> int:
    # Extract timeline or estimate based on decision type
    timeline_weeks = decision_data.get("timeline_weeks", 4)
    return timeline_weeks * 7  # Convert to days

def _generate_cohort_id(self, anonymized_data: Dict) -> str:
    """Generate ID for grouping similar companies"""
    cohort_key = f"{anonymized_data['stage_category']}_{anonymized_data['industry_category']}_{anonymized_data['team_size_range']}"
    return cohort_key

def _create_hash(self, data: Dict) -> str:
    """Create deterministic hash of data"""
    sorted_data = json.dumps(data, sort_keys=True, default=str)
    return hashlib.sha256(sorted_data.encode()).hexdigest()
=====================================================================
COMMUNITY INTELLIGENCE ENGINE
=====================================================================
class CommunityIntelligenceEngine:
def init(self, db_session: Session):
self.db = db_session
self.anonymizer = AnonymizationEngine()
def contribute_company_data(self, company_data: Dict[str, Any]) -> bool:
    """Add company data to community intelligence"""
    try:
        anonymized = self.anonymizer.anonymize_company_profile(company_data)
        
        # Check if profile already exists
        existing = self.db.query(AnonymizedCompanyProfile).filter(
            AnonymizedCompanyProfile.profile_hash == anonymized["profile_hash"]
        ).first()
        
        if existing:
            existing.data_points_contributing += 1
            existing.last_updated = datetime.utcnow()
        else:
            new_profile = AnonymizedCompanyProfile(**anonymized)
            self.db.add(new_profile)
        
        self.db.commit()
        return True
        
    except Exception as e:
        print(f"Error contributing company data: {e}")
        return False

def contribute_tool_decision(self, tool_data: Dict[str, Any], company_context: Dict[str, Any]) -> bool:
    """Add tool decision to community intelligence"""
    try:
        anonymized = self.anonymizer.anonymize_tool_decision(tool_data, company_context)
        
        # Find or create tool pattern
        existing = self.db.query(CommunityToolPattern).filter(
            CommunityToolPattern.tool_name == anonymized["tool_name"],
            CommunityToolPattern.cohort_context == anonymized["cohort_context"]
        ).first()
        
        if existing:
            # Update aggregated metrics
            self._update_tool_metrics(existing, anonymized)
        else:
            # Create new pattern
            new_pattern = CommunityToolPattern(
                tool_name=anonymized["tool_name"],
                tool_category=anonymized["tool_category"],
                cohort_context=anonymized["cohort_context"],
                stage_category=anonymized["stage_category"],
                industry_category=anonymized["industry_category"],
                team_size_range=anonymized["team_size_range"],
                sample_size=1,
                avg_satisfaction_score=anonymized["satisfaction_score"],
                adoption_rate=1.0,  # Will be calculated properly with more data
                typical_use_cases=anonymized["use_cases"],
                selection_reasons=anonymized["selection_reasons"],
                alternative_tools=anonymized["alternatives_considered"],
                confidence_level=self._calculate_confidence(1)
            )
            self.db.add(new_pattern)
        
        self.db.commit()
        return True
        
    except Exception as e:
        print(f"Error contributing tool decision: {e}")
        return False

def contribute_business_decision(self, decision_data: Dict[str, Any], company_context: Dict[str, Any]) -> bool:
    """Add business decision to community intelligence"""
    try:
        anonymized = self.anonymizer.anonymize_decision(decision_data, company_context)
        
        # Create decision context key
        decision_context = f"{anonymized['decision_category']}_{anonymized['cohort_context']}"
        
        # Find or create decision pattern
        existing = self.db.query(CommunityDecisionPattern).filter(
            CommunityDecisionPattern.decision_context == decision_context,
            CommunityDecisionPattern.chosen_option == anonymized["chosen_option"]
        ).first()
        
        if existing:
            self._update_decision_metrics(existing, anonymized)
        else:
            new_pattern = CommunityDecisionPattern(
                decision_category=anonymized["decision_category"],
                decision_type=anonymized["decision_type"],
                decision_context=decision_context,
                cohort_context=anonymized["cohort_context"],
                stage_category=anonymized["stage_category"],
                industry_category=anonymized["industry_category"],
                team_size_range=anonymized["team_size_range"],
                chosen_option=anonymized["chosen_option"],
                alternative_options=anonymized["alternative_options"],
                sample_size=1,
                success_rate=1.0 if anonymized["implementation_success"] else 0.0,
                outcome_satisfaction=anonymized["outcome_satisfaction"],
                avg_time_to_outcome_days=anonymized["time_to_outcome_days"],
                confidence_level=self._calculate_confidence(1)
            )
            self.db.add(new_pattern)
        
        self.db.commit()
        return True
        
    except Exception as e:
        print(f"Error contributing business decision: {e}")
        return False

def get_tool_intelligence(self, context: Dict[str, Any], tool_category: str = None) -> Dict[str, Any]:
    """Get tool adoption intelligence for similar companies"""
    
    # Anonymize context for matching
    anonymized_context = self.anonymizer.anonymize_company_profile(context)
    cohort_id = anonymized_context["cohort_id"]
    
    # Query for tool patterns
    query = self.db.query(CommunityToolPattern).filter(
        CommunityToolPattern.cohort_context == cohort_id
    )
    
    if tool_category:
        query = query.filter(CommunityToolPattern.tool_category == tool_category)
    
    patterns = query.order_by(CommunityToolPattern.adoption_rate.desc()).limit(10).all()
    
    return {
        "context": anonymized_context,
        "tool_recommendations": [
            {
                "tool_name": p.tool_name,
                "tool_category": p.tool_category,
                "adoption_rate": p.adoption_rate,
                "satisfaction_score": p.avg_satisfaction_score,
                "sample_size": p.sample_size,
                "confidence": p.confidence_level,
                "typical_use_cases": p.typical_use_cases,
                "selection_reasons": p.selection_reasons,
                "alternatives": p.alternative_tools
            } for p in patterns
        ],
        "total_companies_in_cohort": self._get_cohort_size(cohort_id)
    }

def get_decision_intelligence(self, context: Dict[str, Any], decision_category: str) -> Dict[str, Any]:
    """Get decision intelligence for similar companies"""
    
    anonymized_context = self.anonymizer.anonymize_company_profile(context)
    cohort_id = anonymized_context["cohort_id"]
    
    # Query for decision patterns
    patterns = self.db.query(CommunityDecisionPattern).filter(
        CommunityDecisionPattern.cohort_context == cohort_id,
        CommunityDecisionPattern.decision_category == decision_category
    ).order_by(CommunityDecisionPattern.success_rate.desc()).limit(10).all()
    
    return {
        "context": anonymized_context,
        "decision_category": decision_category,
        "decision_patterns": [
            {
                "decision_type": p.decision_type,
                "chosen_option": p.chosen_option,
                "success_rate": p.success_rate,
                "sample_size": p.sample_size,
                "confidence": p.confidence_level,
                "avg_satisfaction": p.outcome_satisfaction,
                "typical_timeline_days": p.avg_time_to_outcome_days,
                "alternatives_considered": p.alternative_options
            } for p in patterns
        ],
        "total_companies_in_cohort": self._get_cohort_size(cohort_id)
    }

def get_market_intelligence(self, context: Dict[str, Any]) -> Dict[str, Any]:
    """Get broader market intelligence"""
    
    anonymized_context = self.anonymizer.anonymize_company_profile(context)
    
    # Get industry-wide patterns
    industry_tools = self.db.query(CommunityToolPattern).filter(
        CommunityToolPattern.industry_category == anonymized_context["industry_category"]
    ).order_by(CommunityToolPattern.adoption_rate.desc()).limit(5).all()
    
    stage_decisions = self.db.query(CommunityDecisionPattern).filter(
        CommunityDecisionPattern.stage_category == anonymized_context["stage_category"]
    ).order_by(CommunityDecisionPattern.success_rate.desc()).limit(5).all()
    
    return {
        "industry_trending_tools": [
            {
                "tool_name": t.tool_name,
                "adoption_rate": t.adoption_rate,
                "category": t.tool_category
            } for t in industry_tools
        ],
        "stage_successful_decisions": [
            {
                "decision_type": d.decision_type,
                "success_rate": d.success_rate,
                "sample_size": d.sample_size
            } for d in stage_decisions
        ]
    }

def generate_community_insights_for_prompt(self, user_query: str, context: Dict[str, Any]) -> str:
    """Generate community insights to enhance AI prompts"""
    
    # Classify query to get relevant intelligence
    query_lower = user_query.lower()
    
    insights_text = ""
    
    # Tool-related query
    if any(word in query_lower for word in ["tool", "software", "platform", "app", "service"]):
        tool_intel = self.get_tool_intelligence(context)
        if tool_intel["tool_recommendations"]:
            insights_text += "\nCOMMUNITY TOOL INTELLIGENCE:\n"
            for tool in tool_intel["tool_recommendations"][:3]:
                insights_text += f"• {tool['tool_name']}: {tool['adoption_rate']:.0%} adoption, {tool['satisfaction_score']:.1f}/10 satisfaction ({tool['sample_size']} companies)\n"
    
    # Hiring-related query
    elif any(word in query_lower for word in ["hire", "hiring", "team", "employee", "recruit"]):
        decision_intel = self.get_decision_intelligence(context, "hiring")
        if decision_intel["decision_patterns"]:
            insights_text += "\nCOMMUNITY HIRING INTELLIGENCE:\n"
            for pattern in decision_intel["decision_patterns"][:2]:
                insights_text += f"• {pattern['chosen_option']}: {pattern['success_rate']:.0%} success rate ({pattern['sample_size']} companies)\n"
    
    # Funding-related query
    elif any(word in query_lower for word in ["funding", "investment", "raise", "investor"]):
        decision_intel = self.get_decision_intelligence(context, "funding")
        if decision_intel["decision_patterns"]:
            insights_text += "\nCOMMUNITY FUNDING INTELLIGENCE:\n"
            for pattern in decision_intel["decision_patterns"][:2]:
                insights_text += f"• {pattern['chosen_option']}: {pattern['success_rate']:.0%} success rate, avg {pattern['typical_timeline_days']} days\n"
    
    # General market intelligence
    market_intel = self.get_market_intelligence(context)
    if market_intel["industry_trending_tools"]:
        insights_text += f"\nINDUSTRY TRENDS ({context.get('industry', 'tech').upper()}):\n"
        for tool in market_intel["industry_trending_tools"][:2]:
            insights_text += f"• {tool['tool_name']}: {tool['adoption_rate']:.0%} adoption in {tool['category']}\n"
    
    return insights_text

def _update_tool_metrics(self, existing_pattern: CommunityToolPattern, new_data: Dict[str, Any]):
    """Update aggregated tool metrics with new data point"""
    
    # Update sample size
    old_size = existing_pattern.sample_size
    new_size = old_size + 1
    existing_pattern.sample_size = new_size
    
    # Update satisfaction score (running average)
    old_satisfaction = existing_pattern.avg_satisfaction_score
    new_satisfaction = new_data["satisfaction_score"]
    existing_pattern.avg_satisfaction_score = (old_satisfaction * old_size + new_satisfaction) / new_size
    
    # Update confidence level
    existing_pattern.confidence_level = self._calculate_confidence(new_size)
    
    # Merge use cases and reasons
    existing_use_cases = set(existing_pattern.typical_use_cases or [])
    new_use_cases = set(new_data["use_cases"])
    existing_pattern.typical_use_cases = list(existing_use_cases.union(new_use_cases))
    
    existing_reasons = set(existing_pattern.selection_reasons or [])
    new_reasons = set(new_data["selection_reasons"])
    existing_pattern.selection_reasons = list(existing_reasons.union(new_reasons))
    
    existing_pattern.last_updated = datetime.utcnow()

def _update_decision_metrics(self, existing_pattern: CommunityDecisionPattern, new_data: Dict[str, Any]):
    """Update aggregated decision metrics with new data point"""
    
    # Update sample size
    old_size = existing_pattern.sample_size
    new_size = old_size + 1
    existing_pattern.sample_size = new_size
    
    # Update success rate
    old_success_count = existing_pattern.success_rate * old_size
    new_success = 1 if new_data["implementation_success"] else 0
    existing_pattern.success_rate = (old_success_count + new_success) / new_size
    
    # Update satisfaction (running average)
    old_satisfaction = existing_pattern.outcome_satisfaction
    new_satisfaction = new_data["outcome_satisfaction"]
    existing_pattern.outcome_satisfaction = (old_satisfaction * old_size + new_satisfaction) / new_size
    
    # Update timeline (running average)
    old_timeline = existing_pattern.avg_time_to_outcome_days
    new_timeline = new_data["time_to_outcome_days"]
    existing_pattern.avg_time_to_outcome_days = int((old_timeline * old_size + new_timeline) / new_size)
    
    # Update confidence
    existing_pattern.confidence_level = self._calculate_confidence(new_size)
    
    existing_pattern.last_updated = datetime.utcnow()

def _calculate_confidence(self, sample_size: int) -> str:
    """Calculate confidence level based on sample size"""
    if sample_size < 5:
        return "low"
    elif sample_size < 20:
        return "medium"
    else:
        return "high"

def _get_cohort_size(self, cohort_id: str) -> int:
    """Get total number of companies in a cohort"""
    return self.db.query(AnonymizedCompanyProfile).filter(
        AnonymizedCompanyProfile.cohort_id == cohort_id
    ).count()
=====================================================================
INTEGRATION WITH EXISTING SYSTEM
=====================================================================
def create_community_tables():
"""Create community intelligence tables"""
COMMUNITY_DB_URL = os.getenv("COMMUNITY_DB_URL", os.getenv("ANONYMIZED_DB_URL", "postgresql://user:pass@localhost:5432/community_db"))
engine = create_engine(COMMUNITY_DB_URL)
CommunityBase.metadata.create_all(bind=engine)
print("✅ Community intelligence tables created")
def get_community_session():
"""Get database session for community intelligence"""
COMMUNITY_DB_URL = os.getenv("COMMUNITY_DB_URL", os.getenv("ANONYMIZED_DB_URL", "postgresql://user:pass@localhost:5432/community_db"))
engine = create_engine(COMMUNITY_DB_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
return SessionLocal()
Example usage functions
def enhance_prompt_with_community_intelligence(user_query: str, company_context: Dict[str, Any]) -> str:
"""Enhance AI prompt with community intelligence"""
try:
    with get_community_session() as db:
        community_engine = CommunityIntelligenceEngine(db)
        community_insights = community_engine.generate_community_insights_for_prompt(user_query, company_context)
    
    if community_insights:
        return f"""
COMPANY CONTEXT: {company_context.get('name', 'Startup')} - {company_context.get('stage', 'pre-seed')} {company_context.get('industry', 'tech')}
{community_insights}
USER QUERY: {user_query}
Provide advice that incorporates relevant community intelligence while being specific to this company's situation.
"""
else:
return f"""
COMPANY CONTEXT: {company_context.get('name', 'Startup')} - {company_context.get('stage', 'pre-seed')} {company_context.get('industry', 'tech')}
USER QUERY: {user_query}
Provide personalized startup advice.
"""
except Exception as e:
    print(f"Community intelligence error: {e}")
    return f"""
COMPANY CONTEXT: {company_context.get('name', 'Startup')} - {company_context.get('stage', 'pre-seed')} {company_context.get('industry', 'tech')}
USER QUERY: {user_query}
Provide personalized startup advice.
"""
Background task to contribute data to community intelligence
async def contribute_to_community_intelligence(company_data: Dict[str, Any], interaction_data: Dict[str, Any] = None, tool_data: Dict[str, Any] = None, decision_data: Dict[str, Any] = None):
"""Background task to contribute data to community intelligence"""
try:
    with get_community_session() as db:
        community_engine = CommunityIntelligenceEngine(db)
        
        # Always contribute company profile
        community_engine.contribute_company_data(company_data)
        
        # Contribute specific data if provided
        if tool_data:
            community_engine.contribute_tool_decision(tool_data, company_data)
        
        if decision_data:
            community_engine.contribute_business_decision(decision_data, company_data)
        
        print(f"✅ Contributed data to community intelligence for {company_data.get('stage')} {company_data.get('industry')} company")

except Exception as e:
    print(f"❌ Failed to contribute to community intelligence: {e}")