模拟北京抄家胜率60%的打炸板首版股票回测代码分享

显示全部楼层 · 2023-10-1 10:24:30

北炒是游资中打板一族的一个另类，他总是以打炸板股出名，今天给大家把回测代码也分享了，可以自己去调试优化，如果能卖在早盘10点前高点的80%的话，该策略胜率达到60%，盈亏比是1.54，3%的盈利占比达到26.13%。当然人家不是所有炸板股都买的，有自己的判断，我这里是假设对所有炸板回封股都买的结果。比如北炒在23-8-17日买入重庆啤酒的买点：

模拟北京抄家胜率60%的打炸板首版股票回测代码分享-1.jpg

这是他在23-8-18的交易记录：

模拟北京抄家胜率60%的打炸板首版股票回测代码分享-2.jpg

这是对所有主板股票回测的代码分享：
需要注意的是，上述策略的回测过程未计算滑点对策略收益产生的影响。且过去的表现并不能保证未来的表现，市场风险和不确定性永远存在。后面我会根据自己优化后的策略，用5万元开展实盘来验证这个策略的价值，欢迎大家关注，如果持仓标的发生变动会给大家更新。import csv
import os
import time

import backtrader as bt
import numpy as np
import pandas as pd

from SelfTrade import SelfTrade
import datetime

'''炸首板再次打板策略
1. 当日涨停过，又炸板了，再次快要涨停时候买入（可以尝试炸首板，或者近期涨幅在一定范围内的炸板）
2。第二日早上冲高回落卖出，下跌反弹后卖出，一般10点前挑高点卖出，如果不涨停的话
'''
class ZhaBanSelect(bt.Strategy):
def log(self, Type, Code, PriceType, Price, Date):
      ''' 策略的日志函数'''
      with open(self.logFile, 'a', newline='', encoding='utf-8') as file:
         writer = csv.writer(file)
         writer.writerow([Type, Code, PriceType, Price, Date])
      print('%s, %s, %s, %s, %s' % (Type, Code, PriceType, Price, Date))

params = dict(
      hostDays=1,  # 持有最大天数
      UP_LIMIT = 0.099,  # 涨停标准
      UP_LIMIT_BUY=1.0981,  # 买入起始价位
      START_TIME='9:30:00',
      END_TIME='10:00:00',
      Min_FOLDER=r'/Users/Downloads/tdx/min',

      # 波动低
      LOW_WAVE_RATE=0.3,  # 波动幅度
      LOW_WAVE_PERIOD=30,
)

def __init__(self):
      self.start_dates = {}
      self.start_dates_flag = {}
      self.diff_start_dates_days = {}
      self.selfTrader = SelfTrade(commission_rate=0.0007, cash=10000000)
      self.candidates1 = {}  # 进入第一波候选观察股池
      self.positionsMy = {}  # 持仓股票及其买入价格
      self.days_since_in_host= {}  # 记录每只股票持有后
      self.days_since_in_pool= {}
      self.inds = dict()
      self.first_day_traded = {}

      # 首版
      self.candidatesFirstBan = {}  # 进入候选观察股池
      self.days_since_in_pool_firstBan = {}  # 记录每只股票买入后的交易日数

      # 波动小
      self.candidatesLowWave = {}  # 进入候选观察股池
      self.days_since_in_pool_lowWave = {}  # 记录每只股票买入后的交易日数

      # 构建文件名
      filename = f"首板打炸板持有{self.p.hostDays}天早盘非涨停卖出策略_tdx2023_全部.csv"
      self.logFile = "../out/"+ filename

      with open(self.logFile, 'w', newline='', encoding='utf-8') as file:
         writer = csv.writer(file)
         writer.writerow(['Type', 'Code', 'PriceType', 'Price', 'Date'])

      self.profit_trades = 0  # 交易盈利次数
      self.profit_3percent = 0  # 交易盈利3%比例
      self.stop_loss_trades = 0  # 割肉次数
      self.profit_pct = 0.0  # 盈利时的平均股价涨幅
      self.stop_loss_pct = 0.0  # 割肉时的平均股价跌幅
      self.profit_trades_open = 0  # 交易盈利次数
      self.stop_loss_trades_open = 0  # 割肉次数
      self.profit_pct_open = 0.0  # 盈利时的平均股价涨幅
      self.stop_loss_pct_open = 0.0  # 割肉时的平均股价跌幅
      self.count = 0.0  # 统计资金不足次数

      to_remove = []  # 用于存储需要移除的数据对象
      for i, d in enumerate(self.datas):
         if len(d.array) < self.params.LOW_WAVE_PERIOD:  # 检查数据长度是否足够
            #    print(d._name + '=' + str(len(d.array)))
            to_remove.append(i)  # 将需要移除的数据对象添加到临时列表
         else:
            self.inds[d] = dict()
            self.inds[d]['lowesstWave'] = bt.ind.Lowest(d.low, period=self.params.LOW_WAVE_PERIOD)
            self.inds[d]['highestWave'] = bt.ind.Highest(d.high, period=self.params.LOW_WAVE_PERIOD)

      # 移除数据对象
      for i in reversed(to_remove):
         self.datas.pop(i)  # 使用pop()方法移除指定索引处的数据对象

      # 检查剩余的数据对象数量
      if len(self.datas) == 0:
         print("所有数据对象已移除，无法运行策略")
         self.stop()  # 中止策略的执行

