昨日2020年03月12日时美股历史的第三次熔断，距离上一次熔断03月09号仅3天时间，受全球疫情的影响，股市出现恐慌性抛盘，导致A股也出现较为明显的震荡，A股能否走出独立行情还是不得而知了。
所以我最近学习了一下量化投资的东西，
量化平台Backtrader用起来还不错，A股历史行情数据可从Tushare中获取：
# Create a subclass of Strategy to define the indicators and logic
class SmaCross(bt.Strategy):
# list of parameters which are configurable for the strategy
params = dict(
pfast=10, # period for the fast moving average
pslow=30 # period for the slow moving average
)
def __init__(self):
super().__init__()
sma1 = bt.ind.SMA(period=self.p.pfast) # fast moving average
sma2 = bt.ind.SMA(period=self.p.pslow) # slow moving average
self.crossover = bt.ind.CrossOver(sma1, sma2) # crossover signal
def next(self):
if not self.position: # not in the market
if self.crossover > 0: # if fast crosses slow to the upside
self.order_target_size(target=1) # enter long
# self.buy()
elif self.crossover < 0: # in the market & cross to the downside
self.order_target_size(target=0) # close long position
# self.close()
这是一个简单的均值策略，当10日均线超过30日均线时买入，30日均线超过10日均线卖出。
data_path = './data/'
if not os.path.exists(data_path):
os.makedirs(data_path)
mytoken='your_token'
class Strategy_runner:
def __init__(self, strategy, ts_code, start_date, end_date, data_path=data_path, pro=False, token=mytoken):
self.ts_code = ts_code
self.start_date = start_date
self.end_date = end_date
# convert to datetime
self.start_datetime = datetime.strptime(start_date,'%Y%m%d')
self.end_datetime = datetime.strptime(end_date,'%Y%m%d')
if pro:
csv_name = f'pro_day_{str(ts_code)}-{str(start_date)}-{str(end_date)}.csv'
else:
csv_name = f'day_{str(ts_code)}-{str(start_date)}-{str(end_date)}.csv'
csv_path = os.path.join(data_path,csv_name)
if os.path.exists(csv_path):
if pro:
self.df = pd.read_csv(csv_path)
else:
self.df = pd.read_csv(csv_path,index_col=0)
else:
if pro:
ts.set_token(mytoken)
self.pro = ts.pro_api()
self.df = self.pro.daily(ts_code=self.ts_code, start_date=self.start_date, end_date=self.end_date)
if not self.df.empty:
self.df.to_csv(csv_path, index=False)
else:
self.df = ts.get_hist_data(self.ts_code, str(self.start_datetime), str(self.end_datetime))
if not self.df.empty:
self.df.to_csv(csv_path, index=True)
self.df_bt = self.preprocess(self.df, pro)
print(self.df_bt)
self.strategy = strategy
self.cerebro = bt.Cerebro()
def preprocess(self, df, pro=False):
if pro:
features=['open','high','low','close','vol','trade_date']
# convert_datetime = lambda x:datetime.strptime(x,'%Y%m%d')
convert_datetime = lambda x: pd.to_datetime(str(x))
df['trade_date'] = df['trade_date'].apply(convert_datetime)
print(df)
bt_col_dict = {'vol':'volume','trade_date':'datetime'}
df = df.rename(columns=bt_col_dict)
df = df.set_index('datetime')
# df.index = pd.DatetimeIndex(df.index)
else:
features=['open','high','low','close','volume']
df = df[features]
df['openinterest'] = 0
df.index = pd.DatetimeIndex(df.index)
df = df[::-1]
return df
def run(self):
data = bt.feeds.PandasData(dataname=self.df_bt,
fromdate=self.start_datetime,
todate=self.end_datetime)
self.cerebro.adddata(data) # Add the data feed
self.cerebro.addstrategy(self.strategy) # Add the trading strategy
self.cerebro.broker.setcash(100000.0)
# self.cerebro.addsizer(bt.sizers.FixedSize, stake=10)
# self.cerebro.broker.setcommission(commission=0.0)
self.cerebro.addanalyzer(bt.analyzers.SharpeRatio,_name = 'SharpeRatio')
self.cerebro.addanalyzer(bt.analyzers.DrawDown, _name='DW')
self.results = self.cerebro.run()
strat = self.results[0]
print('Final Portfolio Value: %.2f' % self.cerebro.broker.getvalue())
print('SR:', strat.analyzers.SharpeRatio.get_analysis())
print('DW:', strat.analyzers.DW.get_analysis())
return self.results
def plot(self, iplot=False):
self.cerebro.plot(iplot=iplot)
Tushare有pro版本和普通行情，由于最开始我使用了pro版本但是后面出现了一些连接问题，所以我又写了普通的版本，目前都可以用。
pro：
ts.set_token(mytoken)
self.pro = ts.pro_api()
self.df = self.pro.daily(ts_code=self.ts_code, start_date=self.start_date, end_date=self.end_date)
pro版本需要获取token，这个需要你去官网注册之后可以看到的，之后使用daily即可获取日级历史行情。
ts_code='600515.SH'
start_date='20190101'
end_date='20191231'
strategy_runner = Strategy_runner(strategy=SmaCross, ts_code=ts_code, start_date=start_date, end_date=end_date, pro=True)
results = strategy_runner.run()
strategy_runner.plot()
以去年为例，传入pro=True，即可使用pro接口，然后获取行情，用均值策略分析
添加策略：
self.cerebro.addstrategy(self.strategy)
设置初始资金：
self.cerebro.broker.setcash(100000.0)
加入analyzer为获取夏普率和回撤率：
self.cerebro.addanalyzer(bt.analyzers.SharpeRatio,_name = 'SharpeRatio')
self.cerebro.addanalyzer(bt.analyzers.DrawDown, _name='DW')
运行模拟策略一遍：
self.cerebro.run()
得到分析结果：
Final Portfolio Value: 100000.29
SR: OrderedDict([('sharperatio', None)])
DW: AutoOrderedDict([('len', 178), ('drawdown', 0.0014899734784627702), ('moneydown', 1.4899999999906868), ('max', AutoOrderedDict([('len', 178), ('drawdown', 0.0019099660025940943), ('moneydown', 1.9099999999889405)]))])
因为夏普率默认需要年化的，这里不到一年的数据所以是None。
如果没有token可以直接使用普通接口：
ts_code='600515'
start_date='20190101'
end_date='20191231'
strategy_runner = Strategy_runner(strategy=SmaCross, ts_code=ts_code, start_date=start_date, end_date=end_date, pro=False)
results = strategy_runner.run()
如果使用日级行情，其实两者并没有太大的差别，只是表格的列名有点差别。我这里如果第一次获取之后我会在本地存下该表格，后面就不需要网络获取了。