python pandas数据分析实战:某一小时30万RMB虚拟

最近闲着没事干,玩了一下pandas,拿最近接触过比较大的数据练手了.（就这点东西，因为比较生疏，还花了我两天时间）
这位虚拟主播在国内出道是在五月份,而且出道前就炒作了一波,所以首播的流量相当大了.
经过统计,首播中总共有24590 人发言.
下面是统计数据

总弹幕数：96386
总用户数: 24590
sc数目:338,sc总收益:30360 RMB
gift数目:42598,gift总收益:72175.9 RMB
舰长数目:968,舰长总收益:209664.0 RMB
总付费用户数(舰长+礼物(>1rmb)+sc):6335
不发言的付费用户：0
本次直播总营收:312199.9

直播收入成分占比饼图.png 粉丝牌成分图.png word_cloud.png

其他还有部分分析,因为涉及到有用户信息我就不放上来了.

经过一个半月的时间，热度降了不少，但是还是相当离谱的。现在这时代资本想捧一个人太简单了。

下面是6月29日直播的数据
离开了初期疯狂上舰的时候，礼物收益开始占大头了
4455位用户，相比初次缩水到1/5，说明看热闹的人基本走的差不多了。

总弹幕数：12403
总用户数: 4455
sc数目:23,sc总收益:1308 RMB
gift数目:9456,gift总收益:4016.3 RMB
舰长数目:9,舰长总收益:1782.0 RMB
总付费用户数(舰长+礼物(>1rmb)+sc):248
不发言的付费用户：0
本次直播总营收:7106.3

直播收入成分占比饼图.png

通过粉丝成分我们可以清楚的看到，粉丝提纯已经完成，大部分粉丝牌都转换为主播的粉丝牌了。
其他杂七杂八的，除了王牛奶没有一个超过1%(也印证了我上面说的，看热闹的要么跑了，要么被提纯)
粉丝团在第一次比例也比较高，我估计是d还没有设置粉丝牌的v，他默认的名字是这个。

粉丝牌成分图.png

有些特殊符号还没有完全去除，词典还需要一定的优化

word_cloud.png

解析弹幕xml的函数定义

%%time
danmukuList = get_danmukuList(tree)
scList = get_scList(tree)
giftList = get_giftList(tree)
guardList = get_guardList(tree)

计算直播收入的函数定义

# 计算礼物的收入
# b站的礼物是以金瓜子为单位的，1元等于1000金瓜子
# 这里只计算了大于等于1000金瓜子的礼物价值

