在聚寬平臺上編寫鱷魚交易法則

# 導入函數庫
import jqdata
import numpy as np

# 初始化函數，設定基準等等
def initialize(context):
set_option('use_real_price', True)
set_order_cost(OrderCost(close_tax=0.001, open_commission=0.0003, close_commission=0.0003, min_commission=5), type='stock')
set_benchmark('000300.XSHG')

g.up_price = {} #向上碎形最高價
g.low_price = {} #向下碎形最低價
g.up_fractal_exists = {} #判斷有效向上碎形
g.down_fractal_exists = {} #判斷有效向下碎形
g.AO_index = {} #存放連續的AO指標數據
g.cal_AC_index = {} #計算AC指標中轉存儲
g.AC_index = {} #存放連續的AC指標數據
g.amount = {} #滿倉倉位
g.stock = get_index_stocks('000300.XSHG')
g.buy_stock = []
g.month = context.current_dt.month
run_monthly(select_universe,1,'open')

#重置全局變量
def reset_global():
g.up_price = {} #向上碎形最高價
g.low_price = {} #向下碎形最低價
g.up_fractal_exists = {} #判斷有效向上碎形
g.down_fractal_exists = {} #判斷有效向下碎形
g.AO_index = {} #存放連續的AO指標數據
g.cal_AC_index = {} #計算AC指標中轉存儲
g.AC_index = {} #存放連續的AC指標數據
g.amount = {} #滿倉倉位
g.buy_stock = []

def initial_stock_global(stock):
g.up_price[stock] = 0
g.low_price[stock] = 0
g.up_fractal_exists[stock] = False
g.down_fractal_exists[stock] = False #判斷有效向下碎形
g.AO_index[stock] = [0] #存放連續的AO指標數據
g.cal_AC_index[stock] = [0] #計算AC指標中轉存儲
g.AC_index[stock] = [0] #存放連續的AC指標數據
g.amount[stock] = 0 #滿倉倉位

#輪換選股後清空持倉
def reset_position(context):
for stock in g.buy_stock:
order_target(stock,0)
log.info("sell %s for reset position"%stock)

#選股
def select_universe(context):
#每三個月操做一次
month = context.current_dt.month
if month%6 != g.month%6:
return
#清空全局變量
reset_position(context)
reset_global()
hist = history(30,'1d','close',g.stock,df = False)
for stock in g.stock:
if is_sleeping_alligator(stock,hist,20):
g.buy_stock.append(stock)
#初始化該股票全局變量
initial_stock_global(stock)
print g.buy_stock
return None

#睡着的鱷魚
def is_sleeping_alligator(stock,hist,nday):
for i in range(nday):
if is_struggle(stock,hist,i) == False:
return False
return True

#均線糾纏，BRG三線很是接近
def is_struggle(stock,hist,delta):
blue_line = hist[stock][-21-delta:-8-delta].mean()
red_line = hist[stock][-13-delta:-5-delta].mean()
green_line = hist[stock][-8-delta:-3-delta].mean()
if abs(blue_line/red_line-1)<0.02 and abs(red_line/green_line-1)<0.02:
return True
else:
return False

#判斷 向上 或 向下 碎形
def is_fractal(stock,direction):
hist = attribute_history(stock, 5, fields=[direction])
if direction == 'high':
if np.all(hist.iloc[:2] < hist.iloc[2]) and np.all(hist.iloc[3:] < hist.iloc[2]):
g.up_price[stock] = hist.iloc[2].values
return True
elif direction == 'low':
if np.all(hist.iloc[:2] > hist.iloc[2]) and np.all(hist.iloc[3:] > hist.iloc[2]):
g.low_price[stock] = hist.iloc[2].values
return True
return False

#經過比較碎形與紅線位置，判斷碎形是否有效
def is_effective_fractal(stock, direction):
if is_fractal(stock,direction):
hist = attribute_history(stock, 11)
red_line = hist['close'][:-3].mean()
close_price = hist['close'][-1]
if direction == 'high':
if close_price > red_line:
g.up_fractal_exists[stock] = True
else:
g.up_fractal_exists[stock] = False
elif direction == 'low':
if close_price < red_line:
g.down_fractal_exists[stock] = True
else:
g.down_fractal_exists[stock] = False

#N日內最高價格的N日線
def nday_high_point(stock,n):
hist = history(2*n,'1d','high',[stock],df = False)[stock]
high_point = []
for i in range(n):
high_point.append(max(hist[-5-i:-1-i]))
return np.array(high_point).mean()

#N日內最低價格的N日線
def nday_low_point(stock,n):
hist = history(2*n,'1d','low',[stock],df = False)[stock]
low_point = []
for i in range(n):
low_point.append(max(hist[-5-i:-1-i]))
return np.array(low_point).mean()

