How to use fetch method in lisa

Best Python code snippet using lisa_python

QAQuery_Advance.py

Source:QAQuery_Advance.py Github

copy

Full Screen

1# coding: utf-82#3# The MIT License (MIT)4#5# Copyright (c) 2016-2021 yutiansut/QUANTAXIS6#7# Permission is hereby granted, free of charge, to any person obtaining a copy8# of this software and associated documentation files (the "Software"), to deal9# in the Software without restriction, including without limitation the rights10# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell11# copies of the Software, and to permit persons to whom the Software is12# furnished to do so, subject to the following conditions:13#14# The above copyright notice and this permission notice shall be included in all15# copies or substantial portions of the Software.16#17# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR18# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,19# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE20# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER21# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,22# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE23# SOFTWARE.24import datetime25import re26import pymongo27import pandas as pd28from pandas import DataFrame29from QUANTAXIS.QAData import (30 QA_DataStruct_Index_day,31 QA_DataStruct_Index_min,32 QA_DataStruct_Future_day,33 QA_DataStruct_Future_min,34 QA_DataStruct_Stock_block,35 QA_DataStruct_Financial,36 QA_DataStruct_Stock_day,37 QA_DataStruct_Stock_min,38 QA_DataStruct_CryptoCurrency_day,39 QA_DataStruct_CryptoCurrency_min,40 QA_DataStruct_Stock_transaction,41 QA_DataStruct_Index_min,42 QA_DataStruct_Index_transaction43)44from QUANTAXIS.QAFetch.QAQuery import (45 QA_fetch_index_day,46 QA_fetch_index_min,47 QA_fetch_index_transaction,48 QA_fetch_stock_day,49 QA_fetch_stock_full,50 QA_fetch_stock_min,51 QA_fetch_stock_transaction,52 QA_fetch_future_day,53 QA_fetch_future_min,54 QA_fetch_financial_report,55 QA_fetch_stock_list,56 QA_fetch_index_list,57 QA_fetch_future_list,58 QA_fetch_stock_financial_calendar,59 QA_fetch_stock_divyield,60 QA_fetch_cryptocurrency_day,61 QA_fetch_cryptocurrency_min,62 QA_fetch_cryptocurrency_list63)64from QUANTAXIS.QAUtil.QADate import month_data65from QUANTAXIS.QAUtil import (66 DATABASE,67 QA_Setting,68 QA_util_date_stamp,69 QA_util_date_valid,70 QA_util_log_info,71 QA_util_time_stamp,72 QA_util_getBetweenQuarter,73 QA_util_datetime_to_strdate,74 QA_util_add_months75)76"""77按要求从数据库取数据，并转换成numpy结构78总体思路：79⚙️QA_fetch_***_adv80📍⚙️QA_fetch_*** 🐌 获取数据collections从mongodb中 🐌 返回DataFrame ,81📍📍⚙️用返回的 DataFrame 初始化 ️QA_DataStruct_***82类型***有83_Stock_day84_Stock_min85_Index_day86_Index_min87"""88def QA_fetch_option_day_adv(89 code,90 start='all',91 end=None,92 if_drop_index=True,93 # 🛠 todo collections 参数没有用到， 且数据库是固定的， 这个变量后期去掉94 collections=DATABASE.option_day95):96 '''97 '''98 pass99def QA_fetch_stock_day_adv(100 code,101 start='all',102 end=None,103 if_drop_index=True,104 # 🛠 todo collections 参数没有用到， 且数据库是固定的， 这个变量后期去掉105 collections=DATABASE.stock_day106):107 '''108 :param code: 股票代码109 :param start: 开始日期110 :param end: 结束日期111 :param if_drop_index:112 :param collections: 默认数据库113 :return: 如果股票代码不存 或者开始结束日期不存在 在返回 None ，合法返回 QA_DataStruct_Stock_day 数据114 '''115 '获取股票日线'116 end = start if end is None else end117 start = str(start)[0:10]118 end = str(end)[0:10]119 if start == 'all':120 start = '1990-01-01'121 end = str(datetime.date.today())122 res = QA_fetch_stock_day(code, start, end, format='pd', collections= collections)123 if res is None:124 # 🛠 todo 报告是代码不合法，还是日期不合法125 print(126 "QA Error QA_fetch_stock_day_adv parameter code=%s , start=%s, end=%s call QA_fetch_stock_day return None"127 % (code,128 start,129 end)130 )131 return None132 else:133 res_reset_index = res.set_index(['date', 'code'], drop=if_drop_index)134 # if res_reset_index is None:135 # print("QA Error QA_fetch_stock_day_adv set index 'datetime, code' return None")136 # return None137 return QA_DataStruct_Stock_day(res_reset_index)138def QA_fetch_stock_min_adv(139 code,140 start,141 end=None,142 frequence='1min',143 if_drop_index=True,144 # 🛠 todo collections 参数没有用到， 且数据库是固定的， 这个变量后期去掉145 collections=DATABASE.stock_min146):147 '''148 '获取股票分钟线'149 :param code: 字符串str eg 600085150 :param start: 字符串str 开始日期 eg 2011-01-01151 :param end: 字符串str 结束日期 eg 2011-05-01152 :param frequence: 字符串str 分钟线的类型 支持 1min 1m 5min 5m 15min 15m 30min 30m 60min 60m 类型153 :param if_drop_index: Ture False ， dataframe drop index or not154 :param collections: mongodb 数据库155 :return: QA_DataStruct_Stock_min 类型156 '''157 if frequence in ['1min', '1m']:158 frequence = '1min'159 elif frequence in ['5min', '5m']:160 frequence = '5min'161 elif frequence in ['15min', '15m']:162 frequence = '15min'163 elif frequence in ['30min', '30m']:164 frequence = '30min'165 elif frequence in ['60min', '60m']:166 frequence = '60min'167 else:168 print(169 "QA Error QA_fetch_stock_min_adv parameter frequence=%s is none of 1min 1m 5min 5m 15min 15m 30min 30m 60min 60m"170 % frequence171 )172 return None173 # __data = [] 未使用174 end = start if end is None else end175 if len(start) == 10:176 start = '{} 09:30:00'.format(start)177 if len(end) == 10:178 end = '{} 15:00:00'.format(end)179 if start == end:180 # 🛠 todo 如果相等，根据 frequence 获取开始时间的 时间段 QA_fetch_stock_min， 不支持start end是相等的181 print(182 "QA Error QA_fetch_stock_min_adv parameter code=%s , start=%s, end=%s is equal, should have time span! "183 % (code,184 start,185 end)186 )187 return None188 # 🛠 todo 报告错误 如果开始时间 在 结束时间之后189 res = QA_fetch_stock_min(code, start, end, format='pd', frequence=frequence, collections= collections)190 if res is None:191 print(192 "QA Error QA_fetch_stock_min_adv parameter code=%s , start=%s, end=%s frequence=%s call QA_fetch_stock_min return None"193 % (code,194 start,195 end,196 frequence)197 )198 return None199 else:200 res_set_index = res.set_index(['datetime', 'code'], drop=if_drop_index)201 # if res_set_index is None:202 # print("QA Error QA_fetch_stock_min_adv set index 'datetime, code' return None")203 # return None204 return QA_DataStruct_Stock_min(res_set_index)205def QA_fetch_stock_day_full_adv(date):206 '''207 '返回全市场某一天的数据'208 :param date:209 :return: QA_DataStruct_Stock_day类 型数据210 '''211 # 🛠 todo 检查日期data参数212 res = QA_fetch_stock_full(date, 'pd')213 if res is None:214 print(215 "QA Error QA_fetch_stock_day_full_adv parameter date=%s call QA_fetch_stock_full return None"216 % (date)217 )218 return None219 else:220 res_set_index = res.set_index(['date', 'code'])221 # if res_set_index is None:222 # print("QA Error QA_fetch_stock_day_full set index 'date, code' return None")223 return QA_DataStruct_Stock_day(res_set_index)224def QA_fetch_index_day_adv(225 code,226 start,227 end=None,228 if_drop_index=True,229 # 🛠 todo collections 参数没有用到， 且数据库是固定的， 这个变量后期去掉230 collections=DATABASE.index_day231):232 '''233 :param code: code: 字符串str eg 600085234 :param start: 字符串str 开始日期 eg 2011-01-01235 :param end: 字符串str 结束日期 eg 2011-05-01236 :param if_drop_index: Ture False ， dataframe drop index or not237 :param collections: mongodb 数据库238 :return:239 '''240 '获取指数日线'241 end = start if end is None else end242 start = str(start)[0:10]243 end = str(end)[0:10]244 # 🛠 todo 报告错误 如果开始时间 在 结束时间之后245 # 🛠 todo 如果相等246 res = QA_fetch_index_day(code, start, end, format='pd', collections= collections)247 if res is None:248 print(249 "QA Error QA_fetch_index_day_adv parameter code=%s start=%s end=%s call QA_fetch_index_day return None"250 % (code,251 start,252 end)253 )254 return None255 else:256 res_set_index = res.set_index(['date', 'code'], drop=if_drop_index)257 # if res_set_index is None:258 # print("QA Error QA_fetch_index_day_adv set index 'date, code' return None")259 # return None260 return QA_DataStruct_Index_day(res_set_index)261def QA_fetch_index_min_adv(262 code,263 start,264 end=None,265 frequence='1min',266 if_drop_index=True,267 collections=DATABASE.index_min268):269 '''270 '获取股票分钟线'271 :param code:272 :param start:273 :param end:274 :param frequence:275 :param if_drop_index:276 :param collections:277 :return:278 '''279 if frequence in ['1min', '1m']:280 frequence = '1min'281 elif frequence in ['5min', '5m']:282 frequence = '5min'283 elif frequence in ['15min', '15m']:284 frequence = '15min'285 elif frequence in ['30min', '30m']:286 frequence = '30min'287 elif frequence in ['60min', '60m']:288 frequence = '60min'289 # __data = [] 没有使用290 end = start if end is None else end291 if len(start) == 10:292 start = '{} 09:30:00'.format(start)293 if len(end) == 10:294 end = '{} 15:00:00'.format(end)295 # 🛠 todo 报告错误 如果开始时间 在 结束时间之后296 # if start == end:297 # 🛠 todo 如果相等，根据 frequence 获取开始时间的 时间段 QA_fetch_index_min_adv， 不支持start end是相等的298 # print("QA Error QA_fetch_index_min_adv parameter code=%s , start=%s, end=%s is equal, should have time span! " % (code, start, end))299 # return None300 res = QA_fetch_index_min(code, start, end, format='pd', frequence=frequence, collections= collections)301 if res is None:302 print(303 "QA Error QA_fetch_index_min_adv parameter code=%s start=%s end=%s frequence=%s call QA_fetch_index_min return None"304 % (code,305 start,306 end,307 frequence)308 )309 else:310 res_reset_index = res.set_index(311 ['datetime',312 'code'],313 drop=if_drop_index314 )315 # if res_reset_index is None:316 # print("QA Error QA_fetch_index_min_adv set index 'date, code' return None")317 return QA_DataStruct_Index_min(res_reset_index)318def QA_fetch_stock_transaction_adv(319 code,320 start,321 end=None,322 frequence='tick',323 if_drop_index=True,324 collections=DATABASE.stock_transaction325):326 '''327 :param code:328 :param start:329 :param end:330 :param if_drop_index:331 :param collections:332 :return:333 '''334 end = start if end is None else end335 if len(start) == 10:336 start = '{} 09:30:00'.format(start)337 if len(end) == 10:338 end = '{} 15:00:00'.format(end)339 if start == end:340 # 🛠 todo 如果相等，根据 frequence 获取开始时间的 时间段 QA_fetch_stock_min， 不支持start end是相等的341 print(342 "QA Error QA_fetch_stock_transaction_adv parameter code=%s , start=%s, end=%s is equal, should have time span! "343 % (code,344 start,345 end)346 )347 return None348 # 🛠 todo 报告错误 如果开始时间 在 结束时间之后349 res = QA_fetch_stock_transaction(350 code,351 start,352 end,353 format='pd',354 frequence=frequence, 355 collections= collections356 )357 if res is None:358 print(359 "QA Error QA_fetch_stock_transaction_adv parameter code=%s , start=%s, end=%s frequence=%s call QA_fetch_stock_transaction return None"360 % (code,361 start,362 end,363 frequence)364 )365 return None366 else:367 res_set_index = res.set_index(['datetime', 'code'], drop=if_drop_index)368 # if res_set_index is None:369 # print("QA Error QA_fetch_stock_min_adv set index 'datetime, code' return None")370 # return None371 return QA_DataStruct_Stock_transaction(res_set_index)372# 没有被使用， 和下面的QA_fetch_stock_list_adv函数是一致的373# def QA_fetch_security_list_adv(collections=DATABASE.stock_list):374# '获取股票列表'375# return pd.DataFrame([item for item in collections.find()]).drop('_id', axis=1, inplace=False)376def QA_fetch_index_transaction_adv(377 code,378 start,379 end=None,380 frequence='tick',381 if_drop_index=True,382 collections=DATABASE.index_transaction383):384 '''385 :param code:386 :param start:387 :param end:388 :param if_drop_index:389 :param collections:390 :return:391 '''392 end = start if end is None else end393 if len(start) == 10:394 start = '{} 09:30:00'.format(start)395 if len(end) == 10:396 end = '{} 15:00:00'.format(end)397 if start == end:398 # 🛠 todo 如果相等，根据 frequence 获取开始时间的 时间段 QA_fetch_stock_min， 不支持start end是相等的399 print(400 "QA Error QA_fetch_stock_min_adv parameter code=%s , start=%s, end=%s is equal, should have time span! "401 % (code,402 start,403 end)404 )405 return None406 # 🛠 todo 报告错误 如果开始时间 在 结束时间之后407 res = QA_fetch_index_transaction(408 code,409 start,410 end,411 format='pd',412 frequence=frequence, collections= collections413 )414 if res is None:415 print(416 "QA Error QA_fetch_index_transaction_adv parameter code=%s , start=%s, end=%s frequence=%s call QA_fetch_index_transaction return None"417 % (code,418 start,419 end,420 frequence)421 )422 return None423 else:424 res_set_index = res.set_index(['datetime', 'code'], drop=if_drop_index)425 # if res_set_index is None:426 # print("QA Error QA_fetch_stock_min_adv set index 'datetime, code' return None")427 # return None428 return QA_DataStruct_Index_transaction(res_set_index)429# 没有被使用， 和下面的QA_fetch_stock_list_adv函数是一致的430# def QA_fetch_security_list_adv(collections=DATABASE.stock_list):431# '获取股票列表'432# return pd.DataFrame([item for item in collections.find()]).drop('_id', axis=1, inplace=False)433def QA_fetch_stock_list_adv(collections=DATABASE.stock_list):434 '''435 '获取股票列表'436 :param collections: mongodb 数据库437 :return: DataFrame438 '''439 stock_list_items = QA_fetch_stock_list(collections)440 if len(stock_list_items) == 0:441 print(442 "QA Error QA_fetch_stock_list_adv call item for item in collections.find() return 0 item, maybe the DATABASE.stock_list is empty!"443 )444 return None445 return stock_list_items446def QA_fetch_index_list_adv(collections=DATABASE.index_list):447 '''448 '获取股票列表'449 :param collections: mongodb 数据库450 :return: DataFrame451 '''452 index_list_items = QA_fetch_index_list(collections)453 if len(index_list_items) == 0:454 print(455 "QA Error QA_fetch_index_list_adv call item for item in collections.find() return 0 item, maybe the DATABASE.index_list is empty!"456 )457 return None458 return index_list_items459def QA_fetch_future_day_adv(460 code,461 start,462 end=None,463 if_drop_index=True,464 # 🛠 todo collections 参数没有用到， 且数据库是固定的， 这个变量后期去掉465 collections=DATABASE.future_day466):467 '''468 :param code: code: 字符串str eg 600085469 :param start: 字符串str 开始日期 eg 2011-01-01470 :param end: 字符串str 结束日期 eg 2011-05-01471 :param if_drop_index: Ture False ， dataframe drop index or not472 :param collections: mongodb 数据库473 :return:474 '''475 '获取期货日线'476 end = start if end is None else end477 start = str(start)[0:10]478 end = str(end)[0:10]479 # 🛠 todo 报告错误 如果开始时间 在 结束时间之后480 # 🛠 todo 如果相等481 res = QA_fetch_future_day(code, start, end, format='pd', collections= collections)482 if res is None:483 print(484 "QA Error QA_fetch_future_day_adv parameter code=%s start=%s end=%s call QA_fetch_future_day return None"485 % (code,486 start,487 end)488 )489 else:490 res_set_index = res.set_index(['date', 'code'])491 # if res_set_index is None:492 # print("QA Error QA_fetch_index_day_adv set index 'date, code' return None")493 # return None494 return QA_DataStruct_Future_day(res_set_index)495def QA_fetch_future_min_adv(496 code,497 start,498 end=None,499 frequence='1min',500 if_drop_index=True,501 collections=DATABASE.future_min502):503 '''504 '获取股票分钟线'505 :param code:506 :param start:507 :param end:508 :param frequence:509 :param if_drop_index:510 :param collections:511 :return:512 '''513 if frequence in ['1min', '1m']:514 frequence = '1min'515 elif frequence in ['5min', '5m']:516 frequence = '5min'517 elif frequence in ['15min', '15m']:518 frequence = '15min'519 elif frequence in ['30min', '30m']:520 frequence = '30min'521 elif frequence in ['60min', '60m']:522 frequence = '60min'523 # __data = [] 没有使用524 end = start if end is None else end525 if len(start) == 10:526 start = '{} 00:00:00'.format(start)527 if len(end) == 10:528 end = '{} 15:00:00'.format(end)529 # 🛠 todo 报告错误 如果开始时间 在 结束时间之后530 # if start == end:531 # 🛠 todo 如果相等，根据 frequence 获取开始时间的 时间段 QA_fetch_index_min_adv， 不支持start end是相等的532 # print("QA Error QA_fetch_index_min_adv parameter code=%s , start=%s, end=%s is equal, should have time span! " % (code, start, end))533 # return None534 res = QA_fetch_future_min(535 code,536 start,537 end,538 format='pd',539 frequence=frequence, collections= collections540 )541 if res is None:542 print(543 "QA Error QA_fetch_future_min_adv parameter code=%s start=%s end=%s frequence=%s call QA_fetch_future_min return None"544 % (code,545 start,546 end,547 frequence)548 )549 else:550 res_reset_index = res.set_index(551 ['datetime',552 'code'],553 drop=if_drop_index554 )555 # if res_reset_index is None:556 # print("QA Error QA_fetch_index_min_adv set index 'date, code' return None")557 return QA_DataStruct_Future_min(res_reset_index)558def QA_fetch_future_list_adv(collections=DATABASE.future_list):559 '''560 '获取股票列表'561 :param collections: mongodb 数据库562 :return: DataFrame563 '''564 future_list_items = QA_fetch_future_list()565 if len(future_list_items) == 0:566 print(567 "QA Error QA_fetch_future_list_adv call item for item in collections.find() return 0 item, maybe the DATABASE.future_list is empty!"568 )569 return None570 return future_list_items571def QA_fetch_stock_block_adv(572 code=None,573 blockname=None,574 collections=DATABASE.stock_block575):576 '''577 返回板块 ❌578 :param code:579 :param blockname: 为list时模糊查询多版块交集580 :param collections: 默认数据库 stock_block581 :return: QA_DataStruct_Stock_block582 '''583 if isinstance(blockname, (list,)) and len(blockname) > 0:584 reg_join = "|".join(blockname)585 df = DataFrame([i for i in collections.aggregate([ \586 {"$match": {"blockname": {"$regex": reg_join}}}, \587 {"$group": {"_id": "$code", "count": {"$sum": 1}, "blockname": {"$push": "$blockname"}}}, \588 {"$match": {"count": {"$gte": len(blockname)}}}, \589 {"$project": {"code": "$_id", "blockname": 1, "_id": 0, }}, \590 ])])591 df.blockname = df.blockname.apply(lambda x: ",".join(x))592 return QA_DataStruct_Stock_block(593 df.set_index(["blockname",594 "code"],595 drop=False)596 )597 elif code is not None and blockname is None:598 # 返回这个股票代码所属的板块599 data = pd.DataFrame(600 [item for item in collections.find({'code': {601 '$in': code602 }})]603 )604 data = data.drop(['_id'], axis=1)605 return QA_DataStruct_Stock_block(606 data.set_index(['blockname',607 'code'],608 drop=True).drop_duplicates()609 )610 elif blockname is not None and code is None:611 #612 # 🛠 todo fnished 返回 这个板块所有的股票613 # 返回该板块所属的股票614 # print("QA Error blockname is Not none code none, return all code from its block name have not implemented yet !")615 items_from_collections = [616 item617 for item in collections.find({'blockname': re.compile(blockname)})618 ]619 data = pd.DataFrame(items_from_collections).drop(['_id'], axis=1)620 data_set_index = data.set_index(['blockname', 'code'], drop=True)621 return QA_DataStruct_Stock_block(data_set_index)622 else:623 # 🛠 todo 返回 判断 这个股票是否和属于该板块624 data = pd.DataFrame([item for item in collections.find()]625 ).drop(['_id'],626 axis=1)627 data_set_index = data.set_index(['blockname', 'code'], drop=True)628 return QA_DataStruct_Stock_block(data_set_index)629def QA_fetch_stock_realtime_adv(630 code=None,631 num=1,632 collections=DATABASE.get_collection(633 'realtime_{}'.format(datetime.date.today())634 ),635 verbose=True,636):637 '''638 返回当日的上下五档, code可以是股票可以是list, num是每个股票获取的数量639 :param code:640 :param num:641 :param collections: realtime_XXXX-XX-XX 每天实时时间642 :return: DataFrame643 '''644 if code is not None:645 # code 必须转换成list 去查询数据库646 if isinstance(code, str):647 code = [code]648 elif isinstance(code, list):649 pass650 else:651 print(652 "QA Error QA_fetch_stock_realtime_adv parameter code is not List type or String type"653 )654 items_from_collections = [655 item for item in collections.find(656 {'code': {657 '$in': code658 }},659 limit=num * len(code),660 sort=[('datetime',661 pymongo.DESCENDING)]662 )663 ]664 if (items_from_collections is None) or \665 (len(items_from_collections) == 0):666 if verbose:667 print(668 "QA Error QA_fetch_stock_realtime_adv find parameter code={} num={} collection={} return NOne"669 .format(code,670 num,671 collections)672 )673 return674 data = pd.DataFrame(items_from_collections)675 data_set_index = data.set_index(['datetime',676 'code'],677 drop=False).drop(['_id'],678 axis=1)679 return data_set_index680 else:681 print("QA Error QA_fetch_stock_realtime_adv parameter code is None")682def QA_fetch_financial_report_adv(code, start, end=None, ltype='EN'):683 """高级财务查询接口684 Arguments:685 code {[type]} -- [description]686 start {[type]} -- [description]687 Keyword Arguments:688 end {[type]} -- [description] (default: {None})689 """690 if end is None:691 return QA_DataStruct_Financial(692 QA_fetch_financial_report(code,693 start,694 ltype=ltype)695 )696 else:697 series = pd.Series(698 data=month_data,699 index=pd.to_datetime(month_data, utc=False),700 name='date'701 )702 timerange = series.loc[start:end].tolist()703 return QA_DataStruct_Financial(704 QA_fetch_financial_report(code,705 timerange,706 ltype=ltype)707 )708# def QA_fetch_financial_report_adv(code, start='all', end=None, type='report'):709# """高级财务查询接口710# Arguments:711# code {[type]} -- [description]712# start {[type]} -- [description]713# Keyword Arguments:714# end {[type]} -- [description] (default: {None})715# """716# end = start if end is None else end717# start = str(start)[0:10]718# end = str(end)[0:10]719# if start == 'all':720# start = '1990-01-01'721# end = str(datetime.date.today())722# if end is None:723# end = str(datetime.date.today())724# date_list = list(pd.DataFrame.from_dict(QA_util_getBetweenQuarter(725# start, QA_util_datetime_to_strdate(QA_util_add_months(end, -3)))).T.iloc[:, 1])726# if type == 'report':727# return QA_DataStruct_Financial(QA_fetch_financial_report(code, date_list))728# elif type == 'date':729# return QA_DataStruct_Financial(QA_fetch_financial_report(code, date_list, type='date'))730# else:731# daterange = pd.date_range(start, end)732# timerange = [item.strftime('%Y-%m-%d') for item in list(daterange)]733# if type == 'report':734# return QA_DataStruct_Financial(QA_fetch_financial_report(code, timerange))735# elif type == 'date':736# return QA_DataStruct_Financial(QA_fetch_financial_report(code, timerange, type='date'))737def QA_fetch_stock_financial_calendar_adv(738 code,739 start="all",740 end=None,741 format='pd',742 collections=DATABASE.report_calendar743):744 '获取股票日线'745 # code= [code] if isinstance(code,str) else code746 end = start if end is None else end747 start = str(start)[0:10]748 end = str(end)[0:10]749 # code checking750 if start == 'all':751 start = '1990-01-01'752 end = str(datetime.date.today())753 if end is None:754 return QA_DataStruct_Financial(755 QA_fetch_stock_financial_calendar(756 code,757 start,758 str(datetime.date.today())759 )760 )761 else:762 series = pd.Series(763 data=month_data,764 index=pd.to_datetime(month_data, utc=False),765 name='date'766 )767 timerange = series.loc[start:end].tolist()768 return QA_DataStruct_Financial(769 QA_fetch_stock_financial_calendar(code,770 start,771 end)772 )773def QA_fetch_stock_divyield_adv(774 code,775 start="all",776 end=None,777 format='pd',778 collections=DATABASE.report_calendar779):780 '获取股票日线'781 # code= [code] if isinstance(code,str) else code782 end = start if end is None else end783 start = str(start)[0:10]784 end = str(end)[0:10]785 # code checking786 if start == 'all':787 start = '1990-01-01'788 end = str(datetime.date.today())789 if end is None:790 return QA_DataStruct_Financial(791 QA_fetch_stock_divyield(code,792 start,793 str(datetime.date.today()))794 )795 else:796 series = pd.Series(797 data=month_data,798 index=pd.to_datetime(month_data, utc=False),799 name='date'800 )801 timerange = series.loc[start:end].tolist()802 return QA_DataStruct_Financial(803 QA_fetch_stock_divyield(code,804 start,805 end)806 )807def QA_fetch_cryptocurrency_day_adv(808 code,809 start,810 end=None,811 if_drop_index=True,812 collections=DATABASE.cryptocurrency_day813):814 '''815 '获取数字加密资产日线'816 :param code:817 :param start: 字符串str 开始日期 eg 2011-01-01818 :param end: 字符串str 结束日期 eg 2011-05-01819 :param if_drop_index: Ture False ， dataframe drop index or not820 :param collections: mongodb 数据库821 :return:822 '''823 '获取数字加密资产日线'824 end = start if end is None else end825 start = str(start)[0:10]826 end = str(end)[0:10]827 # 🛠 todo 报告错误 如果开始时间 在 结束时间之后828 # 🛠 todo 如果相等829 res = QA_fetch_cryptocurrency_day(code, start, end, format='pd', collections=collections)830 if res is None:831 print(832 "QA Error QA_fetch_cryptocurrency_day_adv parameter symbol=%s start=%s end=%s call QA_fetch_cryptocurrency_day return None"833 % (code,834 start,835 end)836 )837 else:838 res_set_index = res.set_index(['date', 'code'])839 return QA_DataStruct_CryptoCurrency_day(res_set_index)840def QA_fetch_cryptocurrency_min_adv(841 code,842 start,843 end=None,844 frequence='1min',845 if_drop_index=True,846 collections=DATABASE.cryptocurrency_min847):848 '''849 '获取数字加密资产分钟线'850 :param symbol:851 :param start:852 :param end:853 :param frequence:854 :param if_drop_index:855 :param collections:856 :return:857 '''858 if frequence in ['1min', '1m']:859 frequence = '1min'860 elif frequence in ['5min', '5m']:861 frequence = '5min'862 elif frequence in ['15min', '15m']:863 frequence = '15min'864 elif frequence in ['30min', '30m']:865 frequence = '30min'866 elif frequence in ['60min', '60m']:867 frequence = '60min'868 # __data = [] 没有使用869 end = start if end is None else end870 if len(start) == 10:871 start = '{} 00:00:00'.format(start)872 if len(end) == 10:873 end = '{} 23:59:59'.format(end)874 # 🛠 todo 报告错误 如果开始时间 在 结束时间之后875 # if start == end:876 # 🛠 todo 如果相等，根据 frequence 获取开始时间的 时间段 QA_fetch_cryptocurrency_min_adv， 不支持start end是相等的877 # print("QA Error QA_fetch_cryptocurrency_min_adv parameter code=%s , start=%s, end=%s is equal, should have time span! " % (code, start, end))878 # return None879 res = QA_fetch_cryptocurrency_min(880 code,881 start,882 end,883 format='pd',884 frequence=frequence, collections=collections885 )886 if res is None:887 print(888 "QA Error QA_fetch_cryptocurrency_min_adv parameter symbol=%s start=%s end=%s frequence=%s call QA_fetch_cryptocurrency_min return None"889 % (code,890 start,891 end,892 frequence)893 )894 else:895 res_reset_index = res.set_index(896 ['datetime',897 'code'],898 drop=if_drop_index899 )900 # if res_reset_index is None:901 # print("QA Error QA_fetch_cryptocurrency_min_adv set index 'date, code' return None")902 return QA_DataStruct_CryptoCurrency_min(res_reset_index)903def QA_fetch_cryptocurrency_list_adv(904 market,905 collections=DATABASE.cryptocurrency_list906):907 '''908 '获取数字加密资产列表'909 :param collections: mongodb 数据库910 :return: DataFrame911 '''912 cryptocurrency_list_items = QA_fetch_cryptocurrency_list(market, collections=collections)913 if len(cryptocurrency_list_items) == 0:914 print(915 "QA Error QA_fetch_cryptocurrency_list_adv call item for item in collections.find() return 0 item, maybe the DATABASE.cryptocurrency_list is empty!"916 )917 return None918 return cryptocurrency_list_items919if __name__ == '__main__':920 #st = QA_fetch_stock_block_adv(None, ["北京", "计算机"])921 #QA_fetch_stock_realtime_adv(['000001', '000002'], num=10)922 from QUANTAXIS.QAFetch.QAhuobi import FIRST_PRIORITY923 import QUANTAXIS as QA924 codelist = ['BINANCE.BCHUSDT', 'BINANCE.BSVUSDT', 'BINANCE.BTCUSDT', 'BINANCE.EOSUSDT', 'BINANCE.ETHUSDT', 'BINANCE.ETCUSDT', 'BINANCE.DASHUSDT', 'BINANCE.LTCUSDT', 'BINANCE.XMRUSDT', 'BINANCE.XRPUSDT', 'BINANCE.ZECUSDT']925 data1 = QA_fetch_cryptocurrency_day_adv(926 code=codelist+['HUOBI.{}'.format(code) for code in FIRST_PRIORITY],927 start='2019-08-21',928 end='2020-05-28 18:10:00',929 )930 print(data1.data)931 data2 = QA_fetch_cryptocurrency_min_adv(932 code=[933 'OKEX.BTC-USDT',934 'OKEX.ETH-USDT',935 ],936 start='2017-10-01',937 end='2020-05-28 18:10:00',938 frequence='60min'939 )940 print(data2.data)941 data_4h = QA.QA_DataStruct_CryptoCurrency_min(data2.resample('4h'))...

