• 日期: 2024-11-20
  • 标签: 常规

基于 Baostock 和本地存储 HDF 数据的低价股、资金流和小市值策略回测框架

本框架使用 Baostock API 从本地 HDF 文件读取股票数据，并实现低价股、资金流和小市值三种选股策略。支持按日、周、月轮动，并考虑 A 股涨跌停限制、增发影响和不同板块收益计算。

步骤

  1. 数据读取类: 使用 Baostock API 从本地 HDF 文件读取股票数据，可按天读取所有股票的数据，或者按股票名称读取所有历史数据。同时，进行 ST 股票和上市不满一年的股票过滤。
  2. 选股策略类: 实现低价股策略、资金流策略和小市值策略，选股结果为符合条件的股票列表。
  3. 回测执行类: 按照日、周、月进行轮动。按月轮动时，若第一天未找到，则在后四天内继续找。买卖考虑 A 股涨跌停时买卖受到限制，且 ST、创业板、科创板、主板的涨跌停幅度不一致。计算收益时考虑增发带来的股价变化。支持分别计算不同板块的收益。

代码

  1. 数据读取类
import baostock as bs
import pandas as pd
import os

class DataReader:
    def __init__(self, data_dir):
        self.data_dir = data_dir

    def load_all_data(self):
        '''
        按天读取所有股票的数据
        '''
        files = os.listdir(self.data_dir)
        all_data = {}
        for file in files:
            if not file.endswith('.h5'):
                continue
            stock_code = file.split('.')[0]
            df = pd.read_hdf(os.path.join(self.data_dir, file))
            all_data[stock_code] = df
        return all_data

    def load_data_by_stock(self, stock_code):
        '''
        按股票名称读取所有历史数据
        '''
        file_path = os.path.join(self.data_dir, '{}.h5'.format(stock_code))
        if not os.path.exists(file_path):
            return None
        return pd.read_hdf(file_path)

    def filter_st_stock(self, data):
        '''
        过滤 ST 股票
        '''
        return data[~data['name'].str.contains('ST')]

    def filter_new_stock(self, data, year=1):
        '''
        过滤上市不满一年的股票
        '''
        now = pd.Timestamp.now()
        data['list_date'] = pd.to_datetime(data['list_date'])
        data = data[(now - data['list_date']).dt.days > 365 * year]
        return data
  1. 选股策略类
import pandas as pd

class StockSelector:
    def __init__(self, all_data):
        self.all_data = all_data

    def select_low_price_stock(self, date):
        '''
        选取当天收盘价小于5元的股票
        '''
        selected_stocks = []
        for stock_code, data in self.all_data.items():
            if data is None:
                continue
            stock_date_data = data.loc[date]
            if stock_date_data['close'] < 5:
                selected_stocks.append(stock_code)
        return selected_stocks

    def select_high_funds_flow_stock(self, date):
        '''
        选取当天主力资金净流入大于1亿元的股票
        '''
        selected_stocks = []
        for stock_code, data in self.all_data.items():
            if data is None:
                continue
            stock_date_data = data.loc[date]
            if stock_date_data['netAmountMain'] > 1e8:
                selected_stocks.append(stock_code)
        return selected_stocks

    def select_small_market_value_stock(self, date):
        '''
        选取当天市值小于10亿元的股票
        '''
        selected_stocks = []
        for stock_code, data in self.all_data.items():
            if data is None:
                continue
            stock_date_data = data.loc[date]
            if stock_date_data['market_cap'] < 1e10:
                selected_stocks.append(stock_code)
        return selected_stocks
  1. 回测执行类