#AO=5日內（最高-最低）/2的5日移動平均-34日內（最高-最低）/2的34日移動平均
def AO_index(stock):
g.AO_index[stock].append(nday_high_point(stock,5)/2 + nday_low_point(stock,5)/2\
- nday_high_point(stock,34)/2 - nday_low_point(stock,34)/2)
return None

#AO-AO的5日平均值的5日平均
def AC_index(stock):
AO_index(stock)
if len(g.AO_index[stock]) >= 5:
g.cal_AC_index[stock].append(g.AO_index[stock][-1] - np.array(g.AO_index[stock][-5:]).mean())
if len(g.cal_AC_index[stock]) >=5:
g.AC_index[stock].append(np.array(g.cal_AC_index[stock][-5:]).mean())

#判斷序列n日上行
def is_up_going(alist,n):
if len(alist) < n:
return False
for i in range(n-1):
if alist[-(1+i)] <= alist[-(2+i)]:
return False
return True

#判斷序列n日下行
def is_down_going(alist,n):
if len(alist) < n:
return False
for i in range(n-1):
if alist[-(1+i)] >= alist[-(2+i)]:
return False
return True

#碎形被突破
def active_fractal(stock,direction):
close_price = history(1,'1d','close',[stock],df=False)[stock][0]
if direction == 'up' and close_price > g.up_price[stock]:
return True
elif direction == 'down' and close_price < g.low_price[stock]:
return True
return False

#進場，初始倉位
def set_initial_position(stock,context):
close_price = history(1,'1d','close',[stock],df=False)[stock][0]
g.amount[stock] = context.portfolio.cash/close_price/len(g.buy_stock)*3
order(stock, g.amount[stock])
log.info("buying %s 股數爲 %s"%(stock,g.amount[stock]))
g.down_fractal_exists[stock] = False

#賣出
def sell_all_stock(stock,context):
order_target(stock,0)
log.info("selling %s"%stock)
g.up_fractal_exists[stock] = False

#加倉
def adjust_position(stock,context,position):
order(stock,g.amount[stock]*position)
log.info("adjust position buying %s 股數爲 %s"%(stock,g.amount[stock]*position))

# 計算股票前n日收益率
def security_return(days,security_code):
hist1 = attribute_history(security_code, days + 1, '1d', 'close',df=False)
security_returns = (hist1['close'][-1]-hist1['close'][0])/hist1['close'][0]
return security_returns

# 止損，根據前n日收益率
def conduct_nday_stoploss(context,security_code,days,bench):
if security_return(days,security_code)<= bench:
for stock in g.buy_stock:
order_target_value(stock,0)
log.info("Sell %s for stoploss" %stock)
return True
else:
return False

# 計算股票累計收益率（從建倉至今）
def security_accumulate_return(context,data,stock):
current_price = data[stock].price
cost = context.portfolio.positions[stock].avg_cost
if cost != 0:
return (current_price-cost)/cost
else:
return None

# 個股止損，根據累計收益
def conduct_accumulate_stoploss(context,data,stock,bench):
if security_accumulate_return(context,data,stock) != None\
and security_accumulate_return(context,data,stock) < bench:
order_target_value(stock,0)
log.info("Sell %s for stoploss" %stock)
return True
else:
return False

# 個股止盈，根據累計收益
def conduct_accumulate_stopwin(context,data,stock,bench):
if security_accumulate_return(context,data,stock) != None\
and security_accumulate_return(context,data,stock) > bench:
order_target_value(stock,0)
log.info("Sell %s for stopwin" %stock)
return True
else:
return False

def handle_data(context,data): #大盤止損 if conduct_nday_stoploss(context,'000300.XSHG',3,-0.03): return for stock in g.buy_stock: #個股止損 if conduct_accumulate_stopwin(context,data,stock,0.3)\ or conduct_accumulate_stoploss(context,data,stock,-0.1): return #計算AO，AC指標 AC_index(stock) #空倉時，尋找機會入場 if context.portfolio.positions[stock].amount == 0: #計算向上碎形 is_effective_fractal(stock,'high') #有效向上碎形存在，並被突破，買入 if g.up_fractal_exists and active_fractal(stock,'up'): close_price = history(5, '1d', 'close', [stock],df = False) if is_up_going(g.AO_index[stock],5)\ and is_up_going(g.AC_index[stock],3)\ and is_up_going(close_price[stock],2): set_initial_position(stock,context) #有持倉時，加倉或離場 else: #計算向下碎形 is_effective_fractal(stock,'low') #出場條件1：有效向下碎形存在，並被突破，賣出 if g.down_fractal_exists and active_fractal(stock,'down'): sell_all_stock(stock,context) return