Full Screen

Full Screen

__init__.py

Source:__init__.py Github

copy

Full Screen

1# coding:utf-82#3# The MIT License (MIT)4#5# Copyright (c) 2016-2021 yutiansut/QUANTAXIS6#7# Permission is hereby granted, free of charge, to any person obtaining a copy8# of this software and associated documentation files (the "Software"), to deal9# in the Software without restriction, including without limitation the rights10# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell11# copies of the Software, and to permit persons to whom the Software is12# furnished to do so, subject to the following conditions:13#14# The above copyright notice and this permission notice shall be included in all15# copies or substantial portions of the Software.16#17# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR18# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,19# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE20# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER21# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,22# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE23# SOFTWARE.24"""25QA fetch module26@yutiansut27QAFetch is Under [QAStandard#0.0.2@10x] Protocol28"""29from QUANTAXIS.QAFetch import QAWind as QAWind30from QUANTAXIS.QAFetch import QATushare as QATushare31from QUANTAXIS.QAFetch import QATdx as QATdx32from QUANTAXIS.QAFetch import QAThs as QAThs33from QUANTAXIS.QAFetch import QACrawler as QACL34from QUANTAXIS.QAFetch import QAEastMoney as QAEM35from QUANTAXIS.QAFetch import QAHexun as QAHexun36from QUANTAXIS.QAFetch import QAfinancial37from QUANTAXIS.QAFetch.base import get_stock_market38from QUANTAXIS.QAFetch import QAQAWEB as QAWEB39from QUANTAXIS.QAFetch import QAKQ as QAKQ40def use(package):41 if package in ['wind']:42 try:43 from WindPy import w44 # w.start()45 return QAWind46 except ModuleNotFoundError:47 print('NO WIND CLIENT FOUND')48 elif package in ['tushare', 'ts']:49 return QATushare50 elif package in ['tdx', 'pytdx']:51 return QATdx52 elif package in ['ths', 'THS']:53 return QAThs54 elif package in ['HEXUN', 'Hexun', 'hexun']:55 return QAHexun56 elif package in ['QA']:57 return QAWEB58def QA_fetch_get_stock_day(package, code, start, end, if_fq='00', level='day', type_='pd'):59 Engine = use(package)60 if package in ['ths', 'THS', 'wind']:61 return Engine.QA_fetch_get_stock_day(code, start, end, if_fq)62 elif package in ['ts', 'tushare']:63 return Engine.QA_fetch_get_stock_day(code, start, end, if_fq, type_)64 elif package in ['tdx', 'pytdx']:65 return Engine.QA_fetch_get_stock_day(code, start, end, if_fq, level)66 else:67 return Engine.QA_fetch_get_stock_day(code, start, end)68def QA_fetch_get_stock_realtime(package, code):69 Engine = use(package)70 return Engine.QA_fetch_get_stock_realtime(code)71def QA_fetch_get_stock_indicator(package, code, start, end):72 Engine = use(package)73 return Engine.QA_fetch_get_stock_indicator(code, start, end)74def QA_fetch_get_trade_date(package, end, exchange):75 Engine = use(package)76 return Engine.QA_fetch_get_trade_date(end, exchange)77def QA_fetch_get_stock_min(package, code, start, end, level='1min'):78 Engine = use(package)79 if package in ['tdx', 'pytdx']:80 return Engine.QA_fetch_get_stock_min(code, start, end, level)81 else:82 return 'Unsupport packages'83def QA_fetch_get_stock_transaction(package, code, start, end, retry=2):84 Engine = use(package)85 if package in ['tdx', 'pytdx']:86 return Engine.QA_fetch_get_stock_transaction(code, start, end, retry)87 else:88 return 'Unsupport packages'89def QA_fetch_get_index_transaction(package, code, start, end, retry=2):90 Engine = use(package)91 if package in ['tdx', 'pytdx']:92 return Engine.QA_fetch_get_index_transaction(code, start, end, retry)93 else:94 return 'Unsupport packages'95def QA_fetch_get_stock_transaction_realtime(package, code):96 Engine = use(package)97 if package in ['tdx', 'pytdx']:98 return Engine.QA_fetch_get_stock_transaction_realtime(code)99 else:100 return 'Unsupport packages'101def QA_fetch_get_stock_xdxr(package, code):102 Engine = use(package)103 if package in ['tdx', 'pytdx']:104 return Engine.QA_fetch_get_stock_xdxr(code)105 else:106 return 'Unsupport packages'107def QA_fetch_get_index_day(package, code, start, end, level='day'):108 Engine = use(package)109 if package in ['tdx', 'pytdx']:110 return Engine.QA_fetch_get_index_day(code, start, end, level)111 else:112 return 'Unsupport packages'113def QA_fetch_get_index_min(package, code, start, end, level='1min'):114 Engine = use(package)115 if package in ['tdx', 'pytdx']:116 return Engine.QA_fetch_get_index_min(code, start, end, level)117 else:118 return 'Unsupport packages'119def QA_fetch_get_index_realtime(package, code):120 Engine = use(package)121 return Engine.QA_fetch_get_index_realtime(code)122def QA_fetch_get_bond_day(package, code, start, end, level='day'):123 Engine = use(package)124 if package in ['tdx', 'pytdx']:125 return Engine.QA_fetch_get_bond_day(code, start, end, level)126 else:127 return 'Unsupport packages'128def QA_fetch_get_bond_min(package, code, start, end, level='1min'):129 Engine = use(package)130 if package in ['tdx', 'pytdx']:131 return Engine.QA_fetch_get_bond_min(code, start, end, level)132 else:133 return 'Unsupport packages'134def QA_fetch_get_bond_realtime(package, code):135 Engine = use(package)136 return Engine.QA_fetch_get_bond_realtime(code)137def QA_fetch_get_stock_block(package):138 Engine = use(package)139 if package in ['tdx', 'pytdx', 'ths', 'tushare', 'QA']:140 return Engine.QA_fetch_get_stock_block()141 else:142 return 'Unsupport packages'143def QA_fetch_get_stock_info(package, code):144 Engine = use(package)145 if package in ['tdx', 'pytdx']:146 return Engine.QA_fetch_get_stock_info(code)147 else:148 return 'Unsupport packages'149# LIST150def QA_fetch_get_stock_list(package, type_='stock'):151 Engine = use(package)152 if package in ['tdx', 'pytdx']:153 return Engine.QA_fetch_get_stock_list(type_)154 else:155 return 'Unsupport packages'156def QA_fetch_get_bond_list(package):157 Engine = use(package)158 if package in ['tdx', 'pytdx']:159 return Engine.QA_fetch_get_bond_list()160 else:161 return 'Unsupport packages'162def QA_fetch_get_index_list(package):163 Engine = use(package)164 if package in ['tdx', 'pytdx']:165 return Engine.QA_fetch_get_index_list()166 else:167 return 'Unsupport packages'168def QA_fetch_get_future_list(package,):169 Engine = use(package)170 if package in ['tdx', 'pytdx']:171 return Engine.QA_fetch_get_future_list()172 else:173 return 'Unsupport packages'174def QA_fetch_get_option_list(package,):175 Engine = use(package)176 if package in ['tdx', 'pytdx']:177 return Engine.QA_fetch_get_option_list()178 else:179 return 'Unsupport packages'180def QA_fetch_get_globalfuture_list(package,):181 Engine = use(package)182 if package in ['tdx', 'pytdx']:183 return Engine.QA_fetch_get_globalfuture_list()184 else:185 return 'Unsupport packages'186def QA_fetch_get_hkstock_list(package,):187 Engine = use(package)188 if package in ['tdx', 'pytdx']:189 return Engine.QA_fetch_get_hkstock_list()190 else:191 return 'Unsupport packages'192def QA_fetch_get_hkfund_list(package,):193 Engine = use(package)194 if package in ['tdx', 'pytdx']:195 return Engine.QA_fetch_get_hkfund_list()196 else:197 return 'Unsupport packages'198def QA_fetch_get_hkindex_list(package,):199 Engine = use(package)200 if package in ['tdx', 'pytdx']:201 return Engine.QA_fetch_get_hkindex_list()202 else:203 return 'Unsupport packages'204def QA_fetch_get_usstock_list(package,):205 Engine = use(package)206 if package in ['tdx', 'pytdx']:207 return Engine.QA_fetch_get_usstock_list()208 else:209 return 'Unsupport packages'210def QA_fetch_get_macroindex_list(package,):211 Engine = use(package)212 if package in ['tdx', 'pytdx']:213 return Engine.QA_fetch_get_macroindex_list()214 else:215 return 'Unsupport packages'216def QA_fetch_get_globalindex_list(package,):217 Engine = use(package)218 if package in ['tdx', 'pytdx']:219 return Engine.QA_fetch_get_globalindex_list()220 else:221 return 'Unsupport packages'222def QA_fetch_get_exchangerate_list(package,):223 Engine = use(package)224 if package in ['tdx', 'pytdx']:225 return Engine.QA_fetch_get_exchangerate_list()226 else:227 return 'Unsupport packages'228#######################229def QA_fetch_get_security_bars(code, _type, lens):230 return QATdx.QA_fetch_get_security_bars(code, _type, lens)231def QA_fetch_get_future_transaction(package, code, start, end):232 Engine = use(package)233 if package in ['tdx', 'pytdx']:234 return Engine.QA_fetch_get_future_transaction(code, start, end)235 else:236 return 'Unsupport packages'237def QA_fetch_get_future_transaction_realtime(package, code):238 """239 期货实时tick240 """241 Engine = use(package)242 if package in ['tdx', 'pytdx']:243 return Engine.QA_fetch_get_future_transaction_realtime(code)244 else:245 return 'Unsupport packages'246def QA_fetch_get_future_domain():247 return QAKQ.QA_fetch_get_future_domain()248def QA_fetch_get_future_realtime(package, code):249 Engine = use(package)250 return Engine.QA_fetch_get_future_realtime(code)251def QA_fetch_get_future_day(package, code, start, end, frequence='day'):252 Engine = use(package)253 if package in ['tdx', 'pytdx']:254 return Engine.QA_fetch_get_future_day(code, start, end, frequence=frequence)255 else:256 return 'Unsupport packages'257def QA_fetch_get_future_min(package, code, start, end, frequence='1min'):258 Engine = use(package)259 if package in ['tdx', 'pytdx']:260 return Engine.QA_fetch_get_future_min(code, start, end, frequence=frequence)261 else:262 return 'Unsupport packages'263def QA_fetch_get_chibor(package, frequence):264 Engine = use(package)265 if package in ['Hexun', 'hexun']:266 return Engine.QA_fetch_get_chibor(frequence)267 else:268 return 'Unsupport packages'269QA_fetch_get_option_day = QA_fetch_get_future_day270QA_fetch_get_option_min = QA_fetch_get_future_min271QA_fetch_get_hkstock_day = QA_fetch_get_future_day272QA_fetch_get_hkstock_min = QA_fetch_get_future_min273QA_fetch_get_hkfund_day = QA_fetch_get_future_day274QA_fetch_get_hkfund_min = QA_fetch_get_future_min275QA_fetch_get_hkindex_day = QA_fetch_get_future_day276QA_fetch_get_hkindex_min = QA_fetch_get_future_min277QA_fetch_get_usstock_day = QA_fetch_get_future_day278QA_fetch_get_usstock_min = QA_fetch_get_future_min279QA_fetch_get_option_day = QA_fetch_get_future_day280QA_fetch_get_option_min = QA_fetch_get_future_min281QA_fetch_get_globalfuture_day = QA_fetch_get_future_day282QA_fetch_get_globalfuture_min = QA_fetch_get_future_min283QA_fetch_get_exchangerate_day = QA_fetch_get_future_day284QA_fetch_get_exchangerate_min = QA_fetch_get_future_min285QA_fetch_get_macroindex_day = QA_fetch_get_future_day286QA_fetch_get_macroindex_min = QA_fetch_get_future_min287QA_fetch_get_globalindex_day = QA_fetch_get_future_day...

