如何用Pandas插入多行复杂数据并分析歌曲流媒体播放模式？

一、数据结构合理性分析 #

你提出的**宽表结构（每首歌一行，DayN为列）**虽能直观对比同发行天数的播放量，但存在明显局限：

• 若歌曲生命周期差异大（比如部分歌曲火5年、部分火10年），列数会极度膨胀，导致数据稀疏、存储低效；
• 用0填充早期缺失的1-2年数据会误导模型，让模型误判为真实播放量为0，而非数据缺失。

更推荐长表结构，核心字段包括：

• iscr_code：歌曲唯一标识
• days_since_release：发行后天数（整数，如0、1、2...）
• normalized_streams：归一化后的播放量（总和为1）
• raw_streams：原始播放量（可选）

这种结构更灵活，适配Pandas的分组、聚合操作，也便于后续可视化（如按天数分组绘制均值曲线）和时间序列建模。

二、不破坏数据同步的异常值清洗 #

核心原则是按歌曲分组处理时间序列，避免跨歌曲的全局操作打乱日期与播放量的对应关系：

1. 分组隔离：用df.groupby('iscr_code')对每首歌的时间序列单独处理
2. 异常值识别：
   • 用四分位距（IQR）：计算每首歌播放量的Q1、Q3，超出[Q1-1.5*IQR, Q3+1.5*IQR]范围的标记为异常；
   • 结合时间上下文：流媒体播放量突增可能是推广活动导致的真实数据，不要直接删除，先标记再根据业务规则决策。
3. 异常值处理：
   • 替换：用前后N天的均值、线性插值或时间序列预测值填充异常值；
   • 删除：仅当异常值明确是数据错误（如负数、远超合理范围的数值）时，删除对应行，但需保留天数连续性标记。
4. 缺失值处理：早期缺失的1-2年数据建议保留NaN，不要用0填充，后续建模可选择跳过缺失段，或用ARIMA、Prophet等时间序列方法补全。

示例代码片段：

def clean_outliers(group):
    q1 = group['streams_total'].quantile(0.25)
    q3 = group['streams_total'].quantile(0.75)
    iqr = q3 - q1
    lower_bound = q1 - 1.5 * iqr
    upper_bound = q3 + 1.5 * iqr
    # 标记异常值
    group['is_outlier'] = (group['streams_total'] < lower_bound) | (group['streams_total'] > upper_bound)
    # 用线性插值替换异常值
    group['streams_total'] = group['streams_total'].mask(group['is_outlier'], group['streams_total'].interpolate(method='linear'))
    return group

# 按歌曲分组清洗
cleaned_df = df.groupby('iscr_code').apply(clean_outliers).reset_index(drop=True)

三、几何曲线拟合可行性 #

完全可以进行几何曲线拟合，流媒体播放量的时间分布通常符合指数衰减或幂律分布：

1. 模型选择：
   • 指数模型：y = a * e^(-b*x)（适配初期播放量快速下降、后期趋于平稳的情况）；
   • 幂律模型：y = a * x^(-b)（适配播放量随时间持续缓慢下降的长尾效应）。
2. 拟合步骤：
   • 用归一化后的播放量normalized_streams作为因变量，days_since_release作为自变量；
   • 使用scipy.optimize.curve_fit进行拟合，输出模型参数a、b；
   • 用R²评估拟合优度，选择最优模型。
3. 注意事项：
   • 仅使用有完整数据的时间段（跳过早期缺失的1-2年），避免缺失值干扰拟合结果；
   • 若多首歌曲的拟合参数差异大，可按流派、发行平台分组拟合。

示例代码片段：

from scipy.optimize import curve_fit
import numpy as np

# 定义指数衰减模型
def exponential_model(x, a, b):
    return a * np.exp(-b * x)

# 按歌曲分组拟合
def fit_curve(group):
    x = group['days_since_release'].values
    y = group['normalized_streams'].values
    # 跳过缺失值
    mask = ~np.isnan(y)
    x_clean = x[mask]
    y_clean = y[mask]
    if len(x_clean) < 2:
        return pd.Series([np.nan, np.nan], index=['a', 'b'])
    params, _ = curve_fit(exponential_model, x_clean, y_clean, p0=[1, 0.001])
    return pd.Series(params, index=['a', 'b'])

# 获取每首歌的拟合参数
fit_params = cleaned_df.groupby('iscr_code').apply(fit_curve).reset_index()

四、Pandas插入多行复杂数据 #

根据数据来源选择以下方法：

1. 从JSON批量读取并合并：
   若每首歌的JSON数据是单独文件，循环读取后用pd.concat合并：

   import pandas as pd
   import json
   import os
   
   song_dir = "path/to/song/json/files"
   all_song_data = []
   
   for filename in os.listdir(song_dir):
       if filename.endswith('.json'):
           with open(os.path.join(song_dir, filename), 'r') as f:
               raw_data = json.load(f)
               song_df = pd.DataFrame(raw_data)
               # 从文件名或JSON中提取ISCR
               song_df['iscr_code'] = filename.split('_')[0]
               # 计算发行后天数
               song_df['release_date'] = pd.to_datetime(song_df['release_date'])
               song_df['days_since_release'] = (pd.to_datetime(song_df['date']) - song_df['release_date']).dt.days
               # 归一化播放量
               song_df['normalized_streams'] = song_df['streams_total'] / song_df['streams_total'].sum()
               all_song_data.append(song_df)
   
   # 合并所有数据
   final_df = pd.concat(all_song_data, ignore_index=True)
   

2. 插入字典列表：
   若数据是内存中的字典列表，直接构造新DataFrame后合并：

   new_rows = [
       {'iscr_code': 'USRCAB123B', 'days_since_release': 0, 'normalized_streams': 0.1},
       {'iscr_code': 'USHDB1232H', 'days_since_release': 2, 'normalized_streams': 0.08}
   ]
   new_df = pd.DataFrame(new_rows)
   final_df = pd.concat([final_df, new_df], ignore_index=True)
   

3. 使用df.loc批量插入：
   适合少量行插入，注意设置索引：

   # 获取当前最大索引
   max_idx = final_df.index.max()
   final_df.loc[max_idx+1] = ['USRCAB123B', 1, 0.09]
   final_df.loc[max_idx+2] = ['USHDB1232H', 3, 0.07]
   

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