QUANTAXIS虽然可用于生产环境，但文档跟不上开发，官网论坛也已经不运营了，docker版本是1但github是0，0有些代码运行不了，且文档缺乏。网上代码和示例因版本不同而无从入手，为此，只能从各种片段信息，探索quantaxis的使用。
启动终端
quantaxisdocker安装好后，会安装mongodb，存的数据会存入此处。quantaxis默认使用pytdx进行数据获取。安装好后，命令提示符输入quantaxis，可以使用saveXXX命令，但这种方式不适合调试和跟踪运行，因此我们通过代码来启动，先本地安装quantaxis：
# 版本 1.10.19
pip install quantaxis
建立新的文件，输入代码
from QUANTAXIS.QACmd import QA_cmd
QA_cmd()
运行文件，也可以启动终端，输入save获得指令列表，与运行quantaxis效果一样，如下：
数据保存过程
运行命令如：savesingle_stock_day000001；进入QUANTAXIS.QASU.main中QA_SU_save_single_stock_day，arg即‘000001’，根据engine参数’tdx’调用tdx引擎的对应方法，注释添加至代码如下：
# QUANTAXIS.QASU.main.py
# main.py为不同的数据源提供了统一的对外接口，根据engine参数调用不同的底层引擎获取数据。
def QA_SU_save_single_stock_day(code, engine, client=DATABASE, paralleled=False):
"""save stock_day
Arguments:
code: stock code #如"000001"
engine {[type]} -- [description] #如"tdx"/"tushare"/"jq"/"gm"
Keyword Arguments:
client {[type]} -- [description] (default: {DATABASE})
:param paralleled: 是否并行处理(default: {True})
"""
# 根据engine参数设定底层引擎
engine = select_save_engine(engine, paralleled=paralleled)
#调用底层对应方法，clent默认为mongodb。
engine.QA_SU_save_single_stock_day(code=code, client=client)
# 向数据库插入获取的数据，获取数据使用QA_fetch_get_stock_day，位于 QUANTAXIS.QAFetch.QATdx
coll_stock_day.insert_many(
QA_util_to_json_from_pandas(
QA_fetch_get_stock_day(
str(code),
QA_util_get_next_day(start_date),
end_date,
"00"
)
)
)
QA_fetch_get_stock_day，这个模块源代码注释已比较详细。
@retry(stop_max_attempt_number=3, wait_random_min=50, wait_random_max=100)
def QA_fetch_get_stock_day(code, start_date, end_date, if_fq="00",
frequence="day", ip=None, port=None):
"""获取日线及以上级别的数据
Arguments:
code {str:6} -- code 是一个单独的code 6位长度的str
start_date {str:10} -- 10位长度的日期 比如"2017-01-01"
end_date {str:10} -- 10位长度的日期 比如"2018-01-01"
Keyword Arguments:
if_fq {str} -- "00"/"bfq" -- 不复权 "01"/"qfq" -- 前复权 "02"/"hfq" -- 后复权 "03"/"ddqfq" -- 定点前复权 "04"/"ddhfq" --定点后复权
frequency {str} -- day/week/month/quarter/year 也可以是简写 D/W/M/Q/Y
ip {str} -- [description] (default: None) ip可以通过select_best_ip()函数重新获取
port {int} -- [description] (default: {None})
Returns:
pd.DataFrame/None -- 返回的是dataframe,如果出错比如只获取了一天,而当天停牌,返回None
Exception:
如果出现网络问题/服务器拒绝, 会出现socket:time out 尝试再次获取/更换ip即可, 本函数不做处理
"""
ip, port = get_mainmarket_ip(ip, port)
api = TdxHq_API()
try:
with api.connect(ip, port, time_out=0.7):
if frequence in ["day", "d", "D", "DAY", "Day"]:
frequence = 9
elif frequence in ["w", "W", "Week", "week"]:
frequence = 5
elif frequence in ["month", "M", "m", "Month"]:
frequence = 6
elif frequence in ["quarter", "Q", "Quarter", "q"]:
frequence = 10
elif frequence in ["y", "Y", "year", "Year"]:
frequence = 11
start_date = str(start_date)[0:10]
today_ = datetime.date.today()
lens = QA_util_get_trade_gap(start_date, today_)
data = pd.concat([api.to_df(
api.get_security_bars(frequence, _select_market_code(
code), code, (int(lens / 800) - i) * 800, 800)) for i in
range(int(lens / 800) + 1)], axis=0, sort=False)
# 这里的问题是: 如果只取了一天的股票,而当天停牌, 那么就直接返回None了
if len(data) < 1:
return None
data = data[data["open"] != 0]
data = data.assign(
date=data["datetime"].apply(lambda x: str(x[0:10])),
code=str(code),
date_stamp=data["datetime"].apply(
lambda x: QA_util_date_stamp(str(x)[0:10])))
.set_index("date", drop=False, inplace=False)
end_date = str(end_date)[0:10]
data = data.drop(
["year", "month", "day", "hour", "minute", "datetime"],
axis=1)[
start_date:end_date]
if if_fq in ["00", "bfq"]:
return data
else:
print("CURRENTLY NOT SUPPORT REALTIME FUQUAN")
return None
# xdxr = QA_fetch_get_stock_xdxr(code)
# if if_fq in ["01","qfq"]:
# return QA_data_make_qfq(data,xdxr)
# elif if_fq in ["02","hfq"]:
# return QA_data_make_hfq(data,xdxr)
except Exception as e:
if isinstance(e, TypeError):
print("Tushare内置的pytdx版本和QUANTAXIS使用的pytdx 版本不同, 请重新安装pytdx以解决此问题")
print("pip uninstall pytdx")
print("pip install pytdx")
else:
print(e)
总结
通过源码，可以有几种使用数据的方式从底层引擎抓数据，获取DataFrame：定位到具体的引擎，调用QASU模块下QA_fetch_*。如：
from QUANTAXIS.QAFETCH.QATdx import QA_fetch_get_stock_day
data=QA_fetch_get_stock_day("000001",start_date="2023-01-01",end_date="2023-03-31")
print(data)
这样后面可以扩展自己的数据库存储方案。
调用顶层API
import QUANTAXIS as qa
data=qa.QA_fetch_get_stock_day(package="tdx",code="000001",start="2023-01-01",end="2023-03-31")
print(data)
结果如下：
从数据库获取数据QA_quotation一个统一的获取k线的方法,如果使用mongo,从本地数据库获取,失败则在线获取
import QUANTAXIS as qa
data=qa.QA_quotation("000001", "2019-12-01", "2020-02-03", frequence="day",market="stock_cn")
print(data)
加速数据的获取或保存
# 容易内存不足
from QUANTAXIS.QASU.save_tdx_parallelism import QA_SU_save_stock_day
QA_SU_save_stock_day("tdx",paralleled=True)
常用API
文章为作者独立观点，不代表股票配资公司观点
相关文章
股民评论