Python 量化入门：用 pandas 处理 A 股 K 线数据并计算技术指标

2601_96288572

218人浏览 · 2026-06-09 10:22:44

2601_96288572 · 2026-06-09 10:22:44 发布

Python 量化入门：用 pandas 处理 A 股 K 线数据并计算技术指标

量化交易听起来很高大上，但本质就是用代码处理数据、发现规律。这篇文章从最基础的 K 线数据开始，教你用 pandas 完成数据获取、清洗、技术指标计算的全流程。不需要任何量化框架，纯 Python + pandas 就够了。

一、K 线数据长什么样？

K 线（OHLCV）是量化分析的基础数据，每一根 K 线包含：

字段	含义
open	开盘价
high	最高价
low	最低价
close	收盘价
volume	成交量
amount	成交额

一根日 K 线就是一天的价格摘要。把很多根日 K 线排起来，就是我们常说的"走势图"。

量化分析的核心工作就是对这些数据做计算，比如算均线、判断金叉死叉、计算涨跌幅排名等。

二、获取 K 线数据

获取 A 股数据有很多方式。这里用 alphafeed 做示例（也可以换成你习惯的数据源，后面的 pandas 处理逻辑完全一样）：

pip install alphafeed

from alphafeed import AlphaFeed

af = AlphaFeed(api_key="your-api-key")  # 也支持环境变量 ALPHAFEED_API_KEY

# 获取贵州茅台最近 500 根日 K 线
df = af.klines.get("600519.SH", period="1d", count=500, to_dataframe=True)
print(df.shape)
print(df.columns.tolist())

拿到的 DataFrame 大概长这样：

(500, 9)
['symbol', 'timestamp', 'trade_date', 'open', 'high', 'low', 'close', 'volume', 'amount']

不管用什么数据源，只要最终拿到一个包含 open/high/low/close/volume 的 DataFrame，后面的分析流程都是通用的。

三、数据初步探索

拿到数据后，先做基本检查：

# 看前几行
print(df.head())

# 看数据类型
print(df.dtypes)

# 检查缺失值
print(df.isnull().sum())

# 基本统计
print(df[["open", "high", "low", "close", "volume"]].describe())

常见操作：把 trade_date 设为索引，方便按时间切片：

df["trade_date"] = pd.to_datetime(df["trade_date"])
df = df.set_index("trade_date").sort_index()

# 取最近一年
df_recent = df.loc["2025-06-01":]

四、计算常用技术指标

1. 涨跌幅

df["change_pct"] = df["close"].pct_change() * 100
print(df[["close", "change_pct"]].tail(10))

2. 移动平均线（MA）

均线是最基础的技术指标，用 rolling 一行搞定：

df["ma5"] = df["close"].rolling(5).mean()
df["ma10"] = df["close"].rolling(10).mean()
df["ma20"] = df["close"].rolling(20).mean()
df["ma60"] = df["close"].rolling(60).mean()

print(df[["close", "ma5", "ma10", "ma20", "ma60"]].tail(10))

怎么看均线？

短期均线在长期均线上方 → 多头排列（趋势向上）
短期均线从下往上穿过长期均线 → 金叉（买入信号）
反之 → 死叉（卖出信号）

3. MACD

MACD 由三部分组成：DIF（快线-慢线）、DEA（DIF 的均线）、柱状图（DIF-DEA）。

def calc_macd(close, fast=12, slow=26, signal=9):
    ema_fast = close.ewm(span=fast, adjust=False).mean()
    ema_slow = close.ewm(span=slow, adjust=False).mean()
    dif = ema_fast - ema_slow
    dea = dif.ewm(span=signal, adjust=False).mean()
    macd_bar = (dif - dea) * 2
    return dif, dea, macd_bar

df["dif"], df["dea"], df["macd"] = calc_macd(df["close"])

print(df[["close", "dif", "dea", "macd"]].tail(10))

怎么用？

DIF 上穿 DEA → MACD 金叉
MACD 柱从负变正 → 趋势可能反转

4. RSI（相对强弱指标）

RSI 衡量"最近一段时间涨了多少、跌了多少"，范围 0-100。

