10504041958

Committed 22 Aug 2024 07:52AM UTC coverage: 64.949% (+0.04%) from 64.911%

Build # 10504041958

Build Type

Pull #896

github

Committed by

web-flow

Commit Message

Merge a7da46f69 into e5a93554a

Pull Request Pull Request #896: 存在更新失败的情况时程序状态码不为0

Run Details

4 of 16 new or added lines in 2 files covered. (25.0%)

3 existing lines in 1 file now uncovered.

6704 of 10322 relevant lines covered (64.95%)

3.86 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

27.34

/rqalpha/cmds/bundle.py

# -*- coding: utf-8 -*-
# 版权所有 2020 深圳米筐科技有限公司（下称“米筐科技”）
#
# 除非遵守当前许可，否则不得使用本软件。
#
#     * 非商业用途（非商业用途指个人出于非商业目的使用本软件，或者高校、研究所等非营利机构出于教育、科研等目的使用本软件）：
#         遵守 Apache License 2.0（下称“Apache 2.0 许可”），您可以在以下位置获得 Apache 2.0 许可的副本：
#         http://www.apache.org/licenses/LICENSE-2.0。
#         除非法律有要求或以书面形式达成协议，否则本软件分发时需保持当前许可“原样”不变，且不得附加任何条件。
#
#     * 商业用途（商业用途指个人出于任何商业目的使用本软件，或者法人或其他组织出于任何目的使用本软件）：
#         未经米筐科技授权，任何个人不得出于任何商业目的使用本软件（包括但不限于向第三方提供、销售、出租、出借、转让本软件、本软件的衍生产品、引用或借鉴了本软件功能或源代码的产品或服务），任何法人或其他组织不得出于任何目的使用本软件，否则米筐科技有权追究相应的知识产权侵权责任。
#         在此前提下，对本软件的使用同样需要遵守 Apache 2.0 许可，Apache 2.0 许可与本许可冲突之处，以本许可为准。
#         详细的授权流程，请联系 public@ricequant.com 获取。
import os
import shutil
import tarfile
import tempfile
import time
import datetime
import dateutil
import sys

import click
import requests
import six
import h5py

from rqalpha.utils.i18n import gettext as _

from rqalpha.cmds.entry import cli
from rqalpha.utils import init_rqdatac_env


@cli.command(help=_("create bundle using RQDatac"))
@click.option('-d', '--data-bundle-path', default=os.path.expanduser('~/.rqalpha'), type=click.Path(file_okay=False))
@click.option("rqdatac_uri", '--rqdatac', '--rqdatac-uri', default=None,
              help='rqdatac uri, eg user:password or tcp://user:password@ip:port')
@click.option('--compression', default=False, is_flag=True, help='enable compression to reduce file size')
@click.option('-c', '--concurrency', type=click.INT, default=1)
def create_bundle(data_bundle_path, rqdatac_uri, compression, concurrency):
    try:
        import rqdatac
    except ImportError:
        click.echo(_(
            'rqdatac is required to create bundle. '
            'you can visit https://www.ricequant.com/welcome/rqdata to get rqdatac, '
            'or use "rqalpha download-bundle" to download monthly updated bundle.'
        ))
        return 1

    try:
        init_rqdatac_env(rqdatac_uri)
        rqdatac.init()
    except ValueError as e:
        click.echo(_('rqdatac init failed with error: {}').format(e))
        return 1

    os.makedirs(os.path.join(data_bundle_path, 'bundle'), exist_ok=True)
    from rqalpha.data.bundle import update_bundle as update_bundle_
    update_bundle_(os.path.join(data_bundle_path, 'bundle'), True, compression, concurrency)


@cli.command(help=_("Update bundle using RQDatac"))
@click.option('-d', '--data-bundle-path', default=os.path.expanduser('~/.rqalpha'), type=click.Path(file_okay=False))
@click.option("rqdatac_uri", '--rqdatac', '--rqdatac-uri', default=None,
              help='rqdatac uri, eg user:password or tcp://user:password@ip:port')
