超短策略
from jqdata import *
from jqfactor import *
import pandas as pd
from datetime import datetime,timedelta,date
################################### 初始化设置 #############################################
def initialize(context):
set_option('use_real_price', True)
log.set_level('system', 'error')
set_option('avoid_future_data', True)
def after_code_changed(context):
g.n_days_limit_up_list = [] #重新初始化列表
unschedule_all() # 取消所有定时运行
# run_daily(get_stock_list, '9:05')
run_daily(buy, '09:26')
run_daily(sell, time='11:25', reference_security='000300.XSHG')
run_daily(sell, time='14:50', reference_security='000300.XSHG')
def after_trading_end(context):
print('———————————————————————————————————')
## 定义股票池
def set_stockpool(context):
yesterday = context.previous_date
initial_list = get_all_securities('stock', yesterday).index.tolist()
return initial_list
################################## 交易函数群 ##################################
def buy(context):
current_data = get_current_data()
qualified_stocks = get_stock_list(context)
if qualified_stocks:
value = context.portfolio.available_cash / len(qualified_stocks)
for s in qualified_stocks:
# 下单 #至少够买1手
if context.portfolio.available_cash/current_data[s].last_price>100:
order_value(s, value, MarketOrderStyle(current_data[s].day_open))
print('买入' + s)
def sell(context):
stime = context.current_dt.strftime("%H%M")
current_data = get_current_data()
for s in list(context.portfolio.positions):
close_data = attribute_history(s, 4, '1d', ['close'])
M4=close_data['close'].mean()
MA5=(M4*4+current_data[s].last_price)/5
position=context.portfolio.positions[s]
if ((position.closeable_amount != 0) and (current_data[s].last_price < current_data[s].high_limit) and (current_data[s].last_price > 1*position.avg_cost)):#avg_cost当前持仓成本
order_target_value(s, 0)
ret=100*(position.price/position.avg_cost-1)
print('止盈卖出 ' + get_security_info(s).display_name + s + ' 收益率:{:.2f}%'.format(ret,'.2f'))
#跌破5日线止损
if ((position.closeable_amount != 0) and (current_data[s].last_price < MA5)):
order_target_value(s, 0)
ret=100*(position.price/position.avg_cost-1)
print('止损卖出 ' + get_security_info(s).display_name + s + ' 收益率:{:.2f}%'.format(ret,'.2f'))
##### 选股函数群 #####
选股函数解锁后查看:
# 每日初始股票池
def prepare_stock_list(context):
today = context.current_dt.date()
yesterday = context.previous_date
initial_list = set_stockpool(context)
initial_list = filter_kcbj_stock(initial_list)
initial_list = filter_st_paused_stock(initial_list, today)
initial_list = filter_new_stock(initial_list, today)
# 首次运行,添加前2天的数据
if not g.n_days_limit_up_list:
days = get_trade_days( end_date = yesterday, count=3)[:-1]
for day in days:
g.n_days_limit_up_list.append(get_hl_stock(initial_list, day, 1))
hl_list = get_hl_stock(initial_list, yesterday, 1) # 昨日涨停
g.n_days_limit_up_list.append(hl_list)
hl1_list = set(g.n_days_limit_up_list[-2]) # 前1日曾涨停
#hl2_list = set(g.n_days_limit_up_list[-2] + g.n_days_limit_up_list[-3]) # 前2日曾涨停
hl_list = [stock for stock in hl_list if stock not in hl1_list]
# 昨日曾涨停但未封板
hl_list2 = get_ever_hl_stock2(initial_list, yesterday)
hl_list2 = [stock for stock in hl_list2 if stock not in hl1_list]
g.n_days_limit_up_list.pop(0) # 移除无用的数据
return hl_list, hl_list2
################################### 涨停形态筛选 ##################################
1 、筛选出某一日涨停的股票 2 、筛选出某一日曾经涨停的股票,含炸板的 3 、筛选出某一日曾经涨停但未封板的股票
# 过滤函数
def filter_new_stock(initial_list, date, days=50):
return [stock for stock in initial_list if get_security_info(stock).start_date < date - timedelta(days=days)]
def filter_st_paused_stock(initial_list, date):
current_data = get_current_data()
return [stock for stock in initial_list if not (
current_data[stock].is_st or
current_data[stock].paused or
'退' in current_data[stock].name)]
def filter_kcbj_stock(initial_list):
return [stock for stock in initial_list if stock[0] != '4' and stock[0] != '8' and stock[:2] != '68'] #and stock[0] != '3'
### end ###
最后更新: 2025-03-6 10:56
