当前位置：首页 > news >正文

量化研究--ATR高频量化网格策略实盘研究

news 来源：原创 2025/5/22 7:01:28

今天完成了高频量化网格策略，给了2格模型，固定百分比和自动ATR生成自动网格2格算法

详细的文章介绍

量化研究--ATR高频量化网格策略实盘研究https://mp.weixin.qq.com/s/2NPR1A7HM40LWtU_oTiQvg

策略参数text={

"自定义交易品种交易":"自定义交易类型比如股票，可转债，etf***********","账户支持融资融券":"账户支持融资融券,账户类型STOCK/CREDIT","账户类型":"STOCK","账户":"770","账户类型":"STOCK","是否测试说明":"实盘改成否","是否测试":"否","是否时间测试":"否","测试时间":"20250425","是否立马交易说明":"利用run_time运行策略第一次循环不会下单，但是会记录交易数据需要清空","是否立马交易":"否","买入价格编码":5,"卖出价格编码":5,"是否隔离策略":"否","交易模式说明":"数量/金额","交易模式":"数量","固定交易数量":100,"持有数量限制":300,"固定交易金额":200,"持有金额限制":400,"网格模式说明":"百分比/ATR/","网格模式":"ATR","百分比模式设置":"百分比模式设置","卖出单元格":0.2,"买入单元格":-0.2,
"ATR网格设置":"ATR网格设置********","ATR周期":14,"ATR数据周期":'1d',"是否自动生成ATR倍数":'是',"自定义固定ATR倍数":"自定义固定ATR倍数","ATR倍数":0.01,
"时间设置":"时间设置********","交易时间段":8,"交易开始时间":9,"交易结束时间":24,"是否参加集合竞价":"是","开始交易分钟":0,"监测股票池设置":"监测股票池设置 自定义/持股","监测股票池":"自定义",'自定义股票池':"自定义股票池设置","股票池设置":"持有限制10的股票池设置","自定义股票池":["511090.SH","511130.SH","159655.SZ"],"自定义股票池名称":['30年国债ETF',"30年国债ETF","标普ETF"],
}

我感觉测试了一下ATR比较合适日线网格，所以我在开发日线网格模型，用来做红利ETF，黄金，债券，银行等标的，很合适做网格，波动率稳定

看一下ATR的原理，计算公式，感觉很多是密密麻麻的原理，感觉回到了大学金融工程的课程

def ATR(CLOSE,HIGH,LOW,N=14):    '''    真实波幅    输出MTR:(最高价-最低价)和1日前的收盘价-最高价的绝对值的较大值和1日前的收盘价-最低价的绝对值的较大值    输出真实波幅:MTR的N日简单移动平均    '''    MTR=MAX(MAX((HIGH-LOW),ABS(REF(CLOSE,1)-HIGH)),ABS(REF(CLOSE,1)-LOW))    ATR=MA(MTR,N)    return MTR,ATR

ATR（Average True Range，平均真实波幅）自动生成网格是一种基于市场波动性动态调整交易网格的策略，结合了ATR指标对波动性的衡量和网格交易的机械化操作。以下是其核心原理和关键步骤：

1. ATR指标基础

ATR由韦尔斯·怀尔德提出，用于衡量资产价格的波动性。计算步骤如下：

True Range (TR)：取以下三者最大值
- 当日最高价 - 当日最低价
- 当日最高价 - 前一日收盘价的绝对值
- 当日最低价 - 前一日收盘价的绝对值
ATR：TR的移动平均（通常周期14天）。

ATR值越大，市场波动越剧烈。

2. 网格交易原理

网格交易通过在固定价格区间设置买卖订单，低买高卖赚取差价。传统网格的缺陷是：

固定间距无法适应市场波动变化，可能在剧烈波动时失效。

3. ATR自动生成网格的核心逻辑

动态调整网格间距，使网格间距与当前市场波动性匹配：

确定基准间距：
- 初始网格间距 = 系数 × ATR（例如1.5倍ATR）。
- 系数需根据资产特性和风险偏好调整。
动态调整规则：
- 每隔固定周期（如每日）更新ATR值，重新计算网格间距。
- 波动性上升（ATR增大）→ 拉大网格间距，避免频繁触发订单。
- 波动性下降（ATR减小）→ 缩小网格间距，提高交易频率。
网格层级设计：
- 买入网格：94, 100, 106...
- 卖出网格：100, 106, 112...
- 以当前价格为中心，向上和向下按ATR间距布置买卖订单。
- 例如：当前价100，ATR=5，间距系数=1.2 → 每6点（1.2×5）设置一个订单：
风险控制：
- 总网格宽度：根据最大预期波动范围设定（如±5倍ATR）。
- 资金管理：单网格订单量需适应总资金和波动风险。

