量化研究--ATR高频量化网格策略实盘研究
今天完成了高频量化网格策略，给了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).valuesdef MA(S,N):'''求序列的N日简单移动平均值，返回序列'''return pd.Series(S).rolling(N).mean().valuesdef 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 - lowtr2 = 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).valuesdef MA(S,N):'''求序列的N日简单移动平均值，返回序列'''return pd.Series(S).rolling(N).mean().valuesdef 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_periodself.vol_lookback = volatility_lookbackself.max_layers = max_layersself.risk_per_trade = risk_per_tradeself.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 ** 2vol_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_tradecontract_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 = 10elif vol_metrics['atr_vol'] < 0.03:vol_constrained = 7elif vol_metrics['atr_vol'] < 0.05:vol_constrained = 5else:vol_constrained = 3return 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}")