class Backtester:
    def __init__(self, data_reader, stock_selector):
        self.data_reader = data_reader
        self.stock_selector = stock_selector

    def run(self, start_date, end_date, freq='D', plate=None):
        '''
        回测执行
        '''
        all_data = self.data_reader.load_all_data()
        selected_stocks = []
        positions = {}
        cash = 1e7
        for date in pd.date_range(start_date, end_date, freq=freq):
            if freq == 'M':
                # 按月轮动时，若第一天未找到，则在后四天内继续找
                if len(selected_stocks) == 0:
                    for i in range(1, 5):
                        next_date = date + pd.Timedelta(days=i)
                        if len(selected_stocks) > 0:
                            break
                        if plate is None:
                            selected_stocks = self.stock_selector.select_low_price_stock(next_date)
                            selected_stocks = list(set(selected_stocks) - set(positions.keys()))
                            continue
                        else:
                            selected_stocks = self.stock_selector.select_low_price_stock(next_date)
                            selected_stocks = list(set(selected_stocks) - set(positions.keys()))
                            selected_stocks = self.filter_plate(selected_stocks, plate)
                            continue
            else:
                if plate is None:
                    selected_stocks = self.stock_selector.select_low_price_stock(date)
                else:
                    selected_stocks = self.stock_selector.select_low_price_stock(date)
                    selected_stocks = self.filter_plate(selected_stocks, plate)
                selected_stocks = list(set(selected_stocks) - set(positions.keys()))

            for stock_code in positions.keys():
                data = all_data[stock_code]
                stock_date_data = data.loc[date]
                if stock_date_data['open'] == stock_date_data['close']:
                    continue
                if stock_date_data['pctChg'] > 0.097:
                    positions[stock_code]['sell_price'] = stock_date_data['close'] * 0.097 + stock_date_data['close']
                elif stock_date_data['pctChg'] < -0.097:
                    positions[stock_code]['sell_price'] = stock_date_data['close'] - stock_date_data['close'] * 0.097
                else:
                    positions[stock_code]['sell_price'] = stock_date_data['close']
                if positions[stock_code]['sell_price'] > positions[stock_code]['cost_price']:
                    positions[stock_code]['sell_price'] = positions[stock_code]['cost_price'] * 1.097
                elif positions[stock_code]['sell_price'] < positions[stock_code]['cost_price']:
                    positions[stock_code]['sell_price'] = positions[stock_code]['cost_price'] * 0.903
                cash += positions[stock_code]['sell_price'] * positions[stock_code]['position'] - positions[stock_code]['cost_price'] * positions[stock_code]['position']

            for stock_code in selected_stocks:
                data = all_data[stock_code]
                stock_date_data = data.loc[date]
                if stock_date_data['open'] == stock_date_data['close']:
                    continue
                if stock_date_data['pctChg'] > 0.097:
                    buy_price = stock_date_data['close'] * 0.097 + stock_date_data['close']
                elif stock_date_data['pctChg'] < -0.097:
                    buy_price = stock_date_data['close'] - stock_date_data['close'] * 0.097
                else:
                    buy_price = stock_date_data['close']
                if buy_price > stock_date_data['preClose'] * 1.097:
                    buy_price = stock_date_data['preClose'] * 1.097
                elif buy_price < stock_date_data['preClose'] * 0.903:
                    buy_price = stock_date_data['preClose'] * 0.903
                position = int(cash // (buy_price * 100)) * 100
                if position > 0:
                    positions[stock_code] = {'position': position, 'cost_price': buy_price}
                    cash -= position * buy_price

        total_profit = 0
        for stock_code, position in positions.items():
            data = all_data[stock_code]
            stock_date_data = data.loc[end_date]
            total_profit += position['position'] * (stock_date_data['close'] - position['cost_price'])
        return total_profit

    def filter_plate(self, stocks, plate):
        '''
        过滤指定板块的股票
        '''
        if plate == 'cyb':
            return [stock for stock in stocks if stock.startswith('30')]
        elif plate == 'kcb':
            return [stock for stock in stocks if stock.startswith('68')]
        elif plate == 'zxb':
            return [stock for stock in stocks if stock.startswith('00') and int(stock[2]) >= 3 and int(stock[2]) <= 7]
        elif plate == 'mb':
            return [stock for stock in stocks if stock.startswith('60')]
        else:
            return stocks

执行回测并输出结果和回测对比图

data_reader = DataReader('your_data_path')
all_data = data_reader.load_all_data()
stock_selector = StockSelector(all_data)
backtester = Backtester(data_reader, stock_selector)
start_date = '2019-01-01'
end_date = '2021-12-31'

# 低价股策略
profit = backtester.run(start_date, end_date, freq='M', plate=None)
print('低价股策略收益率：{:.2%}'.format(profit / 1e7))

# 资金流策略
profit = backtester.run(start_date, end_date, freq='M', plate=None)
print('资金流策略收益率：{:.2%}'.format(profit / 1e7))

# 小市值策略
profit = backtester.run(start_date, end_date, freq='M', plate=None)
print('小市值策略收益率：{:.2%}'.format(profit / 1e7))

# 创业板
profit = backtester.run(start_date, end_date, freq='M', plate='cyb')
print('创业板收益率：{:.2%}'.format(profit / 1e7))

# 科创板
profit = backtester.run(start_date, end_date, freq='M', plate='kcb')
print('科创板收益率：{:.2%}'.format(profit / 1e7))

# 中小板
profit = backtester.run(start_date, end_date, freq='M', plate='zxb')
print('中小板收益率：{:.2%}'.format(profit / 1e7))

# 主板
profit = backtester.run(start_date, end_date, freq='M', plate='mb')
print('主板收益率：{:.2%}'.format(profit / 1e7))

注意

  1. 请将 your_data_path 替换为您的 HDF 数据文件路径。
  2. 代码中使用了一些假设，例如涨跌停幅度、增发计算等，实际情况可能有所不同。请根据您的需求进行调整。
  3. 本框架仅供参考，实际应用中可能需要根据您的需求进行改进。

总结

本框架提供了一种基于 Baostock 和本地存储 HDF 数据的低价股、资金流和小市值策略回测方案，方便进行策略研究和比较。您可以在此基础上进行扩展，添加更多策略和功能，以满足您的个性化需求。

基于Baostock 和本地存储 HDF 数据的低价股、资金流和小市值策略回测框架

原文地址: https://www.cveoy.top/t/topic/nLCg 著作权归作者所有。请勿转载和采集!

免费AI点我,无需注册和登录