def calc_rsi(close, period=14):
    delta = close.diff()
    gain = delta.where(delta > 0, 0)
    loss = -delta.where(delta < 0, 0)
    avg_gain = gain.rolling(period).mean()
    avg_loss = loss.rolling(period).mean()
    rs = avg_gain / avg_loss
    return 100 - (100 / (1 + rs))

df["rsi_6"] = calc_rsi(df["close"], 6)
df["rsi_14"] = calc_rsi(df["close"], 14)

print(df[["close", "rsi_6", "rsi_14"]].tail(10))

怎么看？

RSI > 70 → 超买，可能回调
RSI < 30 → 超卖，可能反弹
不同周期的 RSI 组合使用效果更好

5. 布林带（Bollinger Bands）

布林带由中轨（均线）和上下轨（均线 ± N 倍标准差）组成：

period = 20
df["boll_mid"] = df["close"].rolling(period).mean()
df["boll_std"] = df["close"].rolling(period).std()
df["boll_upper"] = df["boll_mid"] + 2 * df["boll_std"]
df["boll_lower"] = df["boll_mid"] - 2 * df["boll_std"]

print(df[["close", "boll_upper", "boll_mid", "boll_lower"]].tail(10))

怎么用？

价格触及上轨 → 可能超买
价格触及下轨 → 可能超卖
布林带收窄 → 可能即将变盘

6. 成交量分析

df["vol_ma5"] = df["volume"].rolling(5).mean()
df["vol_ma20"] = df["volume"].rolling(20).mean()

# 量比：今日成交量 / 近 20 日平均
df["vol_ratio"] = df["volume"] / df["vol_ma20"]

# 筛选放量日（量比 > 2）
big_vol_days = df[df["vol_ratio"] > 2]
print(f"放量交易日: {len(big_vol_days)} 天")
print(big_vol_days[["close", "change_pct", "volume", "vol_ratio"]].tail())

五、生成交易信号

有了技术指标，就可以定义交易规则了。以最经典的双均线策略为例：

MA5 上穿 MA20 → 买入
MA5 下穿 MA20 → 卖出

# 判断金叉死叉
df["signal"] = 0

# 金叉：今天 ma5 > ma20，昨天 ma5 <= ma20
buy_mask = (df["ma5"] > df["ma20"]) & (df["ma5"].shift(1) <= df["ma20"].shift(1))
df.loc[buy_mask, "signal"] = 1

# 死叉
sell_mask = (df["ma5"] < df["ma20"]) & (df["ma5"].shift(1) >= df["ma20"].shift(1))
df.loc[sell_mask, "signal"] = -1

signals = df[df["signal"] != 0]
print(f"共产生 {len(signals)} 个信号")
print(signals[["close", "ma5", "ma20", "signal"]].tail(10))

你可以把这个逻辑换成任何其他指标组合，比如：

# MACD 金叉 + RSI 未超买
buy_macd = (df["dif"] > df["dea"]) & (df["dif"].shift(1) <= df["dea"].shift(1))
buy_rsi = df["rsi_14"] < 70
combined_buy = buy_macd & buy_rsi

六、批量处理多只股票

实际做量化，需要对多只股票批量计算指标：

from alphafeed import AlphaFeed

af = AlphaFeed(api_key="your-api-key")

symbols = ["600519.SH", "000858.SZ", "601318.SH", "000001.SZ", "600036.SH"]
dfs = af.klines.batch(symbols, period="1d", count=200, to_dataframe=True, show_progress=True)

results = []
for symbol, df in dfs.items():
    if len(df) < 60:
        continue

    df["ma5"] = df["close"].rolling(5).mean()
    df["ma20"] = df["close"].rolling(20).mean()
    df["rsi_14"] = calc_rsi(df["close"], 14)

    latest = df.iloc[-1]
    results.append({
        "symbol": symbol,
        "close": latest["close"],
        "ma5>ma20": latest["ma5"] > latest["ma20"],
        "rsi_14": round(latest["rsi_14"], 1),
    })

import pandas as pd
print(pd.DataFrame(results).to_string(index=False))