基于R语言实现基金层级数据聚合及面板数据适配方案问询

我正在处理来自Morningstar的份额级别投资基金面板数据，数据结构如下：

需求为按基金ID将数据聚合至基金层级：

• 基金规模：对应份额净资产之和
• 收益率：按净资产加权平均计算
• 星级评级：按净资产加权平均后取整

示例计算：

• 基金A收益率：(0.01*100 + 0.012*200 + 0.005*150)/(100+200+150) = 0.92%
• 基金B星级评级：(2*50 + 3*120)/(50+120) = 2.70，取整为3

当前数据集包含8000+份额，需要可扩展性的实现方案，同时需了解如何适配包含3个月日度观测的面板数据。

1. 数据预处理 #

首先需要清洗格式不规范的字段：

• 收益率字段：移除百分号%，将逗号,替换为小数点.，转换为浮点型后除以100转为小数格式
• 星级评级字段：提取数字部分，转换为整数型

2. 核心聚合逻辑（基础版，无日期维度） #

使用Python的pandas库实现，该库处理万级以上数据效率足够，且具备良好可扩展性：

import pandas as pd

# 读取数据（示例数据）
data = pd.DataFrame([
    ["A", "A1", 100, "1%", "4 stars"],
    ["A", "A2", 200, "1,2 %", "4 stars"],
    ["A", "A3", 150, "0,5 %", "3 stars"],
    ["B", "B1", 50, "1,1 %", "2 stars"],
    ["B", "B2", 120, "0,75%", "3 stars"],
    ["C", "C1", 300, "0,4%", "5 stars"],
    ["C", "C2", 500, "0,55%", "4 stars"]
], columns=["Fund ID", "Sec ID", "Net Assets", "Return", "Rating"])

# 数据清洗
data["Return"] = data["Return"].str.replace("%", "").str.replace(",", ".").astype(float) / 100
data["Rating"] = data["Rating"].str.extract(r"(\d+)").astype(int)

# 聚合计算
aggregated = data.groupby("Fund ID").apply(
    lambda x: pd.Series({
        "Fund Size": x["Net Assets"].sum(),
        "Weighted Return": (x["Return"] * x["Net Assets"]).sum() / x["Net Assets"].sum(),
        "Weighted Rating": round((x["Rating"] * x["Net Assets"]).sum() / x["Net Assets"].sum())
    })
).reset_index()

# 格式化收益率为百分比
aggregated["Weighted Return"] = aggregated["Weighted Return"].apply(lambda x: f"{x*100:.2f}%")

print(aggregated)

3. 适配3个月日度面板数据 #

如果数据包含日期字段（如Date），只需在分组时增加日期维度，保持聚合逻辑不变即可：

# 假设数据新增Date字段，格式为YYYY-MM-DD
aggregated_daily = data.groupby(["Fund ID", "Date"]).apply(
    lambda x: pd.Series({
        "Fund Size": x["Net Assets"].sum(),
        "Weighted Return": (x["Return"] * x["Net Assets"]).sum() / x["Net Assets"].sum(),
        "Weighted Rating": round((x["Rating"] * x["Net Assets"]).sum() / x["Net Assets"].sum())
    })
).reset_index()

# 格式化收益率
aggregated_daily["Weighted Return"] = aggregated_daily["Weighted Return"].apply(lambda x: f"{x*100:.2f}%")

4. 性能优化建议（针对超大规模数据） #

• 如果数据量超过100万行，可使用pandas的agg方法替代apply，提升效率：

agg_func = {
    "Net Assets": "sum",
    "Return": lambda x: (x * data.loc[x.index, "Net Assets"]).sum() / data.loc[x.index, "Net Assets"].sum(),
    "Rating": lambda x: round((x * data.loc[x.index, "Net Assets"]).sum() / data.loc[x.index, "Net Assets"].sum())
}

aggregated = data.groupby("Fund ID").agg(agg_func).rename(
    columns={"Net Assets": "Fund Size", "Return": "Weighted Return", "Rating": "Weighted Rating"}
).reset_index()

• 若使用SQL处理（如数据存储在数据库中），可直接编写分组聚合SQL，性能更优：

SELECT
    "Fund ID",
    SUM("Net Assets") AS "Fund Size",
    ROUND(SUM("Return" * "Net Assets") / SUM("Net Assets") * 100, 2) || '%' AS "Weighted Return",
    ROUND(SUM("Rating" * "Net Assets") / SUM("Net Assets")) AS "Weighted Rating"
FROM (
    -- 子查询完成数据清洗
    SELECT
        "Fund ID",
        "Net Assets",
        CAST(REPLACE(REPLACE("Return", '%', ''), ',', '.') AS FLOAT) / 100 AS "Return",
        CAST(SUBSTRING("Rating", 1, 1) AS INT) AS "Rating"
    FROM fund_data
) cleaned_data
GROUP BY "Fund ID";

-- 日度面板适配：增加Date分组
SELECT
    "Fund ID",
    "Date",
    SUM("Net Assets") AS "Fund Size",
    ROUND(SUM("Return" * "Net Assets") / SUM("Net Assets") * 100, 2) || '%' AS "Weighted Return",
    ROUND(SUM("Rating" * "Net Assets") / SUM("Net Assets")) AS "Weighted Rating"
FROM (
    SELECT
        "Fund ID",
        "Date",
        "Net Assets",
        CAST(REPLACE(REPLACE("Return", '%', ''), ',', '.') AS FLOAT) / 100 AS "Return",
        CAST(SUBSTRING("Rating", 1, 1) AS INT) AS "Rating"
    FROM fund_daily_data
) cleaned_data
GROUP BY "Fund ID", "Date";

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