Python量化交易：网格策略实现与TqSdk实战

1. 网格交易策略基础解析

网格交易策略是我在量化交易领域最常使用的策略之一，它的核心思想简单却有效。想象一下在渔场撒网捕鱼：我们把网（网格）均匀地撒在价格波动的区间内，无论价格往哪个方向波动，都能"捕获"到交易机会。

1.1 网格策略的核心机制

网格策略的工作原理可以用一个简单的例子说明：假设螺纹钢期货当前价格是3600元，我们设置网格间距为20元，上下各10格。当价格下跌到3580元时买入1手，继续跌到3560元再买入1手；反之当价格上涨到3620元时卖出1手，涨到3640元再卖出1手。

这种策略的关键参数包括：

• 基准价格：网格的中心点（如3600元）
• 网格间距：相邻买卖点的价格差（如20元）
• 网格数量：基准价格上下各设置多少格（如10格）
• 每格手数：每次触发网格时交易的手数（如1手）

注意：网格间距的设置需要结合品种的波动特性和交易成本。间距太小会导致频繁交易被手续费侵蚀利润，太大则可能错过交易机会。

1.2 适用场景与局限性

从我多年的实盘经验来看，网格策略最适合以下市场环境：

1. 震荡行情：价格在一定区间内来回波动时效果最佳
2. 高波动性品种：波动率越高，网格触发的机会越多
3. 流动性好的合约：确保能够及时成交

但必须警惕的是，网格策略在单边趋势行情中会面临较大风险。比如当价格持续下跌时，策略会不断买入，导致持仓亏损不断扩大。因此在实际应用中必须配合严格的风控措施。

2. TqSdk实现详解

天勤量化(TqSdk)是目前国内期货量化交易中较为流行的Python SDK，它的API设计简洁明了，特别适合快速实现网格策略。

2.1 基础版本实现

让我们先看一个最基础的网格策略实现：

from tqsdk import TqApi, TqAuth, TqBacktest from datetime import date # 策略参数配置 SYMBOL = "SHFE.rb2505" # 螺纹钢2505合约 BASE_PRICE = 3600 # 基准价格 GRID_SIZE = 20 # 网格间距(元) GRID_COUNT = 10 # 单边网格数量 LOTS_PER_GRID = 1 # 每格交易手数 # 初始化API连接 api = TqApi( backtest=TqBacktest(start_dt=date(2025,1,1), end_dt=date(2025,6,30)), auth=TqAuth("your_account", "your_password") ) # 获取行情和持仓数据 quote = api.get_quote(SYMBOL) position = api.get_position(SYMBOL) # 生成网格价格列表 grid_prices = [BASE_PRICE + i*GRID_SIZE for i in range(-GRID_COUNT, GRID_COUNT+1)] triggered_grids = set() # 记录已触发网格 while True: api.wait_update() if api.is_changing(quote, "last_price"): current_price = quote.last_price for i, grid_price in enumerate(grid_prices): grid_id = i - GRID_COUNT # 网格编号(-10到+10) if grid_id in triggered_grids: continue # 跳过已触发网格 # 买入信号：价格下穿网格线(网格编号为负) if grid_id < 0 and current_price <= grid_price: api.insert_order(SYMBOL, "BUY", "OPEN", LOTS_PER_GRID, grid_price) triggered_grids.add(grid_id) print(f"触发买入: 网格{grid_id}@{grid_price}") # 卖出信号：价格上穿网格线(网格编号为正)且有持仓 if grid_id > 0 and current_price >= grid_price and position.pos_long >= LOTS_PER_GRID: api.insert_order(SYMBOL, "SELL", "CLOSE", LOTS_PER_GRID, grid_price) triggered_grids.add(grid_id) print(f"触发卖出: 网格{grid_id}@{grid_price}")

这个基础版本已经包含了网格策略的核心逻辑，但实际使用中还需要考虑更多细节。

2.2 面向对象的改进版本

为了提高代码的可维护性和扩展性，我们可以将策略封装成类：