def prenext(self):
      self.next()

def notify_trade(self, trade):
      if trade.isclosed:
         if trade.pnlcomm > 0:
            self.profit_trades_open += 1
            self.profit_pct_open += trade.pnlcomm
         else:
            self.stop_loss_trades_open += 1
            self.stop_loss_pct_open += trade.pnlcomm

      if self.selfTrader.isclose(symbol= trade.data._name):
         if self.selfTrader.positions[trade.data._name]['pnl'] > 0:
            self.profit_trades += 1
            self.profit_pct += self.selfTrader.positions[trade.data._name]['pnl']
            self.log("盈利", trade.data._name, "收益", self.selfTrader.positions[trade.data._name]['pnl'], trade.data.datetime.date(-1))
            if self.selfTrader.positions[trade.data._name]['pnl'] >= 300:
                  self.profit_3percent += 1
         else:
            if self.selfTrader.positions[trade.data._name]['pnl'] > -2500:
                  self.stop_loss_trades += 1
                  self.stop_loss_pct += self.selfTrader.positions[trade.data._name]['pnl']
                  self.log("割肉", trade.data._name, "损失", self.selfTrader.positions[trade.data._name]['pnl'], trade.data.datetime.date(-1))
            else:
                  self.log("可能有除权", trade.data._name, "问题数据", self.selfTrader.positions[trade.data._name]['pnl'], trade.data.datetime.date(-1))

def stop(self):
      if self.profit_trades > 0:
         self.profit_pct /= self.profit_trades
      if self.stop_loss_trades > 0:
         self.stop_loss_pct /= self.stop_loss_trades

      if self.profit_trades_open > 0:
         self.profit_pct_open /= self.profit_trades_open
      if self.stop_loss_trades_open > 0:
         self.stop_loss_pct_open /= self.stop_loss_trades_open
      # 打印交易统计信息
      print("交易盈利次数:", self.profit_trades)
      print("割肉次数:", self.stop_loss_trades)
      print("资金不足次数:", self.count)
      print("盈利时的平均盈利金额: %.2f%%" %self.profit_pct)
      print("割肉时的平均亏损金额: %.2f%%" %self.stop_loss_pct)
      print('Final Portfolio Value by Close trade: %.2f' % self.selfTrader.get_total_assets())
      if self.profit_trades + self.stop_loss_trades > 0:
         percent = self.profit_3percent / (self.profit_trades + self.stop_loss_trades) * 100
         print(f'盈利超过3%的比例: {percent:.2f}%')
      self.log(self.profit_trades, self.stop_loss_trades, self.profit_pct, self.stop_loss_pct, self.selfTrader.get_total_assets())

      current_date = datetime.datetime.now().strftime("%Y-%m-%d")
      outFile = f'../out/candidates{current_date}.txt'
      # 检查文件output.csv是否存在
      if not os.path.exists(outFile):
         with open(outFile, mode='w', newline='', encoding='utf-8') as file:
            writer = csv.writer(file)
            writer.writerow(['代码', 'TargetPrice', 'Type', 'LastClose'])

      with open(outFile, mode='a', newline='', encoding='utf-8') as file:
         writer = csv.writer(file)
         for stock in self.candidatesFirstBan:
            if stock not in self.positionsMy:
                  writer.writerow([stock._name[:-4], round(self.candidatesFirstBan[stock][1] * 1.1 ,2), '炸板',self.candidatesFirstBan[stock][1]])

      allStockList = r'../in/all_stocks.csv'
      df_all = pd.read_csv(allStockList, header=0, na_values='NA')
      df = pd.read_csv(outFile, header=0, na_values='NA')
      merged_df = pd.merge(df, df_all, on='代码', how='left')
      out_df = merged_df[['代码', 'TargetPrice', 'Type', 'LastClose', '名称', '市盈率', '概念']]
      out_df.drop_duplicates(subset=['代码'], keep='first', inplace=True)
      out_df.to_csv('../out/ZhaBancandidates.csv', mode='w', header=True, index=False)

