从单周期到多周期：优雅实现量化交易的多时间框架策略
在之前的文章中，我们用一个五连阳策略来演示策略的开发，但此前的版本不支持多周期联立。在量化交易中，单一时间框架往往难以捕捉市场的全貌，专业交易者通常会结合多个时间周期来分析市场，比如用日线判断趋势，用小时线寻找入场点，用分钟线精确执行。我们在使用缠论交易的过程中经常需要多周期联立，经过多次迭代和重构，终于找到了一套优雅的解决方案。
今天，我们就来聊聊如何在代码层面优雅地实现多时间框架策略。
一、为什么需要多时间框架？
想象你在交易一只股票：
- 日线告诉你大趋势是向上的
- 周线显示当前处于上升通道的中部
- 5分钟线提示短期回调到位，可以入场
这就是多时间框架分析的魅力——不同周期提供不同维度的信息，组合起来能大幅提升决策质量。
二、架构设计：分离数据与策略
在我们的框架中，采用了关注点分离的设计理念，将数据加载和策略逻辑完全解耦。
2.1 InputDataModel：灵活的数据模型
核心思想是用一个独立的数据模型类来定义“策略需要什么数据“：
class InputDataModel:
def __init__(self, data_length: Optional[Dict[str, int]] = None):
"""
data_length: 各时间框架需要的K线数量
例如: {'1d': 1500, '1w': 300, '5m': 1000}
"""
if data_length is None:
self.data_length = {'1d': 1500}
else:
self.data_length = data_length
这个设计的妙处在于：
- 声明式配置：只需声明需要什么数据，不用关心如何获取
- 按需加载：只加载策略真正需要的时间框架数据
- 易于扩展：新增时间框架只需修改配置字典
2.2 支持的时间框架
框架支持丰富的时间框架：
- 分钟级：1m, 3m, 5m, 15m, 30m, 60m, 90m, 120m, 180m
- 日线：1d
- 周线：1w
def load_data(self, code: str, today) -> Dict[str, pd.DataFrame]:
"""加载多时间框架数据"""
result = {}
for period, length in self.data_length.items():
if period in ('day', '1d'):
result['1d'] = self._load_day_data(code, today, length)
elif period in ('week', '1w'):
result['1w'] = self._load_week_data(code, today, length)
elif period.endswith('m'):
result[period] = self._load_min_data(code, today, length, period)
return result
返回的是一个字典，键是时间框架标识，值是对应的DataFrame，使用起来非常直观。
三、BaseStrategy：策略基类的设计哲学
3.1 构造函数：注入数据模型
class BaseStrategy(ABC):
def __init__(
self,
strategy_name,
strategy_id,
init_cash: int = 1000000,
lot_size: int = 3000,
input_data_model: Optional[InputDataModel] = None,
):
self.input_data_model = (
input_data_model if input_data_model else InputDataModel()
)
这种依赖注入的方式让策略类更加灵活：
- 可以使用默认数据配置
- 也可以传入自定义的数据模型
- 甚至可以继承InputDataModel实现完全定制的数据加载逻辑
3.2 核心交易流程
def process_stock(self, code, today):
"""处理单只股票的交易逻辑"""
# 1. 加载多时间框架市场数据
market_data = self.input_data_model.load_data(code, parsed_date)
# 2. 获取当前价格（优先使用高频数据）
timeframes = ['1m', '3m', '5m', '15m', '30m', '60m', '1d', '1w']
for tf in timeframes:
if tf in market_data and len(market_data[tf]) > 0:
current_price = market_data[tf]['close'].iloc[-1]
break
# 3. 检查持仓并执行交易逻辑
pos = self.acc.get_position(code)
if pos.volume_long > 0:
if self.should_sell(pos, market_data):
self._execute_sell_order(...)
else:
if self.should_buy(pos, market_data):
self._execute_buy_order(...)
注意这里的巧妙之处：
- market_data是字典：包含所有配置的时间框架数据
- 价格获取有优先级：优先使用高频数据获取最新价格
- 策略逻辑统一接口：should_buy/should_sell接收完整的market_data
3.3 抽象方法：强制子类实现
@abstractmethod
def should_buy(self, pos, market_data: Dict[str, pd.DataFrame]):
pass
@abstractmethod
def should_sell(self, pos, market_data: Dict[str, pd.DataFrame]):
pass
这是模板方法模式的应用，基类定义框架流程，子类填充具体逻辑。
四、实战案例：五连阳策略
让我们看一个具体的策略实现——五连阳策略。
4.1 自定义数据需求
class FiveYangStrategy(BaseStrategy):
def __init__(self, init_cash=1000000, lot_size=3000, input_data_model=None):
if input_data_model is None:
input_data_model = InputDataModel(data_length={'1d': 30})
super().__init__(
strategy_name="五连阳动量跟随策略",
strategy_id="000001",
init_cash=init_cash,
lot_size=lot_size,
input_data_model=input_data_model,
)
这个策略只需要30根日线K线，所以配置很简单。但如果你想改造成多时间框架版本，只需修改配置：
# 多时间框架版本的五连阳策略
input_data_model = InputDataModel(
data_length={
'1d': 30, # 日线判断五连阳形态
'1w': 20, # 周线判断大趋势
'5m': 500 # 5分钟线精确入场
}
)
4.2 使用多时间框架数据
def should_buy(self, pos, market_data):
"""判断是否买入"""
if '1d' not in market_data:
return False
hist_data = market_data['1d']
# 检查五连阳形态
for i in range(5):
row = hist_data.iloc[current_idx - 4 + i]
is_yang = row['close'] > row['open']
if not is_yang:
return False
# 如果配置了周线，可以加入趋势过滤
if '1w' in market_data:
week_data = market_data['1w']
# 检查周线趋势...
return True
market_data字典让你可以灵活地访问任何配置的时间框架数据，实现复杂的多时间框架逻辑。
4.3 动态止损：成交回调的应用
def on_deal_callback(self, code, price, volume, dt, market_data):
hist_data = market_data['1d']
# 方法1：基于近5天高低点
recent_data = hist_data.iloc[-5:]
stop_loss_1 = (recent_data['high'].max() + recent_data['low'].min()) / 2
# 方法2：基于ATR
atr = self.calculate_atr(hist_data, period=20)
stop_loss_2 = price - 2 * atr
# 取较低者作为止损价
final_stop_loss = min(stop_loss_1, stop_loss_2)
self.set_volume_long_stop_loss_price(code, final_stop_loss)
这个回调函数在成交时自动触发，可以根据市场数据动态设置止损，非常实用。
五、设计模式总结
这套框架运用了多个经典设计模式：
5.1 策略模式（Strategy Pattern）
- InputDataModel是可替换的策略
- 不同策略可以有不同的数据加载逻辑
5.2 模板方法模式（Template Method）
- BaseStrategy定义算法骨架
- 子类实现should_buy/should_sell等具体步骤
5.3 依赖注入（Dependency Injection）
- 通过构造函数注入InputDataModel
- 降低耦合，提高可测试性
5.4 关注点分离（Separation of Concerns）
- 数据加载 → InputDataModel
- 账户管理 → QIFI_Account
- 策略逻辑 → BaseStrategy子类
- 技术指标 → 工具方法（calculate_atr等）
六、扩展建议
基于这个框架，你可以轻松实现：
6.1 多时间框架共振策略
input_data_model = InputDataModel({
'1d': 100, # 日线MACD金叉
'1w': 50, # 周线趋势向上
'15m': 500 # 15分钟回调到位
})
6.2 高频交易策略
input_data_model = InputDataModel({
'1m': 1000, # 1分钟主时间框架
'5m': 500, # 5分钟趋势过滤
})
6.3 自定义数据加载
class CustomInputModel(InputDataModel):
def load_data(self, code, today):
data = super().load_data(code, today)
# 添加自定义数据源，如期货、期权等
data['futures'] = self._load_futures_data(code, today)
return data
七、最佳实践
- 按需配置：只加载策略真正需要的时间框架，避免浪费资源
- 优先级处理：获取价格时优先使用高频数据
- 数据验证：使用前检查数据是否存在（
if '1d' in market_data） - 异常处理：数据加载失败时优雅降级
- 回测一致性：确保回测和实盘使用相同的数据模型
结语
多时间框架策略的核心不在于使用多少个时间框架，而在于如何优雅地组织代码，让策略逻辑清晰、易维护、可扩展。通过InputDataModel和BaseStrategy的配合，我们实现了数据与逻辑的完美分离，让策略开发变得更加高效。
如果你正在构建自己的量化交易系统，不妨尝试这套架构——它会让你的策略开发事半功倍。
关于作者：专注于量化交易系统开发，致力于用工程化的方法解决交易问题。
代码仓库：本文涉及的完整代码在缠论小分队的Freshquant项目中。
下期预告：我将使用这套策略模板，将五连阳策略升级为多时间框架版本，加入缠论形态识别，敬请期待。