第07讲 apply 把函数用到数据上

📎 配套代码：第07讲_apply应用.py
📊 配套数据：本讲用课件内的小样本（2024-01-02 真实收盘快照），无需外部数据

🎬 开场：内置函数不够用的时候

前两讲的操作，用的都是 pandas 给好的函数：mean()、sort_values()、rank()。它们覆盖了大部分日常需求。

但研究里总有些逻辑是 pandas 想不到的：

• 股票代码要按交易所规则加后缀：600519 → 600519.SH，000858 → 000858.SZ（6 开头是上交所，0 和 3 开头是深交所）
• 按市值分档，但档位边界是你自己定的一套业务规则
• 每只股票要跑一次线性回归，取残差
• 每个行业内部做一次特殊的加权

这些逻辑没有现成的函数。apply 就是把你自己写的函数用到数据上的通道。

它是 pandas 里最灵活的操作，也是最容易被滥用的——因为它太方便了，方便到你会拿它去做本来有更好写法的事情。这一讲讲怎么用它，以及什么时候不该用它。

一、🧰 四个层次：作用在什么东西上

apply 家族的关键区别在于：你的函数每次收到的是什么。

搞清楚”函数收到什么”，用法就清楚了。

Series：逐元素

close.map(lambda x: x * 1.1)         # 每个元素乘 1.1
close.apply(lambda x: x * 1.1)       # 一样的效果

在 Series 上 map 和 apply 几乎等价。习惯上：

• map 用于替换/查表（它还能接受字典）
• apply 用于计算

df["industry"].map({"白酒": "消费", "银行": "金融"})     # 字典查表，map 独有

map 传字典时，没匹配上的会变成 NaN。想保留原值用 replace：

df["industry"].replace({"白酒": "消费"})    # 没匹配的保持原样

DataFrame：整列或整行

df.apply(lambda col: col.max() - col.min())            # 默认 axis=0，函数收到一整列
df.apply(lambda row: row["close"] * row["vol"], axis=1) # axis=1，函数收到一整行

axis 的含义很容易记反。记法：axis=1 时函数收到的是一行，你在函数里用 row["列名"] 取值。

DataFrame 每个格子

df"close", "vol".map(lambda v: round(v, 1))

WARNING: ⚠️ applymap 已经废弃
老代码里的 df.applymap(f) 在新版 pandas 会给出 FutureWarning，改用 df.map(f)。
注意 df.map 和 s.map 名字一样但作用不同：前者作用于每个格子，后者作用于每个元素（Series 只有一维，两者其实是一回事）。

🎮 随堂快练

QUESTION: 你要给股票代码加交易所后缀（6 开头 → .SH，其余 → .SZ）。应该用哪个？
TIP: 👉 答案

df["code"].map(lambda c: c + (".SH" if c.startswith("6") else ".SZ"))

作用在一列的每个元素上，用 Series.map（或 .apply，等价）。
顺带一提，这个需求其实有更好的写法——见第三节。

二、🧰 apply 能返回什么

返回一个值：结果是 Series

df.apply(lambda r: r["close"] * r["vol"], axis=1)

600519    53760000.0
000858    13402400.0
601318    11300000.0

每行算出一个数，拼成一个 Series。

返回 Series：结果展开成多列

这是 apply 一个很有用的能力——一次算出多个字段：

df.apply(lambda r: pd.Series({
    "mktcap": r["close"] * r["vol"],
    "half":   r["close"] / 2,
}), axis=1)

            mktcap    half
600519  53760000.0  840.00
000858  13402400.0   76.15
601318  11300000.0   22.60

当你的计算逻辑复杂、且一次要产出好几个指标时，这比分开写几个 apply 更清楚，也只遍历一次。

传额外参数

def scale(x, factor=1.0):
    return x * factor

close.apply(scale, factor=1.1)             # 关键字参数直接传
df.apply(lambda r: f(r, k=2), axis=1)      # 或者用 lambda 包一层

🎮 随堂快练

QUESTION: 每只股票要同时算出市值和”价格是否高于 100″两个字段。写出一次 apply 完成的代码。
TIP: 👉 答案

df.apply(lambda r: pd.Series({
    "mktcap": r["close"] * r["vol"],
    "is_high": r["close"] > 100,
}), axis=1)

返回 pd.Series 就会自动展开成多列。

