def markets(self, currency):
"""Get available coinmarketcap markets data.
It needs a currency as argument.
Args:
currency (str): Currency to get market data
Returns:
list: markets on wich provided currency is currently tradeable
"""
if self.is_symbol(currency):
currency = self.correspondences[currency]
url = urljoin(self.urls["web"], "currencies/%s/" % currency)
html = self._html(url)
response = []
marks = html.find('tbody').find_all('tr')
for m in marks:
_volume_24h = m.find('span', {'class': 'volume'}).getText()
volume_24h = self.parse_int(sub(r'\D', '', _volume_24h))
_price = m.find('span', {'class': 'price'}).getText()
_price = sub(r'\$| |\*|,', '', _price)
price = self.parse_float(_price)
_childs, childs = (m.contents, [])
for c in _childs:
if c != '\n':
childs.append(c)
for n, c in enumerate(childs):
nav = c.string
if n == 1:
exchange = str(nav)
elif n == 2:
pair = str(c.getText()).replace('/', self.pair_separator)
if pair[-1] == '*':
pair = pair.replace(' *', '')
elif n == 5:
percent_volume = self.parse_float(nav.replace('%', ''))
elif n == 6:
updated = nav == "Recently"
market = {'exchange': exchange, 'pair': pair,
'24h_volume_usd': volume_24h,
'price_usd': price,
'percent_volume': percent_volume,
"updated": updated}
response.append(market)
return response
python类urljoin()的实例源码
def exchange(self, name):
"""Obtain data from a exchange passed as argument
Example:
exchange('poloniex')
Args:
name (str): Exchange to retrieve data
Returns:
list: Data from all markets in a exchange
"""
url = urljoin(self.urls["web"], 'exchanges/%s/' % name)
html = self._html(url)
marks = html.find('table').find_all('tr')
response = []
for m in marks[1:]:
_childs, childs = (m.contents, [])
for c in _childs:
if c != '\n':
childs.append(c)
for n, c in enumerate(childs):
if n == 0:
rank = self.parse_int(c.getText())
elif n == 1:
name = str(c.getText())
elif n == 2:
market = str(c.getText().replace(
'/', self.pair_separator
))
elif n == 3:
_volume_24h_usd = sub(r'\$|,', '', c.getText())
volume_24h_usd = self.parse_int(_volume_24h_usd)
elif n == 4:
_price_usd = sub(r'\$| |\*', '', c.getText())
price_usd = self.parse_float(_price_usd)
elif n == 5:
_perc_volume = c.getText().replace('%', '')
perc_volume = self.parse_float(_perc_volume)
indicators = {'rank': rank,
'name': name,
'market': market,
'volume_24h_usd': volume_24h_usd,
'price_usd': price_usd,
'perc_volume': perc_volume}
response.append(indicators)
return response
def get_test_data(fn):
def _dir_content(url):
r = requests.get(url)
r.raise_for_status()
entries = re.findall(r'href="([a-zA-Z0-9_.-]+/?)"', r.text)
files = sorted(set(fn for fn in entries
if not fn.endswith('/')))
return files
def _file_size(url):
r = requests.head(url, headers={'Accept-Encoding': 'identity'})
r.raise_for_status()
return int(r.headers['content-length'])
def _download_file(url, fn_local):
if op.exists(fn_local):
if os.stat(fn_local).st_size == _file_size(url):
logger.info('Using cached file %s' % fn_local)
return fn_local
logger.info('Downloading %s...' % url)
fsize = _file_size(url)
r = requests.get(url, stream=True)
r.raise_for_status()
dl_bytes = 0
with open(fn_local, 'wb') as f:
for d in r.iter_content(chunk_size=1024):
dl_bytes += len(d)
f.write(d)
if dl_bytes != fsize:
raise DownloadError('Download incomplete!')
logger.info('Download completed.')
return fn_local
url = urljoin(data_uri, fn)
dl_dir = data_dir
_makedir(dl_dir)
if fn.endswith('/'):
dl_dir = op.join(data_dir, fn)
_makedir(dl_dir)
dl_files = _dir_content(url)
dl_files = zip([urljoin(url, u) for u in dl_files],
[op.join(dl_dir, f) for f in dl_files])
else:
dl_files = (url, op.join(data_dir, fn))
return _download_file(*dl_files)
return [_download_file(*f) for f in dl_files]
def send(qasm, device='sim_trivial_2', user=None, password=None,
shots=1, verbose=False):
"""
Sends QASM through the IBM API and runs the quantum circuit.
Args:
qasm: QASM representation of the circuit to run.
device (str): 'sim_trivial_2' or 'real' to run on simulator or on the
real chip, respectively.
user (str): IBM quantum experience user.
password (str): IBM quantum experience user password.
shots (int): Number of runs of the same circuit to collect statistics.
verbose (bool): If True, additional information is printed, such as
measurement statistics. Otherwise, the backend simply registers
one measurement result (same behavior as the projectq Simulator).
"""
try:
# check if the device is online
if device in ['ibmqx2', 'ibmqx4']:
url = 'Backends/{}/queue/status'.format(device)
r = requests.get(urljoin(_api_url_status, url))
online = r.json()['state']
if not online:
print("The device is offline (for maintenance?). Use the "
"simulator instead or try again later.")
raise DeviceOfflineError("Device is offline.")
if device == 'ibmqx2':
device = 'real'
if verbose:
print("Authenticating...")
user_id, access_token = _authenticate(user, password)
if verbose:
print("Running code...")
execution_id = _run(qasm, device, user_id, access_token, shots)
if verbose:
print("Waiting for results...")
res = _get_result(execution_id, access_token)
if verbose:
print("Done.")
return res
except requests.exceptions.HTTPError as err:
print("There was an error running your code:")
print(err)
except requests.exceptions.RequestException as err:
print("Looks like something is wrong with server:")
print(err)
except KeyError as err:
print("Failed to parse response:")
print(err)