class GridStrategy: """网格策略封装类""" def __init__(self, api, symbol, base_price, grid_size, grid_count, lots): self.api = api self.symbol = symbol self.base_price = base_price self.grid_size = grid_size self.grid_count = grid_count self.lots = lots # 初始化市场数据 self.quote = api.get_quote(symbol) self.position = api.get_position(symbol) # 初始化网格状态 self.grids = {} for i in range(-grid_count, grid_count + 1): price = base_price + i * grid_size self.grids[i] = { 'price': price, 'triggered': False, 'direction': 'BUY' if i < 0 else 'SELL' } def on_tick(self): """处理行情更新""" current_price = self.quote.last_price for grid_id, grid in self.grids.items(): if grid['triggered']: continue # 检查买入网格 if grid['direction'] == 'BUY' and current_price <= grid['price']: self._execute_trade(grid_id, grid['price'], "BUY", "OPEN") # 检查卖出网格 elif grid['direction'] == 'SELL' and current_price >= grid['price']: if self.position.pos_long >= self.lots: # 确保有持仓可平 self._execute_trade(grid_id, grid['price'], "SELL", "CLOSE") def _execute_trade(self, grid_id, price, direction, offset): """执行交易""" self.api.insert_order( self.symbol, direction, offset, self.lots, limit_price=price ) self.grids[grid_id]['triggered'] = True print(f"{'买入' if direction=='BUY' else '卖出'}执行: 网格{grid_id}@{price}") # 使用示例 api = TqApi(auth=TqAuth("your_account", "your_password")) strategy = GridStrategy(api, "SHFE.rb2505", 3600, 20, 10, 1) while True: api.wait_update() if api.is_changing(strategy.quote): strategy.on_tick()

这个改进版本有以下几个优点：

1. 将策略逻辑封装在类中，提高了代码的模块化程度
2. 网格状态管理更加清晰
3. 更容易添加新功能（如后面会讲到的风控模块）

3. VnPy实现对比

VnPy是另一个广泛使用的量化交易框架，它采用基于事件驱动的策略模板方式。虽然学习曲线比TqSdk陡峭，但更适合构建复杂的交易系统。

3.1 VnPy策略模板实现

from vnpy_ctastrategy import CtaTemplate from vnpy.trader.constant import Interval class GridStrategyVnPy(CtaTemplate): """VnPy网格策略实现""" # 策略参数 base_price = 3600.0 grid_size = 20.0 grid_count = 10 lots_per_grid = 1 # 策略变量 triggered_grids = [] parameters = ["base_price", "grid_size", "grid_count", "lots_per_grid"] variables = ["triggered_grids"] def __init__(self, cta_engine, strategy_name, vt_symbol, setting): super().__init__(cta_engine, strategy_name, vt_symbol, setting) self._init_grids() def _init_grids(self): """初始化网格价格""" self.buy_grids = [ self.base_price - i * self.grid_size for i in range(1, self.grid_count + 1) ] self.sell_grids = [ self.base_price + i * self.grid_size for i in range(1, self.grid_count + 1) ] def on_tick(self, tick): """Tick更新处理""" current_price = tick.last_price # 处理买入网格 for price in self.buy_grids: if price not in self.triggered_grids and current_price <= price: self.buy(price, self.lots_per_grid) self.triggered_grids.append(price) self.write_log(f"触发买入: {price}") # 处理卖出网格 for price in self.sell_grids: if (price not in self.triggered_grids and current_price >= price and self.pos >= self.lots_per_grid): self.sell(price, self.lots_per_grid) self.triggered_grids.append(price) self.write_log(f"触发卖出: {price}") def on_bar(self, bar): """K线更新处理""" pass # 网格策略通常不需要处理K线

3.2 TqSdk与VnPy的深度对比

从实际开发体验来看，这两个框架各有特点：

选择建议：

• 如果你是量化交易新手，或者需要快速实现简单策略，TqSdk是更好的选择
• 如果你需要构建复杂的多策略系统，或者对交易引擎有更高要求，VnPy更适合

4. 高级优化技巧

经过多年的实盘打磨，我总结出几个提升网格策略表现的关键优化点。

4.1 动态调整网格参数

静态网格在行情变化时表现不佳，我们可以让网格参数根据市场状况动态调整。

动态基准价格

def update_base_price(klines, lookback=100): """使用近期均价作为动态基准价格""" return klines['close'].iloc[-lookback:].mean() # 在策略中定期更新 if api.is_changing(klines): BASE_PRICE = update_base_price(klines) # 需要重新初始化网格

自适应网格间距

def calculate_grid_size(klines, atr_period=14, multiplier=0.5): """基于ATR计算动态网格间距""" high = klines['high'] low = klines['low'] close = klines['close'] tr = pd.concat([ high - low, abs(high - close.shift(1)), abs(low - close.shift(1)) ], axis=1).max(axis=1) atr = tr.rolling(atr_period).mean().iloc[-1] return atr * multiplier # ATR的0.5倍作为间距

4.2 增强风控模块

没有风控的网格策略就像没有刹车的汽车。这是我常用的风控方案：

class GridStrategyWithRisk(GridStrategy): """带风控的网格策略""" def __init__(self, *args, max_position=10, max_loss=5000, **kwargs): super().__init__(*args, **kwargs) self.max_position = max_position # 最大持仓限制 self.max_loss = max_loss # 最大允许亏损 def on_tick(self): # 检查持仓限制 if self.position.pos_long >= self.max_position: print("达到最大持仓限制，暂停买入") return # 检查资金风险 account = self.api.get_account() if account.float_profit < -self.max_loss: print(f"达到最大亏损限制{self.max_loss}，执行平仓") self._close_all() return # 执行正常网格逻辑 super().on_tick() def _close_all(self): """平掉所有持仓""" if self.position.pos_long > 0: self.api.insert_order( self.symbol, "SELL", "CLOSE", self.position.pos_long )

