基于Baostock和本地存储的HDF数据构建低价股、资金流和小市值策略回测框架
基于Baostock和本地存储的HDF数据构建低价股、资金流和小市值策略回测框架
本文提供一套基于Baostock和本地存储的HDF数据构建的回测框架，用于测试低价股、资金流和小市值策略。框架支持买卖限制、增发考虑、分板块计算收益和轮动等功能，并包含数据读取类、选股策略类和回测执行类。
步骤及代码如下：
1. 数据读取类
数据读取类主要用于从Baostock下载数据，并将数据存储为HDF格式。代码如下：
import baostock as bs
import pandas as pd
class DataLoader:
def __init__(self, start_date, end_date):
self.start_date = start_date
self.end_date = end_date
def download_data(self, stock_list):
'从Baostock下载数据'
bs.login()
for stock in stock_list:
rs = bs.query_history_k_data_plus(stock, 'date,open,high,low,close,volume,amount',
start_date=self.start_date, end_date=self.end_date,
frequency='d', adjustflag='3')
data_list = []
while (rs.error_code == '0') & rs.next():
data_list.append(rs.get_row_data())
if len(data_list) > 0:
df = pd.DataFrame(data_list, columns=rs.fields)
df.set_index('date', inplace=True)
df.to_hdf(stock+'.h5', key='df', mode='w')
bs.logout()
def read_data(self, stock_list=None):
'从HDF读取数据'
if stock_list is None:
# 读取所有股票数据
pass
else:
# 读取指定股票数据
data = {}
for stock in stock_list:
df = pd.read_hdf(stock+'.h5')
data[stock] = df
return data
2. 选股策略类
选股策略类主要用于实现低价股、资金流和小市值策略。代码如下：
class StockPicker:
def __init__(self, data):
self.data = data
def low_price_strategy(self, price):
'低价股策略，返回价格小于等于price的股票'
selected_stock = []
for stock, df in self.data.items():
if df.iloc[-1]['close'] <= price:
selected_stock.append(stock)
return selected_stock
def capital_flow_strategy(self, n):
'资金流策略，返回最近n天资金流入前n的股票'
selected_stock = []
for stock, df in self.data.items():
net_amount = df.iloc[-n:]['amount'].apply(float).sum()
if net_amount > 0:
selected_stock.append((stock, net_amount))
selected_stock = sorted(selected_stock, key=lambda x:x[1], reverse=True)[:n]
return [s[0] for s in selected_stock]
def small_cap_strategy(self, market_cap):
'小市值策略，返回市值小于等于market_cap的股票'
selected_stock = []
for stock, df in self.data.items():
if df.iloc[-1]['close'] * df.iloc[-1]['outstanding_share'] <= market_cap * 100000000:
selected_stock.append(stock)
return selected_stock
3. 回测执行类
回测执行类主要用于实现买卖限制、考虑增发、支持分板块计算收益、支持轮动等功能。代码如下：
class BackTester:
def __init__(self, data, start_date, end_date, capital, commission_rate):
self.data = data
self.start_date = start_date
self.end_date = end_date
self.capital = capital
self.commission_rate = commission_rate
self.holdings = {}
self.cash = capital
self.stock_prices = {}
self.stock_returns = {}
def run_backtest(self, selected_stock, board=None, frequency='d', rotate=None):
'运行回测'
if board is None:
# 统计所有股票的收益
pass
else:
# 统计指定板块的收益
selected_stock = [s for s in selected_stock if self.get_board(s) == board]
if frequency == 'd':
dates = pd.date_range(self.start_date, self.end_date)
elif frequency == 'w':
dates = pd.date_range(self.start_date, self.end_date, freq='W-FRI')
elif frequency == 'm':
dates = pd.date_range(self.start_date, self.end_date, freq='M')
else:
raise ValueError('Unsupported frequency')
for date in dates:
if rotate is not None:
# 按月轮动
if date.day == 1:
selected_stock = self.rotate(selected_stock, board, rotate)
elif date.day > 5:
continue
for stock in selected_stock:
df = self.data[stock]
if date not in df.index:
continue
price = self.stock_prices.get(stock, None)
if price is None:
price = df.loc[date]['open']
self.stock_prices[stock] = price
if stock in self.holdings:
# 卖出
if df.loc[date]['low'] <= price * 0.9:
sell_price = price * 0.9
sell_amount = self.holdings[stock]
self.cash += sell_price * sell_amount * (1 - self.commission_rate)
self.holdings[stock] = 0
self.stock_prices[stock] = None
elif df.loc[date]['high'] >= price * 1.1:
sell_price = price * 1.1
sell_amount = self.holdings[stock]
self.cash += sell_price * sell_amount * (1 - self.commission_rate)
self.holdings[stock] = 0
self.stock_prices[stock] = None
else:
pass
else:
# 买入
if self.cash <= 0:
continue
if df.loc[date]['low'] <= price * 1.1:
buy_price = price * 1.1
buy_amount = int(self.cash / buy_price / 100) * 100
if buy_amount == 0:
continue
self.cash -= buy_price * buy_amount * (1 + self.commission_rate)
self.holdings[stock] = buy_amount
elif df.loc[date]['high'] >= price * 0.9:
buy_price = price * 0.9
buy_amount = int(self.cash / buy_price / 100) * 100
if buy_amount == 0:
continue
self.cash -= buy_price * buy_amount * (1 + self.commission_rate)
self.holdings[stock] = buy_amount
else:
pass
self.stock_returns[date] = self.get_portfolio_return(selected_stock, date)
def get_portfolio_return(self, selected_stock, date):
'计算组合收益'
total_value = self.cash
for stock, amount in self.holdings.items():
if amount == 0:
continue
df = self.data[stock]
if date not in df.index:
continue
if self.stock_prices[stock] is None:
continue
total_value += df.loc[date]['close'] * amount
total_value += df.loc[date]['incr_holding_shares'] * df.loc[date]['close'] * amount
return (total_value - self.capital) / self.capital
def get_board(self, stock):
'获取股票所在板块'
if stock.startswith('688'):
return '科创板'
elif stock.startswith('300'):
return '创业板'
elif stock.startswith('00'):
return '深市主板'
elif stock.startswith('60'):
return '沪市主板'
else:
return None
def rotate(self, selected_stock, board, rotate):
'轮动策略'
if board is None:
# 所有股票轮动
pass
else:
# 按板块轮动
selected_stock = [s for s in selected_stock if self.get_board(s) == board]
prev_month = (datetime.datetime.strptime(self.start_date, '%Y-%m-%d') + relativedelta(months=rotate-1)).strftime('%Y-%m-%d')
prev_data = self.data.copy()
for stock in selected_stock:
prev_df = pd.read_hdf(stock+'.h5', where='date>='%s' and date<'%s'' % (prev_month, self.start_date))
prev_data[stock] = prev_df.append(self.data[stock])
prev_backtester = BackTester(prev_data, prev_month, self.end_date, self.capital, self.commission_rate)
prev_backtester.run_backtest(selected_stock, board, frequency='d', rotate=None)
selected_stock = sorted(selected_stock, key=lambda x:prev_backtester.stock_returns[x], reverse=True)
return selected_stock[:5]
def plot_returns(self):
'绘制收益曲线'
returns = pd.Series(self.stock_returns)
returns.plot(figsize=(10, 6), title='Portfolio Returns')
plt.show()
4. 示例代码
假设我们要运行低价股策略，资金流策略和小市值策略，回测时间为2020-01-01到2021-01-01，初始资金为100万，交易佣金为0.1%。首先需要从Baostock下载数据：
data_loader = DataLoader('2020-01-01', '2021-01-01')
data_loader.download_data(['000001.SH', '000300.SH', '399001.SZ', '399006.SZ'])
data = data_loader.read_data(['000001.SH', '000300.SH', '399001.SZ', '399006.SZ'])
# 运行低价股策略
stock_picker = StockPicker(data)
selected_stock = stock_picker.low_price_strategy(5)
backtester = BackTester(data, '2020-01-01', '2021-01-01', 1000000, 0.001)
backtester.run_backtest(selected_stock)
backtester.plot_returns()
# 运行资金流策略
selected_stock = stock_picker.capital_flow_strategy(10)
backtester = BackTester(data, '2020-01-01', '2021-01-01', 1000000, 0.001)
backtester.run_backtest(selected_stock)
backtester.plot_returns()
# 运行小市值策略
selected_stock = stock_picker.small_cap_strategy(10)
backtester = BackTester(data, '2020-01-01', '2021-01-01', 1000000, 0.001)
backtester.run_backtest(selected_stock)
backtester.plot_returns()
本框架提供了基本的策略回测功能，用户可以根据自身需求进行扩展。例如，添加新的选股策略、增加交易成本模型、优化回测逻辑等等。
原文地址: https://www.cveoy.top/t/topic/nLB5 著作权归作者所有。请勿转载和采集!