第 28 课：高频交易与网格策略
⏱ 课时：2 小时
🎯 学习目标：学习特殊策略类型——网格交易和高频交易的原理与实现
课程概述
之前我们学习的都是趋势跟随或均值回归策略。本课将介绍两种特殊类型的策略：
- 网格交易：在震荡市场中获利
- 高频交易：捕捉短期价格波动
警告：
⚠️ 这两种策略都有较高的风险和复杂度
⚠️ 需要更深入的理解和更严格的风险控制
⚠️ 建议先在 Dry-run 中充分测试
第一部分：网格交易策略
1.1 网格交易原理
核心理念：
在价格区间内设置多个买入和卖出网格，通过频繁的低买高卖获利。
基本概念
价格网格示例：
────────────────────────────────────
$32,000 卖出网格 5 → 卖出
$31,500 卖出网格 4 → 卖出
$31,000 卖出网格 3 → 卖出
────────────────────────────────────
$30,500 中心价格
────────────────────────────────────
$30,000 买入网格 1 ← 买入
$29,500 买入网格 2 ← 买入
$29,000 买入网格 3 ← 买入
────────────────────────────────────
工作原理：
1. 在 $30,000 买入
2. 价格涨到 $31,000，卖出（盈利 $1,000）
3. 价格跌回 $30,000，再次买入
4. 重复循环
网格交易的优势和劣势
✅ 优势：
1. 适合震荡市场
- 价格来回波动就能盈利
- 不依赖趋势判断
2. 操作简单
- 规则明确
- 易于自动化
3. 频繁获利
- 每次小幅波动都能捕捉
- 心理满足感强
4. 无需预测方向
- 不需要判断涨跌
- 只要有波动就行
❌ 劣势：
1. 单边市场风险
- 持续上涨：卖飞后踏空
- 持续下跌：接飞刀，越跌越买
2. 资金要求高
- 需要分多次买入
- 预留足够资金
3. 手续费侵蚀
- 交易频繁
- 手续费累计高
4. 资金利用率
- 震荡不强时，收益低
- 资金被大量占用
1.2 简单网格策略实现
创建 user_data/strategies/SimpleGridStrategy.py：
from freqtrade.strategy import IStrategy
from pandas import DataFrame
import talib.abstract as ta
class SimpleGridStrategy(IStrategy):
"""
简单网格交易策略
在固定价格网格上低买高卖
"""
INTERFACE_VERSION = 3
# 关闭 ROI（网格策略自己控制卖出）
minimal_roi = {
"0": 100 # 很高的值，实际上不会触发
}
# 关闭全局止损（网格策略有自己的止损逻辑）
stoploss = -0.99
timeframe = '5m'
startup_candle_count: int = 50
# === 网格参数 ===
# 网格间距（百分比）
grid_spacing = 0.01 # 1%
# 网格数量
num_grids = 5
# 中心价格（动态计算，使用最近 N 根 K 线的均价）
center_price_period = 100
# 是否启用动态中心价格
dynamic_center = True
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
计算中心价格
"""
# 动态中心价格（使用 SMA）
dataframe['center_price'] = ta.SMA(dataframe['close'], timeperiod=self.center_price_period)
# 计算价格与中心价格的偏离
dataframe['price_deviation'] = (
(dataframe['close'] - dataframe['center_price']) /
dataframe['center_price'] * 100
)
# 标记当前处于哪个网格
dataframe['grid_level'] = (
dataframe['price_deviation'] / self.grid_spacing
).round()
return dataframe
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
买入信号：价格低于中心价格的网格
"""
dataframe.loc[
(
# 价格低于中心价格
(dataframe['close'] < dataframe['center_price']) &
# 至少偏离 1 个网格
(dataframe['grid_level'] <= -1) &
# 最多偏离 num_grids 个网格（避免过度下跌时买入）
(dataframe['grid_level'] >= -self.num_grids) &
(dataframe['volume'] > 0)
),
'enter_long'] = 1
return dataframe
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
卖出信号：价格高于买入价格 1 个网格以上
"""
dataframe.loc[
(
# 价格高于中心价格
(dataframe['close'] > dataframe['center_price']) &
# 至少偏离 1 个网格
(dataframe['grid_level'] >= 1) &
(dataframe['volume'] > 0)
),
'exit_long'] = 1
return dataframe
def custom_exit(self, pair: str, trade: 'Trade', current_time: 'datetime',
current_rate: float, current_profit: float, **kwargs) -> bool:
"""
自定义卖出：达到 1 个网格间距就卖出
"""
# 如果盈利达到网格间距，卖出
if current_profit >= self.grid_spacing:
return True
return False
1.3 高级网格策略
增强版本，包含动态网格和风险控制：
创建 user_data/strategies/AdvancedGridStrategy.py：
from freqtrade.strategy import IStrategy
from pandas import DataFrame
import talib.abstract as ta
import numpy as np
class AdvancedGridStrategy(IStrategy):
"""
高级网格策略
- 动态调整网格间距（根据波动率）
- 趋势过滤（避免单边市场）
- 资金管理（限制最大持仓）
"""
INTERFACE_VERSION = 3
minimal_roi = {"0": 100}
stoploss = -0.15 # 设置一个安全止损（防止极端情况）
timeframe = '5m'
startup_candle_count: int = 200
# 最大同时持仓网格数
max_open_grids = 3
# 基础网格间距
base_grid_spacing = 0.01 # 1%
# 网格范围（上下各几个网格）
grid_range = 5
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
计算指标
"""
# 1. 中心价格（EMA）
dataframe['center_price'] = ta.EMA(dataframe['close'], timeperiod=100)
# 2. ATR（用于动态调整网格间距）
dataframe['atr'] = ta.ATR(dataframe, timeperiod=14)
dataframe['atr_pct'] = (dataframe['atr'] / dataframe['close']) * 100
# 3. ADX（判断趋势强度，避免单边市场）
dataframe['adx'] = ta.ADX(dataframe, timeperiod=14)
# 4. 动态网格间距（基于 ATR）
dataframe['grid_spacing'] = dataframe['atr_pct'].clip(
lower=self.base_grid_spacing, # 最小 1%
upper=self.base_grid_spacing * 3 # 最大 3%
)
# 5. 价格偏离百分比
dataframe['price_deviation_pct'] = (
(dataframe['close'] - dataframe['center_price']) /
dataframe['center_price'] * 100
)
# 6. 当前网格层级
dataframe['grid_level'] = (
dataframe['price_deviation_pct'] / dataframe['grid_spacing']
).round()
# 7. 成交量
dataframe['volume_mean'] = dataframe['volume'].rolling(window=20).mean()
return dataframe
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
买入信号：
1. 价格低于中心价格
2. 无强趋势（ADX < 25，避免下跌趋势）
3. 在网格范围内
"""
dataframe.loc[
(
# 条件 1：价格低于中心价格
(dataframe['close'] < dataframe['center_price']) &
# 条件 2：趋势不要太强（避免单边市场）
(dataframe['adx'] < 30) &
# 条件 3：在买入网格范围内
(dataframe['grid_level'] >= -self.grid_range) &
(dataframe['grid_level'] <= -1) &
# 条件 4：成交量正常
(dataframe['volume'] > dataframe['volume_mean'] * 0.5) &
(dataframe['volume'] > 0)
),
'enter_long'] = 1
return dataframe
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
卖出信号：价格回到中心价格以上
"""
dataframe.loc[
(
# 价格高于中心价格
(dataframe['close'] > dataframe['center_price']) &
# 至少 1 个网格
(dataframe['grid_level'] >= 1) &
(dataframe['volume'] > 0)
),
'exit_long'] = 1
return dataframe
def custom_stake_amount(self, pair: str, current_time: 'datetime',
current_rate: float, proposed_stake: float,
min_stake: float, max_stake: float, **kwargs) -> float:
"""
自定义仓位：根据网格层级调整
"""
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
if len(dataframe) == 0:
return proposed_stake
last_candle = dataframe.iloc[-1].squeeze()
grid_level = last_candle['grid_level']
# 越低的网格，仓位越大（但要控制总量）
# 例如：
# 网格 -1: 100% 仓位
# 网格 -2: 120% 仓位
# 网格 -3: 150% 仓位
multiplier = 1.0 + (abs(grid_level) - 1) * 0.2
multiplier = min(multiplier, 1.5) # 最大 150%
return proposed_stake * multiplier
def custom_exit(self, pair: str, trade: 'Trade', current_time: 'datetime',
current_rate: float, current_profit: float, **kwargs) -> bool:
"""
自定义卖出：达到 1 个网格间距即卖出
"""
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
if len(dataframe) == 0:
return False
last_candle = dataframe.iloc[-1].squeeze()
grid_spacing = last_candle['grid_spacing'] / 100 # 转为小数
# 盈利超过当前网格间距，卖出
if current_profit >= grid_spacing:
return True
return False
1.4 网格策略测试
# 1. 下载数据
freqtrade download-data -c config.json --days 90 --timeframe 5m
# 2. 回测简单网格策略
freqtrade backtesting \
-c config.json \
--strategy SimpleGridStrategy \
--timerange 20230101-20230331
# 3. 回测高级网格策略
freqtrade backtesting \
-c config.json \
--strategy AdvancedGridStrategy \
--timerange 20230101-20230331
注意事项：
1. 手续费设置要准确
- 网格策略交易频繁
- 手续费影响很大
- "fee": 0.001 # 0.1%
2. 选择合适的交易对
- 流动性好（BTC/USDT, ETH/USDT）
- 波动适中（不要太剧烈）
- 避免长期单边趋势的币种
3. 资金管理
- 预留足够资金应对下跌
- 不要满仓
- 建议用总资金的 50-70%
第二部分：高频交易策略
2.1 高频交易原理
定义：
高频交易（High-Frequency Trading, HFT）是利用极短时间内的价格波动获利的策略。
特点
时间尺度：
- 传统策略：持仓几小时到几天
- 高频策略：持仓几秒到几分钟
交易频率：
- 传统策略：每天 3-5 笔
- 高频策略：每天 50-500 笔
盈利模式：
- 传统策略：捕捉趋势，单笔盈利大
- 高频策略：捕捉微小波动，单笔盈利小但频繁
高频交易的挑战
1. 技术要求
✗ 需要极低延迟
✗ 需要稳定的网络
✗ 需要高性能服务器
✗ Freqtrade 不是专业 HFT 系统
2. 成本问题
✗ 手续费累计高
✗ 滑点影响大
✗ API 限流
3. 策略难度
✗ 信号稳定性差
✗ 假信号多
✗ 难以回测验证
结论：
⚠️ Freqtrade 不适合真正的高频交易（毫秒级）
✅ 但可以做中高频交易（分钟级，每天 20-50 笔）
2.2 中高频策略示例
创建 user_data/strategies/ScalpingStrategy.py：
from freqtrade.strategy import IStrategy
from pandas import DataFrame
import talib.abstract as ta
import freqtrade.vendor.qtpylib.indicators as qtpylib
class ScalpingStrategy(IStrategy):
"""
剥头皮策略（Scalping）
捕捉 1 分钟图上的快速波动
目标：单笔 0.3-1% 的小利润，快进快出
"""
INTERFACE_VERSION = 3
# 快速止盈
minimal_roi = {
"0": 0.01, # 1% 立即止盈
"5": 0.008, # 5 分钟后 0.8%
"10": 0.005, # 10 分钟后 0.5%
"15": 0.003 # 15 分钟后 0.3%
}
# 紧止损
stoploss = -0.015 # -1.5%
# 使用 1 分钟图
timeframe = '1m'
startup_candle_count: int = 30
# 订单类型（市价单，快速成交）
order_types = {
'entry': 'market', # 市价买入
'exit': 'market', # 市价卖出
'stoploss': 'market',
'stoploss_on_exchange': True
}
# 不允许持仓超过 30 分钟（强制平仓）
max_holding_minutes = 30
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
使用快速指标
"""
# 极短期 EMA
dataframe['ema_5'] = ta.EMA(dataframe, timeperiod=5)
dataframe['ema_10'] = ta.EMA(dataframe, timeperiod=10)
# RSI
dataframe['rsi'] = ta.RSI(dataframe, timeperiod=7) # 更短周期
# 布林带（窄周期）
bollinger = qtpylib.bollinger_bands(dataframe['close'], window=10, stds=2)
dataframe['bb_lower'] = bollinger['lower']
dataframe['bb_middle'] = bollinger['mid']
dataframe['bb_upper'] = bollinger['upper']
# 成交量突增
dataframe['volume_mean'] = dataframe['volume'].rolling(window=10).mean()
dataframe['volume_surge'] = dataframe['volume'] / dataframe['volume_mean']
# 价格动量
dataframe['price_momentum'] = (
(dataframe['close'] - dataframe['close'].shift(3)) /
dataframe['close'].shift(3) * 100
)
return dataframe
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
买入信号：快速反弹
"""
dataframe.loc[
(
# 条件 1：快速 EMA 金叉
(qtpylib.crossed_above(dataframe['ema_5'], dataframe['ema_10'])) &
# 条件 2：RSI 从超卖恢复
(dataframe['rsi'] > 30) &
(dataframe['rsi'] < 60) &
(dataframe['rsi'] > dataframe['rsi'].shift(1)) & # RSI 上升
# 条件 3：价格在布林带下半部（有上涨空间）
(dataframe['close'] < dataframe['bb_middle']) &
# 条件 4：成交量突增（确认动能）
(dataframe['volume_surge'] > 1.5) &
# 条件 5：短期动量向上
(dataframe['price_momentum'] > 0) &
(dataframe['volume'] > 0)
),
'enter_long'] = 1
return dataframe
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
卖出信号：快速获利或反转
"""
dataframe.loc[
(
(
# 条件 1：EMA 死叉
(qtpylib.crossed_below(dataframe['ema_5'], dataframe['ema_10'])) |
# 或条件 2：RSI 超买
(dataframe['rsi'] > 70) |
# 或条件 3：价格触及布林带上轨
(dataframe['close'] > dataframe['bb_upper'])
) &
(dataframe['volume'] > 0)
),
'exit_long'] = 1
return dataframe
def custom_exit(self, pair: str, trade: 'Trade', current_time: 'datetime',
current_rate: float, current_profit: float, **kwargs) -> bool:
"""
强制平仓：超过最大持仓时间
"""
# 计算持仓时间（分钟）
trade_duration = (current_time - trade.open_date_utc).total_seconds() / 60
# 超过 30 分钟，无论盈亏都平仓
if trade_duration > self.max_holding_minutes:
return True
# 快速止盈：0.5% 就走
if current_profit >= 0.005:
return True
return False
2.3 高频策略优化技巧
技巧 1：使用市价单
# 高频策略必须使用市价单
order_types = {
'entry': 'market',
'exit': 'market',
'stoploss': 'market',
'stoploss_on_exchange': True
}
# 原因：
# 1. 限价单可能不成交，错过机会
# 2. 高频追求速度，不是最优价格
# 3. 1 秒的延迟可能导致信号失效
技巧 2：严格的持仓时间限制
def custom_exit(self, pair: str, trade: 'Trade', current_time: 'datetime',
current_rate: float, current_profit: float, **kwargs) -> bool:
"""
强制时间止损
"""
trade_duration = (current_time - trade.open_date_utc).total_seconds() / 60
# 超过 30 分钟强制平仓
if trade_duration > 30:
return True
return False
技巧 3：快速止盈止损
# 止盈目标小
minimal_roi = {
"0": 0.01, # 1%
"10": 0.005 # 10 分钟后 0.5%
}
# 止损紧
stoploss = -0.015 # -1.5%
# 盈亏比约 1:1.5（可接受，因为频率高）
技巧 4：成交量过滤
# 必须有成交量突增
(dataframe['volume_surge'] > 1.5) &
# 原因：
# 高频需要流动性支持
# 成交量小会导致滑点大
2.4 高频策略风险管理
风险 1：手续费侵蚀
- 问题：交易 100 次，手续费 0.1% × 2 × 100 = 20%
- 对策：
✓ 使用 VIP 等级降低手续费
✓ 使用交易所代币抵扣（如 BNB）
✓ 计算盈亏时扣除手续费
✓ 确保平均盈利 > 手续费
风险 2：滑点
- 问题：市价单成交价差于预期
- 对策：
✓ 只交易高流动性币种（BTC/ETH）
✓ 避免大额订单
✓ 避免市场剧烈波动时交易
风险 3：过度交易
- 问题：一天交易 200 次，多数是假信号
- 对策：
✓ 设置每日最大交易次数（如 50 次）
✓ 提高入场条件
✓ 增加成交量过滤
风险 4：系统故障
- 问题：网络延迟、API 限流
- 对策：
✓ 使用稳定的服务器
✓ 备用网络连接
✓ 监控 API 调用频率
✓ 设置最大持仓保护
第三部分：策略对比和选择
3.1 三种策略对比
| 特性 | 趋势跟随 | 网格交易 | 高频交易 |
|---|---|---|---|
| 时间框架 | 5m-1h | 5m-15m | 1m-5m |
| 持仓时间 | 数小时-数天 | 数小时 | 数分钟 |
| 交易频率 | 低（3-5/天） | 中（10-20/天） | 高（50+/天） |
| 单笔盈利 | 2-5% | 1-2% | 0.3-1% |
| 胜率 | 50-60% | 60-70% | 55-65% |
| 适合市场 | 趋势市 | 震荡市 | 高波动 |
| 资金要求 | 中 | 高 | 中 |
| 技术要求 | 低 | 中 | 高 |
| 手续费影响 | 小 | 中 | 大 |
| 风险等级 | 中 | 中高 | 高 |
3.2 如何选择
选择趋势跟随，如果你：
✓ 是新手
✓ 每天只能监控几次
✓ 不想频繁交易
✓ 追求稳定增长
✓ 资金有限
选择网格交易，如果你：
✓ 有一定经验
✓ 市场处于震荡期
✓ 有足够资金分批买入
✓ 能接受资金占用
✓ 追求频繁小胜
选择高频交易，如果你：
✓ 有丰富经验
✓ 有稳定的技术环境
✓ 能接受高风险
✓ 手续费率很低
✓ 追求刺激和挑战
⚠️ 不推荐新手使用网格或高频策略
3.3 组合策略
可以同时运行多个策略：
配置示例：
账户 1（70% 资金）：
- 趋势跟随策略
- 5m 时间框架
- 稳健为主
账户 2（20% 资金）：
- 网格策略
- 5m 时间框架
- 震荡市场使用
账户 3（10% 资金）：
- 剥头皮策略
- 1m 时间框架
- 实验性质
优势：
✓ 分散风险
✓ 适应不同市场
✓ 平滑收益曲线
📝 实践任务
任务 1：回测网格策略
- 复制
SimpleGridStrategy到你的环境 - 回测 3 个月数据
- 对比不同网格间距（0.5%, 1%, 2%）的效果
- 记录：
- 交易次数
- 胜率
- 手续费占比
- 最大回撤
任务 2：优化网格参数
修改 AdvancedGridStrategy：
- 调整 ADX 阈值（20, 25, 30）
- 调整网格范围（3, 5, 7）
- 测试不同的动态网格计算方法
- 找出最优组合
任务 3：高频策略实验（谨慎）
- 在 Dry-run 中测试
ScalpingStrategy - 运行 3-7 天
- 统计：
- 每日交易次数
- 平均持仓时间
- 平均单笔盈亏
- 手续费总计
- 判断：是否适合实盘？
任务 4：手续费敏感度分析
对同一策略，测试不同手续费率的影响：
# 0.1% 手续费
"fee": 0.001
# 0.05% 手续费（VIP 优惠）
"fee": 0.0005
# 0.2% 手续费（最差情况）
"fee": 0.002
对比回测结果，了解手续费的影响。
📌 核心要点
网格交易要点
1. 适合震荡市
- ADX < 25
- 价格在区间内波动
- 避免单边趋势
2. 资金管理
- 分批买入
- 预留资金应对下跌
- 不要满仓
3. 参数设置
- 网格间距 1-2%
- 网格数量 5-10 个
- 动态调整更好
4. 风险控制
- 设置最大持仓数
- 设置止损保护
- 监控单边趋势
高频交易要点
1. 技术要求
- 低延迟网络
- 稳定服务器
- 使用市价单
2. 成本控制
- 降低手续费
- 计算滑点
- 只交易高流动性币种
3. 快进快出
- 持仓时间 < 30 分钟
- 止盈 0.5-1%
- 止损 1-1.5%
4. 严格纪律
- 设置交易次数上限
- 避免过度交易
- 及时止损
不适合的情况
❌ 不要使用网格，如果：
- 市场处于强趋势
- 资金不足
- 无法接受资金占用
❌ 不要使用高频，如果：
- 是新手
- 网络不稳定
- 手续费率高（> 0.1%）
- 资金少（< $5000）
🎯 总结
本课介绍了两种特殊策略：
- 网格交易：适合震荡市场，通过频繁的低买高卖获利
- 高频交易：捕捉短期波动，快进快出
重要提醒：
⚠️ 这两种策略都比趋势跟随策略风险更高
⚠️ 需要更多的经验和更严格的风险控制
⚠️ 建议先掌握基础策略，再尝试这些高级策略
⚠️ 务必在 Dry-run 中充分测试
下节课我们将学习如何使用机器学习优化策略。
🎓 学习建议：
- 循序渐进：先掌握趋势策略，再尝试网格和高频
- 充分测试：在 Dry-run 中测试至少 2 周
- 小资金起步：即使测试成功，也要小资金开始
- 严格风控：这两种策略需要更严格的风险控制
- 持续监控：需要更频繁地监控和调整
记住：简单有效 > 复杂华丽。不要为了追求高频而高频，适合自己的才是最好的。
Top comments (0)