@click.option('--compression', default=False, type=click.BOOL, help='enable compression to reduce file size')
@click.option('-c', '--concurrency', type=click.INT, default=1)
def update_bundle(data_bundle_path, rqdatac_uri, compression, concurrency):
    try:
        import rqdatac
    except ImportError:
        click.echo(_(
            'rqdatac is required to update bundle. '
            'you can visit https://www.ricequant.com/welcome/rqdata to get rqdatac, '
            'or use "rqalpha download-bundle" to download monthly updated bundle.'
        ))
        return 1

    try:
        init_rqdatac_env(rqdatac_uri)
        rqdatac.init()
    except ValueError as e:
        click.echo(_('rqdatac init failed with error: {}').format(e))
        return 1

    if not os.path.exists(os.path.join(data_bundle_path, 'bundle')):
        click.echo(_('bundle not exist, use "rqalpha create-bundle" command instead'))
        return 1

    from rqalpha.data.bundle import update_bundle as update_bundle_
    status = update_bundle_(os.path.join(data_bundle_path, 'bundle'), False, compression, concurrency)
    if status != 0:
        sys.exit(status)


@cli.command(help=_("Download bundle (monthly updated)"))
@click.option('-d', '--data-bundle-path', default=os.path.expanduser('~/.rqalpha'), type=click.Path(file_okay=False))
@click.option('--confirm', default=True, is_flag=True)
def download_bundle(data_bundle_path, confirm):
    default_bundle_path = os.path.abspath(os.path.expanduser('~/.rqalpha/bundle'))
    if data_bundle_path is None:
        data_bundle_path = default_bundle_path
    else:
        data_bundle_path = os.path.abspath(os.path.join(data_bundle_path, './bundle/'))
    if (confirm and os.path.exists(data_bundle_path) and data_bundle_path != default_bundle_path and
            os.listdir(data_bundle_path)):
        click.confirm(_(u"""
    [WARNING]
    Target bundle path {data_bundle_path} is not empty.
    The content of this folder will be REMOVED before updating.
    Are you sure to continue?""").format(data_bundle_path=data_bundle_path), abort=True)

    tmp = os.path.join(tempfile.gettempdir(), 'rq.bundle')
    url, total_length = get_exactly_url()

    with open(tmp, 'wb') as out:
        download(out, total_length, url)

    shutil.rmtree(data_bundle_path, ignore_errors=True)
    os.makedirs(data_bundle_path)
    tar = tarfile.open(tmp, 'r:bz2')
    tar.extractall(data_bundle_path)
    tar.close()
    os.remove(tmp)
    six.print_(_(u"Data bundle download successfully in {bundle_path}").format(bundle_path=data_bundle_path))


@cli.command(help=_("Check bundle"))
@click.option('-d', '--data-bundle-path', default=os.path.expanduser('~/.rqalpha'), type=click.Path(file_okay=False))
def check_bundle(data_bundle_path):
    check_bundle_data(os.path.join(data_bundle_path, "bundle"))


CDN_URL = 'http://bundle.assets.ricequant.com/bundles_v4/rqbundle_%04d%02d.tar.bz2'


def get_exactly_url():
    day = datetime.date.today()
    proxy_uri = os.environ.get('RQALPHA_PROXY')
    while True:  # get exact url
        url = CDN_URL % (day.year, day.month)
        six.print_(_(u"try {} ...").format(url))
        r = requests.get(url, stream=True, proxies={'http': proxy_uri, 'https': proxy_uri})
        if r.status_code == 200:
            return url, int(r.headers.get('content-length'))

        day -= dateutil.relativedelta.relativedelta(months=1)


def download(out, total_length, url):
    retry_interval = 3
    retry_times = 5
    proxy_uri = os.environ.get('RQALPHA_PROXY')
    with click.progressbar(length=total_length, label=_(u"downloading ...")) as bar:
        for i in range(retry_times):
            try:
                headers = {'Range': "bytes={}-".format(bar.pos)}
                r = requests.get(url, headers=headers, stream=True, timeout=10, proxies={'http': proxy_uri,
                                                                                         'https': proxy_uri})
                for data in r.iter_content(chunk_size=8192):
                    bar.update(len(data))
                    out.write(data)

                if total_length == bar.pos:
                    return True  # Download complete . exit
            except requests.exceptions.RequestException:
                if i < retry_times - 1:
                    six.print_(_("\nDownload failed, retry in {} seconds.".format(retry_interval)))
                    time.sleep(retry_interval)
                else:
                    raise