三、🤔 什么时候不该用 apply

apply 最大的问题是：它太方便了，方便到你会用它去做本来有更好写法的事。

有内置写法时，别用 apply

df.apply(lambda r: r["close"] * r["vol"], axis=1)     # ❌
df["close"] * df["vol"]                                # ✅

两句结果一样。但第二句更短、更好读，而且快得多——实测 10 万行数据上差 1585 倍：

apply(axis=1)     189.2 ms
向量化 a+b          0.1 ms

原因是 apply(axis=1) 要为每一行造一个 Series 对象，然后退回 Python 逐行执行；而 df["a"] + df["b"] 是整列一次算完（第 01 讲的向量化）。

同样的对比在 Series 上也成立：

close.apply(lambda x: x*2)     8.6 ms
close * 2                       0.1 ms      → 快 155 倍

常见的”其实有内置写法”清单

回到第一节那个加后缀的例子——它其实也有内置写法：

df["code"].map(lambda c: c + (".SH" if c.startswith("6") else ".SZ"))    # 能用
df["code"] + np.where(df["code"].str.startswith("6"), ".SH", ".SZ")      # 更快

那什么时候该用 apply

逻辑确实没法用内置函数表达的时候：

# 每只股票跑一次回归，取 beta
def calc_beta(g):
    cov = g["ret"].cov(g["mkt_ret"])
    return cov / g["mkt_ret"].var()

df.groupby("code").apply(calc_beta)

# 调用一个只吃标量的第三方函数
df["score"].apply(some_external_scoring_function)

# 一次产出多个字段，逻辑较长
def analyze(row):
    ...十几行业务逻辑...
    return pd.Series({...})

df.apply(analyze, axis=1)

判断标准很简单：写完之后回头问一句”这个能不能用整列运算表达”。能，就换掉。

TIP: 💡 数据量小的时候无所谓
几百行、几千行的表，apply 慢那么零点几秒你根本感觉不到，怎么顺手怎么来。
但代码是会被复用的——今天在 1000 行上写的 apply(axis=1)，明天可能被用在 500 万行上。养成”先想内置写法”的习惯，成本很低。

🎮 随堂快练

QUESTION: 下面这句能不能改成内置写法？

df["level"] = df["mktcap"].apply(
    lambda x: "大" if x > 1e11 else ("中" if x > 1e10 else "小"))

TIP: 👉 答案
能，用 pd.cut：

df["level"] = pd.cut(df["mktcap"],
                     bins=[-np.inf, 1e10, 1e11, np.inf],
                     labels=["小", "中", "大"])

好处不只是快：pd.cut 把分档规则集中写在一处，边界一目了然；而嵌套的三元表达式多加一档就得重新捋一遍嵌套关系。
另外 pd.cut 返回的是有序分类类型，后续排序、分组时会按你给的档位顺序，而不是按字符串的字母序。

四、🧰 groupby + apply：最常用的组合

apply 单独用常常有更好的替代，但和 groupby 搭配时它不可替代——因为你的函数收到的是一整个分组，可以做任意复杂的组内计算。

df.groupby("industry").apply(lambda g: g["close"].max() - g["close"].min())

industry
保险       0.0
白酒    1527.7
银行      23.8

函数收到的 g 是一个完整的子 DataFrame，你可以对它做任何事：排序、取前几名、跑回归、多列联合计算。

三个 groupby 方法怎么选

df.groupby("industry")["close"].agg("mean")            # 每行业一个数
df.groupby("industry")["close"].transform("mean")      # 每只股票配上其行业均价
df.groupby("industry").apply(lambda g: ...)            # 任意逻辑

能用 agg / transform 的，就别用 apply——它们更快，意图也更明确。apply 留给”确实要看整个子表”的情况。

WARNING: ⚠️ 新版 pandas 的一个警告
df.groupby("industry").apply(f) 在新版会提示 DeprecationWarning，说分组列的处理方式将改变。加参数明确一下即可：

df.groupby("industry").apply(f, include_groups=False)

或者干脆只对需要的列分组：df.groupby("industry")"close","vol".apply(f)。

🎮 随堂快练

QUESTION: 每个行业取市值最大的 2 只股票。用 agg 还是 apply？
TIP: 👉 答案
apply——因为要返回的是多行（每组 2 行），agg 只能每组返回一个值。

df.groupby("industry", group_keys=False).apply(
    lambda g: g.nlargest(2, "mktcap"))

group_keys=False 是为了不把行业名再加一层到索引上。

🏋️ 训练营

QUESTION: 🟢 训练 1：给每只股票的代码加交易所后缀（6 开头 .SH，其余 .SZ）。写出 apply 写法和内置写法各一种。
TIP: 👉 参考

df["full"] = df.index.map(lambda c: c + (".SH" if c.startswith("6") else ".SZ"))
df["full"] = df.index + np.where(df.index.str.startswith("6"), ".SH", ".SZ")

两种都对。数据量小时随意，大表上第二种明显更快。

QUESTION: 🟡 训练 2：下面三句都能跑，指出各自更好的写法。

a = df.apply(lambda r: r["close"] * r["vol"], axis=1)
b = df["close"].apply(lambda x: 0 if x < 0 else x)
c = df["industry"].apply(lambda x: x in ["白酒", "银行"])

TIP: 👉 参考

a = df["close"] * df["vol"]              # 列间算术，直接向量化
b = df["close"].clip(lower=0)            # 截断
c = df["industry"].isin(["白酒", "银行"]) # 判断属于

三句的共同点：都在做 pandas 已经有专门函数的事。换掉之后不只是快，意图也更清楚——clip(lower=0) 一眼看出是”截断到 0 以上”，而 lambda 要读的人在脑子里跑一遍。

QUESTION: 🔴 训练 3：你要给每只股票算”过去 20 日收益率的标准差”，数据是长表（每行一个股票-日期）。下面这段代码在 500 万行上跑了半小时还没完。指出问题并改写。

def calc_vol(code):
    sub = df[df["code"] == code]
    return sub["ret"].tail(20).std()

result = {c: calc_vol(c) for c in df["code"].unique()}

TIP: 👉 参考
问题：每次调用 calc_vol 都要对整张 500 万行的表做一次 df["code"] == code 的全表扫描。有 5000 只股票就扫 5000 遍，总计算量是 250 亿次比较。
改法：用 groupby 一次分好组，让 pandas 只扫一遍：

result = df.groupby("code")["ret"].apply(lambda s: s.tail(20).std())

如果要的是每一天的滚动波动率（而不是只要最后一个值），连 apply 都不用：

df["vol20"] = df.groupby("code")["ret"].transform(lambda s: s.rolling(20).std())

这个坑的本质不是”apply 慢”，而是在循环里反复筛选整张表。groupby 的价值就是把”分组”这件事只做一次。

🐛 常见坑

• ⚠️ axis 记反：axis=1 时函数收到的是一行。记法：函数里写 row["列名"] 的那种就是 axis=1。
• ⚠️ 有内置写法却用 apply：算术、条件、字符串、查表、分档都有专门函数，更快也更好读。
• ⚠️ apply(axis=1) 在大表上极慢：实测 10 万行差 1585 倍。写之前先想能不能整列算。
• ⚠️ map 传字典时没匹配上的变 NaN：想保留原值用 replace。
• ⚠️ applymap 已废弃：改用 df.map。
• ⚠️ 在循环里反复筛选整表：for c in codes: df[df.code==c] 是全表扫描 × N。用 groupby。
• ⚠️ 能用 transform 却用了 apply：结果要贴回原表时 transform 更合适，也更快。

✍️ 作业

1. 造一张 10 万行的表，分别用 apply(axis=1) 和向量化算两列之积，用 %timeit 记录耗时，算出倍数。
2. 把训练 2 的三句都改成内置写法，并验证结果完全一致（用 .equals()）。
3. 用 df.apply(f, axis=1) 写一个返回 pd.Series 的函数，一次产出三个字段，确认结果自动展开成了三列。
4. 用 groupby().apply() 实现”每个行业取市值最大的 2 只”，再想想能不能用别的方法实现。
5. 思考题：apply(axis=1) 为什么比整列运算慢那么多？（提示：想想它每一行都要做什么——第 01 讲讲过 pandas 一列在内存里是连续的，而一行不是。）

🔮 下讲预告：第 08 讲——数据加载与存储。到目前为止我们的数据都是手工造的。下一讲讲怎么把真实数据读进来：CSV 的各种坑（分隔符、编码、日期解析、类型推断）、Excel、以及为什么存中间结果推荐用 parquet 而不是 CSV。