如何用Python计算日均/月均/年均社交媒体发帖量？

问题背景 #

需求是基于包含unix时间戳的社交媒体帖子数据，计算日均、月均、年均发帖量。原代码存在以下核心问题：

• 未统计无发帖的日期/月份/年份，导致平均值计算偏差（比如某段时间内有几天没发帖，这些天的0值未纳入计算）
• 仅支持按天分组，无法扩展到月/年维度
• 分组逻辑效率低下（嵌套循环处理数千条数据会显著变慢）
• 存在拼写错误与变量名错误（如avarege_number、grop_by_day，以及post_per函数内的变量名误用）

改进方案 #

使用collections.defaultdict高效统计各时间维度的发帖量，再生成完整的时间序列补全无发帖的时间点，最后计算平均值。以下是完整实现：

import datetime
from collections import defaultdict

def unix_to_date(unix_time: int) -> datetime.date:
    """将unix时间戳转换为UTC日期对象"""
    return datetime.datetime.utcfromtimestamp(unix_time).date()

def unix_to_month(unix_time: int) -> tuple:
    """将unix时间戳转换为(year, month)元组"""
    dt = datetime.datetime.utcfromtimestamp(unix_time)
    return (dt.year, dt.month)

def unix_to_year(unix_time: int) -> int:
    """将unix时间戳转换为年份"""
    return datetime.datetime.utcfromtimestamp(unix_time).year

def calculate_daily_average(posts: list) -> float:
    """计算日均发帖量（包含无发帖的日期）"""
    # 统计每天的发帖数
    daily_counts = defaultdict(int)
    min_date = None
    max_date = None
    
    for post in posts:
        date = unix_to_date(post['created_utc'])
        daily_counts[date] += 1
        
        # 更新日期范围
        if min_date is None or date < min_date:
            min_date = date
        if max_date is None or date > max_date:
            max_date = date
    
    # 生成日期范围内的所有日期，补全0值
    all_dates = []
    current_date = min_date
    while current_date <= max_date:
        all_dates.append(daily_counts.get(current_date, 0))
        current_date += datetime.timedelta(days=1)
    
    # 计算平均值
    return sum(all_dates) / len(all_dates) if all_dates else 0.0

def calculate_monthly_average(posts: list) -> float:
    """计算月均发帖量（包含无发帖的月份）"""
    # 统计每月的发帖数
    monthly_counts = defaultdict(int)
    min_year_month = None
    max_year_month = None
    
    for post in posts:
        year_month = unix_to_month(post['created_utc'])
        monthly_counts[year_month] += 1
        
        # 更新月份范围
        if min_year_month is None or year_month < min_year_month:
            min_year_month = year_month
        if max_year_month is None or year_month > max_year_month:
            max_year_month = year_month
    
    # 生成月份范围内的所有月份，补全0值
    all_months = []
    current_year, current_month = min_year_month
    while (current_year, current_month) <= max_year_month:
        all_months.append(monthly_counts.get((current_year, current_month), 0))
        # 切换到下一个月
        if current_month == 12:
            current_year += 1
            current_month = 1
        else:
            current_month += 1
    
    # 计算平均值
    return sum(all_months) / len(all_months) if all_months else 0.0

def calculate_yearly_average(posts: list) -> float:
    """计算年均发帖量（包含无发帖的年份）"""
    # 统计每年的发帖数
    yearly_counts = defaultdict(int)
    min_year = None
    max_year = None
    
    for post in posts:
        year = unix_to_year(post['created_utc'])
        yearly_counts[year] += 1
        
        # 更新年份范围
        if min_year is None or year < min_year:
            min_year = year
        if max_year is None or year > max_year:
            max_year = year
    
    # 生成年份范围内的所有年份，补全0值
    all_years = [yearly_counts.get(year, 0) for year in range(min_year, max_year + 1)]
    
    # 计算平均值
    return sum(all_years) / len(all_years) if all_years else 0.0

# 示例调用
if __name__ == "__main__":
    data = [
        {'created_utc': 1669804989},
        {'created_utc': 1669804782},
        {'created_utc': 1669804772},
        {'created_utc': 1672560000}  # 另一个日期的帖子，用于测试补全无发帖日期
    ]
    
    print(f"日均发帖量: {calculate_daily_average(data):.2f}")
    print(f"月均发帖量: {calculate_monthly_average(data):.2f}")
    print(f"年均发帖量: {calculate_yearly_average(data):.2f}")

代码说明 #

1. 时间转换函数：分别将unix时间戳转换为日期、年月元组、年份，适配不同统计维度
2. 统计逻辑：用defaultdict高效统计各时间单位的发帖数，同时记录时间范围的起止点
3. 补全无发帖时间点：生成起止点之间的所有时间单位，把未统计到的时间点补0，确保平均值计算准确
4. 平均值计算：基于补全后的完整时间序列计算平均值，结果更符合实际情况

内容的提问来源于stack exchange，提问作者user16282401