问题描述
帮我写一个基于 KAMA 自适应均线的量化策略
解决方案
这是一个基于 Kaufman's Adaptive Moving Average (KAMA, 考夫曼自适应均线) 的 QMT 量化策略。
策略原理
KAMA 是一种智能均线，它根据市场的“噪音”水平自动调整灵敏度：
- 趋势明显（噪音低）时：KAMA 变得灵敏，紧随价格，以便快速捕捉趋势。
- 市场震荡（噪音高）时：KAMA 变得平滑（迟钝），避免被频繁的假突破止损。
核心逻辑：
- 买入信号：收盘价 上穿 KAMA 均线。
- 卖出信号：收盘价 下穿 KAMA 均线。
策略代码
# -*- coding: gbk -*-
import pandas as pd
import numpy as np
import talib
def init(ContextInfo):
"""
策略初始化函数
"""
# 1. 设置资金账号 (请替换为您自己的资金账号)
ContextInfo.accID = '6000000000'
ContextInfo.set_account(ContextInfo.accID)
# 2. 设置策略参数
# KAMA 周期 (效率比率 ER 的计算周期)
ContextInfo.n = 10
# 快速均线周期 (用于计算平滑系数，对应高ER)
ContextInfo.fast_period = 2
# 慢速均线周期 (用于计算平滑系数，对应低ER)
ContextInfo.slow_period = 30
# 3. 设置股票池 (示例：平安银行)
# 如果在界面上运行，可以使用 ContextInfo.get_universe() 获取界面勾选的股票
ContextInfo.target_list = ['000001.SZ']
ContextInfo.set_universe(ContextInfo.target_list)
# 4. 设置交易费率等 (回测用)
ContextInfo.set_commission(0, [0.0001, 0.0001, 0.0003, 0.0003, 0.0003, 5])
def calculate_kama_manual(close_series, n, fast_p, slow_p):
"""
手动计算 KAMA (如果不使用 talib 或需要自定义参数)
公式:
ER = Change / Volatility
SC = [ER * (fastest_SC - slowest_SC) + slowest_SC] ^ 2
KAMA = Prior_KAMA + SC * (Price - Prior_KAMA)
"""
prices = np.array(close_series)
length = len(prices)
kama = np.zeros(length)
kama[:] = np.nan
if length <= n:
return kama
# 计算平滑常数
fast_sc = 2 / (fast_p + 1)
slow_sc = 2 / (slow_p + 1)
# 第一个 KAMA 值通常设为第 n 个周期的收盘价或简单移动平均
kama[n-1] = prices[n-1]
for i in range(n, length):
# 1. 效率比率 (Efficiency Ratio, ER)
change = abs(prices[i] - prices[i-n])
volatility = np.sum(np.abs(np.diff(prices[i-n:i+1])))
if volatility == 0:
er = 0
else:
er = change / volatility
# 2. 平滑系数 (Smoothing Constant, SC)
sc = (er * (fast_sc - slow_sc) + slow_sc) ** 2
# 3. KAMA 计算
kama[i] = kama[i-1] + sc * (prices[i] - kama[i-1])
return kama
def get_position(ContextInfo, stock_code):
"""
获取当前持仓数量
"""
# 获取持仓对象列表
positions = get_trade_detail_data(ContextInfo.accID, 'stock', 'position')
for pos in positions:
if pos.m_strInstrumentID + '.' + pos.m_strExchangeID == stock_code:
return pos.m_nVolume
return 0
def handlebar(ContextInfo):
"""
K线周期回调函数
"""
# 获取当前 K 线位置
index = ContextInfo.barpos
# 获取当前时间
realtime = ContextInfo.get_bar_timetag(index)
current_date = timetag_to_datetime(realtime, '%Y-%m-%d %H:%M:%S')
# 遍历股票池
for stock in ContextInfo.target_list:
# 1. 获取历史行情数据
# 我们需要足够的数据来计算 KAMA，这里取 n + 100 根
count = ContextInfo.n + 100
# 使用 get_market_data_ex 获取数据 (推荐)
# 注意：period='follow' 跟随主图周期，也可以强制指定 '1d'
data_map = ContextInfo.get_market_data_ex(
['close'],
[stock],
period='follow',
count=count,
dividend_type='front' # 前复权
)
if stock not in data_map:
continue
df = data_map[stock]
if len(df) < ContextInfo.n + 1:
continue
# 获取收盘价序列
close_prices = df['close'].values
# 2. 计算 KAMA 指标
# 方法A: 使用 talib (如果环境支持，速度快，但参数通常固定为 n)
# kama_values = talib.KAMA(close_prices, timeperiod=ContextInfo.n)
# 方法B: 使用自定义函数 (支持 fast/slow 参数调整)
kama_values = calculate_kama_manual(
close_prices,
ContextInfo.n,
ContextInfo.fast_period,
ContextInfo.slow_period
)
# 获取当前和上一根 K 线的数值
curr_close = close_prices[-1]
prev_close = close_prices[-2]
curr_kama = kama_values[-1]
prev_kama = kama_values[-2]
# 如果 KAMA 计算结果无效(NaN)，跳过
if np.isnan(curr_kama) or np.isnan(prev_kama):
continue
# 3. 生成交易信号
# 金叉：收盘价上穿 KAMA
long_signal = (prev_close <= prev_kama) and (curr_close > curr_kama)
# 死叉：收盘价下穿 KAMA
short_signal = (prev_close >= prev_kama) and (curr_close < curr_kama)
# 4. 执行交易
# 获取当前持仓
pos_vol = get_position(ContextInfo, stock)
# 交易逻辑
if long_signal:
if pos_vol == 0:
# 全仓买入 (示例：可用资金的 90%)
# 注意：实盘中请根据风控调整
ContextInfo.paint('buy', curr_close, -1, 0, 'red', 'noaxis')
print(f"{current_date} [买入信号] {stock} 价格:{curr_close:.2f} 上穿 KAMA:{curr_kama:.2f}")
# 下单：买入，使用最新价，固定数量或金额
# opType=23(买入), orderType=1101(固定数量), volume=1000股
passorder(23, 1101, ContextInfo.accID, stock, 5, -1, 1000, ContextInfo)
elif short_signal:
if pos_vol > 0:
# 清仓卖出
ContextInfo.paint('sell', curr_close, -1, 0, 'green', 'noaxis')
print(f"{current_date} [卖出信号] {stock} 价格:{curr_close:.2f} 下穿 KAMA:{curr_kama:.2f}")
# 下单：卖出，使用最新价，卖出全部持仓
passorder(24, 1101, ContextInfo.accID, stock, 5, -1, pos_vol, ContextInfo)
# 5. 绘图 (可选，用于回测界面展示)
# 绘制 KAMA 线
ContextInfo.paint('KAMA', curr_kama, -1, 0, 'yellow')
代码关键点解析
-
calculate_kama_manual函数：- 虽然 QMT 内置了
talib库，但标准的talib.KAMA函数通常只允许设置效率周期n，而将快速和慢速均线的参数固定。 - 为了完全还原考夫曼的策略逻辑，我编写了一个纯 Python/Numpy 版本的计算函数，允许你自定义
fast_period(通常为2) 和slow_period(通常为30)。
- 虽然 QMT 内置了
-
数据获取 (
get_market_data_ex)：- 使用了
get_market_data_ex接口，这是 QMT 推荐的新版数据接口。 dividend_type='front'设置为前复权，这对于技术指标回测非常重要，避免因分红派息导致的价格跳空产生错误信号。
- 使用了
-
信号逻辑：
- 使用了简单的交叉逻辑：
prev_close <= prev_kama且curr_close > curr_kama判定为上穿。
- 使用了简单的交叉逻辑：
-
交易执行 (
passorder)：opType=23代表股票买入，24代表股票卖出。orderType=1101代表按指定股数下单。prType=5代表使用最新价下单。- 注意：代码中示例为固定买入 1000 股，实盘或回测时建议修改为
order_value(按金额) 或order_target_percent(按仓位比例) 等函数以更好地管理资金。
如何在 QMT 中使用
- 打开 QMT 客户端，进入 “模型研究” -> “策略编辑器”。
- 新建一个 Python 策略。
- 将上述代码完整复制粘贴进去。
- 修改
init函数中的ContextInfo.accID为你自己的资金账号。 - 点击 “编译” 检查是否有语法错误。
- 点击 “回测”，选择标的（如平安银行）、周期（如日线）和时间段进行测试。
常见问题 Q&A
Q: 为什么 KAMA 有时候会走平？
A: 这是 KAMA 的特性。当市场处于震荡期（价格频繁上下波动但位移不大），效率比率（ER）会变得很小，导致平滑系数（SC）趋近于 0。此时 KAMA 会保持前值不变，从而过滤掉震荡噪音，不产生错误的交叉信号。
Q: 如何调整策略的灵敏度？
A:
- 减小
ContextInfo.n(如从 10 改为 5)：ER 计算更敏感，均线变快。 - 减小
ContextInfo.slow_period(如从 30 改为 20)：在震荡市中均线也会稍微移动，不再那么“死板”。
Q: 实盘需要注意什么？
A: 实盘时请确保 ContextInfo.accID 设置正确，并且在 init 中调用了 set_account。此外，建议在 handlebar 开头添加 if not ContextInfo.is_last_bar(): return 逻辑，如果你只想在最新的一根 K 线（实时行情）触发交易，或者使用 do_order 配合定时器来控制下单时机。