def highIntervl(self, data, start_time, end_time):
      stock_name = data._name[2:]  # 获取股票名
      stock_data_path = os.path.join(self.p.Min_FOLDER, stock_name)
      if os.path.isfile(stock_data_path):
         df = pd.read_csv(stock_data_path, parse_dates=['date'])
         # 设置筛选条件
         next_date = data.datetime.date(0)
         next_date_data = df[df['date'].dt.date == next_date]
         while next_date_data.empty:
            next_date += datetime.timedelta(days=1)
            if next_date >= datetime.datetime.now().date():
                  self.log("异常：可能停牌", data._name, "价格", "", data.datetime.date(0))
                  break
            next_date_data = df[df['date'].dt.date == next_date]
         condition = (next_date_data['date'].dt.time > pd.to_datetime(start_time).time()) & (next_date_data['date'].dt.time <= pd.to_datetime(end_time).time())
         # 根据条件筛选数据
         filtered_df = next_date_data[condition]

         # 获取筛选后数据中 'high' 列的最大值
         max_high = filtered_df['high'].max()
         low_min = filtered_df['low'].min()

         return (max_high, low_min)
      else:
         raise f"{stock_data_path} is not exit"

def LowWaveSelect(self, data):
      if not self.upLimit(self.inds[data]['highestWave'][0], self.inds[data]['lowesstWave'][0], self.p.LOW_WAVE_RATE):
         self.candidatesLowWave[data] = data.close[0]  #
      #       self.log("添加lowest后选股池", data._name, "价格", data.close[0], data.datetime.date(0))

      # 判断是否需要移除候选观察股池中的股票，在股票池超过stock_day个交易日
      if data in self.candidatesLowWave:
         if self.upLimit(self.inds[data]['highestWave'][0], self.inds[data]['lowesstWave'][0], self.p.LOW_WAVE_RATE):
            self.candidatesLowWave.pop(data, None)

#          self.log("移除Lowest后选股池", data._name, "价格", data.close[0], data.datetime.date(0))

def FisrtBanSelect(self, data, prev_close):
      if data in self.candidatesFirstBan:
         self.days_since_in_pool_firstBan[data] += 1

      current_date = self.data.datetime.date()
      if self.upLimit(data.high[0],  prev_close, self.p.UP_LIMIT) and not self.upLimit(data.high[-1],  data.close[-2], self.p.UP_LIMIT):  # 乘以1.0998相当于增长9.98%
         stock_name = data._name[2:]  # 获取股票名
         if stock_name not in self.candidatesFirstBan:
            self.candidatesFirstBan[data] = (current_date, prev_close)
            self.days_since_in_pool_firstBan[data] = 0
   #          self.log("添加首版选股池", data._name, "价格", data.close[0], data.datetime.date(0))

      if data in self.candidatesFirstBan:
         if self.days_since_in_pool_firstBan[data] > 0:
            self.candidatesFirstBan.pop(data, None)
            self.days_since_in_pool_firstBan.pop(data, None)
#             self.log("移除首版选股池", data._name, "价格", data.close[0], data.datetime.date(0))

def upLimit(self, currentPrice, lastClose, upLimit):
      if currentPrice >= round(upLimit * lastClose + lastClose,2) :
         return True
      else:
         return False

def downLimit(self, currentPrice, lastClose, downLimit):
      if currentPrice - lastClose < downLimit * lastClose:
         return True
      else:
         return False

def next(self):
      for data in self.datas:
         if data not in self.first_day_traded:
            self.first_day_traded[data] = True
            self.diff_start_dates_days[data] = 0
            continue  # 跳过第一天交易判断

         if data not in self.start_dates:
            self.start_dates[data] = data.datetime.date(0)
            self.start_dates_flag[data] = False
            continue
         if (data.datetime.date(0) - self.start_dates[data]).days != 0 and  not self.start_dates_flag[data]:
            self.start_dates_flag[data] = True
            self.start_dates[data] = data.datetime.date(0)
            continue
         # 当股票当前日大于起始日30天以上才进行计算，保证新股上市30天后才计算，
         if self.start_dates_flag[data] and (data.datetime.date(0) - self.start_dates[data]).days > 30 \
                  and (data.datetime.date(0) - self.start_dates[data]).days > self.diff_start_dates_days[data]:
            self.diff_start_dates_days[data] = (data.datetime.date(0) - self.start_dates[data]).days
            #设置前一日价格
            dividend_flag = False
            prev_close = data.close[-1]
            if data.open[0] - prev_close < -0.11 * prev_close and data._name.startswith(('sh60', 'sz00')) or \
                     data.open[0] - prev_close < -0.21 * prev_close and data._name.startswith(('sh68', 'sz30')):
                  dividend_adjustment = prev_close / data.open[0]
                  prev_close /= dividend_adjustment
                  dividend_flag = True

            self.FisrtBanSelect(data, prev_close)
            self.LowWaveSelect(data)

            # 判断是否需要卖出持仓股票
            if data in self.positionsMy:
                  self.days_since_in_host[data] += 1
                  #除权发生调整成本
                  if dividend_flag:
                     self.selfTrader.positions[data._name]['price'] /= dividend_adjustment
                     self.selfTrader.positions[data._name]['quantity'] *= dividend_adjustment
                     self.log("除权发生", data._name, "调整成本价", self.selfTrader.positions[data._name]['price'], data.datetime.date(0))

                  stock_name = data._name[2:]  # 获取股票名
                  stock_data_path = os.path.join(self.p.Min_FOLDER, stock_name)
                  if os.path.isfile(stock_data_path):
                     df = pd.read_csv(stock_data_path, parse_dates=['date'], index_col='date')
                     next_date = data.datetime.date(0)
                     next_date_data = df[df.index.date == next_date]
                     while next_date_data.empty:
                        next_date += datetime.timedelta(days=1)
                        if next_date >= datetime.datetime.now().date():
                              self.log("异常：可能停牌", data._name, "价格", "",  data.datetime.date(0))
                              break
                        next_date_data = df[df.index.date == next_date]

                     highInterval = self.highIntervl(data, self.p.START_TIME, self.p.END_TIME)
                     close_price = (highInterval[0] - highInterval[1]) * 0.8 + highInterval[1]

                     if not self.upLimit(close_price, prev_close, self.p.UP_LIMIT):
                        sell_price = close_price
                        self.log("早盘卖", data._name, "价格", sell_price, next_date.strftime('%Y-%m-%d'))
                        self.close(data=data, price=sell_price)
                        self.selfTrader.close(symbol=data._name, price=sell_price)
                        self.positionsMy.pop(data, None)
                        # 必须要在卖出股票后移除candidates1股池，否则，卖出当天有可能还会进入买入逻辑，从而导致当天无法完成结算，而且反包规则卖出当天都不符合买入准则
                        self.candidatesFirstBan.pop(data, None)
                     else:
                        self.log("涨停不卖", data._name, "涨停价格", data.close[0], data.datetime.date(0))

            # 判断是否在候选观察股池中，买入炸板股票根据规则
            if data in self.candidatesFirstBan and data in self.candidatesLowWave:
                  stock_name = data._name[2:]  # 获取股票名
                  stock_data_path = os.path.join(self.p.Min_FOLDER, stock_name)
                  if os.path.isfile(stock_data_path):
                     df = pd.read_csv(stock_data_path, parse_dates=['date'], index_col='date')
                     df = df[df.index.date == self.candidatesFirstBan[data][0]]  # 选择日期部分等于current_date的数据
                     firstUpFlag = True
                     secondUpFlag = True
                     for index, row in df.iterrows():
                        high = row['high']
                        low = row['low']
                        datevalue = index.time()

                        if self.upLimit(high, prev_close, self.p.UP_LIMIT) and firstUpFlag and secondUpFlag:
                              # 首次涨停
                              firstUpFlag = False
                              continue
                        if self.upLimit(high, prev_close, self.p.UP_LIMIT) and not firstUpFlag and secondUpFlag and low < high:
                              # 打开涨停版
                              secondUpFlag = False
                              continue
                        if not firstUpFlag and self.upLimit(high, prev_close, self.p.UP_LIMIT) and low < high and not secondUpFlag :
                              # 再次准备涨停
                              cash = self.broker.getcash()
                              if cash < 10000:
                                 self.log("资金不足", data._name, "持有股数量", len(self.positionsMy),
                                          data.datetime.date(0))
                                 self.count = self.count + 1
                              elif data not in self.positionsMy:
                                 buyPrice = round(prev_close * 1.1 ,2)
                                 sizeOpen = 10000 / buyPrice
                                 self.buy(data=data, size=sizeOpen)
                                 self.positionsMy[data] = buyPrice
                                 self.log("买入股票", data._name, "买入价格", buyPrice, index.strftime('%Y-%m-%d %H:%M:%S'))
                                 self.days_since_in_host[data] = 1
                                 sizeClose = 10000 / buyPrice
                                 self.selfTrader.buy(symbol=data._name, price=buyPrice, quantity=sizeClose)
                              break