这里需要注意K值就是ATR的倍数取值，这里我利用波动率优化了一下

def cacal_auto_k_volatility(atr,lookback=30):'''	基于历史波动率自动计算K值	'''# 计算近期ATR波动率	recent_atr = atr[-lookback:]	atr_std = recent_atr.std()	atr_mean = recent_atr.mean()
# 波动率聚类分析	volatility_ratio = atr_std / atr_mean
# 计算辅助指标	skewness =  pd.Series(atr).skew()	kurtosis =  pd.Series(atr).kurt()
# 多条件判定if volatility_ratio < 0.15:		base_k = 0.7if skewness > 0.5: 			base_k *= 0.9  # 右偏修正elif volatility_ratio < 0.3:		base_k = 1.0if kurtosis > 3: 			base_k *= 1.1   # 尖峰修正elif volatility_ratio < 0.6:		base_k = 1.5if skewness < -0.5: 			base_k *= 1.2  # 左偏放大else:		base_k = 2.0# 波动率趋势调整	last_5 = atr[-5:].mean()if last_5 > 0.05:   # 波动率上升趋势return min(base_k * 1.2, 2.5)elif last_5 < -0.05: # 波动率下降趋势return max(base_k * 0.8, 0.5)return base_k

K的取值怎么样定义简单的波动率

1. 静态K值的一般经验范围

波动率水平	ATR倍数（K值）范围	适用场景
低波动市场	1.0 ~ 1.5×ATR	震荡行情（如股指窄幅盘整）
中等波动市场	1.5 ~ 2.5×ATR	常态趋势（如外汇主要货币对）
高波动市场	2.5 ~ 4.0×ATR	极端行情（如加密货币、商品期货）
超高波动市场	>4.0×ATR	黑天鹅事件（需额外风控）

3. 不同资产类别的K值参考

资产类型	低波动K值	高波动K值	备注
股票（大盘股）	1.0 ~ 1.8	1.8 ~ 2.5	如SP500成分股
外汇（主要货币对）	1.2 ~ 2.0	2.0 ~ 3.0	如EUR/USD、USD/JPY
加密货币（BTC/ETH）	2.0 ~ 3.0	3.0 ~ 5.0	需极端行情风控
商品期货（黄金/原油）	1.5 ~ 2.5	2.5 ~ 4.0	受地缘政治影响大

我开发了自动分享计算atr的框架可以测试数据，估计波动率设置atr的K值大小比如513100 纳斯达克etf我们测试一下

atr网格优化 股票：513100 价格1.522  atr:0.022642857142857138 K:1.2 实际值:0.027171428571428565 对应涨跌幅：1.7852449784118638

比如红利etf 512890

atr网格优化 股票：512890 价格1.144  atr:0.010928571428571414 K:1.2 实际值:0.013114285714285696 对应涨跌幅：1.1463536463536448

比如工商银行601398

atr网格优化 股票：601398 价格7.23  atr:0.1385714285714286 K:0.8 实际值:0.11085714285714288 对应涨跌幅：1.533293815451492

我们发现atr可能更合适日线网格，高频网格建议使用30分钟数据计算网格

我给了atr的分析框架

atr的测试框架源代码

