量化打板策略实战源码深度分享,解锁高效交易策略秘诀
量化打板策略是一种在量化交易中常用的策略，其核心思想是捕捉股票价格在短时间内快速上涨的机会。以下是一个简单的量化打板策略的Python源码示例，使用的是pandas和ta-lib库。请注意，这只是一个示例，实际应用中需要根据具体情况进行调整。
```python
import pandas as pd
import numpy as np
import talib
from datetime import datetime, timedelta
import backtrader as bt
# 创建策略
class MyStrategy(bt.Strategy):
params = (
('fast_length', 10),
('slow_length', 20),
('length', 50),
('threshold', 0.02),
)
def __init__(self):
self.rsi = talib.RSI(self.data.close, self.params.length)
self.fast_ma = talib.SMA(self.data.close, self.params.fast_length)
self.slow_ma = talib.SMA(self.data.close, self.params.slow_length)
def next(self):
# 检查RSI是否超过阈值
if self.rsi[-1] > self.params.threshold:
# 检查短期均线是否穿过长期均线
if self.fast_ma[-1] > self.slow_ma[-1]:
self.buy()
# 检查是否需要平仓
if self.getposition(self.data).size > 0:
# 检查RSI
相关内容：
这个策略初版是来自网络，然后根据我的理解进行了改造。由于本人不擅长打板。所以效果一般。有兴趣的可以在此基础上改造，加入自己的理解。
# coding=utf-8
from __future__ import print_function, absolute_import
from gm.api import *
import numpy as np
import datetime
import snowater.common_func_new as cf
"""
策略逻辑：打板策略
盘前筛选出上一交易日首次涨停的股票，盘中监控股票价格达到"标准"则买入，每天买入最多5只票；次日早盘判断是否涨停，卖出非涨停股
标准：tick级别，卖5价格大于涨停价（即卖5价格为0）
改进想法，大盘下跌，买入时机为尾盘，大盘上涨买入时机为早盘。
"""
def init(context):
context.max_holding_num = 5# 最大持股数量
context.frequency = 'tick'
context.first_limit_up_stock = # 每天首板股票池
context.target_stocks = {}# 目标变量,日期：代码
# 盘前定时执行任务
schedule(schedule_func=before_market, date_rule='1d', time_rule='09:20:00')
# 尾盘定时执行任务
schedule(schedule_func=end_market, date_rule='1d', time_rule='10:00:00')
# 盘后定时执行任务
schedule(schedule_func=after_market, date_rule='1d', time_rule='15:20:00')
def before_market(context):
"""筛选上一个交易日首板股票"""
# 获取所有A股代码
all_stocks = cf.get_instruments_all_stocks(context)
# 历史交易日
last_date = get_previous_trading_date(exchange='SZSE', date=context.now)
# 计算首板股票
context.first_limit_up_stock = cal_first_limit_up_stocks(context,all_stocks,last_date)
#根据多个财务指标进行筛选,PEG比率在0到0.5之间、债务总资产比小于50%、每股净现金大于0、市盈率在10到100之间、市净率在0到5之间、换手率在3到8之间
context.first_limit_up_stock = cf.filter_stocks_by_financial_indicators(context.first_limit_up_stock, last_date)
if context.now.date() not in context.target_stocks.keys():
context.target_stocks =
print(f"{context.now}首跌板股票:{context.first_limit_up_stock}")
# 检查是否满仓
if len(context.account().positions()) >= context.max_holding_num:
return # 如果满仓，直接返回，不执行后续买入逻辑
# 订阅数据
subscribe(context.first_limit_up_stock, frequency=context.frequency, count=1, unsubscribe_previous=False)
def on_tick(context, tick):
"""买入"""
# 获取当前时间
now = context.now
if ((now.hour == 9 and now.minute < 30) or
((now.hour == 11 and now.minute >= 30) and (now.hour < 13)) or
now.hour >= 15):
return
symbol = tick# 代码
if symbol in context.first_limit_up_stock:
now = tick# 当前时间
sell_1_price = tick# 卖1挂单价
sell_5_price = tick# 卖5挂单价
# 计算目标下单量
total_asset = context.account().cash
trade_volume = int(np.floor(total_asset/context.max_holding_num/2/sell_1_price/100)*100) if sell_1_price>0 else 0
# 买入条件一、开盘后才买
cond1 = now.hour==9 and now.minute>=30 or now.hour>11
# 买入条件二、买5价格大于涨停价（即卖5价格为0）
cond2 = sell_5_price==0
# 买入条件三、买一挂单价>0(即还未涨停)
cond3 = sell_1_price>0
# 买入条件四、当前未持仓
cond4 = symbol not in context.target_stocks
# 买入条件五、每天最多买入N只股票,总持仓数量为2N
cond5 = len(context.target_stocks)<context.max_holding_num
if cond1 and cond2 and cond3 and cond4 and cond5:
# 买入
context.target_stocks.append(symbol)
order_volume(symbol=symbol, volume=trade_volume, side=OrderSide_Buy, order_type=OrderType_Limit, position_effect=PositionEffect_Open, price=sell_1_price)
if len(context.target_stocks)==context.max_holding_num:
unsubscribe(symbols='*', frequency=context.frequency)
def end_market(context):
"""卖出"""
Account_positions = context.account().positions()
holding_stock = for position in Account_positions]
print(f"{context.now.date()} 持仓股票：{holding_stock}")
# 获取昨日收盘价
last_date = get_previous_trading_date(exchange='SZSE', date=context.now)
for position in Account_positions:
symbol = position
last_close = history(symbol=symbol, frequency='1d', start_time=last_date, end_time=last_date,
fields='symbol,close', adjust=ADJUST_NONE, df=True)
if last_close.empty:
continue
# 获取股票的实时行情数据
data = context.data(symbol=symbol, frequency=context.frequency, count=1)
# 风险控制
cf.apply_stop_loss_strategy(context, current(symbols=symbol))
current_price = data # 当前价格
# 计算涨跌幅
yesterday_close = last_close.close.values
price_change = (current_price - yesterday_close) / yesterday_close * 100
cond1 = position-position>0 if context.mode==MODE_BACKTEST else position>0 # 有可交易持仓
cond4 = price_change < -2 # 涨幅小于-2%则卖出
if cond1 and cond4:
order_volume(symbol=symbol, volume=position, side=OrderSide_Sell,
order_type=OrderType_Market,position_effect=PositionEffect_Close,
price=0)
def after_market(context):
# 更新目标变量
last_date = datetime.datetime.strptime(get_previous_trading_date(exchange='SZSE', date=context.now), '%Y-%m-%d').date()
if last_date in context.target_stocks.keys():
if len(context.target_stocks)==0:
context.target_stocks.pop(last_date)
else:
next_date = datetime.datetime.strptime(get_next_trading_date(exchange='SZSE', date=context.now), '%Y-%m-%d').date()
context.target_stocks = context.target_stocks
context.target_stocks.pop(last_date)
def on_order_status(context, order):
# 标的代码
symbol = order
# 委托价格
price = order
# 委托数量
volume = order
# 目标仓位
target_percent = order
# 查看下单后的委托状态，等于3代表委托全部成交
status = order
# 买卖方向，1为买入，2为卖出
side = order
# 开平仓类型，1为开仓，2为平仓
effect = order
# 委托类型，1为限价委托，2为市价委托
order_type = order
if status == 3:
if effect == 1:
if side == 1:
side_effect = '开多仓'
elif side == 2:
side_effect = '开空仓'
else:
last_date = datetime.datetime.strptime(get_previous_trading_date(exchange='SZSE', date=context.now), '%Y-%m-%d').date()
try:
context.target_stocks.remove(symbol)
except:
context.target_stocks.remove(symbol)
if side == 1:
side_effect = '平空仓'
elif side == 2:
side_effect = '平多仓'
order_type_word = '限价' if order_type==1 else '市价'
print('{}:标的：{}，操作：以{}{}，委托价格：{}，委托数量：{}'.format(datetime.datetime.strftime(context.now,"%Y-%m-%d %H:%M:%S"),symbol,order_type_word,side_effect,price,volume))
def cal_first_limit_up_stocks(context,symbols,date):
"""计算首跌板股票"""
lower_limit = get_history_instruments(symbols, fields='symbol,lower_limit', start_date=date, end_date=date, df=True).set_index('symbol')
#修改列名为data
lower_limit.columns =
close = history(symbols, frequency='1d', start_time=date, end_time=date, fields='symbol,close', adjust=ADJUST_NONE, df=True).set_index('symbol')
close.columns =
common_stock =list( set(list(lower_limit.index))&set(list(close.index)))
lower_limit,close = lower_limit.loc,close.loc
limit_up_stock = list(close.dropna().index)
if not limit_up_stock :
return list()
else :
# 上个交易日
last_date = get_previous_trading_date(exchange='SZSE', date=date)
# 上个交易日不跌停
lower_limit = get_history_instruments(limit_up_stock, fields='symbol,upper_limit', start_date=last_date, end_date=last_date, df=True).set_index('symbol')
if not lower_limit.empty:
lower_limit.columns =
close = history(limit_up_stock, frequency='1d', start_time=last_date, end_time=last_date, fields='symbol,close', adjust=ADJUST_NONE, df=True).set_index('symbol')
close.columns =
common_stock = list(set(list(lower_limit.index))&set(list(close.index)))
upper_limit,close = lower_limit.loc,close.loc
first_limit_up_stock = list(close.dropna().index)
return first_limit_up_stock
else:
return list()
def on_backtest_finished(context, indicator):
print('累计收益率：',str(round(indicator.pnl_ratio*100,2)),',年化收益率：',str(round(indicator.pnl_ratio_annual*100,2)),',最大回撤：',str(round(indicator.max_drawdown*100,2)),
'，夏普比率：',str(round(indicator.sharp_ratio,2)),'，胜率：',str(round(indicator.win_ratio*100,2)),
',开仓次数：',indicator.open_count,'，平仓次数：',indicator.close_count)
if __name__ == '__main__':
'''
strategy_id策略ID, 由系统生成
filename文件名, 请与本文件名保持一致
mode运行模式, 实时模式:MODE_LIVE回测模式:MODE_BACKTEST
token绑定计算机的ID, 可在系统设置-密钥管理中生成
backtest_start_time回测开始时间
backtest_end_time回测结束时间
backtest_adjust股票复权方式, 不复权:ADJUST_NONE前复权:ADJUST_PREV后复权:ADJUST_POST
backtest_initial_cash回测初始资金
backtest_commission_ratio回测佣金比例
backtest_slippage_ratio回测滑点比例
'''
run(strategy_id='3303c0e4-64e7-11ed-927c-025065c1ef9f',
filename='main.py',
mode=MODE_BACKTEST,
token='2a73fd387f79b2c096806a91a6b13f915f981007',
backtest_start_time='2024-10-10 08:00:00',
backtest_end_time='2025-02-14 16:02:00',
backtest_adjust=ADJUST_PREV,
backtest_initial_cash=10000000,
backtest_commission_ratio=0.0008,
backtest_slippage_ratio=0.0001)以上代码是个相对完整的代码，tick级别交易，对于想学习高频交易策略编写的朋友有一定的借鉴意义。
这段代码实现了一个基于 Python 的股票交易策略，具体为“二板打板策略”。该策略的核心逻辑是：在每个交易日开盘前筛选出上一个交易日首次涨停的股票，并在盘中监控股票价格达到特定条件时买入；次日早盘判断是否涨停，卖出非涨停股。以下是代码的主要模块和功能解释：
1. 导入必要的库
from __future__ import print_function, absolute_import
from gm.api import *
import numpy as np
import datetime
import snowater.common_func_new as cf- gm.api 是用于获取市场数据和执行交易操作的 API。
- numpy 用于数值计算。
- datetime 用于处理日期和时间。
- snowater.common_func_new 是自定义模块，提供了辅助函数。
2. 策略初始化 (init 函数)
def init(context):
context.max_holding_num = 5 # 最大持股数量
context.frequency = 'tick' # 数据频率
context.first_limit_up_stock = # 每天首板股票池
context.target_stocks = {} # 目标变量,日期：代码
schedule(schedule_func=before_market, date_rule='1d', time_rule='09:20:00') # 盘前定时任务
schedule(schedule_func=end_market, date_rule='1d', time_rule='10:00:00') # 尾盘定时任务
schedule(schedule_func=after_market, date_rule='1d', time_rule='15:20:00') # 盘后定时任务
- 设置最大持股数量、数据频率等参数。
- 初始化股票池和目标变量。
- 定义三个定时任务，分别在盘前、尾盘和盘后执行。
3. 盘前任务 (before_market 函数)
def before_market(context):
all_stocks = cf.get_instruments_all_stocks(context) # 获取所有A股代码
last_date = get_previous_trading_date(exchange='SZSE', date=context.now) # 历史交易日
context.first_limit_up_stock = cal_first_limit_up_stocks(context, all_stocks, last_date) # 计算首板股票
context.first_limit_up_stock = cf.filter_stocks_by_financial_indicators(context.first_limit_up_stock, last_date) # 根据财务指标筛选
if context.now.date() not in context.target_stocks.keys():
context.target_stocks =
if len(context.account().positions()) >= context.max_holding_num:
return
subscribe(context.first_limit_up_stock, frequency=context.frequency, count=1, unsubscribe_previous=False)
- 获取所有A股代码。
- 计算并筛选出上一个交易日首次涨停的股票。
- 根据多个财务指标进一步筛选。
- 如果当前持仓已满，则直接返回。
- 订阅筛选后的股票数据。
4. 盘中买入逻辑 (on_tick 函数)
def on_tick(context, tick):
now = context.now
if ((now.hour == 9 and now.minute < 30) or
((now.hour == 11 and now.minute >= 30) and (now.hour < 13)) or
now.hour >= 15):
return
symbol = tick
if symbol in context.first_limit_up_stock:
sell_1_price = tick # 卖1挂单价
sell_5_price = tick # 卖5挂单价
total_asset = context.account().cash
trade_volume = int(np.floor(total_asset / context.max_holding_num / 2 / sell_1_price / 100) * 100) if sell_1_price > 0 else 0
cond1 = now.hour == 9 and now.minute >= 30 or now.hour > 11
cond2 = sell_5_price == 0
cond3 = sell_1_price > 0
cond4 = symbol not in context.target_stocks
cond5 = len(context.target_stocks) < context.max_holding_num
if cond1 and cond2 and cond3 and cond4 and cond5:
context.target_stocks.append(symbol)
order_volume(symbol=symbol, volume=trade_volume, side=OrderSide_Buy, order_type=OrderType_Limit, position_effect=PositionEffect_Open, price=sell_1_price)
if len(context.target_stocks) == context.max_holding_num:
unsubscribe(symbols='*', frequency=context.frequency)
- 处理实时行情数据（tick）。
- 判断是否满足买入条件（如开盘后、卖5价格为0、买一挂单价大于0、未持仓、持仓数量未满等）。
- 满足条件则下单买入，并更新目标股票池。
5. 尾盘卖出逻辑 (end_market 函数)
def end_market(context):
Account_positions = context.account().positions()
holding_stock = for position in Account_positions]
print(f"{context.now.date()} 持仓股票：{holding_stock}")
last_date = get_previous_trading_date(exchange='SZSE', date=context.now)
for position in Account_positions:
symbol = position
last_close = history(symbol=symbol, frequency='1d', start_time=last_date, end_time=last_date,
fields='symbol,close', adjust=ADJUST_NONE, df=True)
if last_close.empty:
continue
data = context.data(symbol=symbol, frequency=context.frequency, count=1)
cf.apply_stop_loss_strategy(context, current(symbols=symbol))
current_price = data # 当前价格
yesterday_close = last_close.close.values
price_change = (current_price - yesterday_close) / yesterday_close * 100
cond1 = position-position>0 if context.mode==MODE_BACKTEST else position>0
cond4 = price_change < -2 # 涨幅小于-2%则卖出
if cond1 and cond4:
order_volume(symbol=symbol, volume=position, side=OrderSide_Sell,
order_type=OrderType_Market,position_effect=PositionEffect_Close,
price=0)
- 获取当前持仓股票。
- 遍历持仓股票，获取昨日收盘价和当前价格。
- 应用止损策略。
- 如果涨幅小于-2%，则卖出。
6. 盘后任务 (after_market 函数)
def after_market(context):
last_date = datetime.datetime.strptime(get_previous_trading_date(exchange='SZSE', date=context.now), '%Y-%m-%d').date()
if last_date in context.target_stocks.keys():
if len(context.target_stocks) == 0:
context.target_stocks.pop(last_date)
else:
next_date = datetime.datetime.strptime(get_next_trading_date(exchange='SZSE', date=context.now), '%Y-%m-%d').date()
context.target_stocks = context.target_stocks
context.target_stocks.pop(last_date)
- 更新目标股票池，移除空的日期条目，并将前一天的目标股票池复制到下一天。
7. 订单状态处理 (on_order_status 函数)
def on_order_status(context, order):
symbol = order
price = order
volume = order
target_percent = order
status = order
side = order
effect = order
order_type = order
if status == 3:
if effect == 1:
side_effect = '开多仓' if side == 1 else '开空仓'
else:
last_date = datetime.datetime.strptime(get_previous_trading_date(exchange='SZSE', date=context.now), '%Y-%m-%d').date()
try:
context.target_stocks.remove(symbol)
except:
context.target_stocks.remove(symbol)
side_effect = '平空仓' if side == 1 else '平多仓'
order_type_word = '限价' if order_type == 1 else '市价'
print('{}:标的：{}，操作：以{}{}，委托价格：{}，委托数量：{}'.format(datetime.datetime.strftime(context.now,"%Y-%m-%d %H:%M:%S"), symbol, order_type_word, side_effect, price, volume))
- 处理订单状态变化，记录成交信息并更新目标股票池。
8. 回测结束处理 (on_backtest_finished 函数)
def on_backtest_finished(context, indicator):
print('累计收益率：', str(round(indicator.pnl_ratio*100, 2)), ',年化收益率：', str(round(indicator.pnl_ratio_annual*100, 2)), ',最大回撤：', str(round(indicator.max_drawdown*100, 2)),
'，夏普比率：', str(round(indicator.sharp_ratio, 2)), '，胜率：', str(round(indicator.win_ratio*100, 2)),
',开仓次数：', indicator.open_count, '，平仓次数：', indicator.close_count)
- 打印回测结果，包括累计收益率、年化收益率、最大回撤、夏普比率、胜率等。
9. 主程序入口 (if __name__ == '__main__':)
if __name__ == '__main__':
run(strategy_id='3303c0e4-64e7-11ed-927c-025065c1ef9f',
filename='main.py',
mode=MODE_BACKTEST,
token='2a73fd387f79b2c096806a91a6b13f915f981007',
backtest_start_time='2024-10-10 08:00:00',
backtest_end_time='2025-02-14 16:02:00',
backtest_adjust=ADJUST_PREV,
backtest_initial_cash=10000000,
backtest_commission_ratio=0.0008,
backtest_slippage_ratio=0.0001)
- 设置回测参数并启动策略运行。
import snowater.common_func_new as cf 用的是我个人写的一些方法，在这个策略中用到了，获取股票列表，基本面数据过滤函数和风险控制函数。加入这些并没有显著提升收益率。应该是还有些问题，就不在这里分享。 对这个策略有兴趣的，可以注释掉，或是自己实现这几个函数。
总结
该策略通过定时任务和实时行情监控，实现了对每日首次涨停股票的买入和次日卖出操作，结合了财务指标筛选和风险控制机制，确保了策略的稳健性。
有兴趣的朋友记得留言，关注转发。以后会有更多的干货分享。
1