now = datetime.datetime.now()
one_day = datetime.timedelta(days=1)
next_day = now + one_day
class MyData(bt.feeds.GenericCSVData):
params = (
      ('dtformat', '%Y-%m-%d'),
      ('datetime', 0),
      ('open', 1),
      ('high', 2),
      ('low', 3),
      ('close', 4),
      ('volume', 6),
      ('openinterest', -1),
      ('fromdate', datetime.datetime(2022, 12, 1)),
      ('todate', next_day),
)

if __name__ == '__main__':
time_start = time.time()
cerebro = bt.Cerebro(stdstats=False, cheat_on_open=False)

# 添加数据源
dataFolder = r'/Users/Downloads/tdx/baostock'
allStockList = r'../in/all_stocks.csv'
df_all = pd.read_csv(allStockList, header=0, na_values='NA')

stock_lists = os.listdir(dataFolder)

# stock_lists = ['sh600132.csv']
for stock in stock_lists:
      name = stock.replace(".csv", "")
      new_df = df_all[df_all['代码'] == name]
      if len(new_df) > 0:
         first_row = new_df.iloc[0]
         if 'S' not in first_row['名称'] and '*' not in first_row['名称'] and '退' not in first_row['名称'] and 'X' not in first_row['名称']\
                  and 'D' not in first_row['名称'] and 'C' not in first_row['名称'] and 'U' not in first_row['名称'] :
#             and first_row['市盈率'] > 0 and first_row['市盈率'] < 100:
            filepath = os.path.join(dataFolder, f'{stock}')
            size = os.path.getsize(filepath)
            if size > 10 * 200:
                  if stock.startswith(('sh60', 'sz00')):
                     df = pd.read_csv(filepath)
                     df['date'] = pd.to_datetime(df['date'], format='%Y-%m-%d')
                     last_row = df.tail(1)
                     if last_row['date'].values[0] > np.datetime64('2023-05-18'):
                        cerebro.adddata(MyData(dataname=filepath),  name = stock)

# 添加交易记录
cerebro.broker.setcash(10000000.0)
cerebro.broker.setcommission(commission=0.0007)

cerebro.addstrategy(ZhaBanSelect)

cerebro.run()
time_end = time.time()

print('程序所耗时间：', time_end - time_start)