```python
guard_income = get_guard_income(guardList)
gift_income = get_gift_income(giftList)
sc_income = get_sc_income(scList)

统计用户的函数定义

def get_uidset(list):
    uidSet = set()
    for i in list:
        uidSet.add(i['uid'])
    return uidSet


def get_tiangou_set(scList, giftList, guardList):
    uidSet = set()
    for gift in giftList:
        if gift['price'] > 999:
            uidSet.add(gift['uid'])
    for item in scList+guardList:
        uidSet.add(item['uid'])
    return uidSet


tiangou_set = get_tiangou_set(scList, giftList, guardList)

# 付费用户数目


def get_tiangou_count(scList, giftList, guardList):
    uidSet = get_tiangou_set(scList, giftList, guardList)
    return len(uidSet)


tiangou_count = get_tiangou_count(scList, giftList, guardList)

uidSet = get_uidset(danmukuList+scList+giftList+guardList)
usercount = len(uidSet)

# 粉丝牌数目
# 付费用户可以获得粉丝牌，所以计算本场的付费用户加上未付费的粉丝牌数目即可
def get_medal_count(tiangou_set, danmukuList):
    new_set = tiangou_set.copy()
    for user in danmukuList:
        if user['medal_name'] == medal_name:
            #             print(user)
            new_set.add(user['uid'])
    return len(new_set)


# medal_count = get_medal_count(tiangou_set, danmukuList)

弹幕数,总发言用户数,总营收

total_income= sc_income+gift_income+guard_income
print(f'总弹幕数：{len(danmukuList)}')
print(f'总用户数: {usercount}')
# print(f'有粉丝牌用户数:{medal_count},粉丝牌用户占比:{medal_count/usercount:{5}.{2}}')
print(f'sc数目:{len(scList)},sc总收益:{sc_income} RMB')
print(f'gift数目:{len(giftList)},gift总收益:{gift_income} RMB')
print(f'舰长数目:{len(guardList)},舰长总收益:{guard_income} RMB')
print(f'总付费用户数(舰长+礼物(>1rmb)+sc):{tiangou_count}')
print(f'不发言的付费用户：{len(tiangou_set.difference(uidSet))}')
print(f'本次直播总营收:{total_income}')

直播收入成分占比饼图

plt.rcParams['font.sans-serif'] = ['SimHei']  # 用来正常显示中文标签
# 调大图片大小
plt.figure(figsize=(6, 6))
# 调整字体大小
plt.rcParams.update({"font.size": 16})
plt.title(f'一场直播收入成分占比饼图(总收入:{total_income}元)')
plt.pie([sc_income, gift_income, guard_income], labels=[
        'SC收入', '礼物收入', '舰长收入'], autopct='%1.1f%%', shadow=False, startangle=150)
# 保存饼图
plt.savefig(os.path.join(output_path, '直播收入成分占比饼图'))
plt.show()

弹幕列表

danmuku_df = pd.DataFrame(danmukuList)
danmuku_df

# danmuku_df.groupby(['uid']).agg({'uid': 'count'})

用户发言次数分析

# 添加count新行
danmuku_df['count'] = danmuku_df.groupby(['uid'])['uid'].transform('count')
# danmuku_df

# 对uid去重并排序，取前排
danmuku_count_rank = danmuku_df.drop_duplicates('uid').sort_values(
    'count', ascending=False)
danmuku_count_rank.head(20)

# 发言次数的占比情况，可以看到只发言1次到2次的占60%左右
danmuku_count_rank['count'].describe(
    percentiles=[.25, .50, .60, .70, .80, .90])
# danmuku_count_rank['count'].describe()

粉丝牌、粉丝成分统计

统计单列数据（百分比）的函数定义

# 统计单列数据
def describe_list(list):
    resDict = {}
    for item in list:
        if item == '':
            item = 'Nan'
        if resDict.__contains__(item):
            resDict[item] += 1
        else:
            resDict[item] = 1
    totalCount = len(list)
    describe = pd.DataFrame(pd.Series(resDict), columns=['count'])
    describe['ratio'] = describe['count']/totalCount
    return describe.sort_values('ratio', ascending=False)

medal_describe = describe_list(danmuku_count_rank['medal_name'])
medal_describe

合并num行后统计数据添加到DataDrame尾部一行的函数

# 截取最多前几名,并且把其余的行合并为其他
def medal_describe_sumAfterNum(medal_describe, num):
    sumAfterNum = medal_describe.iloc[num:].sum()
    sumAfterNum.name = '其他'
    return medal_describe.head(num).append(sumAfterNum)

粉丝牌占比饼图

# 粉丝牌成分占比
medal_pie = medal_describe_sumAfterNum(medal_describe, 10)['ratio'].plot.pie(
    figsize=(12, 12), autopct="%.2f", normalize=False, fontsize=12, title='前十名粉丝牌占比(其他为10名后的)')
medal_pie_fig = medal_pie.get_figure()
medal_pie_fig.savefig(os.path.join(output_path, '粉丝牌成分图.png'))

弹幕文本语种分析(找日本人)

danmukuList_df = danmuku_df.loc[:, ['uid', 'user', 'content']]
danmukuList_df

导入自制的语种检测python库

import src.langdetect as langdetect
ld = langdetect.LangDetector()
# def detect_lang(str):
#     lang_list = ['ja', 'zh-cn', 'en']
#     try:
#         lang = langdetect.detect(str)
#     except Exception as e:
#         lang = 'zh-cn'
#     if lang not in lang_list:
#         lang = 'zh-cn'
#     return lang

导入b站字符表情，颜文字表情列表

import src.bilibili_emojis as bilibili_emojis
bilibili_emoji_list = bilibili_emojis.bilibili_emojis

检测语言函数定义（b站颜文字视为中文）

def detect_lang(str):
    lang = ''
    lang = ld.detect(str, specific=True)
    if str in bilibili_emoji_list:
        lang = 'zh'
    return lang

df添加语言种类列

danmukuList_df['lang'] = danmukuList_df['content'].apply(
    lambda x: detect_lang(x))

danmukuList_df.iloc[:10]

选取日语弹幕进行分析

ja_danmukuList_df = danmukuList_df.loc[danmukuList_df['lang'] == 'ja'].copy()

ja_danmukuList_df['count'] = ja_danmukuList_df.groupby(['uid'])[
    'uid'].transform('count')

# 说日语最多的用户名单
freq_ja_df = ja_danmukuList_df.drop_duplicates('uid').sort_values(
    'count', ascending=False)
freq_ja_df.head(100).to_csv(os.path.join(output_path, '用户日语使用次数排名.csv'))
freq_ja_df.head(10)

词频分析、词云

选取中文弹幕进行分析

cndanmuku_df = danmukuList_df.loc[danmukuList_df['lang'] == 'zh'].copy()

cndanmuku_df

载入停用词和本地词典用jieba进行分词

# 载入停用词stopwords，指的是那些没有意义的词和标点符号，分词结果中最好去除掉
def load_stopwords(path):
    stopwords = []
    with open(path, 'r', encoding='utf8')as f:
        for item in f.readlines():
            if word := item.strip('\n'):
                stopwords.append(word)
    return stopwords


stopwords = load_stopwords('dict/stopword.dict')
jieba.load_userdict('dict/food.dict')

使用结巴分词进行分词，过滤停用词的函数定义

# 使用结巴分词对文本分词，并且过滤掉停用词
def lcut_words(text, stopwords):
    res = []
    text = text.strip()
    if text == '':
        return res
    else:
        for word in jieba.cut(text):
            # 过滤掉停用词
            if word not in stopwords:
                res.append(word)
    return res

得到分词后的表格，所有分词的列表

%%time
words_list = []

# 过滤掉bilibili表情


def collect_words(x, stopwords, bilibili_emoji_list):
    global words_list
    if x in bilibili_emoji_list:
        res = [x]
    else:
        res = lcut_words(x, stopwords)
    if len(res) == 1 and res[0] == '':
        print(1)
    words_list += res
    return ' '.join(res)


cndanmuku_df['cut'] = cndanmuku_df['content'].apply(
    lambda x: collect_words(x, stopwords, bilibili_emoji_list))
cndanmuku_df.head(10)

分析词语列表的词频

describe_list(words_list).head(20)

# 使用td-idf方法关键词抽取,这类技术需要语料库的支持,默认的语料库效果一般吧,所以默认不用
# %%time
# danmukuList_df['cut'] = danmukuList_df['content'].apply(
#     lambda x: ' '.join(jieba.analyse.extract_tags(x)))
# danmukuList_df.head(300)

生成词云图

wordcloud_text = ' '.join(words_list)
wc = wordcloud.WordCloud(font_path="msyh.ttc",
                         width=1200,
                         height=800,
                         background_color='white',
                         max_words=150, stopwords=stopwords)
wc.generate(wordcloud_text)
wc.to_file(os.path.join(output_path, 'word_cloud.png'))

plt.figure(figsize=(10, 10))
plt.imshow(wc, interpolation='bilinear')

使用 Pandas Profiling Report 快速生成分析报告

# 使用 Pandas Profiling Report 快速生成分析报告
profile = ProfileReport(
    danmuku_df, title="Pandas Profiling Report", explorative=True)

profile.to_file(os.path.join(output_path, 'profiling_report.html'))