打板策略，直接在书上直接抄下来的代码，但复制上去一直说运行错误，我这个小白就...

狮子之心力交瘁 · 发表于 2024-2-14 21:23

追板逻辑

图是逻辑，我直接在书上直接抄下来的代码，但复制上去一直说运行错误，我这个小白就不懂了。
##连板打板量化交易策略代码实现
##开盘前运行函数
def before_market_open(context):
   #输出运行时间
   log.info('函数运行时间(before_market_open):
'+str(context.current_dt.time()))

   #更新一下股票池
   stock_pool(context)
   #更新账户股票的 str
   for code in context.portfolio.positions.keys():
      position= context.portfolio.positions[code]
      high_price=get_price(code,start_date=position.init_time,end_date=context.current_dt,frequency="1m",fields=['high'],skip_paused=True, fq="pre",count=None) ['high' ] .max ()
      if code not in g.cache_data.keys():
      g.sache_data[code]=dict()
   g.cache_data[code]['high_price']=high_price

   str = calc_history_str(code=code,end_time=get_last_time(position.init_time),timeperiod=str,_WINDOW,unit=LONG_UNIT)
   if code not in g.cache_data.keys ():
      g.cache_data[code] = dict()
   g.cache_data[code] ['str'] = art
##开盘时运行函数
def market_open(context)：
   buy(context)
   sell(context)

##收盘后运行函数
der after_market_close(context):
log.info(str('函数运行时间
(after_market_close):'+str(context.current_dt.time())))
#得到当天所有成交记录
trades = get_trades()
log.info("收盘时的账户记录:",str(context.portfolio.positions))
for_trade in trades.values():
      log.info('成交记录:'+str(_trade))

# if g.stock_pool_update_day % CHANGE_STOCK_POOL_DAY_NUMBER == 0:
      #更新股票池
stock_pools =set()
log.info('一天结束')
log.info('#################################################################')

def load_fundamentals_data(context):# :
      '''
      加载财务数据,选出市值大于10亿元到 350亿元的个股
      '''
      # name==get_all_securities(types=['stock'], date=None)
      df=get_fundamentals(query(valuation).filter (valuation.market_cap>10).filter(valuation.market_cap<350))
      return df['code'].tolist()

def buy(context):
      '''
买入逻辑，开仓前买入
      '''
      for code in g.stock_pool:
         if code in context.portfolio.positions.keys ():
            continue
         current_data=get_current_data ()[code]
         if current_data==None :
               return
         if is_high_limit(code):
               continue

         position_amount=calc_position(context,code)
         log.info("计算出来的仓位量:",position_amount)
         num=g.stockNum-len(context.portfolio.positions)

         if(num>0):
            order_=order_target(security=code,amount=position_amount)
            log.info("成交记录2024:",str(order_))
            log.info("当前的账户记录:",str(context.portfolio.positions))
            if((order_ is not None)and(order_ .filled>0)):
               log.info("成交了吗?")
               log.info("交易买入"，code,"成交均价",order.price，"买入的股数",order .filled)
               atr=calc_history_atr(code=code,end_time=get_last_time
(context.current_dt),timeperiod=ATR_WINDOW,unit=LONG_UNIT)
               if code not in g.cache_data.keys():
                  g.cache_data[code]= dict()
               g.cache_data[code]['atr']= atr
               g.cache_data[code]['high_price'] =
current_data.last_price
         g.bar_number=g.bar_number+1

g.stock_pool=[]
Pass

def is_high_limit(code):
   current_data=get_current_data()[code]
   if current_data.last_price>=current_data.high_limit:
      return True
   if current_data.paused:
      return True
   return False

def is_low_limit(code):
   current_data=get_current_data()[code]
   if current_data.last_price<=current_data.low_limit:
      return True
   if current_data.paused:
      return True
   return False
# m卖票策略
def sell (context):
   sell_list=list(context.portfollo.positions.keys () )
   if(len(sell_list)>0);
      for stock in sell_list:
         close_data1 = get_bars(stoak,count=1, unit='1m',
fields=['close'])[0]['close']
         cost=context.portfolio.positions[stock].acc_avg_cost
         close_data=attribute_history(stock,5,'1d',['close'])
         current_price=get_price(stock, start_date=None,
end_date=context.current_dt,frequency='1m', fields=['close'],
skip_paused=True, count=1).iloc[0]['cloce']
         pre_close=close_datal'close'][-1]
         if(current_price<cost*0.90):
            order_target(stock,0)
            log.info("亏本卖出: %s"%(stock))
         elif current_price>cost*1.20:
            if  (is_high_limit(stock)):
               continue
            order_target(stock,0)
            log.info("赚钱卖出:%s"%(stock))
   pass

def stop_loss(context):
'''
跟踪止损
'''
for code in context.portfolio.positions.keys():
      position = context.portfolio.positions[code]
      if position.closeable_amount <= 0:
         continue
      if is_low_limit(code):
         continue
      current_data = get_current_data()[code]
      if current_data = None :
         continue
      current_price =current_data.last_price

      #获取持仓期间最高价
      start_date= context.current_dt.strftime("%￥-%m-%d")+"00:00:00"
      #为防止发生 start_date 遭遇建仓时间，这里需要进行判断
      #当前时间和建仓时间在同一天时，atart_date设置为建仓时间
      if context.current_dt.strftime ("%￥%m-%d")<=
position.init_time.strftime("%￥-%m-%d"):
               start_date = position.init_time

      high_price = get_price(security=code, start_date=start_date,end_date=context.current_dt, frequency '1m', fields ['high'],skip_paused=True, fq='pre', count=None) [ 'high'] .max()
      #电每日9:30时，get_price获取00:00到09:30之间的最高价时，数据返回的NaN，需要特殊处理。这里采用当前价格和缓存的最高价进行比较
      if not np.isnan(high_price):
         high_price = max (high _price,g.cache_data[code]['high_price'])
      else:
         high_price =
max(current_price,g.cache_data[code]['high_price'])
      g.cache_data[code]['high_price']= high_price
      atr = g.cache_data[code]['atr']
      avg_cost =position.avg_cost

      if current_price <= high_price - atr * TRAILING_STOP_LOSS_ATR;
         # 当前价格小于等于最高价回撤 TRAILING_STOP_LOSS_ATR 倍 ATR，进行止损
卖出
      order_ =order_target(security=code, amount=0)
      if order_ is not None and order .filled>0:
         flag="WIN*" if current_price >avg_cost else "FAIL"
         log.info("交易卖出跟踪止损",
            code，
            "卖出数量",order_.filled,
            "当前价格",current_price,
            "持仓成本"，avg_cost，
            "最高价",high_price，
            "ATR",(atr *TRAILING_STOP_LOSS_ATR)
            "价差",(high_price - current_price)
            )
pass

def stock_pool(context):
   current_dt=context.current_dt,strftime("%￥-%m-%d" )
   codeList=load_fundamentals_data(context)
   current_datas=get_current_data()
   log.info("----股票池更新------")
   i=0
   for code in codelist:
   obestart=code[0:3]
   current_data=current_datas[code]

   '''
   交易日期
   '''
   trade_days=get_trade_days("2014-01-01",current_dt)
   yesterDay=trade_days[-2]
   if current_data.is_st:
      continue
   if current_data.paused:
      coatinve
   name=current_data.name
   if  'ST'  in name or  '*'  in name or  '退'  in name:
      coatinue

   if(codeStart=="300" or codeStart=-"301" or codeStart =="688"):
      continue

   # if (capFilter(code) is False):
   #  continue

   # log.info("昨日换手率:%s"%zrtr)
   if not (tr(code,context)):
      continue

   '''
   远行各种选股条件的判断
   '''
   price=get_price(code,count=30,end_date=current_dt,panel=False,fields=['close’,'open' ,'high' ,'low',' volume' , 'paused'])
   '''
   XG:(进2强势 OR 每日强势) AND 超预期 AND 清洗 AND 高开；
   '''
   if((j2qs (price) or mrgs(price)) and cyq(price,code,current_dt) and gaokai(price)):
   # if(qs(price)):
      log.info("日期",current_dt,"选出股:",code)
      g.stock_pool.append(code)

if(len(g.stock_pool)=-0):
   log.info("没有选到股票")
else:
   log.info("选出股票个数:",len(g.stock_pool))

   pass

def tr(code,context):
      current_dt=context.current_dt.strftime("%￥- %m - %d")
      tr = get_valuation(code,fields=["turnover_ratio" ],end_date=current_dt,count=4)
      # print(tr)
      #昨日换手率
      if(len(tr)>2):
         zrtr=tr.iloc[-2]['turnover_ratio']
      # print(zrtr)

            if((not zrtr is None) and zrtr>3):
               return True
            else:
                  return False
      else :
            return False

def gaokai(pricelist):
   currOpen=priceList.iloc[-1,1]
   preClose=priceList.iloc[-2,0]
   if(currOpen>=preClose*1.002 and currOpen<preClose *1.092):
      return True
   else
      return False

#定义黄金线
def hjx(code,current_dt,price):
   df2=price.iloc[-32:-2]
   gl=df2['high'].mean()
   return gl*(1+13/100)
#定义超预期 # 超预期:=REF(C,1)>黄金线*1.03;
def cyq(priceList,name,current_dt):
   gl=hjx(name,current_dt,priceList)
   if(priceList.iloc[-2,0]>g1*1.03):
         return True
   else:
         return False

#只要中小市值的股票
def caprilter(code):
   q = query(
               valuation.code,
               valuation,market_cap).filter(valuation,code==code)

   df = get_fundamentals (q)
   cap=df.iloc[0]['market_cap']
   if(cap>10 and cap<35):
      return True
   else:
      return False

def aal (price,i):
   b=False
   c=price.iloc[i,0] #收盘
   h=price.iloc[i,2] #high
   r=price .ilocli,0]/price.iloc[i-1,0]
#涨停返回 True
   if(r>1.094 and c==h):
      return True
   else
      return False
def vv(price,dp,i):
   b=False
   s=price.iloc[dp,4]/price.iloc[dp-i,4]
   if(s>1.2):
      # log.info(i,":",s)
      return True
   else:
      return False
#定义前日未涨停
def qrwzt(priceList):
   if aal(priceList,-3):
      return True
   else:
      return False
#定义进二强势
def j2qs (price):
   zrzt=aa1(price,-2)
   zrln=(vv(price,-2,1) or vv(price,-2,2))
   if(qrwzt(price) is False and zrzt and zrln):
      return True
   else:
      return False
#定义每日强势
def mrqs (price):
   qrzt=aal(price,-3)
   zrzt=aal(price,-2)
   zrln=(vv(price,-2,1) or vv(price,-2,2) )
   if(qrzt and zrzt and zrln):
         return True
   else:
         return False

#----------------------------------史出所需要的工具函教-----------
def calc_history_atr(code,end_time,timeperiod=14,unít='1d'):
   '''
   计算标的的 str 值
   Args :
         code 标的的编码
         end_time 计算ATR的时间点
         timeperiod 计算 ATR 的窗囗
         unit 计算ATR的bar的单位
   Returns :
         计算的标的在 end_time 的 ATR 值
   '''
   security_data = get_price(security=code, end_date=end_time,frequency=unit,fields=['close','high','low'], skip_paused=True,fq='pre',count=timeperiod+1)
   nan_count = list(np.isnan(security_data['close'])).count (True)
   if_nan_count==len(security_data['close']):
      log.info("股票 %s 输入数据全是 NaN，该股票可能已退市或刚上市，返回 NaN 值数据。”%stock)
      return np.nan
   else:
      return tl.ATR(np.array(security_data['high']),np.array(security_data['low']),np.array(security_data['close'])timeperiod)[-1]
   pass

def calc_position(context,code):
   '''
   计算建仓头寸依据:资金池每份现金*风险因子/波动率
   Args :
         context 上下文
         code 要计算的标的的代码
   Returns:
         计算得到的头寸，单位为股数
   '''
   #计算 risk_adjust_factor 用到的 sigma 的窗口大小
   RISK_WINDOW=60
   #计算 risk_adjust_factor 用到的两个 sigma 间隔大小
   RISK+DIFF = 30
   #计算 sigma 的窗口大小
   SIGMA_WINDOW=60

   #计算头寸需要用到的数据的数量
   count=RISK_WINDOW+RISK_DIFF*2
   count = max(SIGMA_WINDOW,count)
   history_values =get_price(security=code,end_date=get_last_time(context.current_dt), frequency=LONG_UNIT,fields=['close','high','low'],skip_paused=True,fq='pre', count=count)
   h_array= history values['high']
   l_array = history values['low']
   c_array = history values['close']
   log.info("当前现金:",context.portfolio.starting_cash)

   if(len(history_values.index)<count)or
(list(np.isnan(h_array)).count(True)>0)or
(list(np.isnan(l_array)).count(True)>0)or
(list(np.isnan(c_array)).count(True)>0):
      #数据不足或者数据错误存在 NaN
      return 0

   #数据转换
   value_array =[]
   for i in range(len(h_array)):
         value_array.append((h_array+ l_array+ c_array* 2)/ 4)

   first_sigma =
np.std(value_array[-RISK_WINDOW-(RISK_DIFF*2):-(RISK_DIFF*2)]) # -120:-60
   center_sigma =
np.std(value_array[-RISK_WINDOW-(RISK_DIFF*1):-(RISK_DIFF*1)]) #-90:-30
   last_sigma = np.std(value_array[-RISK_WINDOW :]) #-60:
   sigma =  np.std(value_array[-SIGMA_WINDOW:])
   risk_adjust_factor_ =0
   if last_siqma >center_sigma :
      risk_adjust_factor_ =0.5
   elif last_sigma < center_sigma and last_sigma > first_sigma:
         risk_adjust_factor_ =1.0
   elif last_sigma < center_sigma and last_sigma < first_sigma.
         risk_adjust_factor_ = 1.5
   return int (context.portfolio,starting_cash* 0.055 * risk_adjust_factor_ / (( POSITION_SIGMA * sigma)* 100))*100

xytiger · 发表于 2024-4-17 20:40

这是本什么书？

		自动登录	找回密码
密码