问题描述
请编写一个基于 ASI 振动升降指标的量化策略
解决方案
这是一个基于 ASI (Accumulation Swing Index，振动升降指标) 的量化交易策略。
策略原理
ASI 指标由 Welles Wilder 发明，旨在通过分析开盘价、收盘价、最高价和最低价之间的关系，来反映市场的真实强度。它试图透过价格的短期波动，捕捉潜在的趋势突破。
本策略逻辑：
- 计算 ASI 线：根据 Wilder 的公式计算每日的 Swing Index (SI)，并累加得到 ASI。
- 计算 ASI 均线 (MA_ASI)：对 ASI 线进行移动平均（例如 20 日均线），作为信号线。
- 交易信号：
- 买入：当 ASI 上穿 ASI 均线时，视为上升趋势确立，买入。
- 卖出：当 ASI 下穿 ASI 均线时，视为趋势转弱，卖出。
- 资金管理：等权重分配资金买入符合条件的股票。
策略代码
# -*- coding: utf-8 -*-
import numpy as np
import pandas as pd
def initialize(context):
"""
初始化函数
"""
# 设定基准
set_benchmark('000300.XSHG')
# 开启动态复权模式(真实价格)
set_option('use_real_price', True)
# 过滤掉order系列API产生的比error级别低的log
log.set_level('order', 'error')
# --- 策略参数设置 ---
# 操作的股票池：这里示例使用沪深300成分股的前10只，实际使用可修改
g.security_pool = get_index_stocks('000300.XSHG')[:10]
# ASI计算所需的数据长度 (需要足够长以形成稳定的趋势)
g.asi_window = 60
# ASI均线周期 (信号线)
g.ma_period = 20
# 设置交易费用
set_order_cost(OrderCost(close_tax=0.001, open_commission=0.0003, close_commission=0.0003, min_commission=5), type='stock')
# 每天开盘时运行
run_daily(market_open, time='09:30')
def calculate_asi(df, limit_move_pct=0.10):
"""
计算ASI指标
df: 包含 open, close, high, low, pre_close 的 DataFrame
limit_move_pct: 每日涨跌停限制比例，A股通常为10%
"""
# 提取数据
close = df['close'].values
open_p = df['open'].values
high = df['high'].values
low = df['low'].values
# 获取前一日数据 (shift 1)
prev_close = np.roll(close, 1)
prev_open = np.roll(open_p, 1)
# 第一天数据无效，设为0或处理为NaN
prev_close[0] = close[0]
prev_open[0] = open_p[0]
# 计算 Wilder 公式中的变量
# A = |High - PrevClose|
A = np.abs(high - prev_close)
# B = |Low - PrevClose|
B = np.abs(low - prev_close)
# C = |High - Low|
C = np.abs(high - low)
# D = |Close - PrevClose|
D = np.abs(close - prev_close)
# 计算 R 值 (根据 A, B, C 的最大值决定公式)
# R 的计算比较复杂，需要向量化处理
R = np.zeros_like(close)
# 条件掩码
cond_A = (A >= B) & (A >= C)
cond_B = (B > A) & (B >= C)
cond_C = (C > A) & (C > B)
# R = TR - 0.5 * |Close - PrevClose| + 0.25 * |PrevClose - PrevOpen|
# 注意：Wilder公式中 R 的定义根据最大波动幅度不同而不同
# Case 1: A is max
R[cond_A] = A[cond_A] + 0.5 * A[cond_A] - 0.25 * np.abs(prev_close[cond_A] - prev_open[cond_A]) # 这里简化处理，通常 R ≈ A
# 更精确的 Wilder 公式:
# If A is largest: R = A + 0.5 * B + 0.25 * D (注意这里的D定义可能不同，这里使用标准ASI公式逻辑)
# 让我们使用标准公式：
# K = Max(A, B)
# L = Limit Move (涨跌停限制价格变动值)
# 重新严格按照公式计算 R:
# A = |H1 - C0|, B = |L1 - C0|, C = |H1 - L1|, D = |C1 - O1|? No, formulas vary.
# 采用通用的标准计算方式：
# R = TR + 0.5 * |C1 - C0| + 0.25 * |C0 - O0| (如果 A 最大) -> 这种变体较多
# 我们采用最通用的计算 R 的逻辑：
# R = A + 0.5 * B + 0.25 * |C0 - O0| (If A is Max)
# R = B + 0.5 * A + 0.25 * |C0 - O0| (If B is Max)
# R = C + 0.25 * |C0 - O0| (If C is Max)
abs_prev_close_open = np.abs(prev_close - prev_open)
R[cond_A] = A[cond_A] + 0.5 * B[cond_A] + 0.25 * abs_prev_close_open[cond_A]
R[cond_B] = B[cond_B] + 0.5 * A[cond_B] + 0.25 * abs_prev_close_open[cond_B]
R[cond_C] = C[cond_C] + 0.25 * abs_prev_close_open[cond_C]
# 防止 R 为 0
R[R == 0] = 1e-9
# 计算 X
# X = C1 - C0 + 0.5 * (C1 - O1) + 0.25 * (C0 - O0)
X = (close - prev_close) + 0.5 * (close - open_p) + 0.25 * (prev_close - prev_open)
# 计算 K
# K = Max(A, B)
K = np.maximum(A, B)
# 计算 L (Limit Move)
# L 通常指每日最大允许波动值。在A股是前收盘价 * 10%
L = prev_close * limit_move_pct
# 防止 L 为 0
L[L == 0] = close[0] * limit_move_pct
# 计算 SI (Swing Index)
# SI = 50 * (X / R) * (K / L)
SI = 50 * (X / R) * (K / L)
# 计算 ASI (Accumulation Swing Index)
ASI = np.cumsum(SI)
return ASI
def market_open(context):
"""
每日交易逻辑
"""
# 获取当前持仓
current_positions = context.portfolio.positions.keys()
# 待买入列表
buy_list = []
# 待卖出列表
sell_list = []
for stock in g.security_pool:
# 获取历史数据，多取一些数据以计算均线
# 长度 = ASI计算窗口 + 均线周期
data_len = g.asi_window + g.ma_period
# 获取行情数据
h_data = attribute_history(stock, data_len, '1d', ['open', 'close', 'high', 'low', 'pre_close'], skip_paused=True)
if len(h_data) < data_len:
continue
# 计算 ASI
asi_values = calculate_asi(h_data)
# 计算 ASI 的均线 (MA_ASI)
# 我们只需要最后两个点的均线值来判断交叉
ma_asi = pd.Series(asi_values).rolling(window=g.ma_period).mean().values
# 获取当前和前一天的 ASI 及 MA_ASI
curr_asi = asi_values[-1]
prev_asi = asi_values[-2]
curr_ma = ma_asi[-1]
prev_ma = ma_asi[-2]
# 检查数据有效性
if np.isnan(curr_ma) or np.isnan(prev_ma):
continue
# --- 交易信号判断 ---
# 金叉：ASI 上穿 MA_ASI
if prev_asi < prev_ma and curr_asi > curr_ma:
if stock not in current_positions:
buy_list.append(stock)
# 死叉：ASI 下穿 MA_ASI
elif prev_asi > prev_ma and curr_asi < curr_ma:
if stock in current_positions:
sell_list.append(stock)
# --- 执行交易 ---
# 1. 卖出
for stock in sell_list:
order_target_value(stock, 0)
log.info("ASI死叉，卖出: %s" % stock)
# 2. 买入
# 简单的资金分配：将可用资金平均分配给要买入的股票
if len(buy_list) > 0:
cash_per_stock = context.portfolio.available_cash / len(buy_list)
for stock in buy_list:
order_value(stock, cash_per_stock)
log.info("ASI金叉，买入: %s" % stock)
策略说明
-
ASI 计算核心：
- 代码中
calculate_asi函数实现了 Wilder 的标准公式。 - $R$ 值根据当日最高价、最低价与前一日收盘价的关系动态计算，这是 ASI 指标的核心，用于衡量价格波动的“真实”幅度。
- $L$ (Limit Move) 设定为前收盘价的 10%（A股涨跌停限制），这符合 ASI 设计初衷，即衡量价格相对于最大允许波动的强度。
- 代码中
-
信号生成：
- 策略不直接使用 ASI 的数值大小（因为 ASI 是累积值，数值本身意义不大），而是使用 ASI 及其均线的交叉。
- 当 ASI 突破均线，意味着累积的买卖力量发生了短期趋势性的逆转。
-
注意事项：
- 股票池：示例代码使用了沪深300的前10只股票作为演示。在实盘或正式回测中，建议根据您的需求修改
g.security_pool（例如使用get_index_stocks('000300.XSHG')获取全部成分股，或结合基本面因子选股）。 - 参数调整：
g.ma_period(均线周期) 决定了策略的灵敏度。周期越短，交易越频繁；周期越长，趋势确认越慢但越稳定。
- 股票池：示例代码使用了沪深300的前10只股票作为演示。在实盘或正式回测中，建议根据您的需求修改
常见问题 Q&A
Q: 为什么 ASI 指标在某些软件上看起来和这个不一样？
A: ASI 的计算公式中包含一个 $L$ (Limit Move) 参数，代表每日最大价格变动限制。在美股或其他市场，这个值可能被设为固定常数。在 A 股市场，我们将其动态设为 pre_close * 0.1。不同的 $L$ 值会导致 ASI 的绝对数值不同，但其形态和趋势方向通常是一致的。
Q: 这个策略适合什么行情？
A: ASI 是一个趋势确认指标。它在趋势明显的行情（单边上涨或下跌）中表现较好，能过滤掉虚假的突破。在横盘震荡市中，ASI 可能会频繁产生交叉信号，导致磨损，建议结合其他过滤器（如 ADX 或 价格均线）使用。
Q: 如何查看回测结果？
A: 将上述代码复制到聚宽的“策略研究”或“我的策略”中，点击“编译运行”或“运行回测”，选择回测时间段（例如 2022-01-01 至 2023-01-01），即可查看收益曲线。