def check_bundle_data(data_bundle_path):
    instruments = ["stocks", "indexes", "futures", "funds"]
    corrupt_files, not_exists_instruments = [], []
    for instrument in instruments:
        file = os.path.join(data_bundle_path, "{}.h5".format(instrument))
        if not os.path.exists(file):
            not_exists_instruments.append(instrument)
            continue
        try:
            with h5py.File(file, mode="r") as f:
                for order_book_id in f.keys():
                    __ = f[order_book_id][:1]
        except Exception:
            corrupt_files.append(file)
    if len(corrupt_files):
        click.echo("{}:\n{}".format(_("corrupted files"), corrupt_files))
        is_ok = input("{}(yes/no):".format(_("remove files"))).lower()
        if is_ok in ["yes", "y"]:
            [os.remove(file) for file in corrupt_files]
            click.echo(_("remove success"))
        elif is_ok in ["no", "n"]:
            click.echo(_("corrupted files not remove"))
        else:
            click.echo(_("input error"))
    elif len(not_exists_instruments):
        click.echo(_("bundle's day bar is incomplete, please update bundle"))
    else:
        click.echo(_("good bundle's day bar"))


1	# -- coding: utf-8 --
2	# 版权所有 2020 深圳米筐科技有限公司（下称“米筐科技”）
3	#
4	# 除非遵守当前许可，否则不得使用本软件。
5	#
6	# * 非商业用途（非商业用途指个人出于非商业目的使用本软件，或者高校、研究所等非营利机构出于教育、科研等目的使用本软件）：
7	# 遵守 Apache License 2.0（下称“Apache 2.0 许可”），您可以在以下位置获得 Apache 2.0 许可的副本：
8	# http://www.apache.org/licenses/LICENSE-2.0。
9	# 除非法律有要求或以书面形式达成协议，否则本软件分发时需保持当前许可“原样”不变，且不得附加任何条件。
10	#
11	# * 商业用途（商业用途指个人出于任何商业目的使用本软件，或者法人或其他组织出于任何目的使用本软件）：
12	# 未经米筐科技授权，任何个人不得出于任何商业目的使用本软件（包括但不限于向第三方提供、销售、出租、出借、转让本软件、本软件的衍生产品、引用或借鉴了本软件功能或源代码的产品或服务），任何法人或其他组织不得出于任何目的使用本软件，否则米筐科技有权追究相应的知识产权侵权责任。
13	# 在此前提下，对本软件的使用同样需要遵守 Apache 2.0 许可，Apache 2.0 许可与本许可冲突之处，以本许可为准。
14	# 详细的授权流程，请联系 public@ricequant.com 获取。
15	import os	6✔
16	import shutil	6✔
17	import tarfile	6✔
18	import tempfile	6✔
19	import time	6✔
20	import datetime	6✔
21	import dateutil	6✔
22	import sys	6✔
23
24	import click	6✔
25	import requests	6✔
26	import six	6✔
27	import h5py	6✔
28
29	from rqalpha.utils.i18n import gettext as _	6✔
30
31	from rqalpha.cmds.entry import cli	6✔
32	from rqalpha.utils import init_rqdatac_env	6✔
33
34
35	@cli.command(help=_("create bundle using RQDatac"))	6✔
36	@click.option('-d', '--data-bundle-path', default=os.path.expanduser('~/.rqalpha'), type=click.Path(file_okay=False))	6✔
37	@click.option("rqdatac_uri", '--rqdatac', '--rqdatac-uri', default=None,	6✔
38	help='rqdatac uri, eg user:password or tcp://user:password@ip:port')
39	@click.option('--compression', default=False, is_flag=True, help='enable compression to reduce file size')	6✔
40	@click.option('-c', '--concurrency', type=click.INT, default=1)	6✔
41	def create_bundle(data_bundle_path, rqdatac_uri, compression, concurrency):	5✔
42	try:	×
43	import rqdatac	×
44	except ImportError:	×
45	click.echo(_(	×
46	'rqdatac is required to create bundle. '
47	'you can visit https://www.ricequant.com/welcome/rqdata to get rqdatac, '
48	'or use "rqalpha download-bundle" to download monthly updated bundle.'
49	))
50	return 1	×
51
52	try:	×
53	init_rqdatac_env(rqdatac_uri)	×
54	rqdatac.init()	×
55	except ValueError as e:	×
56	click.echo(_('rqdatac init failed with error: {}').format(e))	×
57	return 1	×
58
59	os.makedirs(os.path.join(data_bundle_path, 'bundle'), exist_ok=True)	×
60	from rqalpha.data.bundle import update_bundle as update_bundle_	×
61	update_bundle_(os.path.join(data_bundle_path, 'bundle'), True, compression, concurrency)	×
62
63
64	@cli.command(help=_("Update bundle using RQDatac"))	6✔
65	@click.option('-d', '--data-bundle-path', default=os.path.expanduser('~/.rqalpha'), type=click.Path(file_okay=False))	6✔
66	@click.option("rqdatac_uri", '--rqdatac', '--rqdatac-uri', default=None,	6✔
67	help='rqdatac uri, eg user:password or tcp://user:password@ip:port')
68	@click.option('--compression', default=False, type=click.BOOL, help='enable compression to reduce file size')	6✔
69	@click.option('-c', '--concurrency', type=click.INT, default=1)	6✔
70	def update_bundle(data_bundle_path, rqdatac_uri, compression, concurrency):	5✔
71	try:	×
72	import rqdatac	×
73	except ImportError:	×
74	click.echo(_(	×
75	'rqdatac is required to update bundle. '
76	'you can visit https://www.ricequant.com/welcome/rqdata to get rqdatac, '
77	'or use "rqalpha download-bundle" to download monthly updated bundle.'
78	))
79	return 1	×
80
81	try:	×
82	init_rqdatac_env(rqdatac_uri)	×
83	rqdatac.init()	×
84	except ValueError as e:	×
85	click.echo(_('rqdatac init failed with error: {}').format(e))	×
86	return 1	×
87
88	if not os.path.exists(os.path.join(data_bundle_path, 'bundle')):	×
89	click.echo(_('bundle not exist, use "rqalpha create-bundle" command instead'))	×
90	return 1	×
91
92	from rqalpha.data.bundle import update_bundle as update_bundle_	×
NEW 93	status = update_bundle_(os.path.join(data_bundle_path, 'bundle'), False, compression, concurrency)	×
NEW 94	if status != 0:	×
NEW 95	sys.exit(status)	×
96
97
98	@cli.command(help=_("Download bundle (monthly updated)"))	6✔
99	@click.option('-d', '--data-bundle-path', default=os.path.expanduser('~/.rqalpha'), type=click.Path(file_okay=False))	6✔
100	@click.option('--confirm', default=True, is_flag=True)	6✔
101	def download_bundle(data_bundle_path, confirm):	5✔
102	default_bundle_path = os.path.abspath(os.path.expanduser('~/.rqalpha/bundle'))	×
103	if data_bundle_path is None:	×
104	data_bundle_path = default_bundle_path	×
105	else:
106	data_bundle_path = os.path.abspath(os.path.join(data_bundle_path, './bundle/'))	×
107	if (confirm and os.path.exists(data_bundle_path) and data_bundle_path != default_bundle_path and	×
108	os.listdir(data_bundle_path)):
109	click.confirm(_(u"""	×
110	[WARNING]
111	Target bundle path {data_bundle_path} is not empty.
112	The content of this folder will be REMOVED before updating.
113	Are you sure to continue?""").format(data_bundle_path=data_bundle_path), abort=True)
114
115	tmp = os.path.join(tempfile.gettempdir(), 'rq.bundle')	×
116	url, total_length = get_exactly_url()	×
117
118	with open(tmp, 'wb') as out:	×
119	download(out, total_length, url)	×
120
121	shutil.rmtree(data_bundle_path, ignore_errors=True)	×
122	os.makedirs(data_bundle_path)	×
123	tar = tarfile.open(tmp, 'r:bz2')	×
124	tar.extractall(data_bundle_path)	×
125	tar.close()	×
126	os.remove(tmp)	×
127	six.print_(_(u"Data bundle download successfully in {bundle_path}").format(bundle_path=data_bundle_path))	×
128
129
130	@cli.command(help=_("Check bundle"))	6✔
131	@click.option('-d', '--data-bundle-path', default=os.path.expanduser('~/.rqalpha'), type=click.Path(file_okay=False))	6✔
132	def check_bundle(data_bundle_path):	5✔
133	check_bundle_data(os.path.join(data_bundle_path, "bundle"))	×
134
135
136	CDN_URL = 'http://bundle.assets.ricequant.com/bundles_v4/rqbundle_%04d%02d.tar.bz2'	6✔
137
138
139	def get_exactly_url():	6✔
140	day = datetime.date.today()	×
141	proxy_uri = os.environ.get('RQALPHA_PROXY')	×
142	while True: # get exact url
143	url = CDN_URL % (day.year, day.month)	×
144	six.print_(_(u"try {} ...").format(url))	×
145	r = requests.get(url, stream=True, proxies={'http': proxy_uri, 'https': proxy_uri})	×
146	if r.status_code == 200:	×
147	return url, int(r.headers.get('content-length'))	×
148
149	day -= dateutil.relativedelta.relativedelta(months=1)	×
150
151
152	def download(out, total_length, url):	6✔
153	retry_interval = 3	×
154	retry_times = 5	×
155	proxy_uri = os.environ.get('RQALPHA_PROXY')	×
156	with click.progressbar(length=total_length, label=_(u"downloading ...")) as bar:	×
157	for i in range(retry_times):	×
158	try:	×
159	headers = {'Range': "bytes={}-".format(bar.pos)}	×
160	r = requests.get(url, headers=headers, stream=True, timeout=10, proxies={'http': proxy_uri,	×
161	'https': proxy_uri})
162	for data in r.iter_content(chunk_size=8192):	×
163	bar.update(len(data))	×
164	out.write(data)	×
165
166	if total_length == bar.pos:	×
167	return True # Download complete . exit	×
168	except requests.exceptions.RequestException:	×
169	if i < retry_times - 1:	×
170	six.print_(_("\nDownload failed, retry in {} seconds.".format(retry_interval)))	×
171	time.sleep(retry_interval)	×
172	else:
173	raise	×
174
175
176	def check_bundle_data(data_bundle_path):	6✔
177	instruments = ["stocks", "indexes", "futures", "funds"]	×
178	corrupt_files, not_exists_instruments = [], []	×
179	for instrument in instruments:	×
180	file = os.path.join(data_bundle_path, "{}.h5".format(instrument))	×
181	if not os.path.exists(file):	×
182	not_exists_instruments.append(instrument)	×
183	continue	×
184	try:	×
185	with h5py.File(file, mode="r") as f:	×
186	for order_book_id in f.keys():	×
187	__ = f[order_book_id][:1]	×
188	except Exception:	×
189	corrupt_files.append(file)	×
190	if len(corrupt_files):	×
191	click.echo("{}:\n{}".format(_("corrupted files"), corrupt_files))	×
192	is_ok = input("{}(yes/no):".format(_("remove files"))).lower()	×
193	if is_ok in ["yes", "y"]:	×
194	[os.remove(file) for file in corrupt_files]	×
195	click.echo(_("remove success"))	×
196	elif is_ok in ["no", "n"]:	×
197	click.echo(_("corrupted files not remove"))	×
198	else:
199	click.echo(_("input error"))	×
200	elif len(not_exists_instruments):	×
201	click.echo(_("bundle's day bar is incomplete, please update bundle"))	×
202	else:
203	click.echo(_("good bundle's day bar"))	×
204

ricequant / rqalpha / 10504041958

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous