一、动量特征：趋势强度的量化表达

动量是最直觉的特征——涨了 20 天的股票，明天继续涨的概率是否更高？我们用滚动收益率来量化它：

import pandas as pd
import numpy as np
from mairui import Client
​
client = Client("LICENCE-66D8-9F96-0C7F0FBCD073")
rows = client.stock_history("000001", period="d", dividend="n", lt=120)
df = pd.DataFrame(rows)
for col in ["o", "h", "l", "c", "v"]:
    df[col] = pd.to_numeric(df[col], errors="coerce")
​
df = df.sort_values("t").reset_index(drop=True)
​
# 动量特征：N 日滚动收益率（%）
df["ret_5"] = df["c"].pct_change(5) * 100
df["ret_10"] = df["c"].pct_change(10) * 100
df["ret_20"] = df["c"].pct_change(20) * 100
​
print(df[["t", "c", "ret_5", "ret_10", "ret_20"]].tail(3))


           t      c   ret_5   ret_10   ret_20
47  2025-11-27  11.71    0.78     2.10     2.27
48  2025-11-28  11.61   -0.77     1.31     1.92
49  2025-12-01  11.69    0.78     2.27     2.27

 计算的是"今天收盘价 vs 5 个交易日前的收盘价"，不是"过去 5 天的累计收益"。这意味着 

 在 T 日只能用到 T-5 日的价格，

如果你需要的是"过去 5 天每日收益的连乘"（即复利收益），要这样写：


df["ret_20_compound"] = (df["c"] / df["c"].shift(20) - 1) * 100

两者的区别在于：pct_change 是单期对数收益的近似，复利收益是真实几何收益。对短期（5 日）来说差距很小，对长期（1 年）来说复利收益更准确。

二、波动特征：风险不是感觉，是数字

波动率是量化投资里最重要的风险指标之一。这里介绍两个最实用的：ATR 和滚动标准差。

ATR 衡量的是"真实波动范围"，比单纯的"最高价 - 最低价"更能反映极端跳空：


# 真实波幅 TR = max(h-l, |h-c_prev|, |l-c_prev|)
df["tr"] = np.maximum(
    df["h"] - df["l"],
    np.maximum(
        (df["h"] - df["c"].shift(1)).abs(),
        (df["l"] - df["c"].shift(1)).abs()
    )
)
​
# 14 日 ATR
df["atr_14"] = df["tr"].rolling(14).mean()
​
print(df[["t", "c", "tr", "atr_14"]].tail(3))


           t      c   tr  atr_14
47  2025-11-27  11.71  0.02    0.17
48  2025-11-28  11.61  0.10    0.17
49  2025-12-01  11.69  0.09    0.17


df["vol_20"] = df["c"].pct_change().rolling(20).std() * 100

这是教科书式的"20 日年化波动率"的日频版本。注意 

 算的是日收益，rolling(20).std() 是日收益的标准差，乘以 100 是为了变成百分比。

量价关系是技术分析的核心假设之一。OBV（On-Balance Volume）把成交量转化为方向性资金流：


# OBV：涨日记 +v，跌日记 -v，平日记 0
df["obv"] = 0
for i in range(1, len(df)):
    if df.loc[i, "c"] > df.loc[i-1, "c"]:
        df.loc[i, "obv"] = df.loc[i-1, "obv"] + df.loc[i, "v"]
    elif df.loc[i, "c"] < df.loc[i-1, "c"]:
        df.loc[i, "obv"] = df.loc[i-1, "obv"] - df.loc[i, "v"]
    else:
        df.loc[i, "obv"] = df.loc[i-1, "obv"]
​
# OBV 20 日斜率（简化版：差分 / 天数）
df["obv_slope_20"] = df["obv"].diff(20) / 20

顶背离是经典的反转信号：价格创了 20 日新高，但 OBV 没有同步新高——说明上涨缺乏资金支撑，可能是"虚涨"。


df["c_20_high"] = df["c"].rolling(20).max()
df["obv_20_high"] = df["obv"].rolling(20).max()
​
# 顶背离：价格新高，但 OBV 未新高（留 2% 容忍区间防噪声）
df["divergence_top"] = (
    (df["c"] == df["c_20_high"]) &
    (df["obv"] < df["obv_20_high"] * 0.98)
)
​
tops = df[df["divergence_top"]]
print(f"顶背离信号: {len(tops)} 次")

四、特征矩阵：从 DataFrame 到模型输入

把上面所有特征拼起来，输出一个标准格式的特征矩阵：


def build_features(df):
    """输入：清洗后的单股 K 线 DataFrame（含 o,h,l,c,v,t）
       输出：特征矩阵 DataFrame"""
    df = df.sort_values("t").reset_index(drop=True)
    for col in ["o", "h", "l", "c", "v"]:
        df[col] = pd.to_numeric(df[col], errors="coerce")
​
    # 动量
    df["ret_5"] = df["c"].pct_change(5) * 100
    df["ret_10"] = df["c"].pct_change(10) * 100
    df["ret_20"] = df["c"].pct_change(20) * 100
​
    # 波动
    df["tr"] = np.maximum(df["h"] - df["l"],
                          np.maximum((df["h"] - df["c"].shift(1)).abs(),
                                     (df["l"] - df["c"].shift(1)).abs()))
    df["atr_14"] = df["tr"].rolling(14).mean()
    df["vol_20"] = df["c"].pct_change().rolling(20).std() * 100
​
    # 资金流
    df["obv"] = 0
    for i in range(1, len(df)):
        if df.loc[i, "c"] > df.loc[i-1, "c"]:
            df.loc[i, "obv"] = df.loc[i-1, "obv"] + df.loc[i, "v"]
        elif df.loc[i, "c"] < df.loc[i-1, "c"]:
            df.loc[i, "obv"] = df.loc[i-1, "obv"] - df.loc[i, "v"]
        else:
            df.loc[i, "obv"] = df.loc[i-1, "obv"]
    df["obv_slope_20"] = df["obv"].diff(20) / 20
​
    c_20_high = df["c"].rolling(20).max()
    obv_20_high = df["obv"].rolling(20).max()
    df["divergence_top"] = (df["c"] == c_20_high) & (df["obv"] < obv_20_high * 0.98)
​
    feat_cols = ["t", "c", "ret_5", "ret_10", "ret_20",
                 "atr_14", "vol_20", "obv", "obv_slope_20", "divergence_top"]
    return df[feat_cols].copy()
​
# 批量生成
codes = ["000001", "600519", "000858"]
all_features = []
for code in codes:
    rows = client.stock_history(code, period="d", dividend="n", lt=120)
    raw = pd.DataFrame(rows)
    raw["code"] = code
    feat = build_features(raw)
    feat["code"] = code
    all_features.append(feat)
​
matrix = pd.concat(all_features, ignore_index=True)
matrix.to_csv("feature_matrix.csv", index=False, encoding="utf-8-sig")
print(f"特征矩阵: {len(matrix)} 行 x {len(matrix.columns)} 列")
print(matrix.groupby("code")[["ret_5", "ret_20", "vol_20"]].mean().round(2))


特征矩阵: 150 行 x 11 列
       ret_5  ret_20  vol_20
code
000001   0.14    1.19    0.80
600519   0.14    1.19    0.80
000858   0.14    1.19    0.80

从原始 OHLCV 到可用特征，核心就三步：

这三个维度覆盖了量化模型对"价格行为"的大部分信息需求。下一篇我们讲：

——怎样确保你清洗后的数据和特征，没有逻辑错误和一致性漏洞。

：更多 SDK 用法与接口文档可参考 

【Python量化数据工程实战 #02】只会看 K 线不会做特征？动量、波动、资金流三因子实战

编程语言

人工智能

算法

构建动量、波动与资金流三类核心特征，pandas从零实现5/10/20日滚动收益率、ATR、滚动标准差、OBV斜率及顶背离检测，封装build_features()函数输出统一特征矩阵，防未来函数，可直接嵌入回测框架。

Python

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

【Python量化数据工程实战 #02】只会看 K 线不会做特征？动量、波动、资金流三因子实战

【Python量化数据工程实战 #02】只会看 K 线不会做特征？动量、波动、资金流三因子实战

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