Community Analytics

Overview

The Community Analytics module provides comprehensive tracking and analysis capabilities for community health, member participation, decision patterns, and collaboration effectiveness. This module helps monitor and optimize community performance.

Status: βœ… The CommunityAnalytics class is now available in the public API.

CommunityAnalytics Class

Initialization

class CommunityAnalytics:
    def __init__(self, community_manager=None)

Parameters:

  • community_manager (Optional): Reference to the community manager instance

Key Features

  1. Decision Analytics - Track decision patterns and effectiveness

  2. Participation Metrics - Monitor member engagement and contributions

  3. Health Metrics - Assess community health and vitality

  4. Collaboration Effectiveness - Measure collaboration quality and outcomes

Methods

get_decision_analytics

Get comprehensive decision analytics for a community.

def get_decision_analytics(
    self,
    community_id: str,
    time_range_days: int = 30
) -> Dict[str, Any]

Parameters:

  • community_id (str): ID of the community

  • time_range_days (int): Time range for analytics in days (default: 30)

Returns: Dictionary containing:

  • total_decisions: Total number of decisions

  • approved: Number of approved decisions

  • rejected: Number of rejected decisions

  • pending: Number of pending decisions

  • avg_decision_time: Average time to reach a decision (hours)

  • decision_types: Distribution of decision types

  • approval_by_type: Approval rates by decision type

  • participation_rate: Average member participation in voting

  • consensus_level: Average consensus level achieved

Example:

analytics = CommunityAnalytics(community_manager)
decision_metrics = analytics.get_decision_analytics("community_001", time_range_days=60)

print(f"Total decisions: {decision_metrics['total_decisions']}")
print(f"Approval rate: {decision_metrics['approved'] / decision_metrics['total_decisions']:.2%}")
print(f"Average decision time: {decision_metrics['avg_decision_time']:.1f} hours")

get_participation_metrics

Get member participation metrics for a community.

def get_participation_metrics(
    self,
    community_id: str,
    time_range_days: int = 30
) -> Dict[str, Any]

Parameters:

  • community_id (str): ID of the community

  • time_range_days (int): Time range for metrics in days (default: 30)

Returns: Dictionary containing:

  • active_members: Number of active members

  • inactive_members: Number of inactive members

  • participation_rate: Overall participation rate

  • avg_contributions: Average contributions per member

  • top_contributors: List of top contributing members

  • engagement_score: Community engagement score

  • participation_by_role: Participation breakdown by role

Example:

participation = analytics.get_participation_metrics("community_001")

print(f"Active members: {participation['active_members']}")
print(f"Participation rate: {participation['participation_rate']:.2%}")
print(f"Top contributors: {participation['top_contributors']}")

get_community_health_metrics

Get overall community health metrics.

def get_community_health_metrics(
    self,
    community_id: str
) -> Dict[str, Any]

Parameters:

  • community_id (str): ID of the community

Returns: Dictionary containing:

  • health_score: Overall community health score (0-100)

  • vitality: Community vitality indicator

  • growth_rate: Member growth rate

  • retention_rate: Member retention rate

  • activity_level: Activity level indicator

  • collaboration_quality: Quality of collaboration

  • decision_effectiveness: Decision-making effectiveness

  • resource_utilization: Resource utilization rate

  • recommendations: List of recommendations for improvement

Example:

health = analytics.get_community_health_metrics("community_001")

print(f"Health score: {health['health_score']}/100")
print(f"Activity level: {health['activity_level']}")
print(f"Recommendations: {health['recommendations']}")

get_collaboration_effectiveness

Measure collaboration effectiveness within the community.

def get_collaboration_effectiveness(
    self,
    community_id: str,
    time_range_days: int = 30
) -> Dict[str, Any]

Parameters:

  • community_id (str): ID of the community

  • time_range_days (int): Time range for analysis in days (default: 30)

Returns: Dictionary containing:

  • collaboration_sessions: Number of collaboration sessions

  • avg_session_duration: Average session duration

  • session_completion_rate: Session completion rate

  • collaboration_score: Overall collaboration effectiveness score

  • resource_sharing_rate: Rate of resource sharing

  • knowledge_sharing_score: Knowledge sharing effectiveness

  • communication_quality: Communication quality metrics

Example:

effectiveness = analytics.get_collaboration_effectiveness("community_001")

print(f"Collaboration score: {effectiveness['collaboration_score']}/100")
print(f"Sessions completed: {effectiveness['collaboration_sessions']}")
print(f"Resource sharing rate: {effectiveness['resource_sharing_rate']:.2%}")

get_member_analytics

Get detailed analytics for a specific member.

def get_member_analytics(
    self,
    community_id: str,
    member_id: str,
    time_range_days: int = 30
) -> Dict[str, Any]

Parameters:

  • community_id (str): ID of the community

  • member_id (str): ID of the member

  • time_range_days (int): Time range for analytics in days (default: 30)

Returns: Dictionary containing:

  • contribution_score: Member’s contribution score

  • participation_rate: Member’s participation rate

  • decisions_proposed: Number of decisions proposed

  • votes_cast: Number of votes cast

  • resources_shared: Number of resources shared

  • collaboration_sessions: Number of sessions participated in

  • reputation_score: Member’s reputation score

  • specializations: Member’s areas of expertise

  • activity_timeline: Activity timeline data

Example:

