这样一来，如果子类没有实现父类中指定要实现的方法，则会自动调用父类中的方法，于是父类方法就会raise将错误抛出，这样替换数据源的时候就会发现是缺少了对指定接口的实现。
于是，搭建一款普适性的数据源框架就显得尤为重要。这样一来，我们可以随意搭配不同的数据源，也能灵活替换不同的数据源，让我们的量化系统因为数据源的变动，改动点足够小。
星球会员好消息！邀请您加入高质量的Python量化编程答疑群
接下来就给大家分享一下如何搭建一个数据源框架。
这两个接口已经可以满足我们量化系统所必须的基础数据了。比如全市场的股票代码映射表、个股历史行情数据、A股市场开市的交易日等等。具体实现代码如下所示：
class MyDataBackend(DataBackend):
def __init__(self):
self._stock_codes_table = {}
@lru_cache(maxsize=4096)
def stock_basics(self):
return pro.stock_basic(exchange="", list_status="L", fields="ts_code,symbol,name,area,industry,list_date")
@lru_cache(maxsize=4096)
def get_price(self, code, start, end, freq):
"""
:param code: e.g. 000002.SH
:param start: "2009-01-01"
:param end: "2019-06-01"
:returns:
:rtype: numpy.rec.array
"""
code = self.convert_code(code)
# 登陆系统
lg = bs.login()
# 获取指数历史行情数据
fields = "date,open,high,low,close,volume,pctChg"
df_bs = bs.query_history_k_data_plus(code, fields, start_date=start, end_date=end,
frequency=freq,
adjustflag="3") # <class "baostock.data.resultset.ResultData">
# frequency="d"取日k线，adjustflag="3"默认不复权，1：后复权；2：前复权
data_list = []
while (df_bs.error_code == "0") & df_bs.next():
# 获取一条记录，将记录合并在一起
data_list.append(df_bs.get_row_data())
result = pd.DataFrame(data_list, columns=df_bs.fields)
result = result.astype({"volume": "uint64", "pctChg": "float64",
"close": "float64", "open": "float64", "low": "float64", "high": "float64"})
result.volume = result.volume / 100 # 单位转换：股-手
result.volume = result.volume.astype("uint64")
result.date = pd.DatetimeIndex(result.date)
result.set_index("date", drop=True, inplace=True)
result.index = result.index.set_names("Date")
recon_data = {"high": result.high, "low": result.low, "open": result.open, "close": result.close,
"volume": result.volume, "pctChg": result.pctChg}
df_recon = pd.DataFrame(recon_data)
# 登出系统
bs.logout()
return df_recon
@lru_cache()
def get_codes_list(self):
"""
获取所有的股票代码列表
"""
code_list = list(self.code_name_map.keys())
return code_list
@property
def code_name_map(self):
code_group = self.stock_basics.loc[:, ["ts_code", "name"]]
if code_group.empty != True:
codes = code_group.ts_code.values
names = code_group.name.values
self._stock_codes_table = dict(zip(codes, names))
else:
raise AttributeError("股票基本信息为空!!!检查tushare的pro.stock_basic接口")
return self._stock_codes_table
def convert_code(self, code):
num, sym = code.lower().split(".")
return sym + "." + num
@lru_cache()
def get_trading_dates(self, start, end):
"""
获取所有的交易日
:param start: 20160101
:param end: 20160201
"""
df = self.get_price("000001.SZ", start=start, end=end, freq="d")
trading_dates = [datetime.datetime.strftime(date, "%Y-%m-%d") for date in df.index.tolist()]
return trading_dates
@lru_cache(maxsize=4096)
def symbol(self, code):
"""
获取 code 对应的名字
:param code str: 股票代码
:returns: 名字
:rtype: str
"""
symbol = self.code_name_map.get(code)
return "{}[{}]".format(code, symbol)我们会把以上完整的源码上传到知识星球《Python量化场景编程技巧与方法》，帮助小伙伴们更好地掌握这个方法。这个星球的用途是分享搭建量化系统中所涉及到的Python及常用第三方库方面的编程方法和技巧。
调用方式和结果如下所示：
data_org = MyDataBackend()
print(data_org.stock_basics)
"""
ts_code symbol name area industry list_date
0 000001.SZ 000001 平安银行 深圳 银行 19910403
1 000002.SZ 000002 万科A 深圳 全国地产 19910129
......
"""
print(data_org.code_name_map)
# {"000001.SZ": "平安银行", "000002.SZ": "万科A"......}
print(data_org.convert_code("000001.SZ"))
# sz.000001
print(data_org.get_price("000001.SZ", "2021-01-01", "2022-01-01", "d"))
"""
high low open close volume pctChg
Date
2021-01-04 19.10 18.44 19.10 18.60 1554216 -3.8263
2021-01-05 18.48 17.80 18.40 18.17 1821352 -2.3118
2021-01-06 19.56 18.00 18.08 19.56 1934945 7.6500
......
"""
print(data_org.get_codes_list())
# ["000001.SZ", "000002.SZ", "000004.SZ", "000005.SZ"......]
print(data_org.get_trading_dates("2021-01-01", "2022-01-01"))
# ["2021-01-04", "2021-01-05", "2021-01-06", "2021-01-07"......]
print(data_org.symbol("000001.SZ"))
# 000001.SZ[平安银行]
首先定一个父类DataBackend，在这个父类中定义量化系统中所必须的几个接口函数，比如：
def get_price(self, code, start, end, freq)
def get_codes_list(self)
def get_trading_dates(self, start, end)
def symbol(self, code)接下来再实现子类MyDataBackend，子类是继承父类DataBackend的。
子类中我们使用了tusharepro的stock_basic接口，以及baostock的query_history_k_data_plus接口。
在量化交易系统中，数据是第一环节。虽然目前市面上的数据源多种多样，比如tushare、baostock、JQData、pytdx、akshare等等，但是无论什么数据源，必须要满足我们量化系统最基本的几种数据，比如股票代码表、个股行情数据等等。
父类中只是预留这几个接口并不实现，要求在子类中必须实现，如果子类中未实现该方法，我们使用raiseNotImplementedError报错。如下所示：
def get_trading_dates(self, start, end):
"""
获取所有的交易日
:param start: "2009-01-01"
:param end: "2019-06-01"
"""
raise NotImplementedError搭建过程
文章为作者独立观点，不代表股票交易接口观点
股民评论