import akshare as akimport pandas as pdimport numpy  as npfrom trader_tool.unification_data import unification_datadata=unification_data(trader_tool='qmt',data_api='dfcf')data=data.get_unification_data()def RET(S,N=1):    '''    返回序列倒数第N个值,默认返回最后一个    '''    return np.array(S)[-N]     def ABS(S):      '''    返回N的绝对值    '''       return np.abs(S)            def MAX(S1,S2):     '''    序列max    '''    return np.maximum(S1,S2)    def MIN(S1,S2):      '''    序列min    '''    return np.minimum(S1,S2)   def IF(S,A,B):       '''    序列布尔判断 return=A  if S==True  else  B    '''    return np.where(S,A,B)      def REF(S, N=1):      '''            对序列整体下移动N,返回序列(shift后会产生NAN)        '''    return pd.Series(S).shift(N).values  def MA(S,N):        '''    求序列的N日简单移动平均值，返回序列         '''                             return pd.Series(S).rolling(N).mean().values  def ATR(CLOSE,HIGH,LOW,N=14):    '''    真实波幅    输出MTR:(最高价-最低价)和1日前的收盘价-最高价的绝对值的较大值和1日前的收盘价-最低价的绝对值的较大值    输出真实波幅:MTR的N日简单移动平均    '''    MTR=MAX(MAX((HIGH-LOW),ABS(REF(CLOSE,1)-HIGH)),ABS(REF(CLOSE,1)-LOW))    ATR=MA(MTR,N)    return MTR,ATRdef calculate_atr(data, window=14):    """    计算ATR（Average True Range）    """    high = data['high']    low = data['low']    close = data['close']        # 计算 True Range (TR)    tr1 = high - low    tr2 = abs(high - close.shift(1))    tr3 = abs(low - close.shift(1))    tr = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1)        # 计算 ATR（SMA平滑）    atr = tr.rolling(window).mean()    return atrdef cacal_auto_k_volatility(atr,lookback=30):    '''    基于历史波动率自动计算K值    '''    # 计算近期ATR波动率    recent_atr = atr[-lookback:]    atr_std = recent_atr.std()    atr_mean = recent_atr.mean()        # 波动率聚类分析    volatility_ratio = atr_std / atr_mean        # 动态K值计算    if volatility_ratio < 0.2:        k = 0.8  # 低波动环境    elif volatility_ratio < 0.5:        k = 1.2  # 正常波动    else:        k = 1.8  # 高波动环境    return k#print(data)stock='601398'start_date='20240101'end_date='20500101'df=data.get_hist_data_em(stock=stock,start_date=start_date,end_date=end_date)CLOSE=df['close']HIGH=df['high']LOW=df['low']price=CLOSE.tolist()[-1]mtr,atr=ATR(CLOSE,HIGH,LOW)atr_value=atr.tolist()[-1]K=cacal_auto_k_volatility(atr)adjust_value=atr_value*Kzdf=(adjust_value/price)*100print("atr网格优化 股票：{} 价格{}  atr:{} K:{} 实际值:{} 对应涨跌幅：{} ".format(stock,price,atr_value,K,adjust_value,zdf))

atr的调整框架

