策略示例
纸上得来终觉浅。本章给出五个典型策略的完整代码，覆盖日线趋势、多股票择时、日内做 T、盘前抢板、tick 高频四种节奏，每个都标注运行频率和关键逻辑，改参数就能变成自己的策略。
一、双均线策略（日线）
思路：5 日均线上穿 20 日均线全仓买入，下穿则清仓。运行频率：日线。
def initialize(context):
g.security = '600570.SS'
set_universe(g.security)
def before_trading_start(context, data):
# 盘前取近 20 日收盘价
h = get_history(20, '1d', 'close', g.security, fq='dypre',
include=False, is_dict=True)
g.close = h[g.security]['close']
def handle_data(context, data):
current_price = data[g.security].close
# 拼上当前价合成完整序列
import numpy as np
close = np.concatenate((g.close, [current_price]))
ma5 = close[-5:].mean()
ma20 = close[-20:].mean()
if ma5 > ma20 and get_position(g.security).amount == 0:
order_value(g.security, context.portfolio.cash)
log.info('金叉买入')
elif ma5 < ma20 and get_position(g.security).enable_amount > 0:
order_target(g.security, 0)
log.info('死叉清仓')要点：盘前把历史数据缓存到 g.close，盘中只拼一根当前 K 线，避免 handle_data 内重复拉数据。
二、MACD 多股票金叉策略（日线）
思路：在沪深 300 成分股中，找 DIF、DEA 均为正且 MACD 由负转正（强势金叉）的标的买入；DIF、DEA 均为负且 MACD 由正转负（弱势死叉）则卖出。等权持仓 10 只。运行频率：日线。
def initialize(context):
g.hold_num = 10
set_benchmark('000300.SS')
def before_trading_start(context, data):
g.stocks = get_index_stocks('000300.SS')
g.stocks = filter_stock_by_status(g.stocks, ['ST', 'HALT', 'DELISTING'])
h = get_history(100, '1d', 'close', security_list=g.stocks,
fq='dypre', include=False, is_dict=True)
g.close_dict = {s: h[s]['close'] for s in g.stocks}
g.every_value = context.portfolio.portfolio_value / g.hold_num
def handle_data(context, data):
for s in g.stocks:
close = g.close_dict[s]
dif, dea, macd = get_MACD(close, 12, 26, 9)
DIF, DEA = dif[-1], dea[-1]
macd_now, macd_pre = macd[-1], macd[-2]
pos = get_position(s)
if pos.amount == 0:
# 强势金叉 + 资金充足
if DIF > 0 and DEA > 0 and macd_pre < 0 and macd_now >= 0:
if context.portfolio.cash >= g.every_value * 0.8:
order_target_value(s, g.every_value)
log.info('买入 %s' % s)
else:
# 弱势死叉
if DIF < 0 and DEA < 0 and macd_pre >= 0 and macd_now < 0:
order_target(s, 0)
log.info('卖出 %s' % s)要点：盘前一次性批量拉 100 日收盘价（is_dict=True 取数更快），handle_data 里只做计算和下单，避免循环中重复请求接口。
三、日内 T+0 策略（分钟）
思路：先持有一份底仓，盘中用 5 分钟和 15 分钟 RSI 多空共振判断做正 T（先买后卖）或反 T（先卖后买），盈利 1% 恢复头寸，收盘前清算。运行频率：分钟。
import numpy as np
def initialize(context):
g.amount = 100 # 一份标准头寸
g.rate = 1 # 做 T 涨跌幅 1%
g.L, g.S = 50, 80 # 长短周期 RSI 阈值
g.security = '510500.SS'
g.ini_buy = False
if not is_trade():
set_limit_mode('UNLIMITED')
def before_trading_start(context, data):
g.B_flag = False # 正 T 开关
g.S_flag = False # 反 T 开关
g.handle_flag = True
def handle_data(context, data):
if not g.handle_flag:
return
k = get_current_kline_count() # 当日已走完的分钟数
if not g.ini_buy:
order(g.security, g.amount) # 开盘先建底仓
g.ini_buy = True
g.handle_flag = False
return
if k < 30:
return
if k % 5 == 0: # 每 5 分钟判断一次
h5 = get_history(100, '5m', 'close', g.security, is_dict=True)[g.security]['close']
h15 = get_history(100, '15m', 'close', g.security, is_dict=True)[g.security]['close']
now_price = data[g.security].close
h15 = np.concatenate((h15, [now_price]))
rsi5 = get_rsi(h5)[-1]
rsi15 = get_rsi(h15)[-1]
if rsi15 > g.L and rsi5 > g.S and not g.B_flag:
order(g.security, g.amount); g.B_flag = True; g.B_cost = now_price
if rsi15 < 100-g.L and rsi5 < 100-g.S and not g.S_flag:
order(g.security, -g.amount); g.S_flag = True; g.S_cost = now_price
# 盈利恢复头寸
if g.B_flag and data[g.security].price >= g.B_cost * (1 + g.rate/100):
order(g.security, -g.amount); g.B_flag = False
if g.S_flag and data[g.security].price <= g.S_cost * (1 - g.rate/100):
order(g.security, g.amount); g.S_flag = False
# 收盘前清算回底仓量
if k >= 238:
order_target(g.security, g.amount)
def get_rsi(arr, periods=14):
# 自实现 RSI，返回与输入等长的列表
rsi = [np.nan] * len(arr)
if len(arr) <= periods:
return rsi
up = down = 0
for i in range(1, periods + 1):
if arr[i] >= arr[i-1]: up += arr[i] - arr[i-1]
else: down += arr[i-1] - arr[i]
up, down = up/periods, down/periods
rsi[periods] = 100 - 100/(1 + up/(down or 1e-9))
for j in range(periods+1, len(arr)):
u = max(arr[j]-arr[j-1], 0); d = max(arr[j-1]-arr[j], 0)
up = (up*(periods-1)+u)/periods
down = (down*(periods-1)+d)/periods
rsi[j] = 100 - 100/(1 + up/(down or 1e-9))
return rsi要点：日内做 T 的前提是先有底仓（绕开 T+1）；用 get_current_kline_count() 控制判断节奏；收盘前用 order_target 恢复到初始头寸。
四、集合竞价追涨停（定时任务）
思路：每天 9:23 用
run_daily触发，若最新价已触及涨停价则买入。运行频率：日线/分钟均可。
def initialize(context):
g.security = '600570.SS'
set_universe(g.security)
run_daily(context, auction_func, time='9:23')
def auction_func(context):
snap = get_snapshot(g.security)
last_px = float(snap[g.security]['last_px'])
up_px = float(snap[g.security]['up_px'])
if last_px >= up_px:
order(g.security, 100, limit_price=up_px)
log.info('追涨停买入 %s' % g.security)
def handle_data(context, data):
pass要点：run_daily 只能在 initialize 中注册；集合竞价阶段的实时行情要用 get_snapshot 而非 data 对象。
五、tick 级别均线策略（仅交易）
思路：每 3 秒触发一次，用最新 tick 价合成实时均线，金叉买入、死叉卖出。运行频率：tick（仅交易）。
def initialize(context):
g.security = '600570.SS'
set_universe(g.security)
run_interval(context, tick_ma, seconds=3)
def before_trading_start(context, data):
h = get_history(10, '1d', 'close', g.security, fq='pre', include=False)
g.close_arr = h['close'].values
def tick_ma(context):
snap = get_snapshot(g.security)
price = snap[g.security]['last_px']
ma5 = (price + g.close_arr[-4:].sum()) / 5
ma10 = (price + g.close_arr[-9:].sum()) / 10
if ma5 > ma10:
order_value(g.security, context.portfolio.cash)
log.info('tick 金叉买入')
elif ma5 < ma10 and get_position(g.security).amount > 0:
order_target(g.security, 0)
log.info('tick 死叉卖出')
def handle_data(context, data):
pass要点：run_interval 只在交易时段（9:30~15:00）触发；tick_data 和 run_interval 累计线程数上限为 5，超过会堵塞。
小结
五个示例覆盖了 PTrade 的四种节奏：日线趋势、多股择时、分钟日内、盘前定时、tick 高频。抄起来改股票池和参数就能跑，但记住先回测、再模拟、最后小仓位试水。进阶内容看期货与定时任务：期货专用与定时任务。
