基于ATR通道的金肯纳特交易系统

import numpy as np
import pandas as pd
import talib  as ta
import datetime


start = '2015-01-01'                       # 回测起始时间
end = '2015-12-30'                         # 回测结束时间
universe = DynamicUniverse('A')        # 证券池，支持股票、基金、期货、指数四种资产
benchmark = 'HS300'                        # 策略参考标准
freq = 'd'                                 # 策略类型，'d'表示日间策略使用日线回测，'m'表示日内策略使用分钟线回测
refresh_rate = 10                           # 调仓频率，表示执行handle_data的时间间隔，若freq = 'd'时间间隔的单位为交易日，若freq = 'm'时间间隔为分钟
  
# 配置账户信息，支持多资产多账户
accounts = {
    'fantasy_account': AccountConfig(account_type='security', capital_base=10000000)
}




Max_position_per = 0.1
Max_history_window = 250
Max_time_range = 30


atrlength = 20
avglength =20




def initialize(context):
    pass


def handle_data(context):  
    tradingDic = timing_CK(context)
    trading(tradingDic,context)
复制

import talib  as ta


def timing_CK(context):
    account =  context.get_account('fantasy_account')
    current_universe = context.get_universe(exclude_halt=True)
    security_position = account.get_positions()
    history = context.history(current_universe,['closePrice','lowPrice','highPrice'],Max_time_range,rtype= 'array')
    
    buy_list = []
    sell_list = []
    for sec in  current_universe:
        close = history[sec]['closePrice']
        low = history[sec]['lowPrice']
        high = history[sec]['highPrice']
     #数据处理 计算ATR通道上轨和止损点  计算三价均线  
        atr = ta.ATR(high,low,close,atrlength)[-1]
        movavgval = ta.MA((high+low+close)/3,avglength)
        upband = movavgval[-1] + atr
        liquidpoint = movavgval         #  三价均线向上 突破上轨做多 止损点为3价均线
        
        if  movavgval[-1] >movavgval[-2] and high[-1]>=upband and sec not in security_position:
            buy_list.append(sec)
        elif low[-1]<=  liquidpoint[-1] and sec in security_position:
            sell_list.append(sec)
            
    TradingDic = {'buy_list':buy_list,'sell_list':sell_list}
    return TradingDic
复制

def trading(TradingDic,context):
    account =  context.get_account('fantasy_account')
    current_universe = context.get_universe( exclude_halt=True)
    security_position = account.get_positions()
    cash = account.cash
    buy_list = TradingDic['buy_list']
    sell_list = TradingDic['sell_list']
    
    for sec in current_universe:
        if sec   in sell_list and sec in security_position:
            account.order_to(sec,0)
            cash += security_position[sec].amount * context.current_price(sec)
            
    if len(buy_list) >0:
        weight = min(Max_position_per, 1.0/len(buy_list))
    else:
        weight = 0
    for sec in buy_list:
        if sec not in security_position:
            account.order_pct_to(sec,weight)
复制