import numpy as npimport pandas as pdfrom typing import Dict, Listdef RET(S,N=1):    '''    返回序列倒数第N个值,默认返回最后一个    '''    return np.array(S)[-N]     def ABS(S):      '''    返回N的绝对值    '''       return np.abs(S)            def MAX(S1,S2):     '''    序列max    '''    return np.maximum(S1,S2)    def MIN(S1,S2):      '''    序列min    '''    return np.minimum(S1,S2)   def IF(S,A,B):       '''    序列布尔判断 return=A  if S==True  else  B    '''    return np.where(S,A,B)      def REF(S, N=1):      '''            对序列整体下移动N,返回序列(shift后会产生NAN)        '''    return pd.Series(S).shift(N).values  def MA(S,N):        '''    求序列的N日简单移动平均值，返回序列         '''                             return pd.Series(S).rolling(N).mean().values  def ATR(CLOSE,HIGH,LOW,N=14):    '''    真实波幅    输出MTR:(最高价-最低价)和1日前的收盘价-最高价的绝对值的较大值和1日前的收盘价-最低价的绝对值的较大值    输出真实波幅:MTR的N日简单移动平均    '''    MTR=MAX(MAX((HIGH-LOW),ABS(REF(CLOSE,1)-HIGH)),ABS(REF(CLOSE,1)-LOW))    ATR=MA(MTR,N)    return MTR,ATRclass VolatilityGridGenerator:    def __init__(self,                  atr_period: int = 14,                  volatility_lookback: int = 30,                 max_layers: int = 10,                 risk_per_trade: float = 0.01):        """        基于波动率自动生成网格的交易系统                参数:            atr_period: ATR计算周期            volatility_lookback: 波动率分析回溯期            max_layers: 最大网格层数            risk_per_trade: 单笔交易风险比例        """        self.atr_period = atr_period        self.vol_lookback = volatility_lookback        self.max_layers = max_layers        self.risk_per_trade = risk_per_trade        self.volatility_history = []            def calculate_volatility(self, prices: pd.DataFrame) -> Dict[str, float]:        """计算多维度波动率指标"""        mtr,atr = ATR(prices['high'], prices['low'], prices['close'], self.atr_period)        current_atr = atr.tolist()[-1]        current_price = prices['close'].iloc[-1]                # 标准化ATR波动率        atr_vol = current_atr / current_price               # 历史波动率(20日年化)        log_returns = np.log(prices['close']/prices['close'].shift(1))        hist_vol = log_returns.std() * np.sqrt(252)                # 极差波动率        daily_range = (prices['high'] - prices['low']) / prices['close']        range_vol = daily_range.rolling(5).mean().iloc[-1]                # 波动率聚集指标        squared_returns = log_returns ** 2        vol_clustering = squared_returns.autocorr(lag=1)        return {            'atr': current_atr,            'atr_vol': atr_vol,            'hist_vol': hist_vol,            'range_vol': range_vol,            'vol_clustering': vol_clustering        }    def determine_volatility_regime(self, vol_metrics: Dict[str, float]) -> str:        """识别当前波动率状态"""        composite_vol = 0.5*vol_metrics['atr_vol'] + 0.3*vol_metrics['hist_vol'] + 0.2*vol_metrics['range_vol']                if composite_vol < 0.01:            return 'ultra_low'        elif composite_vol < 0.02:            return 'low'        elif composite_vol < 0.04:            return 'moderate'        elif composite_vol < 0.06:            return 'high'        else:            return 'extreme'        def calculate_dynamic_k(self, vol_regime: str) -> float:        """根据波动状态确定K值"""        regime_params = {            'ultra_low': 0.6,            'low': 0.8,            'moderate': 1.2,            'high': 1.6,            'extreme': 2.0        }        return regime_params.get(vol_regime, 1.0)        def calculate_optimal_layers(self,                                vol_metrics: Dict[str, float],                               account_size: float,                               current_price: float) -> int:        """计算最优网格层数"""        # 资金约束计算        risk_capital = account_size * self.risk_per_trade        contract_size = risk_capital / vol_metrics['atr']        margin_per_grid = contract_size * current_price * 0.1  # 假设10%保证金        capital_constrained = int(account_size * 0.5 / margin_per_grid)                # 波动率约束        if vol_metrics['atr_vol'] < 0.01:            vol_constrained = 10        elif vol_metrics['atr_vol'] < 0.03:            vol_constrained = 7        elif vol_metrics['atr_vol'] < 0.05:            vol_constrained = 5        else:            vol_constrained = 3                    return min(vol_constrained, capital_constrained, self.max_layers)        def generate_grid(self,                     prices: pd.DataFrame,                     account_size: float = 10000) -> Dict[str, List[float]]:        """        生成动态网格                返回:            {                'buy_levels': [买入网格价格],                'sell_levels': [卖出网格价格],                'current_price': 当前价格,                'atr': 当前ATR值,                'k': 使用的K值,                'layers': 网格层数,                'vol_regime': 波动状态            }        """        # 计算波动率指标        vol_metrics = self.calculate_volatility(prices)        current_price = prices['close'].iloc[-1]                # 识别波动状态        vol_regime = self.determine_volatility_regime(vol_metrics)                # 计算动态参数        k = self.calculate_dynamic_k(vol_regime)        layers = self.calculate_optimal_layers(vol_metrics, account_size, current_price)                # 计算网格间距        grid_spacing = vol_metrics['atr'] * k                # 生成网格        buy_levels = [round(current_price - (i * grid_spacing), 4)                      for i in range(1, layers+1)]        sell_levels = [round(current_price + (i * grid_spacing), 4)                      for i in range(1, layers+1)]                return {            'buy_levels': sorted(buy_levels),            'sell_levels': sorted(sell_levels),            'current_price': round(current_price, 4),            'atr': round(vol_metrics['atr'], 4),            'k': round(k, 2),            'layers': layers,            'vol_regime': vol_regime        }# 使用示例if __name__ == "__main__":    # 生成示例数据    np.random.seed(42)    dates = pd.date_range(start='2023-01-01', periods=100)    prices = pd.DataFrame({        'open': np.cumprod(1 + np.random.normal(0.001, 0.02, 100)),        'high': np.cumprod(1 + np.random.normal(0.001, 0.025, 100)),        'low': np.cumprod(1 + np.random.normal(0.001, 0.025, 100)),        'close': np.cumprod(1 + np.random.normal(0.001, 0.02, 100))    }, index=dates)        # 初始化网格生成器    grid_generator = VolatilityGridGenerator(max_layers=8)        # 生成网格    grid = grid_generator.generate_grid(prices, account_size=50000)        # 打印结果    print(f"当前价格: {grid['current_price']}")    print(f"ATR值: {grid['atr']}")    print(f"波动状态: {grid['vol_regime']}")    print(f"K值: {grid['k']}")    print(f"网格层数: {grid['layers']}")    print("\n买入网格:")    for level in grid['buy_levels']:        print(f"{level:.4f}")    print("\n卖出网格:")    for level in grid['sell_levels']:        print(f"{level:.4f}")