4.3 多品种网格策略

对于资金量较大的账户，可以考虑同时运行多个品种的网格策略，分散风险。

class MultiSymbolGrid: """多品种网格策略管理器""" def __init__(self, api, symbol_configs): self.api = api self.strategies = {} for symbol, config in symbol_configs.items(): self.strategies[symbol] = GridStrategy( api, symbol, config['base_price'], config['grid_size'], config['grid_count'], config['lots'] ) def run(self): while True: self.api.wait_update() for symbol, strategy in self.strategies.items(): if self.api.is_changing(strategy.quote): strategy.on_tick() # 配置示例 symbol_configs = { "SHFE.rb2505": { "base_price": 3600, "grid_size": 20, "grid_count": 10, "lots": 1 }, "DCE.m2505": { "base_price": 3000, "grid_size": 15, "grid_count": 8, "lots": 1 } } api = TqApi(auth=TqAuth("your_account", "your_password")) multi_grid = MultiSymbolGrid(api, symbol_configs) multi_grid.run()

5. 实盘经验分享

在过去的实盘交易中，我总结了以下宝贵经验：

5.1 品种选择要点

不是所有品种都适合网格交易。我通常考虑以下因素：

1. 波动性：选择日均波动1%-3%的品种
2. 流动性：确保买卖价差小，成交量充足
3. 手续费：选择手续费率较低的品种
4. 保证金：根据账户资金选择适当保证金要求的品种

表现较好的品种通常包括：

• 黑色系：螺纹钢、热卷
• 农产品：豆粕、菜粕
• 化工品：PTA、甲醇

5.2 参数优化方法

网格策略参数对绩效影响很大，我的优化流程是：

1. 历史回测：至少覆盖3种不同市场环境
2. 参数扫描：对网格间距和数量进行敏感性分析
3. 蒙特卡洛测试：验证策略的稳健性
4. 模拟盘验证：至少1个月的模拟交易
5. 实盘小资金测试：确认实际执行效果

5.3 常见问题处理

在实际运行中常遇到以下问题及解决方法：

1. 滑点问题：

   • 原因：行情快速波动时限价单无法成交
   • 解决：适当放宽成交条件，或使用市价单

2. 网格耗尽：

   • 现象：价格超出网格范围后策略失效
   • 解决：设置动态网格调整机制

3. 资金不足：

   • 现象：保证金不足导致无法开仓
   • 解决：严格控制网格数量和每格手数

4. 极端行情：

   • 风险：单边行情造成重大亏损
   • 解决：设置硬止损，或配合趋势过滤

6. 绩效评估与改进

一个完整的网格策略还需要科学的绩效评估体系。

6.1 关键绩效指标

def analyze_performance(trades): """分析策略绩效""" # 基础统计 total_trades = len(trades) win_trades = len([t for t in trades if t['pnl'] > 0]) win_rate = win_trades / total_trades # 盈亏统计 total_pnl = sum(t['pnl'] for t in trades) avg_win = sum(t['pnl'] for t in trades if t['pnl'] > 0) / win_trades avg_loss = sum(t['pnl'] for t in trades if t['pnl'] <= 0) / (total_trades - win_trades) # 风险指标 max_drawdown = calculate_max_drawdown(trades) sharpe_ratio = calculate_sharpe(trades) print(f"总交易次数: {total_trades}") print(f"胜率: {win_rate:.1%}") print(f"总盈亏: {total_pnl:.0f}") print(f"平均盈利: {avg_win:.0f} | 平均亏损: {avg_loss:.0f}") print(f"最大回撤: {max_drawdown:.0f}") print(f"夏普比率: {sharpe_ratio:.2f}")

6.2 典型绩效特征

从历史回测和实盘来看，网格策略通常呈现以下特征：

• 胜率高：通常在60%-80%之间
• 盈亏比低：平均盈利通常小于平均亏损
• 回撤可控：好的参数设置下回撤通常在5%-15%
• 资金曲线平稳：在震荡市中呈现稳定上升趋势

6.3 持续优化方向

为了保持策略竞争力，我通常会从以下几个方向进行持续优化：

1. 动态参数调整：根据市场波动性自动调整网格参数
2. 品种轮动：选择当前最适合网格交易的品种
3. 混合策略：结合趋势过滤或其他策略信号
4. 执行优化：改进订单执行算法，减少滑点
5. 资金管理：根据账户规模动态调整仓位

网格策略虽然原理简单，但要真正用好需要大量的实践和优化。建议新手从小资金开始，逐步积累经验后再加大投入。记住，在量化交易中，稳健性永远比高收益更重要。