JoinQuant策略代碼示例

• 總體回測前

'''
================================================================================
總體回測前
================================================================================
'''
#總體回測前要做的事情
def initialize(context):
set_params() #1設置策參數
set_variables() #2設置中間變量
set_backtest() #3設置回測條件

#1
#設置策略參數
def set_params():
g.tc=15 # 調倉頻率
g.N=4 #持倉數目
g.security = ["000001.XSHE","000002.XSHE","000006.XSHE","000007.XSHE","000009.XSHE"]#設置股票池

#2
#設置中間變量
def set_variables():
return

#3
#設置回測條件
def set_backtest():
set_option('use_real_price', True) #用真實價格交易
log.set_level('order', 'error')

 

• 每天開盤前

'''
================================================================================
每天開盤前
================================================================================
'''
#每天開盤前要做的事情
def before_trading_start(context):
    set_slip_fee(context) 

#4 
# 根據不同的時間段設置滑點與手續費
def set_slip_fee(context):
    # 將滑點設置為0
    set_slippage(FixedSlippage(0)) 
    # 根據不同的時間段設置手續費
    dt=context.current_dt
    
    if dt>datetime.datetime(2013,1, 1):
        set_commission(PerTrade(buy_cost=0.0003, sell_cost=0.0013, min_cost=5)) 
        
    elif dt>datetime.datetime(2011,1, 1):
        set_commission(PerTrade(buy_cost=0.001, sell_cost=0.002, min_cost=5))
            
    elif dt>datetime.datetime(2009,1, 1):
        set_commission(PerTrade(buy_cost=0.002, sell_cost=0.003, min_cost=5))
                
    else:
        set_commission(PerTrade(buy_cost=0.003, sell_cost=0.004, min_cost=5))

 

• 每天交易時（以雙均線策略為例）

'''
================================================================================
每天交易時
================================================================================
'''
def handle_data(context, data):
    # 將總資金等分為g.N份，為每只股票配資
    capital_unit = context.portfolio.portfolio_value/g.N
    toSell = signal_stock_sell(context,data)
    toBuy = signal_stock_buy(context,data)
    # 執行賣出操作以騰出資金
    for i in range(len(g.security)):
        if toSell[i]==1:
            order_target_value(g.security[i],0)
    # 執行買入操作
    for i in range(len(g.security)):
        if toBuy[i]==1:
            order_target_value(g.security[i],capital_unit)  
    if not (1 in toBuy) or (1 in toSell):
        # log.info("今日無操作")
        send_message("今日無操作")


#5
#獲得賣出信號
#輸入：context, data
#輸出：sell - list
def signal_stock_sell(context,data):
    sell = [0]*len(g.security)
    for i in range(len(g.security)):
    # 算出今天和昨天的兩個指數移動均線的值，我們這里假設長線是60天，短線是1天(前一天的收盤價)
        (ema_long_pre,ema_long_now) = get_EMA(g.security[i],60,data)
        (ema_short_pre,ema_short_now) = get_EMA(g.security[i],1,data)
        # 如果短均線從上往下穿越長均線，則為死叉信號，標記賣出
        if ema_short_now < ema_long_now and ema_short_pre > ema_long_pre and context.portfolio.positions[g.security[i]].sellable_amount > 0:
            sell[i]=1
    return sell
        

#6
#獲得買入信號
#輸入：context, data
#輸出：buy - list
def signal_stock_buy(context,data):
    buy = [0]*len(g.security)
    for i in range(len(g.security)):
    # 算出今天和昨天的兩個指數移動均線的值，我們這里假設長線是60天，短線是1天(前一天的收盤價)
        (ema_long_pre,ema_long_now) = get_EMA(g.security[i],60,data)
        (ema_short_pre,ema_short_now) = get_EMA(g.security[i],1,data)
        # 如果短均線從下往上穿越長均線，則為金叉信號，標記買入
        if ema_short_now > ema_long_now and ema_short_pre < ema_long_pre and context.portfolio.positions[g.security[i]].sellable_amount == 0 :
            buy[i]=1
    return buy


#7
# 計算移動平均線數據
# 輸入：股票代碼-字符串，移動平均線天數-整數
# 輸出：算術平均值-浮點數
def get_MA(security_code,days):
    # 獲得前days天的數據，詳見API
    a=attribute_history(security_code, days, '1d', ('close'))
    # 定義一個局部變量sum，用於求和
    sum=0
    # 對前days天的收盤價進行求和
    for i in range(1,days+1):
        sum+=a['close'][-i]
    # 求和之后除以天數就可以的得到算術平均值啦
    return sum/days

#8
# 計算指數移動平均線數據
# 輸入：股票代碼-字符串，移動指數平均線天數-整數，data
# 輸出：今天和昨天的移動指數平均數-浮點數
def get_EMA(security_code,days,data):
    # 如果只有一天的話,前一天的收盤價就是移動平均
    if days==1:
    # 獲得前兩天的收盤價數據，一個作為上一期的移動平均值，后一個作為當期的移動平均值
        t = attribute_history(security_code, 2, '1d', ('close'))
        return t['close'][-2],t['close'][-1]
    else:
    # 如果全局變量g.EMAs不存在的話，創建一個字典類型的變量，用來記錄已經計算出來的EMA值
        if 'EMAs' not in dir(g):
            g.EMAs={}
        # 字典的關鍵字用股票編碼和天數連接起來唯一確定，以免不同股票或者不同天數的指數移動平均弄在一起了
        key="%s%d" %(security_code,days)
        # 如果關鍵字存在，說明之前已經計算過EMA了，直接迭代即可
        if key in g.EMAs:
            #計算alpha值
            alpha=(days-1.0)/(days+1.0)
            # 獲得前一天的EMA（這個是保存下來的了）
            EMA_pre=g.EMAs[key]
            # EMA迭代計算
            EMA_now=EMA_pre*alpha+data[security_code].close*(1.0-alpha)
            # 寫入新的EMA值
            g.EMAs[key]=EMA_now
            # 給用戶返回昨天和今天的兩個EMA值
            return (EMA_pre,EMA_now)
        # 如果關鍵字不存在，說明之前沒有計算過這個EMA，因此要初始化
        else:
            # 獲得days天的移動平均
            ma=get_MA(security_code,days) 
            # 如果滑動平均存在（不返回NaN）的話，那么我們已經有足夠數據可以對這個EMA初始化了
            if not(isnan(ma)):
                g.EMAs[key]=ma
                # 因為剛剛初始化，所以前一期的EMA還不存在
                return (float("nan"),ma)
            else:
                # 移動平均數據不足days天，只好返回NaN值
                return (float("nan"),float("nan"))

• 每天收盤后

'''
================================================================================
每天收盤后
================================================================================
'''
# 每日收盤后要做的事情（本策略中不需要）
def after_trading_end(context):
    return