member_stats = analytics.get_member_analytics("community_001", "member_001")

print(f"Contribution score: {member_stats['contribution_score']}")
print(f"Decisions proposed: {member_stats['decisions_proposed']}")
print(f"Resources shared: {member_stats['resources_shared']}")

get_trend_analysis

Get trend analysis for community metrics over time.

def get_trend_analysis(
    self,
    community_id: str,
    metrics: List[str],
    time_range_days: int = 90
) -> Dict[str, List[Any]]

Parameters:

  • community_id (str): ID of the community

  • metrics (List[str]): List of metrics to analyze

  • time_range_days (int): Time range for analysis in days (default: 90)

Returns: Dictionary mapping metric names to time-series data

Available Metrics:

  • member_count: Number of members over time

  • activity_level: Activity level over time

  • decision_rate: Decision-making rate over time

  • participation_rate: Participation rate over time

  • resource_creation: Resource creation rate over time

Example:

trends = analytics.get_trend_analysis(
    "community_001",
    metrics=["member_count", "activity_level", "decision_rate"],
    time_range_days=90
)

# Plot trends
import matplotlib.pyplot as plt

plt.plot(trends['member_count'], label='Members')
plt.plot(trends['activity_level'], label='Activity')
plt.legend()
plt.show()

Analytics Metrics Reference

Health Score Components

The community health score (0-100) is calculated from:

  1. Activity Level (25%) - Recent activity frequency and intensity

  2. Participation Rate (25%) - Member engagement in decisions and discussions

  3. Collaboration Quality (20%) - Effectiveness of collaborative efforts

  4. Decision Effectiveness (15%) - Quality and timeliness of decisions

  5. Resource Utilization (15%) - Usage of shared resources and knowledge

Vitality Indicators

  • Growing: Increasing membership and activity

  • Stable: Consistent membership and activity

  • Declining: Decreasing membership or activity

  • Inactive: Minimal recent activity

Engagement Score Calculation

Engagement Score = (
    0.3 Γ— Participation Rate +
    0.3 Γ— Contribution Frequency +
    0.2 Γ— Resource Sharing +
    0.2 Γ— Collaboration Sessions
) Γ— 100

Usage Examples

Complete Analytics Report

async def generate_community_report(community_id: str):
    """Generate a comprehensive community analytics report."""
    
    analytics = CommunityAnalytics(community_manager)
    
    # Get all metrics
    decision_metrics = analytics.get_decision_analytics(community_id, time_range_days=30)
    participation = analytics.get_participation_metrics(community_id, time_range_days=30)
    health = analytics.get_community_health_metrics(community_id)
    effectiveness = analytics.get_collaboration_effectiveness(community_id, time_range_days=30)
    
    # Generate report
    report = {
        "community_id": community_id,
        "report_date": datetime.utcnow().isoformat(),
        "summary": {
            "health_score": health['health_score'],
            "vitality": health['vitality'],
            "active_members": participation['active_members'],
            "total_decisions": decision_metrics['total_decisions'],
            "collaboration_sessions": effectiveness['collaboration_sessions']
        },
        "decisions": decision_metrics,
        "participation": participation,
        "health": health,
        "collaboration": effectiveness,
        "recommendations": health['recommendations']
    }
    
    return report

Real-time Monitoring

async def monitor_community_health(community_id: str):
    """Monitor community health in real-time."""
    
    analytics = CommunityAnalytics(community_manager)
    
    while True:
        health = analytics.get_community_health_metrics(community_id)
        
        # Alert on low health score
        if health['health_score'] < 50:
            print(f"⚠️ WARNING: Community health score is low: {health['health_score']}")
            print(f"Recommendations: {health['recommendations']}")
        
        # Alert on declining vitality
        if health['vitality'] == "declining":
            print(f"⚠️ WARNING: Community vitality is declining")
        
        # Sleep for monitoring interval
        await asyncio.sleep(3600)  # Check every hour

Member Performance Review

async def review_member_performance(community_id: str, member_id: str):
    """Generate a performance review for a member."""
    
    analytics = CommunityAnalytics(community_manager)
    
    # Get member analytics for different time ranges
    month_stats = analytics.get_member_analytics(community_id, member_id, time_range_days=30)
    quarter_stats = analytics.get_member_analytics(community_id, member_id, time_range_days=90)
    
    review = {
        "member_id": member_id,
        "review_date": datetime.utcnow().isoformat(),
        "monthly_performance": month_stats,
        "quarterly_performance": quarter_stats,
        "growth": {
            "contribution": quarter_stats['contribution_score'] - month_stats['contribution_score'],
            "participation": quarter_stats['participation_rate'] - month_stats['participation_rate']
        },
        "strengths": _identify_strengths(month_stats),
        "areas_for_improvement": _identify_improvements(month_stats)
    }
    
    return review

Import and Usage

The CommunityAnalytics class is now available in the public API:

# Import from domain.community
from aiecs.domain.community import CommunityAnalytics

# Or import from domain
from aiecs.domain import CommunityAnalytics

# Initialize
analytics = CommunityAnalytics(community_manager)

# Use analytics methods
health = analytics.get_community_health_metrics("community_001")

Notes

  • Analytics data is cached for performance optimization

  • Cache expiration can be configured per analytics instance

  • Historical data storage depends on the underlying persistence layer

  • Real-time metrics are computed on-demand

  • Trend analysis requires sufficient historical data