Full Screen

Full Screen

test_cache.py

Source:test_cache.py Github

copy

Full Screen

...26 value = 'value'27 given_fetch_method = mock.Mock(return_value=iter([('key', value)]))28 given_cache = cache.Cache(given_fetch_method)29 # When value with key is fetched30 result = given_cache.fetch('key', 60.0)31 # Result is returned32 self.assertIs(value, result)33 # Then fetch method is called once34 given_fetch_method.assert_called_once_with(('key',))35 def test_fecth_with_no_result(self):36 given_fetch_method = mock.Mock(return_value=iter([]))37 given_cache = cache.Cache(given_fetch_method)38 # When value with key is fetched39 try:40 given_cache.fetch('key', 60.0)41 except cache.CacheFetchError as error:42 given_fetch_method.assert_called_once_with(('key',))43 self.assertRaises(KeyError, error.reraise_cause)44 else:45 self.fail('Expecting CacheFetchError to be raised.')46 @mock.patch.object(cache, 'LOG')47 def test_fecth_with_failure(self, logger):48 # pylint: disable=unused-argument49 given_error = RuntimeError("It doesn't work like this!")50 def failing_function(keys):51 raise given_error52 given_fetch_method = mock.Mock(side_effect=failing_function)53 given_cache = cache.Cache(given_fetch_method)54 # When value with key is fetched55 try:56 given_cache.fetch('key', 60.0)57 except cache.CacheFetchError as error:58 given_fetch_method.assert_called_once_with(('key',))59 self.assertRaises(RuntimeError, error.reraise_cause)60 else:61 self.fail('Expecting CacheFetchError to be raised.')62 logger.warning.assert_called_once_with(63 'Error fetching values for keys: %r', "'key'",64 exc_info=(type(given_error), given_error, mock.ANY))65 def test_fecth_again_after_clear(self):66 value1 = 'value1'67 value2 = 'value2'68 given_fetch_method = mock.Mock(69 side_effect=[iter([('key', value1)]),70 iter([('key', value2)])])71 given_cache = cache.Cache(given_fetch_method)72 # When value with key is fetched73 result1 = given_cache.fetch('key', 60.0)74 # When cache is cleared75 given_cache.clear()76 # When value with same key is fetched again77 result2 = given_cache.fetch('key', 0.0)78 # Then first result is returned79 self.assertIs(value1, result1)80 # Then fetch method is called twice81 self.assertEqual(82 [mock.call(('key',)), mock.call(('key',))],83 given_fetch_method.mock_calls)84 # Then second result is returned85 self.assertIs(value2, result2)86 def test_fecth_again_before_timeout(self):87 value1 = 'value1'88 value2 = 'value2'89 given_fetch_method = mock.Mock(90 side_effect=[iter([('key', value1)]),91 iter([('key', value2)])])92 given_cache = cache.Cache(given_fetch_method)93 # When value with key is fetched94 result1 = given_cache.fetch('key', 1.0)95 # When value with same key is fetched again and cached entry is not96 # expired97 result2 = given_cache.fetch('key', 0.0)98 # First result is returned99 self.assertIs(value1, result1)100 # Then fetch method is called once101 given_fetch_method.assert_called_once_with(('key',))102 # Then first result is returned twice103 self.assertIs(value1, result2)104 def test_fecth_again_after_timeout(self):105 value1 = 'value1'106 value2 = 'value2'107 given_fetch_method = mock.Mock(108 side_effect=[iter([('key', value1)]),109 iter([('key', value2)])])110 given_cache = cache.Cache(given_fetch_method)111 # When value with key is fetched112 result1 = given_cache.fetch('key', 0.0)113 # When value with same key is fetched again and cached entry is114 # expired115 result2 = given_cache.fetch('key', 0.0)116 # Then first result is returned117 self.assertIs(value1, result1)118 # Then fetch method is called twice119 self.assertEqual(120 [mock.call(('key',)), mock.call(('key',))],121 given_fetch_method.mock_calls)122 # Then second result is returned123 self.assertIs(value2, result2)124 def test_fecth_two_values_yielding_both_before_timeout(self):125 value1 = 'value1'126 value2 = 'value2'127 given_fetch_method = mock.Mock(128 return_value=iter([('key1', value1),129 ('key2', value2)]))130 given_cache = cache.Cache(given_fetch_method)131 # When value with key is fetched132 result1 = given_cache.fetch('key1', 60.0)133 # When value with another key is fetched and cached entry is not134 # expired135 result2 = given_cache.fetch('key2', 60.0)136 # Then first result is returned137 self.assertIs(value1, result1)138 # Then fetch method is called once139 given_fetch_method.assert_called_once_with(('key1',))140 # Then second result is returned141 self.assertIs(value2, result2)142 def test_fecth_two_values_yielding_both_after_timeout(self):143 value1 = 'value1'144 value2 = 'value2'145 given_fetch_method = mock.Mock(146 return_value=[('key1', value1), ('key2', value2)])147 given_cache = cache.Cache(given_fetch_method)148 # When value with key is fetched149 result1 = given_cache.fetch('key1', 0.0)150 # When value with another key is fetched and cached entry is151 # expired152 result2 = given_cache.fetch('key2', 0.0)153 # Then first result is returned154 self.assertIs(value1, result1)155 # Then fetch method is called twice156 self.assertEqual(157 [mock.call(('key1',)), mock.call(('key2',))],158 given_fetch_method.mock_calls)159 # Then second result is returned160 self.assertIs(value2, result2)161 def test_fecth_all_with_multiple_entries(self):162 given_fetch_method = mock.Mock(163 return_value=iter([('key', 'value1'),164 ('key', 'value2')]))165 given_cache = cache.Cache(given_fetch_method)166 # When value with key is fetched...

Full Screen

Full Screen

edgetest.py

Source:edgetest.py Github

copy

Full Screen

...46 fetch_object)47 return requests.get(request_url, verify=verify, timeout=const.FETCH_TIMEOUT,48 headers={"Host": fetch_host, "User-Agent": USER_AGENT})49 pass50 def fetch(self, fetch_host, fetch_object, proto="https", port=80, verify=False):51 """52 fetch_host: The Host header to use when fetching53 fetch_object: The path to the object to be fetched54 """55 try:56 response = self.make_request(fetch_host, fetch_object, proto, port, verify)57 except requests.exceptions.Timeout as e:58 # Just assume it took the maximum amount of time59 return const.FETCH_TIMEOUT60 except requests.exceptions.ConnectionError as e:61 logging.error("Connection error when fetching from %s: %s", self.edgename, str(e))62 for i in range(const.FETCH_RETRY-1):63 logging.warning("Retrying connection to %s", self.edgename)64 try:...

Full Screen

Full Screen

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.

LambdaTest Learning Hubs:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run lisa automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

... View sample repo... View DocumentationSign up Free
›
_

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

...
Sign up with GoogleSign up with Email
Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful